Remediates reviewer findings F1, F2, and F3 from review 483 on PR #770. F1 (blocking): _current_runtime_mode_report derived the runtime-gate alignment input as realpath(workspace_root) == process_project_root, redefining workspace_roots_aligned. Its established meaning is ctx["roots_aligned"] (canonical_repo_root == process_project_root) — a repository-level question. Because the global worktree rule requires all task work to live in a branches/ worktree, the old derivation reported aligned False for exactly the sessions that are configured correctly, raising unsafe_process_root_workspace_alignment and refusing every non-read operation once an author, reviewer, or merger namespace held a worktree binding. The gate is now fed ctx["roots_aligned"]. F2 (blocking): _current_runtime_mode_report cached its first result into _STARTUP_RUNTIME_MODE, which was initialised to None rather than captured at import, and the read-only refresh=True path seeded it too. Session-scoped fields (active_task_workspace and its derived alignment) and mutable fields (dirty_files) were frozen for the process lifetime, so the acceptance criterion 7 dirty-runtime blocker stopped applying after the snapshot and one session's binding decided alignment for every later session. Immutable process facts (process root, branch, head, checkout-ness) are now captured at import as _STARTUP_RUNTIME_FACTS, matching the #420 parity baseline. Dirty state, workspace binding, and alignment are recomputed on every call. A new stable_control_runtime.observe_dirty_files() splits out the one runtime fact that legitimately changes during a process lifetime; observe_runtime() delegates to it so the parsing lives in one place. refresh is retained for the read-only reporting path but no longer selects a cache, so a read-only call can neither seed nor weaken a later mutation decision. F3 (major): no test exercised the real derivation — every server-wiring test asserting a healthy runtime permits mutations patched _current_runtime_mode_report with a fixture whose workspace_roots_aligned was True. New TestServerWiringRealDerivation patches only the derivation's inputs and lets the real function build the report: - a clean stable control checkout plus a correctly bound branches/ worktree permits an otherwise authorized mutation; - misaligned process/canonical roots fail closed; - newly dirty task state is detected after an earlier clean read; - a read-only refresh cannot freeze a permissive mutation result; - an unresolvable binding reports unknown alignment, never alignment proof. Acceptance criteria 6, 8, 10, and 11 are unchanged and untouched. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
+43
-28
@@ -2010,12 +2010,21 @@ import stable_control_runtime # noqa: E402
|
||||
# Read-only operations are never blocked by staleness.
|
||||
_STARTUP_PARITY = master_parity_gate.capture_startup_parity(PROJECT_ROOT)
|
||||
|
||||
# Stable-control runtime snapshot (#615): which runtime this process serves
|
||||
# from. Captured lazily on first use and then reused, so the mutation gate never
|
||||
# shells out per call. Whether the gate enforces at all is decided from how the
|
||||
# process was loaded -- a suite running from a branches/ worktree is dev-test by
|
||||
# design, and per-test production simulation must not switch this gate on.
|
||||
_STARTUP_RUNTIME_MODE: dict | None = None
|
||||
# Stable-control runtime facts (#615): which runtime this process serves from.
|
||||
# These are the *immutable* facts -- process root, branch, head, checkout-ness --
|
||||
# and they are captured at import exactly like the #420 parity baseline, because
|
||||
# the runtime a process serves from is fixed when it loads its code.
|
||||
#
|
||||
# Dirty state and the session's workspace binding are deliberately NOT captured
|
||||
# here. Both change during a process lifetime, so freezing them would retire the
|
||||
# acceptance-criterion-7 dirty-runtime blocker after the first snapshot and let
|
||||
# whichever session happened to call first decide alignment for every session
|
||||
# after it. _current_runtime_mode_report() re-evaluates them per call.
|
||||
#
|
||||
# Whether the gate enforces at all is decided from how the process was loaded --
|
||||
# a suite running from a branches/ worktree is dev-test by design, and per-test
|
||||
# production simulation must not switch this gate on.
|
||||
_STARTUP_RUNTIME_FACTS = stable_control_runtime.observe_runtime(PROJECT_ROOT)
|
||||
_RUNTIME_MODE_GATE_UNDER_TEST = "pytest" in sys.modules or "unittest" in sys.modules
|
||||
import worktree_cleanup_audit # noqa: E402
|
||||
import canonical_comment_validator as ccv # noqa: E402
|
||||
@@ -11372,18 +11381,22 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
|
||||
binding this session resolved, so the report answers both "what code is the
|
||||
control plane running" and "does it agree with the task workspace".
|
||||
|
||||
The mutation gate uses the **startup snapshot** (``refresh=False``), for the
|
||||
same reason the #420 parity gate compares against a startup baseline: the
|
||||
runtime a process serves from is fixed when it loads its code, and editing
|
||||
files afterwards does not change what is already in memory. That also keeps
|
||||
the mutation path free of per-call subprocess reads. ``refresh=True`` re-reads
|
||||
live state for the read-only reporting path.
|
||||
"""
|
||||
global _STARTUP_RUNTIME_MODE
|
||||
The immutable facts -- process root, branch, head, checkout-ness -- come from
|
||||
the import-time snapshot, for the same reason the #420 parity gate compares
|
||||
against a startup baseline: the runtime a process serves from is fixed when
|
||||
it loads its code, and editing files afterwards does not change what is
|
||||
already in memory.
|
||||
|
||||
if not refresh and _STARTUP_RUNTIME_MODE is not None:
|
||||
return _STARTUP_RUNTIME_MODE
|
||||
observed = stable_control_runtime.observe_runtime(PROJECT_ROOT)
|
||||
Everything mutation-sensitive is recomputed on **every** call: the dirty-file
|
||||
list, the session's workspace binding, and the alignment derived from it.
|
||||
Caching those would mean a checkout that goes dirty after the first call is
|
||||
never caught again, and that one session's binding decides alignment for the
|
||||
whole process. ``refresh`` is retained for the read-only reporting path; it
|
||||
no longer selects between a cache and live state, so a read-only call can
|
||||
neither seed nor weaken a later mutation decision.
|
||||
"""
|
||||
facts = _STARTUP_RUNTIME_FACTS
|
||||
dirty_files = stable_control_runtime.observe_dirty_files(PROJECT_ROOT)
|
||||
workspace_root = None
|
||||
aligned = None
|
||||
canonical_root = None
|
||||
@@ -11391,9 +11404,14 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
|
||||
ctx = _resolve_namespace_mutation_context(None)
|
||||
workspace_root = ctx.get("workspace_path")
|
||||
canonical_root = ctx.get("canonical_repo_root")
|
||||
aligned = os.path.realpath(workspace_root or "") == (
|
||||
ctx.get("process_project_root") or ""
|
||||
)
|
||||
# Alignment keeps its established repository-level meaning (#615 F1):
|
||||
# does this namespace target the repository the process is installed in,
|
||||
# i.e. canonical_repo_root == process_project_root. It is deliberately
|
||||
# NOT path equality between the task workspace and the process root --
|
||||
# the global worktree rule requires task work to live in a branches/
|
||||
# worktree, so that comparison would classify every correctly bound
|
||||
# author, reviewer, and merger session as unsafe.
|
||||
aligned = ctx.get("roots_aligned")
|
||||
except Exception:
|
||||
# An unresolvable binding is reported as unknown alignment, never as
|
||||
# proof of alignment.
|
||||
@@ -11402,21 +11420,18 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
|
||||
profile_name = get_profile()["profile_name"]
|
||||
except Exception:
|
||||
profile_name = None
|
||||
report = stable_control_runtime.build_runtime_report(
|
||||
return stable_control_runtime.build_runtime_report(
|
||||
process_root=PROJECT_ROOT,
|
||||
checkout_branch=observed["checkout_branch"],
|
||||
runtime_head=observed["runtime_head"],
|
||||
is_git_checkout=observed["is_git_checkout"],
|
||||
dirty_files=observed["dirty_files"],
|
||||
checkout_branch=facts["checkout_branch"],
|
||||
runtime_head=facts["runtime_head"],
|
||||
is_git_checkout=facts["is_git_checkout"],
|
||||
dirty_files=dirty_files,
|
||||
active_task_workspace=workspace_root,
|
||||
canonical_repository_root=canonical_root,
|
||||
workspace_roots_aligned=aligned,
|
||||
profile=profile_name,
|
||||
declared_mode=stable_control_runtime.declared_runtime_mode(),
|
||||
)
|
||||
if _STARTUP_RUNTIME_MODE is None:
|
||||
_STARTUP_RUNTIME_MODE = report
|
||||
return report
|
||||
|
||||
|
||||
def _runtime_mode_block(op: str) -> list[str]:
|
||||
|
||||
@@ -597,6 +597,20 @@ def _git_capture(root: str, *args: str) -> str | None:
|
||||
return (res.stdout or "").strip() or None
|
||||
|
||||
|
||||
def observe_dirty_files(process_root: str | None) -> list[str]:
|
||||
"""Read the live dirty-file list at *process_root*.
|
||||
|
||||
Split out from :func:`observe_runtime` because dirtiness is the one runtime
|
||||
fact that legitimately changes during a process lifetime. The mutation gate
|
||||
must re-read it per call rather than trust a startup snapshot, or a checkout
|
||||
that goes dirty after the snapshot is never blocked again (#615).
|
||||
"""
|
||||
if not process_root:
|
||||
return []
|
||||
porcelain = _git_capture(process_root, "status", "--porcelain") or ""
|
||||
return [line[3:].strip() for line in porcelain.splitlines() if line.strip()]
|
||||
|
||||
|
||||
def observe_runtime(process_root: str | None) -> dict:
|
||||
"""Read the runtime facts classification needs from *process_root*.
|
||||
|
||||
@@ -618,8 +632,7 @@ def observe_runtime(process_root: str | None) -> dict:
|
||||
branch = _git_capture(process_root, "rev-parse", "--abbrev-ref", "HEAD")
|
||||
if branch == "HEAD": # detached HEAD has no branch name
|
||||
branch = None
|
||||
porcelain = _git_capture(process_root, "status", "--porcelain") or ""
|
||||
dirty = [line[3:].strip() for line in porcelain.splitlines() if line.strip()]
|
||||
dirty = observe_dirty_files(process_root)
|
||||
return {
|
||||
"checkout_branch": branch,
|
||||
"runtime_head": _git_capture(process_root, "rev-parse", "HEAD"),
|
||||
|
||||
@@ -468,5 +468,136 @@ class TestServerWiring(unittest.TestCase):
|
||||
self.assertEqual(report["mcp_process_root"], self.srv.PROJECT_ROOT)
|
||||
|
||||
|
||||
class TestServerWiringRealDerivation(unittest.TestCase):
|
||||
"""Drive the *real* report derivation, not a pre-built fixture (#615 F3).
|
||||
|
||||
Every other server-wiring test patches ``_current_runtime_mode_report`` with
|
||||
a fixture, so the derivation the daemon actually runs was never executed by
|
||||
the suite. These tests patch only its *inputs* -- the import-time facts, the
|
||||
dirty-file read, and the resolved namespace binding -- and let the real
|
||||
function build the report.
|
||||
"""
|
||||
|
||||
TASK_WORKTREE = STABLE_ROOT + "/branches/issue-615-runtime-mode-enforcement"
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as srv # imported lazily: heavy module
|
||||
|
||||
self.srv = srv
|
||||
|
||||
def _stable_facts(self):
|
||||
"""Immutable facts of a promoted stable-control runtime."""
|
||||
return {
|
||||
"checkout_branch": "master",
|
||||
"runtime_head": SHA_A,
|
||||
"is_git_checkout": True,
|
||||
"dirty_files": [],
|
||||
}
|
||||
|
||||
def _binding(self, *, roots_aligned=True, workspace=None):
|
||||
"""A resolved namespace binding, as the server's resolver returns it."""
|
||||
return {
|
||||
"workspace_path": workspace or self.TASK_WORKTREE,
|
||||
"canonical_repo_root": STABLE_ROOT,
|
||||
"process_project_root": STABLE_ROOT,
|
||||
"roots_aligned": roots_aligned,
|
||||
}
|
||||
|
||||
def _real_derivation(self, *, dirty=None, roots_aligned=True, workspace=None):
|
||||
"""Context managers that patch only the inputs, never the derivation."""
|
||||
return (
|
||||
patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}),
|
||||
patch.object(self.srv, "PROJECT_ROOT", STABLE_ROOT),
|
||||
patch.object(self.srv, "_STARTUP_RUNTIME_FACTS", self._stable_facts()),
|
||||
patch.object(
|
||||
self.srv,
|
||||
"_resolve_namespace_mutation_context",
|
||||
return_value=self._binding(
|
||||
roots_aligned=roots_aligned, workspace=workspace),
|
||||
),
|
||||
patch.object(scr, "observe_dirty_files", return_value=list(dirty or [])),
|
||||
)
|
||||
|
||||
def test_clean_stable_checkout_with_bound_task_worktree_permits_mutation(self):
|
||||
# The sanctioned configuration: a clean control checkout on master plus a
|
||||
# correctly bound branches/ worktree. Before the F1 fix this failed, because
|
||||
# alignment was path equality between the task workspace and the process
|
||||
# root, which a branches/ worktree can never satisfy.
|
||||
env, root, facts, ctx, dirty = self._real_derivation()
|
||||
with env, root, facts, ctx, dirty:
|
||||
report = self.srv._current_runtime_mode_report()
|
||||
self.assertEqual(report["runtime_mode"], scr.RUNTIME_MODE_STABLE)
|
||||
self.assertEqual(report["active_task_workspace"], self.TASK_WORKTREE)
|
||||
self.assertTrue(report["workspace_roots_aligned"])
|
||||
self.assertTrue(
|
||||
report["real_mutations_allowed"], report["mutation_block_reasons"])
|
||||
self.assertEqual(self.srv._runtime_mode_block("gitea.pr.create"), [])
|
||||
|
||||
def test_misaligned_process_and_canonical_roots_fail_closed(self):
|
||||
# Alignment keeps its repository-level meaning: the namespace targeting a
|
||||
# different repository than the process is installed in is the unsafe case.
|
||||
env, root, facts, ctx, dirty = self._real_derivation(roots_aligned=False)
|
||||
with env, root, facts, ctx, dirty:
|
||||
report = self.srv._current_runtime_mode_report()
|
||||
self.assertFalse(report["workspace_roots_aligned"])
|
||||
self.assertFalse(report["real_mutations_allowed"])
|
||||
reasons = self.srv._runtime_mode_block("gitea.pr.create")
|
||||
self.assertTrue(reasons)
|
||||
self.assertTrue(any("alignment" in reason for reason in reasons))
|
||||
|
||||
def test_newly_dirty_task_state_is_detected_after_an_earlier_clean_read(self):
|
||||
# A clean read must not license every later mutation: the acceptance
|
||||
# criterion 7 dirty blocker has to keep applying for the process lifetime.
|
||||
env, root, facts, ctx, dirty = self._real_derivation(dirty=[])
|
||||
with env, root, facts, ctx, dirty:
|
||||
self.assertTrue(
|
||||
self.srv._current_runtime_mode_report()["real_mutations_allowed"])
|
||||
self.assertEqual(self.srv._runtime_mode_block("gitea.pr.create"), [])
|
||||
|
||||
env, root, facts, ctx, dirty = self._real_derivation(
|
||||
dirty=["gitea_mcp_server.py"])
|
||||
with env, root, facts, ctx, dirty:
|
||||
report = self.srv._current_runtime_mode_report()
|
||||
self.assertEqual(report["dirty_files"], ["gitea_mcp_server.py"])
|
||||
self.assertFalse(report["real_mutations_allowed"])
|
||||
self.assertTrue(self.srv._runtime_mode_block("gitea.pr.create"))
|
||||
|
||||
def test_read_only_refresh_cannot_freeze_a_permissive_mutation_result(self):
|
||||
# gitea_get_runtime_context() calls with refresh=True. That read-only call
|
||||
# must not seed a cache that a later mutation gate would then trust.
|
||||
env, root, facts, ctx, dirty = self._real_derivation(dirty=[])
|
||||
with env, root, facts, ctx, dirty:
|
||||
self.assertTrue(
|
||||
self.srv._current_runtime_mode_report(refresh=True)[
|
||||
"real_mutations_allowed"]
|
||||
)
|
||||
|
||||
env, root, facts, ctx, dirty = self._real_derivation(
|
||||
dirty=["stable_control_runtime.py"])
|
||||
with env, root, facts, ctx, dirty:
|
||||
self.assertFalse(
|
||||
self.srv._current_runtime_mode_report()["real_mutations_allowed"])
|
||||
self.assertTrue(self.srv._runtime_mode_block("gitea.pr.create"))
|
||||
|
||||
def test_unresolvable_binding_reports_unknown_alignment_never_alignment_proof(self):
|
||||
# An unresolvable binding must report alignment as unknown (None), never
|
||||
# as True. Only *definite* misalignment blocks: a session with no task
|
||||
# binding resolved is the ordinary case, and failing it closed would
|
||||
# reintroduce exactly the F1 breakage this change removes.
|
||||
env, root, facts, _, dirty = self._real_derivation()
|
||||
broken = patch.object(
|
||||
self.srv,
|
||||
"_resolve_namespace_mutation_context",
|
||||
side_effect=RuntimeError("no binding"),
|
||||
)
|
||||
with env, root, facts, broken, dirty:
|
||||
report = self.srv._current_runtime_mode_report()
|
||||
self.assertIsNone(report["workspace_roots_aligned"])
|
||||
self.assertNotIn(
|
||||
scr.BLOCKER_UNSAFE_ALIGNMENT,
|
||||
scr.assess_runtime_mutation_gate(report)["blocker_kinds"],
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user