Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysadmin
|
e344e68a13 | ||
 sysadminandClaude Opus 4.8
|
c0c6d14b73 |
@@ -355,10 +355,6 @@ def assess_anti_stomp_preflight(
|
||||
root_head_sha: str | None = None,
|
||||
root_porcelain: 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_worktree: bool = True,
|
||||
create_issue_bootstrap_assessment: dict[str, Any] | None = None,
|
||||
@@ -557,7 +553,6 @@ def assess_anti_stomp_preflight(
|
||||
remote_master_sha=remote_master_sha,
|
||||
resolved_role=req_role or role,
|
||||
actual_role=role,
|
||||
remote_master_ref=remote_master_ref,
|
||||
)
|
||||
checks["root_checkout"] = {
|
||||
"block": bool(root_assessment.get("block")),
|
||||
|
||||
+14
-456
@@ -37,34 +37,6 @@ CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT"
|
||||
# Candidate git remote names probed when deriving repository identity.
|
||||
_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.
|
||||
# ``mode`` selects how repository authority is established, so an unrecognised
|
||||
# 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
|
||||
|
||||
|
||||
def _identity_remote_candidates(path: str, remote: str | None) -> list[str]:
|
||||
"""Ordered remote names to probe for identity at *path*.
|
||||
def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
|
||||
"""``owner/repository`` derived from a git remote configured at *path*.
|
||||
|
||||
Names are used verbatim — never case-folded. Git config subsection names are
|
||||
case-sensitive, so a repository whose remote is ``MDCPS`` is reached only by
|
||||
the exact string ``MDCPS``; the lowercase entry in
|
||||
:data:`_IDENTITY_REMOTE_CANDIDATES` simply does not resolve, and the exact
|
||||
name arrives from ``git remote`` below (#983).
|
||||
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).
|
||||
"""
|
||||
text = (path or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
|
||||
ordered: list[str] = []
|
||||
for name in (remote, *_IDENTITY_REMOTE_CANDIDATES):
|
||||
clean = (name or "").strip()
|
||||
@@ -165,7 +139,7 @@ def _identity_remote_candidates(path: str, remote: str | None) -> list[str]:
|
||||
|
||||
try:
|
||||
listed = subprocess.run(
|
||||
["git", "-C", path, "remote"],
|
||||
["git", "-C", text, "remote"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
@@ -175,20 +149,11 @@ def _identity_remote_candidates(path: str, remote: str | None) -> list[str]:
|
||||
for name in listed:
|
||||
if name and name not in ordered:
|
||||
ordered.append(name)
|
||||
return 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):
|
||||
for name in ordered:
|
||||
try:
|
||||
url = subprocess.run(
|
||||
["git", "-C", path, "remote", "get-url", name],
|
||||
["git", "-C", text, "remote", "get-url", name],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
@@ -197,415 +162,8 @@ def _configured_remote_identities(path: str, remote: str | None) -> list[tuple[s
|
||||
continue
|
||||
parsed = remote_repo_guard.parse_org_repo_from_remote_url(url)
|
||||
if parsed:
|
||||
found.append((name, f"{parsed[0]}/{parsed[1]}"))
|
||||
return found
|
||||
|
||||
|
||||
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,
|
||||
)
|
||||
return f"{parsed[0]}/{parsed[1]}"
|
||||
return None
|
||||
|
||||
|
||||
def assess_canonical_repository_root(
|
||||
|
||||
@@ -34,10 +34,8 @@ namespaces.
|
||||
|
||||
1. Multiple active instances **may** share the same `client_type`.
|
||||
2. Every application launch receives a **distinct** `client_instance_id`.
|
||||
3. All namespace workers of one launch report the **same**
|
||||
`client_instance_id`. For a whole-fleet launch that is five workers; for a
|
||||
project-scoped launch it is exactly the namespaces that launch started
|
||||
(#985).
|
||||
3. All five namespace workers of one launch report the **same**
|
||||
`client_instance_id`.
|
||||
4. Each namespace worker has a **distinct** `worker_identity`, process
|
||||
identity, generation, and PID.
|
||||
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
|
||||
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
|
||||
|
||||
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).
|
||||
* `gitea_snapshot_instance_fleet` — sanctioned MCP tool (#978).
|
||||
* `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
|
||||
|
||||
|
||||
@@ -51,6 +51,7 @@ that gates each call, not which tools exist.
|
||||
- `gitea_adopt_merger_pr_lease`
|
||||
- `gitea_adopt_workflow_lease`
|
||||
- `gitea_allocate_next_work`
|
||||
- `gitea_apply_stale_worker_retirement`
|
||||
- `gitea_assess_already_landed_reconciliation`
|
||||
- `gitea_assess_conflict_fix_classification`
|
||||
- `gitea_assess_conflict_fix_push`
|
||||
@@ -124,6 +125,7 @@ that gates each call, not which tools exist.
|
||||
- `gitea_observability_link_issue`
|
||||
- `gitea_observability_list_projects`
|
||||
- `gitea_observability_reconcile_incident`
|
||||
- `gitea_plan_stale_worker_retirement`
|
||||
- `gitea_post_heartbeat`
|
||||
- `gitea_publish_unpublished_issue_branch`
|
||||
- `gitea_quarantine_contaminated_review`
|
||||
|
||||
@@ -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
@@ -1286,19 +1286,16 @@ def multi_namespace_launcher_entries(
|
||||
profile_by_namespace,
|
||||
*,
|
||||
client_type,
|
||||
namespaces=None,
|
||||
config_path=None,
|
||||
client_instance_id=None,
|
||||
fleet_run_id=None,
|
||||
session_id=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
|
||||
propagated to every namespace worker env. *namespaces* selects a validated
|
||||
project-scoped subset (#985); omitting it launches all five sanctioned
|
||||
namespaces as before. See
|
||||
propagated to every namespace worker env. See
|
||||
:func:`mcp_application_launcher.build_application_mcp_servers`.
|
||||
"""
|
||||
import mcp_application_launcher as app_launcher
|
||||
@@ -1306,7 +1303,6 @@ def multi_namespace_launcher_entries(
|
||||
return app_launcher.build_application_mcp_servers(
|
||||
profile_by_namespace,
|
||||
client_type=client_type,
|
||||
namespaces=namespaces,
|
||||
config_path=config_path or DEFAULT_CONFIG_PATH,
|
||||
client_instance_id=client_instance_id,
|
||||
fleet_run_id=fleet_run_id,
|
||||
|
||||
+699
-27
@@ -1052,16 +1052,9 @@ def _create_issue_bootstrap_assessment(
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
|
||||
remote_master_sha_error: str | None = None
|
||||
try:
|
||||
# #983: consume the resolved-target state so an *underivable* base ref
|
||||
# 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(
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
||||
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:
|
||||
remote_master_sha = None
|
||||
remote_master_sha_error = (
|
||||
@@ -1089,14 +1082,9 @@ def _create_issue_bootstrap_assessment(
|
||||
# closed instead of proceeding without base-equivalence proof.
|
||||
remote_master_sha_error: str | None = None
|
||||
try:
|
||||
# #983: same resolved-target consumption as the author bootstrap above —
|
||||
# a target whose base ref cannot be derived must say why.
|
||||
_base_state = root_checkout_guard.resolve_remote_master_ref_state(
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
||||
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:
|
||||
remote_master_sha = None
|
||||
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"]
|
||||
canonical_root = ctx["canonical_repo_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
|
||||
# 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")
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
|
||||
|
||||
# Repo/org facts (best-effort; explicit org/repo when provided).
|
||||
resolved_org = org
|
||||
@@ -1449,7 +1432,6 @@ def _run_anti_stomp_preflight(
|
||||
root_head_sha=git_state.get("head_sha"),
|
||||
root_porcelain=git_state.get("porcelain_status") or "",
|
||||
remote_master_sha=remote_master_sha,
|
||||
remote_master_ref=remote_master_ref,
|
||||
startup_head=startup_head,
|
||||
current_code_head=current_code_head,
|
||||
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"]
|
||||
workspace = ctx["workspace_path"]
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
|
||||
# #983: consume the resolved target so the contamination message names the
|
||||
# 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"]
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
|
||||
assessment = root_checkout_guard.assess_root_checkout_guard(
|
||||
workspace_path=workspace,
|
||||
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,
|
||||
resolved_role=_preflight_resolved_role,
|
||||
actual_role=_actual_profile_role(),
|
||||
remote_master_ref=base_state.get("ref"),
|
||||
)
|
||||
if assessment["block"]:
|
||||
raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
|
||||
@@ -19646,6 +19623,701 @@ def gitea_snapshot_instance_fleet(
|
||||
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()
|
||||
def gitea_get_runtime_context(
|
||||
remote: str = "dadeschools",
|
||||
|
||||
+11
-60
@@ -378,11 +378,6 @@ def format_parity(assessment: dict) -> str:
|
||||
# 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"
|
||||
|
||||
|
||||
@@ -413,7 +408,7 @@ def assess_target_repository_parity(
|
||||
*,
|
||||
canonical_root: str | None,
|
||||
source: str | None,
|
||||
tracking_ref: str | None = None,
|
||||
tracking_ref: str = DEFAULT_TARGET_TRACKING_REF,
|
||||
) -> dict:
|
||||
"""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
|
||||
single-repository default has no second dimension to be stale about. A
|
||||
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 = {
|
||||
"configured": bool(canonical_root),
|
||||
@@ -439,13 +429,9 @@ def assess_target_repository_parity(
|
||||
"repository_slug": None,
|
||||
"checkout_head": None,
|
||||
"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,
|
||||
"determinable": False,
|
||||
"stale": False,
|
||||
"reason_code": None,
|
||||
"reasons": [],
|
||||
}
|
||||
if not canonical_root:
|
||||
@@ -477,53 +463,18 @@ def assess_target_repository_parity(
|
||||
result["checkout_head"] = head
|
||||
result["determinable"] = True
|
||||
|
||||
# Local import keeps this module dependency-light for its startup role.
|
||||
import canonical_repository_root as _crr
|
||||
remote_url = _git_capture(canonical_root, "remote", "get-url", "origin")
|
||||
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
|
||||
# remote assumed to be named 'origin'. In a checkout whose only remote is
|
||||
# 'prgs', the old lookup failed outright and reported the identity as
|
||||
# underivable while a leftover refs/remotes/origin/master still resolved.
|
||||
#
|
||||
# #983 B2: this must be the *ambiguity-aware* resolver. The first-wins
|
||||
# `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"]
|
||||
parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url)
|
||||
if parsed:
|
||||
result["repository_slug"] = f"{parsed[0]}/{parsed[1]}"
|
||||
if not result["repository_slug"]:
|
||||
result["reasons"].append(
|
||||
"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)
|
||||
if not tracking_head:
|
||||
|
||||
+19
-379
@@ -15,7 +15,7 @@ from __future__ import annotations
|
||||
|
||||
import os
|
||||
import secrets
|
||||
from typing import Any, Mapping, Sequence
|
||||
from typing import Any, Mapping
|
||||
|
||||
import mcp_fleet_snapshot as fleet
|
||||
import mcp_worker_identity as mwi
|
||||
@@ -152,54 +152,10 @@ def namespace_worker_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(
|
||||
profile_by_namespace: Mapping[str, str],
|
||||
*,
|
||||
client_type: str | None,
|
||||
namespaces: "Sequence[str] | None" = None,
|
||||
config_path: str | None = None,
|
||||
client_instance_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
|
||||
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 = [
|
||||
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:
|
||||
raise ValueError(
|
||||
"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:
|
||||
@@ -265,13 +213,6 @@ def build_application_mcp_servers(
|
||||
"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.
|
||||
if command is None or args is None:
|
||||
import gitea_config
|
||||
@@ -282,7 +223,7 @@ def build_application_mcp_servers(
|
||||
|
||||
servers: dict[str, Any] = {}
|
||||
shared_id = launch["client_instance_id"]
|
||||
for namespace in resolved_namespaces:
|
||||
for namespace in SANCTIONED_NAMESPACES:
|
||||
server_name = f"gitea-{namespace}"
|
||||
profile = profile_by_namespace[namespace]
|
||||
env = namespace_worker_env(
|
||||
@@ -299,17 +240,14 @@ def build_application_mcp_servers(
|
||||
"env": env,
|
||||
}
|
||||
|
||||
excluded = [ns for ns in SANCTIONED_NAMESPACES if ns not in resolved_namespaces]
|
||||
return {
|
||||
"mcpServers": servers,
|
||||
"launch": launch,
|
||||
"client_instance_id": shared_id,
|
||||
"client_type": launch["client_type"],
|
||||
"namespaces": list(resolved_namespaces),
|
||||
"excluded_namespaces": excluded,
|
||||
"project_scoped": bool(excluded),
|
||||
"namespaces": list(SANCTIONED_NAMESPACES),
|
||||
"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(
|
||||
mcp_servers: Mapping[str, Any],
|
||||
*,
|
||||
namespaces: "Sequence[str] | None" = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Inspect a production mcpServers map for shared instance attribution.
|
||||
|
||||
Returns the unique set of client_instance_id values across the gitea-*
|
||||
servers of one launch and whether every worker of that launch shares
|
||||
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.
|
||||
Returns the unique set of client_instance_id values across gitea-* servers
|
||||
and whether all five namespaces share exactly one trusted ID.
|
||||
"""
|
||||
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] = []
|
||||
by_server: dict[str, str | None] = {}
|
||||
for name in expected_servers:
|
||||
for name in NAMESPACE_SERVER_NAMES:
|
||||
entry = mcp_servers.get(name) or {}
|
||||
env = entry.get("env") or {}
|
||||
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
|
||||
if fleet.assess_instance_identity(i)["trusted"]
|
||||
]
|
||||
missing_servers = [n for n in expected_servers if n not in mcp_servers]
|
||||
return {
|
||||
"server_instance_ids": by_server,
|
||||
"unique_instance_ids": unique,
|
||||
"trusted_instance_ids": trusted,
|
||||
"inspected_servers": list(expected_servers),
|
||||
"missing_servers": missing_servers,
|
||||
"shared_single_trusted_id": (
|
||||
bool(expected_servers)
|
||||
and not missing_servers
|
||||
and len(unique) == 1
|
||||
len(unique) == 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(
|
||||
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)
|
||||
raw = (source.get(CLIENT_INSTANCE_ENV) or "").strip() or None
|
||||
provenance_marker = (source.get(INSTANCE_PROVENANCE_ENV) or "").strip()
|
||||
assessment = dict(fleet.assess_instance_identity(raw))
|
||||
# The ``inst-…`` format is public and reproducible, so format alone cannot
|
||||
# establish trust: anyone able to set one env var could otherwise hand-write
|
||||
# a well-formed ID and be believed. Trust therefore requires the launcher's
|
||||
# own seal as well, and a well-formed but unsealed identity fails closed
|
||||
# rather than degrading to a mere annotation (#985).
|
||||
assessment = fleet.assess_instance_identity(raw)
|
||||
# Ordinary user-supplied values without launcher provenance marker are
|
||||
# still format-checked by assess_instance_identity. When the format is
|
||||
# trusted but the launcher marker is absent, keep the ID but note that
|
||||
# provenance is not launcher-sealed (operator hand-set or legacy config).
|
||||
if assessment["trusted"] and provenance_marker != LAUNCHER_PROVENANCE_VALUE:
|
||||
assessment = dict(assessment)
|
||||
assessment["launcher_sealed"] = False
|
||||
assessment["trusted"] = False
|
||||
assessment["complete"] = False
|
||||
assessment["provenance"] = fleet.INSTANCE_ID_PROVENANCE_UNSEALED
|
||||
assessment["reasons"] = list(assessment.get("reasons") or []) + [
|
||||
f"{INSTANCE_PROVENANCE_ENV} is not {LAUNCHER_PROVENANCE_VALUE!r}; "
|
||||
"identity format is valid but not sealed by the production "
|
||||
"launcher, so trusted attribution is refused"
|
||||
"identity format is valid but not sealed by the production launcher"
|
||||
]
|
||||
else:
|
||||
assessment = dict(assessment)
|
||||
assessment["launcher_sealed"] = bool(
|
||||
assessment["trusted"]
|
||||
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
|
||||
)
|
||||
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())
|
||||
|
||||
@@ -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"),
|
||||
}
|
||||
@@ -84,10 +84,6 @@ SANCTIONED_NAMESPACES = frozenset(
|
||||
INSTANCE_ID_PROVENANCE_TRUSTED = "trusted_launcher"
|
||||
INSTANCE_ID_PROVENANCE_LEGACY = "legacy_incomplete"
|
||||
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"
|
||||
FLEET_RUN_ENV = "GITEA_MCP_FLEET_RUN_ID"
|
||||
|
||||
@@ -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
@@ -39,7 +39,7 @@ import threading
|
||||
import time
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Iterator
|
||||
from typing import Any, Callable, Iterator, Sequence
|
||||
|
||||
|
||||
# --- Provenance verdicts -------------------------------------------------
|
||||
@@ -224,6 +224,13 @@ def _heartbeat_expectation_drift(
|
||||
STATUS_ACTIVE = "active"
|
||||
STATUS_SUPERSEDED = "superseded"
|
||||
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"})
|
||||
_FALSE_VALUES = frozenset({"0", "false", "no", "manual", "manual_launch"})
|
||||
@@ -315,13 +322,53 @@ _SCHEMA_OPTIONAL_COLUMNS: tuple[tuple[str, str], ...] = (
|
||||
("parity_revision", "TEXT"),
|
||||
("live_revision", "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):
|
||||
"""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:
|
||||
return datetime.now(timezone.utc)
|
||||
|
||||
@@ -791,6 +838,9 @@ class WorkerRegistry:
|
||||
parity_revision: str | None = None,
|
||||
live_revision: 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]:
|
||||
"""Atomically register one worker identity.
|
||||
|
||||
@@ -823,6 +873,20 @@ class WorkerRegistry:
|
||||
proc_id = process_identity or (
|
||||
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:
|
||||
existing = conn.execute(
|
||||
"SELECT * FROM worker_registrations WHERE worker_identity = ?",
|
||||
@@ -905,8 +969,9 @@ class WorkerRegistry:
|
||||
fencing_epoch, status,
|
||||
fleet_run_id, authenticated_account, process_identity,
|
||||
startup_revision, loaded_revision, parity_revision,
|
||||
live_revision, instance_id_provenance
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
live_revision, instance_id_provenance,
|
||||
host_id, boot_id, process_start_time
|
||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
""",
|
||||
(
|
||||
worker_identity,
|
||||
@@ -935,6 +1000,9 @@ class WorkerRegistry:
|
||||
(parity_revision or "").strip() or None,
|
||||
(live_revision 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(
|
||||
@@ -1217,6 +1285,363 @@ class WorkerRegistry:
|
||||
"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]:
|
||||
"""Registry row minus anything that should not travel to an LLM surface."""
|
||||
|
||||
+9
-146
@@ -1,16 +1,9 @@
|
||||
"""Root checkout guard (#475).
|
||||
|
||||
The project root checkout is the stable control checkout on its integration
|
||||
branch. Author/reviewer/merge flows must fail closed when the control checkout
|
||||
is contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of the
|
||||
tracking integration ref). 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.
|
||||
The project root checkout is the stable control checkout on master/prgs/master.
|
||||
Author/reviewer/merge flows must fail closed when the control checkout is
|
||||
contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
|
||||
prgs/master). Isolated ``branches/...`` worktrees remain allowed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -18,7 +11,6 @@ from __future__ import annotations
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
import canonical_repository_root
|
||||
from author_mutation_worktree import is_path_under_branches
|
||||
from reviewer_worktree import parse_dirty_tracked_files
|
||||
|
||||
@@ -28,65 +20,19 @@ REMEDIATION = (
|
||||
)
|
||||
|
||||
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")
|
||||
|
||||
|
||||
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(
|
||||
canonical_repo_root: str,
|
||||
*,
|
||||
remote_refs: tuple[str, ...] | None = None,
|
||||
explicit_remote: str | None = None,
|
||||
) -> str | None:
|
||||
"""Return the commit SHA for the tracking integration 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.
|
||||
"""
|
||||
"""Return the commit SHA for the tracking master ref when available."""
|
||||
root = (canonical_repo_root or "").strip()
|
||||
if not root:
|
||||
return None
|
||||
if remote_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:
|
||||
for ref in remote_refs or REMOTE_MASTER_REFS:
|
||||
res = subprocess.run(
|
||||
["git", "-C", root, "rev-parse", "--verify", ref],
|
||||
capture_output=True,
|
||||
@@ -100,84 +46,6 @@ def resolve_remote_master_sha(
|
||||
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
|
||||
|
||||
|
||||
@@ -191,9 +59,8 @@ def assess_root_checkout_guard(
|
||||
remote_master_sha: str | None,
|
||||
resolved_role: str | None = None,
|
||||
actual_role: str | None = None,
|
||||
remote_master_ref: str | None = None,
|
||||
) -> 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``
|
||||
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:
|
||||
# #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(
|
||||
f"control checkout HEAD does not match {ref_label} "
|
||||
f"(HEAD {head_sha[:12]}, {ref_label} {remote_master_sha[:12]})"
|
||||
"control checkout HEAD does not match prgs/master "
|
||||
f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
|
||||
@@ -175,6 +175,44 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
"permission": "gitea.read",
|
||||
"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.
|
||||
"clear_stale_binding": {
|
||||
"permission": "gitea.read",
|
||||
|
||||
@@ -1032,15 +1032,6 @@ class ClientHintsTests(unittest.TestCase):
|
||||
)
|
||||
|
||||
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
|
||||
|
||||
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_INSTANCE": inst,
|
||||
"GITEA_MCP_INSTANCE_PROVENANCE": (
|
||||
fleet.INSTANCE_ID_PROVENANCE_TRUSTED
|
||||
),
|
||||
"GITEA_MCP_FLEET_RUN_ID": "run-1",
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
hints = server._client_identity_hints()
|
||||
self.assertTrue(hints["instance_identity_trusted"])
|
||||
self.assertTrue(hints["instance_launcher_sealed"])
|
||||
self.assertEqual(hints["client_instance_id"], inst)
|
||||
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):
|
||||
import gitea_mcp_server as server
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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()
|
||||
@@ -87,27 +87,13 @@ class TestAssessRootCheckoutGuard(unittest.TestCase):
|
||||
self.assertTrue(result["block"])
|
||||
self.assertIn("tracked local edits", result["reasons"][0])
|
||||
|
||||
def test_head_behind_tracking_base_ref_blocked(self):
|
||||
"""#983: the base ref is derived, so the message no longer hardcodes PRGS."""
|
||||
def test_head_behind_prgs_master_blocked(self):
|
||||
result = self._assess(
|
||||
head_sha=OTHER_SHA,
|
||||
remote_master_sha=MASTER_SHA,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertIn(
|
||||
"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])
|
||||
self.assertIn("does not match prgs/master", result["reasons"][0])
|
||||
|
||||
def test_merger_requires_clean_control_checkout(self):
|
||||
result = self._assess(
|
||||
|
||||
Reference in New Issue
Block a user