tegwick
0068e34644
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
267 lines
9.8 KiB
Python
267 lines
9.8 KiB
Python
"""Fast, guarded trigger for State Hub's forge-derived repository projection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from repo_manager.gitops import GitError, commit_paths, head_sha, push_ff
|
|
from repo_manager.identifiers import ensure_missing_work_record_identifiers
|
|
|
|
|
|
def _git(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, check=False)
|
|
|
|
|
|
def _ahead_behind(repo: Path) -> dict[str, Any]:
|
|
upstream = _git(repo, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
|
|
if upstream.returncode != 0:
|
|
raise ValueError("current branch has no upstream; central can only derive pushed work")
|
|
counts = _git(repo, "rev-list", "--left-right", "--count", "@{u}...HEAD")
|
|
if counts.returncode != 0:
|
|
raise ValueError(counts.stderr.strip() or "could not compare HEAD with upstream")
|
|
try:
|
|
behind, ahead = (int(value) for value in counts.stdout.split())
|
|
except ValueError as exc:
|
|
raise ValueError("could not parse Git ahead/behind state") from exc
|
|
return {"upstream": upstream.stdout.strip(), "behind": behind, "ahead": ahead}
|
|
|
|
|
|
def _dirty_authoritative_paths(repo: Path) -> list[str]:
|
|
result = _git(repo, "status", "--porcelain", "--", "workplans")
|
|
if result.returncode != 0:
|
|
raise ValueError(result.stderr.strip() or "could not inspect workplan files")
|
|
return [line[3:] for line in result.stdout.splitlines() if len(line) > 3]
|
|
|
|
|
|
def _pending_path(repo: Path) -> Path:
|
|
resolved = _git(
|
|
repo, "rev-parse", "--git-path", "repo-manager/pending-work-record-projection.json"
|
|
)
|
|
if resolved.returncode != 0 or not resolved.stdout.strip():
|
|
raise ValueError("could not resolve the repository-local Git state directory")
|
|
path = Path(resolved.stdout.strip())
|
|
return path if path.is_absolute() else repo / path
|
|
|
|
|
|
def _write_pending(repo: Path, payload: dict[str, Any]) -> Path:
|
|
path = _pending_path(repo)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary: Path | None = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False
|
|
) as handle:
|
|
temporary = Path(handle.name)
|
|
handle.write(json.dumps(payload, indent=2) + "\n")
|
|
temporary.replace(path)
|
|
temporary = None
|
|
finally:
|
|
if temporary is not None:
|
|
temporary.unlink(missing_ok=True)
|
|
return path
|
|
|
|
|
|
def sync_repository_projection(
|
|
repo_root: Path,
|
|
*,
|
|
api_base: str,
|
|
repo_slug: str | None = None,
|
|
expected_instance_label: str | None = "railiance01",
|
|
commit_identifiers: bool = True,
|
|
push: bool = False,
|
|
acknowledge_retirements: bool = False,
|
|
queue_on_unavailable: bool = True,
|
|
transport: httpx.BaseTransport | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Assign missing IDs, prove the commit is pushed, then reconcile once."""
|
|
started = time.perf_counter()
|
|
repo_root = repo_root.resolve()
|
|
slug = repo_slug or repo_root.name
|
|
|
|
dirty = _dirty_authoritative_paths(repo_root)
|
|
if dirty:
|
|
return {
|
|
"ok": False,
|
|
"status": "pending_commit",
|
|
"error": "authoritative workplan files have uncommitted changes",
|
|
"dirty_paths": dirty,
|
|
}
|
|
|
|
assignments = ensure_missing_work_record_identifiers(repo_root)
|
|
if assignments["files_changed"] and commit_identifiers:
|
|
try:
|
|
commit_paths(
|
|
repo_root,
|
|
assignments["files_changed"],
|
|
message=(
|
|
"repo.work.assign_missing_identifiers\n\n"
|
|
"source: repo-manager\nreason: deterministic projection registration\n"
|
|
),
|
|
)
|
|
except GitError as exc:
|
|
return {"ok": False, "status": "failed", "error": str(exc), "identifiers": assignments}
|
|
|
|
dirty = _dirty_authoritative_paths(repo_root)
|
|
if dirty:
|
|
return {
|
|
"ok": False,
|
|
"status": "pending_commit",
|
|
"error": "authoritative workplan files have uncommitted changes",
|
|
"dirty_paths": dirty,
|
|
"identifiers": assignments,
|
|
}
|
|
|
|
try:
|
|
git = _ahead_behind(repo_root)
|
|
except ValueError as exc:
|
|
return {"ok": False, "status": "failed", "error": str(exc), "identifiers": assignments}
|
|
if git["behind"]:
|
|
return {
|
|
"ok": False,
|
|
"status": "behind_upstream",
|
|
"error": "branch is behind upstream; reconcile Git before projection",
|
|
"git": git,
|
|
}
|
|
if git["ahead"] and push:
|
|
pushed, message = push_ff(repo_root)
|
|
if not pushed:
|
|
return {
|
|
"ok": False,
|
|
"status": "push_failed",
|
|
"error": message,
|
|
"git": git,
|
|
}
|
|
git = _ahead_behind(repo_root)
|
|
if git["ahead"]:
|
|
return {
|
|
"ok": True,
|
|
"status": "pending_push",
|
|
"message": "commit is not yet visible to the authoritative forge",
|
|
"git": git,
|
|
"identifiers": assignments,
|
|
}
|
|
|
|
commit = head_sha(repo_root)
|
|
if not commit:
|
|
return {"ok": False, "status": "failed", "error": "repository HEAD is unavailable"}
|
|
|
|
request_path = f"/repos/{slug}/work-record-projection/reconcile"
|
|
request_body = {
|
|
"expected_commit": commit,
|
|
"acknowledge_retirements": acknowledge_retirements,
|
|
}
|
|
# The review acknowledgement changes request semantics at the same commit.
|
|
# Bind the key to the complete POST, retaining identical keys for retries.
|
|
request_digest = hashlib.sha256(json.dumps(
|
|
{"method": "POST", "path": request_path, "body": request_body},
|
|
sort_keys=True, separators=(",", ":"),
|
|
).encode()).hexdigest()
|
|
pending = {
|
|
"schema": "repo-manager.pending-work-record-projection.v1",
|
|
"repo_slug": slug,
|
|
"expected_commit": commit,
|
|
"api_base": api_base.rstrip("/"),
|
|
"acknowledge_retirements": acknowledge_retirements,
|
|
"expected_instance_label": expected_instance_label,
|
|
}
|
|
headers = {
|
|
"Idempotency-Key": f"rmgr-projection:v2:{request_digest}",
|
|
"X-StateHub-Source-Agent": "repo-manager",
|
|
}
|
|
try:
|
|
with httpx.Client(
|
|
base_url=api_base.rstrip("/"),
|
|
timeout=httpx.Timeout(120.0, connect=5.0),
|
|
follow_redirects=True,
|
|
transport=transport,
|
|
) as client:
|
|
health = client.get("/state/health")
|
|
health.raise_for_status()
|
|
identity = health.json()
|
|
identity_ok = identity.get("instance_role") == "primary" and (
|
|
expected_instance_label is None
|
|
or identity.get("instance_label") == expected_instance_label
|
|
)
|
|
if not identity_ok:
|
|
if not queue_on_unavailable:
|
|
return {
|
|
"ok": False,
|
|
"status": "wrong_instance",
|
|
"instance": identity,
|
|
}
|
|
path = _write_pending(repo_root, {**pending, "reason": "wrong_instance"})
|
|
return {
|
|
"ok": True,
|
|
"status": "queued",
|
|
"reason": "wrong_instance",
|
|
"instance": identity,
|
|
"pending_path": str(path),
|
|
}
|
|
response = client.post(
|
|
request_path,
|
|
json=request_body,
|
|
headers=headers,
|
|
)
|
|
response.raise_for_status()
|
|
receipt = response.json()
|
|
except (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError) as exc:
|
|
if not queue_on_unavailable:
|
|
return {"ok": False, "status": "unavailable", "error": str(exc)}
|
|
path = _write_pending(repo_root, {**pending, "reason": "hub_unavailable"})
|
|
return {
|
|
"ok": True,
|
|
"status": "queued",
|
|
"reason": "hub_unavailable",
|
|
"pending_path": str(path),
|
|
}
|
|
except httpx.HTTPStatusError as exc:
|
|
detail: Any
|
|
try:
|
|
detail = exc.response.json()
|
|
except ValueError:
|
|
detail = exc.response.text[:500]
|
|
return {
|
|
"ok": False,
|
|
"status": "rejected",
|
|
"http_status": exc.response.status_code,
|
|
"error": detail,
|
|
}
|
|
|
|
# A successful HTTP response alone is not an exact-commit primary receipt.
|
|
outcome = receipt.get("outcome") if isinstance(receipt, dict) else None
|
|
if not (
|
|
isinstance(receipt, dict)
|
|
and receipt.get("schema") == "state-hub.repository-projection-reconcile.v1"
|
|
and receipt.get("instance_role") == "primary"
|
|
and receipt.get("instance_label") == identity.get("instance_label")
|
|
and receipt.get("expected_commit") == commit
|
|
and receipt.get("derived_commit") == commit
|
|
and isinstance(outcome, dict)
|
|
and outcome.get("repo_slug") == slug
|
|
and outcome.get("commit") == commit
|
|
and outcome.get("status") in {"applied", "noop", "refused"}
|
|
):
|
|
return {"ok": False, "status": "invalid_receipt", "repo_slug": slug,
|
|
"commit": commit, "error": "response does not bind the requested primary/repository/commit"}
|
|
outcome_status = outcome["status"]
|
|
if outcome_status in {"applied", "noop"}:
|
|
_pending_path(repo_root).unlink(missing_ok=True)
|
|
return {
|
|
"ok": outcome_status in {"applied", "noop"},
|
|
"status": outcome_status,
|
|
"repo_slug": slug,
|
|
"commit": commit,
|
|
"requests": 2,
|
|
"elapsed_ms": round((time.perf_counter() - started) * 1000, 1),
|
|
"git": git,
|
|
"identifiers": assignments,
|
|
"receipt": receipt,
|
|
}
|