Merge pull request 'feat: enforce stable-control vs dev runtime modes for Gitea MCP (Closes #615)' (#770) from feat/issue-615-runtime-mode-enforcement into master
This commit was merged in pull request #770.
This commit is contained in:
@@ -171,13 +171,18 @@ then:
|
||||
- Does not replace CI or code review for MCP changes
|
||||
- Does not authorize editing stable checkout “because tests need a quick fix”
|
||||
|
||||
## 5. Implementation follow-ups (optional tooling)
|
||||
## 5. Implementation follow-ups
|
||||
|
||||
These may land in later issues; the **policy binds sessions now**:
|
||||
The **policy binds sessions now**. The enforcement layer landed with issue #615
|
||||
acceptance criteria 6–11 in `stable_control_runtime.py`:
|
||||
|
||||
1. Session preflight that refuses mutations if workspace root equals a `branches/` feature worktree configured as “dev only.”
|
||||
2. Explicit `runtime_kind=stable|dev` in MCP config and `gitea_whoami` profile metadata.
|
||||
3. Promotion checklist script that emits the durable promotion marker fields.
|
||||
1. ~~Session preflight that refuses mutations if workspace root equals a `branches/` feature worktree configured as “dev only.”~~ **Landed.** `_runtime_mode_block()` refuses every mutating operation from a `dev-test`, dev-worktree-launched, dirty-stable, misaligned, or `unknown` runtime; `gitea.read` is never blocked, so an operator can still diagnose a sick runtime.
|
||||
2. ~~Explicit `runtime_kind=stable|dev` in MCP config and `gitea_whoami` profile metadata.~~ **Landed** as `runtime_mode` (`stable-control` | `dev-test` | `unknown`), reported by `gitea_get_runtime_context` under `stable_control_runtime` together with the runtime git SHA, branch, checkout path, process root, active workspace, alignment, dirty files, and `real_mutations_allowed`. Operators running a packaged layout with no git checkout declare the mode explicitly with `GITEA_MCP_RUNTIME_MODE`.
|
||||
3. ~~Promotion checklist script that emits the durable promotion marker fields.~~ **Landed** as `scripts/promote-stable-runtime` (read-only; emits and validates the record) plus [`../stable-runtime-promotion-runbook.md`](../stable-runtime-promotion-runbook.md).
|
||||
|
||||
Post-transport-flap proof is enforced per namespace: a flap invalidates every
|
||||
`gitea-*` namespace at once, and author proof never transfers to reviewer,
|
||||
merger, or reconciler (`namespace_not_reproven_after_flap`).
|
||||
|
||||
**Not optional (issue #615 acceptance criterion 2):** operator guide and runbooks **must** cross-link this ADR (see §6). Cross-links are documentation acceptance, not deferred tooling.
|
||||
|
||||
|
||||
@@ -1241,6 +1241,7 @@ When posting a Canonical Thread Handoff after a binding blocker:
|
||||
## Related documents
|
||||
|
||||
- [`architecture/mcp-stable-control-runtime-policy-adr.md`](architecture/mcp-stable-control-runtime-policy-adr.md) — stable control runtime vs dev runtime; LLM must not kill/restart MCP; operator-owned reload and promotions; routine post-merge parity staleness (#615).
|
||||
- [`stable-runtime-promotion-runbook.md`](stable-runtime-promotion-runbook.md) — operator promotion procedure, required promotion-record fields, per-namespace post-flap re-proving, and rollback for the stable control runtime (#615).
|
||||
- [`reviewer-handoff-consistency.md`](reviewer-handoff-consistency.md) — reject contradictory reviewer handoffs (#501).
|
||||
- [`issue-acceptance-gate.md`](issue-acceptance-gate.md) — controller issue-acceptance audit after PR merge (#500).
|
||||
- [`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) — portable cross-project LLM workflow skill.
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
# Stable control runtime — promotion runbook (#615)
|
||||
|
||||
Operator / release-manager procedure for promoting a revision into the **stable
|
||||
control runtime**: the Gitea MCP server that performs real issue/PR mutations.
|
||||
|
||||
Policy source: [`architecture/mcp-stable-control-runtime-policy-adr.md`](architecture/mcp-stable-control-runtime-policy-adr.md).
|
||||
Enforcement: `stable_control_runtime.py` (runtime mode classification, mutation
|
||||
gates, per-namespace post-flap re-proving, promotion-record validation).
|
||||
|
||||
**Promotion is operator-owned.** Normal author / reviewer / merger / reconciler
|
||||
sessions must never kill, restart, or relaunch the MCP server, and must never
|
||||
edit the stable runtime checkout. A session that needs newer server code stops
|
||||
with `BLOCKED + DIAGNOSE` and hands off to the operator.
|
||||
|
||||
---
|
||||
|
||||
## 1. When a promotion is required
|
||||
|
||||
- A merged PR changes MCP server code the control plane must now enforce.
|
||||
- `gitea_assess_master_parity` reports `stale: true` / `restart_required: true`.
|
||||
- `gitea_get_runtime_context` reports a `runtime_mode` other than
|
||||
`stable-control`, or `real_mutations_allowed: false`.
|
||||
|
||||
## 2. Pre-promotion checks
|
||||
|
||||
Run these **before** advancing the stable checkout:
|
||||
|
||||
1. The target revision is on remote `master` and was merged through
|
||||
`gitea_merge_pr` (never a direct stable-branch push — see #671).
|
||||
2. The stable control checkout is clean (`git status --porcelain` empty) and on
|
||||
`master`. A dirty stable runtime is itself a mutation blocker.
|
||||
3. The advance is strictly fast-forwardable: local `master` is an ancestor of
|
||||
`prgs/master`.
|
||||
4. No active workflow lease is mid-mutation (`gitea_list_workflow_leases`).
|
||||
|
||||
## 3. Promotion steps
|
||||
|
||||
1. Record the **previous** runtime SHA (`gitea_assess_master_parity` →
|
||||
`startup_head`).
|
||||
2. `git fetch --prune prgs` in the stable control checkout.
|
||||
3. `git merge --ff-only prgs/master` — never rebase, reset, or force.
|
||||
4. Record the **promoted** runtime SHA (`git rev-parse HEAD`).
|
||||
5. Reload the runtime using the sanctioned client path (IDE/client reconnect or
|
||||
the operator's supervised service reload). Never `pkill` the daemon from a
|
||||
workflow session.
|
||||
6. Re-prove **each** namespace independently (see §5).
|
||||
7. Record the promotion (see §4) and post it as a durable comment on the
|
||||
tracking issue.
|
||||
|
||||
## 4. Promotion record (required fields)
|
||||
|
||||
Every promotion must record all of the following. `assess_promotion_record()`
|
||||
validates them and fails closed on any missing field, or when
|
||||
`previous_runtime_sha` equals `promoted_runtime_sha` (nothing was promoted).
|
||||
|
||||
| Field | Meaning |
|
||||
|-------|---------|
|
||||
| `previous_runtime_sha` | SHA the stable runtime was serving before promotion |
|
||||
| `promoted_runtime_sha` | SHA the stable runtime serves after promotion |
|
||||
| `source_branch` | Branch the promoted revision came from |
|
||||
| `source_pr` | PR number that merged it |
|
||||
| `restart_method` | Exact reload/restart mechanism the operator used |
|
||||
| `health_check_proof` | `gitea_assess_mcp_namespace_health` result per namespace |
|
||||
| `identity_proof` | `gitea_whoami` username + profile per namespace |
|
||||
| `profile_proof` | `gitea_get_runtime_context` active profile per namespace |
|
||||
| `workspace_proof` | Process root, canonical root, alignment, clean state |
|
||||
| `mutation_capability_proof` | `gitea_resolve_task_capability` for the intended task |
|
||||
| `rollback_instructions` | Exact steps to return to `previous_runtime_sha` |
|
||||
|
||||
Helper: `scripts/promote-stable-runtime` emits and validates the record. It
|
||||
never restarts anything — it reads state and prints the record for the operator
|
||||
to act on and archive.
|
||||
|
||||
## 5. Post-promotion namespace re-proving
|
||||
|
||||
A restart or transport flap drops every `gitea-*` namespace together. Author
|
||||
proof is **not** global proof. For each of `author`, `reviewer`, `merger`,
|
||||
`reconciler`, in that namespace:
|
||||
|
||||
1. `gitea_whoami`
|
||||
2. `gitea_get_runtime_context`
|
||||
3. `gitea_resolve_task_capability` immediately before the intended mutation
|
||||
4. Mutate only when no reconnect / restart / stale-runtime gate is reported
|
||||
|
||||
Until a namespace passes all four, its mutations stay blocked with
|
||||
`namespace_not_reproven_after_flap`.
|
||||
|
||||
## 6. Rollback
|
||||
|
||||
If the promoted runtime is unhealthy — namespace EOF that does not recover,
|
||||
identity or profile mismatch, capability resolution failure, or an unexpected
|
||||
`runtime_mode`:
|
||||
|
||||
1. **Stop all PR/review/merge work.** An unhealthy stable runtime fails closed;
|
||||
do not route around it.
|
||||
2. Fast-forward or check out `previous_runtime_sha` in the stable checkout.
|
||||
3. Reload the runtime by the same sanctioned method.
|
||||
4. Re-prove every namespace (§5).
|
||||
5. Record the rollback as a promotion record whose `promoted_runtime_sha` is the
|
||||
restored SHA, with the failure evidence in `health_check_proof`.
|
||||
|
||||
## 7. Runtime modes seen in reports
|
||||
|
||||
| Mode | Meaning | Real mutations |
|
||||
|------|---------|----------------|
|
||||
| `stable-control` | Promoted revision, stable branch, clean checkout | Allowed |
|
||||
| `dev-test` | Launched from a `branches/` worktree or a feature branch | Blocked against production |
|
||||
| `unknown` | Root unresolvable, not a git checkout, or detached HEAD with no declaration | Blocked |
|
||||
|
||||
A packaged deployment with no git checkout must declare itself explicitly with
|
||||
`GITEA_MCP_RUNTIME_MODE=stable-control`; an unset or misspelled value falls back
|
||||
to inference and, failing that, to `unknown`.
|
||||
|
||||
## 8. Related
|
||||
|
||||
- `architecture/mcp-stable-control-runtime-policy-adr.md` — the policy (#615)
|
||||
- `mcp-namespace-health.md` — client-namespace health (#543)
|
||||
- `mcp-namespace-eof-recovery.md` — reconnect-only EOF recovery
|
||||
- `mcp-daemon-import-guard.md` — sanctioned daemon only (#558)
|
||||
- `bootstrap-review-path.md` — controller bootstrap when the live runtime cannot
|
||||
review its own fix (#557)
|
||||
+139
-1
@@ -2002,12 +2002,30 @@ import native_mcp_preference # noqa: E402
|
||||
import branch_cleanup_guard # noqa: E402
|
||||
import thread_state_ledger_validator # noqa: E402
|
||||
import master_parity_gate # noqa: E402
|
||||
import stable_control_runtime # noqa: E402
|
||||
|
||||
# Master-parity baseline (#420): the commit this server process started at.
|
||||
# Captured once so that, at mutation time, we can detect when the on-disk
|
||||
# master has advanced past the running code and fail closed until restart.
|
||||
# Read-only operations are never blocked by staleness.
|
||||
_STARTUP_PARITY = master_parity_gate.capture_startup_parity(PROJECT_ROOT)
|
||||
|
||||
# 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
|
||||
|
||||
@@ -11356,6 +11374,98 @@ def _current_master_parity() -> dict:
|
||||
return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
|
||||
|
||||
|
||||
def _current_runtime_mode_report(refresh: bool = False) -> dict:
|
||||
"""Describe the runtime this process is serving from (#615).
|
||||
|
||||
Combines the observed checkout facts at ``PROJECT_ROOT`` with the workspace
|
||||
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 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.
|
||||
|
||||
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
|
||||
try:
|
||||
ctx = _resolve_namespace_mutation_context(None)
|
||||
workspace_root = ctx.get("workspace_path")
|
||||
canonical_root = ctx.get("canonical_repo_root")
|
||||
# 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.
|
||||
aligned = None
|
||||
try:
|
||||
profile_name = get_profile()["profile_name"]
|
||||
except Exception:
|
||||
profile_name = None
|
||||
return stable_control_runtime.build_runtime_report(
|
||||
process_root=PROJECT_ROOT,
|
||||
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(),
|
||||
)
|
||||
|
||||
|
||||
def _runtime_mode_block(op: str) -> list[str]:
|
||||
"""Fail-closed runtime-mode reasons for a mutation *op* (#615).
|
||||
|
||||
Real Gitea mutations may only run on the promoted stable control runtime.
|
||||
Read-only operations are never blocked: a dev/test or unknown runtime can
|
||||
still be inspected, which is exactly how an operator diagnoses it.
|
||||
|
||||
Under unit-test isolation the gate is skipped unless the explicit #683
|
||||
force-on flag requests production behaviour: the suite legitimately runs
|
||||
from ``branches/`` worktrees, which classify as ``dev-test`` by design. That
|
||||
decision is taken from how the *process* was loaded, not from a per-call
|
||||
check, so a test that simulates production for some other guard does not
|
||||
silently switch this one on. The legacy dirty/porcelain force flags are
|
||||
deliberately not honoured here -- they request dirtiness paths, not a claim
|
||||
that the runtime is promoted.
|
||||
"""
|
||||
if op == "gitea.read":
|
||||
return []
|
||||
if _RUNTIME_MODE_GATE_UNDER_TEST and not (
|
||||
workflow_scope_guard.production_guards_forced()
|
||||
):
|
||||
return []
|
||||
try:
|
||||
report = _current_runtime_mode_report()
|
||||
except Exception as exc:
|
||||
return [
|
||||
"runtime mode could not be assessed (fail closed): "
|
||||
f"{_redact(str(exc))}"
|
||||
]
|
||||
return stable_control_runtime.runtime_block_reasons(report)
|
||||
|
||||
|
||||
def _master_parity_block(op: str) -> list[str]:
|
||||
"""Fail-closed staleness reasons for a mutation *op* (#420).
|
||||
|
||||
@@ -11380,11 +11490,16 @@ def _profile_operation_gate(op: str) -> list[str]:
|
||||
|
||||
A mutating operation is additionally refused when the server code is stale
|
||||
relative to master (#420), so a long-running process cannot bypass a
|
||||
capability gate that has since been merged.
|
||||
capability gate that has since been merged, and when the runtime itself is
|
||||
not the promoted stable control runtime (#615) -- a dev/test or unknown
|
||||
runtime holds production credentials but has not been promoted.
|
||||
"""
|
||||
stale_reasons = _master_parity_block(op)
|
||||
if stale_reasons:
|
||||
return stale_reasons
|
||||
runtime_reasons = _runtime_mode_block(op)
|
||||
if runtime_reasons:
|
||||
return runtime_reasons
|
||||
try:
|
||||
profile = get_profile()
|
||||
except Exception as exc:
|
||||
@@ -15027,6 +15142,29 @@ def gitea_get_runtime_context(
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# #615: runtime mode + SHA reporting, so a session can tell whether it is
|
||||
# talking to the promoted stable control runtime before it mutates.
|
||||
try:
|
||||
runtime_mode_report = _current_runtime_mode_report(refresh=True)
|
||||
result["stable_control_runtime"] = runtime_mode_report
|
||||
result["stable_control_runtime"]["summary"] = (
|
||||
stable_control_runtime.format_runtime_mode(runtime_mode_report)
|
||||
)
|
||||
# The advisory only replaces safe_next_action when the gate is live for
|
||||
# this process; under test isolation the report stays informational.
|
||||
gate_live = not _RUNTIME_MODE_GATE_UNDER_TEST or (
|
||||
workflow_scope_guard.production_guards_forced()
|
||||
)
|
||||
if gate_live and not runtime_mode_report["real_mutations_allowed"]:
|
||||
result["safe_next_action"] = (
|
||||
"Runtime is not the promoted stable control runtime "
|
||||
f"({stable_control_runtime.format_runtime_mode(runtime_mode_report)}); "
|
||||
"real mutations are blocked. Operator promotion is required — "
|
||||
"see docs/stable-runtime-promotion-runbook.md."
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
parity = _current_master_parity()
|
||||
result["master_parity"] = {
|
||||
"in_parity": parity["in_parity"],
|
||||
|
||||
Executable
+171
@@ -0,0 +1,171 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
usage() {
|
||||
cat <<'EOF'
|
||||
usage: scripts/promote-stable-runtime [--root <path>] [--promoted <sha>] \
|
||||
[--source-branch <branch>] [--source-pr <n>] \
|
||||
[--restart-method <text>] [--rollback <text>] \
|
||||
[--health-check-proof <text>] \
|
||||
[--identity-proof <text>] [--profile-proof <text>] \
|
||||
[--workspace-proof <text>] \
|
||||
[--mutation-capability-proof <text>]
|
||||
|
||||
Emit and validate a stable-control-runtime promotion record (#615).
|
||||
|
||||
This helper is READ-ONLY. It never fetches, merges, restarts, reloads, or kills
|
||||
anything: promotion itself is an operator action documented in
|
||||
docs/stable-runtime-promotion-runbook.md. The helper reads the current runtime
|
||||
state, assembles the required record, validates it with
|
||||
stable_control_runtime.assess_promotion_record(), and prints it for the operator
|
||||
to act on and archive.
|
||||
|
||||
Defaults:
|
||||
--root the repository root containing this script
|
||||
--promoted HEAD of that root
|
||||
|
||||
Exit status is non-zero when the assembled record is incomplete, so a promotion
|
||||
cannot be recorded without its proof fields.
|
||||
|
||||
Example:
|
||||
scripts/promote-stable-runtime \
|
||||
--source-branch feat/issue-615-runtime-mode-enforcement \
|
||||
--source-pr 770 \
|
||||
--restart-method "IDE client reconnect (/mcp)" \
|
||||
--rollback "git -C <root> merge --ff-only <previous-sha>; reconnect client" \
|
||||
--health-check-proof "gitea_assess_mcp_namespace_health: all four healthy" \
|
||||
--identity-proof "gitea_whoami per namespace" \
|
||||
--profile-proof "gitea_get_runtime_context per namespace" \
|
||||
--workspace-proof "process root == canonical root; clean" \
|
||||
--mutation-capability-proof "gitea_resolve_task_capability: allowed"
|
||||
EOF
|
||||
}
|
||||
|
||||
ROOT=""
|
||||
PROMOTED=""
|
||||
SOURCE_BRANCH=""
|
||||
SOURCE_PR=""
|
||||
RESTART_METHOD=""
|
||||
ROLLBACK=""
|
||||
HEALTH_PROOF=""
|
||||
IDENTITY_PROOF=""
|
||||
PROFILE_PROOF=""
|
||||
WORKSPACE_PROOF=""
|
||||
CAPABILITY_PROOF=""
|
||||
|
||||
while [[ $# -gt 0 ]]; do
|
||||
case "$1" in
|
||||
--root) ROOT="${2:-}"; shift 2 ;;
|
||||
--promoted) PROMOTED="${2:-}"; shift 2 ;;
|
||||
--source-branch) SOURCE_BRANCH="${2:-}"; shift 2 ;;
|
||||
--source-pr) SOURCE_PR="${2:-}"; shift 2 ;;
|
||||
--restart-method) RESTART_METHOD="${2:-}"; shift 2 ;;
|
||||
--rollback) ROLLBACK="${2:-}"; shift 2 ;;
|
||||
--health-check-proof) HEALTH_PROOF="${2:-}"; shift 2 ;;
|
||||
--identity-proof) IDENTITY_PROOF="${2:-}"; shift 2 ;;
|
||||
--profile-proof) PROFILE_PROOF="${2:-}"; shift 2 ;;
|
||||
--workspace-proof) WORKSPACE_PROOF="${2:-}"; shift 2 ;;
|
||||
--mutation-capability-proof) CAPABILITY_PROOF="${2:-}"; shift 2 ;;
|
||||
-h|--help) usage; exit 0 ;;
|
||||
*) echo "unknown argument: $1" >&2; usage; exit 2 ;;
|
||||
esac
|
||||
done
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT="${ROOT:-$(cd "$SCRIPT_DIR/.." && pwd)}"
|
||||
|
||||
if ! git -C "$ROOT" rev-parse --show-toplevel >/dev/null 2>&1; then
|
||||
echo "error: '$ROOT' is not a git checkout; cannot read runtime SHAs" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
PREVIOUS="${GITEA_MCP_PREVIOUS_RUNTIME_SHA:-}"
|
||||
if [[ -z "$PREVIOUS" ]]; then
|
||||
# The runtime the operator is replacing. Best-effort: the commit master
|
||||
# pointed at before the fast-forward, recorded in the reflog.
|
||||
PREVIOUS="$(git -C "$ROOT" rev-parse 'master@{1}' 2>/dev/null || true)"
|
||||
fi
|
||||
PROMOTED="${PROMOTED:-$(git -C "$ROOT" rev-parse HEAD)}"
|
||||
BRANCH="$(git -C "$ROOT" rev-parse --abbrev-ref HEAD)"
|
||||
DIRTY="$(git -C "$ROOT" status --porcelain | wc -l | tr -d ' ')"
|
||||
STAMP="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
|
||||
|
||||
if [[ -z "$WORKSPACE_PROOF" ]]; then
|
||||
WORKSPACE_PROOF="root=$ROOT branch=$BRANCH dirty_files=$DIRTY"
|
||||
fi
|
||||
if [[ -z "$ROLLBACK" && -n "$PREVIOUS" ]]; then
|
||||
ROLLBACK="git -C $ROOT merge --ff-only $PREVIOUS (or checkout $PREVIOUS), then reload the runtime by the same sanctioned method and re-prove every namespace"
|
||||
fi
|
||||
|
||||
cat <<EOF
|
||||
# Stable control runtime promotion record (#615)
|
||||
# Generated $STAMP by scripts/promote-stable-runtime (read-only)
|
||||
|
||||
previous_runtime_sha: ${PREVIOUS:-<MISSING: record the SHA the runtime served before promotion>}
|
||||
promoted_runtime_sha: ${PROMOTED}
|
||||
source_branch: ${SOURCE_BRANCH:-<MISSING: pass --source-branch>}
|
||||
source_pr: ${SOURCE_PR:-<MISSING: pass --source-pr>}
|
||||
restart_method: ${RESTART_METHOD:-<MISSING: pass --restart-method>}
|
||||
health_check_proof: ${HEALTH_PROOF:-<MISSING: pass --health-check-proof>}
|
||||
identity_proof: ${IDENTITY_PROOF:-<MISSING: pass --identity-proof>}
|
||||
profile_proof: ${PROFILE_PROOF:-<MISSING: pass --profile-proof>}
|
||||
workspace_proof: ${WORKSPACE_PROOF}
|
||||
mutation_capability_proof: ${CAPABILITY_PROOF:-<MISSING: pass --mutation-capability-proof>}
|
||||
rollback_instructions: ${ROLLBACK:-<MISSING: pass --rollback>}
|
||||
EOF
|
||||
|
||||
if [[ "$DIRTY" != "0" ]]; then
|
||||
{
|
||||
echo
|
||||
echo "WARNING: the stable checkout has $DIRTY dirty file(s); a dirty stable"
|
||||
echo " runtime is itself a mutation blocker (dirty_stable_runtime_checkout)."
|
||||
} >&2
|
||||
fi
|
||||
|
||||
PYTHON_BIN="${PYTHON_BIN:-python3}"
|
||||
RECORD_JSON="$(
|
||||
ROOT="$ROOT" \
|
||||
PREVIOUS="$PREVIOUS" PROMOTED="$PROMOTED" \
|
||||
SOURCE_BRANCH="$SOURCE_BRANCH" SOURCE_PR="$SOURCE_PR" \
|
||||
RESTART_METHOD="$RESTART_METHOD" HEALTH_PROOF="$HEALTH_PROOF" \
|
||||
IDENTITY_PROOF="$IDENTITY_PROOF" PROFILE_PROOF="$PROFILE_PROOF" \
|
||||
WORKSPACE_PROOF="$WORKSPACE_PROOF" CAPABILITY_PROOF="$CAPABILITY_PROOF" \
|
||||
ROLLBACK="$ROLLBACK" \
|
||||
"$PYTHON_BIN" -c '
|
||||
import json
|
||||
import os
|
||||
|
||||
print(json.dumps({
|
||||
"previous_runtime_sha": os.environ.get("PREVIOUS", ""),
|
||||
"promoted_runtime_sha": os.environ.get("PROMOTED", ""),
|
||||
"source_branch": os.environ.get("SOURCE_BRANCH", ""),
|
||||
"source_pr": os.environ.get("SOURCE_PR", ""),
|
||||
"restart_method": os.environ.get("RESTART_METHOD", ""),
|
||||
"health_check_proof": os.environ.get("HEALTH_PROOF", ""),
|
||||
"identity_proof": os.environ.get("IDENTITY_PROOF", ""),
|
||||
"profile_proof": os.environ.get("PROFILE_PROOF", ""),
|
||||
"workspace_proof": os.environ.get("WORKSPACE_PROOF", ""),
|
||||
"mutation_capability_proof": os.environ.get("CAPABILITY_PROOF", ""),
|
||||
"rollback_instructions": os.environ.get("ROLLBACK", ""),
|
||||
}))
|
||||
'
|
||||
)"
|
||||
|
||||
RECORD_JSON="$RECORD_JSON" ROOT="$ROOT" "$PYTHON_BIN" -c '
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.environ["ROOT"])
|
||||
import stable_control_runtime as scr
|
||||
|
||||
record = json.loads(os.environ["RECORD_JSON"])
|
||||
assessment = scr.assess_promotion_record(record)
|
||||
print()
|
||||
print("validation:", json.dumps(assessment, indent=2))
|
||||
print()
|
||||
if not assessment["valid"]:
|
||||
print("Promotion record is INCOMPLETE - do not archive it as a promotion.")
|
||||
sys.exit(1)
|
||||
print("Promotion record is complete. Archive it on the tracking issue.")
|
||||
'
|
||||
@@ -0,0 +1,641 @@
|
||||
"""Stable-control vs dev/test runtime classification and mutation gates (#615).
|
||||
|
||||
``docs/architecture/mcp-stable-control-runtime-policy-adr.md`` states the policy:
|
||||
real Gitea mutations may only be performed by the **stable control runtime**,
|
||||
while MCP server development happens in isolated ``branches/`` worktrees and
|
||||
optional dev/test runtimes. The ADR alone is not enforcement — a daemon
|
||||
relaunched from a feature worktree still holds production credentials and will
|
||||
happily mutate production issues.
|
||||
|
||||
This module supplies the runtime half of that policy:
|
||||
|
||||
* :func:`classify_runtime_mode` decides whether the running process is a
|
||||
``stable-control``, ``dev-test``, or ``unknown`` runtime.
|
||||
* :func:`build_runtime_report` collects the reporting fields the ADR requires
|
||||
(mode, SHA, branch, checkout path, process root, workspace, binding, dirty
|
||||
files, alignment, and whether real mutations are allowed).
|
||||
* :func:`assess_runtime_mutation_gate` turns that report into a fail-closed
|
||||
mutation gate.
|
||||
* The post-transport-flap helpers keep namespace re-proving **per namespace**,
|
||||
so proving the author namespace never implies the reviewer, merger, or
|
||||
reconciler namespace is callable.
|
||||
|
||||
Every assessment is pure: callers inject the observed facts, so the logic is
|
||||
unit-testable without a git checkout or a live daemon. Only the thin
|
||||
:func:`observe_runtime` reader touches the filesystem.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
# Operator declaration of the runtime this process is. An explicit, valid
|
||||
# declaration wins over inference — an operator running a packaged release
|
||||
# layout may have no git checkout to infer from.
|
||||
ENV_RUNTIME_MODE = "GITEA_MCP_RUNTIME_MODE"
|
||||
# Escape hatch mirroring the #420 parity gate: disables enforcement only.
|
||||
ENV_DISABLE = "GITEA_MCP_DISABLE_RUNTIME_MODE_GATE"
|
||||
|
||||
RUNTIME_MODE_STABLE = "stable-control"
|
||||
RUNTIME_MODE_DEV_TEST = "dev-test"
|
||||
RUNTIME_MODE_UNKNOWN = "unknown"
|
||||
|
||||
VALID_RUNTIME_MODES = frozenset(
|
||||
{RUNTIME_MODE_STABLE, RUNTIME_MODE_DEV_TEST, RUNTIME_MODE_UNKNOWN}
|
||||
)
|
||||
|
||||
# Branches a stable control checkout is allowed to sit on. Anything else is a
|
||||
# development checkout by definition (the global worktree rule keeps the
|
||||
# control checkout on a stable branch).
|
||||
STABLE_BRANCHES = frozenset({"master", "main", "dev"})
|
||||
|
||||
# Path segment that marks an isolated development worktree.
|
||||
DEV_WORKTREE_SEGMENT = "branches"
|
||||
|
||||
BLOCKER_DEV_TEST_PRODUCTION = "dev_test_runtime_targets_production"
|
||||
BLOCKER_UNKNOWN_RUNTIME = "unknown_runtime_mode"
|
||||
BLOCKER_DIRTY_STABLE_RUNTIME = "dirty_stable_runtime_checkout"
|
||||
BLOCKER_DEV_WORKTREE_LAUNCH = "runtime_launched_from_dev_worktree"
|
||||
BLOCKER_UNSAFE_ALIGNMENT = "unsafe_process_root_workspace_alignment"
|
||||
BLOCKER_NAMESPACE_NOT_REPROVEN = "namespace_not_reproven_after_flap"
|
||||
|
||||
# Namespaces that must each be re-proven independently after a transport flap.
|
||||
WORKFLOW_NAMESPACES = ("author", "reviewer", "merger", "reconciler")
|
||||
|
||||
|
||||
def gate_disabled() -> bool:
|
||||
"""Whether the runtime-mode gate is disabled by env escape hatch."""
|
||||
return bool((os.environ.get(ENV_DISABLE) or "").strip())
|
||||
|
||||
|
||||
def declared_runtime_mode() -> str | None:
|
||||
"""Return the operator-declared runtime mode, if a valid one is set.
|
||||
|
||||
An unset or unrecognised value returns ``None`` so classification falls
|
||||
back to inference rather than trusting a typo.
|
||||
"""
|
||||
value = (os.environ.get(ENV_RUNTIME_MODE) or "").strip().lower()
|
||||
if value in VALID_RUNTIME_MODES:
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _path_segments(path: str) -> list[str]:
|
||||
return [seg for seg in os.path.normpath(path).split(os.sep) if seg]
|
||||
|
||||
|
||||
def launched_from_dev_worktree(process_root: str | None) -> bool:
|
||||
"""Whether *process_root* sits inside a ``branches/`` development worktree."""
|
||||
if not process_root:
|
||||
return False
|
||||
return DEV_WORKTREE_SEGMENT in _path_segments(process_root)
|
||||
|
||||
|
||||
def classify_runtime_mode(
|
||||
*,
|
||||
process_root: str | None,
|
||||
checkout_branch: str | None,
|
||||
is_git_checkout: bool = True,
|
||||
declared_mode: str | None = None,
|
||||
) -> dict:
|
||||
"""Classify the runtime this process is serving from.
|
||||
|
||||
``declared_mode`` (normally :func:`declared_runtime_mode`) is authoritative
|
||||
when supplied and valid. Otherwise the mode is inferred:
|
||||
|
||||
* no resolvable root, or a root that is not a git checkout → ``unknown``
|
||||
(a packaged deployment must declare its mode explicitly);
|
||||
* a root inside a ``branches/`` worktree → ``dev-test``;
|
||||
* an unreadable branch → ``unknown``;
|
||||
* a stable branch (``master``/``main``/``dev``) → ``stable-control``;
|
||||
* any other branch → ``dev-test``.
|
||||
|
||||
Returns the mode plus the discriminating facts and human-readable reasons.
|
||||
"""
|
||||
reasons: list[str] = []
|
||||
dev_worktree = launched_from_dev_worktree(process_root)
|
||||
|
||||
if declared_mode in VALID_RUNTIME_MODES:
|
||||
reasons.append(
|
||||
f"runtime mode declared by operator via {ENV_RUNTIME_MODE}="
|
||||
f"{declared_mode}"
|
||||
)
|
||||
return _mode_result(declared_mode, dev_worktree, True, reasons)
|
||||
|
||||
if not process_root:
|
||||
reasons.append(
|
||||
"runtime process root could not be resolved; runtime mode is "
|
||||
"indeterminate"
|
||||
)
|
||||
return _mode_result(RUNTIME_MODE_UNKNOWN, dev_worktree, False, reasons)
|
||||
|
||||
if not is_git_checkout:
|
||||
reasons.append(
|
||||
f"runtime process root '{process_root}' is not a git checkout and "
|
||||
f"no {ENV_RUNTIME_MODE} declaration was supplied"
|
||||
)
|
||||
return _mode_result(RUNTIME_MODE_UNKNOWN, dev_worktree, False, reasons)
|
||||
|
||||
if dev_worktree:
|
||||
reasons.append(
|
||||
f"runtime was launched from development worktree '{process_root}' "
|
||||
f"(inside '{DEV_WORKTREE_SEGMENT}/')"
|
||||
)
|
||||
return _mode_result(RUNTIME_MODE_DEV_TEST, True, False, reasons)
|
||||
|
||||
branch = (checkout_branch or "").strip()
|
||||
if not branch:
|
||||
reasons.append(
|
||||
f"runtime checkout branch at '{process_root}' could not be read "
|
||||
f"(detached HEAD or unreadable); runtime mode is indeterminate"
|
||||
)
|
||||
return _mode_result(RUNTIME_MODE_UNKNOWN, False, False, reasons)
|
||||
|
||||
if branch in STABLE_BRANCHES:
|
||||
reasons.append(
|
||||
f"runtime checkout '{process_root}' is on stable branch '{branch}'"
|
||||
)
|
||||
return _mode_result(RUNTIME_MODE_STABLE, False, False, reasons)
|
||||
|
||||
reasons.append(
|
||||
f"runtime checkout '{process_root}' is on development branch "
|
||||
f"'{branch}', not a stable branch "
|
||||
f"({', '.join(sorted(STABLE_BRANCHES))})"
|
||||
)
|
||||
return _mode_result(RUNTIME_MODE_DEV_TEST, False, False, reasons)
|
||||
|
||||
|
||||
def _mode_result(mode, dev_worktree, declared, reasons) -> dict:
|
||||
return {
|
||||
"runtime_mode": mode,
|
||||
"dev_worktree_launched": bool(dev_worktree),
|
||||
"declared": bool(declared),
|
||||
"reasons": list(reasons),
|
||||
}
|
||||
|
||||
|
||||
def build_runtime_report(
|
||||
*,
|
||||
process_root: str | None,
|
||||
checkout_branch: str | None,
|
||||
runtime_head: str | None,
|
||||
active_task_workspace: str | None = None,
|
||||
canonical_repository_root: str | None = None,
|
||||
repository_slug: str | None = None,
|
||||
profile: str | None = None,
|
||||
authenticated_identity: str | None = None,
|
||||
dirty_files: list[str] | tuple[str, ...] | None = None,
|
||||
workspace_roots_aligned: bool | None = None,
|
||||
is_git_checkout: bool = True,
|
||||
declared_mode: str | None = None,
|
||||
) -> dict:
|
||||
"""Build the ADR-required runtime report (#615 acceptance criterion 6).
|
||||
|
||||
``real_mutations_allowed`` is the summary bit: it is true only when the
|
||||
corresponding mutation gate finds nothing to block on for a production
|
||||
target.
|
||||
"""
|
||||
classification = classify_runtime_mode(
|
||||
process_root=process_root,
|
||||
checkout_branch=checkout_branch,
|
||||
is_git_checkout=is_git_checkout,
|
||||
declared_mode=declared_mode,
|
||||
)
|
||||
report = {
|
||||
"runtime_mode": classification["runtime_mode"],
|
||||
"runtime_mode_declared": classification["declared"],
|
||||
"runtime_mode_reasons": classification["reasons"],
|
||||
"dev_worktree_launched": classification["dev_worktree_launched"],
|
||||
"runtime_git_sha": runtime_head,
|
||||
"runtime_branch": checkout_branch,
|
||||
"runtime_checkout_path": process_root,
|
||||
"mcp_process_root": process_root,
|
||||
"active_task_workspace": active_task_workspace,
|
||||
"canonical_repository_root": canonical_repository_root,
|
||||
"repository_slug": repository_slug,
|
||||
"profile": profile,
|
||||
"authenticated_identity": authenticated_identity,
|
||||
"dirty_files": sorted(dirty_files or []),
|
||||
"workspace_roots_aligned": workspace_roots_aligned,
|
||||
"gate_enforced": not gate_disabled(),
|
||||
}
|
||||
gate = assess_runtime_mutation_gate(report)
|
||||
report["real_mutations_allowed"] = not gate["block"]
|
||||
report["mutation_block_reasons"] = gate["reasons"]
|
||||
return report
|
||||
|
||||
|
||||
def assess_runtime_mutation_gate(
|
||||
report: dict,
|
||||
*,
|
||||
target_is_production: bool = True,
|
||||
namespace: str | None = None,
|
||||
namespace_reproof: dict | None = None,
|
||||
) -> dict:
|
||||
"""Fail-closed mutation gate for the runtime a mutation would execute in.
|
||||
|
||||
Blocks when (acceptance criterion 7):
|
||||
|
||||
* the runtime is ``dev-test`` and the mutation targets the production
|
||||
repository;
|
||||
* the runtime mode is ``unknown``;
|
||||
* the stable runtime checkout is dirty;
|
||||
* the runtime was launched from a development worktree;
|
||||
* process-root / workspace alignment is unsafe;
|
||||
* (criterion 8) the namespace has not been re-proven since a transport flap.
|
||||
|
||||
The disabled escape hatch never blocks; the caller decides read-vs-mutate
|
||||
before calling.
|
||||
"""
|
||||
reasons: list[str] = []
|
||||
blockers: list[str] = []
|
||||
|
||||
if gate_disabled():
|
||||
return _gate_result(False, blockers, reasons, disabled=True)
|
||||
|
||||
mode = (report or {}).get("runtime_mode")
|
||||
|
||||
if mode == RUNTIME_MODE_UNKNOWN:
|
||||
blockers.append(BLOCKER_UNKNOWN_RUNTIME)
|
||||
reasons.append(
|
||||
"runtime mode is 'unknown'; a runtime that cannot prove it is the "
|
||||
f"stable control runtime must not mutate production (declare "
|
||||
f"{ENV_RUNTIME_MODE} or run from a stable checkout)"
|
||||
)
|
||||
|
||||
if mode == RUNTIME_MODE_DEV_TEST and target_is_production:
|
||||
blockers.append(BLOCKER_DEV_TEST_PRODUCTION)
|
||||
reasons.append(
|
||||
"runtime mode is 'dev-test' and the mutation targets the "
|
||||
"production repository; dev/test runtimes must not mutate real "
|
||||
"issues or PRs (ADR: stable control runtime vs dev runtime)"
|
||||
)
|
||||
|
||||
if report.get("dev_worktree_launched") and target_is_production:
|
||||
blockers.append(BLOCKER_DEV_WORKTREE_LAUNCH)
|
||||
reasons.append(
|
||||
f"runtime was launched from a '{DEV_WORKTREE_SEGMENT}/' development "
|
||||
f"worktree ('{report.get('mcp_process_root')}'); production "
|
||||
f"mutations require the promoted stable control runtime"
|
||||
)
|
||||
|
||||
dirty = list(report.get("dirty_files") or [])
|
||||
if mode == RUNTIME_MODE_STABLE and dirty:
|
||||
blockers.append(BLOCKER_DIRTY_STABLE_RUNTIME)
|
||||
reasons.append(
|
||||
"stable control runtime checkout is dirty "
|
||||
f"({len(dirty)} file(s): {', '.join(dirty[:5])}"
|
||||
f"{'...' if len(dirty) > 5 else ''}); the control plane must run "
|
||||
"promoted, unmodified code"
|
||||
)
|
||||
|
||||
if report.get("workspace_roots_aligned") is False:
|
||||
blockers.append(BLOCKER_UNSAFE_ALIGNMENT)
|
||||
reasons.append(
|
||||
"process-root / active-workspace alignment is unsafe; the runtime "
|
||||
"and the task workspace disagree about which checkout is being "
|
||||
"mutated"
|
||||
)
|
||||
|
||||
if namespace:
|
||||
reproof = assess_namespace_reproof(namespace_reproof, namespace)
|
||||
if reproof["reproof_required"] and not reproof["proven"]:
|
||||
blockers.append(BLOCKER_NAMESPACE_NOT_REPROVEN)
|
||||
reasons.extend(reproof["reasons"])
|
||||
|
||||
return _gate_result(bool(blockers), blockers, reasons)
|
||||
|
||||
|
||||
def _gate_result(block, blockers, reasons, *, disabled=False) -> dict:
|
||||
return {
|
||||
"block": bool(block),
|
||||
"blocker_kinds": list(blockers),
|
||||
"blocker_kind": blockers[0] if blockers else None,
|
||||
"reasons": list(reasons),
|
||||
"gate_disabled": bool(disabled),
|
||||
}
|
||||
|
||||
|
||||
def runtime_block_reasons(
|
||||
report: dict,
|
||||
*,
|
||||
target_is_production: bool = True,
|
||||
namespace: str | None = None,
|
||||
namespace_reproof: dict | None = None,
|
||||
) -> list[str]:
|
||||
"""Block reasons for a mutation gate (empty when the mutation may proceed)."""
|
||||
gate = assess_runtime_mutation_gate(
|
||||
report,
|
||||
target_is_production=target_is_production,
|
||||
namespace=namespace,
|
||||
namespace_reproof=namespace_reproof,
|
||||
)
|
||||
return gate["reasons"]
|
||||
|
||||
|
||||
def runtime_report_payload(report: dict, gate: dict | None = None) -> dict:
|
||||
"""Structured recovery payload for permission-block responses."""
|
||||
gate = gate or assess_runtime_mutation_gate(report)
|
||||
return {
|
||||
"kind": "runtime_mode_block",
|
||||
"runtime_mode": report.get("runtime_mode"),
|
||||
"runtime_git_sha": report.get("runtime_git_sha"),
|
||||
"runtime_branch": report.get("runtime_branch"),
|
||||
"runtime_checkout_path": report.get("runtime_checkout_path"),
|
||||
"blocker_kind": gate.get("blocker_kind"),
|
||||
"blocker_kinds": list(gate.get("blocker_kinds") or []),
|
||||
"reasons": list(gate.get("reasons") or []),
|
||||
"recovery": [
|
||||
"Real workflow mutations run only on the promoted stable control "
|
||||
"runtime (see docs/architecture/"
|
||||
"mcp-stable-control-runtime-policy-adr.md).",
|
||||
"Operator action: promote the intended revision into the stable "
|
||||
"runtime and reload it — see "
|
||||
"docs/stable-runtime-promotion-runbook.md.",
|
||||
"Normal author/reviewer/merger/reconciler sessions must not kill, "
|
||||
"restart, or relaunch the MCP server themselves.",
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def format_runtime_mode(report: dict) -> str:
|
||||
"""One-line human summary for logs / runtime context."""
|
||||
mode = report.get("runtime_mode") or RUNTIME_MODE_UNKNOWN
|
||||
sha = report.get("runtime_git_sha")
|
||||
branch = report.get("runtime_branch") or "unknown-branch"
|
||||
short = sha[:12] if sha else "unknown-sha"
|
||||
suffix = (
|
||||
"" if report.get("real_mutations_allowed", True) else " (mutations blocked)"
|
||||
)
|
||||
return f"{mode} at {short} on {branch}{suffix}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Post-transport-flap namespace re-proving (#615 acceptance criterion 8)
|
||||
#
|
||||
# A transport flap (#584) drops every gitea-* namespace at once. Proving the
|
||||
# author namespace afterwards says nothing about the reviewer, merger, or
|
||||
# reconciler namespace, so proof is tracked per namespace and a flap
|
||||
# invalidates all of them.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
REQUIRED_NAMESPACE_PROOF_STEPS = (
|
||||
"whoami",
|
||||
"runtime_context",
|
||||
"capability_resolved",
|
||||
)
|
||||
|
||||
|
||||
def new_reproof_state() -> dict:
|
||||
"""Return an empty post-flap re-proving state."""
|
||||
return {"flap_at": None, "namespaces": {}}
|
||||
|
||||
|
||||
def record_transport_flap(state: dict | None, *, at: str) -> dict:
|
||||
"""Record a transport flap: every namespace must be re-proven after *at*.
|
||||
|
||||
Existing per-namespace proofs are kept for audit but no longer satisfy the
|
||||
gate, because they were recorded before the flap.
|
||||
"""
|
||||
result = dict(state or new_reproof_state())
|
||||
result["flap_at"] = at
|
||||
result["namespaces"] = dict(result.get("namespaces") or {})
|
||||
return result
|
||||
|
||||
|
||||
def record_namespace_proof(
|
||||
state: dict | None,
|
||||
namespace: str,
|
||||
*,
|
||||
at: str,
|
||||
whoami: bool = False,
|
||||
runtime_context: bool = False,
|
||||
capability_resolved: bool = False,
|
||||
stale_runtime_reported: bool = False,
|
||||
) -> dict:
|
||||
"""Record proof steps completed for exactly one namespace.
|
||||
|
||||
A namespace whose proof reported a reconnect/restart/stale-runtime gate is
|
||||
never counted as proven, regardless of which steps ran.
|
||||
"""
|
||||
result = dict(state or new_reproof_state())
|
||||
namespaces = dict(result.get("namespaces") or {})
|
||||
namespaces[(namespace or "").strip()] = {
|
||||
"at": at,
|
||||
"whoami": bool(whoami),
|
||||
"runtime_context": bool(runtime_context),
|
||||
"capability_resolved": bool(capability_resolved),
|
||||
"stale_runtime_reported": bool(stale_runtime_reported),
|
||||
}
|
||||
result["namespaces"] = namespaces
|
||||
return result
|
||||
|
||||
|
||||
def assess_namespace_reproof(state: dict | None, namespace: str) -> dict:
|
||||
"""Whether *namespace* is re-proven after the most recent transport flap.
|
||||
|
||||
``reproof_required`` is false when no flap has been recorded — this gate
|
||||
only speaks to post-flap proof and never invents a requirement.
|
||||
"""
|
||||
ns = (namespace or "").strip()
|
||||
store = state or {}
|
||||
flap_at = store.get("flap_at")
|
||||
reasons: list[str] = []
|
||||
|
||||
if not flap_at:
|
||||
return {
|
||||
"namespace": ns,
|
||||
"reproof_required": False,
|
||||
"proven": True,
|
||||
"flap_at": None,
|
||||
"proof_at": None,
|
||||
"missing_steps": [],
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
entry = (store.get("namespaces") or {}).get(ns)
|
||||
if not entry:
|
||||
reasons.append(
|
||||
f"MCP namespace '{ns}' has not been re-proven since the transport "
|
||||
f"flap at {flap_at}; run whoami, runtime context, and capability "
|
||||
f"resolve for '{ns}' itself (proof of another namespace does not "
|
||||
f"transfer)"
|
||||
)
|
||||
return {
|
||||
"namespace": ns,
|
||||
"reproof_required": True,
|
||||
"proven": False,
|
||||
"flap_at": flap_at,
|
||||
"proof_at": None,
|
||||
"missing_steps": list(REQUIRED_NAMESPACE_PROOF_STEPS),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
proof_at = entry.get("at")
|
||||
if proof_at is not None and str(proof_at) < str(flap_at):
|
||||
reasons.append(
|
||||
f"MCP namespace '{ns}' proof at {proof_at} predates the transport "
|
||||
f"flap at {flap_at}; re-prove the namespace before mutating"
|
||||
)
|
||||
return {
|
||||
"namespace": ns,
|
||||
"reproof_required": True,
|
||||
"proven": False,
|
||||
"flap_at": flap_at,
|
||||
"proof_at": proof_at,
|
||||
"missing_steps": list(REQUIRED_NAMESPACE_PROOF_STEPS),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
missing = [step for step in REQUIRED_NAMESPACE_PROOF_STEPS if not entry.get(step)]
|
||||
if missing:
|
||||
reasons.append(
|
||||
f"MCP namespace '{ns}' post-flap proof is incomplete; missing: "
|
||||
f"{', '.join(missing)}"
|
||||
)
|
||||
if entry.get("stale_runtime_reported"):
|
||||
missing = missing or ["stale_runtime_clear"]
|
||||
reasons.append(
|
||||
f"MCP namespace '{ns}' reported a reconnect/restart/stale-runtime "
|
||||
f"gate during re-proving; mutation stays blocked until the "
|
||||
f"namespace reconnects cleanly"
|
||||
)
|
||||
|
||||
return {
|
||||
"namespace": ns,
|
||||
"reproof_required": True,
|
||||
"proven": not missing,
|
||||
"flap_at": flap_at,
|
||||
"proof_at": proof_at,
|
||||
"missing_steps": list(missing),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def unproven_namespaces(
|
||||
state: dict | None, namespaces=WORKFLOW_NAMESPACES
|
||||
) -> list[str]:
|
||||
"""Return the namespaces still requiring post-flap re-proving."""
|
||||
out = []
|
||||
for ns in namespaces:
|
||||
assessment = assess_namespace_reproof(state, ns)
|
||||
if assessment["reproof_required"] and not assessment["proven"]:
|
||||
out.append(ns)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Promotion records (#615 acceptance criterion 4 / 10)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
PROMOTION_REQUIRED_FIELDS = (
|
||||
"previous_runtime_sha",
|
||||
"promoted_runtime_sha",
|
||||
"source_branch",
|
||||
"source_pr",
|
||||
"restart_method",
|
||||
"health_check_proof",
|
||||
"identity_proof",
|
||||
"profile_proof",
|
||||
"workspace_proof",
|
||||
"mutation_capability_proof",
|
||||
"rollback_instructions",
|
||||
)
|
||||
|
||||
|
||||
def assess_promotion_record(record: dict | None) -> dict:
|
||||
"""Validate an operator promotion record against the ADR checklist.
|
||||
|
||||
A promotion that does not record both the previous and the promoted SHA is
|
||||
not a promotion — it is an undocumented restart.
|
||||
"""
|
||||
data = record or {}
|
||||
missing = [
|
||||
field
|
||||
for field in PROMOTION_REQUIRED_FIELDS
|
||||
if not str(data.get(field) or "").strip()
|
||||
]
|
||||
reasons = []
|
||||
if missing:
|
||||
reasons.append(
|
||||
"promotion record is incomplete; missing: " + ", ".join(missing)
|
||||
)
|
||||
previous = str(data.get("previous_runtime_sha") or "").strip()
|
||||
promoted = str(data.get("promoted_runtime_sha") or "").strip()
|
||||
if previous and promoted and previous == promoted:
|
||||
reasons.append(
|
||||
"promotion record lists the same previous and promoted SHA "
|
||||
f"({previous[:12]}); nothing was promoted"
|
||||
)
|
||||
return {
|
||||
"valid": not reasons,
|
||||
"missing_fields": missing,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Filesystem observation (the only impure helper)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _git_capture(root: str, *args: str) -> str | None:
|
||||
if not root:
|
||||
return None
|
||||
try:
|
||||
res = subprocess.run(
|
||||
["git", "-C", root, *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 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*.
|
||||
|
||||
Returns ``checkout_branch``, ``runtime_head``, ``is_git_checkout``, and
|
||||
``dirty_files``. Every read failure degrades to ``None``/empty rather than
|
||||
raising, so a runtime that cannot be inspected classifies as ``unknown``
|
||||
instead of crashing the caller.
|
||||
"""
|
||||
empty = {
|
||||
"checkout_branch": None,
|
||||
"runtime_head": None,
|
||||
"is_git_checkout": False,
|
||||
"dirty_files": [],
|
||||
}
|
||||
if not process_root:
|
||||
return empty
|
||||
if not _git_capture(process_root, "rev-parse", "--show-toplevel"):
|
||||
return empty
|
||||
branch = _git_capture(process_root, "rev-parse", "--abbrev-ref", "HEAD")
|
||||
if branch == "HEAD": # detached HEAD has no branch name
|
||||
branch = None
|
||||
dirty = observe_dirty_files(process_root)
|
||||
return {
|
||||
"checkout_branch": branch,
|
||||
"runtime_head": _git_capture(process_root, "rev-parse", "HEAD"),
|
||||
"is_git_checkout": True,
|
||||
"dirty_files": dirty,
|
||||
}
|
||||
@@ -0,0 +1,603 @@
|
||||
"""Tests for the stable-control runtime mode gates (#615).
|
||||
|
||||
Covers acceptance criteria 6-11: runtime mode + SHA reporting, the fail-closed
|
||||
mutation gates (dev-test targeting production, unknown runtime, dirty stable
|
||||
checkout, dev-worktree launch, unsafe alignment), per-namespace post-flap
|
||||
re-proving, promotion-record completeness, and the policy statements that keep
|
||||
normal sessions from restarting the stable MCP runtime.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
import stable_control_runtime as scr # noqa: E402
|
||||
|
||||
|
||||
SHA_A = "a" * 40
|
||||
SHA_B = "b" * 40
|
||||
|
||||
STABLE_ROOT = "/Users/dev/Development/Gitea-Tools"
|
||||
DEV_WORKTREE_ROOT = "/Users/dev/Development/Gitea-Tools/branches/issue-615-work"
|
||||
|
||||
|
||||
def stable_report(**overrides):
|
||||
"""A healthy stable-control runtime report, overridable per test."""
|
||||
base = dict(
|
||||
process_root=STABLE_ROOT,
|
||||
checkout_branch="master",
|
||||
runtime_head=SHA_A,
|
||||
active_task_workspace=STABLE_ROOT,
|
||||
canonical_repository_root=STABLE_ROOT,
|
||||
repository_slug="Scaled-Tech-Consulting/Gitea-Tools",
|
||||
profile="prgs-author",
|
||||
authenticated_identity="jcwalker3",
|
||||
dirty_files=[],
|
||||
workspace_roots_aligned=True,
|
||||
)
|
||||
base.update(overrides)
|
||||
return scr.build_runtime_report(**base)
|
||||
|
||||
|
||||
class TestClassifyRuntimeMode(unittest.TestCase):
|
||||
def test_stable_branch_checkout_is_stable_control(self):
|
||||
res = scr.classify_runtime_mode(
|
||||
process_root=STABLE_ROOT, checkout_branch="master")
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_STABLE)
|
||||
self.assertFalse(res["dev_worktree_launched"])
|
||||
|
||||
def test_main_and_dev_are_also_stable(self):
|
||||
for branch in ("main", "dev"):
|
||||
res = scr.classify_runtime_mode(
|
||||
process_root=STABLE_ROOT, checkout_branch=branch)
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_STABLE, branch)
|
||||
|
||||
def test_branches_worktree_launch_is_dev_test(self):
|
||||
res = scr.classify_runtime_mode(
|
||||
process_root=DEV_WORKTREE_ROOT,
|
||||
checkout_branch="feat/issue-615-runtime-mode-enforcement",
|
||||
)
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_DEV_TEST)
|
||||
self.assertTrue(res["dev_worktree_launched"])
|
||||
|
||||
def test_feature_branch_outside_branches_is_still_dev_test(self):
|
||||
res = scr.classify_runtime_mode(
|
||||
process_root="/Users/dev/Development/scratch-clone",
|
||||
checkout_branch="feat/experiment",
|
||||
)
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_DEV_TEST)
|
||||
self.assertFalse(res["dev_worktree_launched"])
|
||||
|
||||
def test_unresolvable_root_is_unknown(self):
|
||||
res = scr.classify_runtime_mode(process_root=None, checkout_branch=None)
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_UNKNOWN)
|
||||
|
||||
def test_non_git_root_is_unknown(self):
|
||||
res = scr.classify_runtime_mode(
|
||||
process_root="/opt/gitea-tools-release",
|
||||
checkout_branch=None,
|
||||
is_git_checkout=False,
|
||||
)
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_UNKNOWN)
|
||||
|
||||
def test_detached_head_is_unknown(self):
|
||||
res = scr.classify_runtime_mode(
|
||||
process_root=STABLE_ROOT, checkout_branch=None)
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_UNKNOWN)
|
||||
|
||||
def test_operator_declaration_wins_over_inference(self):
|
||||
res = scr.classify_runtime_mode(
|
||||
process_root="/opt/gitea-tools-release",
|
||||
checkout_branch=None,
|
||||
is_git_checkout=False,
|
||||
declared_mode=scr.RUNTIME_MODE_STABLE,
|
||||
)
|
||||
self.assertEqual(res["runtime_mode"], scr.RUNTIME_MODE_STABLE)
|
||||
self.assertTrue(res["declared"])
|
||||
|
||||
def test_invalid_declaration_is_ignored(self):
|
||||
with patch.dict(os.environ, {scr.ENV_RUNTIME_MODE: "production-ish"}):
|
||||
self.assertIsNone(scr.declared_runtime_mode())
|
||||
|
||||
def test_valid_declaration_is_read_from_env(self):
|
||||
with patch.dict(os.environ, {scr.ENV_RUNTIME_MODE: "dev-test"}):
|
||||
self.assertEqual(scr.declared_runtime_mode(), scr.RUNTIME_MODE_DEV_TEST)
|
||||
|
||||
|
||||
class TestRuntimeReport(unittest.TestCase):
|
||||
"""Acceptance criterion 6: runtime mode and SHA reporting."""
|
||||
|
||||
def test_report_carries_every_required_field(self):
|
||||
report = stable_report()
|
||||
for field in (
|
||||
"runtime_mode",
|
||||
"runtime_git_sha",
|
||||
"runtime_branch",
|
||||
"runtime_checkout_path",
|
||||
"mcp_process_root",
|
||||
"active_task_workspace",
|
||||
"repository_slug",
|
||||
"profile",
|
||||
"authenticated_identity",
|
||||
"dirty_files",
|
||||
"workspace_roots_aligned",
|
||||
"real_mutations_allowed",
|
||||
):
|
||||
self.assertIn(field, report, field)
|
||||
|
||||
def test_report_records_the_runtime_sha(self):
|
||||
self.assertEqual(stable_report()["runtime_git_sha"], SHA_A)
|
||||
|
||||
def test_format_summarises_mode_sha_and_branch(self):
|
||||
summary = scr.format_runtime_mode(stable_report())
|
||||
self.assertIn(scr.RUNTIME_MODE_STABLE, summary)
|
||||
self.assertIn(SHA_A[:12], summary)
|
||||
self.assertIn("master", summary)
|
||||
|
||||
|
||||
class TestMutationGate(unittest.TestCase):
|
||||
"""Acceptance criterion 7: fail-closed mutation gates."""
|
||||
|
||||
def test_stable_healthy_runtime_allows_real_mutations(self):
|
||||
report = stable_report()
|
||||
gate = scr.assess_runtime_mutation_gate(report)
|
||||
self.assertFalse(gate["block"])
|
||||
self.assertEqual(gate["reasons"], [])
|
||||
self.assertTrue(report["real_mutations_allowed"])
|
||||
|
||||
def test_dev_test_runtime_blocks_real_production_mutations(self):
|
||||
report = stable_report(
|
||||
process_root=DEV_WORKTREE_ROOT,
|
||||
checkout_branch="feat/issue-615-runtime-mode-enforcement",
|
||||
active_task_workspace=DEV_WORKTREE_ROOT,
|
||||
)
|
||||
gate = scr.assess_runtime_mutation_gate(report)
|
||||
self.assertTrue(gate["block"])
|
||||
self.assertIn(scr.BLOCKER_DEV_TEST_PRODUCTION, gate["blocker_kinds"])
|
||||
self.assertFalse(report["real_mutations_allowed"])
|
||||
|
||||
def test_dev_test_runtime_may_mutate_a_non_production_target(self):
|
||||
report = stable_report(
|
||||
process_root=DEV_WORKTREE_ROOT,
|
||||
checkout_branch="feat/issue-615-runtime-mode-enforcement",
|
||||
)
|
||||
gate = scr.assess_runtime_mutation_gate(
|
||||
report, target_is_production=False)
|
||||
self.assertFalse(gate["block"])
|
||||
|
||||
def test_unknown_runtime_blocks_mutations(self):
|
||||
report = stable_report(checkout_branch=None)
|
||||
gate = scr.assess_runtime_mutation_gate(report)
|
||||
self.assertTrue(gate["block"])
|
||||
self.assertIn(scr.BLOCKER_UNKNOWN_RUNTIME, gate["blocker_kinds"])
|
||||
|
||||
def test_unknown_runtime_blocks_even_a_non_production_target(self):
|
||||
report = stable_report(checkout_branch=None)
|
||||
gate = scr.assess_runtime_mutation_gate(
|
||||
report, target_is_production=False)
|
||||
self.assertTrue(gate["block"])
|
||||
|
||||
def test_dirty_stable_runtime_blocks_mutations(self):
|
||||
report = stable_report(dirty_files=["gitea_mcp_server.py"])
|
||||
gate = scr.assess_runtime_mutation_gate(report)
|
||||
self.assertTrue(gate["block"])
|
||||
self.assertIn(scr.BLOCKER_DIRTY_STABLE_RUNTIME, gate["blocker_kinds"])
|
||||
self.assertTrue(
|
||||
any("dirty" in reason for reason in gate["reasons"]))
|
||||
|
||||
def test_dev_worktree_launch_is_reported_as_its_own_blocker(self):
|
||||
report = stable_report(
|
||||
process_root=DEV_WORKTREE_ROOT,
|
||||
checkout_branch="feat/issue-615-runtime-mode-enforcement",
|
||||
)
|
||||
gate = scr.assess_runtime_mutation_gate(report)
|
||||
self.assertIn(scr.BLOCKER_DEV_WORKTREE_LAUNCH, gate["blocker_kinds"])
|
||||
|
||||
def test_unsafe_workspace_alignment_blocks_mutations(self):
|
||||
report = stable_report(workspace_roots_aligned=False)
|
||||
gate = scr.assess_runtime_mutation_gate(report)
|
||||
self.assertTrue(gate["block"])
|
||||
self.assertIn(scr.BLOCKER_UNSAFE_ALIGNMENT, gate["blocker_kinds"])
|
||||
|
||||
def test_unknown_alignment_does_not_block(self):
|
||||
report = stable_report(workspace_roots_aligned=None)
|
||||
self.assertFalse(scr.assess_runtime_mutation_gate(report)["block"])
|
||||
|
||||
def test_env_escape_hatch_disables_the_gate(self):
|
||||
report = stable_report(checkout_branch=None)
|
||||
with patch.dict(os.environ, {scr.ENV_DISABLE: "1"}):
|
||||
gate = scr.assess_runtime_mutation_gate(report)
|
||||
self.assertFalse(gate["block"])
|
||||
self.assertTrue(gate["gate_disabled"])
|
||||
|
||||
def test_block_reasons_helper_matches_the_gate(self):
|
||||
report = stable_report(checkout_branch=None)
|
||||
self.assertEqual(
|
||||
scr.runtime_block_reasons(report),
|
||||
scr.assess_runtime_mutation_gate(report)["reasons"],
|
||||
)
|
||||
|
||||
def test_block_payload_names_the_operator_recovery_path(self):
|
||||
report = stable_report(checkout_branch=None)
|
||||
payload = scr.runtime_report_payload(report)
|
||||
self.assertEqual(payload["kind"], "runtime_mode_block")
|
||||
self.assertEqual(payload["blocker_kind"], scr.BLOCKER_UNKNOWN_RUNTIME)
|
||||
self.assertTrue(
|
||||
any("promotion-runbook" in line for line in payload["recovery"]))
|
||||
|
||||
|
||||
class TestPostFlapReproving(unittest.TestCase):
|
||||
"""Acceptance criterion 8: per-namespace post-flap re-proving."""
|
||||
|
||||
def test_no_flap_means_no_reproof_required(self):
|
||||
state = scr.new_reproof_state()
|
||||
res = scr.assess_namespace_reproof(state, "reviewer")
|
||||
self.assertFalse(res["reproof_required"])
|
||||
self.assertTrue(res["proven"])
|
||||
|
||||
def test_transport_recovery_requires_namespace_specific_reproving(self):
|
||||
state = scr.record_transport_flap(
|
||||
scr.new_reproof_state(), at="2026-07-20T14:00:00Z")
|
||||
res = scr.assess_namespace_reproof(state, "reviewer")
|
||||
self.assertTrue(res["reproof_required"])
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertEqual(
|
||||
res["missing_steps"], list(scr.REQUIRED_NAMESPACE_PROOF_STEPS))
|
||||
|
||||
def test_author_proof_does_not_imply_other_namespaces(self):
|
||||
state = scr.record_transport_flap(
|
||||
scr.new_reproof_state(), at="2026-07-20T14:00:00Z")
|
||||
state = scr.record_namespace_proof(
|
||||
state,
|
||||
"author",
|
||||
at="2026-07-20T14:05:00Z",
|
||||
whoami=True,
|
||||
runtime_context=True,
|
||||
capability_resolved=True,
|
||||
)
|
||||
self.assertTrue(scr.assess_namespace_reproof(state, "author")["proven"])
|
||||
for other in ("reviewer", "merger", "reconciler"):
|
||||
assessment = scr.assess_namespace_reproof(state, other)
|
||||
self.assertFalse(assessment["proven"], other)
|
||||
self.assertTrue(
|
||||
any("does not transfer" in reason
|
||||
for reason in assessment["reasons"]),
|
||||
other,
|
||||
)
|
||||
self.assertEqual(
|
||||
scr.unproven_namespaces(state),
|
||||
["reviewer", "merger", "reconciler"],
|
||||
)
|
||||
|
||||
def test_proof_recorded_before_the_flap_does_not_count(self):
|
||||
state = scr.record_namespace_proof(
|
||||
scr.new_reproof_state(),
|
||||
"merger",
|
||||
at="2026-07-20T13:00:00Z",
|
||||
whoami=True,
|
||||
runtime_context=True,
|
||||
capability_resolved=True,
|
||||
)
|
||||
state = scr.record_transport_flap(state, at="2026-07-20T14:00:00Z")
|
||||
res = scr.assess_namespace_reproof(state, "merger")
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(any("predates" in reason for reason in res["reasons"]))
|
||||
|
||||
def test_incomplete_proof_lists_the_missing_steps(self):
|
||||
state = scr.record_transport_flap(
|
||||
scr.new_reproof_state(), at="2026-07-20T14:00:00Z")
|
||||
state = scr.record_namespace_proof(
|
||||
state, "reviewer", at="2026-07-20T14:05:00Z", whoami=True)
|
||||
res = scr.assess_namespace_reproof(state, "reviewer")
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertEqual(
|
||||
res["missing_steps"], ["runtime_context", "capability_resolved"])
|
||||
|
||||
def test_stale_runtime_report_keeps_the_namespace_unproven(self):
|
||||
state = scr.record_transport_flap(
|
||||
scr.new_reproof_state(), at="2026-07-20T14:00:00Z")
|
||||
state = scr.record_namespace_proof(
|
||||
state,
|
||||
"reviewer",
|
||||
at="2026-07-20T14:05:00Z",
|
||||
whoami=True,
|
||||
runtime_context=True,
|
||||
capability_resolved=True,
|
||||
stale_runtime_reported=True,
|
||||
)
|
||||
res = scr.assess_namespace_reproof(state, "reviewer")
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(
|
||||
any("stale-runtime" in reason for reason in res["reasons"]))
|
||||
|
||||
def test_unproven_namespace_blocks_the_mutation_gate(self):
|
||||
state = scr.record_transport_flap(
|
||||
scr.new_reproof_state(), at="2026-07-20T14:00:00Z")
|
||||
gate = scr.assess_runtime_mutation_gate(
|
||||
stable_report(), namespace="reviewer", namespace_reproof=state)
|
||||
self.assertTrue(gate["block"])
|
||||
self.assertIn(scr.BLOCKER_NAMESPACE_NOT_REPROVEN, gate["blocker_kinds"])
|
||||
|
||||
def test_reproven_namespace_clears_the_mutation_gate(self):
|
||||
state = scr.record_transport_flap(
|
||||
scr.new_reproof_state(), at="2026-07-20T14:00:00Z")
|
||||
state = scr.record_namespace_proof(
|
||||
state,
|
||||
"reviewer",
|
||||
at="2026-07-20T14:05:00Z",
|
||||
whoami=True,
|
||||
runtime_context=True,
|
||||
capability_resolved=True,
|
||||
)
|
||||
gate = scr.assess_runtime_mutation_gate(
|
||||
stable_report(), namespace="reviewer", namespace_reproof=state)
|
||||
self.assertFalse(gate["block"])
|
||||
|
||||
|
||||
class TestPromotionRecord(unittest.TestCase):
|
||||
"""Acceptance criteria 4 / 10: promotion records previous and promoted SHAs."""
|
||||
|
||||
def complete_record(self, **overrides):
|
||||
record = {
|
||||
"previous_runtime_sha": SHA_A,
|
||||
"promoted_runtime_sha": SHA_B,
|
||||
"source_branch": "feat/issue-615-runtime-mode-enforcement",
|
||||
"source_pr": "770",
|
||||
"restart_method": "operator reload of the stable control runtime",
|
||||
"health_check_proof": "gitea_assess_mcp_namespace_health: healthy",
|
||||
"identity_proof": "gitea_whoami: sysadmin / prgs-reviewer",
|
||||
"profile_proof": "runtime context: prgs-reviewer",
|
||||
"workspace_proof": "process root == canonical root, clean",
|
||||
"mutation_capability_proof": "resolve review_pr: allowed",
|
||||
"rollback_instructions": "re-promote " + SHA_A,
|
||||
}
|
||||
record.update(overrides)
|
||||
return record
|
||||
|
||||
def test_complete_record_is_valid(self):
|
||||
res = scr.assess_promotion_record(self.complete_record())
|
||||
self.assertTrue(res["valid"])
|
||||
self.assertEqual(res["missing_fields"], [])
|
||||
|
||||
def test_promotion_records_previous_and_promoted_shas(self):
|
||||
res = scr.assess_promotion_record(
|
||||
self.complete_record(previous_runtime_sha="", promoted_runtime_sha=""))
|
||||
self.assertFalse(res["valid"])
|
||||
self.assertIn("previous_runtime_sha", res["missing_fields"])
|
||||
self.assertIn("promoted_runtime_sha", res["missing_fields"])
|
||||
|
||||
def test_identical_shas_are_not_a_promotion(self):
|
||||
res = scr.assess_promotion_record(
|
||||
self.complete_record(promoted_runtime_sha=SHA_A))
|
||||
self.assertFalse(res["valid"])
|
||||
self.assertTrue(
|
||||
any("nothing was promoted" in reason for reason in res["reasons"]))
|
||||
|
||||
def test_missing_rollback_instructions_fail_closed(self):
|
||||
res = scr.assess_promotion_record(
|
||||
self.complete_record(rollback_instructions=""))
|
||||
self.assertFalse(res["valid"])
|
||||
self.assertIn("rollback_instructions", res["missing_fields"])
|
||||
|
||||
def test_empty_record_is_invalid(self):
|
||||
self.assertFalse(scr.assess_promotion_record(None)["valid"])
|
||||
|
||||
|
||||
class TestNormalSessionsCannotRestartStableRuntime(unittest.TestCase):
|
||||
"""Acceptance criterion 3: normal sessions do not restart the stable MCP."""
|
||||
|
||||
def test_adr_forbids_kill_restart_and_relaunch(self):
|
||||
adr = (
|
||||
REPO_ROOT
|
||||
/ "docs"
|
||||
/ "architecture"
|
||||
/ "mcp-stable-control-runtime-policy-adr.md"
|
||||
).read_text()
|
||||
for phrase in ("Kill the running MCP server process",
|
||||
"Restart / relaunch the MCP server process",
|
||||
"Relaunch MCP from a development worktree"):
|
||||
self.assertIn(phrase, adr, phrase)
|
||||
|
||||
def test_promotion_runbook_exists_and_lists_every_record_field(self):
|
||||
runbook = (
|
||||
REPO_ROOT / "docs" / "stable-runtime-promotion-runbook.md"
|
||||
).read_text()
|
||||
for field in scr.PROMOTION_REQUIRED_FIELDS:
|
||||
self.assertIn(field, runbook, field)
|
||||
|
||||
def test_no_mcp_tool_offers_a_runtime_restart(self):
|
||||
server = (REPO_ROOT / "gitea_mcp_server.py").read_text()
|
||||
for forbidden in ("def gitea_restart_", "def gitea_kill_"):
|
||||
self.assertNotIn(forbidden, server, forbidden)
|
||||
|
||||
|
||||
class TestServerWiring(unittest.TestCase):
|
||||
"""The gate is wired into the server's mutation permission path."""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server as srv # imported lazily: heavy module
|
||||
|
||||
self.srv = srv
|
||||
|
||||
def test_reads_are_never_blocked_by_runtime_mode(self):
|
||||
with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}):
|
||||
self.assertEqual(self.srv._runtime_mode_block("gitea.read"), [])
|
||||
|
||||
def test_gate_is_skipped_under_pure_unit_test_isolation(self):
|
||||
# The suite itself runs from a branches/ worktree (dev-test by design);
|
||||
# without forced production guards the gate must not fire.
|
||||
self.assertEqual(self.srv._runtime_mode_block("gitea.pr.create"), [])
|
||||
|
||||
def test_dev_worktree_runtime_blocks_mutations_when_guards_forced(self):
|
||||
report = stable_report(
|
||||
process_root=DEV_WORKTREE_ROOT,
|
||||
checkout_branch="feat/issue-615-runtime-mode-enforcement",
|
||||
)
|
||||
with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}), \
|
||||
patch.object(
|
||||
self.srv, "_current_runtime_mode_report", return_value=report):
|
||||
reasons = self.srv._runtime_mode_block("gitea.pr.create")
|
||||
self.assertTrue(reasons)
|
||||
self.assertTrue(any("dev-test" in reason for reason in reasons))
|
||||
|
||||
def test_stable_runtime_allows_mutations_when_guards_forced(self):
|
||||
with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}), \
|
||||
patch.object(
|
||||
self.srv,
|
||||
"_current_runtime_mode_report",
|
||||
return_value=stable_report()):
|
||||
self.assertEqual(self.srv._runtime_mode_block("gitea.pr.create"), [])
|
||||
|
||||
def test_unassessable_runtime_fails_closed(self):
|
||||
with patch.dict(os.environ, {"GITEA_TEST_FORCE_PRODUCTION_GUARDS": "1"}), \
|
||||
patch.object(
|
||||
self.srv,
|
||||
"_current_runtime_mode_report",
|
||||
side_effect=RuntimeError("boom")):
|
||||
reasons = self.srv._runtime_mode_block("gitea.pr.create")
|
||||
self.assertTrue(reasons)
|
||||
self.assertTrue(any("fail closed" in reason for reason in reasons))
|
||||
|
||||
def test_live_report_describes_this_checkout(self):
|
||||
report = self.srv._current_runtime_mode_report()
|
||||
self.assertIn(report["runtime_mode"], scr.VALID_RUNTIME_MODES)
|
||||
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()
|
||||
@@ -304,6 +304,193 @@ Who/what acts next:
|
||||
)
|
||||
)
|
||||
|
||||
# ── Stable control runtime states (#615) ─────────────────────────────────────
|
||||
|
||||
EXAMPLES.append(
|
||||
_example(
|
||||
"runtime_healthy",
|
||||
"""
|
||||
[CONTROLLER HANDOFF] Runtime check — stable control runtime healthy
|
||||
|
||||
Server-side mutation ledger:
|
||||
- none — no server-side state changed
|
||||
|
||||
Blockers:
|
||||
- none
|
||||
""",
|
||||
f"""
|
||||
[THREAD STATE LEDGER] Runtime — stable control runtime healthy
|
||||
|
||||
What is true now:
|
||||
- Runtime mode: stable-control
|
||||
- Runtime git SHA: {HEAD_SHA}
|
||||
- Server-side decision state: no server-side state changed
|
||||
- Local verdict/state: runtime reported real_mutations_allowed=true
|
||||
- Latest known validation: gitea_get_runtime_context read in this session
|
||||
|
||||
What changed:
|
||||
- nothing; this is a read-only runtime observation
|
||||
|
||||
What is blocked:
|
||||
- Blocker classification: no blocker
|
||||
|
||||
Who/what acts next:
|
||||
- Next actor: author
|
||||
- Required action: proceed with the allocated workflow phase
|
||||
- Do not do: restart or relaunch the stable runtime
|
||||
- Resume from: gitea_workflow_dashboard
|
||||
""",
|
||||
)
|
||||
)
|
||||
|
||||
EXAMPLES.append(
|
||||
_example(
|
||||
"transport_flap_recovered",
|
||||
"""
|
||||
[CONTROLLER HANDOFF] Runtime check — transport flap recovered
|
||||
|
||||
Server-side mutation ledger:
|
||||
- none — no server-side state changed
|
||||
|
||||
Blockers:
|
||||
- environment/tooling blocker: MCP transport dropped mid-session and recovered
|
||||
""",
|
||||
f"""
|
||||
[THREAD STATE LEDGER] Runtime — transport flap recovered, namespaces re-proven
|
||||
|
||||
What is true now:
|
||||
- Runtime mode: stable-control
|
||||
- Runtime git SHA: {HEAD_SHA}
|
||||
- Server-side decision state: no server-side state changed
|
||||
- Local verdict/state: all four namespaces re-proven after the flap
|
||||
- Latest known validation: whoami + runtime context + capability resolve per namespace
|
||||
|
||||
What changed:
|
||||
- author, reviewer, merger, and reconciler namespaces each re-proven independently
|
||||
|
||||
What is blocked:
|
||||
- Blocker classification: no blocker
|
||||
|
||||
Who/what acts next:
|
||||
- Next actor: author
|
||||
- Required action: resume the interrupted workflow phase from its last durable state
|
||||
- Do not do: treat author proof as proof of the other namespaces
|
||||
- Resume from: the phase handoff that preceded the flap
|
||||
""",
|
||||
)
|
||||
)
|
||||
|
||||
EXAMPLES.append(
|
||||
_example(
|
||||
"namespace_not_yet_reproven",
|
||||
"""
|
||||
[CONTROLLER HANDOFF] Runtime check — reviewer namespace not re-proven
|
||||
|
||||
Server-side mutation ledger:
|
||||
- none — no server-side state changed
|
||||
|
||||
Blockers:
|
||||
- environment/tooling blocker: reviewer namespace not re-proven since the transport flap
|
||||
""",
|
||||
f"""
|
||||
[THREAD STATE LEDGER] Runtime — reviewer namespace not re-proven after flap
|
||||
|
||||
What is true now:
|
||||
- Runtime mode: stable-control
|
||||
- Runtime git SHA: {HEAD_SHA}
|
||||
- Server-side decision state: no server-side state changed
|
||||
- Local verdict/state: reviewer namespace unproven; mutation gate fails closed
|
||||
- Latest known validation: author namespace re-proven; reviewer not attempted
|
||||
|
||||
What changed:
|
||||
- reviewer mutations blocked with namespace_not_reproven_after_flap
|
||||
|
||||
What is blocked:
|
||||
- Blocker classification: environment/tooling blocker
|
||||
|
||||
Who/what acts next:
|
||||
- Next actor: reviewer
|
||||
- Required action: run whoami, runtime context, and capability resolve in the reviewer namespace
|
||||
- Do not do: substitute author proof for reviewer proof
|
||||
- Resume from: docs/stable-runtime-promotion-runbook.md section 5
|
||||
""",
|
||||
)
|
||||
)
|
||||
|
||||
EXAMPLES.append(
|
||||
_example(
|
||||
"promotion_completed",
|
||||
"""
|
||||
[CONTROLLER HANDOFF] Runtime promotion — completed
|
||||
|
||||
Server-side mutation ledger:
|
||||
- gitea_create_issue_comment on #615 with the promotion record
|
||||
|
||||
Blockers:
|
||||
- none
|
||||
""",
|
||||
f"""
|
||||
[THREAD STATE LEDGER] Runtime — promotion completed and re-proven
|
||||
|
||||
What is true now:
|
||||
- Runtime mode: stable-control
|
||||
- Runtime git SHA: {HEAD_SHA}
|
||||
- Server-side decision state: server-side state changed
|
||||
- Local verdict/state: promotion record carries every required field
|
||||
- Latest known validation: assess_promotion_record valid=true; all namespaces re-proven
|
||||
|
||||
What changed:
|
||||
- stable control runtime advanced to the promoted SHA and reloaded by the operator
|
||||
|
||||
What is blocked:
|
||||
- Blocker classification: no blocker
|
||||
|
||||
Who/what acts next:
|
||||
- Next actor: author
|
||||
- Required action: resume normal workflow phases on the promoted runtime
|
||||
- Do not do: promote again without a fresh record
|
||||
- Resume from: docs/stable-runtime-promotion-runbook.md section 4
|
||||
""",
|
||||
)
|
||||
)
|
||||
|
||||
EXAMPLES.append(
|
||||
_example(
|
||||
"rollback_required",
|
||||
"""
|
||||
[CONTROLLER HANDOFF] Runtime promotion — rollback required
|
||||
|
||||
Server-side mutation ledger:
|
||||
- gitea_create_issue_comment on #615 with the rollback evidence
|
||||
|
||||
Blockers:
|
||||
- environment/tooling blocker: promoted runtime unhealthy, rollback required
|
||||
""",
|
||||
f"""
|
||||
[THREAD STATE LEDGER] Runtime — promoted runtime unhealthy, rollback required
|
||||
|
||||
What is true now:
|
||||
- Runtime mode: unknown
|
||||
- Runtime git SHA: {HEAD_SHA}
|
||||
- Server-side decision state: no server-side state changed after the promotion record
|
||||
- Local verdict/state: promoted runtime failed namespace health; mutations blocked
|
||||
- Latest known validation: namespace health probe reported EOF after reload
|
||||
|
||||
What changed:
|
||||
- all PR/review/merge work stopped pending rollback to the previous runtime SHA
|
||||
|
||||
What is blocked:
|
||||
- Blocker classification: environment/tooling blocker
|
||||
|
||||
Who/what acts next:
|
||||
- Next actor: controller
|
||||
- Required action: operator rolls back to the previous runtime SHA and re-proves every namespace
|
||||
- Do not do: route around the unhealthy runtime or mutate from a dev/test runtime
|
||||
- Resume from: docs/stable-runtime-promotion-runbook.md section 6
|
||||
""",
|
||||
)
|
||||
)
|
||||
|
||||
EXAMPLES.append(
|
||||
_example(
|
||||
"duplicate_canonicalization_blocker",
|
||||
|
||||
Reference in New Issue
Block a user