tegwick
470ece82ed
Nine repositories are invisible to derivation because central may not read them. This adds the consuming half of the credential lane MASON-WP-0003 built. The cluster has no agent injector and no secrets-store CSI driver, so the pod authenticates to OpenBao with a projected ServiceAccount token (audience `openbao`, not the API server) and reads the KV path itself. `forgeRead.*` carries coordinates only; no credential is a chart value, an image layer, or a Kubernetes Secret. The credential reaches git through GIT_CONFIG_* setting http.extraHeader, not through `-c` and not through userinfo in the clone URL — both of those put the token in the process listing. It is redacted from ForgeDeriveError, which is logged, stored in reset outcomes, and returned over the API. Absent stays a supported state: with no credential, or with OpenBao unreachable, resolution returns None and public derivation runs unchanged. Raising would turn "nine repositories are unreadable" into "the pass failed", which is what T01 exists to prevent. Chart default is disabled. 717 pass. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 2583210@bnt-lap001 Assistant-Session: f2bff2d5-e9b2-4338-92ca-10282a927006
907 lines
35 KiB
Python
907 lines
35 KiB
Python
"""Derive a repository's work-record projection from the forge (STATE-WP-0083-T01).
|
|
|
|
`ADR-012` decision 1 makes the forge the projection source. Central does its own
|
|
reading: it clones the repository's default branch and derives from that, rather
|
|
than accepting a projection computed elsewhere — which `ADR-010` decision 5
|
|
forbids.
|
|
|
|
This module is read-only. Deriving must be safe to run at any time, because it is
|
|
what makes the reset in `T03` verifiable: you can always ask what the projection
|
|
*should* be without changing anything.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import base64
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
# Same derivation as ADR-007 / repo-manager, so a forge-derived projection and a
|
|
# preliminary overlay compute identical identities for the same record.
|
|
_WORK_RECORD_NAMESPACE = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")
|
|
_TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL)
|
|
_HEADING_RE = re.compile(r"^(#{1,4})\s+(.+?)$", re.MULTILINE)
|
|
DEFAULT_FORGE_BASE = "https://forgejo.coulomb.social/coulomb"
|
|
|
|
|
|
def derived_record_uuid(record_id: str) -> str:
|
|
return str(uuid.uuid5(_WORK_RECORD_NAMESPACE, f"helixforge\n{record_id}"))
|
|
|
|
|
|
class ForgeDeriveError(RuntimeError):
|
|
"""The repository could not be read from the forge."""
|
|
|
|
|
|
class ForgeUnreadableError(ForgeDeriveError):
|
|
"""Central is not permitted to read this repository — a policy, not a fault.
|
|
|
|
`ADR-012`'s premise (the forge is the projection source) holds only for
|
|
repositories central can read, and until `STATE-WP-0084` nothing said so:
|
|
a private repository failed the same way a broken one did, so "cannot read"
|
|
and "does not exist" were indistinguishable from the outside.
|
|
|
|
They must never be confused, because they authorise opposite things. A
|
|
repository that does not derive may have had its files removed deliberately;
|
|
a repository we cannot read tells us nothing at all about its files. Only
|
|
the first can justify retiring a record.
|
|
|
|
Forgejo answers an unauthenticated request for a private repository with a
|
|
404, so "not found" is classified as unreadable too. That is deliberate: the
|
|
two cases are genuinely indistinguishable at this layer, and the safe
|
|
reading of an ambiguous answer is the one that cannot destroy a record.
|
|
"""
|
|
|
|
|
|
# git says this in several ways depending on version, transport, and whether a
|
|
# credential helper is installed; all of them mean the same thing here.
|
|
_UNREADABLE_MARKERS = (
|
|
"could not read username",
|
|
"could not read password",
|
|
"authentication failed",
|
|
"terminal prompts disabled",
|
|
"invalid username or password",
|
|
"403 forbidden",
|
|
"the requested url returned error: 403",
|
|
"the requested url returned error: 401",
|
|
"repository not found",
|
|
"remote: not found",
|
|
"does not appear to be a git repository",
|
|
)
|
|
|
|
|
|
def _is_unreadable(message: str) -> bool:
|
|
low = message.lower()
|
|
if "not found" in low and "fatal: repository" in low:
|
|
return True
|
|
return any(marker in low for marker in _UNREADABLE_MARKERS)
|
|
|
|
|
|
@dataclass
|
|
class DerivedTask:
|
|
record_id: str
|
|
uuid: str
|
|
title: str | None
|
|
status: str | None
|
|
priority: str | None
|
|
|
|
|
|
@dataclass
|
|
class DerivedWorkplan:
|
|
record_id: str
|
|
uuid: str
|
|
title: str | None
|
|
status: str | None
|
|
relative_path: str
|
|
archived: bool
|
|
tasks: list[DerivedTask] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class DerivedProjection:
|
|
repo_slug: str
|
|
commit: str
|
|
workplans: list[DerivedWorkplan] = field(default_factory=list)
|
|
# False when the checkout has no `workplans/` directory at all. An empty
|
|
# projection then means "we did not find the records", which is not the same
|
|
# claim as "this repository has no records" — and only the second one could
|
|
# ever justify retiring anything.
|
|
records_source_present: bool = True
|
|
|
|
@property
|
|
def task_count(self) -> int:
|
|
return sum(len(w.tasks) for w in self.workplans)
|
|
|
|
@property
|
|
def retirement_eligible(self) -> bool:
|
|
"""Whether an absence in this projection is evidence of an absence.
|
|
|
|
A projection that could not be read never reaches this: it raises. What
|
|
this rules out is the quieter case — a clone that succeeded and returned
|
|
nothing, which is what would have retired every record in a repository
|
|
had a clone ever come back empty instead of failing (`STATE-WP-0084-T01`;
|
|
the near-miss was `vergabe-teilnahme`).
|
|
"""
|
|
return self.records_source_present
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schema": "state-hub.forge-projection.v1",
|
|
"repo_slug": self.repo_slug,
|
|
# Provenance is not optional: a projection that cannot name the
|
|
# commit it came from cannot be audited (ADR-012 decision 2).
|
|
"commit": self.commit,
|
|
"records_source_present": self.records_source_present,
|
|
"retirement_eligible": self.retirement_eligible,
|
|
"workplans": [
|
|
{
|
|
"record_id": w.record_id,
|
|
"uuid": w.uuid,
|
|
"title": w.title,
|
|
"status": w.status,
|
|
"relative_path": w.relative_path,
|
|
"archived": w.archived,
|
|
"tasks": [
|
|
{
|
|
"record_id": t.record_id,
|
|
"uuid": t.uuid,
|
|
"title": t.title,
|
|
"status": t.status,
|
|
"priority": t.priority,
|
|
}
|
|
for t in w.tasks
|
|
],
|
|
}
|
|
for w in self.workplans
|
|
],
|
|
}
|
|
|
|
|
|
from api.services.forge_credential import forge_read_token # noqa: F401
|
|
|
|
# Kept as module attributes so callers and tests that reached for them here
|
|
# still resolve after the sources moved to `forge_credential`.
|
|
FORGE_TOKEN_ENV = "FORGE_READ_TOKEN"
|
|
FORGE_TOKEN_FILE_ENV = "FORGE_READ_TOKEN_FILE"
|
|
|
|
|
|
def _credential_env(token: str | None) -> dict[str, str]:
|
|
"""Git config carrying the credential, passed by environment not argv.
|
|
|
|
`-c http.extraHeader=...` would place the token in the process command line,
|
|
where it is readable by anything that can run `ps` and lands in any log that
|
|
records invocations. GIT_CONFIG_* achieves the same configuration without
|
|
that exposure.
|
|
"""
|
|
if not token:
|
|
return {}
|
|
header = base64.b64encode(f"x-access-token:{token}".encode()).decode()
|
|
return {
|
|
"GIT_CONFIG_COUNT": "1",
|
|
"GIT_CONFIG_KEY_0": "http.extraHeader",
|
|
"GIT_CONFIG_VALUE_0": f"Authorization: Basic {header}",
|
|
}
|
|
|
|
|
|
def _run_git(
|
|
*args: str, cwd: str | None = None, timeout: float = 120.0, token: str | None = None
|
|
) -> str:
|
|
# Without this a clone of a private repository blocks on a username prompt
|
|
# instead of failing, and an unattended derivation pass hangs rather than
|
|
# reporting. Failing fast is what makes the unreadable case observable.
|
|
env = {**os.environ, "GIT_TERMINAL_PROMPT": "0", "GIT_ASKPASS": "", "GCM_INTERACTIVE": "never"}
|
|
env.update(_credential_env(token))
|
|
proc = subprocess.run(
|
|
["git", *args], cwd=cwd, capture_output=True, text=True, timeout=timeout, env=env
|
|
)
|
|
if proc.returncode != 0:
|
|
detail = (proc.stderr or proc.stdout).strip()[:400]
|
|
if token:
|
|
# Never let a credential reach an exception that is logged, stored
|
|
# in a reset outcome, or returned over the API.
|
|
detail = detail.replace(token, "***")
|
|
raise ForgeDeriveError(detail)
|
|
return proc.stdout.strip()
|
|
|
|
|
|
def _split_frontmatter(text: str) -> tuple[dict, str]:
|
|
if not text.startswith("---"):
|
|
return {}, text
|
|
end = text.find("\n---", 3)
|
|
if end == -1:
|
|
return {}, text
|
|
raw = text[3:end]
|
|
body = text[end + 4 :]
|
|
try:
|
|
meta = yaml.safe_load(raw) or {}
|
|
except yaml.YAMLError:
|
|
return {}, text
|
|
return (meta if isinstance(meta, dict) else {}), body
|
|
|
|
|
|
def _parse_tasks(body: str) -> list[DerivedTask]:
|
|
headings = [
|
|
(m.start(), m.group(2).strip()) for m in _HEADING_RE.finditer(body)
|
|
]
|
|
out: list[DerivedTask] = []
|
|
for m in _TASK_BLOCK_RE.finditer(body):
|
|
try:
|
|
block = yaml.safe_load(m.group(1).strip()) or {}
|
|
except yaml.YAMLError:
|
|
continue
|
|
if not isinstance(block, dict):
|
|
continue
|
|
rid = str(block.get("id") or "").strip()
|
|
if not rid:
|
|
continue
|
|
title = block.get("title")
|
|
if not title:
|
|
prev = [t for pos, t in headings if pos < m.start()]
|
|
title = prev[-1] if prev else None
|
|
out.append(
|
|
DerivedTask(
|
|
record_id=rid,
|
|
# Derived, not read from the file: the forge projection must not
|
|
# inherit an identifier the file happens to carry.
|
|
uuid=derived_record_uuid(rid),
|
|
title=title,
|
|
status=(str(block["status"]).strip() if block.get("status") else None),
|
|
priority=(str(block["priority"]).strip() if block.get("priority") else None),
|
|
)
|
|
)
|
|
return out
|
|
|
|
|
|
def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> DerivedProjection:
|
|
"""Derive a projection from an already-materialised checkout."""
|
|
proj = DerivedProjection(repo_slug=repo_slug, commit=commit)
|
|
wp_dir = repo_root / "workplans"
|
|
if not wp_dir.is_dir():
|
|
proj.records_source_present = False
|
|
return proj
|
|
for path in sorted(wp_dir.rglob("*.md")):
|
|
if path.name.startswith("."):
|
|
continue
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeDecodeError):
|
|
continue
|
|
meta, body = _split_frontmatter(text)
|
|
if str(meta.get("type") or "").strip() != "workplan":
|
|
continue
|
|
rid = str(meta.get("id") or "").strip()
|
|
if not rid:
|
|
continue
|
|
proj.workplans.append(
|
|
DerivedWorkplan(
|
|
record_id=rid,
|
|
uuid=derived_record_uuid(rid),
|
|
title=(str(meta["title"]).strip() if meta.get("title") else None),
|
|
status=(str(meta["status"]).strip() if meta.get("status") else None),
|
|
relative_path=str(path.relative_to(repo_root).as_posix()),
|
|
archived=path.parent.name == "archived",
|
|
tasks=_parse_tasks(body),
|
|
)
|
|
)
|
|
proj.workplans.sort(key=lambda w: w.record_id)
|
|
return proj
|
|
|
|
|
|
def derive_from_forge(
|
|
repo_slug: str, *, forge_base: str = DEFAULT_FORGE_BASE, ref: str | None = None
|
|
) -> DerivedProjection:
|
|
"""Clone the repository's default branch from the forge and derive from it.
|
|
|
|
A fresh shallow clone every time, deliberately: the source is what the forge
|
|
holds now, and a reused working copy is how a projection ends up reflecting
|
|
someone's local state instead (ADR-012 context).
|
|
"""
|
|
url = f"{forge_base.rstrip('/')}/{repo_slug}.git"
|
|
token = forge_read_token()
|
|
with tempfile.TemporaryDirectory(prefix=f"forge-{repo_slug}-") as tmp:
|
|
args = ["clone", "--depth", "1", "--quiet"]
|
|
if ref:
|
|
args += ["--branch", ref]
|
|
try:
|
|
_run_git(*args, url, tmp, token=token)
|
|
except subprocess.TimeoutExpired as exc:
|
|
raise ForgeDeriveError(f"clone timed out for {repo_slug}") from exc
|
|
except ForgeDeriveError as exc:
|
|
# Classify before propagating. A caller that cannot tell "not
|
|
# permitted" from "broken" will eventually treat one as the other.
|
|
if _is_unreadable(str(exc)):
|
|
raise ForgeUnreadableError(
|
|
f"{repo_slug} could not be read from the forge: {exc}"
|
|
) from exc
|
|
raise
|
|
commit = _run_git("rev-parse", "HEAD", cwd=tmp)
|
|
return derive_from_checkout(Path(tmp), repo_slug, commit)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Comparison against what the hub currently holds (STATE-WP-0083-T02)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
|
|
_ARCHIVE_PREFIX_RE = re.compile(r"^\d{6}-")
|
|
|
|
|
|
def _path_key(path: str) -> str:
|
|
"""Normalise a workplan path so an archived copy matches its live one."""
|
|
name = path.rsplit("/", 1)[-1]
|
|
return _ARCHIVE_PREFIX_RE.sub("", name).strip().lower()
|
|
|
|
|
|
@dataclass
|
|
class ProjectionDiff:
|
|
"""What a reset would change, computed without changing anything.
|
|
|
|
This is what makes `ADR-012` decision 7's "verifiable" real: the reset can
|
|
always be inspected before it runs, and its result compared against the forge
|
|
afterwards.
|
|
"""
|
|
|
|
repo_slug: str
|
|
commit: str
|
|
missing: list[dict[str, Any]] = field(default_factory=list) # forge has, hub lacks
|
|
stale: list[dict[str, Any]] = field(default_factory=list) # hub has, forge lacks
|
|
differing: list[dict[str, Any]] = field(default_factory=list) # both, fields differ
|
|
# Set when the source could not support a claim of absence, so `stale` was
|
|
# deliberately left empty rather than computed (`STATE-WP-0084-T01`).
|
|
stale_withheld: str | None = None
|
|
|
|
@property
|
|
def clean(self) -> bool:
|
|
return not (self.missing or self.stale or self.differing)
|
|
|
|
@property
|
|
def would_remove(self) -> int:
|
|
return len(self.stale)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schema": "state-hub.projection-diff.v1",
|
|
"repo_slug": self.repo_slug,
|
|
"commit": self.commit,
|
|
"clean": self.clean,
|
|
"counts": {
|
|
"missing": len(self.missing),
|
|
"stale": len(self.stale),
|
|
"differing": len(self.differing),
|
|
},
|
|
"stale_withheld": self.stale_withheld,
|
|
"missing": self.missing,
|
|
"stale": self.stale,
|
|
"differing": self.differing,
|
|
}
|
|
|
|
|
|
def diff_against_hub(
|
|
derived: DerivedProjection,
|
|
hub_workplans: list[dict[str, Any]],
|
|
hub_tasks_by_workplan: dict[str, list[dict[str, Any]]],
|
|
) -> ProjectionDiff:
|
|
"""Compare a derived projection with the hub's current records.
|
|
|
|
Pure: takes the hub's state as data rather than reading it, so the comparison
|
|
is testable without a database and cannot accidentally mutate anything.
|
|
"""
|
|
d = ProjectionDiff(repo_slug=derived.repo_slug, commit=derived.commit)
|
|
if not derived.retirement_eligible:
|
|
# Compute what is missing and what differs as usual — those only ever
|
|
# add or correct. Absence is the one conclusion this source cannot
|
|
# support, so it is not drawn at all rather than drawn and then filtered.
|
|
d.stale_withheld = (
|
|
"the checkout has no workplans/ directory, so an absent record is "
|
|
"unexplained rather than evidence of removal"
|
|
)
|
|
|
|
# Match on canonical identity, never on UUID. Most hub records still carry
|
|
# pre-ADR-007 random identifiers, so a UUID-keyed comparison reports every
|
|
# record as simultaneously missing and stale — and a reset built on that
|
|
# would destroy and recreate the entire projection. The canonical record id
|
|
# is what is stable across the identifier migration; the backing file is the
|
|
# fallback when a hub slug was derived from a filename rather than an id.
|
|
def _wp_key(record_id: str | None, slug: str | None, path: str | None) -> str:
|
|
if record_id:
|
|
return record_id.strip().lower()
|
|
if path:
|
|
return _path_key(path)
|
|
return (slug or "").strip().lower()
|
|
|
|
want_wp = {w.record_id.strip().lower(): w for w in derived.workplans}
|
|
have_wp: dict[str, dict[str, Any]] = {}
|
|
want_paths = {_path_key(w.relative_path): k for k, w in want_wp.items()}
|
|
for w in hub_workplans:
|
|
slug = str(w.get("slug") or "")
|
|
key = slug.strip().lower()
|
|
if key not in want_wp:
|
|
# Slugs were not always the canonical id; fall back to the file.
|
|
bp = w.get("backing_relative_path")
|
|
if bp and _path_key(bp) in want_paths:
|
|
key = want_paths[_path_key(bp)]
|
|
else:
|
|
cand = [k for k in want_wp if slug.lower().startswith(k + "-")]
|
|
if len(cand) == 1:
|
|
key = cand[0]
|
|
have_wp[key] = w
|
|
|
|
for key, w in want_wp.items():
|
|
if key not in have_wp:
|
|
d.missing.append({"kind": "workplan", "record_id": w.record_id, "uuid": w.uuid})
|
|
for key, w in have_wp.items():
|
|
if key not in want_wp:
|
|
if d.stale_withheld:
|
|
continue
|
|
uid = str(w["id"])
|
|
d.stale.append(
|
|
{
|
|
"kind": "workplan",
|
|
"uuid": uid,
|
|
"slug": w.get("slug"),
|
|
"status": w.get("status"),
|
|
# A stale record with no backing file is the case that must
|
|
# never be destroyed silently (ADR-012 decision 7).
|
|
"has_backing_file": bool(
|
|
w.get("backing_filename") or w.get("backing_relative_path")
|
|
),
|
|
}
|
|
)
|
|
for key, w in want_wp.items():
|
|
cur = have_wp.get(key)
|
|
uid = str(cur["id"]) if cur else w.uuid
|
|
if not cur:
|
|
continue
|
|
changed = {}
|
|
if (cur.get("status") or None) != (w.status or None):
|
|
changed["status"] = {"hub": cur.get("status"), "forge": w.status}
|
|
cur_path = cur.get("backing_relative_path") or None
|
|
if cur_path != w.relative_path:
|
|
changed["backing_relative_path"] = {"hub": cur_path, "forge": w.relative_path}
|
|
if changed:
|
|
d.differing.append(
|
|
{"kind": "workplan", "record_id": w.record_id, "uuid": uid, "changed": changed}
|
|
)
|
|
|
|
for w in derived.workplans:
|
|
hub_rows = hub_tasks_by_workplan.get(w.uuid, [])
|
|
if not hub_rows:
|
|
cur = have_wp.get(w.record_id.strip().lower())
|
|
if cur:
|
|
hub_rows = hub_tasks_by_workplan.get(str(cur["id"]), [])
|
|
# Match on the canonical record id where the hub has one. Rows created
|
|
# before STATE-WP-0083-T06 fall back to title, which is why those are
|
|
# reported rather than acted on: a renamed heading is indistinguishable
|
|
# from a replaced task under title matching.
|
|
def _task_key(record_id: str | None, title: str | None) -> str:
|
|
if record_id:
|
|
return record_id.strip().lower()
|
|
return "title:" + (title or "").strip().lower()
|
|
|
|
have = {
|
|
_task_key(t.get("record_id"), t.get("title")): t for t in hub_rows
|
|
}
|
|
want = {_task_key(t.record_id, t.title): t for t in w.tasks}
|
|
for key, t in want.items():
|
|
if key not in have:
|
|
d.missing.append(
|
|
{"kind": "task", "record_id": t.record_id, "uuid": t.uuid,
|
|
"workplan": w.record_id}
|
|
)
|
|
for key, t in have.items():
|
|
if key not in want:
|
|
if d.stale_withheld:
|
|
continue
|
|
uid = str(t["id"])
|
|
d.stale.append(
|
|
{"kind": "task", "uuid": uid, "title": t.get("title"),
|
|
"status": t.get("status"), "workplan": w.record_id,
|
|
"has_backing_file": False}
|
|
)
|
|
for key, t in want.items():
|
|
cur = have.get(key)
|
|
if cur and (cur.get("status") or None) != (t.status or None):
|
|
d.differing.append(
|
|
{"kind": "task", "record_id": t.record_id, "uuid": t.uuid,
|
|
"workplan": w.record_id,
|
|
"changed": {"status": {"hub": cur.get("status"), "forge": t.status}}}
|
|
)
|
|
return d
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Applying the reset (STATE-WP-0083-T03)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class ResetOutcome:
|
|
repo_slug: str
|
|
commit: str
|
|
status: str # applied | refused | noop | unreadable
|
|
created: list[str] = field(default_factory=list)
|
|
updated: list[str] = field(default_factory=list)
|
|
retired: list[str] = field(default_factory=list)
|
|
refused: list[dict[str, Any]] = field(default_factory=list)
|
|
notes: list[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schema": "state-hub.projection-reset.v1",
|
|
"repo_slug": self.repo_slug,
|
|
"commit": self.commit,
|
|
"status": self.status,
|
|
"counts": {
|
|
"created": len(self.created),
|
|
"updated": len(self.updated),
|
|
"retired": len(self.retired),
|
|
"refused": len(self.refused),
|
|
},
|
|
"created": self.created,
|
|
"updated": self.updated,
|
|
"retired": self.retired,
|
|
"refused": self.refused,
|
|
"notes": self.notes,
|
|
}
|
|
|
|
|
|
RETIRE_REASON = "no longer derived from the forge"
|
|
|
|
|
|
async def reset_repository_projection(
|
|
session: Any,
|
|
repo_slug: str,
|
|
*,
|
|
acknowledge_retirements: bool = False,
|
|
forge_base: str = DEFAULT_FORGE_BASE,
|
|
derived: DerivedProjection | None = None,
|
|
) -> ResetOutcome:
|
|
"""Reconcile one repository's workplan projection against the forge.
|
|
|
|
Creates what the forge has and the hub lacks, updates what differs, and
|
|
retires what no longer derives. It does not delete: hub-native records
|
|
reference workplans with `ON DELETE RESTRICT`, and destroying a progress
|
|
event to tidy a derived projection would lose hub-native truth to fix a
|
|
derived-state problem (`ADR-012` decision 7 as amended).
|
|
|
|
Retirement is refused by default. A record that stops deriving may mean the
|
|
file was removed deliberately — or that someone pointed this at the wrong
|
|
branch. The caller must say which.
|
|
|
|
Scope: workplans, and the tasks of workplans being created. Tasks of
|
|
*existing* workplans are left alone, because hub task rows carry no
|
|
canonical identifier and can only be matched by title — renaming a heading
|
|
would otherwise destroy and recreate its record (`T06`).
|
|
"""
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import select
|
|
|
|
from api.models.managed_repo import ManagedRepo
|
|
from api.models.task import Task
|
|
from api.models.workplan import Workplan
|
|
|
|
if derived is None:
|
|
try:
|
|
derived = derive_from_forge(repo_slug, forge_base=forge_base)
|
|
except ForgeUnreadableError as exc:
|
|
# Not an error: a statement about what central is permitted to see.
|
|
# Reported as its own status so a caller cannot mistake it for a
|
|
# repository whose records stopped deriving (`STATE-WP-0084-T01`).
|
|
out = ResetOutcome(repo_slug=repo_slug, commit="", status="unreadable")
|
|
out.refused.append(
|
|
{"reason": "repository could not be read from the forge", "slug": repo_slug,
|
|
"detail": str(exc)[:300]}
|
|
)
|
|
out.notes.append(
|
|
"Nothing was changed and nothing was retired. This says nothing "
|
|
"about whether the repository's records still exist."
|
|
)
|
|
return out
|
|
outcome = ResetOutcome(repo_slug=repo_slug, commit=derived.commit, status="noop")
|
|
|
|
repo = (
|
|
await session.execute(select(ManagedRepo).where(ManagedRepo.slug == repo_slug))
|
|
).scalar_one_or_none()
|
|
if repo is None:
|
|
outcome.status = "refused"
|
|
outcome.refused.append({"reason": "repository is not registered", "slug": repo_slug})
|
|
return outcome
|
|
|
|
rows = list(
|
|
(
|
|
await session.execute(select(Workplan).where(Workplan.repo_id == repo.id))
|
|
).scalars()
|
|
)
|
|
|
|
want = {w.record_id.strip().lower(): w for w in derived.workplans}
|
|
want_paths = {_path_key(w.relative_path): k for k, w in want.items()}
|
|
|
|
matched: dict[str, Any] = {}
|
|
for row in rows:
|
|
key = (row.slug or "").strip().lower()
|
|
if key not in want:
|
|
bp = row.backing_relative_path
|
|
if bp and _path_key(bp) in want_paths:
|
|
key = want_paths[_path_key(bp)]
|
|
else:
|
|
cand = [k for k in want if key.startswith(k + "-")]
|
|
key = cand[0] if len(cand) == 1 else key
|
|
matched[key] = row
|
|
|
|
# An identifier this repository would create may already belong to another
|
|
# repository. Two repositories creating the same daily identifier on the
|
|
# same day is a documented case (CUST-WP-0066), and the derivation is
|
|
# deliberately deterministic, so the collision is real rather than
|
|
# incidental. Refuse and say so: a constraint violation is a stack trace,
|
|
# a refusal is something the caller can rule on.
|
|
creating = [w for k, w in want.items() if k not in matched]
|
|
if creating:
|
|
foreign = list(
|
|
(
|
|
await session.execute(
|
|
select(Workplan).where(
|
|
Workplan.id.in_([uuid.UUID(w.uuid) for w in creating]),
|
|
Workplan.repo_id != repo.id,
|
|
)
|
|
)
|
|
).scalars()
|
|
)
|
|
# `slug` carries its own unique constraint across the whole table, so an
|
|
# identifier check alone is not enough: two repositories can derive
|
|
# different identifiers whose slugs still collide. Missing this is what
|
|
# left disaster-control raising IntegrityError after the identifier
|
|
# refusal was added.
|
|
slug_clash = list(
|
|
(
|
|
await session.execute(
|
|
select(Workplan).where(
|
|
Workplan.slug.in_([w.record_id.lower() for w in creating]),
|
|
Workplan.repo_id != repo.id,
|
|
)
|
|
)
|
|
).scalars()
|
|
)
|
|
if slug_clash:
|
|
held = {(r.slug or "").lower(): r for r in slug_clash}
|
|
outcome.status = "refused"
|
|
for w in creating:
|
|
row = held.get(w.record_id.lower())
|
|
if row is None:
|
|
continue
|
|
outcome.refused.append(
|
|
{
|
|
"reason": "slug already belongs to another repository",
|
|
"record_id": w.record_id,
|
|
"slug": w.record_id.lower(),
|
|
"held_by_id": str(row.id),
|
|
}
|
|
)
|
|
outcome.notes.append(
|
|
"Identifier collision is an identity decision, not a projection "
|
|
"one; acknowledging retirements does not authorise it."
|
|
)
|
|
return outcome
|
|
|
|
if foreign:
|
|
owned = {str(r.id): r for r in foreign}
|
|
outcome.status = "refused"
|
|
for w in creating:
|
|
held = owned.get(w.uuid)
|
|
if held is None:
|
|
continue
|
|
outcome.refused.append(
|
|
{
|
|
"reason": "derived identifier already belongs to another repository",
|
|
"record_id": w.record_id,
|
|
"uuid": w.uuid,
|
|
"held_by_slug": held.slug,
|
|
}
|
|
)
|
|
outcome.notes.append(
|
|
"Identifier collision is an identity decision, not a projection "
|
|
"one; acknowledging retirements does not authorise it."
|
|
)
|
|
return outcome
|
|
|
|
stale = [
|
|
r for k, r in matched.items()
|
|
if k not in want and r.projection_retired_at is None
|
|
]
|
|
if stale and not derived.retirement_eligible:
|
|
# Acknowledgement cannot authorise this. The caller is confirming that
|
|
# records which stopped deriving should be retired; here nothing has
|
|
# been shown to have stopped deriving, because the source produced no
|
|
# records to compare against. Consenting to a conclusion is not the
|
|
# same as the evidence for it existing.
|
|
outcome.status = "refused"
|
|
for r in stale:
|
|
outcome.refused.append(
|
|
{
|
|
"reason": "source produced no records; absence is unexplained",
|
|
"slug": r.slug,
|
|
"status": r.status,
|
|
"backing_relative_path": r.backing_relative_path,
|
|
}
|
|
)
|
|
outcome.notes.append(
|
|
"The checkout has no workplans/ directory. Retirement withheld "
|
|
"regardless of acknowledgement (STATE-WP-0084-T01)."
|
|
)
|
|
return outcome
|
|
if stale and not acknowledge_retirements:
|
|
outcome.status = "refused"
|
|
for r in stale:
|
|
outcome.refused.append(
|
|
{
|
|
"reason": "would be retired; the forge no longer derives it",
|
|
"slug": r.slug,
|
|
"status": r.status,
|
|
"backing_relative_path": r.backing_relative_path,
|
|
}
|
|
)
|
|
outcome.notes.append(
|
|
"Re-run with acknowledgement to retire these. Nothing was changed."
|
|
)
|
|
return outcome
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
|
|
for key, w in want.items():
|
|
row = matched.get(key)
|
|
if row is None:
|
|
row = Workplan(
|
|
id=uuid.UUID(w.uuid),
|
|
repo_id=repo.id,
|
|
topic_id=repo.topic_id,
|
|
slug=w.record_id.lower(),
|
|
title=w.title or w.record_id,
|
|
status=w.status or "proposed",
|
|
backing_filename=w.relative_path.rsplit("/", 1)[-1],
|
|
backing_relative_path=w.relative_path,
|
|
backing_archived=w.archived,
|
|
derived_from_commit=derived.commit,
|
|
)
|
|
session.add(row)
|
|
await session.flush()
|
|
for t in w.tasks:
|
|
# Safe only because nothing exists to mis-match against: this
|
|
# workplan is new to the hub.
|
|
session.add(
|
|
Task(
|
|
id=uuid.UUID(t.uuid),
|
|
workplan_id=row.id,
|
|
record_id=t.record_id,
|
|
title=t.title or t.record_id,
|
|
status=t.status or "todo",
|
|
priority=t.priority or "medium",
|
|
)
|
|
)
|
|
outcome.created.append(w.record_id)
|
|
continue
|
|
|
|
changed = False
|
|
if w.status and row.status != w.status:
|
|
row.status = w.status
|
|
changed = True
|
|
if row.backing_relative_path != w.relative_path:
|
|
row.backing_relative_path = w.relative_path
|
|
row.backing_filename = w.relative_path.rsplit("/", 1)[-1]
|
|
row.backing_archived = w.archived
|
|
changed = True
|
|
if row.projection_retired_at is not None:
|
|
# It derives again; un-retire rather than leaving a contradiction.
|
|
row.projection_retired_at = None
|
|
row.projection_retired_reason = None
|
|
changed = True
|
|
if row.derived_from_commit != derived.commit:
|
|
row.derived_from_commit = derived.commit
|
|
changed = True
|
|
if changed:
|
|
outcome.updated.append(w.record_id)
|
|
|
|
for r in stale:
|
|
r.projection_retired_at = now
|
|
r.projection_retired_reason = RETIRE_REASON
|
|
outcome.retired.append(r.slug or str(r.id))
|
|
|
|
if outcome.created or outcome.updated or outcome.retired:
|
|
outcome.status = "applied"
|
|
outcome.notes.append(
|
|
"Tasks of existing workplans were not touched; hub tasks carry no "
|
|
"canonical identifier (STATE-WP-0083-T06)."
|
|
)
|
|
return outcome
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fleet form (STATE-WP-0083-T04)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class FleetResetOutcome:
|
|
results: dict[str, dict[str, Any]] = field(default_factory=dict)
|
|
errors: dict[str, str] = field(default_factory=dict)
|
|
# Kept apart from `errors` on purpose. Nine repositories sitting in an error
|
|
# bucket read as nine broken repositories; they were nine we were not
|
|
# allowed to read, which is a different thing to go and fix
|
|
# (`STATE-WP-0083-T04`, 2026-08-26).
|
|
unreadable: dict[str, str] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
by_status: dict[str, int] = {}
|
|
for r in self.results.values():
|
|
by_status[r["status"]] = by_status.get(r["status"], 0) + 1
|
|
return {
|
|
"schema": "state-hub.fleet-projection-reset.v1",
|
|
"repositories": len(self.results) + len(self.errors) + len(self.unreadable),
|
|
"by_status": by_status,
|
|
"errored": len(self.errors),
|
|
"unreadable_count": len(self.unreadable),
|
|
"unreadable": self.unreadable,
|
|
"totals": {
|
|
k: sum(r["counts"][k] for r in self.results.values())
|
|
for k in ("created", "updated", "retired", "refused")
|
|
},
|
|
"results": self.results,
|
|
"errors": self.errors,
|
|
}
|
|
|
|
|
|
async def reset_fleet_projection(
|
|
session_factory: Any,
|
|
repo_slugs: list[str],
|
|
*,
|
|
acknowledge_retirements: bool = False,
|
|
forge_base: str = DEFAULT_FORGE_BASE,
|
|
) -> FleetResetOutcome:
|
|
"""Reset every repository, one at a time, sharing the per-repository path.
|
|
|
|
The fleet form is a loop over the repository form and nothing else
|
|
(`ADR-012` decision 7). The rarely-run wide operation must be the frequently
|
|
run narrow one, or the wide one is trusted on the strength of never having
|
|
been exercised.
|
|
|
|
A repository that refuses or errors is recorded and the pass continues.
|
|
Aborting on the first refusal would mean one unresolved repository blocks
|
|
reconstruction everywhere — which in practice means permanently.
|
|
|
|
Each repository gets its own session, so one failure cannot roll back
|
|
another's work or leave a poisoned transaction behind.
|
|
"""
|
|
outcome = FleetResetOutcome()
|
|
for slug in repo_slugs:
|
|
try:
|
|
async with session_factory() as session:
|
|
result = await reset_repository_projection(
|
|
session,
|
|
slug,
|
|
acknowledge_retirements=acknowledge_retirements,
|
|
forge_base=forge_base,
|
|
)
|
|
if result.status == "applied":
|
|
await session.commit()
|
|
else:
|
|
await session.rollback()
|
|
if result.status == "unreadable":
|
|
detail = next(
|
|
(r.get("detail", "") for r in result.refused), ""
|
|
)
|
|
outcome.unreadable[slug] = detail or "could not be read from the forge"
|
|
else:
|
|
outcome.results[slug] = result.to_dict()
|
|
except ForgeUnreadableError as exc:
|
|
# Reachable when the caller supplied its own derivation path.
|
|
outcome.unreadable[slug] = str(exc)[:300]
|
|
except Exception as exc: # noqa: BLE001 - one repo must not end the pass
|
|
outcome.errors[slug] = f"{type(exc).__name__}: {exc}"[:300]
|
|
return outcome
|