Compare commits

..
Author SHA1 Message Date
sysadmin e344e68a13 fix(fleet): dedicated retirement capability, identity proof, external fencing
Addresses the three blocking findings in review 657 on PR #982 and nothing
else.

B1 — apply was authorized by gitea.read
---------------------------------------
Plan and apply shared the observational permission class, so any profile that
could look could also destroy; the mutation landing in the local control-plane
registry rather than in Gitea makes it no less a mutation.

Apply now requires its own capability, gitea.worker_registry.retire, checked at
entry and re-resolved immediately before the registry mutation. Plan stays on
gitea.read. No profile holds the new permission by default, so author,
reviewer, merger, and read-only profiles fail closed on the permission itself
rather than on the role check alone; the controller/reconciler role restriction
remains as defence in depth. Granting it is a deliberate operator edit to
profiles.json, and removing it revokes apply completely. No new Gitea write
permission is introduced and no author permission is broadened.

B2 — complete legacy rows could be retired without identity proof
-----------------------------------------------------------------
"Incomplete identity" previously meant only that pre-existing columns were
null, which a legacy-pid row satisfies trivially. Retirement now requires an
affirmative two-part proof: launcher-minted inst- attribution, plus fencing
evidence (host_id, boot_id, process_start_time) that turns a bare pid into a
statement about one process incarnation.

New mcp_process_fencing supplies those probes; every one returns None rather
than guessing, and None always preserves. Rows written before these columns
existed, and rows on legacy instance identities, are preserved permanently —
they are retired only after re-registering under a trusted identity. That
legacy rows would otherwise remain outstanding is explicitly not treated as
grounds for a weaker proof. A live pid stays an absolute block even across a
boot boundary, and pid reuse is reported distinctly from a live worker.

B3 — lease and OS liveness sat outside the registry transaction
---------------------------------------------------------------
BEGIN IMMEDIATE locks the worker registry only, and the per-target loop runs
after revalidation, so a lease acquired or a pid revived in between would go
unnoticed — the registry-column guard cannot catch it because no registry
column changed.

Two mechanisms now close that window, both applied per target immediately
before its own write: an external_fence_fn version token over active leases,
captured inside the transaction before the authoritative read and re-compared
before every guarded UPDATE (movement aborts the whole transaction; an
unreadable lease store raises rather than comparing equal), and a liveness_fn
re-probe that must affirmatively re-establish that this exact process is gone,
comparing process_start_time so a reused pid is refused. Omitting the re-probe
retires nothing rather than proceeding unfenced. The guarded UPDATE also
asserts the fencing triple is unchanged, and all three columns participate in
the CAS token.

Preserved from the accepted work: registry_fingerprint still excludes
observation time and is order- and numeric-typing stable, plan still mutates
nothing, drift still retires zero, retired rows stay historical, and ordinary
author operations remain ungated by fleet state.

Tests: tests/test_issue_980_stale_worker_retirement.py 87 passed, 46 subtests
(was 40 passed, 23 subtests), covering all three blockers' required
regressions. Adjacent suites (#980, #978, #975, #948, task-capability role
invariants) 242 passed, 156 subtests. Full suite from the branch worktree
28 failed, 6253 passed, 6 skipped, 1152 subtests; master baseline at
108cbfa173 from branches/baseline-980-108cbfa 28 failed, 6166 passed, 6
skipped, 1106 subtests. The failing sets are identical under comm, so zero
regressions and zero masked pre-existing failures; the +87 passing delta is
this branch's tests.

No live worker-registry row was retired — every test uses a throwaway SQLite
database. BAA, issue #981, PR #906, issue #650, review 624, unrelated branches
and worktrees, profiles, configuration, credentials, sessions, and running
processes were not touched.

