Spaces:
Running
Running
| """Bootstrap permanent maintenance state from deploy-generated source files.""" | |
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| from typing import Any, Dict | |
| from .maintenance import MaintenanceLedger | |
| def bootstrap(ledger: MaintenanceLedger, *, root: str | Path = "workspace") -> Dict[str, Any]: | |
| root = Path(root) | |
| seed = root / "maintenance_seed.jsonl" | |
| source = root / "maintenance_source.json" | |
| events = [] | |
| if seed.exists(): | |
| for line in seed.read_text(encoding="utf-8").splitlines(): | |
| try: | |
| value = json.loads(line) | |
| except Exception: | |
| continue | |
| if isinstance(value, dict): | |
| events.append(value) | |
| imported = ledger.import_events(events, source="deployment.seed") if events else {"added": 0, "duplicate": 0} | |
| state = {} | |
| if source.exists(): | |
| try: | |
| value = json.loads(source.read_text(encoding="utf-8")) | |
| if isinstance(value, dict): | |
| state = value | |
| except Exception: | |
| state = {} | |
| if state: | |
| git_state = { | |
| "repository": state.get("repository"), | |
| "branch": state.get("branch"), | |
| "head": state.get("head"), | |
| "parents": state.get("parents") or [], | |
| "history_commits": state.get("history_commits", 0), | |
| "source_generated_at": state.get("generated_at"), | |
| } | |
| ledger.set_cell("maint.git.state", git_state, source="deployment.seed", summary="Canonical deployed git state") | |
| ledger.set_cell( | |
| "maint.git.parity", | |
| { | |
| "repository": state.get("repository"), | |
| "github_head": state.get("head"), | |
| "deployed_source_head": state.get("head"), | |
| "parity": "matched", | |
| }, | |
| source="deployment.seed", | |
| summary="GitHub source and deployed HF source match", | |
| ) | |
| ledger.set_cell( | |
| "maint.deploy.state", | |
| { | |
| "state": "source-synced", | |
| "repository": state.get("repository"), | |
| "head": state.get("head"), | |
| "github_run_id": state.get("github_run_id"), | |
| "github_run_attempt": state.get("github_run_attempt"), | |
| "github_workflow": state.get("github_workflow"), | |
| }, | |
| source="github.deploy", | |
| summary="Canonical source sync reached Hugging Face repository", | |
| ) | |
| ledger.set_cell( | |
| "maint.change.cursor", | |
| {"cursor": ledger.state().get("cursor", 0), "seed_events": len(events), **imported}, | |
| source="maintenance.bootstrap", | |
| summary="Maintenance ledger cursor", | |
| ) | |
| return {"events": len(events), "imported": imported, "source": state} | |