Refs #980
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-30 18:01:20 -04:00
sysadminandClaude Opus 4.8 c0c6d14b73 feat(fleet): add CAS-protected stale worker retirement capability
Adds a sanctioned controller/reconciler capability that retires conclusively
stale MCP worker-registry rows through a dry-run-first, compare-and-swap
protected workflow (#980).

The #978 fleet snapshot is read-only, and its registry_revision is seeded with
snapshot_at at second precision, so a CAS gated on it can never pass. That
token is left unchanged; retirement gets its own stable registry_fingerprint
derived exclusively from canonical retirement-relevant registry content, plus a
candidate_fingerprint pinning the approved set. Identical contents observed at
different times produce the same token; row order never affects it; any
retirement-relevant change moves it.

WorkerRegistry.retire_stale_workers performs the whole decision inside one
BEGIN IMMEDIATE transaction: re-read rows, recompute the fingerprint from those
rows, recompute the eligibility plan from those rows (re-reading active workflow
leases), compare the candidate fingerprint, revalidate every target, then retire
each survivor with a guarded UPDATE asserting its status, heartbeat, generation,
session, fencing epoch, and pid are unchanged. Drift retires zero workers and
reports registry_revision_moved or candidate_set_moved; any exception rolls back
and reports transaction_failed, so a partial write is never reported as success.

Retirement requires a full conjunction: active status, complete registry fields,
parsable heartbeat, not live, pid_alive false, expired heartbeat, stale
ownership, canonical repository binding, no identity evidence shared with a live
or unprobeable worker, and no active workflow-lease ownership. Everything else
is preserved with a structured reason code. Retired rows become historical
rather than stale and keep their history; nothing is deleted.

Retirement does not repair untrusted live identity: live workers on legacy
instance identities are preserved and keep their legacy_incomplete_identity
blockers, so the result never claims the fleet became safe.

Exposed as gitea_plan_stale_worker_retirement and
gitea_apply_stale_worker_retirement, restricted to controller/reconciler role
kinds. The Gitea operation gate stays gitea.read because the mutation lands in
the local control-plane registry, matching the #601 lease lifecycle; no new
Gitea write permission is introduced and no author permission is broadened.

Tests: tests/test_issue_980_stale_worker_retirement.py (40 passed, 23 subtests),
including the regression test proving the old snapshot_at derivation moved the
token one second apart while the new registry CAS token does not. Full suite
from the branch worktree matches the master baseline exactly: 28 failed / 6206
passed vs 28 failed / 6166 passed, identical failure set.

Closes #980

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-30 15:22:30 -04:00
20 changed files with 4167 additions and 2588 deletions
-5
View File
@@ -355,10 +355,6 @@ def assess_anti_stomp_preflight(
root_head_sha: str | None = None, root_head_sha: str | None = None,
root_porcelain: str | None = None, root_porcelain: str | None = None,
remote_master_sha: str | None = None, remote_master_sha: str | None = None,
# #983: the tracking integration ref the SHA above came from, so root-checkout
# contamination names the ref actually compared instead of a hardcoded
# 'prgs/master'. None preserves the previous generic wording.
remote_master_ref: str | None = None,
check_root_checkout: bool = True, check_root_checkout: bool = True,
check_worktree: bool = True, check_worktree: bool = True,
create_issue_bootstrap_assessment: dict[str, Any] | None = None, create_issue_bootstrap_assessment: dict[str, Any] | None = None,
@@ -557,7 +553,6 @@ def assess_anti_stomp_preflight(
remote_master_sha=remote_master_sha, remote_master_sha=remote_master_sha,
resolved_role=req_role or role, resolved_role=req_role or role,
actual_role=role, actual_role=role,
remote_master_ref=remote_master_ref,
) )
checks["root_checkout"] = { checks["root_checkout"] = {
"block": bool(root_assessment.get("block")), "block": bool(root_assessment.get("block")),
+14 -456
View File
@@ -37,34 +37,6 @@ CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT"
# Candidate git remote names probed when deriving repository identity. # Candidate git remote names probed when deriving repository identity.
_IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps") _IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps")
# Fallback integration-branch names, probed only when the checkout declares no
# configured upstream (#983). Mirrors the stable base branches recognised
# elsewhere in the workflow (``stacked_pr_support``, ``root_checkout_guard``).
# The fallback is deliberately *not* ordered-first-wins: when more than one of
# these refs exists and the checkout records no upstream, the integration branch
# is genuinely ambiguous and resolution fails closed instead of guessing.
INTEGRATION_BRANCH_CANDIDATES: tuple[str, ...] = ("master", "main", "dev")
# Sources for a *proven* base-ref derivation (#983 B1).
#
# ``refs/remotes/<remote>/HEAD`` is deliberately absent from this list. It is a
# local symbolic-ref *cache* written once at clone time and refreshed only by an
# explicit ``git remote set-head``; an ordinary fetch never updates it. When the
# upstream default branch changes afterwards the cache keeps naming the old
# branch, so trusting it derives the wrong integration branch for a checkout
# that is sitting exactly on its tip. The cache is still read, but only as a
# corroborating observation reported back to the caller — never as an authority,
# and never as a tie-breaker between otherwise ambiguous candidates.
BASE_REF_SOURCE_CONFIGURED_UPSTREAM = "configured_branch_upstream"
BASE_REF_SOURCE_UNIQUE_CANDIDATE = "unique_integration_branch_ref"
# Reason codes for base-ref derivation outcomes (#983), so callers and tests can
# assert the refusal cause instead of string-matching prose.
DENY_NO_IDENTITY_REMOTE = "no_identity_remote"
DENY_AMBIGUOUS_REMOTE = "ambiguous_identity_remote"
DENY_AMBIGUOUS_BASE_BRANCH = "ambiguous_integration_branch"
DENY_NO_BASE_BRANCH = "no_integration_branch_ref"
# Repository-authority modes (#973 B10). Exactly two values are supported. # Repository-authority modes (#973 B10). Exactly two values are supported.
# ``mode`` selects how repository authority is established, so an unrecognised # ``mode`` selects how repository authority is established, so an unrecognised
# value must never be normalised onto one of these: aliasing a trusted mode is # value must never be normalised onto one of these: aliasing a trusted mode is
@@ -148,15 +120,17 @@ def resolve_repo_toplevel(path: str) -> str | None:
return os.path.realpath(top) if top else None return os.path.realpath(top) if top else None
def _identity_remote_candidates(path: str, remote: str | None) -> list[str]: def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
"""Ordered remote names to probe for identity at *path*. """``owner/repository`` derived from a git remote configured at *path*.
Names are used verbatim — never case-folded. Git config subsection names are Tries the caller-named remote first, then a small set of known remote names,
case-sensitive, so a repository whose remote is ``MDCPS`` is reached only by then whatever remote the repository actually has. Returns None when no remote
the exact string ``MDCPS``; the lowercase entry in URL is parseable (identity cannot be proven).
:data:`_IDENTITY_REMOTE_CANDIDATES` simply does not resolve, and the exact
name arrives from ``git remote`` below (#983).
""" """
text = (path or "").strip()
if not text:
return None
ordered: list[str] = [] ordered: list[str] = []
for name in (remote, *_IDENTITY_REMOTE_CANDIDATES): for name in (remote, *_IDENTITY_REMOTE_CANDIDATES):
clean = (name or "").strip() clean = (name or "").strip()
@@ -165,7 +139,7 @@ def _identity_remote_candidates(path: str, remote: str | None) -> list[str]:
try: try:
listed = subprocess.run( listed = subprocess.run(
["git", "-C", path, "remote"], ["git", "-C", text, "remote"],
capture_output=True, capture_output=True,
text=True, text=True,
check=True, check=True,
@@ -175,20 +149,11 @@ def _identity_remote_candidates(path: str, remote: str | None) -> list[str]:
for name in listed: for name in listed:
if name and name not in ordered: if name and name not in ordered:
ordered.append(name) ordered.append(name)
return ordered
for name in ordered:
def _configured_remote_identities(path: str, remote: str | None) -> list[tuple[str, str]]:
"""``(remote_name, owner/repository)`` for every probe name that resolves.
Remote names are returned exactly as configured so downstream tracking refs
(``refs/remotes/<remote>/<branch>``) address the real ref (#983).
"""
found: list[tuple[str, str]] = []
for name in _identity_remote_candidates(path, remote):
try: try:
url = subprocess.run( url = subprocess.run(
["git", "-C", path, "remote", "get-url", name], ["git", "-C", text, "remote", "get-url", name],
capture_output=True, capture_output=True,
text=True, text=True,
check=True, check=True,
@@ -197,415 +162,8 @@ def _configured_remote_identities(path: str, remote: str | None) -> list[tuple[s
continue continue
parsed = remote_repo_guard.parse_org_repo_from_remote_url(url) parsed = remote_repo_guard.parse_org_repo_from_remote_url(url)
if parsed: if parsed:
found.append((name, f"{parsed[0]}/{parsed[1]}")) return f"{parsed[0]}/{parsed[1]}"
return found return None
def _configured_branch_upstream(path: str) -> tuple[str | None, str | None]:
"""``(remote_name, branch)`` the checked-out branch is configured to track.
Reads ``branch.<current>.remote`` and ``branch.<current>.merge`` — the
checkout's own explicitly configured integration target, equivalent to
``@{upstream}``. Unlike ``refs/remotes/<remote>/HEAD`` this is not a
clone-time cache: it is written when the branch is set up to track an
upstream and rewritten whenever that tracking changes, so it states what the
checkout actually integrates onto today (#983 B1).
Returns ``(None, None)`` on a detached HEAD or an untracked branch. Names are
returned verbatim; git remote names are case-sensitive.
"""
text = (path or "").strip()
if not text:
return None, None
branch = _git_read(text, "symbolic-ref", "--quiet", "--short", "HEAD")
if not branch:
return None, None
remote = _git_read(text, "config", "--get", f"branch.{branch}.remote")
merge = _git_read(text, "config", "--get", f"branch.{branch}.merge")
if not remote or not merge:
return None, None
prefix = "refs/heads/"
upstream_branch = merge[len(prefix):].strip() if merge.startswith(prefix) else merge.strip()
if not upstream_branch:
return None, None
return remote, upstream_branch
def _cached_remote_head_branch(path: str, remote: str) -> str | None:
"""Branch named by the *cached* ``refs/remotes/<remote>/HEAD`` symref.
Read for observability only. This value is never authoritative (see
:data:`BASE_REF_SOURCE_CONFIGURED_UPSTREAM`); it is surfaced so an operator
can see that the local cache disagrees with the configured upstream, and so
a refusal can name the misleading signal explicitly.
"""
prefix = f"refs/remotes/{remote}/"
symref = _git_read(path, "symbolic-ref", "--quiet", f"{prefix}HEAD")
if not symref or not symref.startswith(prefix):
return None
branch = symref[len(prefix):].strip()
return branch or None
def assess_identity_remote(
path: str, *, explicit_remote: str | None = None
) -> dict:
"""Resolve the identity remote, failing closed when the target is ambiguous.
This is the ambiguity-aware counterpart to :func:`resolve_identity_remote`,
and the reason gating and reporting can no longer disagree (#983 B2).
*explicit_remote* is a **caller-supplied disambiguation** and nothing else.
It must come from an operator or an explicitly sanctioned repository
context; a value this module inferred while probing must never be handed
back in through it, because doing so re-labels an internal first-wins guess
as deliberate caller intent and silently suppresses the ambiguity gate.
Decision table:
* No remote yields a parseable identity -> :data:`DENY_NO_IDENTITY_REMOTE`.
* *explicit_remote* names one of the resolving remotes -> that remote is
authoritative, ``explicit`` True.
* Otherwise, when every resolving remote claims the **same** repository the
target is unambiguous and is accepted, ``explicit`` False. The remote
named by the checkout's configured upstream is preferred among equals so
the choice is deterministic rather than probe-order dependent.
* Otherwise distinct remotes claim different repositories and there is no
sanctioned disambiguation -> :data:`DENY_AMBIGUOUS_REMOTE`.
Returns a dict with ``remote``, ``slug``, ``identities``, ``ambiguous``,
``explicit``, ``reason_code`` and ``reasons``. ``remote``/``slug`` are None
on any refusal, so a caller cannot report a repository the gate refuses.
"""
text = (path or "").strip()
result: dict = {
"remote": None,
"slug": None,
"identities": [],
"ambiguous": False,
"explicit": False,
"reason_code": None,
"reasons": [],
}
if not text:
result["reason_code"] = DENY_NO_IDENTITY_REMOTE
result["reasons"].append(
"no repository path supplied for identity-remote resolution (fail closed)"
)
return result
named = (explicit_remote or "").strip() or None
identities = _configured_remote_identities(text, named)
result["identities"] = list(identities)
if not identities:
result["reason_code"] = DENY_NO_IDENTITY_REMOTE
result["reasons"].append(
f"no git remote at '{text}' yields a parseable repository identity "
"(fail closed)"
)
return result
if named:
for name, slug in identities:
if name == named:
result["remote"] = name
result["slug"] = slug
result["explicit"] = True
return result
distinct = {slug for _, slug in identities}
if len(distinct) > 1:
listed = ", ".join(f"{name} -> {slug}" for name, slug in identities)
result["ambiguous"] = True
result["reason_code"] = DENY_AMBIGUOUS_REMOTE
detail = (
f"explicitly named remote '{named}' does not resolve a repository identity "
"there, so it cannot disambiguate; "
if named
else ""
)
result["reasons"].append(
f"ambiguous repository identity at '{text}': {detail}remotes resolve to "
f"different repositories ({listed}); no single authoritative target can "
"be established (fail closed)"
)
return result
# One repository, possibly reachable through several remote names (a mirror).
# Prefer the remote the checkout is actually configured to track so the
# choice is deterministic instead of probe-order dependent.
upstream_remote, _ = _configured_branch_upstream(text)
chosen = identities[0]
if upstream_remote:
for entry in identities:
if entry[0] == upstream_remote:
chosen = entry
break
result["remote"], result["slug"] = chosen
return result
def resolve_identity_remote(
path: str, *, remote: str | None = None
) -> tuple[str | None, str | None]:
"""``(remote_name, owner/repository)`` for the remote that proves identity.
The remote *name* is the piece historically thrown away by
:func:`repository_identity_slug`, even though resolving the slug already
required discovering it. Cross-repository base-ref derivation needs that
name to build ``refs/remotes/<remote>/<branch>``, so it is now returned
rather than discarded (#983). Returns ``(None, None)`` when no remote URL is
parseable (identity cannot be proven).
First-wins by design: this is the identity lookup behind
:func:`repository_identity_slug` and the #706/#973 canonical-root
validation, which compare an observed slug against an independently trusted
expected slug and therefore do not need an ambiguity verdict. Callers that
*derive* a target rather than validate one — the mutation guard and the
parity report — must use :func:`assess_identity_remote`, which fails closed
on ambiguity (#983 B2).
"""
text = (path or "").strip()
if not text:
return None, None
for name, slug in _configured_remote_identities(text, remote):
return name, slug
return None, None
def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
"""``owner/repository`` derived from a git remote configured at *path*.
Tries the caller-named remote first, then a small set of known remote names,
then whatever remote the repository actually has. Returns None when no remote
URL is parseable (identity cannot be proven).
"""
return resolve_identity_remote(path, remote=remote)[1]
def _base_ref_result(
*,
proven: bool,
reasons: list[str],
remote: str | None = None,
branch: str | None = None,
repository_slug: str | None = None,
source: str | None = None,
reason_code: str | None = None,
identity_explicit: bool = False,
configured_upstream_remote: str | None = None,
configured_upstream_branch: str | None = None,
cached_remote_head_branch: str | None = None,
) -> dict:
"""Build the base-ref derivation payload.
``tracking_refs`` is the ordered probe tuple downstream guards hand to
``git rev-parse``: the ``<remote>/<branch>`` shorthand first, then the fully
qualified ``refs/remotes/<remote>/<branch>``. It is empty whenever the
derivation is not ``proven``, so an unresolved target can never be probed
against some other repository's ref.
``cached_remote_head_branch`` reports what the local
``refs/remotes/<remote>/HEAD`` cache claims, and
``cached_remote_head_conflicts`` whether that claim disagrees with the branch
actually derived. Both are observability only: the cache never decides the
outcome (#983 B1).
"""
tracking_ref = f"refs/remotes/{remote}/{branch}" if proven and remote and branch else None
tracking_refs: tuple[str, ...] = (
(f"{remote}/{branch}", tracking_ref) if tracking_ref else ()
)
return {
"proven": proven,
"block": not proven,
"remote": remote,
"branch": branch,
"repository_slug": repository_slug,
"tracking_ref": tracking_ref,
"tracking_refs": tracking_refs,
"source": source,
"reason_code": reason_code,
"identity_explicit": identity_explicit,
"configured_upstream_remote": configured_upstream_remote,
"configured_upstream_branch": configured_upstream_branch,
"cached_remote_head_branch": cached_remote_head_branch,
"cached_remote_head_conflicts": bool(
cached_remote_head_branch and branch and cached_remote_head_branch != branch
),
"reasons": list(reasons),
}
def _git_read(path: str, *args: str) -> str | None:
"""Run a read-only git command in *path*; ``None`` on any failure."""
try:
res = subprocess.run(
["git", "-C", path, *args],
capture_output=True,
text=True,
check=False,
)
except Exception:
return None
if res.returncode != 0:
return None
return (res.stdout or "").strip() or None
def _ref_exists(path: str, ref: str) -> bool:
"""Whether *ref* resolves in the checkout at *path*."""
try:
res = subprocess.run(
["git", "-C", path, "rev-parse", "--verify", "--quiet", ref],
capture_output=True,
text=True,
check=False,
)
except Exception:
return False
return res.returncode == 0 and bool((res.stdout or "").strip())
def resolve_target_base_ref(path: str, *, explicit_remote: str | None = None) -> dict:
"""Derive the authoritative integration base ref for the checkout at *path*.
This is the single resolved target shared by cross-repository mutation
gating and parity reporting, so the two can never disagree about which
commit a checkout is supposed to match (#983).
*explicit_remote* is a caller-supplied disambiguation only; see
:func:`assess_identity_remote` for why an internally inferred remote must
never be passed back in here (#983 B2).
Resolution order:
1. **Identity remote** — via :func:`assess_identity_remote`, with its exact
configured case preserved (``MDCPS`` stays ``MDCPS``). Ambiguous identity
fails closed here rather than resolving to whichever remote probed first.
2. **The checkout's configured upstream** — ``branch.<current>.remote`` plus
``branch.<current>.merge``, accepted when it names the identity remote
and its remote-tracking ref actually exists. This is the checkout's own
declaration of what it integrates onto, and unlike the remote-HEAD cache
it is rewritten whenever that tracking changes.
3. **Fallback** — only when the checkout declares no usable upstream,
exactly one of :data:`INTEGRATION_BRANCH_CANDIDATES` present as a
remote-tracking ref.
``refs/remotes/<remote>/HEAD`` is **not** a step. It is read for reporting
(``cached_remote_head_branch`` / ``cached_remote_head_conflicts``) and never
decides the branch, because it is a clone-time cache that an ordinary fetch
does not refresh: a checkout whose upstream default moved on still has the
old branch cached, and trusting it gates that checkout against a ref it does
not integrate onto (#983 B1).
Fails closed — ``proven`` False, empty ``tracking_refs``, and a
``reason_code`` — when identity is unprovable, when distinct remotes claim
different repositories, when several candidate branches exist with no
configured upstream, or when no candidate exists at all. Nothing here
invents a branch, writes a ref, runs a fetch, or falls back to another
repository's base. Every ``proven`` result names a tracking ref that
resolves in this checkout.
"""
text = (path or "").strip()
if not text:
return _base_ref_result(
proven=False,
reasons=["no repository path supplied for base-ref derivation (fail closed)"],
reason_code=DENY_NO_IDENTITY_REMOTE,
)
identity = assess_identity_remote(text, explicit_remote=explicit_remote)
remote_name, slug = identity["remote"], identity["slug"]
if not remote_name:
return _base_ref_result(
proven=False,
reasons=list(identity["reasons"]),
reason_code=identity["reason_code"],
identity_explicit=bool(identity["explicit"]),
)
prefix = f"refs/remotes/{remote_name}/"
cached = _cached_remote_head_branch(text, remote_name)
upstream_remote, upstream_branch = _configured_branch_upstream(text)
common = {
"repository_slug": slug,
"identity_explicit": bool(identity["explicit"]),
"configured_upstream_remote": upstream_remote,
"configured_upstream_branch": upstream_branch,
"cached_remote_head_branch": cached,
}
# 2. The checkout's configured upstream, when it belongs to the identity
# remote and its tracking ref is actually present. Requiring the ref to
# exist keeps 'proven' honest: a proven target is always resolvable.
if (
upstream_remote == remote_name
and upstream_branch
and _ref_exists(text, f"{prefix}{upstream_branch}")
):
return _base_ref_result(
proven=True,
reasons=[],
remote=remote_name,
branch=upstream_branch,
source=BASE_REF_SOURCE_CONFIGURED_UPSTREAM,
**common,
)
# 3. Exactly one known integration branch present as a remote-tracking ref.
present = [
candidate
for candidate in INTEGRATION_BRANCH_CANDIDATES
if _ref_exists(text, f"{prefix}{candidate}")
]
if len(present) == 1:
return _base_ref_result(
proven=True,
reasons=[],
remote=remote_name,
branch=present[0],
source=BASE_REF_SOURCE_UNIQUE_CANDIDATE,
**common,
)
# The cached remote HEAD is named in the refusal so the operator can see the
# signal that looks authoritative but is not, and is told the read-only fix.
cache_note = (
f" the cached '{prefix}HEAD' names '{cached}', but that cache is written at "
"clone time and is not refreshed by fetch, so it cannot break the tie;"
if cached
else ""
)
remedy = (
f" Configure the checkout's upstream (git branch --set-upstream-to={remote_name}/"
"<branch>) so the integration target is declared rather than guessed."
)
if len(present) > 1:
return _base_ref_result(
proven=False,
reasons=[
f"the checkout at '{text}' declares no upstream on remote "
f"'{remote_name}' and several integration branches exist "
f"({', '.join(present)});{cache_note} the integration base ref is "
f"ambiguous (fail closed).{remedy}"
],
reason_code=DENY_AMBIGUOUS_BASE_BRANCH,
**common,
)
return _base_ref_result(
proven=False,
reasons=[
f"the checkout at '{text}' declares no upstream on remote '{remote_name}' "
f"and none of {'/'.join(INTEGRATION_BRANCH_CANDIDATES)} exists as a "
f"remote-tracking ref under '{prefix}';{cache_note} the integration base "
f"ref cannot be derived (fail closed; no fetch is performed here).{remedy}"
],
reason_code=DENY_NO_BASE_BRANCH,
**common,
)
def assess_canonical_repository_root( def assess_canonical_repository_root(
+7 -90
View File
@@ -34,10 +34,8 @@ namespaces.
1. Multiple active instances **may** share the same `client_type`. 1. Multiple active instances **may** share the same `client_type`.
2. Every application launch receives a **distinct** `client_instance_id`. 2. Every application launch receives a **distinct** `client_instance_id`.
3. All namespace workers of one launch report the **same** 3. All five namespace workers of one launch report the **same**
`client_instance_id`. For a whole-fleet launch that is five workers; for a `client_instance_id`.
project-scoped launch it is exactly the namespaces that launch started
(#985).
4. Each namespace worker has a **distinct** `worker_identity`, process 4. Each namespace worker has a **distinct** `worker_identity`, process
identity, generation, and PID. identity, generation, and PID.
5. Instance identity is **never** inferred from PID proximity, timestamps, or 5. Instance identity is **never** inferred from PID proximity, timestamps, or
@@ -48,92 +46,6 @@ namespaces.
This deliberately **replaces** any permanent `exactly_one_per_profile` fleet This deliberately **replaces** any permanent `exactly_one_per_profile` fleet
model (#949 assumption) as the operating rule for multi-instance fleets. model (#949 assumption) as the operating rule for multi-instance fleets.
## Project-scoped launches and the runnable entry point (#985)
Not every project exposes all five namespaces. A project-scoped configuration
such as Weekly Briefings intentionally has **author, reviewer, and merger
only**, and has no controller or reconciler profile to supply. Requiring all
five would force operators to invent profiles that must not exist, so the
launcher accepts an explicitly validated subset.
### Running a launch
```bash
python3 -m mcp_application_launcher \
--client claude_code \
--namespaces author,reviewer,merger \
--profile author=prgs-author \
--profile reviewer=prgs-reviewer \
--profile merger=prgs-merger
```
That mints one trusted `client_instance_id`, writes a per-launch `mcpServers`
config, and executes:
```text
claude --mcp-config <per-launch.json> --strict-mcp-config
```
Add `--dry-run` to print the plan (namespaces, excluded namespaces, minted ID,
argv, config path) as JSON and start nothing. Omit `--namespaces` to launch all
five exactly as before. Arguments after the flags are forwarded to the client.
Other supported clients register one entry in
`mcp_application_launcher.CLIENT_LAUNCH_SPECS`; because the argv builder only
ever receives a config this module wrote, every client necessarily goes through
the same mint-once/propagate-to-all mechanism.
### Why the config is per-launch and never `.mcp.json`
Persisting a trusted ID into a shared, reused `.mcp.json` would give two
concurrent sessions **the same** trusted identity — precisely the reuse case
the duplicate gate must reject. The launcher therefore writes a fresh
owner-readable-only (`0600`) config per launch. Do not commit one, and do not
copy a minted `GITEA_MCP_CLIENT_INSTANCE` into any checked-in configuration.
### Provenance sealing
The `inst-…` format is public and reproducible, so format alone cannot
establish trust: anyone able to set one environment variable could otherwise
hand-write a valid-looking ID and be believed. Trust therefore requires **both**
the format and `GITEA_MCP_INSTANCE_PROVENANCE=trusted_launcher`, which only the
launcher writes. A well-formed but unsealed identity is classified
`unsealed_launcher` and **fails closed** — it is still reported for diagnosis,
but never authorizes trusted attribution. Manually asserted trust is not
possible.
### Migration and coordinated relaunch
Existing configurations that predate this work set no instance key at all, so
their workers register under `legacy-pid-*` (`legacy_incomplete`) and remain
mutually indistinguishable when two launches share a profile.
Migrating is a **coordinated relaunch**, not an in-place edit — a running
worker cannot acquire an identity it was not started with:
1. Stop every LLM client currently running Gitea MCP workers. A single
surviving legacy cohort keeps the fleet ambiguous.
2. Relaunch each application through the command above.
3. Repeat per application. Concurrent launches are expected and safe: each
receives its own trusted ID.
### Post-launch verification
* `gitea_get_runtime_context``provenance_assessment.attachment` should show
an `inst-…` `client_instance_id` with
`instance_id_provenance: trusted_launcher`, not `legacy-pid-*` /
`legacy_incomplete`.
* Every namespace of the same launch must report that **same** ID; two
different launches must report different IDs.
* `gitea_resolve_task_capability` should return
`exact_safe_next_action: "None; ready for operations."` with no
`blocker_kind`. A `runtime_reconnect_required` naming duplicate PIDs per
profile means at least one cohort is still on legacy identity, or a second
cohort is genuinely running.
* `mcp_application_launcher.collect_instance_ids_from_mcp_servers(servers,
namespaces=[...])` returns `shared_single_trusted_id` for a built config, and
reports `missing_servers` when an expected namespace entry is absent.
## How five workers join one instance ## How five workers join one instance
1. The host starts one application instance (for example one Codex session). 1. The host starts one application instance (for example one Codex session).
@@ -277,6 +189,11 @@ mutation. Diagnostic reads remain available where `gitea.read` allows.
* `mcp_fleet_snapshot` — pure snapshot + classification (#978). * `mcp_fleet_snapshot` — pure snapshot + classification (#978).
* `gitea_snapshot_instance_fleet` — sanctioned MCP tool (#978). * `gitea_snapshot_instance_fleet` — sanctioned MCP tool (#978).
* `gitea_get_runtime_context` — single-process view (not fleet-wide). * `gitea_get_runtime_context` — single-process view (not fleet-wide).
* [stale-worker-retirement.md](stale-worker-retirement.md) — CAS-protected
retirement of conclusively stale registrations (#980). Note that the
`registry_revision` this snapshot returns is **time-seeded** and unsuitable
for compare-and-swap; retirement derives its own stable
`registry_fingerprint` from registry content alone.
## Non-goals ## Non-goals
+2
View File
@@ -51,6 +51,7 @@ that gates each call, not which tools exist.
- `gitea_adopt_merger_pr_lease` - `gitea_adopt_merger_pr_lease`
- `gitea_adopt_workflow_lease` - `gitea_adopt_workflow_lease`
- `gitea_allocate_next_work` - `gitea_allocate_next_work`
- `gitea_apply_stale_worker_retirement`
- `gitea_assess_already_landed_reconciliation` - `gitea_assess_already_landed_reconciliation`
- `gitea_assess_conflict_fix_classification` - `gitea_assess_conflict_fix_classification`
- `gitea_assess_conflict_fix_push` - `gitea_assess_conflict_fix_push`
@@ -124,6 +125,7 @@ that gates each call, not which tools exist.
- `gitea_observability_link_issue` - `gitea_observability_link_issue`
- `gitea_observability_list_projects` - `gitea_observability_list_projects`
- `gitea_observability_reconcile_incident` - `gitea_observability_reconcile_incident`
- `gitea_plan_stale_worker_retirement`
- `gitea_post_heartbeat` - `gitea_post_heartbeat`
- `gitea_publish_unpublished_issue_branch` - `gitea_publish_unpublished_issue_branch`
- `gitea_quarantine_contaminated_review` - `gitea_quarantine_contaminated_review`
+282
View File
@@ -0,0 +1,282 @@
# Stale worker retirement (#980)
The #978 instance-fleet snapshot made registry accuracy observable but
deliberately read-only: a registry full of rows whose owning processes are long
gone stays full. This document describes the sanctioned way to retire those
rows — a dry-run-first, compare-and-swap-protected workflow available only to
controller and reconciler namespaces.
Related: [instance-fleet-identity.md](instance-fleet-identity.md) (#978),
[post-restart-reconcile.md](post-restart-reconcile.md) (#662).
## Why a dedicated registry token
`mcp_fleet_snapshot._consistency_token` seeds its digest with `snapshot_at`,
formatted at second precision. Its output — surfaced as `registry_revision` and
`consistency_token` on the snapshot — therefore changes on **every call**, even
when no registry row changed. Any compare-and-swap gated on it can never pass:
a dry-run/apply cycle spanning more than one second aborts unconditionally.
That token remains useful as an observation stamp and is unchanged. #980 adds a
separate, *stable* token instead:
| Token | Module | Derived from | Stable across time? |
| --- | --- | --- | --- |
| `registry_revision` / `consistency_token` | `mcp_fleet_snapshot` | `snapshot_at` + a subset of row fields | **No** — moves every second |
| `registry_fingerprint` | `mcp_fleet_retirement` | canonical retirement-relevant row content only | **Yes** |
| `candidate_fingerprint` | `mcp_fleet_retirement` | canonical content of the selected candidate rows | **Yes** |
`registry_fingerprint` guarantees:
* identical canonical registry contents always produce the same token, whenever
they are observed;
* row iteration order never affects the token (serialized rows are sorted);
* any retirement-relevant change moves it — row creation or deletion, identity
change, heartbeat or TTL change, ownership change, registration-state change,
PID change, or repository-binding change.
The exact field set is `mcp_fleet_retirement.FINGERPRINT_FIELDS`. Deliberately
excluded: `token_fingerprint` (credential-adjacent, never a retirement input),
the four `*_revision` columns (revision drift is an independent restart concern
and is not part of the eligibility conjunction), and the `retired_*` bookkeeping
columns this feature adds. Numeric values are canonicalised, so a TTL that
round-trips through SQLite as `900.0` hashes identically to `900`.
## Eligibility — the conjunction
A registration is retired only when **every** one of these holds. Any missing
or contradictory evidence preserves the row.
| Requirement | Preserve reason code when it fails |
| --- | --- |
| `status` is `active` | `already_terminal_registration` |
| Every field the conjunction reads is present (`REQUIRED_IDENTITY_FIELDS`) | `incomplete_registry_identity` |
| `last_heartbeat_at` parses as a UTC stamp | `unparsable_heartbeat` |
| Worker is not live | `worker_live` |
| PID probe returns a definite answer | `pid_liveness_unknown` |
| PID probe says the process is gone | `pid_alive` |
| Heartbeat has expired under the canonical TTL | `heartbeat_not_expired` |
| `ownership_state` is exactly `stale` | `ambiguous_ownership_state` |
| Repository binding present and canonical | `repository_binding_ambiguous` |
| No identity evidence shared with a live or unprobeable worker | `conflicting_identity_evidence` |
| No other active row claims the same (instance, namespace) while one may be live | `client_instance_conflict` |
| Not an active workflow-lease owner | `protected_active_workflow_owner` |
| Instance identity is launcher-minted (`inst-…`) | `untrusted_identity_provenance` |
| Row's `host_id` matches the host running retirement | `host_binding_unproven` |
| Boot identity is known on both sides | `boot_identity_unknown` |
| The pid number is not occupied by a different incarnation | `pid_reuse_detected` |
Eligible rows carry `eligible_stale_orphan`.
Three properties are worth stating explicitly:
* **`pid_alive` can only withdraw liveness, never grant it**
(`WorkerRegistry.is_live`, #948 AC7). A heartbeat-lapsed but still-running
process therefore classifies as `stale` in the snapshot, yet #980's added
`pid_alive is False` requirement preserves it. An unprobeable PID (`None`)
also fails closed.
* **Affirmative identity proof is required (review 657 B2).** An earlier
revision required only that the pre-existing registry columns were non-null —
which a legacy `legacy-pid-…` row satisfies trivially, so a row that proved
nothing about *which* process it described was retireable. Retirement now
needs both halves of a positive proof:
* **Attribution** — a launcher-minted `inst-…` `client_instance_id`, so the
row is known to belong to one specific application launch rather than
having been inferred from pid proximity.
* **Fencing** — `host_id`, `boot_id`, and `process_start_time`, which turn a
bare pid into a statement about one process: which machine it ran on, which
boot of that machine, and which incarnation of that pid number.
**Consequence, stated plainly:** registrations written before these columns
existed, and any row on a legacy instance identity, are preserved
*permanently*. They are retired only after their worker re-registers under a
trusted identity — never on weaker evidence. That the alternative would leave
legacy rows outstanding indefinitely is not a reason to relax the proof.
* **A live pid is an absolute block.** Even across a boot boundary, where the
number provably cannot belong to the registered process, an occupied pid
preserves the row rather than being argued away by the fencing proof.
Multiple processes belonging to one legitimate worker cohort are not treated as
multiple independent workers: the fleet model from #948/#978 is preserved
unchanged, and sharing a role or profile is never a duplicate.
## Tools
### `gitea_plan_stale_worker_retirement`
Read-only. Controller and reconciler only.
| Parameter | Meaning |
| --- | --- |
| `remote` | `dadeschools` or `prgs` |
| `host`, `org`, `repo` | Optional overrides (audit context) |
| `canonical_repository` | Expected repository binding; defaults to the process root |
Returns `registry_fingerprint`, `candidate_fingerprint`,
`candidate_worker_identities`, per-worker `candidates` and `preserved` entries
(each with `reason_code`, `detail`, and structured `evidence`),
`preserved_reason_counts`, `assessed_count`, `candidate_count`,
`preserved_count`, and `protected_active_workflow_owners`.
`mutation_performed` is always `false` and `read_only` is always `true`.
Planning is deterministic: the same authoritative registry contents produce the
same plan and the same tokens regardless of when they are observed.
### `gitea_apply_stale_worker_retirement`
Mutating. Controller and reconciler only.
| Parameter | Meaning |
| --- | --- |
| `registry_fingerprint` | The exact stable token the plan returned |
| `candidate_fingerprint` | The exact candidate-set token the plan returned |
| `worker_identities` | The exact candidate identities (list, or JSON / comma-separated string) |
| `remote`, `host`, `org`, `repo` | As above |
| `canonical_repository` | Must match the value the plan used |
Before touching the registry, apply fails closed on: profile permission, role
kind, master parity (`mutation_safe`), stable-runtime mode, capability
resolution refreshed immediately before mutation, worker-registry availability,
workflow-lease enumeration failure, and daemon-cohort uniqueness
(`classify_cohort`).
A matching token is necessary but never sufficient. Inside one
`BEGIN IMMEDIATE` transaction (`WorkerRegistry.retire_stale_workers`) the
server:
1. re-reads the authoritative rows;
2. recomputes `registry_fingerprint` from *those* rows and compares — a mismatch
returns `registry_revision_moved` with `retired_count: 0` and no write;
3. recomputes the eligibility plan from *those* rows — re-reading the active
workflow leases rather than reusing the set captured before the transaction
opened, so a lease acquired after planning still preserves its worker — and
compares `candidate_fingerprint`. A mismatch returns `candidate_set_moved`
with `retired_count: 0` and no write; a lease-enumeration failure raises and
rolls the transaction back;
4. revalidates every requested identity against that fresh plan;
5. retires each survivor with a guarded `UPDATE` that additionally asserts
`status`, `last_heartbeat_at`, `generation_id`, `session_id`,
`fencing_epoch`, and `pid` are unchanged. A guard that matches no row
preserves the worker with `row_changed_since_plan`.
There is no window between a safety check and its matching write, so a worker
that comes back to life, changes ownership, or is retired concurrently cannot be
removed on the strength of a stale observation. Any exception — including a
commit failure — rolls the whole transaction back and returns
`transaction_failed` with `success: false`, `retired_count: 0`, and
`mutation_performed: false`; a partial write can never be reported as success.
### Outcomes
| `outcome` | Meaning | `mutation_performed` |
| --- | --- | --- |
| `planned` | Dry-run result | `false` |
| `applied` | Transaction ran; see `retired` / `preserved` | `true` only if something was retired |
| `registry_revision_moved` | Registry changed between plan and apply | `false` |
| `candidate_set_moved` | Eligibility verdict changed between plan and apply | `false` |
| `already_retired` | Every requested row is already retired (idempotent replay) | `false` |
| `nothing_requested` | Empty target list | `false` |
| `transaction_failed` | Rolled back; nothing retired | `false` |
## What retirement does to the fleet snapshot
A retired row keeps its history: `status` moves to `retired` and `retired_at`,
`retired_by`, `retirement_reason` are recorded. Nothing is deleted. Because
`retired` is not `active`, the #978 snapshot counts the row as **historical**,
not stale, so `stale_worker_count` falls and historical rows never make the live
fleet unsafe by themselves.
**Retirement does not repair untrusted live identity.** Live workers registered
under legacy `pid-`/`proc-` instance identities are preserved untouched and
keep their `legacy_incomplete_identity` blockers. Retiring every stale row can
therefore legitimately produce:
* `stale_worker_count: 0`
* residual live `legacy_incomplete_identity` blockers
* `live_fleet_safe: false`
That is a truthful result, and the `post_apply` block reports the remaining
blockers rather than claiming the fleet became safe. Trusted
`client_instance_id` propagation through launchers is a separate enrollment
problem.
## Permissions
Plan and apply are authorized differently, and deliberately so (review 657 B1).
| | Plan | Apply |
| --- | --- | --- |
| Capability | `gitea.read` | `gitea.worker_registry.retire` |
| Nature | observational; opens no transaction, writes nothing | mutation |
| Roles | `controller`, `reconciler` | `controller`, `reconciler` |
An earlier revision authorized apply with `gitea.read` alone, reasoning that
the mutation lands in the local control-plane registry rather than in Gitea.
That the write is local makes it **no less a mutation**: sharing an
observational permission class with plan meant any profile that could *look*
could also *destroy*. Apply now requires its own capability.
* **Denied:** author, reviewer, merger, and every ordinary read-only profile —
they lack the capability, so they fail closed on the permission itself rather
than on the role check alone. The role restriction remains as defence in
depth: a profile mistakenly granted the capability still cannot reach apply
from an author, reviewer, or merger role.
* The capability is checked at entry **and** re-resolved immediately before the
registry mutation, so a profile change mid-call cannot be outrun.
* **No new Gitea write permission is introduced.**
`gitea.worker_registry.retire` authorizes exactly one local control-plane
transition (`worker_registrations.status -> retired`) and grants no branch,
issue, PR, review, merge, or restart authority. No author permission is
broadened.
* The fleet snapshot remains observational: nothing here turns it into a gate on
ordinary author work.
### Operator step
No profile holds `gitea.worker_registry.retire` by default, so apply is inert
until an operator adds it to the `allowed_operations` of the controller or
reconciler profile in `profiles.json`. Removing it again immediately and
completely revokes apply, while leaving plan and every other capability
untouched. That grant is a configuration change and is outside the scope of the
code that implements this feature.
## External-state fencing
`BEGIN IMMEDIATE` locks the worker registry and nothing else, so two inputs the
decision depends on sit outside the transaction's isolation domain: the
workflow-lease table in a separate control-plane database, and OS process
liveness. Re-reading them once during revalidation is not sufficient — the
per-target loop runs afterwards, so a lease acquired (or a pid revived) after
revalidation but before a given row's `UPDATE` would go unnoticed, and the
registry-column guard cannot catch it because no registry column changed.
Two mechanisms close that window, both applied per target immediately before
its own write:
* **`external_fence_fn`** — a version token over active leases
(`external_state_fingerprint`), captured inside the transaction *before* the
authoritative read and re-compared before every guarded `UPDATE`. Any movement
raises, rolling back the whole transaction: once the world has changed, every
remaining per-row decision was computed against a world that no longer exists.
An unreadable lease store raises rather than returning a token, because
"unreadable" must not silently compare equal to "unchanged".
* **`liveness_fn`** — a re-probe of process liveness and fencing identity that
must affirmatively re-establish that this exact process is gone. It compares
`process_start_time`, so a pid number reused since the plan is refused rather
than accepted.
A caller that supplies no `liveness_fn` retires nothing
(`liveness_reprobe_unavailable`) rather than proceeding unfenced. The guarded
`UPDATE` additionally asserts `host_id`, `boot_id`, and `process_start_time` are
unchanged, and all three participate in the CAS token, so fencing movement
alone is enough to abort.
## Non-goals
* Killing or restarting processes.
* Editing session files or configuration.
* Direct database cleanup outside the sanctioned transaction.
* Retiring live workers.
* Backfilling trusted identity for legacy workers.
* Rewriting worker ownership.
* Cleaning unrelated workflow-lease or issue-claim registries.
+2 -6
View File
@@ -1286,19 +1286,16 @@ def multi_namespace_launcher_entries(
profile_by_namespace, profile_by_namespace,
*, *,
client_type, client_type,
namespaces=None,
config_path=None, config_path=None,
client_instance_id=None, client_instance_id=None,
fleet_run_id=None, fleet_run_id=None,
session_id=None, session_id=None,
launch_nonce=None, launch_nonce=None,
): ):
"""Build production mcpServers for the namespaces of one application launch. """Build production mcpServers for all five namespaces of one application launch.
One shared trusted ``client_instance_id`` is minted (or reused) and One shared trusted ``client_instance_id`` is minted (or reused) and
propagated to every namespace worker env. *namespaces* selects a validated propagated to every namespace worker env. See
project-scoped subset (#985); omitting it launches all five sanctioned
namespaces as before. See
:func:`mcp_application_launcher.build_application_mcp_servers`. :func:`mcp_application_launcher.build_application_mcp_servers`.
""" """
import mcp_application_launcher as app_launcher import mcp_application_launcher as app_launcher
@@ -1306,7 +1303,6 @@ def multi_namespace_launcher_entries(
return app_launcher.build_application_mcp_servers( return app_launcher.build_application_mcp_servers(
profile_by_namespace, profile_by_namespace,
client_type=client_type, client_type=client_type,
namespaces=namespaces,
config_path=config_path or DEFAULT_CONFIG_PATH, config_path=config_path or DEFAULT_CONFIG_PATH,
client_instance_id=client_instance_id, client_instance_id=client_instance_id,
fleet_run_id=fleet_run_id, fleet_run_id=fleet_run_id,
+699 -27
View File
@@ -1052,16 +1052,9 @@ def _create_issue_bootstrap_assessment(
git_state = issue_lock_worktree.read_worktree_git_state(workspace) git_state = issue_lock_worktree.read_worktree_git_state(workspace)
remote_master_sha_error: str | None = None remote_master_sha_error: str | None = None
try: try:
# #983: consume the resolved-target state so an *underivable* base ref remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
# reaches the assessor as named missing evidence rather than a bare
# None, which the bootstrap would otherwise report only as "live
# master tip is unknown" with no cause.
_base_state = root_checkout_guard.resolve_remote_master_ref_state(
ctx["canonical_repo_root"] ctx["canonical_repo_root"]
) )
remote_master_sha = _base_state["sha"]
if not remote_master_sha and _base_state.get("reasons"):
remote_master_sha_error = "; ".join(_base_state["reasons"])
except Exception as exc: except Exception as exc:
remote_master_sha = None remote_master_sha = None
remote_master_sha_error = ( remote_master_sha_error = (
@@ -1089,14 +1082,9 @@ def _create_issue_bootstrap_assessment(
# closed instead of proceeding without base-equivalence proof. # closed instead of proceeding without base-equivalence proof.
remote_master_sha_error: str | None = None remote_master_sha_error: str | None = None
try: try:
# #983: same resolved-target consumption as the author bootstrap above — remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
# a target whose base ref cannot be derived must say why.
_base_state = root_checkout_guard.resolve_remote_master_ref_state(
ctx["canonical_repo_root"] ctx["canonical_repo_root"]
) )
remote_master_sha = _base_state["sha"]
if not remote_master_sha and _base_state.get("reasons"):
remote_master_sha_error = "; ".join(_base_state["reasons"])
except Exception as exc: except Exception as exc:
remote_master_sha = None remote_master_sha = None
remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed" remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
@@ -1315,12 +1303,7 @@ def _run_anti_stomp_preflight(
workspace = ctx["workspace_path"] workspace = ctx["workspace_path"]
canonical_root = ctx["canonical_repo_root"] canonical_root = ctx["canonical_repo_root"]
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root) git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
# #983: derive the target base ref for this repository and carry the ref remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
# itself alongside the SHA, so the anti-stomp root-checkout check reports the
# ref it actually compared.
_root_base_state = root_checkout_guard.resolve_remote_master_ref_state(canonical_root)
remote_master_sha = _root_base_state["sha"]
remote_master_ref = _root_base_state.get("ref")
# Repo/org facts (best-effort; explicit org/repo when provided). # Repo/org facts (best-effort; explicit org/repo when provided).
resolved_org = org resolved_org = org
@@ -1449,7 +1432,6 @@ def _run_anti_stomp_preflight(
root_head_sha=git_state.get("head_sha"), root_head_sha=git_state.get("head_sha"),
root_porcelain=git_state.get("porcelain_status") or "", root_porcelain=git_state.get("porcelain_status") or "",
remote_master_sha=remote_master_sha, remote_master_sha=remote_master_sha,
remote_master_ref=remote_master_ref,
startup_head=startup_head, startup_head=startup_head,
current_code_head=current_code_head, current_code_head=current_code_head,
lease_required=lease_required, lease_required=lease_required,
@@ -2005,11 +1987,7 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
canonical_root = ctx["canonical_repo_root"] canonical_root = ctx["canonical_repo_root"]
workspace = ctx["workspace_path"] workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root) git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
# #983: consume the resolved target so the contamination message names the remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
# ref that was actually compared (refs/remotes/<remote>/<branch>) instead of
# a hardcoded 'prgs/master' the target repository may not have.
base_state = root_checkout_guard.resolve_remote_master_ref_state(canonical_root)
remote_master_sha = base_state["sha"]
assessment = root_checkout_guard.assess_root_checkout_guard( assessment = root_checkout_guard.assess_root_checkout_guard(
workspace_path=workspace, workspace_path=workspace,
canonical_repo_root=canonical_root, canonical_repo_root=canonical_root,
@@ -2019,7 +1997,6 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
remote_master_sha=remote_master_sha, remote_master_sha=remote_master_sha,
resolved_role=_preflight_resolved_role, resolved_role=_preflight_resolved_role,
actual_role=_actual_profile_role(), actual_role=_actual_profile_role(),
remote_master_ref=base_state.get("ref"),
) )
if assessment["block"]: if assessment["block"]:
raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment)) raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
@@ -19646,6 +19623,701 @@ def gitea_snapshot_instance_fleet(
return snapshot return snapshot
# --- #980 CAS-protected stale worker retirement ---------------------------
#: #980 review 657 B1. Apply is a mutation and must not be authorized by the
#: observational ``gitea.read`` class. Declared once here and consumed by both
#: the entry gate and the immediately-pre-mutation re-check so the two can
#: never drift apart; ``task_capability_map`` maps the apply tasks to the same
#: literal.
RETIREMENT_MUTATION_PERMISSION = "gitea.worker_registry.retire"
def _retirement_role_block(task: str) -> dict | None:
"""Refuse the retirement surface to any non-controller/reconciler role.
Role is a *secondary* control. The primary authority for apply is the
dedicated ``gitea.worker_registry.retire`` capability (#980 review 657 B1);
this check additionally pins the surface to the two roles that own fleet
reconciliation, so a profile mistakenly granted the permission still cannot
reach it from an author, reviewer, or merger role. Plan remains
``gitea.read`` and is observational.
"""
profile = get_profile()
role = _profile_role_kind(profile)
if role in {"controller", "reconciler"}:
return None
return {
"success": False,
"allowed": False,
"mutation_performed": False,
"retired_count": 0,
"denied_role": role,
"required_roles": ["controller", "reconciler"],
"requested_task": task,
"reasons": [
f"{task} is restricted to controller and reconciler roles; active "
f"role_kind is {role!r}. Author, reviewer, and merger profiles keep "
"gitea.read for diagnosis elsewhere but never receive worker-"
"registry retirement, and no unrelated mutation permission is "
"granted."
],
"exact_next_action": (
"Re-run from a prgs-controller or prgs-reconciler namespace."
),
}
def _retirement_runtime_block() -> list[str]:
"""Fail-closed runtime reasons that must stop a retirement apply (#980).
``gitea.read`` deliberately bypasses the #420 parity gate and the #615
stable-runtime gate, because a stale server may still be *inspected*. An
apply is a mutation, so both gates are re-asserted explicitly here rather
than inherited.
"""
reasons: list[str] = []
try:
parity = _current_master_parity()
except Exception as exc:
return [
f"master parity could not be assessed (fail closed): {_redact(str(exc))}"
]
if not parity.get("mutation_safe"):
reasons.append(
"runtime parity is not mutation-safe: "
f"{parity.get('summary') or 'stale runtime'}"
)
try:
reasons.extend(
stable_control_runtime.runtime_block_reasons(
_current_runtime_mode_report()
)
)
except Exception as exc:
reasons.append(
f"runtime mode could not be assessed (fail closed): {_redact(str(exc))}"
)
return reasons
def _retirement_protected_owners() -> tuple[dict | None, dict]:
"""Workflow owners that must never be retired as stale workers.
A registration whose process still owns an active control-plane lease is a
live workflow participant needing its own reconciliation, not a stale
orphan. Failure to enumerate leases is ambiguity, so it aborts rather than
proceeding with an empty protection set.
"""
db, errs = _control_plane_db_or_error()
if db is None:
return (
{
"success": False,
"mutation_performed": False,
"retired_count": 0,
"reasons": [
"active workflow leases could not be enumerated, so "
"protected owners are unknown (fail closed)",
*errs,
],
},
{},
)
try:
leases = db.list_leases(statuses=["active"], limit=1000)
except Exception as exc:
return (
{
"success": False,
"mutation_performed": False,
"retired_count": 0,
"reasons": [
"active workflow leases could not be enumerated, so "
"protected owners are unknown (fail closed): "
f"{_redact(str(exc))}"
],
},
{},
)
session_ids: set[str] = set()
pids: set[int] = set()
for lease in leases:
owner = lease.get("session_id") or lease.get("owner_session_id")
if owner:
session_ids.add(str(owner))
for key in ("owner_pid", "session_pid"):
value = lease.get(key)
if value is None:
continue
try:
pids.add(int(value))
except (TypeError, ValueError):
continue
return None, {
"session_ids": sorted(session_ids),
"pids": sorted(pids),
"active_lease_count": len(leases),
}
def _retirement_plan(records, canonical_repository: str, protected: dict) -> dict:
"""The single planning path shared by dry run and in-transaction revalidation."""
import mcp_fleet_retirement
import mcp_process_fencing
return mcp_fleet_retirement.plan_stale_worker_retirement(
records,
pid_alive_probe=issue_lock_store.is_process_alive,
canonical_repository=canonical_repository,
protected_session_ids=(protected or {}).get("session_ids"),
protected_pids=(protected or {}).get("pids"),
# #980 review 657 B2: the fencing evidence that makes a recorded pid
# interpretable. Probed live so a plan produced on one host can never
# authorize a retirement carried out on another.
current_host_id=mcp_process_fencing.current_host_id(),
current_boot_id=mcp_process_fencing.current_boot_id(),
start_time_probe=mcp_process_fencing.process_start_time,
)
def _retirement_external_fence() -> str:
"""Version token over lease + liveness state consumed by a retirement (#980 B3).
Recomputed inside the retirement transaction immediately before every
guarded write. Because neither the control-plane lease database nor the OS
process table is covered by the registry's ``BEGIN IMMEDIATE``, this token
is the only thing that can detect either of them moving mid-transaction.
A failure to read leases raises rather than returning a token: an
unreadable external world is indistinguishable from a changed one, so it
must abort the transaction rather than silently compare equal.
"""
import mcp_fleet_retirement
db, errs = _control_plane_db_or_error()
if db is None:
raise RuntimeError(
"active workflow leases could not be read while fencing the "
f"retirement transaction: {errs}"
)
leases = db.list_leases(statuses=["active"], limit=1000)
return mcp_fleet_retirement.external_state_fingerprint(leases)
def _retirement_liveness_reprobe(row: dict) -> dict:
"""Immediate pre-write re-establishment that one specific process is gone.
#980 review 657 B3: revalidation happens once per transaction, but the
per-target loop runs afterwards, so this re-probe closes the remaining
window between "this row was judged safe" and "this row is written". It
re-reads OS state rather than trusting the plan, and it compares the
fencing triple so a pid number reused since the plan cannot pass.
"""
import mcp_fleet_retirement
import mcp_process_fencing
pid = row.get("pid")
try:
alive = issue_lock_store.is_process_alive(pid)
except Exception as exc:
return {
"safe": False,
"reason_code": mcp_fleet_retirement.REASON_PID_UNKNOWN,
"detail": (
f"pid {pid!r} could not be re-probed immediately before the "
f"write: {_redact(str(exc))}"
),
}
if alive:
live_start = mcp_process_fencing.process_start_time(pid)
recorded_start = (row.get("process_start_time") or "").strip()
reused = bool(live_start) and live_start != recorded_start
return {
"safe": False,
"reason_code": (
mcp_fleet_retirement.REASON_PID_REUSED
if reused
else mcp_fleet_retirement.REASON_PID_ALIVE
),
"detail": (
f"pid {pid!r} is alive at write time"
+ (
f" but is a different incarnation ({live_start!r} != "
f"{recorded_start!r})"
if reused
else ""
)
),
"evidence": {
"pid": pid,
"pid_alive": True,
"live_process_start_time": live_start,
"recorded_process_start_time": recorded_start,
},
}
host_now = mcp_process_fencing.current_host_id()
boot_now = mcp_process_fencing.current_boot_id()
if not host_now or (row.get("host_id") or "").strip() != host_now:
return {
"safe": False,
"reason_code": mcp_fleet_retirement.REASON_HOST_UNPROVEN,
"detail": (
"host identity no longer agrees with the registration at write "
f"time (row {row.get('host_id')!r}, now {host_now!r})"
),
}
if not boot_now:
return {
"safe": False,
"reason_code": mcp_fleet_retirement.REASON_BOOT_UNKNOWN,
"detail": "boot identity became unobtainable before the write",
}
return {
"safe": True,
"evidence": {
"pid": pid,
"pid_alive": False,
"host_id": host_now,
"boot_id": boot_now,
},
}
def _retirement_revalidation_plan(records, canonical_repository: str) -> dict:
"""Revalidation planner used *inside* the retirement transaction.
Deliberately re-reads the active workflow leases rather than reusing the
set captured before the transaction opened: a lease acquired after planning
must still preserve its worker. An enumeration failure raises, which rolls
the transaction back and retires nothing.
"""
block, protected = _retirement_protected_owners()
if block:
raise RuntimeError(
"active workflow leases could not be re-read inside the retirement "
"transaction; refusing to retire anything"
)
return _retirement_plan(records, canonical_repository, protected)
@mcp.tool()
def gitea_plan_stale_worker_retirement(
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
canonical_repository: str | None = None,
) -> dict:
"""Read-only: authoritative retirement plan for stale worker rows (#980).
Controller and reconciler only. Reads the worker registry, classifies every
registration with the same assessor the #978 fleet snapshot uses, and
returns the exact set of registrations that are conclusively stale orphans
together with a **stable** ``registry_fingerprint`` and an exact
``candidate_fingerprint``.
The fingerprint is derived only from canonical retirement-relevant registry
content never from ``snapshot_at``, wall-clock, request, or report time
so two plans over an unchanged registry agree and the apply compare-and-swap
can actually pass. Row order never affects it.
Retires nothing. Live workers, workers whose PID cannot be probed, workers
with unparsable heartbeats, workers sharing identity evidence with a live or
unprobeable worker, foreign or unbound repositories, and workers that still
own an active workflow lease are all preserved with a reason code.
Args:
remote: Known instance 'dadeschools' or 'prgs'.
host: Optional host override.
org: Optional org override (audit context only).
repo: Optional repo override (audit context only).
canonical_repository: Expected repository binding for
foreign-repository classification (defaults to the process root).
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"read_only": True,
"mutation_performed": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
role_block = _retirement_role_block("gitea_plan_stale_worker_retirement")
if role_block:
role_block["read_only"] = True
return role_block
registry = _worker_registry()
if registry is None:
return {
"success": False,
"read_only": True,
"mutation_performed": False,
"reasons": [
"worker registry is unavailable; cannot produce an authoritative "
"retirement plan (fail closed)"
],
"exact_next_action": (
"Ensure GITEA_WORKER_REGISTRY_DB is writable and re-run after "
"workers have registered."
),
}
protected_block, protected = _retirement_protected_owners()
if protected_block:
protected_block["read_only"] = True
return protected_block
try:
records = registry.list_workers(status=None)
except Exception as exc:
return {
"success": False,
"read_only": True,
"mutation_performed": False,
"reasons": [
f"failed to read worker registry: {type(exc).__name__}: "
f"{_redact(str(exc))}"
],
}
canon = canonical_repository or PROJECT_ROOT
plan = _retirement_plan(records, canon, protected)
profile = get_profile()
plan["role_kind"] = _profile_role_kind(profile)
plan["profile"] = profile.get("profile_name")
plan["remote"] = _effective_remote(remote)
plan["repository"] = {"org": org, "repo": repo, "canonical_repository": canon}
plan["protected_active_workflow_owners"] = protected
plan["apply_tool"] = "gitea_apply_stale_worker_retirement"
plan["permission_scope"] = {
"read_only": True,
"granted_operations": ["gitea.read"],
"denied_unrelated_mutations": True,
"note": (
"Planning is strictly observational. It does not authorize branch, "
"issue, PR, review, merge, or restart mutations, and it retires "
"nothing."
),
}
plan["exact_next_action"] = (
"Pass registry_fingerprint, candidate_fingerprint, and the exact "
"candidate_worker_identities to gitea_apply_stale_worker_retirement."
if plan.get("candidate_count")
else "No registration is conclusively stale; nothing to apply."
)
return plan
@mcp.tool()
def gitea_apply_stale_worker_retirement(
registry_fingerprint: str,
candidate_fingerprint: str,
worker_identities: list | str,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
canonical_repository: str | None = None,
) -> dict:
"""Retire conclusively stale worker registrations under CAS (#980).
Controller and reconciler only. Requires the exact ``registry_fingerprint``
and ``candidate_fingerprint`` returned by
``gitea_plan_stale_worker_retirement`` plus the exact candidate identity
list. A matching token is necessary but never sufficient: inside a single
``BEGIN IMMEDIATE`` transaction the registry is re-read, the fingerprint is
recomputed from those rows, the eligibility plan is recomputed from those
rows, and every target is independently revalidated immediately before its
own guarded ``UPDATE``. Any drift retires zero workers and reports
``registry_revision_moved`` or ``candidate_set_moved``.
Retiring stale rows does not repair untrusted live identity. Live workers
registered under legacy ``pid-``/``proc-`` instance identities remain
untouched and their ``legacy_incomplete_identity`` blockers remain
outstanding, so the result never claims the fleet became safe.
Args:
registry_fingerprint: Stable token from the plan (CAS expectation).
candidate_fingerprint: Exact candidate-set token from the plan.
worker_identities: The exact candidate identities the plan returned
(list, or a JSON / comma-separated string).
remote: Known instance 'dadeschools' or 'prgs'.
host: Optional host override.
org: Optional org override (audit context only).
repo: Optional repo override (audit context only).
canonical_repository: Expected repository binding (defaults to the
process root); must match the value the plan used.
"""
import json as _json
import mcp_fleet_retirement
# #980 review 657 B1: the dedicated mutation capability, not gitea.read.
# A profile holding only the observational permission cannot get past this
# line, and the role check below is defence in depth rather than the sole
# authority.
mutation_block = _profile_operation_gate(RETIREMENT_MUTATION_PERMISSION)
if mutation_block:
return {
"success": False,
"mutation_performed": False,
"retired_count": 0,
"requested_task": "apply_stale_worker_retirement",
"required_operation_permission": RETIREMENT_MUTATION_PERMISSION,
"reasons": mutation_block,
"permission_report": _permission_block_report(
RETIREMENT_MUTATION_PERMISSION
),
}
role_block = _retirement_role_block("gitea_apply_stale_worker_retirement")
if role_block:
return role_block
runtime_reasons = _retirement_runtime_block()
if runtime_reasons:
return {
"success": False,
"mutation_performed": False,
"retired_count": 0,
"blocker_kind": "runtime_not_mutation_safe",
"reasons": runtime_reasons,
"exact_next_action": (
"Restore runtime parity on the stable control checkout, then "
"re-plan and re-apply."
),
}
# Fresh identity + capability resolution immediately before mutation.
profile = get_profile()
role_kind = _profile_role_kind(profile)
required_permission = task_capability_map.required_permission(
"apply_stale_worker_retirement"
)
required_role = task_capability_map.required_role(
"apply_stale_worker_retirement"
)
permission_ok, permission_reason = gitea_config.check_operation(
required_permission,
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
if not permission_ok:
return {
"success": False,
"mutation_performed": False,
"retired_count": 0,
"requested_task": "apply_stale_worker_retirement",
"required_operation_permission": required_permission,
"required_role_kind": required_role,
"reasons": [
"capability resolution immediately before apply refused this "
f"session: {permission_reason}"
],
}
try:
resolved_host = host or REMOTES[_effective_remote(remote)]["host"]
authenticated_username = _authenticated_username(resolved_host)
except Exception:
authenticated_username = None
registry = _worker_registry()
if registry is None:
return {
"success": False,
"mutation_performed": False,
"retired_count": 0,
"reasons": [
"worker registry is unavailable; refusing to retire anything "
"(fail closed)"
],
}
raw = worker_identities
if isinstance(raw, str):
text = raw.strip()
try:
raw = _json.loads(text)
except Exception:
raw = [part.strip() for part in text.split(",") if part.strip()]
if isinstance(raw, str):
raw = [raw]
if not isinstance(raw, list):
return {
"success": False,
"mutation_performed": False,
"retired_count": 0,
"reasons": ["worker_identities must be a list of worker identities"],
}
targets = [str(item).strip() for item in raw if str(item).strip()]
protected_block, protected = _retirement_protected_owners()
if protected_block:
return protected_block
# #948 daemon-cohort uniqueness: a contested generation or a reused worker
# identity means ownership is ambiguous fleet-wide, so retire nothing.
try:
cohort = mcp_worker_identity.classify_cohort(
registry.list_workers(status=mcp_worker_identity.STATUS_ACTIVE),
pid_alive_probe=issue_lock_store.is_process_alive,
)
except Exception as exc:
return {
"success": False,
"mutation_performed": False,
"retired_count": 0,
"reasons": [
"daemon-cohort uniqueness could not be assessed (fail closed): "
f"{_redact(str(exc))}"
],
}
if cohort.get("blocked"):
return {
"success": False,
"mutation_performed": False,
"retired_count": 0,
"blocker_kind": cohort.get("blocker_kind"),
"reasons": [
"daemon-cohort uniqueness failed; worker ownership is contested",
*(cohort.get("reasons") or []),
],
"cohort": {
"blocked_worker_identities": cohort.get("blocked_worker_identities"),
"duplicate_identities": cohort.get("duplicate_identities"),
"contested_generations": cohort.get("contested_generations"),
},
}
canon = canonical_repository or PROJECT_ROOT
acting = "/".join(
part
for part in (authenticated_username, profile.get("profile_name"))
if part
) or "unknown"
result = registry.retire_stale_workers(
expected_registry_fingerprint=registry_fingerprint,
expected_candidate_fingerprint=candidate_fingerprint,
worker_identities=targets,
fingerprint_fn=mcp_fleet_retirement.registry_fingerprint,
plan_fn=lambda rows: _retirement_revalidation_plan(rows, canon),
retired_by=acting,
retirement_reason=mcp_fleet_retirement.REASON_ELIGIBLE,
external_fence_fn=_retirement_external_fence,
liveness_fn=_retirement_liveness_reprobe,
)
result["role_kind"] = role_kind
result["profile"] = profile.get("profile_name")
result["remote"] = _effective_remote(remote)
result["repository"] = {"org": org, "repo": repo, "canonical_repository": canon}
result["protected_active_workflow_owners"] = protected
result["permission_scope"] = {
"granted_operations": [RETIREMENT_MUTATION_PERMISSION],
"mutation_capability": RETIREMENT_MUTATION_PERMISSION,
"plan_capability": "gitea.read",
"control_plane_mutation": "worker_registrations.status -> retired",
"denied_unrelated_mutations": True,
"note": (
"This capability retires local worker-registry rows only. It grants "
"no branch, issue, PR, review, merge, or restart authority, and it "
"never kills or restarts a process."
),
}
result["post_apply"] = _retirement_post_apply(
registry, result, canonical_repository=canon
)
try:
gitea_audit.write_event(
gitea_audit.build_event(
action="gitea_apply_stale_worker_retirement",
result=(
gitea_audit.SUCCEEDED
if result.get("mutation_performed")
else gitea_audit.BLOCKED
if not result.get("success")
else gitea_audit.ALLOWED
),
remote=_effective_remote(remote),
repository=canon,
profile_name=profile.get("profile_name"),
audit_label=profile.get("audit_label"),
authenticated_username=authenticated_username,
task_role=role_kind,
operation="worker_registry.retire_stale_workers",
reason=result.get("outcome"),
request_metadata=mcp_fleet_retirement.summarize_plan(result),
)
)
except Exception:
pass
return result
def _retirement_post_apply(
registry, result: dict, *, canonical_repository: str
) -> dict:
"""Fresh fleet verification after a retirement attempt (#980 requirement 6)."""
import mcp_fleet_retirement
import mcp_fleet_snapshot as _fleet_after
try:
after_records = registry.list_workers(status=None)
after = _fleet_after.snapshot_instance_fleet(
after_records,
pid_alive_probe=issue_lock_store.is_process_alive,
canonical_repository=canonical_repository,
)
except Exception as exc:
return {
"available": False,
"reasons": [
f"post-apply verification could not be produced: {_redact(str(exc))}"
],
}
retired_ids = {
str(r.get("worker_identity")) for r in result.get("retired") or []
}
return {
"available": True,
"registry_fingerprint": mcp_fleet_retirement.registry_fingerprint(
after_records
),
"live_worker_count": after.get("live_worker_count"),
"stale_worker_count": after.get("stale_worker_count"),
"historical_worker_count": after.get("historical_worker_count"),
"retired_still_counted_live": sorted(
str(w.get("worker_identity"))
for w in after.get("live_workers") or []
if str(w.get("worker_identity")) in retired_ids
),
"retired_still_counted_stale": sorted(
str(w.get("worker_identity"))
for w in after.get("stale_workers") or []
if str(w.get("worker_identity")) in retired_ids
),
"live_fleet_safe": after.get("live_fleet_safe"),
"remaining_blockers": [
{"classification": f.get("classification"), "detail": f.get("detail")}
for f in after.get("active_blockers") or []
],
"note": (
"Stale retirement does not repair untrusted live identity; residual "
"legacy_incomplete_identity blockers keep live_fleet_safe false and "
"that is a truthful result."
),
}
@mcp.tool() @mcp.tool()
def gitea_get_runtime_context( def gitea_get_runtime_context(
remote: str = "dadeschools", remote: str = "dadeschools",
+11 -60
View File
@@ -378,11 +378,6 @@ def format_parity(assessment: dict) -> str:
# feeds the mutation gate and never changes startup_head/current_head. # feeds the mutation gate and never changes startup_head/current_head.
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Legacy hardcoded target tracking ref. Retained for callers that still pass an
# explicit ref, but no longer the default: assuming a remote named ``origin``
# read an unrelated (often orphaned) remote-tracking ref in any checkout whose
# remote is named something else, and reported the target stale against a commit
# from a remote that may no longer even be configured (#983).
DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master" DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master"
@@ -413,7 +408,7 @@ def assess_target_repository_parity(
*, *,
canonical_root: str | None, canonical_root: str | None,
source: str | None, source: str | None,
tracking_ref: str | None = None, tracking_ref: str = DEFAULT_TARGET_TRACKING_REF,
) -> dict: ) -> dict:
"""Assess the configured cross-repository target checkout. """Assess the configured cross-repository target checkout.
@@ -426,11 +421,6 @@ def assess_target_repository_parity(
An unconfigured namespace is ``configured=False`` and never ``stale`` — the An unconfigured namespace is ``configured=False`` and never ``stale`` — the
single-repository default has no second dimension to be stale about. A single-repository default has no second dimension to be stale about. A
configured root that cannot be read is ``determinable=False`` with reasons. configured root that cannot be read is ``determinable=False`` with reasons.
*tracking_ref* defaults to None, meaning **derive the target from the
repository itself** through the same resolver the mutation guard uses, so
gating and reporting can never disagree about which ref is authoritative
(#983). Passing an explicit ref preserves the previous behaviour verbatim.
""" """
result = { result = {
"configured": bool(canonical_root), "configured": bool(canonical_root),
@@ -439,13 +429,9 @@ def assess_target_repository_parity(
"repository_slug": None, "repository_slug": None,
"checkout_head": None, "checkout_head": None,
"tracking_ref": tracking_ref, "tracking_ref": tracking_ref,
"base_remote": None,
"base_branch": None,
"tracking_ref_source": "explicit_tracking_ref" if tracking_ref else None,
"remote_tracking_head": None, "remote_tracking_head": None,
"determinable": False, "determinable": False,
"stale": False, "stale": False,
"reason_code": None,
"reasons": [], "reasons": [],
} }
if not canonical_root: if not canonical_root:
@@ -477,53 +463,18 @@ def assess_target_repository_parity(
result["checkout_head"] = head result["checkout_head"] = head
result["determinable"] = True result["determinable"] = True
# Local import keeps this module dependency-light for its startup role. remote_url = _git_capture(canonical_root, "remote", "get-url", "origin")
import canonical_repository_root as _crr if remote_url:
# Local import keeps this module dependency-light for its startup role.
import remote_repo_guard
# #983: identity comes from whichever remote actually proves it, not from a parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url)
# remote assumed to be named 'origin'. In a checkout whose only remote is if parsed:
# 'prgs', the old lookup failed outright and reported the identity as result["repository_slug"] = f"{parsed[0]}/{parsed[1]}"
# underivable while a leftover refs/remotes/origin/master still resolved. if not result["repository_slug"]:
# result["reasons"].append(
# #983 B2: this must be the *ambiguity-aware* resolver. The first-wins "target repository identity could not be derived from its git remote"
# `resolve_identity_remote` picks whichever remote probes first, so on a
# target where distinct remotes claim different repositories the report
# confidently named one of them while the mutation gate refused the same
# target — gating and reporting evaluating different repositories, which is
# precisely the divergence this issue exists to end.
identity = _crr.assess_identity_remote(canonical_root)
result["repository_slug"] = identity["slug"]
if not identity["slug"]:
result["reason_code"] = identity["reason_code"]
result["reasons"].extend(
identity["reasons"]
or ["target repository identity could not be derived from its git remote"]
) )
if identity["ambiguous"]:
# An ambiguous target has no single authoritative base, so reporting one
# would be a guess. Fail closed here exactly as the gate does.
return result
# Identity is otherwise resolved independently of the base ref: a target that
# has never been fetched still has a provable repository identity, and
# reporting it as unidentifiable would lose real information over an
# unrelated missing ref.
if not tracking_ref:
# No remote argument. The identity remote resolved just above was
# *inferred here*, and feeding it back in would tell the resolver a
# caller had explicitly disambiguated the repository, suppressing its
# ambiguity gate (#983 B2). Only an operator-supplied remote may do that,
# and this call site has none.
base = _crr.resolve_target_base_ref(canonical_root)
if not base.get("proven"):
result["reason_code"] = base.get("reason_code")
result["reasons"].extend(base.get("reasons") or [])
return result
tracking_ref = base["tracking_ref"]
result["tracking_ref"] = tracking_ref
result["tracking_ref_source"] = base.get("source")
result["base_remote"] = base.get("remote")
result["base_branch"] = base.get("branch")
tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref) tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref)
if not tracking_head: if not tracking_head:
+19 -379
View File
@@ -15,7 +15,7 @@ from __future__ import annotations
import os import os
import secrets import secrets
from typing import Any, Mapping, Sequence from typing import Any, Mapping
import mcp_fleet_snapshot as fleet import mcp_fleet_snapshot as fleet
import mcp_worker_identity as mwi import mcp_worker_identity as mwi
@@ -152,54 +152,10 @@ def namespace_worker_env(
return env return env
def resolve_launch_namespaces(
namespaces: "Sequence[str] | None" = None,
) -> tuple[str, ...]:
"""Validate an explicit project-scoped namespace subset (#985).
``None`` keeps the historical whole-fleet behaviour and resolves to all five
sanctioned namespaces. An explicit sequence is validated against
:data:`SANCTIONED_NAMESPACES` and returned in canonical fleet order, so a
project-scoped configuration (for example author/reviewer/merger only) can
launch without inventing controller or reconciler profiles.
This is an allow-list, not a filter: anything outside the sanctioned set is
refused rather than silently dropped, so a typo can never quietly shrink a
launch to fewer workers than the operator intended.
"""
if namespaces is None:
return tuple(SANCTIONED_NAMESPACES)
requested = [str(ns).strip().lower() for ns in namespaces]
if not requested or any(not ns for ns in requested):
raise ValueError(
"namespaces must be a non-empty sequence of sanctioned namespace "
f"names; choose from {list(SANCTIONED_NAMESPACES)}"
)
unknown = sorted({ns for ns in requested if ns not in SANCTIONED_NAMESPACES})
if unknown:
raise ValueError(
f"unsanctioned namespace(s) {unknown} requested; sanctioned "
f"namespaces are {list(SANCTIONED_NAMESPACES)}"
)
duplicates = sorted({ns for ns in requested if requested.count(ns) > 1})
if duplicates:
raise ValueError(
f"duplicate namespace(s) {duplicates} requested; each sanctioned "
"namespace may appear at most once in one launch"
)
selected = set(requested)
return tuple(ns for ns in SANCTIONED_NAMESPACES if ns in selected)
def build_application_mcp_servers( def build_application_mcp_servers(
profile_by_namespace: Mapping[str, str], profile_by_namespace: Mapping[str, str],
*, *,
client_type: str | None, client_type: str | None,
namespaces: "Sequence[str] | None" = None,
config_path: str | None = None, config_path: str | None = None,
client_instance_id: str | None = None, client_instance_id: str | None = None,
fleet_run_id: str | None = None, fleet_run_id: str | None = None,
@@ -219,22 +175,14 @@ def build_application_mcp_servers(
Every namespace server entry receives the **same** instance ID. Separate Every namespace server entry receives the **same** instance ID. Separate
calls with no supplied ID receive distinct IDs. calls with no supplied ID receive distinct IDs.
*namespaces* selects an explicitly validated project-scoped subset (#985).
Omitting it preserves the original whole-fleet behaviour, so existing
five-namespace callers are unaffected. Only the resolved namespaces need a
profile, and only they are started; excluded namespaces are reported so a
caller can prove a controller/reconciler worker was never launched.
""" """
resolved_namespaces = resolve_launch_namespaces(namespaces)
missing = [ missing = [
ns for ns in resolved_namespaces if not profile_by_namespace.get(ns) ns for ns in SANCTIONED_NAMESPACES if not profile_by_namespace.get(ns)
] ]
if missing: if missing:
raise ValueError( raise ValueError(
"build_application_mcp_servers requires a profile for every " "build_application_mcp_servers requires a profile for every "
f"requested namespace; missing: {missing}" f"sanctioned namespace; missing: {missing}"
) )
if client_instance_id is None: if client_instance_id is None:
@@ -265,13 +213,6 @@ def build_application_mcp_servers(
"namespace_server_names": list(NAMESPACE_SERVER_NAMES), "namespace_server_names": list(NAMESPACE_SERVER_NAMES),
} }
# The launch record describes only the namespaces this launch actually
# starts, so downstream attribution never implies an unstarted worker.
launch["namespaces"] = list(resolved_namespaces)
launch["namespace_server_names"] = [
f"gitea-{ns}" for ns in resolved_namespaces
]
# Resolve command/args from the production server entry when not provided. # Resolve command/args from the production server entry when not provided.
if command is None or args is None: if command is None or args is None:
import gitea_config import gitea_config
@@ -282,7 +223,7 @@ def build_application_mcp_servers(
servers: dict[str, Any] = {} servers: dict[str, Any] = {}
shared_id = launch["client_instance_id"] shared_id = launch["client_instance_id"]
for namespace in resolved_namespaces: for namespace in SANCTIONED_NAMESPACES:
server_name = f"gitea-{namespace}" server_name = f"gitea-{namespace}"
profile = profile_by_namespace[namespace] profile = profile_by_namespace[namespace]
env = namespace_worker_env( env = namespace_worker_env(
@@ -299,17 +240,14 @@ def build_application_mcp_servers(
"env": env, "env": env,
} }
excluded = [ns for ns in SANCTIONED_NAMESPACES if ns not in resolved_namespaces]
return { return {
"mcpServers": servers, "mcpServers": servers,
"launch": launch, "launch": launch,
"client_instance_id": shared_id, "client_instance_id": shared_id,
"client_type": launch["client_type"], "client_type": launch["client_type"],
"namespaces": list(resolved_namespaces), "namespaces": list(SANCTIONED_NAMESPACES),
"excluded_namespaces": excluded,
"project_scoped": bool(excluded),
"shared_instance_id_across_namespaces": True, "shared_instance_id_across_namespaces": True,
"namespace_count": len(resolved_namespaces), "namespace_count": len(SANCTIONED_NAMESPACES),
} }
@@ -378,33 +316,15 @@ def launcher_entry_for_profile(
def collect_instance_ids_from_mcp_servers( def collect_instance_ids_from_mcp_servers(
mcp_servers: Mapping[str, Any], mcp_servers: Mapping[str, Any],
*,
namespaces: "Sequence[str] | None" = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Inspect a production mcpServers map for shared instance attribution. """Inspect a production mcpServers map for shared instance attribution.
Returns the unique set of client_instance_id values across the gitea-* Returns the unique set of client_instance_id values across gitea-* servers
servers of one launch and whether every worker of that launch shares and whether all five namespaces share exactly one trusted ID.
exactly one trusted ID.
The inspected set is the launch's own namespaces, not an assumed five
(#985): a project-scoped author/reviewer/merger launch must be able to
prove shared attribution without a controller or reconciler entry counting
as a missing worker. Pass *namespaces* to assert an exact expected subset;
omit it to inspect whichever sanctioned namespace servers are present.
""" """
if namespaces is None:
expected_servers = [
name for name in NAMESPACE_SERVER_NAMES if name in mcp_servers
]
else:
expected_servers = [
f"gitea-{ns}" for ns in resolve_launch_namespaces(namespaces)
]
ids: list[str] = [] ids: list[str] = []
by_server: dict[str, str | None] = {} by_server: dict[str, str | None] = {}
for name in expected_servers: for name in NAMESPACE_SERVER_NAMES:
entry = mcp_servers.get(name) or {} entry = mcp_servers.get(name) or {}
env = entry.get("env") or {} env = entry.get("env") or {}
raw = (env.get(CLIENT_INSTANCE_ENV) or "").strip() or None raw = (env.get(CLIENT_INSTANCE_ENV) or "").strip() or None
@@ -417,19 +337,14 @@ def collect_instance_ids_from_mcp_servers(
for i in unique for i in unique
if fleet.assess_instance_identity(i)["trusted"] if fleet.assess_instance_identity(i)["trusted"]
] ]
missing_servers = [n for n in expected_servers if n not in mcp_servers]
return { return {
"server_instance_ids": by_server, "server_instance_ids": by_server,
"unique_instance_ids": unique, "unique_instance_ids": unique,
"trusted_instance_ids": trusted, "trusted_instance_ids": trusted,
"inspected_servers": list(expected_servers),
"missing_servers": missing_servers,
"shared_single_trusted_id": ( "shared_single_trusted_id": (
bool(expected_servers) len(unique) == 1
and not missing_servers
and len(unique) == 1
and len(trusted) == 1 and len(trusted) == 1
and all(by_server.get(n) == unique[0] for n in expected_servers) and all(by_server.get(n) == unique[0] for n in NAMESPACE_SERVER_NAMES)
), ),
"namespace_server_count": sum( "namespace_server_count": sum(
1 for n in NAMESPACE_SERVER_NAMES if n in mcp_servers 1 for n in NAMESPACE_SERVER_NAMES if n in mcp_servers
@@ -450,23 +365,20 @@ def inherit_or_refuse_client_instance(
source = dict(env if env is not None else os.environ) source = dict(env if env is not None else os.environ)
raw = (source.get(CLIENT_INSTANCE_ENV) or "").strip() or None raw = (source.get(CLIENT_INSTANCE_ENV) or "").strip() or None
provenance_marker = (source.get(INSTANCE_PROVENANCE_ENV) or "").strip() provenance_marker = (source.get(INSTANCE_PROVENANCE_ENV) or "").strip()
assessment = dict(fleet.assess_instance_identity(raw)) assessment = fleet.assess_instance_identity(raw)
# The ``inst-…`` format is public and reproducible, so format alone cannot # Ordinary user-supplied values without launcher provenance marker are
# establish trust: anyone able to set one env var could otherwise hand-write # still format-checked by assess_instance_identity. When the format is
# a well-formed ID and be believed. Trust therefore requires the launcher's # trusted but the launcher marker is absent, keep the ID but note that
# own seal as well, and a well-formed but unsealed identity fails closed # provenance is not launcher-sealed (operator hand-set or legacy config).
# rather than degrading to a mere annotation (#985).
if assessment["trusted"] and provenance_marker != LAUNCHER_PROVENANCE_VALUE: if assessment["trusted"] and provenance_marker != LAUNCHER_PROVENANCE_VALUE:
assessment = dict(assessment)
assessment["launcher_sealed"] = False assessment["launcher_sealed"] = False
assessment["trusted"] = False
assessment["complete"] = False
assessment["provenance"] = fleet.INSTANCE_ID_PROVENANCE_UNSEALED
assessment["reasons"] = list(assessment.get("reasons") or []) + [ assessment["reasons"] = list(assessment.get("reasons") or []) + [
f"{INSTANCE_PROVENANCE_ENV} is not {LAUNCHER_PROVENANCE_VALUE!r}; " f"{INSTANCE_PROVENANCE_ENV} is not {LAUNCHER_PROVENANCE_VALUE!r}; "
"identity format is valid but not sealed by the production " "identity format is valid but not sealed by the production launcher"
"launcher, so trusted attribution is refused"
] ]
else: else:
assessment = dict(assessment)
assessment["launcher_sealed"] = bool( assessment["launcher_sealed"] = bool(
assessment["trusted"] assessment["trusted"]
and provenance_marker == LAUNCHER_PROVENANCE_VALUE and provenance_marker == LAUNCHER_PROVENANCE_VALUE
@@ -479,275 +391,3 @@ def inherit_or_refuse_client_instance(
(source.get("GITEA_MCP_CLIENT") or "").strip() or None (source.get("GITEA_MCP_CLIENT") or "").strip() or None
) )
return assessment return assessment
# ---------------------------------------------------------------------------
# Runnable launch path (#985)
# ---------------------------------------------------------------------------
#: How each supported LLM client is started with a per-launch MCP config.
#:
#: The launch interface is deliberately a data-driven registry rather than a
#: Claude-specific code path: adding another supported client is one entry, and
#: every client necessarily goes through the same mint-once/propagate-to-all
#: identity mechanism because the argv builder only ever receives a config file
#: this module wrote.
CLIENT_LAUNCH_SPECS: dict[str, dict[str, Any]] = {
"claude_code": {
"command": "claude",
"config_args": lambda path: ["--mcp-config", path, "--strict-mcp-config"],
"docs": "claude --mcp-config <per-launch.json> --strict-mcp-config",
},
}
def supported_launch_clients() -> tuple[str, ...]:
"""Client types with a sanctioned runnable launch path."""
return tuple(sorted(CLIENT_LAUNCH_SPECS))
def build_client_launch_argv(
client_type: str,
config_path: str,
*,
extra_args: "Sequence[str] | None" = None,
) -> list[str]:
"""Build the argv that starts *client_type* against a per-launch config."""
client = mwi.normalize_client_name(client_type)
spec = CLIENT_LAUNCH_SPECS.get(client)
if spec is None:
raise ValueError(
f"no sanctioned runnable launch path for client {client!r}; "
f"supported clients: {list(supported_launch_clients())}"
)
argv = [str(spec["command"])] + list(spec["config_args"](str(config_path)))
if extra_args:
argv.extend(str(a) for a in extra_args)
return argv
def write_launch_config(
mcp_servers: Mapping[str, Any],
*,
directory: str | None = None,
) -> str:
"""Write one launch's ``mcpServers`` config to a fresh per-launch file.
The file is per-launch and owner-readable only. #985 explicitly rules out
persisting an instance ID into a shared, reused ``.mcp.json``: two
concurrent sessions reading one static trusted ID is precisely the reuse
case the duplicate guard must reject. A distinct file per launch keeps the
minted identity bound to the launch that minted it.
"""
import json
import tempfile
fd, path = tempfile.mkstemp(
prefix="gitea-mcp-launch-", suffix=".json", dir=directory
)
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump({"mcpServers": dict(mcp_servers)}, handle, indent=2)
handle.write("\n")
except Exception:
try:
os.unlink(path)
except OSError:
pass
raise
return path
def prepare_application_launch(
profile_by_namespace: Mapping[str, str],
*,
client_type: str,
namespaces: "Sequence[str] | None" = None,
config_path: str | None = None,
fleet_run_id: str | None = None,
session_id: str | None = None,
launch_nonce: str | None = None,
extra_args: "Sequence[str] | None" = None,
config_directory: str | None = None,
now=None,
) -> dict[str, Any]:
"""Mint one launch and produce everything needed to exec the client.
Returns the built servers, the per-launch config file path, the argv to
exec, and the shared trusted identity — without starting anything, so the
same preparation is testable and inspectable via ``--dry-run``.
"""
built = build_application_mcp_servers(
profile_by_namespace,
client_type=client_type,
namespaces=namespaces,
config_path=config_path,
fleet_run_id=fleet_run_id,
session_id=session_id,
launch_nonce=launch_nonce,
now=now,
)
# Fail before writing anything if this client has no runnable path.
argv_probe = build_client_launch_argv(
client_type, "<pending>", extra_args=extra_args
)
launch_config_path = write_launch_config(
built["mcpServers"], directory=config_directory
)
argv = build_client_launch_argv(
client_type, launch_config_path, extra_args=extra_args
)
proof = collect_instance_ids_from_mcp_servers(
built["mcpServers"], namespaces=built["namespaces"]
)
return {
"client_type": built["client_type"],
"client_instance_id": built["client_instance_id"],
"namespaces": built["namespaces"],
"excluded_namespaces": built["excluded_namespaces"],
"project_scoped": built["project_scoped"],
"launch_config_path": launch_config_path,
"argv": argv,
"argv_template": argv_probe,
"mcpServers": built["mcpServers"],
"launch": built["launch"],
"shared_single_trusted_id": proof["shared_single_trusted_id"],
"attribution_proof": proof,
}
def _parse_profile_assignments(values: "Sequence[str]") -> dict[str, str]:
"""Parse ``namespace=profile`` CLI pairs into a mapping."""
profiles: dict[str, str] = {}
for raw in values or ():
text = str(raw)
if "=" not in text:
raise ValueError(
f"invalid --profile {text!r}; expected namespace=profile-name"
)
namespace, _, profile = text.partition("=")
namespace = namespace.strip().lower()
profile = profile.strip()
if not namespace or not profile:
raise ValueError(
f"invalid --profile {text!r}; expected namespace=profile-name"
)
if namespace in profiles:
raise ValueError(f"duplicate --profile entry for namespace {namespace!r}")
profiles[namespace] = profile
return profiles
def main(argv: "Sequence[str] | None" = None) -> int:
"""Runnable entry point: start a supported LLM client for one launch.
Example (project-scoped, three namespaces)::
python3 -m mcp_application_launcher \\
--client claude_code \\
--namespaces author,reviewer,merger \\
--profile author=prgs-author \\
--profile reviewer=prgs-reviewer \\
--profile merger=prgs-merger
"""
import argparse
import json
parser = argparse.ArgumentParser(
prog="mcp_application_launcher",
description=(
"Start a supported LLM client with one trusted, per-launch "
"GITEA_MCP_CLIENT_INSTANCE shared by every MCP namespace worker."
),
)
parser.add_argument(
"--client",
default="claude_code",
help=f"client to launch; supported: {list(supported_launch_clients())}",
)
parser.add_argument(
"--namespaces",
default=None,
help=(
"comma-separated sanctioned namespace subset (e.g. "
"'author,reviewer,merger'); omit to launch all five"
),
)
parser.add_argument(
"--profile",
action="append",
default=[],
metavar="NAMESPACE=PROFILE",
help="profile for one namespace; repeat once per launched namespace",
)
parser.add_argument("--config-path", default=None, help="profiles.json path")
parser.add_argument("--fleet-run-id", default=None)
parser.add_argument("--session-id", default=None)
parser.add_argument(
"--dry-run",
action="store_true",
help="print the launch plan as JSON and exit without starting anything",
)
parser.add_argument(
"client_args",
nargs=argparse.REMAINDER,
help="arguments forwarded to the client after '--'",
)
ns = parser.parse_args(list(argv) if argv is not None else None)
try:
profiles = _parse_profile_assignments(ns.profile)
requested = (
[part.strip() for part in ns.namespaces.split(",") if part.strip()]
if ns.namespaces
else None
)
resolved = resolve_launch_namespaces(requested)
# Profiles for namespaces that are not being launched are almost always
# a mistake (a stale five-namespace invocation), so refuse rather than
# silently ignore them.
stray = sorted(set(profiles) - set(resolved))
if stray:
raise ValueError(
f"--profile given for namespace(s) {stray} that this launch "
f"does not start; launched namespaces are {list(resolved)}"
)
forwarded = [a for a in (ns.client_args or []) if a != "--"]
prepared = prepare_application_launch(
profiles,
client_type=ns.client,
namespaces=requested,
config_path=ns.config_path,
fleet_run_id=ns.fleet_run_id,
session_id=ns.session_id,
extra_args=forwarded,
)
except ValueError as exc:
parser.error(str(exc))
return 2 # pragma: no cover - argparse.error raises SystemExit
if ns.dry_run:
plan = {
key: prepared[key]
for key in (
"client_type",
"client_instance_id",
"namespaces",
"excluded_namespaces",
"project_scoped",
"launch_config_path",
"argv",
"shared_single_trusted_id",
)
}
print(json.dumps(plan, indent=2))
return 0
argv_to_exec = prepared["argv"]
os.execvp(argv_to_exec[0], argv_to_exec)
return 0 # pragma: no cover - execvp does not return
if __name__ == "__main__": # pragma: no cover - process entry point
raise SystemExit(main())
+776
View File
@@ -0,0 +1,776 @@
"""CAS-protected retirement planning for stale worker registrations (#980).
#978 (merged PR #979) made the fleet observable: every registered namespace
worker, its instance attribution, its heartbeat freshness, and a structured
classification. It deliberately stopped there — the snapshot is read-only and
the control plane still had no sanctioned way to retire registry rows whose
owning process is conclusively gone.
This module is the *decision layer* for that retirement. It is pure: callers
supply registry rows, a clock, and a PID probe; nothing here opens SQLite,
scans process tables, or mutates state. The transactional apply lives in
:meth:`mcp_worker_identity.WorkerRegistry.retire_stale_workers`, which calls
back into these same pure functions so plan and apply can never disagree about
what "the registry looks like" or "which rows are eligible".
Why a separate token
--------------------
``mcp_fleet_snapshot._consistency_token`` seeds its digest with ``snapshot_at``
at second precision, so ``registry_revision`` changes on every call even when
no registry row changed. A compare-and-swap gated on it can never pass — a
dry-run/apply cycle spanning more than one second aborts unconditionally. That
token is still useful as an observation stamp, so it is left exactly as it is;
#980 gets its own :func:`registry_fingerprint`, derived *only* from canonical
retirement-relevant row content:
* identical registry contents observed at any two times produce the same token,
* row order never affects the token (serialized rows are sorted),
* any create/delete/identity/liveness/ownership/registration-state change to a
retirement-relevant field changes the token.
Fail-closed posture
-------------------
A worker is retired only when the control plane *conclusively* establishes it
is a stale orphan. Missing evidence is never read as permission: an unprobeable
PID, an unparsable heartbeat, a row that shares identity evidence with a live
or unprobeable worker, a foreign or absent repository binding, or a worker that
still owns an active workflow lease all preserve the row.
Trusted launcher identity (``inst-…`` provenance) is deliberately *not* part of
the conjunction. #980 places trusted ``client_instance_id`` propagation out of
scope and lists "backfilling trusted identity for legacy workers" as a non-goal;
requiring it here would preserve every legacy row forever and make the feature
inert. What *is* required is that the registry fields the conjunction reads are
actually present — see :data:`REQUIRED_IDENTITY_FIELDS`.
"""
from __future__ import annotations
import hashlib
from datetime import datetime
from typing import Any, Callable, Iterable, Mapping, Sequence
import mcp_fleet_snapshot as fleet
import mcp_worker_identity as mwi
# --- Outcomes -------------------------------------------------------------
OUTCOME_PLANNED = "planned"
OUTCOME_APPLIED = "applied"
OUTCOME_REGISTRY_MOVED = "registry_revision_moved"
OUTCOME_CANDIDATES_MOVED = "candidate_set_moved"
OUTCOME_ALREADY_RETIRED = "already_retired"
OUTCOME_NOTHING_REQUESTED = "nothing_requested"
# --- Reason codes ---------------------------------------------------------
#: The only reason code that authorizes retirement.
REASON_ELIGIBLE = "eligible_stale_orphan"
REASON_ALREADY_TERMINAL = "already_terminal_registration"
REASON_AMBIGUOUS_OWNERSHIP = "ambiguous_ownership_state"
REASON_CONFLICTING_IDENTITY = "conflicting_identity_evidence"
REASON_FOREIGN_REPOSITORY = "repository_binding_ambiguous"
REASON_HEARTBEAT_FRESH = "heartbeat_not_expired"
REASON_INCOMPLETE_IDENTITY = "incomplete_registry_identity"
REASON_NOT_IN_PLAN = "not_in_current_plan"
REASON_PID_ALIVE = "pid_alive"
REASON_PID_UNKNOWN = "pid_liveness_unknown"
REASON_PROTECTED_OWNER = "protected_active_workflow_owner"
REASON_ROW_CHANGED = "row_changed_since_plan"
REASON_ROW_MISSING = "registration_missing"
REASON_UNPARSABLE_HEARTBEAT = "unparsable_heartbeat"
REASON_WORKER_LIVE = "worker_live"
# --- #980 review 657 B2: affirmative identity/liveness proof ---------------
#: The registration's instance identity is not launcher-minted (``inst-…``),
#: so nothing proves which application launch this row belongs to.
REASON_UNTRUSTED_PROVENANCE = "untrusted_identity_provenance"
#: The row does not record which host its pid belongs to, or records a
#: different host than the one probing. A local pid probe cannot speak for a
#: process on another machine.
REASON_HOST_UNPROVEN = "host_binding_unproven"
#: Boot identity is missing on the row or unobtainable here, so a recorded pid
#: cannot be compared against a live pid at all.
REASON_BOOT_UNKNOWN = "boot_identity_unknown"
#: The pid is alive but belongs to a different process incarnation than the one
#: registered — reported distinctly from a plain live worker.
REASON_PID_REUSED = "pid_reuse_detected"
#: Two active registrations claim one client instance within one namespace.
REASON_INSTANCE_CONFLICT = "client_instance_conflict"
#: The immediate pre-write re-probe could not re-establish death (#980 B3).
REASON_LIVENESS_REPROBE = "liveness_reprobe_refused"
#: Instance-identity prefix minted by the trusted launcher. Kept in sync with
#: ``mcp_fleet_snapshot._TRUSTED_INSTANCE_PREFIX`` through
#: :func:`mcp_fleet_snapshot.assess_instance_identity`, which stays the single
#: authority on what "trusted" means — this module never re-implements it.
TRUSTED_INSTANCE_PREFIX = "inst-"
#: Registry columns that must carry a usable value before the eligibility
#: conjunction can even be evaluated. Absence is ambiguity, not permission.
#:
#: #980 review 657 B2 added the fencing triple. Before it, "complete identity"
#: meant only that the pre-existing columns were non-null, which a legacy
#: ``legacy-pid-…`` row satisfies trivially — so a row that proved nothing about
#: *which* process it described was retireable. The triple is what makes a
#: recorded pid interpretable: which machine it ran on, which boot of that
#: machine, and which incarnation of that pid number. A registration written
#: before these columns existed carries NULL and is therefore preserved
#: permanently, which is the intended fail-closed outcome.
REQUIRED_IDENTITY_FIELDS: tuple[str, ...] = (
"worker_identity",
"client_instance_id",
"session_id",
"generation_id",
"status",
"started_at",
"last_heartbeat_at",
"heartbeat_ttl_seconds",
"pid",
"host_id",
"boot_id",
"process_start_time",
)
#: Canonical retirement-relevant content. Ordering here is fixed and part of
#: the token contract; adding a field changes every fingerprint, so a change
#: here is a deliberate contract revision.
#:
#: Deliberately excluded: ``token_fingerprint`` (credential-adjacent, never a
#: retirement input), the four ``*_revision`` columns (revision drift is an
#: independent restart concern and is not part of the eligibility conjunction),
#: and the ``retired_*`` bookkeeping columns this feature adds.
FINGERPRINT_FIELDS: tuple[str, ...] = (
"worker_identity",
"client_name",
"client_instance_id",
"session_id",
"generation_id",
"role",
"profile",
"namespace",
"remote",
"repository_binding",
"pid",
"process_identity",
"transport",
"started_at",
"last_heartbeat_at",
"heartbeat_ttl_seconds",
"fencing_epoch",
"status",
"fleet_run_id",
"authenticated_account",
"instance_id_provenance",
# #980 review 657 B3: fencing evidence is a retirement input, so moving it
# must move the CAS token. Without these, a row whose host, boot, or
# process incarnation changed would hash identically to the row the plan
# approved.
"host_id",
"boot_id",
"process_start_time",
)
_FINGERPRINT_VERSION = "registryfp-v1"
_CANDIDATE_VERSION = "candidatefp-v1"
_UNIT = "\x1f"
_RECORD = "\x1e"
def _canon(value: Any) -> str:
"""Stable text for one field value, independent of Python/SQLite typing.
``900`` and ``900.0`` are the same TTL and must hash the same; a value that
round-trips through SQLite as REAL must not produce a different token than
the same value supplied by a caller as ``int``.
"""
if value is None:
return ""
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, float):
if value != value or value in (float("inf"), float("-inf")):
return repr(value)
if value.is_integer():
return str(int(value))
return repr(value)
if isinstance(value, int):
return str(value)
return str(value)
def _serialize_row(row: Mapping[str, Any]) -> str:
return _UNIT.join(f"{name}={_canon(row.get(name))}" for name in FINGERPRINT_FIELDS)
def _digest(version: str, serialized: Sequence[str], prefix: str) -> str:
ordered = sorted(serialized)
material = _RECORD.join([version, str(len(ordered)), *ordered])
return f"{prefix}-{hashlib.sha256(material.encode('utf-8')).hexdigest()[:32]}"
def registry_fingerprint(rows: Iterable[Mapping[str, Any]]) -> str:
"""Content-derived compare-and-swap token for the worker registry (#980).
Derived exclusively from :data:`FINGERPRINT_FIELDS` across every row. It
contains no ``snapshot_at``, wall-clock, request, or report-generation
time, so two observations of an unchanged registry always agree, and the
serialized rows are sorted so iteration order cannot perturb the digest.
"""
return _digest(
_FINGERPRINT_VERSION,
[_serialize_row(row) for row in rows],
"registryfp",
)
def candidate_fingerprint(candidate_rows: Iterable[Mapping[str, Any]]) -> str:
"""Exact-candidate-set token over the selected rows' canonical content.
A matching :func:`registry_fingerprint` already implies these rows are
unchanged; this second token additionally pins *which* rows the operator
approved, so an apply can never widen or narrow the approved set.
"""
return _digest(
_CANDIDATE_VERSION,
[_serialize_row(row) for row in candidate_rows],
"candidatefp",
)
def _probe_pid(
pid: Any, pid_alive_probe: Callable[[int | None], bool | None] | None
) -> bool | None:
if pid_alive_probe is None or pid is None:
return None
try:
result = pid_alive_probe(pid)
except Exception:
return None
return None if result is None else bool(result)
def _reuse_detected(
row: Mapping[str, Any],
live_start_time: str | None,
current_host_id: str | None,
current_boot_id: str | None,
) -> bool:
"""Is the pid occupied by a *different* incarnation than the one recorded?
Only meaningful when the recorded pid is comparable to the live one — same
machine, same boot. Across hosts or boots the number is unrelated by
construction and reuse is not the interesting question.
"""
recorded_host = (row.get("host_id") or "").strip()
recorded_boot = (row.get("boot_id") or "").strip()
if not current_host_id or recorded_host != current_host_id:
return False
if not current_boot_id or recorded_boot != current_boot_id:
return False
recorded_start = (row.get("process_start_time") or "").strip()
return bool(live_start_time) and live_start_time != recorded_start
def _instance_key(row: Mapping[str, Any]) -> tuple[str, str] | None:
"""The (instance, namespace) pair #978 requires to be unique among live rows."""
instance = _canon(row.get("client_instance_id"))
namespace = _canon(row.get("namespace"))
if not instance or not namespace:
return None
return (instance, namespace)
def _probe_start_time(
pid: Any, start_time_probe: Callable[[Any], str | None] | None
) -> str | None:
if start_time_probe is None or pid is None:
return None
try:
return start_time_probe(pid)
except Exception:
return None
def _evidence(
row: Mapping[str, Any],
snapshot_row: Mapping[str, Any],
pid_alive: bool | None,
) -> dict[str, Any]:
liveness = snapshot_row.get("liveness") or {}
return {
"worker_identity": row.get("worker_identity"),
"client_type": snapshot_row.get("client_type"),
"client_instance_id": row.get("client_instance_id"),
"fleet_run_id": row.get("fleet_run_id"),
"namespace": row.get("namespace"),
"profile": row.get("profile"),
"declared_role": row.get("role"),
"session_id": row.get("session_id"),
"generation_id": row.get("generation_id"),
"fencing_epoch": row.get("fencing_epoch"),
"process_identity": snapshot_row.get("process_identity"),
"pid": row.get("pid"),
"pid_alive": pid_alive,
"host_id": row.get("host_id"),
"boot_id": row.get("boot_id"),
"process_start_time": row.get("process_start_time"),
"repository_binding": row.get("repository_binding"),
"foreign_repository": bool(snapshot_row.get("foreign_repository")),
"status": row.get("status"),
"started_at": row.get("started_at"),
"last_heartbeat_at": row.get("last_heartbeat_at"),
"heartbeat_ttl_seconds": row.get("heartbeat_ttl_seconds"),
"heartbeat_age_seconds": liveness.get("heartbeat_age_seconds"),
"heartbeat_fresh": liveness.get("heartbeat_fresh"),
"live": bool(snapshot_row.get("live")),
"ownership_state": snapshot_row.get("ownership_state"),
"instance_id_provenance": snapshot_row.get("instance_id_provenance"),
"instance_identity_trusted": bool(
snapshot_row.get("instance_identity_trusted")
),
}
def _conflict_keys(
row: Mapping[str, Any], snapshot_row: Mapping[str, Any]
) -> list[tuple[str, str]]:
keys: list[tuple[str, str]] = []
for name, value in (
("session_id", row.get("session_id")),
("generation_id", row.get("generation_id")),
("process_identity", snapshot_row.get("process_identity")),
("pid", row.get("pid")),
):
text = _canon(value)
if text:
keys.append((name, text))
return keys
def external_state_fingerprint(
leases: Iterable[Mapping[str, Any]],
*,
liveness: Iterable[tuple[Any, Any]] = (),
) -> str:
"""Version token over the external state a retirement decision consumed.
#980 review 657 B3: ``BEGIN IMMEDIATE`` on the worker registry does not
cover the control-plane lease table or the OS process table, so those
inputs can move while the transaction is open. This token lets the
transaction detect that movement: it is captured before the authoritative
read and re-compared immediately before every guarded write, and any
difference aborts rather than retiring against evidence that has changed.
Only ownership-relevant lease fields participate, so unrelated churn (a
heartbeat timestamp advancing on an unrelated lease) does not cause
spurious aborts, while an acquire, release, or owner change always does.
"""
lease_units: list[str] = []
for lease in leases:
lease_units.append(
_UNIT.join(
f"{name}={_canon(lease.get(name))}"
for name in (
"lease_id",
"role",
"target",
"status",
"session_id",
"owner_session_id",
"owner_pid",
"session_pid",
"generation",
)
)
)
for pid, alive in liveness:
lease_units.append(f"liveness{_UNIT}pid={_canon(pid)}{_UNIT}alive={_canon(alive)}")
return _digest("externalfp-v1", lease_units, "externalfp")
def assess_retirement_identity_proof(
row: Mapping[str, Any],
snapshot_row: Mapping[str, Any],
*,
current_host_id: str | None,
current_boot_id: str | None,
live_start_time: str | None,
pid_alive: bool | None,
) -> tuple[str, str] | None:
"""Affirmative proof that this row names one specific, now-dead process.
Returns ``None`` when the proof holds, or ``(reason_code, detail)`` naming
the first thing that could not be established. #980 review 657 B2: absence
of evidence is never read as staleness, so every branch here refuses on
*missing* information exactly as firmly as on contradictory information.
The proof has two independent halves and needs both:
* **Attribution** — a launcher-minted ``inst-…`` instance identity, so the
row is known to belong to one specific application launch rather than
having been inferred from pid proximity.
* **Fencing** — the row's host matches the host doing the probing, boot
identity is known on both sides, and the recorded process incarnation
agrees with whatever currently occupies that pid number.
Requiring trusted attribution means pre-#978 ``legacy-pid-…`` rows are
preserved permanently. That is deliberate. The reviewer specifically
rejected the argument that legacy rows "would remain forever" as grounds
for a weaker proof, and #980 lists backfilling trusted identity for legacy
workers as a non-goal — so those rows are retired only after their worker
re-registers under a trusted identity, never on weaker evidence.
"""
if not snapshot_row.get("instance_identity_trusted"):
return (
REASON_UNTRUSTED_PROVENANCE,
"client_instance_id "
f"{row.get('client_instance_id')!r} is not launcher-minted "
f"({snapshot_row.get('instance_id_provenance')!r}); nothing proves "
"which application launch this registration belongs to",
)
recorded_host = (row.get("host_id") or "").strip()
if not current_host_id:
return (
REASON_HOST_UNPROVEN,
"this process cannot establish its own host identity, so a local "
"pid probe cannot be attributed to any machine",
)
if recorded_host != current_host_id:
return (
REASON_HOST_UNPROVEN,
f"registration is bound to host {recorded_host!r} but retirement is "
f"running on {current_host_id!r}; a local pid probe says nothing "
"about a process on another machine",
)
recorded_boot = (row.get("boot_id") or "").strip()
if not current_boot_id:
return (
REASON_BOOT_UNKNOWN,
"the current boot identity could not be determined, so a recorded "
"pid cannot be compared against a live pid",
)
recorded_start = (row.get("process_start_time") or "").strip()
if recorded_boot != current_boot_id:
# A different boot is the strongest possible death evidence: every pid
# from a previous boot is gone, and pid numbers restart, so whatever
# occupies this number now is unrelated by construction.
return None
# Same boot: the pid number is directly comparable, so the recorded
# incarnation must still agree with whatever holds that number.
if pid_alive and live_start_time and live_start_time != recorded_start:
return (
REASON_PID_REUSED,
f"pid {row.get('pid')!r} is alive but started at "
f"{live_start_time!r}, not the registered {recorded_start!r}; the "
"number was reused by an unrelated process and this registration's "
"own liveness is therefore unproven",
)
if pid_alive:
return (
REASON_PID_ALIVE,
f"recorded pid {row.get('pid')!r} is still running on this host and "
"boot",
)
return None
def _missing_identity_fields(row: Mapping[str, Any]) -> list[str]:
missing: list[str] = []
for name in REQUIRED_IDENTITY_FIELDS:
value = row.get(name)
if value is None or (isinstance(value, str) and not value.strip()):
missing.append(name)
return missing
def plan_stale_worker_retirement(
rows: Iterable[Mapping[str, Any]],
*,
now: datetime | None = None,
pid_alive_probe: Callable[[int | None], bool | None] | None = None,
canonical_repository: str | None = None,
protected_worker_identities: Iterable[str] | None = None,
protected_session_ids: Iterable[str] | None = None,
protected_pids: Iterable[Any] | None = None,
current_host_id: str | None = None,
current_boot_id: str | None = None,
start_time_probe: Callable[[Any], str | None] | None = None,
) -> dict[str, Any]:
"""Decide, without mutating anything, which registrations may be retired.
Every row lands in exactly one of ``candidates`` (eligible) or
``preserved`` (with the reason code that stopped it), so the output
explains the whole registry rather than only the interesting part.
"""
all_rows = [dict(row) for row in rows]
protected_ids = {str(w) for w in (protected_worker_identities or []) if w}
protected_sessions = {str(s) for s in (protected_session_ids or []) if s}
protected_pid_set = {_canon(p) for p in (protected_pids or []) if p is not None}
snapshots: dict[int, dict[str, Any]] = {}
pid_alive_by_index: dict[int, bool | None] = {}
start_time_by_index: dict[int, str | None] = {}
for index, row in enumerate(all_rows):
pid_alive = _probe_pid(row.get("pid"), pid_alive_probe)
pid_alive_by_index[index] = pid_alive
start_time_by_index[index] = _probe_start_time(
row.get("pid"), start_time_probe
)
snapshots[index] = fleet.build_worker_snapshot_row(
row,
now=now,
pid_alive_probe=(lambda _pid, _value=pid_alive: _value),
canonical_repository=canonical_repository,
)
# Identity evidence owned by a worker that is live, or whose liveness could
# not be established, is ambiguous: anything sharing it is preserved.
ambiguous_keys: set[tuple[str, str]] = set()
identity_counts: dict[str, int] = {}
for index, row in enumerate(all_rows):
identity = _canon(row.get("worker_identity"))
if identity:
identity_counts[identity] = identity_counts.get(identity, 0) + 1
snapshot_row = snapshots[index]
liveness = snapshot_row.get("liveness") or {}
unresolved = (
pid_alive_by_index[index] is None
or liveness.get("heartbeat_fresh") is None
)
if snapshot_row.get("live") or (
str(row.get("status") or "") == mwi.STATUS_ACTIVE and unresolved
):
ambiguous_keys.update(_conflict_keys(row, snapshot_row))
# Two active registrations claiming one (client_instance_id, namespace)
# violate the #978 uniqueness invariant — but only when one of them might
# still be running. Several *dead* rows accumulating on one slot across
# restarts is ordinary history and every one of them is safely retirable;
# a slot shared with a live or unprobeable worker is genuinely ambiguous,
# because which row that process belongs to cannot be settled from the
# registry alone.
#
# The key is deliberately the (instance, namespace) pair, not the instance
# alone: one legitimate cohort is exactly one instance spread across
# distinct namespaces, so keying on the instance would make every cohort
# look self-conflicting and preserve the whole fleet forever.
instance_members: dict[tuple[str, str], list[int]] = {}
for index, row in enumerate(all_rows):
if str(row.get("status") or "") != mwi.STATUS_ACTIVE:
continue
key = _instance_key(row)
if key is None:
continue
instance_members.setdefault(key, []).append(index)
instance_conflicts: dict[tuple[str, str], bool] = {}
for key, members in instance_members.items():
if len(members) < 2:
continue
contested = any(
snapshots[i].get("live") or pid_alive_by_index[i] is None
for i in members
)
if contested:
instance_conflicts[key] = True
candidates: list[dict[str, Any]] = []
candidate_rows: list[Mapping[str, Any]] = []
preserved: list[dict[str, Any]] = []
for index, row in enumerate(all_rows):
snapshot_row = snapshots[index]
pid_alive = pid_alive_by_index[index]
liveness = snapshot_row.get("liveness") or {}
evidence = _evidence(row, snapshot_row, pid_alive)
blocked: tuple[str, str] | None = None
identity = _canon(row.get("worker_identity"))
missing = _missing_identity_fields(row)
shared = sorted(
f"{name}={value}"
for name, value in _conflict_keys(row, snapshot_row)
if (name, value) in ambiguous_keys
)
binding = (row.get("repository_binding") or "").strip()
protected_hits: list[str] = []
if identity and identity in protected_ids:
protected_hits.append(f"worker_identity={identity}")
if _canon(row.get("session_id")) in protected_sessions:
protected_hits.append(f"session_id={_canon(row.get('session_id'))}")
if _canon(row.get("pid")) in protected_pid_set:
protected_hits.append(f"pid={_canon(row.get('pid'))}")
if identity and identity_counts.get(identity, 0) > 1:
blocked = (
REASON_CONFLICTING_IDENTITY,
f"worker identity {identity!r} appears on more than one registry row",
)
elif str(row.get("status") or "") != mwi.STATUS_ACTIVE:
blocked = (
REASON_ALREADY_TERMINAL,
f"registration status is {row.get('status')!r}; nothing to retire",
)
elif missing:
blocked = (
REASON_INCOMPLETE_IDENTITY,
"registry row is missing field(s) the retirement conjunction "
f"reads: {missing}",
)
elif mwi._parse_ts(row.get("last_heartbeat_at")) is None:
blocked = (
REASON_UNPARSABLE_HEARTBEAT,
"last_heartbeat_at is not a parsable UTC stamp; liveness is unknown",
)
elif snapshot_row.get("live"):
blocked = (REASON_WORKER_LIVE, "worker is live and must not be retired")
elif pid_alive is None:
blocked = (
REASON_PID_UNKNOWN,
f"pid {row.get('pid')!r} could not be probed; liveness is unproven",
)
elif pid_alive and _reuse_detected(
row, start_time_by_index[index], current_host_id, current_boot_id
):
# Reported before the generic live-pid branch so the operator sees
# *why* the number is occupied: an unrelated process inherited it,
# which means this registration's own liveness is unproven rather
# than positively established.
blocked = (
REASON_PID_REUSED,
f"pid {row.get('pid')!r} is alive but started at "
f"{start_time_by_index[index]!r}, not the registered "
f"{row.get('process_start_time')!r}; the number was reused",
)
elif pid_alive:
blocked = (
REASON_PID_ALIVE,
f"recorded pid {row.get('pid')!r} is still running",
)
elif liveness.get("heartbeat_fresh") is not False:
blocked = (
REASON_HEARTBEAT_FRESH,
"heartbeat has not expired under the canonical TTL policy",
)
elif snapshot_row.get("ownership_state") != "stale":
blocked = (
REASON_AMBIGUOUS_OWNERSHIP,
"ownership_state is "
f"{snapshot_row.get('ownership_state')!r}, not 'stale'",
)
elif not binding or snapshot_row.get("foreign_repository"):
blocked = (
REASON_FOREIGN_REPOSITORY,
"repository binding is absent or does not match the canonical "
"repository; retirement scope is ambiguous",
)
elif shared:
blocked = (
REASON_CONFLICTING_IDENTITY,
"identity evidence is shared with a live or unprobeable worker: "
f"{shared}",
)
elif protected_hits:
blocked = (
REASON_PROTECTED_OWNER,
"worker still owns active workflow state requiring separate "
f"reconciliation: {protected_hits}",
)
elif instance_conflicts.get(_instance_key(row)):
blocked = (
REASON_INSTANCE_CONFLICT,
"another active registration claims client_instance_id "
f"{row.get('client_instance_id')!r} in namespace "
f"{row.get('namespace')!r}; instance ownership is ambiguous",
)
else:
# Affirmative identity + fencing proof runs last: everything above
# establishes the row is *inert*, and this establishes it is
# unambiguously *this* worker (#980 review 657 B2).
blocked = assess_retirement_identity_proof(
row,
snapshot_row,
current_host_id=current_host_id,
current_boot_id=current_boot_id,
live_start_time=start_time_by_index[index],
pid_alive=pid_alive,
)
if blocked is not None:
preserved.append(
{
"worker_identity": row.get("worker_identity"),
"reason_code": blocked[0],
"detail": blocked[1],
"evidence": evidence,
}
)
continue
candidates.append(
{
"worker_identity": row.get("worker_identity"),
"reason_code": REASON_ELIGIBLE,
"detail": (
"dead pid, expired heartbeat, stale ownership, unambiguous "
"identity, canonical repository binding, no active workflow "
"ownership"
),
"evidence": evidence,
}
)
candidate_rows.append(row)
counts: dict[str, int] = {}
for entry in preserved:
counts[entry["reason_code"]] = counts.get(entry["reason_code"], 0) + 1
return {
"success": True,
"read_only": True,
"mutation_performed": False,
"outcome": OUTCOME_PLANNED,
"registry_fingerprint": registry_fingerprint(all_rows),
"candidate_fingerprint": candidate_fingerprint(candidate_rows),
"assessed_count": len(all_rows),
"candidate_count": len(candidates),
"preserved_count": len(preserved),
"candidates": candidates,
"candidate_worker_identities": [c["worker_identity"] for c in candidates],
"preserved": preserved,
"preserved_reason_counts": counts,
"protected_inputs": {
"worker_identities": sorted(protected_ids),
"session_ids": sorted(protected_sessions),
"pids": sorted(protected_pid_set),
},
"canonical_repository": canonical_repository,
"fencing_context": {
"current_host_id": current_host_id,
"current_boot_id": current_boot_id,
"start_time_probe_available": start_time_probe is not None,
},
}
def summarize_plan(plan: Mapping[str, Any]) -> dict[str, Any]:
"""Compact, log-safe view of a plan or apply result."""
return {
"outcome": plan.get("outcome"),
"registry_fingerprint": plan.get("registry_fingerprint"),
"candidate_fingerprint": plan.get("candidate_fingerprint"),
"assessed_count": plan.get("assessed_count"),
"candidate_count": plan.get("candidate_count"),
"retired_count": plan.get("retired_count"),
"preserved_count": plan.get("preserved_count"),
"mutation_performed": plan.get("mutation_performed"),
}
-4
View File
@@ -84,10 +84,6 @@ SANCTIONED_NAMESPACES = frozenset(
INSTANCE_ID_PROVENANCE_TRUSTED = "trusted_launcher" INSTANCE_ID_PROVENANCE_TRUSTED = "trusted_launcher"
INSTANCE_ID_PROVENANCE_LEGACY = "legacy_incomplete" INSTANCE_ID_PROVENANCE_LEGACY = "legacy_incomplete"
INSTANCE_ID_PROVENANCE_MISSING = "missing" INSTANCE_ID_PROVENANCE_MISSING = "missing"
#: Well-formed ``inst-…`` identity presented without the launcher's own
#: provenance seal (#985). The format alone is public and reproducible, so an
#: unsealed value is treated as manually asserted trust and fails closed.
INSTANCE_ID_PROVENANCE_UNSEALED = "unsealed_launcher"
CLIENT_INSTANCE_ENV = "GITEA_MCP_CLIENT_INSTANCE" CLIENT_INSTANCE_ENV = "GITEA_MCP_CLIENT_INSTANCE"
FLEET_RUN_ENV = "GITEA_MCP_FLEET_RUN_ID" FLEET_RUN_ENV = "GITEA_MCP_FLEET_RUN_ID"
+205
View File
@@ -0,0 +1,205 @@
"""Host, boot, and process-start fencing evidence for retirement safety (#980).
Review 657 B2/B3 established that a local ``os.kill(pid, 0)`` probe is not, on
its own, evidence that a *particular registered worker* is gone:
* **Host ambiguity.** A registry row written on host A records pid 1234. Probing
pid 1234 on host B answers a question nobody asked. "Not running here" is not
"not running".
* **PID reuse.** Pid 1234 may be alive and belong to an unrelated process that
the kernel handed the number to after the original exited. The naive probe
reads that as "the worker is live" (safe, over-preserving) — but the converse
matters too: evidence captured about pid 1234 at time T must not be honoured
at time T+n if the process behind that number changed in between.
* **Boot boundaries.** Every pid from a previous boot is conclusively gone, and
pid numbers restart, so a recorded pid is only comparable to a live pid when
both belong to the same boot.
This module supplies the three pieces of evidence that turn a bare pid into a
statement about one specific process:
``host_id``
Which machine the pid belongs to.
``boot_id``
Which boot of that machine the pid belongs to. Pids are only comparable
within a single boot.
``process_start_time``
Which *incarnation* of that pid number. Two processes on the same host and
boot sharing a pid number cannot share a start time, so comparing start
times defeats reuse.
Every probe here is read-only, never raises, and returns ``None`` when the
evidence cannot be established. ``None`` means *unknown*, and the retirement
conjunction is required to treat unknown as "preserve", never as "safe".
"""
from __future__ import annotations
import os
import platform
import subprocess
# --- Host identity --------------------------------------------------------
def current_host_id() -> str | None:
"""Stable identifier for the machine this process runs on.
Deliberately the kernel node name rather than anything network-derived: it
does not change when an interface goes down or a VPN reassigns an address,
and retirement must not become unsafe because DNS moved.
"""
try:
node = (platform.node() or "").strip()
except Exception:
return None
return node or None
# --- Boot identity --------------------------------------------------------
def _linux_boot_id() -> str | None:
try:
with open("/proc/sys/kernel/random/boot_id", encoding="utf-8") as handle:
value = handle.read().strip()
except Exception:
return None
return value or None
def _darwin_boot_id() -> str | None:
"""macOS boot identity, derived from ``kern.boottime``.
``sysctl`` prints e.g. ``{ sec = 1785400000, usec = 123456 } Wed Jul 30 ...``.
Only the integer seconds are kept: the trailing human-readable date is
locale-dependent and would make the identifier unstable across environments
for the very same boot.
"""
try:
completed = subprocess.run(
["/usr/sbin/sysctl", "-n", "kern.boottime"],
capture_output=True,
text=True,
timeout=5,
check=False,
)
except Exception:
return None
if completed.returncode != 0:
return None
text = (completed.stdout or "").strip()
marker = "sec = "
start = text.find(marker)
if start < 0:
return None
tail = text[start + len(marker) :]
digits = ""
for char in tail:
if char.isdigit():
digits += char
else:
break
return f"boot-{digits}" if digits else None
def current_boot_id() -> str | None:
"""Identifier for the current boot, or ``None`` when it cannot be proven."""
system = ""
try:
system = (platform.system() or "").strip().lower()
except Exception:
system = ""
if system == "linux":
return _linux_boot_id()
if system == "darwin":
return _darwin_boot_id()
# An unrecognised platform yields no boot evidence rather than a guess.
return None
# --- Process start time ---------------------------------------------------
def _linux_process_start_time(pid: int) -> str | None:
"""Field 22 of ``/proc/<pid>/stat`` — start time in clock ticks since boot.
The executable name in field 2 is parenthesised and may itself contain
spaces and parentheses, so the fields are located from the *last* ``)``
rather than by splitting the whole line.
"""
try:
with open(f"/proc/{pid}/stat", encoding="utf-8") as handle:
raw = handle.read()
except Exception:
return None
close = raw.rfind(")")
if close < 0:
return None
fields = raw[close + 1 :].split()
# After the ')' the next field is state (field 3), so field 22 is index 19.
if len(fields) < 20:
return None
value = fields[19].strip()
return f"ticks-{value}" if value else None
def _darwin_process_start_time(pid: int) -> str | None:
"""macOS process start, from ``ps -o lstart=``.
``lstart`` is the absolute wall-clock start of that pid's current
incarnation. Two processes reusing one pid number report different values,
which is exactly the discrimination reuse detection needs.
"""
try:
completed = subprocess.run(
["/bin/ps", "-o", "lstart=", "-p", str(int(pid))],
capture_output=True,
text=True,
timeout=5,
check=False,
)
except Exception:
return None
if completed.returncode != 0:
return None
value = " ".join((completed.stdout or "").split())
return f"lstart-{value}" if value else None
def process_start_time(pid: int | None) -> str | None:
"""Start-time token for *pid*'s current incarnation, or ``None``.
``None`` is returned both when the pid does not exist and when the platform
cannot answer. Callers must not read either case as evidence of death — a
dead pid is established by the liveness probe, and this value only ever
*withdraws* a retirement that pid-level evidence would otherwise allow.
"""
if pid is None:
return None
try:
numeric = int(pid)
except (TypeError, ValueError):
return None
if numeric <= 0:
return None
system = ""
try:
system = (platform.system() or "").strip().lower()
except Exception:
system = ""
if system == "linux":
return _linux_process_start_time(numeric)
if system == "darwin":
return _darwin_process_start_time(numeric)
return None
def current_process_fencing(pid: int | None = None) -> dict[str, str | None]:
"""The full fencing triple for *pid* (defaults to this process)."""
target = os.getpid() if pid is None else pid
return {
"host_id": current_host_id(),
"boot_id": current_boot_id(),
"process_start_time": process_start_time(target),
}
+428 -3
View File
@@ -39,7 +39,7 @@ import threading
import time import time
from contextlib import contextmanager from contextlib import contextmanager
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Any, Iterator from typing import Any, Callable, Iterator, Sequence
# --- Provenance verdicts ------------------------------------------------- # --- Provenance verdicts -------------------------------------------------
@@ -224,6 +224,13 @@ def _heartbeat_expectation_drift(
STATUS_ACTIVE = "active" STATUS_ACTIVE = "active"
STATUS_SUPERSEDED = "superseded" STATUS_SUPERSEDED = "superseded"
STATUS_RELEASED = "released" STATUS_RELEASED = "released"
#: #980 terminal state for a registration whose owning process is conclusively
#: gone. Distinct from ``released`` (the worker said goodbye) and
#: ``superseded`` (a newer generation took over): ``retired`` records that the
#: *control plane* concluded the row was a stale orphan and retired it under a
#: compare-and-swap. Like every non-active status it is not live, so a retired
#: row counts as historical rather than stale in the #978 fleet snapshot.
STATUS_RETIRED = "retired"
_TRUE_VALUES = frozenset({"1", "true", "yes", "client_managed"}) _TRUE_VALUES = frozenset({"1", "true", "yes", "client_managed"})
_FALSE_VALUES = frozenset({"0", "false", "no", "manual", "manual_launch"}) _FALSE_VALUES = frozenset({"0", "false", "no", "manual", "manual_launch"})
@@ -315,13 +322,53 @@ _SCHEMA_OPTIONAL_COLUMNS: tuple[tuple[str, str], ...] = (
("parity_revision", "TEXT"), ("parity_revision", "TEXT"),
("live_revision", "TEXT"), ("live_revision", "TEXT"),
("instance_id_provenance", "TEXT"), ("instance_id_provenance", "TEXT"),
# #980 review 657 B2/B3: fencing evidence that turns a bare pid into a
# statement about one specific process. A registration written before these
# columns existed carries NULL and can never satisfy the retirement identity
# proof, which is the intended fail-closed outcome — absence of evidence is
# not evidence of staleness.
("host_id", "TEXT"),
("boot_id", "TEXT"),
("process_start_time", "TEXT"),
# #980 retirement bookkeeping. Deliberately outside the CAS fingerprint
# field set: they record *that* a retirement happened, and are written only
# by the retirement transaction itself.
("retired_at", "TEXT"),
("retired_by", "TEXT"),
("retirement_reason", "TEXT"),
) )
def _default_process_fencing(pid: int | None) -> dict[str, str | None]:
"""Fencing triple for *pid*, degrading to unknowns rather than raising.
Imported lazily so this storage module keeps no import-time dependency on
the probe layer, and so a platform where the probes are unavailable still
registers workers — it simply records no fencing evidence, and those rows
are then permanently ineligible for retirement.
"""
try:
import mcp_process_fencing
return mcp_process_fencing.current_process_fencing(pid)
except Exception:
return {"host_id": None, "boot_id": None, "process_start_time": None}
class WorkerRegistryError(RuntimeError): class WorkerRegistryError(RuntimeError):
"""Raised for registry misuse that is a programming error, not a refusal.""" """Raised for registry misuse that is a programming error, not a refusal."""
class _ExternalStateMoved(RuntimeError):
"""External safety evidence changed inside the retirement transaction.
Raised so the surrounding ``with self._tx()`` rolls back: once lease state
or process liveness has moved, every remaining per-row decision was computed
against a world that no longer exists, so the whole attempt is abandoned
rather than partially applied.
"""
def _utc_now() -> datetime: def _utc_now() -> datetime:
return datetime.now(timezone.utc) return datetime.now(timezone.utc)
@@ -791,6 +838,9 @@ class WorkerRegistry:
parity_revision: str | None = None, parity_revision: str | None = None,
live_revision: str | None = None, live_revision: str | None = None,
instance_id_provenance: str | None = None, instance_id_provenance: str | None = None,
host_id: str | None = None,
boot_id: str | None = None,
process_start_time: str | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Atomically register one worker identity. """Atomically register one worker identity.
@@ -823,6 +873,20 @@ class WorkerRegistry:
proc_id = process_identity or ( proc_id = process_identity or (
f"pid-{int(pid)}" if pid is not None else None f"pid-{int(pid)}" if pid is not None else None
) )
# #980 B2/B3: capture the fencing triple for the pid being registered.
# A caller may supply it (tests, or a launcher that already probed);
# otherwise it is probed here, at the only moment the process is known
# to be the one that owns this registration. Any probe that cannot
# answer stores NULL, which permanently withholds retirement eligibility
# from the row rather than granting it on absent evidence.
fencing = _default_process_fencing(pid)
host_id = host_id if host_id is not None else fencing["host_id"]
boot_id = boot_id if boot_id is not None else fencing["boot_id"]
process_start_time = (
process_start_time
if process_start_time is not None
else fencing["process_start_time"]
)
with self._tx() as conn: with self._tx() as conn:
existing = conn.execute( existing = conn.execute(
"SELECT * FROM worker_registrations WHERE worker_identity = ?", "SELECT * FROM worker_registrations WHERE worker_identity = ?",
@@ -905,8 +969,9 @@ class WorkerRegistry:
fencing_epoch, status, fencing_epoch, status,
fleet_run_id, authenticated_account, process_identity, fleet_run_id, authenticated_account, process_identity,
startup_revision, loaded_revision, parity_revision, startup_revision, loaded_revision, parity_revision,
live_revision, instance_id_provenance live_revision, instance_id_provenance,
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) host_id, boot_id, process_start_time
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
""", """,
( (
worker_identity, worker_identity,
@@ -935,6 +1000,9 @@ class WorkerRegistry:
(parity_revision or "").strip() or None, (parity_revision or "").strip() or None,
(live_revision or "").strip() or None, (live_revision or "").strip() or None,
(instance_id_provenance or "").strip() or None, (instance_id_provenance or "").strip() or None,
(host_id or "").strip() or None,
(boot_id or "").strip() or None,
(process_start_time or "").strip() or None,
), ),
) )
row = conn.execute( row = conn.execute(
@@ -1217,6 +1285,363 @@ class WorkerRegistry:
"reasons": [], "reasons": [],
} }
def retire_stale_workers(
self,
*,
expected_registry_fingerprint: str,
expected_candidate_fingerprint: str,
worker_identities: Sequence[str],
fingerprint_fn: Callable[[list[dict[str, Any]]], str],
plan_fn: Callable[[list[dict[str, Any]]], dict[str, Any]],
retired_by: str | None = None,
retirement_reason: str = "",
now: datetime | None = None,
external_fence_fn: Callable[[], str] | None = None,
liveness_fn: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
) -> dict[str, Any]:
"""Compare-and-swap retirement of conclusively stale registrations (#980).
The whole decision happens inside one ``BEGIN IMMEDIATE`` transaction:
the authoritative rows are re-read, the stable registry fingerprint is
recomputed from *those* rows, the eligibility plan is recomputed from
*those* rows, and only then are the approved targets retired — each
with a per-row guarded ``UPDATE`` that also asserts the row's identity,
liveness, and ownership columns are byte-identical to what the
revalidation just read. There is no window in which a safety check and
its matching write are separated by another statement, so a worker that
comes back to life, changes ownership, or is retired concurrently
cannot be deleted on the strength of a stale observation.
``fingerprint_fn`` and ``plan_fn`` are injected rather than imported so
the storage layer never depends on the decision layer; production wires
in :func:`mcp_fleet_retirement.registry_fingerprint` and
:func:`mcp_fleet_retirement.plan_stale_worker_retirement`, which is
exactly what the plan surface used.
Any exception — including a failure to commit — rolls the transaction
back and is reported as ``transaction_failed`` with zero retirements
and ``mutation_performed`` false, so a partial write can never be
reported as success.
**External state (#980 review 657 B3).** ``BEGIN IMMEDIATE`` locks this
registry and nothing else, so two critical inputs live outside the
transaction's isolation domain: the workflow-lease table in a separate
control-plane database, and OS process liveness. Re-reading them once
during revalidation is not enough — the per-target loop takes time, so a
lease acquired (or a pid revived) after ``plan_fn`` returned but before
*this* row's ``UPDATE`` would go unnoticed, and the registry-column guard
cannot catch it because no registry column changed.
Two mechanisms close that gap, both applied per target and immediately
before its own write:
``external_fence_fn``
A version token over all external state the decision consumed. It is
captured inside the transaction before revalidation and re-read
before every guarded ``UPDATE``; any movement aborts the whole
transaction rather than retiring against evidence that has changed.
``liveness_fn``
A per-row re-probe of process liveness and fencing identity (host,
boot, start time). It runs immediately before the row's write and
must affirmatively re-establish that this exact process is gone.
Both default to ``None`` only so the storage layer stays independent of
the decision and control-plane layers; production always supplies them,
and a caller that omits ``liveness_fn`` gets no retirement at all rather
than an unfenced one.
"""
requested = [str(w) for w in (worker_identities or []) if str(w).strip()]
stamp = _ts(now or _utc_now())
result: dict[str, Any] | None = None
def _base(outcome: str) -> dict[str, Any]:
return {
"success": True,
"outcome": outcome,
"mutation_performed": False,
"retired": [],
"retired_count": 0,
"preserved": [],
"preserved_count": 0,
"requested_count": len(requested),
"expected_registry_fingerprint": expected_registry_fingerprint,
"expected_candidate_fingerprint": expected_candidate_fingerprint,
"acting_identity": retired_by,
"reasons": [],
}
if not requested:
outcome = _base("nothing_requested")
outcome["reasons"] = [
"no worker identities were supplied; nothing to retire"
]
return outcome
try:
with self._tx() as conn:
# Captured *after* BEGIN IMMEDIATE and *before* the
# authoritative read, so every later comparison is against the
# external state this decision was actually built on.
fence_at_plan = external_fence_fn() if external_fence_fn else None
rows = [
self._row_to_record(r)
for r in conn.execute(
"SELECT * FROM worker_registrations"
).fetchall()
]
by_identity = {
str(row.get("worker_identity")): row for row in rows
}
current_registry_fingerprint = fingerprint_fn(rows)
if current_registry_fingerprint != expected_registry_fingerprint:
already = [
wid
for wid in requested
if str(
(by_identity.get(wid) or {}).get("status") or ""
)
== STATUS_RETIRED
]
idempotent = len(already) == len(requested)
result = _base(
"already_retired" if idempotent else "registry_revision_moved"
)
result["idempotent"] = idempotent
result["current_registry_fingerprint"] = (
current_registry_fingerprint
)
result["reasons"] = [
"the worker registry changed between plan and apply; "
"retiring zero workers"
if not idempotent
else "every requested registration is already retired; "
"safe no-op"
]
result["preserved"] = [
{
"worker_identity": wid,
"reason_code": "registry_revision_moved",
"detail": (
"aborted before any retirement: registry "
"fingerprint moved"
),
}
for wid in requested
]
result["preserved_count"] = len(requested)
return result
fresh_plan = plan_fn(rows)
current_candidate_fingerprint = fresh_plan.get(
"candidate_fingerprint"
)
if current_candidate_fingerprint != expected_candidate_fingerprint:
result = _base("candidate_set_moved")
result["current_registry_fingerprint"] = (
current_registry_fingerprint
)
result["current_candidate_fingerprint"] = (
current_candidate_fingerprint
)
result["reasons"] = [
"the retirement candidate set changed between plan and "
"apply; retiring zero workers"
]
result["preserved"] = [
{
"worker_identity": wid,
"reason_code": "candidate_set_moved",
"detail": (
"aborted before any retirement: candidate "
"fingerprint moved"
),
}
for wid in requested
]
result["preserved_count"] = len(requested)
return result
eligible = {
str(c.get("worker_identity")): c
for c in fresh_plan.get("candidates") or []
}
preserved_index = {
str(p.get("worker_identity")): p
for p in fresh_plan.get("preserved") or []
}
retired: list[dict[str, Any]] = []
preserved: list[dict[str, Any]] = []
for wid in requested:
row = by_identity.get(wid)
if row is None:
preserved.append(
{
"worker_identity": wid,
"reason_code": "registration_missing",
"detail": "no registration row with this identity",
}
)
continue
if wid not in eligible:
blocked = preserved_index.get(wid) or {}
preserved.append(
{
"worker_identity": wid,
"reason_code": blocked.get("reason_code")
or "not_in_current_plan",
"detail": blocked.get("detail")
or (
"revalidation immediately before retirement "
"no longer finds this worker eligible"
),
"evidence": blocked.get("evidence"),
}
)
continue
# --- External-state fence, immediately before this write ---
#
# Lease state and OS liveness are outside this transaction,
# so they are re-checked here rather than trusted from
# revalidation. Movement aborts the whole transaction: a
# changed world invalidates every remaining decision, not
# only this row's.
if external_fence_fn is not None:
fence_now = external_fence_fn()
if fence_now != fence_at_plan:
raise _ExternalStateMoved(
"external safety state (workflow leases or "
"process liveness) changed inside the retirement "
"transaction; rolling back and retiring nothing"
)
if liveness_fn is None:
preserved.append(
{
"worker_identity": wid,
"reason_code": "liveness_reprobe_unavailable",
"detail": (
"no immediate pre-write liveness re-probe was "
"supplied; refusing to retire on revalidation "
"evidence alone (fail closed)"
),
}
)
continue
verdict = liveness_fn(dict(row))
if not verdict.get("safe"):
preserved.append(
{
"worker_identity": wid,
"reason_code": verdict.get("reason_code")
or "liveness_reprobe_refused",
"detail": verdict.get("detail")
or (
"immediate pre-write re-probe could not "
"re-establish that this process is gone"
),
"evidence": verdict.get("evidence"),
}
)
continue
cursor = conn.execute(
"UPDATE worker_registrations "
"SET status = ?, retired_at = ?, retired_by = ?, "
" retirement_reason = ? "
"WHERE worker_identity = ? "
" AND status = ? "
" AND last_heartbeat_at = ? "
" AND generation_id = ? "
" AND session_id = ? "
" AND fencing_epoch = ? "
" AND IFNULL(pid, -1) = IFNULL(?, -1) "
" AND IFNULL(host_id, '') = IFNULL(?, '') "
" AND IFNULL(boot_id, '') = IFNULL(?, '') "
" AND IFNULL(process_start_time, '') = IFNULL(?, '')",
(
STATUS_RETIRED,
stamp,
retired_by,
retirement_reason
or (eligible[wid].get("reason_code") or ""),
wid,
STATUS_ACTIVE,
row.get("last_heartbeat_at"),
row.get("generation_id"),
row.get("session_id"),
row.get("fencing_epoch"),
row.get("pid"),
row.get("host_id"),
row.get("boot_id"),
row.get("process_start_time"),
),
)
if cursor.rowcount == 1:
retired.append(
{
"worker_identity": wid,
"reason_code": eligible[wid].get("reason_code"),
"retired_at": stamp,
"retired_by": retired_by,
"evidence": eligible[wid].get("evidence"),
}
)
else:
preserved.append(
{
"worker_identity": wid,
"reason_code": "row_changed_since_plan",
"detail": (
"guarded update matched no row; the "
"registration changed inside the retirement "
"transaction"
),
}
)
result = _base("applied")
result["mutation_performed"] = bool(retired)
result["retired"] = retired
result["retired_count"] = len(retired)
result["preserved"] = preserved
result["preserved_count"] = len(preserved)
result["current_registry_fingerprint"] = (
current_registry_fingerprint
)
result["current_candidate_fingerprint"] = (
current_candidate_fingerprint
)
result["retired_at"] = stamp if retired else None
result["external_fence"] = fence_at_plan
except Exception as exc: # rolled back by _tx; report, never half-claim
moved = isinstance(exc, _ExternalStateMoved)
failure = _base("external_state_moved" if moved else "transaction_failed")
failure["success"] = False
failure["reasons"] = [
str(exc)
if moved
else "retirement transaction failed and was rolled back; zero "
f"registrations were retired: {type(exc).__name__}: {exc}"
]
failure["preserved"] = [
{
"worker_identity": wid,
"reason_code": (
"external_state_moved" if moved else "transaction_failed"
),
"detail": "transaction rolled back before any commit",
}
for wid in requested
]
failure["preserved_count"] = len(requested)
return failure
return result
def _public_record(record: dict[str, Any]) -> dict[str, Any]: def _public_record(record: dict[str, Any]) -> dict[str, Any]:
"""Registry row minus anything that should not travel to an LLM surface.""" """Registry row minus anything that should not travel to an LLM surface."""
+9 -146
View File
@@ -1,16 +1,9 @@
"""Root checkout guard (#475). """Root checkout guard (#475).
The project root checkout is the stable control checkout on its integration The project root checkout is the stable control checkout on master/prgs/master.
branch. Author/reviewer/merge flows must fail closed when the control checkout Author/reviewer/merge flows must fail closed when the control checkout is
is contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of the contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
tracking integration ref). Isolated ``branches/...`` worktrees remain allowed. prgs/master). Isolated ``branches/...`` worktrees remain allowed.
The tracking ref is *derived per repository* (#983) rather than assumed to be
``prgs/master``: a namespace bound to another repository — say remote ``MDCPS``
on branch ``dev`` — is gated against ``refs/remotes/MDCPS/dev``. Derivation is
delegated to :mod:`canonical_repository_root`, the authoritative
repository/context resolver, so gating and parity reporting share one resolved
target instead of maintaining two disagreeing hardcoded defaults.
""" """
from __future__ import annotations from __future__ import annotations
@@ -18,7 +11,6 @@ from __future__ import annotations
import os import os
import subprocess import subprocess
import canonical_repository_root
from author_mutation_worktree import is_path_under_branches from author_mutation_worktree import is_path_under_branches
from reviewer_worktree import parse_dirty_tracked_files from reviewer_worktree import parse_dirty_tracked_files
@@ -28,65 +20,19 @@ REMEDIATION = (
) )
BASE_BRANCHES = frozenset({"master", "main", "dev"}) BASE_BRANCHES = frozenset({"master", "main", "dev"})
# Legacy PRGS-specific probe order. Retained only for callers that pass an
# explicit ``remote_refs`` override; it is no longer the silent default, because
# inheriting it in a non-PRGS checkout compared that checkout against a ref it
# can never have (#983).
REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master") REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")
def _derive_probe_refs(root: str, explicit_remote: str | None) -> dict:
"""Derive the ordered tracking refs to probe for *root*.
Returns the derivation payload from :mod:`canonical_repository_root` plus a
``refs`` tuple, which is empty when the target is not provable.
"""
derived = canonical_repository_root.resolve_target_base_ref(
root, explicit_remote=explicit_remote
)
return {
"refs": tuple(derived.get("tracking_refs") or ()),
"remote": derived.get("remote"),
"branch": derived.get("branch"),
"source": derived.get("source"),
"proven": bool(derived.get("proven")),
"reason_code": derived.get("reason_code"),
"reasons": list(derived.get("reasons") or []),
"cached_remote_head_branch": derived.get("cached_remote_head_branch"),
"cached_remote_head_conflicts": bool(derived.get("cached_remote_head_conflicts")),
}
def resolve_remote_master_sha( def resolve_remote_master_sha(
canonical_repo_root: str, canonical_repo_root: str,
*, *,
remote_refs: tuple[str, ...] | None = None, remote_refs: tuple[str, ...] | None = None,
explicit_remote: str | None = None,
) -> str | None: ) -> str | None:
"""Return the commit SHA for the tracking integration ref when available. """Return the commit SHA for the tracking master ref when available."""
This remains the single place that turns a ref into a SHA. Returns None when
the target cannot be derived or resolved — exactly what this function already
returned when ``rev-parse`` failed. Callers that must fail closed on missing
evidence (the #749/#757 bootstrap path) surface that None as *missing
evidence*, never as "no constraint".
*explicit_remote* is a caller-supplied disambiguation and is named that way
deliberately: an internally inferred remote handed back in would suppress the
resolver's ambiguity gate (#983 B2). No production caller supplies it.
"""
root = (canonical_repo_root or "").strip() root = (canonical_repo_root or "").strip()
if not root: if not root:
return None return None
if remote_refs: for ref in remote_refs or REMOTE_MASTER_REFS:
probe: tuple[str, ...] = tuple(remote_refs)
else:
derived = _derive_probe_refs(root, explicit_remote)
if not derived["proven"]:
return None
probe = derived["refs"]
for ref in probe:
res = subprocess.run( res = subprocess.run(
["git", "-C", root, "rev-parse", "--verify", ref], ["git", "-C", root, "rev-parse", "--verify", ref],
capture_output=True, capture_output=True,
@@ -100,84 +46,6 @@ def resolve_remote_master_sha(
return None return None
def resolve_remote_master_ref_state(
canonical_repo_root: str,
*,
remote_refs: tuple[str, ...] | None = None,
explicit_remote: str | None = None,
) -> dict:
"""Resolve the tracking integration ref together with its commit SHA.
Returns ``sha``, the ``ref`` it came from, the derived ``remote`` /
``branch``, a machine-checkable ``reason_code``, and ``reasons``. ``sha`` is
None whenever the target cannot be resolved — never a fallback to some other
repository's commit.
The SHA itself is obtained through :func:`resolve_remote_master_sha` rather
than by re-probing here, so one public function stays authoritative for
ref-to-SHA resolution. An explicit *remote_refs* keeps the historical
behaviour exactly: those refs are probed in order and no derivation happens.
"""
root = (canonical_repo_root or "").strip()
state: dict = {
"sha": None,
"ref": None,
"remote": None,
"branch": None,
"source": None,
"reason_code": None,
"reasons": [],
"cached_remote_head_branch": None,
"cached_remote_head_conflicts": False,
}
if not root:
state["reasons"].append("no canonical repository root supplied (fail closed)")
return state
if remote_refs:
probe: tuple[str, ...] = tuple(remote_refs)
state["source"] = "explicit_remote_refs"
else:
derived = _derive_probe_refs(root, explicit_remote)
state["remote"] = derived["remote"]
state["branch"] = derived["branch"]
state["source"] = derived["source"]
state["cached_remote_head_branch"] = derived["cached_remote_head_branch"]
state["cached_remote_head_conflicts"] = derived["cached_remote_head_conflicts"]
if not derived["proven"]:
state["reason_code"] = derived["reason_code"]
state["reasons"] = derived["reasons"]
return state
probe = derived["refs"]
sha = resolve_remote_master_sha(root, remote_refs=probe, explicit_remote=explicit_remote)
if not sha:
state["reasons"].append(
"tracking integration ref "
f"{' / '.join(probe) if probe else '(none derived)'} does not resolve in "
f"'{root}' (fail closed)"
)
return state
state["sha"] = sha
# Name the ref that actually carries this commit. When the SHA comes from a
# test double no probe will match, so fall back to the first derived ref,
# which is the one the guard is conceptually comparing against.
for ref in probe:
res = subprocess.run(
["git", "-C", root, "rev-parse", "--verify", ref],
capture_output=True,
text=True,
check=False,
)
if res.returncode == 0 and (res.stdout or "").strip() == sha:
state["ref"] = ref
break
else:
state["ref"] = probe[0] if probe else None
return state
resolve_tracking_master_sha = resolve_remote_master_sha resolve_tracking_master_sha = resolve_remote_master_sha
@@ -191,9 +59,8 @@ def assess_root_checkout_guard(
remote_master_sha: str | None, remote_master_sha: str | None,
resolved_role: str | None = None, resolved_role: str | None = None,
actual_role: str | None = None, actual_role: str | None = None,
remote_master_ref: str | None = None,
) -> dict: ) -> dict:
"""Fail closed when the control checkout is not clean on its integration ref. """Fail closed when the control checkout is not clean master/prgs/master.
``resolved_role`` is the preflight-resolved *task* role and ``actual_role`` ``resolved_role`` is the preflight-resolved *task* role and ``actual_role``
is the *active profile* role (#540). The reconciler exemption honours either is the *active profile* role (#540). The reconciler exemption honours either
@@ -235,13 +102,9 @@ def assess_root_checkout_guard(
) )
if remote_master_sha and head_sha and head_sha != remote_master_sha: if remote_master_sha and head_sha and head_sha != remote_master_sha:
# #983: name the ref that was actually compared. Reporting a literal
# 'prgs/master' in a checkout gated against refs/remotes/MDCPS/dev sends
# the operator to inspect a ref that repository does not have.
ref_label = (remote_master_ref or "").strip() or "the tracking integration ref"
reasons.append( reasons.append(
f"control checkout HEAD does not match {ref_label} " "control checkout HEAD does not match prgs/master "
f"(HEAD {head_sha[:12]}, {ref_label} {remote_master_sha[:12]})" f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
) )
proven = not reasons proven = not reasons
+38
View File
@@ -175,6 +175,44 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.read", "permission": "gitea.read",
"role": "controller", "role": "controller",
}, },
# #980: CAS-protected retirement of conclusively stale worker
# registrations.
#
# Planning is observational and stays on ``gitea.read``: it opens no
# transaction, writes nothing, and returns only what a fleet snapshot
# already exposes to the same roles.
#
# Applying is a mutation and review 657 B1 established that ``gitea.read``
# cannot authorize it. The mutation landing in the local control-plane
# registry rather than in Gitea makes it *no less* a mutation, and sharing
# an observational permission class with plan meant any profile that could
# look could also destroy. It now requires its own permission,
# ``gitea.worker_registry.retire``, which no profile holds by default — so
# author, reviewer, merger, and ordinary read-only profiles fail closed on
# the permission itself rather than relying on the role check alone. The
# role restriction (controller/reconciler), runtime parity, cohort
# uniqueness, and the exact registry + candidate fingerprints all remain,
# and are now defence in depth behind the capability rather than a
# substitute for it.
#
# Granting the permission is a deliberate operator act in profiles.json;
# removing it from a profile immediately and completely revokes apply.
"plan_stale_worker_retirement": {
"permission": "gitea.read",
"role": "controller",
},
"gitea_plan_stale_worker_retirement": {
"permission": "gitea.read",
"role": "controller",
},
"apply_stale_worker_retirement": {
"permission": "gitea.worker_registry.retire",
"role": "controller",
},
"gitea_apply_stale_worker_retirement": {
"permission": "gitea.worker_registry.retire",
"role": "controller",
},
# #644: Phase 2 Web Console recovery tasks. # #644: Phase 2 Web Console recovery tasks.
"clear_stale_binding": { "clear_stale_binding": {
"permission": "gitea.read", "permission": "gitea.read",
@@ -1032,15 +1032,6 @@ class ClientHintsTests(unittest.TestCase):
) )
def test_client_hints_trusted_when_set(self): def test_client_hints_trusted_when_set(self):
"""Trusted requires the launcher seal as well as the format (#985).
Before #985 a well-formed ``inst-…`` value alone was accepted as
trusted. The format is public and reproducible, so anyone able to set
one environment variable could hand-write a valid-looking ID; #985
therefore additionally requires the provenance marker that only the
production launcher writes. This case now supplies the full sealed env
the launcher actually emits.
"""
import gitea_mcp_server as server import gitea_mcp_server as server
inst = fleet.generate_client_instance_id("codex", launch_nonce="t", now=NOW) inst = fleet.generate_client_instance_id("codex", launch_nonce="t", now=NOW)
@@ -1049,46 +1040,15 @@ class ClientHintsTests(unittest.TestCase):
{ {
"GITEA_MCP_CLIENT": "codex", "GITEA_MCP_CLIENT": "codex",
"GITEA_MCP_CLIENT_INSTANCE": inst, "GITEA_MCP_CLIENT_INSTANCE": inst,
"GITEA_MCP_INSTANCE_PROVENANCE": (
fleet.INSTANCE_ID_PROVENANCE_TRUSTED
),
"GITEA_MCP_FLEET_RUN_ID": "run-1", "GITEA_MCP_FLEET_RUN_ID": "run-1",
}, },
clear=False, clear=False,
): ):
hints = server._client_identity_hints() hints = server._client_identity_hints()
self.assertTrue(hints["instance_identity_trusted"]) self.assertTrue(hints["instance_identity_trusted"])
self.assertTrue(hints["instance_launcher_sealed"])
self.assertEqual(hints["client_instance_id"], inst) self.assertEqual(hints["client_instance_id"], inst)
self.assertEqual(hints["fleet_run_id"], "run-1") self.assertEqual(hints["fleet_run_id"], "run-1")
def test_client_hints_wellformed_but_unsealed_not_trusted(self):
"""#985: a valid-looking ID without the launcher seal fails closed."""
import gitea_mcp_server as server
inst = fleet.generate_client_instance_id(
"codex", launch_nonce="unsealed", now=NOW
)
with mock.patch.dict(
os.environ,
{
"GITEA_MCP_CLIENT": "codex",
"GITEA_MCP_CLIENT_INSTANCE": inst,
"GITEA_MCP_FLEET_RUN_ID": "run-1",
},
clear=False,
):
os.environ.pop("GITEA_MCP_INSTANCE_PROVENANCE", None)
hints = server._client_identity_hints()
self.assertFalse(hints["instance_identity_trusted"])
self.assertFalse(hints["instance_launcher_sealed"])
# Still reported for diagnosis rather than silently discarded.
self.assertEqual(hints["client_instance_id"], inst)
self.assertEqual(
hints["instance_id_provenance"],
fleet.INSTANCE_ID_PROVENANCE_UNSEALED,
)
def test_malformed_user_supplied_instance_not_trusted(self): def test_malformed_user_supplied_instance_not_trusted(self):
import gitea_mcp_server as server import gitea_mcp_server as server
File diff suppressed because it is too large Load Diff
-692
View File
@@ -1,692 +0,0 @@
"""Regression tests for Issue #983: derived target base ref for cross-repository checkouts.
The mutation guard previously assumed ``prgs/master`` and the parity report
assumed ``origin/master``. Any repository using neither — for example remote
``MDCPS`` on integration branch ``dev`` — could not prove base equivalence, so
every gated mutation failed closed with no reachable remedy.
Two further defects were found by review at head ``2d5d5c9d`` and are covered
here:
* **B1** — the first fix derived the integration branch from
``refs/remotes/<remote>/HEAD``. That symref is a *local cache* written at clone
time and never refreshed by fetch, so on the real Weekly Briefings target it
still named ``main`` while the checkout tracked and sat exactly on ``dev``. The
authoritative signal is the checkout's own configured upstream. The fixtures
below therefore reproduce the **disagreement**: the cache says ``main``, the
configured upstream says ``dev``, and ``dev`` must win.
* **B2** — the parity report passed its internally inferred identity remote back
into the resolver, which reads a caller-supplied remote as "the caller already
disambiguated" and skips its ambiguity gate. Gating and reporting could then
evaluate different repositories. An inferred remote is now never laundered into
explicit caller intent, and ambiguity fails closed on both sides.
These tests build hermetic git repositories on disk (no network, no fetch) and
assert the derived target end to end: identity remote, integration branch,
tracking ref, fail-closed refusals, and agreement between the mutation guard and
the parity report.
"""
from __future__ import annotations
import inspect
import os
import subprocess
import tempfile
import unittest
import anti_stomp_preflight
import canonical_repository_root as crr
import master_parity_gate
import root_checkout_guard
MDCPS_URL = "https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git"
MDCPS_SLUG = "MDCPS/WeeklyBriefings-Meta"
PRGS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
PRGS_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
def _git(root: str, *args: str) -> str:
res = subprocess.run(
["git", "-C", root, *args],
capture_output=True,
text=True,
check=True,
)
return (res.stdout or "").strip()
def _make_repo(root: str, *, remote: str | None, url: str | None) -> str:
"""Initialise a repository with one commit and an optional named remote."""
os.makedirs(root, exist_ok=True)
_git(root, "init", "--quiet")
_git(root, "config", "user.email", "[email protected]")
_git(root, "config", "user.name", "Issue983 Test")
_git(root, "config", "commit.gpgsign", "false")
with open(os.path.join(root, "seed.txt"), "w", encoding="utf-8") as fh:
fh.write("seed\n")
_git(root, "add", "seed.txt")
_git(root, "commit", "--quiet", "-m", "seed")
if remote and url:
_git(root, "remote", "add", remote, url)
return _git(root, "rev-parse", "HEAD")
def _set_remote_branch(root: str, remote: str, branch: str, sha: str) -> None:
"""Create refs/remotes/<remote>/<branch> without contacting a network."""
_git(root, "update-ref", f"refs/remotes/{remote}/{branch}", sha)
def _set_remote_head(root: str, remote: str, branch: str) -> None:
"""Write the *cached* refs/remotes/<remote>/HEAD symref.
This is the signal B1 proved untrustworthy. Fixtures use it to reproduce a
stale cache, never to manufacture the answer under test.
"""
_git(
root,
"symbolic-ref",
f"refs/remotes/{remote}/HEAD",
f"refs/remotes/{remote}/{branch}",
)
def _set_upstream(root: str, remote: str, branch: str) -> None:
"""Configure the current branch's upstream, exactly as git tracking does.
Writes ``branch.<current>.remote`` / ``branch.<current>.merge`` directly
rather than via ``--set-upstream-to`` so no ref is required to pre-exist and
no network is touched.
"""
current = _git(root, "symbolic-ref", "--short", "HEAD")
_git(root, "config", f"branch.{current}.remote", remote)
_git(root, "config", f"branch.{current}.merge", f"refs/heads/{branch}")
def _checkout_new_branch(root: str, branch: str) -> None:
_git(root, "checkout", "--quiet", "-b", branch)
def _advance(root: str, message: str) -> str:
with open(os.path.join(root, "seed.txt"), "a", encoding="utf-8") as fh:
fh.write(message + "\n")
_git(root, "add", "seed.txt")
_git(root, "commit", "--quiet", "-m", message)
return _git(root, "rev-parse", "HEAD")
class _RepoCase(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.root = os.path.join(self._tmp.name, "repo")
def _weekly_briefings_shape(self) -> str:
"""The real Weekly Briefings target, including its stale cache.
Remote ``MDCPS``; checked out on ``dev``; upstream configured to
``MDCPS/dev``; both ``dev`` and ``main`` present as tracking refs; and
``refs/remotes/MDCPS/HEAD`` still cached at ``main`` from clone time.
"""
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
_checkout_new_branch(self.root, "dev")
_set_remote_branch(self.root, "MDCPS", "dev", head)
stale = _advance(self.root, "main diverged long ago")
_set_remote_branch(self.root, "MDCPS", "main", stale)
_git(self.root, "reset", "--hard", "--quiet", head)
_set_remote_head(self.root, "MDCPS", "main") # stale clone-time cache
_set_upstream(self.root, "MDCPS", "dev") # authoritative
return head
class TestPrgsBehaviourPreserved(_RepoCase):
"""Required coverage 1: PRGS prgs/master compatibility."""
def _prgs(self) -> str:
head = _make_repo(self.root, remote="prgs", url=PRGS_URL)
_set_remote_branch(self.root, "prgs", "master", head)
_set_remote_head(self.root, "prgs", "master")
_set_upstream(self.root, "prgs", "master")
return head
def test_prgs_master_resolves_unchanged(self):
head = self._prgs()
got = crr.resolve_target_base_ref(self.root)
self.assertTrue(got["proven"], got["reasons"])
self.assertEqual(got["remote"], "prgs")
self.assertEqual(got["branch"], "master")
self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master")
self.assertEqual(got["repository_slug"], PRGS_SLUG)
self.assertEqual(got["source"], crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM)
# Cache and upstream agree here, which is the ordinary PRGS state.
self.assertFalse(got["cached_remote_head_conflicts"])
self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head)
def test_prgs_resolves_without_a_configured_upstream(self):
"""A PRGS checkout with no tracking config still resolves master."""
head = _make_repo(self.root, remote="prgs", url=PRGS_URL)
_set_remote_branch(self.root, "prgs", "master", head)
got = crr.resolve_target_base_ref(self.root)
self.assertTrue(got["proven"], got["reasons"])
self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master")
self.assertEqual(got["source"], crr.BASE_REF_SOURCE_UNIQUE_CANDIDATE)
def test_legacy_explicit_remote_refs_path_is_untouched(self):
"""An explicit remote_refs override still short-circuits derivation."""
head = _make_repo(self.root, remote="prgs", url=PRGS_URL)
_set_remote_branch(self.root, "prgs", "master", head)
state = root_checkout_guard.resolve_remote_master_ref_state(
self.root, remote_refs=root_checkout_guard.REMOTE_MASTER_REFS
)
self.assertEqual(state["sha"], head)
self.assertEqual(state["source"], "explicit_remote_refs")
class TestStaleCachedRemoteHead(_RepoCase):
"""Required coverage 3: configured upstream MDCPS/dev vs stale cache -> main.
This is B1. The fixture deliberately does **not** point
``refs/remotes/MDCPS/HEAD`` at ``dev``; it reproduces the disagreement that
was live on ``/Users/jasonwalker/Development/weekly-briefings``.
"""
def test_configured_upstream_beats_stale_cached_remote_head(self):
head = self._weekly_briefings_shape()
# Precondition: the fixture really is in the defective state.
self.assertEqual(
_git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD"),
"refs/remotes/MDCPS/main",
)
self.assertEqual(_git(self.root, "config", "--get", "branch.dev.merge"), "refs/heads/dev")
self.assertNotEqual(
_git(self.root, "rev-parse", "refs/remotes/MDCPS/main"),
_git(self.root, "rev-parse", "refs/remotes/MDCPS/dev"),
)
got = crr.resolve_target_base_ref(self.root)
self.assertTrue(got["proven"], got["reasons"])
self.assertEqual(got["branch"], "dev")
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
self.assertEqual(got["source"], crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM)
# The stale cache is reported, never obeyed.
self.assertEqual(got["cached_remote_head_branch"], "main")
self.assertTrue(got["cached_remote_head_conflicts"])
self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head)
def test_checkout_on_its_integration_tip_is_not_blocked(self):
"""The live symptom: a checkout exactly on its tip was reported stale."""
head = self._weekly_briefings_shape()
state = root_checkout_guard.resolve_remote_master_ref_state(self.root)
self.assertEqual(state["sha"], head)
self.assertTrue(state["cached_remote_head_conflicts"])
assessment = root_checkout_guard.assess_root_checkout_guard(
workspace_path=self.root,
canonical_repo_root=self.root,
current_branch="dev",
head_sha=head,
porcelain_status="",
remote_master_sha=state["sha"],
remote_master_ref=state["ref"],
)
self.assertTrue(assessment["proven"], assessment["reasons"])
report = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertFalse(report["stale"])
self.assertEqual(report["base_branch"], "dev")
self.assertEqual(report["reasons"], [])
def test_cached_remote_head_alone_never_proves_a_target(self):
"""With no configured upstream, the cache cannot supply the branch."""
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
# Only a non-candidate branch exists, and only the cache names it.
_set_remote_branch(self.root, "MDCPS", "trunk", head)
_set_remote_head(self.root, "MDCPS", "trunk")
got = crr.resolve_target_base_ref(self.root)
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH)
self.assertEqual(got["tracking_refs"], ())
self.assertEqual(got["cached_remote_head_branch"], "trunk")
# The refusal names the misleading signal so an operator is not sent
# chasing a ref that looks authoritative.
self.assertIn("trunk", " ".join(got["reasons"]))
def test_cached_remote_head_never_breaks_a_tie(self):
"""Two candidates, no upstream: the cache must not decide."""
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
_set_remote_branch(self.root, "MDCPS", "dev", head)
_set_remote_branch(self.root, "MDCPS", "main", head)
_set_remote_head(self.root, "MDCPS", "dev")
got = crr.resolve_target_base_ref(self.root)
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH)
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
def test_no_proven_source_is_the_remote_head_cache(self):
"""Structural guard: the cache is not in the set of proving sources."""
sources = {
crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM,
crr.BASE_REF_SOURCE_UNIQUE_CANDIDATE,
}
self.assertNotIn("remote_head_symref", sources)
self.assertFalse(hasattr(crr, "BASE_REF_SOURCE_REMOTE_HEAD"))
class TestCrossRepositoryTarget(_RepoCase):
"""Required coverage 2/4/5: MDCPS/dev, no origin remote, exact case."""
def test_mdcps_dev_resolves(self):
head = self._weekly_briefings_shape()
got = crr.resolve_target_base_ref(self.root)
self.assertTrue(got["proven"], got["reasons"])
self.assertEqual(got["remote"], "MDCPS")
self.assertEqual(got["branch"], "dev")
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
self.assertEqual(got["repository_slug"], MDCPS_SLUG)
self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head)
def test_no_remote_named_origin(self):
self._weekly_briefings_shape()
self.assertEqual(_git(self.root, "remote"), "MDCPS")
got = crr.resolve_target_base_ref(self.root)
self.assertTrue(got["proven"], got["reasons"])
self.assertNotIn("origin", got["tracking_ref"])
def test_remote_name_case_is_preserved_exactly(self):
self._weekly_briefings_shape()
got = crr.resolve_target_base_ref(self.root)
self.assertEqual(got["remote"], "MDCPS")
self.assertNotEqual(got["remote"], "mdcps")
# The tracking ref must address the real ref, which is case-sensitive.
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
self.assertTrue(
_git(self.root, "rev-parse", "--verify", got["tracking_ref"]),
"case-preserved tracking ref must resolve",
)
def test_lowercase_candidate_never_supplies_the_remote_name(self):
"""Guards against silently case-folding MDCPS to the candidate 'mdcps'.
``_IDENTITY_REMOTE_CANDIDATES`` contains a lowercase ``mdcps`` entry and
is probed *before* the repository's own remote listing. Git remote names
live in case-sensitive config subsections on every platform, so the
lowercase probe cannot resolve and the exact-case name must arrive from
``git remote``. Asserted through config rather than ref lookup because a
case-insensitive filesystem (macOS) resolves loose refs either way, which
would make a ref-based assertion test the filesystem instead of the code.
"""
self._weekly_briefings_shape()
res = subprocess.run(
["git", "-C", self.root, "remote", "get-url", "mdcps"],
capture_output=True,
text=True,
check=False,
)
self.assertNotEqual(res.returncode, 0, "git remote names are case-sensitive")
# A caller naming the wrong case cannot disambiguate, but the target is
# unambiguous anyway, so the exact-case name is still resolved.
got = crr.resolve_target_base_ref(self.root, explicit_remote="mdcps")
self.assertTrue(got["proven"], got["reasons"])
self.assertEqual(got["remote"], "MDCPS")
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
self.assertFalse(got["identity_explicit"], "a non-matching name is not explicit intent")
name, slug = crr.resolve_identity_remote(self.root)
self.assertEqual(name, "MDCPS")
self.assertEqual(slug, MDCPS_SLUG)
class TestTargetStaleness(_RepoCase):
"""Required coverage 6/7: matching tip, and behind or divergent checkout."""
def test_local_equal_to_resolved_tip_is_not_stale(self):
self._weekly_briefings_shape()
got = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertTrue(got["determinable"])
self.assertFalse(got["stale"])
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
self.assertEqual(got["base_remote"], "MDCPS")
self.assertEqual(got["base_branch"], "dev")
self.assertIsNone(got["reason_code"])
def test_local_behind_resolved_tip_is_stale(self):
head = self._weekly_briefings_shape()
advanced = _advance(self.root, "remote moved on")
_set_remote_branch(self.root, "MDCPS", "dev", advanced)
_git(self.root, "reset", "--hard", "--quiet", head)
got = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertTrue(got["determinable"])
self.assertTrue(got["stale"])
self.assertEqual(got["checkout_head"], head)
self.assertEqual(got["remote_tracking_head"], advanced)
def test_local_divergent_from_resolved_tip_is_stale(self):
head = self._weekly_briefings_shape()
remote_side = _advance(self.root, "remote side")
_set_remote_branch(self.root, "MDCPS", "dev", remote_side)
_git(self.root, "reset", "--hard", "--quiet", head)
local_side = _advance(self.root, "local side")
got = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertTrue(got["stale"])
self.assertEqual(got["checkout_head"], local_side)
self.assertNotEqual(local_side, remote_side)
class TestFailClosed(_RepoCase):
"""Required coverage 8/9: missing remote or ref, and ambiguous targets."""
def test_missing_remote_fails_closed(self):
_make_repo(self.root, remote=None, url=None)
got = crr.resolve_target_base_ref(self.root)
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_NO_IDENTITY_REMOTE)
self.assertEqual(got["tracking_refs"], ())
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
def test_missing_tracking_ref_fails_closed(self):
_make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
# Remote configured, but nothing has ever been fetched.
got = crr.resolve_target_base_ref(self.root)
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH)
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
def test_configured_upstream_without_a_tracking_ref_fails_closed(self):
"""A proven target always names a ref that resolves."""
_make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
_set_upstream(self.root, "MDCPS", "dev") # declared, never fetched
got = crr.resolve_target_base_ref(self.root)
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH)
self.assertIsNone(got["tracking_ref"])
def test_every_proven_target_resolves(self):
"""No 'proven' result may name an unresolvable tracking ref."""
head = self._weekly_briefings_shape()
got = crr.resolve_target_base_ref(self.root)
self.assertTrue(got["proven"])
self.assertEqual(_git(self.root, "rev-parse", got["tracking_ref"]), head)
def test_ambiguous_integration_branch_fails_closed(self):
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
# Two candidate integration branches and no configured upstream.
_set_remote_branch(self.root, "MDCPS", "dev", head)
_set_remote_branch(self.root, "MDCPS", "main", head)
got = crr.resolve_target_base_ref(self.root)
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH)
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
def test_ambiguous_identity_remote_fails_closed(self):
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
_git(self.root, "remote", "add", "prgs", PRGS_URL)
_set_remote_branch(self.root, "MDCPS", "dev", head)
_set_remote_branch(self.root, "prgs", "master", head)
got = crr.resolve_target_base_ref(self.root)
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_REMOTE)
self.assertEqual(got["tracking_refs"], ())
def test_orphan_tracking_ref_from_removed_remote_is_ignored(self):
"""The live Gitea-Tools symptom: refs/remotes/origin/* outlives its remote."""
head = _make_repo(self.root, remote="prgs", url=PRGS_URL)
_set_remote_branch(self.root, "prgs", "master", head)
_set_upstream(self.root, "prgs", "master")
# An abandoned ref left behind by a remote that no longer exists.
_set_remote_branch(self.root, "origin", "master", head)
_advance(self.root, "orphan must not be consulted")
got = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master")
self.assertEqual(got["repository_slug"], PRGS_SLUG)
self.assertNotIn(
"target repository identity could not be derived from its git remote",
got["reasons"],
)
class TestExplicitVersusInferredRemote(_RepoCase):
"""Required coverage 11: explicit disambiguation stays distinct from inference.
This is B2. A remote the module inferred while probing must never re-enter
the resolver as though an operator had named it.
"""
def _two_remotes(self) -> str:
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
_git(self.root, "remote", "add", "prgs", PRGS_URL)
_set_remote_branch(self.root, "MDCPS", "dev", head)
_set_remote_branch(self.root, "prgs", "master", head)
return head
def test_explicit_remote_disambiguates(self):
self._two_remotes()
got = crr.resolve_target_base_ref(self.root, explicit_remote="MDCPS")
self.assertTrue(got["proven"], got["reasons"])
self.assertEqual(got["remote"], "MDCPS")
self.assertEqual(got["branch"], "dev")
self.assertTrue(got["identity_explicit"])
def test_inferred_remote_does_not_disambiguate(self):
"""Feeding the inferred remote back in must not unlock the target."""
self._two_remotes()
inferred, _ = crr.resolve_identity_remote(self.root)
self.assertIsNotNone(inferred, "the first-wins probe still returns a name")
# The report infers internally and must still refuse.
report = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertEqual(report["reason_code"], crr.DENY_AMBIGUOUS_REMOTE)
self.assertIsNone(report["repository_slug"])
self.assertIsNone(report["tracking_ref"])
self.assertFalse(report["stale"])
self.assertTrue(report["reasons"])
def test_report_never_passes_a_remote_into_the_resolver(self):
"""Structural guard against the exact B2 regression."""
src = inspect.getsource(master_parity_gate.assess_target_repository_parity)
self.assertIn("resolve_target_base_ref(canonical_root)", src)
self.assertNotIn("resolve_target_base_ref(canonical_root, remote=", src)
self.assertNotIn("explicit_remote=identity", src)
# Reporting must use the ambiguity-aware identity resolver.
self.assertIn("assess_identity_remote(canonical_root)", src)
def test_explicit_parameter_is_named_for_its_meaning(self):
for fn in (
crr.resolve_target_base_ref,
root_checkout_guard.resolve_remote_master_sha,
root_checkout_guard.resolve_remote_master_ref_state,
):
params = inspect.signature(fn).parameters
self.assertIn("explicit_remote", params, fn.__name__)
self.assertNotIn("remote", params, fn.__name__)
def test_unmatched_explicit_remote_cannot_unlock_an_ambiguous_target(self):
self._two_remotes()
got = crr.resolve_target_base_ref(self.root, explicit_remote="nonexistent")
self.assertFalse(got["proven"])
self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_REMOTE)
class TestGatingAndReportingAgree(_RepoCase):
"""Required coverage 10: guard and parity report make identical decisions."""
def test_same_resolved_target_for_gate_and_report(self):
self._weekly_briefings_shape()
gate = root_checkout_guard.resolve_remote_master_ref_state(self.root)
report = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertEqual(gate["remote"], report["base_remote"])
self.assertEqual(gate["branch"], report["base_branch"])
self.assertEqual(gate["sha"], report["remote_tracking_head"])
self.assertIn(
gate["ref"],
(report["tracking_ref"], f"{gate['remote']}/{gate['branch']}"),
)
def test_both_sides_refuse_the_same_unresolvable_target(self):
_make_repo(self.root, remote=None, url=None)
gate = root_checkout_guard.resolve_remote_master_ref_state(self.root)
report = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertIsNone(gate["sha"])
self.assertIsNone(report["remote_tracking_head"])
self.assertFalse(report["stale"])
self.assertTrue(report["reasons"])
self.assertEqual(gate["reason_code"], report["reason_code"])
def test_both_sides_refuse_the_same_ambiguous_target(self):
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
_git(self.root, "remote", "add", "prgs", PRGS_URL)
_set_remote_branch(self.root, "MDCPS", "dev", head)
_set_remote_branch(self.root, "prgs", "master", head)
gate = root_checkout_guard.resolve_remote_master_ref_state(self.root)
report = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertIsNone(gate["sha"])
self.assertEqual(gate["reason_code"], crr.DENY_AMBIGUOUS_REMOTE)
# The report must not name a repository the gate refuses to act on.
self.assertEqual(report["reason_code"], gate["reason_code"])
self.assertIsNone(report["repository_slug"])
self.assertIsNone(report["base_remote"])
self.assertIsNone(report["base_branch"])
self.assertFalse(report["stale"])
def test_both_sides_refuse_the_same_ambiguous_branch(self):
head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL)
_set_remote_branch(self.root, "MDCPS", "dev", head)
_set_remote_branch(self.root, "MDCPS", "main", head)
gate = root_checkout_guard.resolve_remote_master_ref_state(self.root)
report = master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertEqual(gate["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH)
self.assertEqual(report["reason_code"], gate["reason_code"])
self.assertIsNone(report["tracking_ref"])
class TestProductionCallers(unittest.TestCase):
"""Required coverage 13: every production caller consumes the same target."""
def test_guard_reports_the_ref_it_actually_compared(self):
assessment = root_checkout_guard.assess_root_checkout_guard(
workspace_path="/tmp/nonexistent-workspace-983",
canonical_repo_root="/tmp/nonexistent-root-983",
current_branch="dev",
head_sha="a" * 40,
porcelain_status="",
remote_master_sha="b" * 40,
remote_master_ref="refs/remotes/MDCPS/dev",
)
self.assertTrue(assessment["block"])
joined = " ".join(assessment["reasons"])
self.assertIn("refs/remotes/MDCPS/dev", joined)
self.assertNotIn("prgs/master", joined)
def test_guard_message_without_a_ref_stays_generic(self):
assessment = root_checkout_guard.assess_root_checkout_guard(
workspace_path="/tmp/nonexistent-workspace-983",
canonical_repo_root="/tmp/nonexistent-root-983",
current_branch="master",
head_sha="a" * 40,
porcelain_status="",
remote_master_sha="b" * 40,
)
joined = " ".join(assessment["reasons"])
self.assertIn("the tracking integration ref", joined)
self.assertNotIn("prgs/master", joined)
def test_anti_stomp_preflight_forwards_the_resolved_ref(self):
sig = inspect.signature(anti_stomp_preflight.assess_anti_stomp_preflight)
self.assertIn("remote_master_ref", sig.parameters)
src = inspect.getsource(anti_stomp_preflight.assess_anti_stomp_preflight)
self.assertIn("remote_master_ref=remote_master_ref", src)
def test_no_production_caller_inherits_the_prgs_default(self):
"""Every resolve site must derive, or pass remote_refs explicitly."""
import gitea_mcp_server
src = inspect.getsource(gitea_mcp_server)
# The four historical call sites now consume the resolved-target state.
self.assertGreaterEqual(src.count("resolve_remote_master_ref_state("), 4)
self.assertNotIn("resolve_remote_master_sha(canonical_root)", src)
def test_no_production_caller_supplies_an_explicit_remote(self):
"""Nothing in production may suppress the ambiguity gate (#983 B2)."""
import gitea_mcp_server
for module in (gitea_mcp_server, anti_stomp_preflight, master_parity_gate):
src = inspect.getsource(module)
self.assertNotIn("explicit_remote=", src, module.__name__)
class TestRepositoryStructureUntouched(_RepoCase):
"""Required coverage 12: resolution mutates no ref, remote, config, or checkout."""
def test_resolution_creates_no_refs_or_branches(self):
self._weekly_briefings_shape()
fmt = "--format=%(refname) %(objectname)"
before_refs = _git(self.root, "for-each-ref", fmt)
before_remotes = _git(self.root, "remote")
before_config = _git(self.root, "config", "--local", "--list")
before_head = _git(self.root, "rev-parse", "HEAD")
before_branch = _git(self.root, "symbolic-ref", "--short", "HEAD")
before_status = _git(self.root, "status", "--porcelain", "--untracked-files=all")
before_symref = _git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD")
crr.resolve_target_base_ref(self.root)
crr.assess_identity_remote(self.root)
root_checkout_guard.resolve_remote_master_sha(self.root)
root_checkout_guard.resolve_remote_master_ref_state(self.root)
master_parity_gate.assess_target_repository_parity(
canonical_root=self.root, source="test"
)
self.assertEqual(_git(self.root, "for-each-ref", fmt), before_refs)
self.assertEqual(_git(self.root, "remote"), before_remotes)
self.assertEqual(_git(self.root, "config", "--local", "--list"), before_config)
self.assertEqual(_git(self.root, "rev-parse", "HEAD"), before_head)
self.assertEqual(_git(self.root, "symbolic-ref", "--short", "HEAD"), before_branch)
self.assertEqual(
_git(self.root, "status", "--porcelain", "--untracked-files=all"), before_status
)
# The stale cache is specifically NOT repaired: that would be a mutation.
self.assertEqual(_git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD"), before_symref)
if __name__ == "__main__":
unittest.main()
@@ -1,664 +0,0 @@
"""#985 — trusted client-instance identity for project-scoped LLM launches.
Covers the acceptance criteria of issue #985:
* a three-namespace (author/reviewer/merger) launch succeeds with no
controller or reconciler profile,
* every worker of one launch shares exactly one trusted instance ID,
* two concurrent launches receive distinct IDs and stay distinguishable,
* reuse, missing, legacy, malformed, unsanctioned, and unsealed identities all
fail closed,
* the existing five-namespace behaviour is unchanged,
* Claude Code has a documented, runnable launch command.
"""
from __future__ import annotations
import datetime as _dt
import json
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import mcp_application_launcher as launcher # noqa: E402
import mcp_fleet_snapshot as fleet # noqa: E402
NOW = _dt.datetime(2026, 7, 31, 12, 0, 0, tzinfo=_dt.timezone.utc)
PROJECT_NAMESPACES = ("author", "reviewer", "merger")
ALL_NAMESPACES = ("author", "reviewer", "merger", "controller", "reconciler")
#: A project-scoped configuration intentionally has no controller/reconciler
#: profile — that absence is the condition under test, not an omission.
PROJECT_PROFILES = {ns: f"prgs-{ns}" for ns in PROJECT_NAMESPACES}
FULL_PROFILES = {ns: f"prgs-{ns}" for ns in ALL_NAMESPACES}
def _instance_ids(servers):
return {
name: entry["env"][launcher.CLIENT_INSTANCE_ENV]
for name, entry in servers.items()
}
class ResolveLaunchNamespacesTests(unittest.TestCase):
"""The validated subset allow-list."""
def test_none_resolves_to_all_five_in_canonical_order(self):
self.assertEqual(
launcher.resolve_launch_namespaces(None), tuple(ALL_NAMESPACES)
)
def test_project_subset_resolves_in_canonical_order(self):
# Deliberately out of order on input; canonical order on output.
self.assertEqual(
launcher.resolve_launch_namespaces(["merger", "author", "reviewer"]),
("author", "reviewer", "merger"),
)
def test_case_and_whitespace_normalised(self):
self.assertEqual(
launcher.resolve_launch_namespaces([" Author ", "REVIEWER"]),
("author", "reviewer"),
)
def test_unsanctioned_namespace_refused(self):
with self.assertRaises(ValueError) as ctx:
launcher.resolve_launch_namespaces(["author", "admin"])
self.assertIn("admin", str(ctx.exception))
def test_typo_is_refused_not_silently_dropped(self):
with self.assertRaises(ValueError):
launcher.resolve_launch_namespaces(["author", "reviewr"])
def test_empty_subset_refused(self):
with self.assertRaises(ValueError):
launcher.resolve_launch_namespaces([])
with self.assertRaises(ValueError):
launcher.resolve_launch_namespaces([" "])
def test_duplicate_namespace_refused(self):
with self.assertRaises(ValueError):
launcher.resolve_launch_namespaces(["author", "author"])
class ProjectScopedLaunchTests(unittest.TestCase):
"""AC: three-namespace launch without controller/reconciler profiles."""
def _build(self, **kwargs):
kwargs.setdefault("client_type", "claude_code")
kwargs.setdefault("namespaces", PROJECT_NAMESPACES)
kwargs.setdefault("command", "python3")
kwargs.setdefault("args", ["mcp_server.py"])
profiles = kwargs.pop("profiles", PROJECT_PROFILES)
return launcher.build_application_mcp_servers(profiles, **kwargs)
def test_three_namespace_launch_succeeds_without_controller_reconciler(self):
built = self._build(launch_nonce="proj-1", now=NOW)
servers = built["mcpServers"]
self.assertEqual(
sorted(servers), ["gitea-author", "gitea-merger", "gitea-reviewer"]
)
self.assertEqual(built["namespace_count"], 3)
self.assertEqual(list(built["namespaces"]), list(PROJECT_NAMESPACES))
def test_excluded_namespaces_are_not_started(self):
built = self._build(launch_nonce="proj-2", now=NOW)
self.assertNotIn("gitea-controller", built["mcpServers"])
self.assertNotIn("gitea-reconciler", built["mcpServers"])
self.assertEqual(
sorted(built["excluded_namespaces"]), ["controller", "reconciler"]
)
self.assertTrue(built["project_scoped"])
def test_all_workers_of_one_launch_share_one_trusted_id(self):
built = self._build(launch_nonce="proj-3", now=NOW)
ids = set(_instance_ids(built["mcpServers"]).values())
self.assertEqual(len(ids), 1, ids)
shared = next(iter(ids))
self.assertTrue(fleet.assess_instance_identity(shared)["trusted"])
self.assertEqual(built["client_instance_id"], shared)
def test_every_worker_env_is_launcher_sealed(self):
built = self._build(launch_nonce="proj-4", now=NOW)
for name, entry in built["mcpServers"].items():
env = entry["env"]
self.assertEqual(
env[launcher.INSTANCE_PROVENANCE_ENV],
fleet.INSTANCE_ID_PROVENANCE_TRUSTED,
name,
)
self.assertEqual(env["GITEA_CLIENT_MANAGED"], "1", name)
self.assertFalse(
env[launcher.CLIENT_INSTANCE_ENV].startswith("legacy-"), name
)
def test_attribution_proof_holds_for_three_namespace_launch(self):
"""A subset launch must not read as 'missing two workers'."""
built = self._build(launch_nonce="proj-5", now=NOW)
proof = launcher.collect_instance_ids_from_mcp_servers(
built["mcpServers"], namespaces=PROJECT_NAMESPACES
)
self.assertTrue(proof["shared_single_trusted_id"], proof)
self.assertEqual(proof["missing_servers"], [])
# And without an explicit expectation it inspects what is present.
inferred = launcher.collect_instance_ids_from_mcp_servers(
built["mcpServers"]
)
self.assertTrue(inferred["shared_single_trusted_id"], inferred)
def test_missing_profile_for_requested_namespace_refused(self):
with self.assertRaises(ValueError) as ctx:
self._build(profiles={"author": "prgs-author"})
self.assertIn("merger", str(ctx.exception))
def test_profile_for_unlaunched_namespace_is_ignored_not_started(self):
"""Extra profiles never widen the launch beyond the validated subset."""
built = self._build(profiles=FULL_PROFILES, launch_nonce="proj-6", now=NOW)
self.assertEqual(len(built["mcpServers"]), 3)
self.assertNotIn("gitea-controller", built["mcpServers"])
def test_unsanctioned_namespace_refused_at_build(self):
with self.assertRaises(ValueError):
self._build(
profiles={**PROJECT_PROFILES, "admin": "prgs-admin"},
namespaces=["author", "admin"],
)
class ConcurrentLaunchDistinctionTests(unittest.TestCase):
"""AC: concurrent launches are distinct and distinguishable."""
def _build(self, nonce, namespaces=PROJECT_NAMESPACES, profiles=None):
return launcher.build_application_mcp_servers(
profiles or PROJECT_PROFILES,
client_type="claude_code",
namespaces=namespaces,
launch_nonce=nonce,
now=NOW,
command="python3",
args=["mcp_server.py"],
)
def test_two_concurrent_launches_receive_different_ids(self):
a = self._build("concurrent-a")
b = self._build("concurrent-b")
self.assertNotEqual(a["client_instance_id"], b["client_instance_id"])
def test_same_profile_across_launches_stays_distinguishable(self):
"""Same profile, two launches: distinguishable, so not a false duplicate."""
a = self._build("dist-a")
b = self._build("dist-b")
a_author = a["mcpServers"]["gitea-author"]["env"]
b_author = b["mcpServers"]["gitea-author"]["env"]
# Same profile on purpose — that alone must never imply duplication.
self.assertEqual(a_author["GITEA_MCP_PROFILE"], b_author["GITEA_MCP_PROFILE"])
self.assertNotEqual(
a_author[launcher.CLIENT_INSTANCE_ENV],
b_author[launcher.CLIENT_INSTANCE_ENV],
)
for env in (a_author, b_author):
assessed = launcher.inherit_or_refuse_client_instance(env)
self.assertTrue(assessed["trusted"], assessed)
self.assertTrue(assessed["launcher_sealed"], assessed)
def test_no_nonce_still_yields_distinct_ids(self):
a = self._build(None)
b = self._build(None)
self.assertNotEqual(a["client_instance_id"], b["client_instance_id"])
def test_ids_are_never_persisted_between_launches(self):
"""No static per-project ID: two runs of one config differ."""
first = self._build("static-check-1")
second = self._build("static-check-2")
self.assertNotEqual(
first["mcpServers"]["gitea-author"]["env"][launcher.CLIENT_INSTANCE_ENV],
second["mcpServers"]["gitea-author"]["env"][launcher.CLIENT_INSTANCE_ENV],
)
class FailClosedIdentityTests(unittest.TestCase):
"""AC: missing / legacy / reused / conflicting / unsealed all fail closed."""
def test_missing_identity_fails_closed(self):
assessed = launcher.inherit_or_refuse_client_instance({})
self.assertFalse(assessed["trusted"])
self.assertFalse(assessed["launcher_sealed"])
self.assertEqual(
assessed["provenance"], fleet.INSTANCE_ID_PROVENANCE_MISSING
)
def test_legacy_pid_identity_fails_closed(self):
for legacy in ("legacy-pid-40578", "pid-1234", "proc-99"):
with self.subTest(legacy=legacy):
assessed = launcher.inherit_or_refuse_client_instance(
{
launcher.CLIENT_INSTANCE_ENV: legacy,
launcher.INSTANCE_PROVENANCE_ENV: (
fleet.INSTANCE_ID_PROVENANCE_TRUSTED
),
}
)
self.assertFalse(assessed["trusted"], legacy)
self.assertFalse(assessed["launcher_sealed"], legacy)
def test_wellformed_but_unsealed_identity_fails_closed(self):
"""The decisive case: format alone must not manufacture trust."""
forged = fleet.generate_client_instance_id(
"claude_code", launch_nonce="forged", now=NOW
)
# A hand-set env var with a valid-looking ID but no launcher seal.
assessed = launcher.inherit_or_refuse_client_instance(
{launcher.CLIENT_INSTANCE_ENV: forged}
)
self.assertFalse(assessed["trusted"], assessed)
self.assertFalse(assessed["launcher_sealed"], assessed)
self.assertEqual(
assessed["provenance"], fleet.INSTANCE_ID_PROVENANCE_UNSEALED
)
# The value is still reported for diagnosis, not silently dropped.
self.assertEqual(assessed["client_instance_id"], forged)
def test_wrong_provenance_marker_fails_closed(self):
forged = fleet.generate_client_instance_id(
"claude_code", launch_nonce="wrong-marker", now=NOW
)
assessed = launcher.inherit_or_refuse_client_instance(
{
launcher.CLIENT_INSTANCE_ENV: forged,
launcher.INSTANCE_PROVENANCE_ENV: "totally_trusted",
}
)
self.assertFalse(assessed["trusted"], assessed)
self.assertEqual(
assessed["provenance"], fleet.INSTANCE_ID_PROVENANCE_UNSEALED
)
def test_sealed_launcher_env_is_trusted(self):
env = launcher.namespace_worker_env(
profile_name="prgs-author",
client_type="claude_code",
client_instance_id=fleet.generate_client_instance_id(
"claude_code", launch_nonce="sealed", now=NOW
),
)
assessed = launcher.inherit_or_refuse_client_instance(env)
self.assertTrue(assessed["trusted"], assessed)
self.assertTrue(assessed["launcher_sealed"], assessed)
def test_untrusted_id_refused_on_build_and_worker_env(self):
with self.assertRaises(ValueError):
launcher.build_application_mcp_servers(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
client_instance_id="legacy-pid-1",
command="python3",
args=["mcp_server.py"],
)
with self.assertRaises(ValueError):
launcher.namespace_worker_env(
profile_name="prgs-author",
client_type="claude_code",
client_instance_id="hand-written",
)
def test_extra_env_cannot_forge_seal_or_identity(self):
trusted = fleet.generate_client_instance_id(
"claude_code", launch_nonce="seal-guard", now=NOW
)
env = launcher.namespace_worker_env(
profile_name="prgs-author",
client_type="claude_code",
client_instance_id=trusted,
extra_env={
launcher.CLIENT_INSTANCE_ENV: "inst-spoof-20260101T000000Z-abcdef",
launcher.INSTANCE_PROVENANCE_ENV: "trusted_launcher",
"GITEA_CLIENT_MANAGED": "0",
},
)
self.assertEqual(env[launcher.CLIENT_INSTANCE_ENV], trusted)
self.assertEqual(env["GITEA_CLIENT_MANAGED"], "1")
def test_conflicting_ids_across_workers_break_shared_attribution(self):
"""One launch whose workers disagree must not read as shared."""
built = launcher.build_application_mcp_servers(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
launch_nonce="conflict",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
servers = built["mcpServers"]
servers["gitea-merger"]["env"][launcher.CLIENT_INSTANCE_ENV] = (
fleet.generate_client_instance_id(
"claude_code", launch_nonce="other", now=NOW
)
)
proof = launcher.collect_instance_ids_from_mcp_servers(
servers, namespaces=PROJECT_NAMESPACES
)
self.assertFalse(proof["shared_single_trusted_id"], proof)
self.assertEqual(len(proof["unique_instance_ids"]), 2)
def test_absent_expected_worker_breaks_shared_attribution(self):
built = launcher.build_application_mcp_servers(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
launch_nonce="absent",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
servers = dict(built["mcpServers"])
servers.pop("gitea-merger")
proof = launcher.collect_instance_ids_from_mcp_servers(
servers, namespaces=PROJECT_NAMESPACES
)
self.assertFalse(proof["shared_single_trusted_id"], proof)
self.assertEqual(proof["missing_servers"], ["gitea-merger"])
def test_reused_trusted_id_is_reuse_not_a_fresh_launch(self):
"""Resume reuses by design; two *live* launches sharing it is the hazard."""
first = launcher.build_application_mcp_servers(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
launch_nonce="reuse-src",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
resumed = launcher.build_application_mcp_servers(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
client_instance_id=first["client_instance_id"],
command="python3",
args=["mcp_server.py"],
)
self.assertEqual(
first["client_instance_id"], resumed["client_instance_id"]
)
# Two independent launches must never collide by default.
independent = launcher.build_application_mcp_servers(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
command="python3",
args=["mcp_server.py"],
)
self.assertNotEqual(
first["client_instance_id"], independent["client_instance_id"]
)
class BackwardCompatibilityTests(unittest.TestCase):
"""AC: existing full five-namespace behaviour remains compatible."""
def test_omitting_namespaces_launches_all_five(self):
built = launcher.build_application_mcp_servers(
FULL_PROFILES,
client_type="codex",
launch_nonce="compat",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
self.assertEqual(len(built["mcpServers"]), 5)
self.assertEqual(built["namespace_count"], 5)
self.assertEqual(built["excluded_namespaces"], [])
self.assertFalse(built["project_scoped"])
ids = set(_instance_ids(built["mcpServers"]).values())
self.assertEqual(len(ids), 1)
def test_five_namespace_missing_profile_still_refused(self):
incomplete = {ns: f"prgs-{ns}" for ns in PROJECT_NAMESPACES}
with self.assertRaises(ValueError) as ctx:
launcher.build_application_mcp_servers(
incomplete,
client_type="codex",
command="python3",
args=["mcp_server.py"],
)
message = str(ctx.exception)
self.assertIn("controller", message)
self.assertIn("reconciler", message)
def test_legacy_collect_call_without_namespaces_still_works(self):
built = launcher.build_application_mcp_servers(
FULL_PROFILES,
client_type="codex",
launch_nonce="compat-proof",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
proof = launcher.collect_instance_ids_from_mcp_servers(built["mcpServers"])
self.assertTrue(proof["shared_single_trusted_id"], proof)
self.assertEqual(proof["namespace_server_count"], 5)
def test_gitea_config_wrapper_supports_subset(self):
import gitea_config
built = gitea_config.multi_namespace_launcher_entries(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
config_path="/cfg/profiles.json",
launch_nonce="cfg-subset",
)
self.assertEqual(built["namespace_count"], 3)
self.assertEqual(
sorted(built["excluded_namespaces"]), ["controller", "reconciler"]
)
def test_gitea_config_wrapper_default_is_five(self):
import gitea_config
built = gitea_config.multi_namespace_launcher_entries(
FULL_PROFILES,
client_type="claude_code",
config_path="/cfg/profiles.json",
launch_nonce="cfg-full",
)
self.assertEqual(built["namespace_count"], 5)
class RunnableLaunchPathTests(unittest.TestCase):
"""AC: Claude Code has a documented supported launch command."""
def test_claude_code_is_a_supported_client(self):
self.assertIn("claude_code", launcher.supported_launch_clients())
def test_claude_code_argv_uses_per_launch_config(self):
argv = launcher.build_client_launch_argv("claude_code", "/tmp/launch.json")
self.assertEqual(argv[0], "claude")
self.assertIn("--mcp-config", argv)
self.assertIn("/tmp/launch.json", argv)
self.assertIn("--strict-mcp-config", argv)
def test_extra_args_are_forwarded(self):
argv = launcher.build_client_launch_argv(
"claude_code", "/tmp/launch.json", extra_args=["--resume"]
)
self.assertEqual(argv[-1], "--resume")
def test_unsupported_client_refused(self):
with self.assertRaises(ValueError):
launcher.build_client_launch_argv("not_a_client", "/tmp/x.json")
def test_prepare_launch_writes_private_per_launch_config(self):
prepared = launcher.prepare_application_launch(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
launch_nonce="prep-1",
now=NOW,
)
path = prepared["launch_config_path"]
self.addCleanup(lambda: os.path.exists(path) and os.unlink(path))
self.assertTrue(os.path.exists(path))
self.assertEqual(os.stat(path).st_mode & 0o777, 0o600)
with open(path, encoding="utf-8") as handle:
payload = json.load(handle)
self.assertEqual(
sorted(payload["mcpServers"]),
["gitea-author", "gitea-merger", "gitea-reviewer"],
)
self.assertTrue(prepared["shared_single_trusted_id"], prepared)
self.assertIn(path, prepared["argv"])
def test_two_prepared_launches_use_distinct_files_and_ids(self):
first = launcher.prepare_application_launch(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
now=NOW,
)
second = launcher.prepare_application_launch(
PROJECT_PROFILES,
client_type="claude_code",
namespaces=PROJECT_NAMESPACES,
now=NOW,
)
for prepared in (first, second):
path = prepared["launch_config_path"]
self.addCleanup(lambda p=path: os.path.exists(p) and os.unlink(p))
self.assertNotEqual(
first["launch_config_path"], second["launch_config_path"]
)
self.assertNotEqual(
first["client_instance_id"], second["client_instance_id"]
)
def test_unsupported_client_writes_no_config_file(self):
"""Fail before creating a file, so a bad invocation leaves no residue."""
with self.assertRaises(ValueError):
launcher.prepare_application_launch(
PROJECT_PROFILES,
client_type="not_a_client",
namespaces=PROJECT_NAMESPACES,
now=NOW,
)
class CliTests(unittest.TestCase):
"""The runnable entry point itself."""
def _run_dry(self, argv):
import contextlib
import io
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer):
code = launcher.main(argv)
return code, buffer.getvalue()
def test_dry_run_project_scoped_plan(self):
code, out = self._run_dry(
[
"--client",
"claude_code",
"--namespaces",
"author,reviewer,merger",
"--profile",
"author=prgs-author",
"--profile",
"reviewer=prgs-reviewer",
"--profile",
"merger=prgs-merger",
"--dry-run",
]
)
self.assertEqual(code, 0)
plan = json.loads(out)
path = plan["launch_config_path"]
self.addCleanup(lambda: os.path.exists(path) and os.unlink(path))
self.assertEqual(plan["namespaces"], ["author", "reviewer", "merger"])
self.assertEqual(
sorted(plan["excluded_namespaces"]), ["controller", "reconciler"]
)
self.assertTrue(plan["shared_single_trusted_id"])
self.assertTrue(
fleet.assess_instance_identity(plan["client_instance_id"])["trusted"]
)
self.assertEqual(plan["argv"][0], "claude")
def test_cli_refuses_profile_for_unlaunched_namespace(self):
with self.assertRaises(SystemExit):
self._run_dry(
[
"--namespaces",
"author,reviewer,merger",
"--profile",
"author=prgs-author",
"--profile",
"reviewer=prgs-reviewer",
"--profile",
"merger=prgs-merger",
"--profile",
"controller=prgs-controller",
"--dry-run",
]
)
def test_cli_refuses_unsanctioned_namespace(self):
with self.assertRaises(SystemExit):
self._run_dry(
[
"--namespaces",
"author,admin",
"--profile",
"author=prgs-author",
"--dry-run",
]
)
def test_cli_refuses_missing_profile(self):
with self.assertRaises(SystemExit):
self._run_dry(
[
"--namespaces",
"author,reviewer",
"--profile",
"author=prgs-author",
"--dry-run",
]
)
def test_cli_refuses_malformed_profile_pair(self):
with self.assertRaises(SystemExit):
self._run_dry(
["--namespaces", "author", "--profile", "prgs-author", "--dry-run"]
)
def test_two_cli_runs_receive_distinct_ids(self):
args = [
"--namespaces",
"author,reviewer,merger",
"--profile",
"author=prgs-author",
"--profile",
"reviewer=prgs-reviewer",
"--profile",
"merger=prgs-merger",
"--dry-run",
]
_, first = self._run_dry(args)
_, second = self._run_dry(args)
a = json.loads(first)
b = json.loads(second)
for plan in (a, b):
path = plan["launch_config_path"]
self.addCleanup(lambda p=path: os.path.exists(p) and os.unlink(p))
self.assertNotEqual(a["client_instance_id"], b["client_instance_id"])
if __name__ == "__main__": # pragma: no cover
unittest.main()
+2 -16
View File
@@ -87,27 +87,13 @@ class TestAssessRootCheckoutGuard(unittest.TestCase):
self.assertTrue(result["block"]) self.assertTrue(result["block"])
self.assertIn("tracked local edits", result["reasons"][0]) self.assertIn("tracked local edits", result["reasons"][0])
def test_head_behind_tracking_base_ref_blocked(self): def test_head_behind_prgs_master_blocked(self):
"""#983: the base ref is derived, so the message no longer hardcodes PRGS."""
result = self._assess( result = self._assess(
head_sha=OTHER_SHA, head_sha=OTHER_SHA,
remote_master_sha=MASTER_SHA, remote_master_sha=MASTER_SHA,
) )
self.assertTrue(result["block"]) self.assertTrue(result["block"])
self.assertIn( self.assertIn("does not match prgs/master", result["reasons"][0])
"does not match the tracking integration ref", result["reasons"][0]
)
def test_head_behind_named_base_ref_reports_that_ref(self):
"""The resolved ref is named, so a non-PRGS target is reported accurately."""
result = self._assess(
head_sha=OTHER_SHA,
remote_master_sha=MASTER_SHA,
remote_master_ref="refs/remotes/MDCPS/dev",
)
self.assertTrue(result["block"])
self.assertIn("does not match refs/remotes/MDCPS/dev", result["reasons"][0])
self.assertNotIn("prgs/master", result["reasons"][0])
def test_merger_requires_clean_control_checkout(self): def test_merger_requires_clean_control_checkout(self):
result = self._assess( result = self._assess(