Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 78cc37a977 feat(observability): optional self-hosted Sentry instrumentation for MCP workflow failures (Closes #606)
Add an env-var-gated, off-by-default Sentry SDK integration so MCP runtime
errors, fail-closed workflow blockers, lease/terminal-lock/stale-runtime
collisions, and recurring watchdog check-ins are visible in a self-hosted
Sentry at https://sentry.prgs.cc/. Gitea stays the source of truth; Sentry is
observe-only.

New module `sentry_observability.py` mirrors the `gitea_audit` conventions
(env-gated, best-effort, redacting):

- Config from MCP_SENTRY_ENABLED / SENTRY_DSN / SENTRY_ENVIRONMENT /
  SENTRY_RELEASE / MCP_SENTRY_TRACES_SAMPLE_RATE / MCP_SENTRY_ENABLE_LOGS.
  Active only when enabled AND a DSN is present; otherwise a hard no-op.
- Fail OPEN for observability (never blocks a tool success path) and fail
  CLOSED for redaction (drop a field rather than risk leaking it).
- `scrub_event` before_send/before_send_log hook + allowlisted tags: no
  tokens/passwords/keychain ids/DSNs/cookies, no raw session-state or full
  prompt bodies (session_id -> 12-char hash), no full filesystem paths
  (worktree path -> coarse category). Reuses incident_bridge + gitea_audit
  scrubbers.
- capture_exception, capture_workflow_blocker (with canonical next action),
  and monitor_checkin with six stable cron slugs (stale lease scan, terminal
  lock scan, allocator health, namespace health, dashboard freshness,
  reconciler cleanup).
- `sentry_sdk` is a lazily-imported optional dependency; the module imports
  and no-ops cleanly when the package is absent.

Wiring in gitea_mcp_server.py (additive, guarded, best-effort):
- init_sentry() in __main__ before mcp.run.
- capture_exception in the `_audited` failure path; capture_workflow_blocker
  in `_audit_pr_result` BLOCKED/FAILED path.
- allocator + namespace-health watchdog check-ins at their MCP tool sites
  (domain modules left pure).

Also: pin `sentry-sdk==2.20.0` (optional), document the six env vars in
`.env.example`, and add `docs/observability/sentry-integration.md` covering
project creation in https://sentry.prgs.cc/, DSN handling, local/dev/prod
config, redaction guarantees, and coexistence with the #612 incident bridge.

Tests: tests/test_sentry_observability.py (36 cases) cover disabled / enabled /
missing-DSN / missing-SDK, redaction, exception capture, workflow-blocker
capture, and cron check-in behaviour. Full suite: 2632 passed, 6 skipped.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-12 02:35:47 -04:00
22 changed files with 1358 additions and 4165 deletions
+21
View File
@@ -39,6 +39,27 @@ GITEA_AUDIT_LOG=/path/to/gitea-mcp-audit.log
# only — never the token value. Surfaced by gitea_get_profile. # only — never the token value. Surfaced by gitea_get_profile.
GITEA_TOKEN_SOURCE=GITEA_TOKEN GITEA_TOKEN_SOURCE=GITEA_TOKEN
# ── Optional self-hosted Sentry observability (#606) ────────────────────────
# Emits runtime errors, fail-closed workflow blockers, lease/terminal-lock/
# stale-runtime collisions, and watchdog cron check-ins to a SELF-HOSTED Sentry
# (https://sentry.prgs.cc/) — never Sentry Cloud. Gitea stays the source of
# truth; Sentry is observe-only. OFF by default: with MCP_SENTRY_ENABLED unset
# or SENTRY_DSN empty, nothing is initialised and no events are sent.
#
# Master gate. Truthy = 1/true/yes/on. Both this AND SENTRY_DSN are required.
MCP_SENTRY_ENABLED=0
# DSN for the self-hosted project (create a `gitea-tools-mcp` project in
# https://sentry.prgs.cc/ and copy its DSN). Never commit a real DSN.
SENTRY_DSN=
# Deployment environment tag (local/dev/prod). Defaults to "development".
SENTRY_ENVIRONMENT=development
# Optional release identifier (e.g. a git SHA or version string).
SENTRY_RELEASE=
# Performance-trace sample rate, 0.01.0 (clamped). Default 0.0 (traces off).
MCP_SENTRY_TRACES_SAMPLE_RATE=0.0
# Set to 1 to forward Python logs to Sentry as structured logs. Default off.
MCP_SENTRY_ENABLE_LOGS=0
# Optional canonical runtime-profile config (#19). Instead of the fields above, # Optional canonical runtime-profile config (#19). Instead of the fields above,
# point every LLM launcher at ONE JSON file of named profiles and select one. # point every LLM launcher at ONE JSON file of named profiles and select one.
# Secrets are referenced (keychain id / env var name), never inlined. See # Secrets are referenced (keychain id / env var name), never inlined. See
+138
View File
@@ -0,0 +1,138 @@
# Self-hosted Sentry observability for the Gitea MCP server (#606)
Optional, **off-by-default** instrumentation that reports MCP runtime errors,
fail-closed workflow blockers, lease / terminal-lock / stale-runtime
collisions, and recurring watchdog check-ins to a **self-hosted** Sentry at
`https://sentry.prgs.cc/`.
> **Gitea remains the source of truth.** Sentry is observe-only. It never
> approves, merges, closes, or otherwise mutates Gitea workflow state, and it
> never bypasses leases, #332, workflow roles, or the MCP gates. Sentry alerts
> may only feed the *sanctioned* Gitea issue/comment path via the #612 incident
> bridge — never a direct write.
Implemented by [`sentry_observability.py`](../../sentry_observability.py).
---
## 1. Create the Sentry project
1. Sign in to the self-hosted Sentry at **`https://sentry.prgs.cc/`** (this is
**not** Sentry Cloud — do not use `*.ingest.sentry.io`).
2. Create a new **Python** project named **`gitea-tools-mcp`**.
3. Open **Settings → Projects → gitea-tools-mcp → Client Keys (DSN)** and copy
the DSN. It looks like `https://<publickey>@sentry.prgs.cc/<project-id>`.
4. **Never commit the DSN.** It is a runtime secret supplied via env var only.
## 2. Configure the environment
All configuration is env-var driven (see [`.env.example`](../../.env.example)):
| Variable | Purpose | Default |
|----------|---------|---------|
| `MCP_SENTRY_ENABLED` | Master gate (`1/true/yes/on`). Required. | off |
| `SENTRY_DSN` | Self-hosted DSN. Required. | *(empty)* |
| `SENTRY_ENVIRONMENT` | `local` / `dev` / `prod` tag. | `development` |
| `SENTRY_RELEASE` | Release id (git SHA or version). | *(none)* |
| `MCP_SENTRY_TRACES_SAMPLE_RATE` | Perf-trace sample rate `0.01.0` (clamped). | `0.0` |
| `MCP_SENTRY_ENABLE_LOGS` | Forward Python logs as structured logs. | off |
**The feature stays completely off unless `MCP_SENTRY_ENABLED` is truthy *and*
`SENTRY_DSN` is non-empty.** With either missing, `init_sentry()` is a no-op,
the SDK is never initialised, and no events are sent — existing tool behaviour
and API-call patterns are unchanged.
### Per-environment examples
```bash
# local (quiet: capture errors/blockers, no traces)
export MCP_SENTRY_ENABLED=1
export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
export SENTRY_ENVIRONMENT=local
# dev (light tracing + logs)
export MCP_SENTRY_ENABLED=1
export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
export SENTRY_ENVIRONMENT=dev
export MCP_SENTRY_TRACES_SAMPLE_RATE=0.2
export MCP_SENTRY_ENABLE_LOGS=1
# prod (errors/blockers + low-rate tracing, release-tagged)
export MCP_SENTRY_ENABLED=1
export SENTRY_DSN="https://<key>@sentry.prgs.cc/<id>"
export SENTRY_ENVIRONMENT=prod
export SENTRY_RELEASE="$(git rev-parse --short HEAD)"
export MCP_SENTRY_TRACES_SAMPLE_RATE=0.05
```
The optional SDK is pinned in [`requirements.txt`](../../requirements.txt)
(`sentry-sdk==2.20.0`). It is imported lazily: if the package is absent, the
module still imports and every entry point is a safe no-op.
## 3. What is instrumented
| Signal | Where | Notes |
|--------|-------|-------|
| Startup init | `gitea_mcp_server.py` `__main__`, before `mcp.run` | Prints a redaction-safe status line to stderr. |
| Failing mutations (exceptions) | `_audited(...)` context manager | `capture_exception` with scrubbed tags. |
| Fail-closed blockers / failed mutations | `_audit_pr_result(...)` (BLOCKED/FAILED) | Structured `capture_workflow_blocker` event incl. the canonical next action when available (criterion 7). |
| Allocator watchdog check-ins | `gitea_allocate_next_work` tool | `allocator_health`, `stale_lease_scan`, `terminal_lock_scan`. |
| Namespace-health check-in | `gitea_assess_mcp_namespace_health` tool | `namespace_health`. |
All capture paths are **best-effort / fail open**: a Sentry outage or capture
error never breaks an MCP tool success path.
## 4. Cron / watchdog monitors
`sentry_observability.MONITOR_SLUGS` defines stable check-in slugs:
| Registry key | Sentry monitor slug | Wired at |
|--------------|--------------------|----------|
| `stale_lease_scan` | `gitea-mcp-stale-lease-scan` | allocator run (global lease expiry) |
| `terminal_lock_scan` | `gitea-mcp-terminal-lock-scan` | allocator run (terminal-lock lookup) |
| `allocator_health` | `gitea-mcp-allocator-health` | allocator run |
| `namespace_health` | `gitea-mcp-namespace-health` | namespace-health probe |
| `dashboard_freshness` | `gitea-mcp-dashboard-freshness` | call `monitor_checkin("dashboard_freshness", ...)` from the dashboard refresh job (#605) |
| `reconciler_cleanup` | `gitea-mcp-reconciler-cleanup` | call `monitor_checkin("reconciler_cleanup", ...)` from the reconciler cleanup entrypoint |
Create matching Cron monitors in Sentry with those slugs. Emit an
`in_progress` check-in at job start and `ok`/`error` at completion via
`sentry_observability.monitor_checkin(slug_key, status)`.
## 5. Redaction guarantees (fail closed)
Redaction fails *closed*: if a field cannot be proven safe it is dropped rather
than sent. The `before_send` (and `before_send_log`) hook `scrub_event`
recursively redacts every outgoing event; on any error it drops the event
entirely. Guarantees, proven by `tests/test_sentry_observability.py`:
- **No** tokens, passwords, keychain IDs, DSNs, cookies, or `user:pass@host`.
- **No** raw session-state or full prompt/comment bodies — `session_id` is only
ever surfaced as a 12-char `session_id_hash`.
- **No** private config contents or raw credential headers.
- **No** full local filesystem paths — a worktree path collapses to a coarse
`worktree_category` (`author` / `reviewer` / `merger` / `reconciler` /
`branches` / `root` / `other`).
- Only the allowlisted tag keys in `ALLOWED_TAG_KEYS` are ever attached.
## 6. Coexistence with GlitchTip / the #612 incident bridge
This is the **outbound** path (MCP → Sentry SDK). It complements — it does not
replace — the **inbound** [`incident_bridge.py`](../../incident_bridge.py)
(#612), which turns Sentry/GlitchTip *observations* into durable Gitea issues
and `incident_links` rows.
- Prefer **one** observability path per environment. Point the MCP server's
`SENTRY_DSN` at the same self-hosted `gitea-tools-mcp` project that the #612
bridge reconciles from, so an MCP-reported error and its Gitea issue line up.
- GlitchTip is Sentry-protocol compatible; if an existing GlitchTip DSN is in
use, either migrate it to `https://sentry.prgs.cc/` or document the split
(MCP → Sentry, legacy → GlitchTip) explicitly for operators.
- The bridge remains the **only** sanctioned route from an alert back into
Gitea workflow state.
## 7. Non-goals
- Sentry must **not** become the workflow source of truth.
- Sentry must **not** approve, merge, close, or mutate Gitea workflow state.
- Sentry must **not** bypass leases, #332, workflow roles, or the MCP gates.
+19 -207
View File
@@ -243,192 +243,6 @@ def _redact(text):
return str(text) return str(text)
# ── Classified client failures (#699) ─────────────────────────────────────────
# Subclasses of RuntimeError preserve existing ``except RuntimeError`` call
# sites. Exception *messages* are fixed constants only — HTTP response bodies,
# Keychain material, and arbitrary exception text are never stored on the
# exception or re-emitted to tool results / daemon logs.
# Fixed messages (must match mcp_tool_error_boundary.FIXED_MESSAGES keys used here).
_MSG_AUTH_INVALID = "Gitea authentication failed: invalid or revoked credentials"
_MSG_AUTH_FAILED = "Gitea authentication failed"
_MSG_AUTHZ_SCOPE = "Gitea authorization failed: insufficient token scope"
_MSG_AUTHZ_DENIED = "Gitea authorization failed: access denied"
_MSG_NETWORK = "Network error contacting Gitea"
_MSG_CONFIG = "Gitea configuration or credential resolution failed"
_MSG_UPSTREAM = "Gitea upstream unavailable"
_MSG_HTTP = "Gitea HTTP request failed"
class GiteaClientError(RuntimeError):
"""Base for known Gitea client failures with stable reason_code metadata."""
reason_code = "client_error"
error_class = "client"
http_status = None
def __init__(self, message=None, *, reason_code=None, http_status=None):
if reason_code is not None:
self.reason_code = reason_code
if http_status is not None:
self.http_status = http_status
# Message is always a fixed constant; callers cannot inject bodies.
fixed = message if message is not None else _MSG_HTTP
super().__init__(fixed)
class GiteaAuthError(GiteaClientError):
"""Authentication failure (invalid/revoked credentials → typically HTTP 401)."""
reason_code = "auth_invalid_token"
error_class = "authentication"
http_status = 401
def __init__(self, message=None, *, reason_code=None, http_status=None):
super().__init__(
message if message is not None else _MSG_AUTH_INVALID,
reason_code=reason_code or "auth_invalid_token",
http_status=http_status if http_status is not None else 401,
)
class GiteaAuthzError(GiteaClientError):
"""Authorization failure (HTTP 403 — scope deficiency or access denied)."""
reason_code = "authz_denied"
error_class = "authorization"
http_status = 403
def __init__(self, message=None, *, reason_code=None, http_status=None):
code = reason_code or "authz_denied"
if code == "authz_insufficient_scope":
fixed = _MSG_AUTHZ_SCOPE
else:
fixed = _MSG_AUTHZ_DENIED
code = "authz_denied"
super().__init__(
message if message is not None else fixed,
reason_code=code,
http_status=http_status if http_status is not None else 403,
)
class GiteaNetworkError(GiteaClientError):
"""Transport / DNS / timeout failure contacting Gitea."""
reason_code = "network_error"
error_class = "network"
http_status = None
def __init__(self, message=None, *, reason_code=None, http_status=None):
super().__init__(
message if message is not None else _MSG_NETWORK,
reason_code=reason_code or "network_error",
http_status=http_status,
)
class GiteaConfigError(GiteaClientError):
"""Local configuration / credential resolution failure (not HTTP auth)."""
reason_code = "config_error"
error_class = "configuration"
http_status = None
def __init__(self, message=None, *, reason_code=None, http_status=None):
super().__init__(
message if message is not None else _MSG_CONFIG,
reason_code=reason_code or "config_error",
http_status=http_status,
)
class GiteaHttpError(GiteaClientError):
"""Non-auth HTTP failure with fixed message (no response body)."""
reason_code = "http_error"
error_class = "client"
http_status = None
def __init__(self, message=None, *, reason_code=None, http_status=None):
code = reason_code or "http_error"
if code == "upstream_unavailable":
fixed = _MSG_UPSTREAM
else:
fixed = _MSG_HTTP
code = "http_error"
super().__init__(
message if message is not None else fixed,
reason_code=code,
http_status=http_status,
)
def _looks_like_insufficient_scope(detail: str) -> bool:
"""Internal: inspect redacted body *only* to refine 403 reason_code.
The body is never stored on the exception or returned to callers.
"""
lower = (detail or "").lower()
markers = (
"insufficient scope",
"required scope",
"does not have at least one of required scope",
"token does not have",
"missing scope",
"scope(s)",
)
return any(m in lower for m in markers)
def classify_http_status(code: int, *, body_hint: str = "") -> tuple[type, str, int]:
"""Central HTTP status → (exception_class, reason_code, http_status).
Every HTTP 403 becomes authorization-class. Body text is used only as a
local hint for scope vs denied reason_code and is never returned.
"""
if code == 401:
return (GiteaAuthError, "auth_invalid_token", 401)
if code == 403:
if _looks_like_insufficient_scope(body_hint or ""):
return (GiteaAuthzError, "authz_insufficient_scope", 403)
return (GiteaAuthzError, "authz_denied", 403)
if code in (502, 503, 504):
return (GiteaHttpError, "upstream_unavailable", code)
return (GiteaHttpError, "http_error", code)
def raise_for_http_status(code: int, body: str = "") -> None:
"""Raise a typed client error for *code* without embedding *body*.
*body* may be inspected only to choose scope vs denied for 403; it is
never placed on the exception message.
"""
# Redact before any inspection; discard after classification.
try:
hint = _redact(body or "").strip()
except Exception:
hint = ""
exc_cls, reason, status = classify_http_status(code, body_hint=hint)
# Explicitly construct without passing body/hint into message.
if exc_cls is GiteaAuthError:
raise GiteaAuthError(reason_code=reason, http_status=status)
if exc_cls is GiteaAuthzError:
raise GiteaAuthzError(reason_code=reason, http_status=status)
if reason == "upstream_unavailable":
raise GiteaHttpError(
reason_code="upstream_unavailable",
http_status=status,
)
raise GiteaHttpError(reason_code="http_error", http_status=status)
def _raise_http_error(code: int, detail: str = "") -> None:
"""Backward-compatible alias — *detail* is never embedded in the error."""
raise_for_http_status(code, detail)
def _add_query(url, **params): def _add_query(url, **params):
"""Return *url* with the given query parameters added or overridden. """Return *url* with the given query parameters added or overridden.
@@ -501,24 +315,23 @@ def api_request(method, url, auth_header, payload=None, *,
"""Make an authenticated JSON request to the Gitea API. """Make an authenticated JSON request to the Gitea API.
Returns parsed JSON on success (or ``None`` for an empty body), and raises Returns parsed JSON on success (or ``None`` for an empty body), and raises
a classified client error on failure. ``RuntimeError`` on failure.
On HTTP 429 the request is retried up to *max_retries* times: honoring a On HTTP 429 the request is retried up to *max_retries* times: honoring a
valid ``Retry-After`` header (seconds or HTTP-date) when present, otherwise valid ``Retry-After`` header (seconds or HTTP-date) when present, otherwise
using capped jittered exponential backoff. Successful responses are using capped jittered exponential backoff. Successful responses are
unchanged. unchanged.
Failures raise typed exceptions with **fixed messages only** (#699). HTTP All failures are converted to a ``RuntimeError`` with a clear, secret
response bodies are read solely for local 403 reason refinement and are -redacted message (no raw stack traces or credential material):
never stored on exceptions or returned to callers:
- HTTP 401 → :class:`GiteaAuthError` (``auth_invalid_token``) - Non-429 HTTP errors surface the status code and a redacted response body.
- HTTP 403 → :class:`GiteaAuthzError` (scope or denied) 502/503/504 upstream errors get an explicit "Gitea upstream unavailable"
- 502/503/504 → :class:`GiteaHttpError` (``upstream_unavailable``) message.
- Other non-429 HTTP → :class:`GiteaHttpError` (``http_error``) - Timeouts and network/DNS failures (``URLError`` / ``TimeoutError``) surface
- Timeouts / DNS / ``URLError`` → :class:`GiteaNetworkError` a generic "network error contacting Gitea" message.
- Malformed success JSON → plain ``RuntimeError`` (programming/protocol; - A malformed (non-JSON) success body surfaces a "malformed JSON response"
not reclassified as authentication) message rather than a raw decode error.
The ``*_func`` parameters and ``timeout`` are injection points for The ``*_func`` parameters and ``timeout`` are injection points for
deterministic testing. deterministic testing.
@@ -556,23 +369,22 @@ def api_request(method, url, auth_header, payload=None, *,
error_body = e.read().decode("utf-8", errors="replace") error_body = e.read().decode("utf-8", errors="replace")
except Exception: except Exception:
error_body = "" error_body = ""
# Classify from status (+ local body hint). Body is not embedded. detail = _redact(error_body).strip()
try: if e.code in (502, 503, 504):
raise_for_http_status(e.code, error_body) msg = f"HTTP {e.code}: Gitea upstream unavailable"
except GiteaClientError: raise RuntimeError(f"{msg}: {detail}" if detail else msg) from e
raise raise RuntimeError(f"HTTP {e.code}: {detail}") from e
# Defensive: raise_for_http_status always raises.
raise GiteaHttpError(http_status=e.code) from e # pragma: no cover
except (urllib.error.URLError, TimeoutError) as e: except (urllib.error.URLError, TimeoutError) as e:
# Fixed message only — do not embed URLError reason (may leak paths). reason = getattr(e, "reason", e)
raise GiteaNetworkError(reason_code="network_error") from e raise RuntimeError(
f"network error contacting Gitea: {_redact(reason)}"
) from e
if not body: if not body:
return None return None
try: try:
return json.loads(body) return json.loads(body)
except ValueError as e: except ValueError as e:
# Programming/protocol failure — not authentication.
raise RuntimeError("malformed JSON response from Gitea") from e raise RuntimeError("malformed JSON response from Gitea") from e
+145 -586
View File
@@ -623,280 +623,101 @@ def verify_preflight_purity(
remote: str | None = None, remote: str | None = None,
worktree_path: str | None = None, worktree_path: str | None = None,
task: str | None = None, task: str | None = None,
*,
target_issue_number: int | None = None,
require_author_lock: bool = False,
): ):
"""Verify identity/capability order, then production workspace guards. """Verify that identity and capability were verified prior to session edits."""
#683: pytest/unittest must not skip production root/branches/scope
enforcement when force-on signals request production behavior. The
early return below only skips *preflight-order* purity checks under
pure unit-test isolation never when production guards are active.
"""
global _preflight_reviewer_violation_files global _preflight_reviewer_violation_files
in_test = _preflight_in_test_mode() in_test = _preflight_in_test_mode()
production_active = workflow_scope_guard.production_guards_active( if in_test and not (
in_test_mode=in_test os.environ.get("GITEA_TEST_FORCE_DIRTY")
) or os.environ.get("GITEA_TEST_PORCELAIN") is not None
# Pure unit-test isolation: skip purity-order unless legacy dirty/porcelain ):
# force flags request the dirtiness path. #683 FORCE_PRODUCTION_GUARDS alone
# runs production root/branches/scope without requiring whoami/capability.
skip_purity_order = in_test and not workflow_scope_guard.purity_order_forced()
if not skip_purity_order:
if not _preflight_whoami_called:
raise RuntimeError(
"Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)"
)
if not _preflight_capability_called:
raise RuntimeError(
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
)
if (
task is not None
and _preflight_resolved_task is not None
and task != _preflight_resolved_task
):
raise RuntimeError(
"Pre-flight task mismatch: "
f"resolved '{_preflight_resolved_task}' but mutation requires "
f"'{task}' (fail closed)"
)
# #671: block review/merge/close/completion mutations while the session is
# contaminated by a direct stable-branch push attempt (reconciler-exempt).
_enforce_stable_branch_contamination_gate(task, remote)
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
canonical_root = ctx["canonical_repo_root"]
process_root = ctx["process_project_root"]
real_workspace = os.path.realpath(workspace)
role = ctx.get("workspace_role_kind") or _effective_workspace_role()
if real_workspace != process_root:
if not _preflight_in_test_mode():
membership = author_mutation_worktree.assess_workspace_repo_membership(
workspace_path=workspace,
canonical_repo_root=canonical_root,
)
if membership["block"]:
raise RuntimeError(
author_mutation_worktree.format_workspace_repo_membership_error(
membership
)
)
dirty_files = sorted(
_parse_porcelain_entries(_get_workspace_porcelain(workspace))
)
if dirty_files:
raise RuntimeError(
nwb.format_namespace_workspace_binding_error(
role_kind=role,
workspace_path=workspace,
binding_source=ctx.get("workspace_binding_source")
or "unknown binding source",
dirty_files=dirty_files,
ignored_bindings=ctx.get("ignored_bindings"),
)
)
else:
if _preflight_whoami_violation:
raise RuntimeError(
"Pre-flight order violation: Workspace file edits occurred before "
f"gitea_whoami verification (fail closed). Offending files: "
f"{_format_preflight_files(_preflight_whoami_violation_files)}"
)
if _preflight_capability_violation:
raise RuntimeError(
"Pre-flight order violation: Workspace file edits occurred before "
f"gitea_resolve_task_capability verification (fail closed). Offending files: "
f"{_format_preflight_files(_preflight_capability_violation_files)}"
)
if role in {"reviewer", "merger"}:
current = _get_workspace_porcelain()
baseline = _preflight_capability_baseline_porcelain or ""
reviewer_delta = _new_tracked_changes_since(baseline, current)
_preflight_reviewer_violation_files = reviewer_delta
if reviewer_delta:
raise RuntimeError(
f"{role.title()} role violation: profile is forbidden from modifying "
"tracked workspace files (fail closed). Offending files: "
f"{_format_preflight_files(reviewer_delta)}"
)
# Historical path: root + branches after purity-order when dirty paths live.
_enforce_root_checkout_guard(worktree_path)
_enforce_branches_only_author_mutation(worktree_path)
_enforce_issue_scope_guard(
worktree_path,
task=task,
target_issue_number=target_issue_number,
require_author_lock=require_author_lock,
)
_clear_preflight_capability_state()
return return
# #683: under pytest unit isolation, FORCE_PRODUCTION_GUARDS still runs if not _preflight_whoami_called:
# production root + branches + issue scope (no silent no-op of guards). raise RuntimeError(
if production_active: "Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)"
_enforce_root_checkout_guard(worktree_path) )
_enforce_branches_only_author_mutation(worktree_path) if not _preflight_capability_called:
_enforce_issue_scope_guard( raise RuntimeError(
worktree_path, "Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
task=task, )
target_issue_number=target_issue_number, if (
require_author_lock=require_author_lock, task is not None
and _preflight_resolved_task is not None
and task != _preflight_resolved_task
):
raise RuntimeError(
"Pre-flight task mismatch: "
f"resolved '{_preflight_resolved_task}' but mutation requires "
f"'{task}' (fail closed)"
) )
# #671: block review/merge/close/completion mutations while the session is
# contaminated by a direct stable-branch push attempt (reconciler-exempt).
_enforce_stable_branch_contamination_gate(task, remote)
def _session_issue_lock_snapshot(
workspace_path: str | None = None,
) -> dict:
"""Return session lock fields relevant to #683 scope enforcement.
Branch-vs-lock comparison uses the live workspace branch only when the
lock's worktree matches the mutation workspace. That prevents a foreign
or leftover session lock from poisoning unrelated test worktrees while
still fail-closing when the bound worktree drifts to another issue.
"""
lock = issue_lock_store.read_session_issue_lock() or {}
raw = lock.get("issue_number")
locked: int | None
try:
locked = int(raw) if raw is not None else None
except (TypeError, ValueError):
locked = None
lock_wt = (lock.get("worktree_path") or "").strip()
workspace = (workspace_path or "").strip()
worktrees_match = False
if lock_wt and workspace:
try:
worktrees_match = os.path.realpath(lock_wt) == os.path.realpath(workspace)
except OSError:
worktrees_match = False
return {
"locked_issue_number": locked,
"lock_branch_name": (lock.get("branch_name") or "").strip() or None,
"lock_worktree_path": lock_wt or None,
"worktrees_match": worktrees_match,
}
def _session_locked_issue_number() -> int | None:
"""Return the active session issue lock number when present (#683)."""
return _session_issue_lock_snapshot().get("locked_issue_number")
def _enforce_issue_scope_guard(
worktree_path: str | None = None,
*,
task: str | None = None,
target_issue_number: int | None = None,
require_author_lock: bool = False,
) -> None:
"""#683: fail closed on missing/out-of-scope issue ownership for mutations."""
ctx = _resolve_namespace_mutation_context(worktree_path) ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"] workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(workspace) canonical_root = ctx["canonical_repo_root"]
# Honour actual profile role as well as poisoned task role (#540 / #683): process_root = ctx["process_project_root"]
# comment_issue preflight stamps required_role_kind=author, which must not real_workspace = os.path.realpath(workspace)
# strip a genuine reconciler of control-checkout exemptions.
role = ctx.get("workspace_role_kind") or _effective_workspace_role() role = ctx.get("workspace_role_kind") or _effective_workspace_role()
actual = _actual_profile_role()
if actual in nwb.NON_AUTHOR_ROLES:
role = actual
snap = _session_issue_lock_snapshot(workspace)
# Scope uses the lock's recorded branch for issue-number matching.
# Live workspace branch can inherit the parent control checkout's branch
# name when a temp branches/ dir is not its own worktree tip — that must
# not invent a false out-of-scope failure. Live branch drift is enforced
# by issue_lock_store.verify_lock_for_mutation elsewhere.
branch_for_scope = snap.get("lock_branch_name")
if (
snap.get("worktrees_match")
and workflow_scope_guard.production_guards_forced()
):
live_branch = git_state.get("current_branch")
live_issue = workflow_scope_guard.extract_issue_number_from_branch(
live_branch
)
locked = snap.get("locked_issue_number")
if (
live_issue is not None
and locked is not None
and live_issue != locked
):
branch_for_scope = live_branch
# Author implementation / source-adjacent mutations need ownership when forced.
authorish = role == "author" or (
task
in {
"create_issue",
"comment_issue",
"lock_issue",
"create_pr",
"commit_files",
"gitea_commit_files",
"mark_issue",
}
)
require_lock = bool(require_author_lock) or (
authorish
and workflow_scope_guard.production_guards_forced()
and role == "author"
)
assessment = workflow_scope_guard.assess_production_mutation_guards(
workspace_path=workspace,
canonical_repo_root=ctx["canonical_repo_root"],
porcelain_status=git_state.get("porcelain_status") or "",
current_branch=branch_for_scope,
locked_issue_number=snap.get("locked_issue_number"),
target_issue_number=target_issue_number,
role_kind=role,
require_author_lock=require_lock,
in_test_mode=_preflight_in_test_mode(),
)
workflow_scope_guard.raise_if_blocked(assessment)
if real_workspace != process_root:
if not _preflight_in_test_mode():
membership = author_mutation_worktree.assess_workspace_repo_membership(
workspace_path=workspace,
canonical_repo_root=canonical_root,
)
if membership["block"]:
raise RuntimeError(
author_mutation_worktree.format_workspace_repo_membership_error(
membership
)
)
def _production_guard_block_from_exc(exc: BaseException, **extra) -> dict | None: dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
"""Map production-guard exceptions to typed tool block responses (#683).""" if dirty_files:
if isinstance(exc, workflow_scope_guard.ProductionGuardError): raise RuntimeError(
return workflow_scope_guard.block_response(exc, **extra) nwb.format_namespace_workspace_binding_error(
text = str(exc) role_kind=role,
if "Workflow scope guard (#683)" in text or "Root checkout guard (#475)" in text: workspace_path=workspace,
kind = workflow_scope_guard.BLOCKER_PRODUCTION_GUARD binding_source=ctx.get("workspace_binding_source")
if "root_diagnostic_edit" in text or "tracked source or test edits" in text: or "unknown binding source",
kind = workflow_scope_guard.BLOCKER_ROOT_DIAGNOSTIC_EDIT dirty_files=dirty_files,
elif "Branches-only mutation guard" in text or "stable control checkout" in text: ignored_bindings=ctx.get("ignored_bindings"),
kind = workflow_scope_guard.BLOCKER_MISSING_WORKTREE )
elif "out-of-scope" in text or "locked to issue" in text: )
kind = workflow_scope_guard.BLOCKER_OUT_OF_SCOPE_ISSUE else:
elif "no owning issue" in text: if _preflight_whoami_violation:
kind = workflow_scope_guard.BLOCKER_MISSING_ISSUE_SCOPE raise RuntimeError(
return workflow_scope_guard.block_response( "Pre-flight order violation: Workspace file edits occurred before "
blocker_kind=kind, f"gitea_whoami verification (fail closed). Offending files: "
reasons=[text], f"{_format_preflight_files(_preflight_whoami_violation_files)}"
**extra, )
) if _preflight_capability_violation:
if "Branches-only mutation guard" in text: raise RuntimeError(
return workflow_scope_guard.block_response( "Pre-flight order violation: Workspace file edits occurred before "
blocker_kind=workflow_scope_guard.BLOCKER_MISSING_WORKTREE, f"gitea_resolve_task_capability verification (fail closed). Offending files: "
reasons=[text], f"{_format_preflight_files(_preflight_capability_violation_files)}"
**extra, )
)
if "Root checkout guard" in text: if role in {"reviewer", "merger"}:
return workflow_scope_guard.block_response( current = _get_workspace_porcelain()
blocker_kind=workflow_scope_guard.BLOCKER_ROOT_DIAGNOSTIC_EDIT, baseline = _preflight_capability_baseline_porcelain or ""
reasons=[text], reviewer_delta = _new_tracked_changes_since(baseline, current)
**extra, _preflight_reviewer_violation_files = reviewer_delta
) if reviewer_delta:
return None raise RuntimeError(
f"{role.title()} role violation: profile is forbidden from modifying "
"tracked workspace files (fail closed). Offending files: "
f"{_format_preflight_files(reviewer_delta)}"
)
_enforce_root_checkout_guard(worktree_path)
_enforce_branches_only_author_mutation(worktree_path)
_clear_preflight_capability_state()
def _verify_role_mutation_workspace( def _verify_role_mutation_workspace(
@@ -906,12 +727,7 @@ def _verify_role_mutation_workspace(
worktree: str | None = None, worktree: str | None = None,
task: str | None = None, task: str | None = None,
) -> str: ) -> str:
"""Bind reviewer/merger mutations to the active namespace workspace (#510). """Bind reviewer/merger mutations to the active namespace workspace (#510)."""
#683: must NOT early-return solely because pytest/unittest is loaded.
Production workspace binding always runs; test isolation uses explicit
env fixtures / force-on flags, never a production short-circuit here.
"""
# Check running runtimes to prevent stale mutations # Check running runtimes to prevent stale mutations
try: try:
@@ -1084,9 +900,7 @@ from gitea_auth import ( # noqa: E402
repo_api_url, repo_api_url,
get_profile, get_profile,
gitea_url, gitea_url,
GiteaConfigError,
) )
import mcp_tool_error_boundary # noqa: E402
import gitea_audit # noqa: E402 import gitea_audit # noqa: E402
import gitea_config # noqa: E402 import gitea_config # noqa: E402
import capability_stop_terminal # noqa: E402 import capability_stop_terminal # noqa: E402
@@ -1104,6 +918,7 @@ import allocator_service # noqa: E402
import control_plane_db # noqa: E402 import control_plane_db # noqa: E402
import lease_lifecycle # noqa: E402 import lease_lifecycle # noqa: E402
import incident_bridge # noqa: E402 import incident_bridge # noqa: E402
import sentry_observability # noqa: E402 (#606 optional Sentry observability)
import agent_temp_artifacts import agent_temp_artifacts
import issue_lock_worktree # noqa: E402 import issue_lock_worktree # noqa: E402
import issue_lock_provenance # noqa: E402 import issue_lock_provenance # noqa: E402
@@ -1114,7 +929,6 @@ import merge_approval_gate # noqa: E402
import already_landed_reconcile # noqa: E402 import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402 import author_mutation_worktree # noqa: E402
import root_checkout_guard # noqa: E402 import root_checkout_guard # noqa: E402
import workflow_scope_guard # noqa: E402 # #683 production scope / force-on guards
import stable_branch_push_guard # noqa: E402 import stable_branch_push_guard # noqa: E402
import remote_repo_guard # noqa: E402 import remote_repo_guard # noqa: E402
import issue_claim_heartbeat # noqa: E402 import issue_claim_heartbeat # noqa: E402
@@ -1466,12 +1280,6 @@ def _with_optional_url(result: dict, url: str | None) -> dict:
result["url"] = url result["url"] = url
return result return result
# #699: known auth/authz/network/config failures → structured CallToolResult
# isError; stdio transport must survive (no unhandled raise / process exit).
from mcp.server.fastmcp.tools.base import Tool as _FastMCPTool # noqa: E402
mcp_tool_error_boundary.install_tool_run_boundary(_FastMCPTool)
mcp = FastMCP("gitea-tools", instructions=( mcp = FastMCP("gitea-tools", instructions=(
"Gitea issue tracker and PR management for dadeschools and prgs instances. " "Gitea issue tracker and PR management for dadeschools and prgs instances. "
"Use the gitea_ prefixed tools to create issues, PRs, list issues, etc." "Use the gitea_ prefixed tools to create issues, PRs, list issues, etc."
@@ -1809,16 +1617,13 @@ def _enforce_remote_repo_guard(
def _auth(host: str) -> str: def _auth(host: str) -> str:
"""Get auth header, raise if unavailable. """Get auth header, raise if unavailable."""
Missing credentials are a configuration failure, not a silent internal
crash. Typed as :class:`gitea_auth.GiteaConfigError` so the tool-error
boundary (#699) maps them to a structured isError result without EOF.
The exception message is a fixed constant (no host/token material).
"""
header = get_auth_header(host) header = get_auth_header(host)
if header is None: if header is None:
raise GiteaConfigError(reason_code="config_error") raise RuntimeError(
f"No credentials for {host}. "
"Ensure you've logged in via HTTPS at least once."
)
return header return header
@@ -1973,6 +1778,18 @@ def _audited(action: str, *, host, remote, org=None, repo=None,
result=gitea_audit.FAILED, reason=_redact(str(exc)), result=gitea_audit.FAILED, reason=_redact(str(exc)),
request_metadata=request_metadata, issue_number=issue_number, request_metadata=request_metadata, issue_number=issue_number,
pr_number=pr_number, target_branch=target_branch) pr_number=pr_number, target_branch=target_branch)
# #606: best-effort Sentry capture of the failing mutation (fail open).
sentry_observability.capture_exception(
exc,
tags={
"mutation_tool": action,
"remote": remote,
"repo": repo,
"org": org,
"issue_number": issue_number,
"pr_number": pr_number,
},
)
raise raise
_audit(action, host=host, remote=remote, org=org, repo=repo, _audit(action, host=host, remote=remote, org=org, repo=repo,
result=gitea_audit.SUCCEEDED, request_metadata=request_metadata, result=gitea_audit.SUCCEEDED, request_metadata=request_metadata,
@@ -2019,6 +1836,20 @@ def _audit_pr_result(action: str):
"merge_method": result.get("merge_method"), "merge_method": result.get("merge_method"),
}, },
) )
# #606: surface fail-closed blockers / failed mutations to
# Sentry as structured events (best-effort, fail open).
if status in (gitea_audit.BLOCKED, gitea_audit.FAILED):
sentry_observability.capture_workflow_blocker(
action,
message="; ".join(reasons) or action,
next_action=result.get("safe_next_action"),
level="error" if status == gitea_audit.FAILED else "warning",
tags={
"mutation_tool": action,
"pr_number": result.get("pr_number"),
"current_head_sha": result.get("head_sha"),
},
)
except Exception: except Exception:
pass # best-effort; never break the tool pass # best-effort; never break the tool
return result return result
@@ -2092,15 +1923,7 @@ def gitea_create_issue(
) )
if blocked: if blocked:
return blocked return blocked
try: verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
verify_preflight_purity(
remote, worktree_path=worktree_path, task="create_issue"
)
except Exception as exc:
typed = _production_guard_block_from_exc(exc, number=None)
if typed is not None:
return typed
raise
content_gate = issue_content_gate.pre_create_issue_content_gate( content_gate = issue_content_gate.pre_create_issue_content_gate(
title, title,
body, body,
@@ -7975,12 +7798,11 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
org: str | None = None, org: str | None = None,
repo: str | None = None, repo: str | None = None,
) -> dict: ) -> dict:
"""Read-only: diagnose open-PR reviewer lease handoff and emit next action (#599, #691). """Read-only: diagnose open-PR reviewer lease handoff and emit next action (#599).
Classifies no-lease / own / foreign / superseded-head / expired / orphan / Classifies no-lease / own / foreign / instructed-lease-missing-with-replacement
instructed-lease-missing-with-replacement and worktree binding mismatch. and worktree binding mismatch. Returns a canonical ``next_action`` without
Returns a canonical ``next_action`` (including guarded obsolete-lease cleanup) mutating Gitea state. Never steals foreign leases.
without mutating Gitea state. Never steals foreign leases.
""" """
read_block = _profile_operation_gate("gitea.read") read_block = _profile_operation_gate("gitea.read")
if read_block: if read_block:
@@ -7991,7 +7813,7 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
} }
comments = _fetch_pr_comments( comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo) pr_number, remote=remote, host=host, org=org, repo=repo)
h, o, r = _resolve(remote, host, org, repo) h, _o, _r = _resolve(remote, host, org, repo)
try: try:
username = _authenticated_username(h) username = _authenticated_username(h)
except Exception: except Exception:
@@ -8005,64 +7827,6 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
# Prefer in-session lease id when present; otherwise diagnose as a fresh # Prefer in-session lease id when present; otherwise diagnose as a fresh
# caller without inventing ownership of an on-thread lease. # caller without inventing ownership of an on-thread lease.
session_id = (session_lease.get("session_id") or "").strip() or None session_id = (session_lease.get("session_id") or "").strip() or None
# Live PR head + formal reviews for #691 superseded/completed classification.
current_head_sha = None
formal_reviews: list[dict] = []
try:
auth = _auth(h)
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth
) or {}
head = pr_live.get("head") or {}
current_head_sha = head.get("sha") or pr_live.get("head_commit_sha")
try:
reviews = api_request(
"GET",
f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews",
auth,
) or []
for rev in reviews if isinstance(reviews, list) else []:
formal_reviews.append({
"verdict": rev.get("state") or rev.get("verdict"),
"reviewed_head_sha": rev.get("commit_id") or rev.get(
"reviewed_head_sha"
),
"dismissed": bool(rev.get("dismissed")),
"reviewer": ((rev.get("user") or {}).get("login")),
"submitted_at": rev.get("submitted_at"),
})
except Exception:
formal_reviews = []
except Exception:
current_head_sha = None
lease_wt = None
active = reviewer_pr_lease.find_newest_nonterminal_lease(
comments, pr_number=pr_number, include_expired=True
)
if active:
lease_wt = (active.get("worktree") or "").strip() or None
worktree_exists = None
worktree_clean = None
if lease_wt:
worktree_exists = os.path.isdir(lease_wt)
if worktree_exists:
try:
import subprocess
st = subprocess.run(
["git", "-C", lease_wt, "status", "--porcelain"],
capture_output=True,
text=True,
timeout=30,
check=False,
)
if st.returncode == 0:
worktree_clean = not bool((st.stdout or "").strip())
except Exception:
worktree_clean = None
diagnosis = reviewer_pr_lease.diagnose_reviewer_pr_lease_handoff( diagnosis = reviewer_pr_lease.diagnose_reviewer_pr_lease_handoff(
comments, comments,
pr_number=pr_number, pr_number=pr_number,
@@ -8072,11 +7836,6 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
env_bound_worktree=env_wt, env_bound_worktree=env_wt,
instructed_session_id=instructed_session_id, instructed_session_id=instructed_session_id,
instructed_comment_id=instructed_comment_id, instructed_comment_id=instructed_comment_id,
current_head_sha=current_head_sha,
formal_reviews=formal_reviews,
worktree_exists=worktree_exists,
worktree_clean=worktree_clean,
owner_process_alive=None, # PID never proves ownership; leave unset
) )
diagnosis["success"] = True diagnosis["success"] = True
diagnosis["remote"] = remote diagnosis["remote"] = remote
@@ -8204,212 +7963,6 @@ def gitea_cleanup_post_merge_moot_lease(
return report return report
@mcp.tool()
def gitea_cleanup_obsolete_reviewer_comment_lease(
pr_number: int,
confirmation: str = "",
apply: bool = False,
controller_recovery_authorized: bool = False,
expected_lease_comment_id: int | None = None,
expected_session_id: str | None = None,
expected_leased_head: str | None = None,
owner_process_alive: bool | None = None,
current_head_review_in_progress: bool = False,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Guarded cleanup for obsolete comment-backed reviewer leases on open PRs (#691).
Neutralises an expired and/or superseded-head foreign lease by posting a
terminal ``phase: released`` lease marker (append-only audit). Never deletes
comments, never repoints the lease to the new head, never transfers
validation/decision/workflow state, never uses PID equality as ownership,
and never steals a genuinely active foreign lease on the current head.
Read-first when ``apply`` is false. Apply requires:
- ``controller_recovery_authorized=True`` (explicit operator/controller
recovery capability not implied by profile alone)
- ``confirmation`` exactly ``CLEANUP OBSOLETE REVIEWER LEASE <pr_number>``
- full eligibility evidence from
``reviewer_pr_lease.assess_obsolete_reviewer_comment_lease_cleanup``
Comment-backed leases need no control-plane DB ``lease_id``.
"""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"cleanup_performed": False,
"no_steal_or_adoption": True,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
expected_repo = f"{o}/{r}"
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo
)
pr_live = api_request(
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth
) or {}
head = pr_live.get("head") or {}
current_head_sha = head.get("sha") or pr_live.get("head_commit_sha")
formal_reviews: list[dict] = []
try:
reviews = api_request(
"GET",
f"{repo_api_url(h, o, r)}/pulls/{pr_number}/reviews",
auth,
) or []
for rev in reviews if isinstance(reviews, list) else []:
formal_reviews.append({
"verdict": rev.get("state") or rev.get("verdict"),
"reviewed_head_sha": rev.get("commit_id")
or rev.get("reviewed_head_sha"),
"dismissed": bool(rev.get("dismissed")),
"reviewer": ((rev.get("user") or {}).get("login")),
"submitted_at": rev.get("submitted_at"),
})
except Exception:
formal_reviews = []
lease = reviewer_pr_lease.find_newest_nonterminal_lease(
comments, pr_number=pr_number, include_expired=True
)
lease_wt = ((lease or {}).get("worktree") or "").strip() or None
worktree_exists = None
worktree_clean = None
worktree_has_unpreserved_work = None
if lease_wt:
worktree_exists = os.path.isdir(lease_wt)
if worktree_exists:
try:
import subprocess
st = subprocess.run(
["git", "-C", lease_wt, "status", "--porcelain"],
capture_output=True,
text=True,
timeout=30,
check=False,
)
if st.returncode == 0:
dirty = bool((st.stdout or "").strip())
worktree_clean = not dirty
worktree_has_unpreserved_work = dirty
except Exception:
worktree_clean = None
session = reviewer_pr_lease.get_session_lease() or {}
assessment = reviewer_pr_lease.assess_obsolete_reviewer_comment_lease_cleanup(
comments,
pr_number=pr_number,
current_head_sha=current_head_sha,
formal_reviews=formal_reviews,
repo=expected_repo,
expected_repo=expected_repo,
requesting_session_id=(session.get("session_id") or "").strip() or None,
controller_recovery_authorized=controller_recovery_authorized,
worktree_exists=worktree_exists,
worktree_clean=worktree_clean,
worktree_has_unpreserved_work=worktree_has_unpreserved_work,
owner_process_alive=owner_process_alive,
owner_pid_observed=None,
requesting_pid=os.getpid(),
current_head_review_in_progress=current_head_review_in_progress,
expected_lease_comment_id=expected_lease_comment_id,
expected_session_id=expected_session_id,
expected_leased_head=expected_leased_head,
confirmation=confirmation,
apply=apply,
)
report = {
"success": True,
"pr_number": pr_number,
"pr_state": pr_live.get("state"),
"cleanup_allowed": assessment.get("cleanup_allowed"),
"cleanup_performed": False,
"classification": assessment.get("classification"),
"blocker_kind": assessment.get("blocker_kind"),
"exact_next_action": assessment.get("exact_next_action"),
"mutation_eligibility": assessment.get("mutation_eligibility"),
"leased_head": assessment.get("leased_head"),
"current_head": assessment.get("current_head"),
"expires_at": assessment.get("expires_at"),
"terminal_review": assessment.get("terminal_review"),
"worktree_state": assessment.get("worktree_state"),
"owner_session_evidence": assessment.get("owner_session_evidence"),
"cleanup_tool": assessment.get("cleanup_tool"),
"required_confirmation": assessment.get("required_confirmation"),
"active_lease": assessment.get("active_lease"),
"no_steal_or_adoption": True,
"no_repoint": True,
"no_validation_transfer": True,
"mode": "apply" if apply else "read_only",
"reasons": assessment.get("reasons") or [],
"forbidden": assessment.get("forbidden") or [],
}
if not apply:
return report
if not assessment.get("cleanup_allowed"):
report["success"] = False
report["cleanup_skipped_reason"] = (
assessment.get("fail_closed_reasons")
or assessment.get("reasons")
or ["cleanup not allowed"]
)
return report
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
report["success"] = False
report["reasons"] = comment_block
report["permission_report"] = _permission_block_report("gitea.pr.comment")
return report
verify_preflight_purity(remote)
# Terminal lease marker (preserves history; does not delete comment 10749-style
# entries — newest released marker neutralises the ledger).
release_body = assessment.get("release_body") or ""
audit_body = assessment.get("audit_comment_body") or ""
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
with _audited(
"comment_pr",
host=h,
remote=remote,
org=o,
repo=r,
pr_number=pr_number,
request_metadata={
"source": "cleanup_obsolete_reviewer_comment_lease",
"classification": assessment.get("classification"),
},
):
posted_release = api_request(
"POST", comment_url, auth, {"body": release_body}
)
posted_audit = None
if audit_body.strip():
posted_audit = api_request(
"POST", comment_url, auth, {"body": audit_body}
)
# Never seed session lease from cleaned foreign lease.
report["cleanup_performed"] = True
report["released_comment_id"] = (posted_release or {}).get("id")
report["audit_comment_id"] = (posted_audit or {}).get("id")
report["session_lease_seeded"] = False
return report
@mcp.tool() @mcp.tool()
def gitea_release_reviewer_pr_lease( def gitea_release_reviewer_pr_lease(
pr_number: int, pr_number: int,
@@ -8641,26 +8194,9 @@ def gitea_create_issue_comment(
with the reveal opt-in); on a permission block or empty body, with the reveal opt-in); on a permission block or empty body,
'success'/'performed' False and 'reasons' with no API call made 'success'/'performed' False and 'reasons' with no API call made
(permission blocks also carry a structured 'permission_report', (permission blocks also carry a structured 'permission_report',
#142). On production-guard blocks (#683): 'blocker_kind' and #142).
'exact_next_action' with no API side effect.
""" """
try: verify_preflight_purity(remote, worktree_path=worktree_path, task="comment_issue")
# Do not pass target_issue_number: comments on other issues remain
# allowed while an author holds a different implementation lock.
# Scope ownership for source edits is enforced via branch/worktree
# binding + root diagnostic checks (#683).
verify_preflight_purity(
remote,
worktree_path=worktree_path,
task="comment_issue",
)
except Exception as exc:
typed = _production_guard_block_from_exc(
exc, issue_number=issue_number
)
if typed is not None:
return typed
raise
gate_reasons = _profile_operation_gate("gitea.issue.comment") gate_reasons = _profile_operation_gate("gitea.issue.comment")
reasons = list(gate_reasons) reasons = list(gate_reasons)
if not (body or "").strip(): if not (body or "").strip():
@@ -10445,6 +9981,11 @@ def gitea_assess_mcp_namespace_health(
probe_source=probe_source, probe_source=probe_source,
) )
_record_live_namespace_health(result) _record_live_namespace_health(result)
# #606: namespace-health watchdog check-in (best-effort, fail open).
sentry_observability.monitor_checkin(
"namespace_health",
"ok" if result.get("healthy", result.get("callable", True)) else "error",
)
return result return result
@@ -12372,6 +11913,18 @@ def gitea_allocate_next_work(
result["inventory_source"] = ( result["inventory_source"] = (
"candidates_json" if candidates_json else "gitea_live" "candidates_json" if candidates_json else "gitea_live"
) )
# #606: watchdog check-ins for the recurring jobs this allocator run
# performs — global stale-lease expiry, terminal-lock lookup, and the
# allocator itself. Best-effort; a failed selection reports "error".
_alloc_ok = bool(result.get("success"))
sentry_observability.monitor_checkin(
"allocator_health", "ok" if _alloc_ok else "error"
)
if _alloc_ok:
# These two scans complete inside allocate_next_work before selection;
# a successful result proves both ran.
sentry_observability.monitor_checkin("stale_lease_scan", "ok")
sentry_observability.monitor_checkin("terminal_lock_scan", "ok")
return result return result
@@ -12696,4 +12249,10 @@ if __name__ == "__main__":
# processes (e.g. review_pr.py) can detect and refuse profile # processes (e.g. review_pr.py) can detect and refuse profile
# side-channel overrides (#199). # side-channel overrides (#199).
_export_session_profile_lock() _export_session_profile_lock()
# #606: optional self-hosted Sentry observability. No-op unless
# MCP_SENTRY_ENABLED is truthy and SENTRY_DSN is set; never blocks startup.
_sentry_status = sentry_observability.init_sentry()
sys.stderr.write(
f"--- Sentry observability: {_sentry_status.get('reason')} ---\n"
)
mcp.run(transport="stdio") mcp.run(transport="stdio")
-1
View File
@@ -13,7 +13,6 @@ from typing import Any
AUTHORIZED_CLEANUP_TOOLS = frozenset({ AUTHORIZED_CLEANUP_TOOLS = frozenset({
"gitea_cleanup_post_merge_moot_lease", "gitea_cleanup_post_merge_moot_lease",
"gitea_cleanup_obsolete_reviewer_comment_lease",
"gitea_reconcile_merged_cleanups", "gitea_reconcile_merged_cleanups",
"gitea_delete_branch", "gitea_delete_branch",
"gitea_cleanup_merged_pr_branch", "gitea_cleanup_merged_pr_branch",
-451
View File
@@ -1,451 +0,0 @@
"""MCP tool-boundary error mapping for known Gitea client failures (#699).
Known authentication / authorization / network / configuration failures leave
the tool boundary as a sanitized structured ``CallToolResult`` with
``isError=True``. Stdio transport remains connected.
Design constraints (reviewer-ratified, #699 / PR #701):
1. **No secret material** in tool results or daemon logs. Messages are fixed
constants keyed by ``reason_code``; HTTP bodies / exception text never
surface. Sanitization failure fails closed to ``internal_error``.
2. **Narrow boundary** wraps the original FastMCP ``Tool.run`` success path;
re-raises framework control-flow exceptions (``UrlElicitationRequiredError``);
maps only typed client failures. Installation is idempotent.
3. **No RuntimeError substring heuristics.** Only explicit typed
authentication / authorization / network / configuration exceptions
receive those labels.
4. **HTTP classification** lives in ``gitea_auth.classify_http_status``;
every 403 is authorization-class.
"""
from __future__ import annotations
import json
import logging
import sys
from typing import Any
logger = logging.getLogger("gitea_mcp.tool_error_boundary")
# Stable reason codes (#699).
REASON_AUTH_FAILED = "auth_failed"
REASON_AUTH_INVALID_TOKEN = "auth_invalid_token"
REASON_AUTHZ_INSUFFICIENT_SCOPE = "authz_insufficient_scope"
REASON_AUTHZ_DENIED = "authz_denied"
REASON_NETWORK_ERROR = "network_error"
REASON_CONFIG_ERROR = "config_error"
REASON_INTERNAL_ERROR = "internal_error"
REASON_UPSTREAM_UNAVAILABLE = "upstream_unavailable"
REASON_HTTP_ERROR = "http_error"
ERROR_CLASS_AUTHENTICATION = "authentication"
ERROR_CLASS_AUTHORIZATION = "authorization"
ERROR_CLASS_NETWORK = "network"
ERROR_CLASS_CONFIGURATION = "configuration"
ERROR_CLASS_INTERNAL = "internal"
ERROR_CLASS_UPSTREAM = "upstream"
# Fixed, secret-free operator messages. Never interpolate HTTP bodies,
# Keychain contents, tokens, or arbitrary exception text.
FIXED_MESSAGES: dict[str, str] = {
REASON_AUTH_FAILED: "Gitea authentication failed",
REASON_AUTH_INVALID_TOKEN: (
"Gitea authentication failed: invalid or revoked credentials"
),
REASON_AUTHZ_INSUFFICIENT_SCOPE: (
"Gitea authorization failed: insufficient token scope"
),
REASON_AUTHZ_DENIED: "Gitea authorization failed: access denied",
REASON_NETWORK_ERROR: "Network error contacting Gitea",
REASON_CONFIG_ERROR: "Gitea configuration or credential resolution failed",
REASON_INTERNAL_ERROR: "Internal tool error",
REASON_UPSTREAM_UNAVAILABLE: "Gitea upstream unavailable",
REASON_HTTP_ERROR: "Gitea HTTP request failed",
}
_INSTALL_FLAG = "_gitea_auth_boundary_installed"
_ORIGINAL_ATTR = "_gitea_auth_boundary_original"
def fixed_message(reason_code: str) -> str:
"""Return the fixed sanitized message for *reason_code* (fail closed)."""
return FIXED_MESSAGES.get(reason_code, FIXED_MESSAGES[REASON_INTERNAL_ERROR])
def _safe_profile_name() -> str | None:
try:
from gitea_auth import get_profile
name = (get_profile() or {}).get("profile_name")
if name is None:
return None
text = str(name).strip()
# Profile names are non-secret identifiers; still bound length.
return text[:80] if text else None
except Exception:
return None
def _is_framework_control_flow(exc: BaseException) -> bool:
"""True for exceptions the MCP framework must re-raise unchanged."""
try:
from mcp.shared.exceptions import UrlElicitationRequiredError
if isinstance(exc, UrlElicitationRequiredError):
return True
except Exception:
pass
# BaseException subclasses that must never become tool isError payloads.
if isinstance(exc, (KeyboardInterrupt, SystemExit, GeneratorExit)):
return True
return False
def is_known_client_failure(exc: BaseException) -> bool:
"""True only for explicit typed Gitea client / config failures.
Never true for arbitrary ``RuntimeError`` (reviewer finding #3).
"""
try:
import gitea_auth
if isinstance(
exc,
(
gitea_auth.GiteaAuthError,
gitea_auth.GiteaAuthzError,
gitea_auth.GiteaNetworkError,
gitea_auth.GiteaConfigError,
gitea_auth.GiteaHttpError,
),
):
return True
except Exception:
return False
try:
import gitea_config
if isinstance(exc, gitea_config.ConfigError):
return True
except Exception:
pass
return False
def classify_exception(exc: BaseException) -> dict[str, Any]:
"""Return structured classification with **fixed** messages only.
Only typed authentication / authorization / network / configuration
failures receive those labels. Unexpected programming failures are
``internal_error``. HTTP bodies and exception text are never copied
into ``message``.
"""
try:
return _classify_exception_impl(exc)
except Exception:
# Fail closed: sanitization / classification failure must not leak.
return {
"reason_code": REASON_INTERNAL_ERROR,
"error_class": ERROR_CLASS_INTERNAL,
"http_status": None,
"message": fixed_message(REASON_INTERNAL_ERROR),
"transport_survives": True,
}
def _classify_exception_impl(exc: BaseException) -> dict[str, Any]:
import gitea_auth
if isinstance(exc, gitea_auth.GiteaAuthError):
code = getattr(exc, "reason_code", None) or REASON_AUTH_INVALID_TOKEN
if code not in (
REASON_AUTH_FAILED,
REASON_AUTH_INVALID_TOKEN,
):
code = REASON_AUTH_INVALID_TOKEN
return {
"reason_code": code,
"error_class": ERROR_CLASS_AUTHENTICATION,
"http_status": getattr(exc, "http_status", None) or 401,
"message": fixed_message(code),
"transport_survives": True,
}
if isinstance(exc, gitea_auth.GiteaAuthzError):
code = getattr(exc, "reason_code", None) or REASON_AUTHZ_DENIED
if code not in (REASON_AUTHZ_DENIED, REASON_AUTHZ_INSUFFICIENT_SCOPE):
code = REASON_AUTHZ_DENIED
return {
"reason_code": code,
"error_class": ERROR_CLASS_AUTHORIZATION,
"http_status": getattr(exc, "http_status", None) or 403,
"message": fixed_message(code),
"transport_survives": True,
}
if isinstance(exc, gitea_auth.GiteaNetworkError):
return {
"reason_code": REASON_NETWORK_ERROR,
"error_class": ERROR_CLASS_NETWORK,
"http_status": getattr(exc, "http_status", None),
"message": fixed_message(REASON_NETWORK_ERROR),
"transport_survives": True,
}
if isinstance(exc, gitea_auth.GiteaConfigError):
return {
"reason_code": REASON_CONFIG_ERROR,
"error_class": ERROR_CLASS_CONFIGURATION,
"http_status": getattr(exc, "http_status", None),
"message": fixed_message(REASON_CONFIG_ERROR),
"transport_survives": True,
}
if isinstance(exc, gitea_auth.GiteaHttpError):
code = getattr(exc, "reason_code", None) or REASON_HTTP_ERROR
if code == REASON_UPSTREAM_UNAVAILABLE:
error_class = ERROR_CLASS_UPSTREAM
else:
error_class = ERROR_CLASS_INTERNAL
code = REASON_HTTP_ERROR
return {
"reason_code": code,
"error_class": error_class,
"http_status": getattr(exc, "http_status", None),
"message": fixed_message(code),
"transport_survives": True,
}
try:
import gitea_config
if isinstance(exc, gitea_config.ConfigError):
return {
"reason_code": REASON_CONFIG_ERROR,
"error_class": ERROR_CLASS_CONFIGURATION,
"http_status": None,
"message": fixed_message(REASON_CONFIG_ERROR),
"transport_survives": True,
}
except Exception:
pass
# Unwrap FastMCP ToolError cause when the original was a typed failure.
cause = getattr(exc, "__cause__", None)
if cause is not None and cause is not exc and is_known_client_failure(cause):
return _classify_exception_impl(cause)
# No message-substring authentication heuristics (reviewer finding #3).
return {
"reason_code": REASON_INTERNAL_ERROR,
"error_class": ERROR_CLASS_INTERNAL,
"http_status": None,
"message": fixed_message(REASON_INTERNAL_ERROR),
"transport_survives": True,
}
def build_structured_error_payload(
classification: dict[str, Any],
*,
tool_name: str | None = None,
profile_name: str | None = None,
) -> dict[str, Any]:
"""LLM-safe structured payload — fixed message + typed metadata only."""
try:
reason = str(classification.get("reason_code") or REASON_INTERNAL_ERROR)
message = fixed_message(reason)
# Refuse to emit any classification message that is not the fixed constant.
if classification.get("message") != message:
message = fixed_message(reason)
payload: dict[str, Any] = {
"success": False,
"isError": True,
"reason_code": reason if reason in FIXED_MESSAGES else REASON_INTERNAL_ERROR,
"error_class": classification.get("error_class") or ERROR_CLASS_INTERNAL,
"message": message,
"transport_survives": True,
"retryable": classification.get("error_class")
in {
ERROR_CLASS_AUTHENTICATION,
ERROR_CLASS_NETWORK,
ERROR_CLASS_CONFIGURATION,
},
}
status = classification.get("http_status")
if isinstance(status, int):
payload["http_status"] = status
if tool_name and isinstance(tool_name, str):
# Tool names are identifiers, not secrets; bound length.
payload["tool"] = tool_name[:120]
if profile_name and isinstance(profile_name, str):
payload["profile"] = profile_name[:80]
return payload
except Exception:
return {
"success": False,
"isError": True,
"reason_code": REASON_INTERNAL_ERROR,
"error_class": ERROR_CLASS_INTERNAL,
"message": fixed_message(REASON_INTERNAL_ERROR),
"transport_survives": True,
"retryable": False,
}
def log_sanitized_daemon_reason(
classification: dict[str, Any],
*,
tool_name: str | None = None,
stream=None,
) -> None:
"""Daemon log: reason codes only — never exception text or response bodies."""
stream = stream if stream is not None else sys.stderr
try:
reason = classification.get("reason_code") or REASON_INTERNAL_ERROR
error_class = classification.get("error_class") or ERROR_CLASS_INTERNAL
# Only emit known tokens; never classification['message'] from callers
# that might have been poisoned.
if reason not in FIXED_MESSAGES:
reason = REASON_INTERNAL_ERROR
error_class = ERROR_CLASS_INTERNAL
parts = [
"mcp_tool_error",
f"reason_code={reason}",
f"error_class={error_class}",
]
if tool_name and isinstance(tool_name, str):
parts.append(f"tool={tool_name[:120]}")
status = classification.get("http_status")
if isinstance(status, int):
parts.append(f"http_status={status}")
# Intentionally no detail= / message= field — secrets lived there.
line = " ".join(parts)
stream.write(line + "\n")
if hasattr(stream, "flush"):
stream.flush()
logger.warning(line)
except Exception:
# Fail closed: never fall back to logging the exception.
try:
stream.write(
"mcp_tool_error reason_code=internal_error "
"error_class=internal\n"
)
if hasattr(stream, "flush"):
stream.flush()
except Exception:
pass
def to_call_tool_result(
exc: BaseException,
*,
tool_name: str | None = None,
profile_name: str | None = None,
log: bool = True,
) -> Any:
"""Build a FastMCP ``CallToolResult`` with ``isError=True`` for *exc*."""
from mcp.types import CallToolResult, TextContent
try:
classification = classify_exception(exc)
if log:
log_sanitized_daemon_reason(classification, tool_name=tool_name)
payload = build_structured_error_payload(
classification, tool_name=tool_name, profile_name=profile_name
)
text = json.dumps(payload, indent=2, sort_keys=True)
return CallToolResult(
content=[TextContent(type="text", text=text)],
structuredContent=payload,
isError=True,
)
except Exception:
# Absolute fail-closed path — no exception text.
fallback = {
"success": False,
"isError": True,
"reason_code": REASON_INTERNAL_ERROR,
"error_class": ERROR_CLASS_INTERNAL,
"message": fixed_message(REASON_INTERNAL_ERROR),
"transport_survives": True,
"retryable": False,
}
return CallToolResult(
content=[
TextContent(
type="text",
text=json.dumps(fallback, indent=2, sort_keys=True),
)
],
structuredContent=fallback,
isError=True,
)
def install_tool_run_boundary(Tool) -> bool:
"""Install a **narrow** error boundary around FastMCP ``Tool.run``.
Strategy (preserves framework semantics):
* Call the **original** ``Tool.run`` for the success path (async, return
types, convert_result, protocol behavior unchanged).
* Re-raise ``UrlElicitationRequiredError`` and other control-flow
exceptions without mapping to ``internal_error``.
* Map typed Gitea client failures (and ToolError whose ``__cause__`` is
typed) to structured ``CallToolResult(isError=True)``.
* Map remaining unexpected tool failures to fixed ``internal_error``
isError results (transport survival) without secret-bearing text.
* Idempotent: second install is a no-op and returns ``False``.
"""
if getattr(Tool.run, _INSTALL_FLAG, False):
return False
from mcp.server.fastmcp.exceptions import ToolError
from mcp.shared.exceptions import UrlElicitationRequiredError
original_run = Tool.run
async def run_boundary(
self,
arguments: dict[str, Any],
context=None,
convert_result: bool = False,
) -> Any:
try:
return await original_run(
self,
arguments,
context=context,
convert_result=convert_result,
)
except UrlElicitationRequiredError:
# Framework control-flow — must not become internal_error.
raise
except BaseException as exc:
if _is_framework_control_flow(exc):
raise
if not isinstance(exc, Exception):
raise
# Prefer typed cause under FastMCP ToolError wrappers.
target: BaseException = exc
if isinstance(exc, ToolError) and exc.__cause__ is not None:
target = exc.__cause__
# Known client failures → structured isError with reason codes.
# Unexpected failures → fixed internal_error isError (no secrets).
profile_name = _safe_profile_name()
return to_call_tool_result(
target,
tool_name=getattr(self, "name", None),
profile_name=profile_name,
)
setattr(run_boundary, _INSTALL_FLAG, True)
setattr(run_boundary, _ORIGINAL_ATTR, original_run)
Tool.run = run_boundary # type: ignore[method-assign]
return True
def boundary_is_installed(Tool) -> bool:
"""True when the #699 boundary is active on *Tool.run*."""
return bool(getattr(Tool.run, _INSTALL_FLAG, False))
+1
View File
@@ -31,6 +31,7 @@ python-multipart==0.0.32
referencing==0.37.0 referencing==0.37.0
rich==15.0.0 rich==15.0.0
rpds-py==2026.5.1 rpds-py==2026.5.1
sentry-sdk==2.20.0
shellingham==1.5.4 shellingham==1.5.4
sse-starlette==3.4.5 sse-starlette==3.4.5
starlette==1.3.1 starlette==1.3.1
+43 -657
View File
@@ -524,26 +524,13 @@ def assess_lease_inventory(
"reclaimable_review_leases": reclaimable, "reclaimable_review_leases": reclaimable,
} }
# Canonical next-action vocabulary for reviewer lease handoff (#599, #691). # Canonical next-action vocabulary for reviewer lease handoff (#599).
NEXT_ACTION_ACQUIRE = "acquire" NEXT_ACTION_ACQUIRE = "acquire"
NEXT_ACTION_WAIT = "wait" NEXT_ACTION_WAIT = "wait"
NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session" NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease" NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup" NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding" NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding"
NEXT_ACTION_CLEANUP_OBSOLETE_LEASE = "cleanup_obsolete_reviewer_comment_lease"
CLEANUP_OBSOLETE_LEASE_TOOL = "gitea_cleanup_obsolete_reviewer_comment_lease"
CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX = "CLEANUP OBSOLETE REVIEWER LEASE "
# Formal terminal review verdicts that complete a leased-head workflow (#691).
_TERMINAL_REVIEW_VERDICTS = frozenset({
"APPROVED",
"REQUEST_CHANGES",
"approved",
"request_changes",
"REQUESTCHANGES",
})
_HANDOFF_CLASSIFICATIONS = frozenset({ _HANDOFF_CLASSIFICATIONS = frozenset({
"no_lease", "no_lease",
@@ -552,12 +539,6 @@ _HANDOFF_CLASSIFICATIONS = frozenset({
"foreign_active", "foreign_active",
"foreign_reclaimable", "foreign_reclaimable",
"foreign_expired", "foreign_expired",
"foreign_active_current_head",
"foreign_expired_current_head",
"foreign_completed_superseded_head",
"foreign_expired_superseded_head",
"orphaned_owner_missing",
"ambiguous_conflicting_evidence",
"instructed_lease_missing_with_replacement", "instructed_lease_missing_with_replacement",
"worktree_binding_mismatch", "worktree_binding_mismatch",
}) })
@@ -570,476 +551,6 @@ def _norm_path(value: str | None) -> str:
return os.path.normpath(text.rstrip("/")) return os.path.normpath(text.rstrip("/"))
def _normalize_verdict(value: str | None) -> str:
text = (value or "").strip().upper().replace("-", "_").replace(" ", "_")
if text in {"REQUESTCHANGES", "REQUEST_CHANGES", "CHANGES_REQUESTED"}:
return "REQUEST_CHANGES"
if text in {"APPROVED", "APPROVE"}:
return "APPROVED"
return text
def formal_terminal_review_for_head(
formal_reviews: list[dict] | None,
*,
leased_head: str | None,
) -> dict[str, Any] | None:
"""Return the latest undismissed terminal formal review for *leased_head*."""
head = _normalize_sha(leased_head)
if not head:
return None
matches: list[dict[str, Any]] = []
for review in formal_reviews or []:
if review.get("dismissed"):
continue
verdict = _normalize_verdict(
review.get("verdict") or review.get("state") or review.get("body_state")
)
if verdict not in {"APPROVED", "REQUEST_CHANGES"}:
continue
rhead = _normalize_sha(
review.get("reviewed_head_sha")
or review.get("commit_id")
or review.get("head_sha")
)
if rhead != head:
continue
matches.append({**review, "verdict": verdict, "reviewed_head_sha": rhead})
if not matches:
return None
return matches[-1]
def find_newest_nonterminal_lease(
comments: list[dict],
*,
pr_number: int,
now: datetime | None = None,
include_expired: bool = True,
) -> dict[str, Any] | None:
"""Newest non-terminal lease marker, optionally including expired ones (#691).
Unlike ``find_active_reviewer_lease``, this retains expired non-terminal
markers so diagnosis/cleanup can still name the obsolete lease after
``expires_at`` (comment-backed ledger; no control-plane ``lease_id``).
"""
now = now or datetime.now(timezone.utc)
entries = list(reversed(_lease_entries(comments, pr_number=pr_number)))
for entry in entries:
phase = (entry.get("phase") or "").strip().lower()
if phase in _TERMINAL_PHASES:
# Newest terminal ends the ledger chain for active acquisition, but
# an older non-terminal is not "active". Stop at newest marker.
return None
expired = _lease_expired(entry, now=now)
if expired and not include_expired:
return None
lease = dict(entry)
lease["freshness"] = classify_lease_freshness(lease, now=now)
lease["expired"] = expired
return lease
return None
def cleanup_confirmation_for_pr(pr_number: int) -> str:
return f"{CLEANUP_OBSOLETE_LEASE_CONFIRMATION_PREFIX}{int(pr_number)}"
def assess_obsolete_reviewer_comment_lease_cleanup(
comments: list[dict],
*,
pr_number: int,
current_head_sha: str | None,
formal_reviews: list[dict] | None = None,
repo: str | None = None,
expected_repo: str | None = None,
requesting_session_id: str | None = None,
controller_recovery_authorized: bool = False,
worktree_exists: bool | None = None,
worktree_clean: bool | None = None,
worktree_has_unpreserved_work: bool | None = None,
owner_process_alive: bool | None = None,
owner_pid_observed: int | None = None,
requesting_pid: int | None = None,
current_head_review_in_progress: bool = False,
expected_lease_comment_id: int | None = None,
expected_session_id: str | None = None,
expected_leased_head: str | None = None,
confirmation: str | None = None,
apply: bool = False,
now: datetime | None = None,
) -> dict[str, Any]:
"""Guarded cleanup eligibility for obsolete comment-backed reviewer leases (#691).
Cleanup is never transfer/adoption/repoint and never uses PID equality as
ownership. Eligible only when evidence proves the lease is past expiry and/or
pinned to a superseded head with a completed formal terminal review, and
every safety boundary holds.
"""
now = now or datetime.now(timezone.utc)
reasons: list[str] = []
fail_closed_reasons: list[str] = []
current_head = _normalize_sha(current_head_sha)
req_session = (requesting_session_id or "").strip()
lease = find_newest_nonterminal_lease(
comments, pr_number=pr_number, now=now, include_expired=True
)
if not lease:
# Fall back to active finder (unexpired only) for report consistency.
lease = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
classification = "no_lease"
cleanup_allowed = False
release_body: str | None = None
audit_comment_body: str | None = None
terminal_review = None
leased_head = None
head_superseded = False
past_expiry = False
expires_parseable = True
if not lease:
fail_closed_reasons.append(
f"PR #{pr_number}: no non-terminal comment-backed reviewer lease to clean"
)
classification = "no_lease"
else:
leased_head = _normalize_sha(lease.get("candidate_head"))
expires_raw = lease.get("expires_at")
expires_at = _parse_timestamp(expires_raw)
if expires_raw and expires_at is None:
expires_parseable = False
fail_closed_reasons.append(
"lease expires_at cannot be parsed or trusted (fail closed)"
)
past_expiry = bool(expires_at and expires_at <= now)
freshness = lease.get("freshness") or classify_lease_freshness(lease, now=now)
owner_session = (lease.get("session_id") or "").strip()
is_requesting_owner = bool(
req_session and owner_session and req_session == owner_session
)
# Identity pins (repo / PR / session / head / comment) must match when
# the caller supplies expected evidence.
lease_repo = (lease.get("repo") or "").strip()
exp_repo = (expected_repo or repo or "").strip()
if exp_repo and lease_repo and exp_repo != lease_repo:
fail_closed_reasons.append(
f"repository identity mismatch: lease repo={lease_repo!r} "
f"expected={exp_repo!r}"
)
if lease.get("pr_number") not in (None, pr_number):
fail_closed_reasons.append(
f"PR identity mismatch: lease pr={lease.get('pr_number')} "
f"requested={pr_number}"
)
if expected_lease_comment_id is not None:
cid = lease.get("comment_id")
if cid is None or int(cid) != int(expected_lease_comment_id):
fail_closed_reasons.append(
"lease comment_id does not match expected evidence "
f"(lease={cid}, expected={expected_lease_comment_id})"
)
if expected_session_id:
if owner_session != expected_session_id.strip():
fail_closed_reasons.append(
"lease session_id does not match expected evidence "
f"(lease={owner_session}, expected={expected_session_id})"
)
if expected_leased_head:
exp_head = _normalize_sha(expected_leased_head)
if not exp_head or exp_head != leased_head:
fail_closed_reasons.append(
"leased head does not match expected evidence "
f"(lease={leased_head}, expected={exp_head})"
)
# PID equality is never ownership proof (#691).
if (
owner_pid_observed is not None
and requesting_pid is not None
and int(owner_pid_observed) == int(requesting_pid)
):
reasons.append(
"observed PID equals requesting PID but PID equality is NOT "
"ownership proof; ignored for authorization"
)
if is_requesting_owner:
fail_closed_reasons.append(
"requesting session owns the lease; use "
"gitea_release_reviewer_pr_lease (owner path), not non-owner cleanup"
)
if current_head_review_in_progress:
fail_closed_reasons.append(
"a current-head review submission is in progress; cleanup denied"
)
if not current_head:
fail_closed_reasons.append(
"current PR head SHA missing or unparseable (fail closed)"
)
if not leased_head:
fail_closed_reasons.append(
"lease candidate_head missing or unparseable (fail closed)"
)
head_superseded = bool(
current_head and leased_head and current_head != leased_head
)
head_matches_current = bool(
current_head and leased_head and current_head == leased_head
)
terminal_review = formal_terminal_review_for_head(
formal_reviews, leased_head=leased_head
)
has_terminal = terminal_review is not None
# Worktree safety.
if worktree_has_unpreserved_work is True or worktree_clean is False:
fail_closed_reasons.append(
"lease worktree is dirty or has unpreserved work; cleanup denied"
)
if (
worktree_exists is True
and worktree_clean is None
and worktree_has_unpreserved_work is None
):
# Unknown cleanliness when path exists — fail closed.
fail_closed_reasons.append(
"lease worktree exists but cleanliness is unknown (fail closed)"
)
# Classification matrix (#691).
if head_matches_current and freshness in {"active", "stale_warning"}:
classification = "foreign_active_current_head"
fail_closed_reasons.append(
"genuinely active foreign lease on the current PR head; "
"cleanup denied (fail closed)"
)
elif head_matches_current and past_expiry:
classification = "foreign_expired_current_head"
# Expired on current head: owner-missing / orphan path may apply.
if owner_process_alive is True:
fail_closed_reasons.append(
"lease expired on current head but owner process still alive "
"with plausible activity; cleanup denied"
)
elif worktree_clean is False:
fail_closed_reasons.append(
"owner process absent but worktree dirty; cleanup denied"
)
elif owner_process_alive is False and (
worktree_clean is True or worktree_exists is False
):
if controller_recovery_authorized:
classification = "orphaned_owner_missing"
else:
fail_closed_reasons.append(
"controller/recovery capability not authorized for orphan cleanup"
)
else:
fail_closed_reasons.append(
"insufficient orphan evidence for expired current-head lease "
"(need owner_process_alive=false and clean/absent worktree)"
)
elif head_superseded and has_terminal and past_expiry:
classification = "foreign_expired_superseded_head"
elif head_superseded and has_terminal and not past_expiry:
classification = "foreign_completed_superseded_head"
elif head_superseded and not has_terminal:
classification = "ambiguous_conflicting_evidence"
fail_closed_reasons.append(
"lease head is superseded but no formal terminal review exists "
"for the leased head (fail closed)"
)
elif not head_superseded and not past_expiry and freshness in {
"active", "stale_warning"
}:
classification = "foreign_active_current_head"
fail_closed_reasons.append(
"foreign lease still active on current head; wait or use owner release"
)
elif past_expiry and not head_superseded:
classification = "foreign_expired_current_head"
else:
classification = "ambiguous_conflicting_evidence"
fail_closed_reasons.append(
"lease evidence is ambiguous; cleanup denied (fail closed)"
)
# Recent owner progress on a non-superseded active lease blocks cleanup.
minutes = _minutes_since_activity(lease, now=now)
if (
head_matches_current
and freshness == "active"
and minutes is not None
and minutes < STALE_WARNING_MINUTES
):
fail_closed_reasons.append(
"owner session has recent authenticated progress on current head; "
"cleanup denied"
)
# Authority + confirmation for apply.
if not controller_recovery_authorized:
if classification in {
"foreign_completed_superseded_head",
"foreign_expired_superseded_head",
"foreign_expired_current_head",
"orphaned_owner_missing",
}:
fail_closed_reasons.append(
"controller/recovery capability required "
"(controller_recovery_authorized=true)"
)
expected_conf = cleanup_confirmation_for_pr(pr_number)
if apply:
if (confirmation or "").strip() != expected_conf:
fail_closed_reasons.append(
f"confirmation must equal exactly {expected_conf!r}"
)
# Superseded + terminal path does not require past_expiry (AC1).
eligible_class = classification in {
"foreign_completed_superseded_head",
"foreign_expired_superseded_head",
"orphaned_owner_missing",
"foreign_expired_current_head",
}
# foreign_expired_current_head needs orphan/clean worktree + authority.
if classification == "foreign_expired_current_head":
if worktree_clean is not True and worktree_exists is not False:
if "worktree" not in " ".join(fail_closed_reasons):
fail_closed_reasons.append(
"expired current-head cleanup requires proven-clean "
"worktree or absent worktree"
)
if owner_process_alive is True:
fail_closed_reasons.append(
"owner process still alive on expired current-head lease"
)
cleanup_allowed = (
eligible_class
and expires_parseable
and not fail_closed_reasons
and not is_requesting_owner
)
if cleanup_allowed:
release_body = format_lease_body(
repo=lease.get("repo") or exp_repo or "",
pr_number=pr_number,
issue_number=lease.get("issue_number"),
reviewer_identity=lease.get("reviewer_identity") or "",
profile=lease.get("profile") or "unknown",
session_id=owner_session or "unknown",
worktree=lease.get("worktree") or "",
phase="released",
candidate_head=leased_head,
target_branch=lease.get("target_branch") or "master",
target_branch_sha=lease.get("target_branch_sha"),
last_activity=now,
blocker="obsolete-superseded-or-expired-lease",
)
audit_comment_body = (
"## Canonical obsolete reviewer lease cleanup (#691)\n\n"
f"- pr: #{pr_number}\n"
f"- lease_comment_id: {lease.get('comment_id')}\n"
f"- session_id: {owner_session}\n"
f"- leased_head: {leased_head}\n"
f"- current_head: {current_head}\n"
f"- expires_at: {lease.get('expires_at')}\n"
f"- classification: {classification}\n"
f"- terminal_review_verdict: "
f"{(terminal_review or {}).get('verdict')}\n"
f"- tool: {CLEANUP_OBSOLETE_LEASE_TOOL}\n"
"- action: posted terminal phase=released lease marker; "
"did not transfer validation, decision, or workflow proof; "
"did not repoint lease to the new head; did not delete history\n"
)
reasons.append(
f"cleanup eligible ({classification}); post terminal released "
"marker via sanctioned tool"
)
else:
reasons.extend(fail_closed_reasons)
return {
"pr_number": pr_number,
"classification": classification,
"blocker_kind": classification if not cleanup_allowed else "none",
"cleanup_allowed": cleanup_allowed,
"mutation_allowed": False, # never grants review mutation
"mutation_eligibility": "prohibited" if not cleanup_allowed else "cleanup_only",
"exact_next_action": (
NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
if cleanup_allowed
else (
NEXT_ACTION_WAIT
if classification
in {
"foreign_active_current_head",
"ambiguous_conflicting_evidence",
}
else NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP
)
),
"cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
"required_confirmation": cleanup_confirmation_for_pr(pr_number),
"leased_head": leased_head,
"current_head": current_head,
"head_superseded": head_superseded,
"past_expiry": past_expiry,
"expires_at": (lease or {}).get("expires_at") if lease else None,
"expires_parseable": expires_parseable,
"terminal_review": terminal_review,
"terminal_review_present": terminal_review is not None,
"worktree_state": {
"path": (lease or {}).get("worktree") if lease else None,
"exists": worktree_exists,
"clean": worktree_clean,
"has_unpreserved_work": worktree_has_unpreserved_work,
},
"owner_session_evidence": {
"session_id": (lease or {}).get("session_id") if lease else None,
"reviewer_identity": (
(lease or {}).get("reviewer_identity") if lease else None
),
"process_alive": owner_process_alive,
"pid_observed": owner_pid_observed,
"pid_is_not_ownership_proof": True,
"requesting_session_is_owner": bool(
lease
and req_session
and (lease.get("session_id") or "").strip() == req_session
),
},
"active_lease": lease,
"release_body": release_body,
"audit_comment_body": audit_comment_body,
"controller_recovery_authorized": controller_recovery_authorized,
"reasons": reasons if reasons else fail_closed_reasons,
"fail_closed_reasons": fail_closed_reasons,
"forbidden": [
"manual comment deletion",
"database edits",
"mtime manipulation",
"PID-based ownership",
"direct session-state seeding",
"lease stealing or adoption",
"repointing old lease to new head",
"transfer of validation or decision state",
"worktree reuse by replacement reviewer",
],
}
def diagnose_reviewer_pr_lease_handoff( def diagnose_reviewer_pr_lease_handoff(
comments: list[dict], comments: list[dict],
*, *,
@@ -1050,50 +561,40 @@ def diagnose_reviewer_pr_lease_handoff(
env_bound_worktree: str | None = None, env_bound_worktree: str | None = None,
instructed_session_id: str | None = None, instructed_session_id: str | None = None,
instructed_comment_id: int | None = None, instructed_comment_id: int | None = None,
current_head_sha: str | None = None,
formal_reviews: list[dict] | None = None,
worktree_exists: bool | None = None,
worktree_clean: bool | None = None,
owner_process_alive: bool | None = None,
now: datetime | None = None, now: datetime | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Classify open-PR reviewer lease handoff and emit a canonical next action (#599, #691). """Classify open-PR reviewer lease handoff and emit a canonical next action (#599).
Read-only diagnosis. Never steals, releases, or adopts a foreign lease. Read-only diagnosis. Never steals, releases, or adopts a foreign lease.
Fail-closed acquisition rules for active foreign leases remain intact. Fail-closed acquisition rules for active foreign leases remain intact.
Returns a structured diagnosis with: Returns a structured diagnosis with:
- classification (including #691 superseded/expired distinctions) - classification
- next_action (including cleanup_obsolete_reviewer_comment_lease) - next_action (one of wait / resume_exact_owner_session /
release_expired_lease / operator_authorized_cleanup /
repair_worktree_binding / acquire)
- active_lease identity fields when present - active_lease identity fields when present
- worktree_binding match result - worktree_binding match result
- instructed-lease mismatch flags - instructed-lease mismatch flags
- cleanup tool/confirmation when cleanup-eligible
""" """
now = now or datetime.now(timezone.utc) now = now or datetime.now(timezone.utc)
session_id = (current_session_id or "").strip() session_id = (current_session_id or "").strip()
identity = (current_reviewer_identity or "").strip() identity = (current_reviewer_identity or "").strip()
instructed_sid = (instructed_session_id or "").strip() instructed_sid = (instructed_session_id or "").strip()
reasons: list[str] = [] reasons: list[str] = []
current_head = _normalize_sha(current_head_sha)
active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now) active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
# Also surface expired non-terminal newest marker for #691 diagnosis.
newest_nt = find_newest_nonterminal_lease(
comments, pr_number=pr_number, now=now, include_expired=True
)
lease_for_class = active or newest_nt
session_lease = get_session_lease() session_lease = get_session_lease()
# Worktree binding: env-bound vs proposed vs active lease worktree. # Worktree binding: env-bound vs proposed vs active lease worktree.
env_wt = _norm_path(env_bound_worktree) env_wt = _norm_path(env_bound_worktree)
prop_wt = _norm_path(proposed_worktree) prop_wt = _norm_path(proposed_worktree)
lease_wt = _norm_path((lease_for_class or {}).get("worktree") if lease_for_class else None) lease_wt = _norm_path((active or {}).get("worktree") if active else None)
binding_mismatch = False binding_mismatch = False
binding_details: dict[str, Any] = { binding_details: dict[str, Any] = {
"env_bound_worktree": env_bound_worktree or None, "env_bound_worktree": env_bound_worktree or None,
"proposed_worktree": proposed_worktree or None, "proposed_worktree": proposed_worktree or None,
"lease_worktree": (lease_for_class or {}).get("worktree") if lease_for_class else None, "lease_worktree": (active or {}).get("worktree") if active else None,
"match": True, "match": True,
} }
paths = [p for p in (env_wt, prop_wt, lease_wt) if p] paths = [p for p in (env_wt, prop_wt, lease_wt) if p]
@@ -1107,13 +608,13 @@ def diagnose_reviewer_pr_lease_handoff(
# Instructed lease gone while a different lease is active (PR #592-style). # Instructed lease gone while a different lease is active (PR #592-style).
instructed_missing_with_replacement = False instructed_missing_with_replacement = False
if instructed_sid or instructed_comment_id is not None: if instructed_sid or instructed_comment_id is not None:
if not lease_for_class: if not active:
reasons.append( reasons.append(
"instructed lease is gone and no active replacement lease remains" "instructed lease is gone and no active replacement lease remains"
) )
else: else:
owner = (lease_for_class.get("session_id") or "").strip() owner = (active.get("session_id") or "").strip()
cid = lease_for_class.get("comment_id") cid = active.get("comment_id")
sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid) sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
cid_mismatch = ( cid_mismatch = (
instructed_comment_id is not None instructed_comment_id is not None
@@ -1130,94 +631,35 @@ def diagnose_reviewer_pr_lease_handoff(
# Classification + next_action. # Classification + next_action.
classification = "no_lease" classification = "no_lease"
next_action = NEXT_ACTION_ACQUIRE next_action = NEXT_ACTION_ACQUIRE
cleanup_hint: dict[str, Any] | None = None
if lease_for_class: if active:
owner = (lease_for_class.get("session_id") or "").strip() owner = (active.get("session_id") or "").strip()
freshness = lease_for_class.get("freshness") or classify_lease_freshness( freshness = active.get("freshness") or classify_lease_freshness(
lease_for_class, now=now active, now=now
) )
owner_identity = (lease_for_class.get("reviewer_identity") or "").strip() owner_identity = (active.get("reviewer_identity") or "").strip()
is_own = bool(session_id and owner and owner == session_id) is_own = bool(session_id and owner and owner == session_id)
# Same identity alone is NOT ownership for resume; session_id must match. # Same identity alone is NOT ownership for resume; session_id must match.
same_identity = bool( same_identity = bool(
identity and owner_identity and identity == owner_identity identity and owner_identity and identity == owner_identity
) )
leased_head = _normalize_sha(lease_for_class.get("candidate_head"))
head_superseded = bool(
current_head and leased_head and current_head != leased_head
)
head_current = bool(
current_head and leased_head and current_head == leased_head
)
past_expiry = bool(lease_for_class.get("expired")) or freshness == "expired"
if not past_expiry:
past_expiry = _lease_expired(lease_for_class, now=now)
terminal = formal_terminal_review_for_head(
formal_reviews, leased_head=leased_head
)
if is_own and freshness in {"active", "stale_warning"} and not past_expiry: if is_own and freshness in {"active", "stale_warning"}:
classification = "own_active" classification = "own_active"
next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
elif is_own and (freshness in {"reclaimable", "expired"} or past_expiry): elif is_own and freshness in {"reclaimable", "expired"}:
classification = "own_expired" classification = "own_expired"
next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
reasons.append( reasons.append(
f"own lease freshness is '{freshness}'; release via " f"own lease freshness is '{freshness}'; release via "
"gitea_release_reviewer_pr_lease then re-acquire" "gitea_release_reviewer_pr_lease then re-acquire"
) )
elif not is_own and head_superseded and terminal and past_expiry: elif not is_own and freshness in {"active", "stale_warning"}:
classification = "foreign_expired_superseded_head" classification = "foreign_active"
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign lease expired and pinned to superseded head with "
"formal terminal review; use guarded obsolete-lease cleanup"
)
elif not is_own and head_superseded and terminal and not past_expiry:
classification = "foreign_completed_superseded_head"
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign lease pinned to superseded head with completed formal "
"review; use guarded obsolete-lease cleanup (do not wait indefinitely)"
)
elif not is_own and head_superseded and not terminal:
classification = "ambiguous_conflicting_evidence"
next_action = NEXT_ACTION_WAIT
reasons.append(
"lease head superseded but no formal terminal review for leased "
"head; fail closed / wait (no indefinite steal)"
)
elif not is_own and head_current and past_expiry:
classification = "foreign_expired_current_head"
if owner_process_alive is False and worktree_clean is True:
classification = "orphaned_owner_missing"
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign expired lease on current head; owner process absent "
"and worktree clean — guarded orphan cleanup eligible"
)
elif owner_process_alive is False and worktree_clean is False:
classification = "ambiguous_conflicting_evidence"
next_action = NEXT_ACTION_WAIT
reasons.append(
"owner process absent but worktree dirty; cleanup denied"
)
else:
next_action = NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
reasons.append(
"foreign expired lease on current head; use guarded cleanup "
"when worktree/process evidence permits"
)
elif not is_own and freshness in {"active", "stale_warning"} and not past_expiry:
classification = (
"foreign_active_current_head" if head_current or not current_head
else "foreign_active"
)
next_action = NEXT_ACTION_WAIT next_action = NEXT_ACTION_WAIT
reasons.append( reasons.append(
f"foreign active reviewer lease (session_id={owner}, " f"foreign active reviewer lease (session_id={owner}, "
f"phase={lease_for_class.get('phase')}, freshness={freshness}); " f"phase={active.get('phase')}, freshness={freshness}); "
"do not submit; do not steal" "do not submit; do not steal"
) )
if same_identity: if same_identity:
@@ -1232,12 +674,11 @@ def diagnose_reviewer_pr_lease_handoff(
f"foreign reclaimable lease (session_id={owner}); clear only via " f"foreign reclaimable lease (session_id={owner}); clear only via "
"sanctioned gitea_release_reviewer_pr_lease when reclaimable" "sanctioned gitea_release_reviewer_pr_lease when reclaimable"
) )
elif not is_own and (freshness == "expired" or past_expiry): elif not is_own and freshness == "expired":
classification = "foreign_expired" classification = "foreign_expired"
next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
reasons.append( reasons.append(
f"foreign expired lease (session_id={owner}); use sanctioned release " f"foreign expired lease (session_id={owner}); use sanctioned release"
f"or {CLEANUP_OBSOLETE_LEASE_TOOL}"
) )
else: else:
classification = "foreign_active" classification = "foreign_active"
@@ -1250,26 +691,13 @@ def diagnose_reviewer_pr_lease_handoff(
if instructed_missing_with_replacement and classification.startswith( if instructed_missing_with_replacement and classification.startswith(
"foreign" "foreign"
): ):
# Preserve #691 cleanup path when superseded/expired; otherwise classification = "instructed_lease_missing_with_replacement"
# keep the PR #592 instructed-missing label. # Foreign active still means wait; reclaimable still release.
if next_action != NEXT_ACTION_CLEANUP_OBSOLETE_LEASE: if next_action == NEXT_ACTION_WAIT:
classification = "instructed_lease_missing_with_replacement" reasons.append(
if next_action == NEXT_ACTION_WAIT: "replacement foreign lease is active — wait; "
reasons.append( "operator_authorized_cleanup only with explicit operator authority"
"replacement foreign lease is active — wait; " )
"operator_authorized_cleanup only with explicit operator authority"
)
if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
cleanup_hint = {
"cleanup_tool": CLEANUP_OBSOLETE_LEASE_TOOL,
"required_confirmation": cleanup_confirmation_for_pr(pr_number),
"controller_recovery_authorized_required": True,
"leased_head": leased_head,
"current_head": current_head,
"expires_at": lease_for_class.get("expires_at"),
"terminal_review_verdict": (terminal or {}).get("verdict"),
}
else: else:
classification = "no_lease" classification = "no_lease"
next_action = NEXT_ACTION_ACQUIRE next_action = NEXT_ACTION_ACQUIRE
@@ -1292,55 +720,29 @@ def diagnose_reviewer_pr_lease_handoff(
) )
lease_summary = None lease_summary = None
if lease_for_class: if active:
lease_summary = { lease_summary = {
"comment_id": lease_for_class.get("comment_id"), "comment_id": active.get("comment_id"),
"session_id": lease_for_class.get("session_id"), "session_id": active.get("session_id"),
"phase": lease_for_class.get("phase"), "phase": active.get("phase"),
"candidate_head": lease_for_class.get("candidate_head"), "candidate_head": active.get("candidate_head"),
"expires_at": lease_for_class.get("expires_at"), "expires_at": active.get("expires_at"),
"last_activity": lease_for_class.get("last_activity"), "last_activity": active.get("last_activity"),
"freshness": lease_for_class.get("freshness") "freshness": active.get("freshness")
or classify_lease_freshness(lease_for_class, now=now), or classify_lease_freshness(active, now=now),
"reviewer_identity": lease_for_class.get("reviewer_identity"), "reviewer_identity": active.get("reviewer_identity"),
"profile": lease_for_class.get("profile"), "profile": active.get("profile"),
"worktree": lease_for_class.get("worktree"), "worktree": active.get("worktree"),
"blocker": lease_for_class.get("blocker"), "blocker": active.get("blocker"),
} }
return { return {
"pr_number": pr_number, "pr_number": pr_number,
"classification": classification, "classification": classification,
"blocker_kind": classification,
"next_action": next_action, "next_action": next_action,
"exact_next_action": next_action,
"active_lease": lease_summary, "active_lease": lease_summary,
"session_lease": session_lease, "session_lease": session_lease,
"worktree_binding": binding_details, "worktree_binding": binding_details,
"leased_head": (lease_summary or {}).get("candidate_head"),
"current_head": current_head,
"expires_at": (lease_summary or {}).get("expires_at"),
"terminal_review_state": (
formal_terminal_review_for_head(
formal_reviews,
leased_head=(lease_summary or {}).get("candidate_head"),
)
if lease_summary
else None
),
"worktree_state": {
"exists": worktree_exists,
"clean": worktree_clean,
"path": (lease_summary or {}).get("worktree"),
},
"owner_session_evidence": {
"session_id": (lease_summary or {}).get("session_id"),
"process_alive": owner_process_alive,
"pid_is_not_ownership_proof": True,
},
"cleanup": cleanup_hint,
"cleanup_tool": (cleanup_hint or {}).get("cleanup_tool"),
"required_confirmation": (cleanup_hint or {}).get("required_confirmation"),
"instructed_session_id": instructed_session_id, "instructed_session_id": instructed_session_id,
"instructed_comment_id": instructed_comment_id, "instructed_comment_id": instructed_comment_id,
"instructed_lease_missing_with_replacement": instructed_missing_with_replacement, "instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
@@ -1349,19 +751,6 @@ def diagnose_reviewer_pr_lease_handoff(
and not binding_mismatch and not binding_mismatch
and bool(session_lease) and bool(session_lease)
), ),
"mutation_eligibility": (
"allowed"
if (
next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
and not binding_mismatch
and bool(session_lease)
)
else (
"cleanup_only"
if next_action == NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
else "prohibited"
)
),
"reasons": reasons, "reasons": reasons,
"forbidden": [ "forbidden": [
"manual lock deletion", "manual lock deletion",
@@ -1369,8 +758,5 @@ def diagnose_reviewer_pr_lease_handoff(
"mtime manipulation", "mtime manipulation",
"direct _SESSION_LEASE seeding", "direct _SESSION_LEASE seeding",
"silent foreign lease steal", "silent foreign lease steal",
"PID-based ownership",
"repointing old lease to new head",
"transfer of validation or decision state",
], ],
} }
+535
View File
@@ -0,0 +1,535 @@
"""Optional self-hosted Sentry observability for the Gitea MCP server (#606).
Adds env-var-gated Sentry SDK instrumentation so runtime errors, fail-closed
workflow blockers, lease/terminal-lock/stale-runtime collisions, and recurring
watchdog check-ins are visible in a *self-hosted* Sentry at
``https://sentry.prgs.cc/`` — never Sentry Cloud, and never as the workflow
source of truth (Gitea stays canonical).
Design constraints (mirror ``gitea_audit`` and the #612 incident bridge):
- **Off by default.** With ``MCP_SENTRY_ENABLED`` false/unset *or* ``SENTRY_DSN``
empty, ``init_sentry`` is a no-op and no events are ever sent — existing tool
behaviour and API-call patterns are unchanged (acceptance criterion 1).
- **Fail *open* for observability.** A Sentry outage, a missing ``sentry_sdk``
package, or any capture error must never break an MCP tool success path. Every
public entry point swallows its own exceptions.
- **Fail *closed* for redaction.** If a field cannot be proven safe it is dropped
rather than sent. Tokens, passwords, keychain IDs, DSNs, private config, raw
session-state, full prompt bodies, and full filesystem paths never leave here.
- **No hard dependency.** ``sentry_sdk`` is imported lazily; the module is fully
importable and testable without it installed.
Sentry is observe-only: it must not approve, merge, close, or otherwise mutate
Gitea workflow state, nor bypass leases, #332, or MCP gates. Alerts may only feed
the sanctioned Gitea issue/comment path via the #612 incident bridge.
"""
from __future__ import annotations
import hashlib
import os
import re
from dataclasses import dataclass
from typing import Any
# Reuse the most comprehensive existing scrubber so redaction stays consistent
# with the #612 incident bridge (tokens, DSNs, cookies, bearer/basic, keychain
# ids, session ids, user:pass@host).
from incident_bridge import redact_text as _redact_text
# Second, complementary scrubber: catches bare ``token <value>`` /
# ``Bearer <value>`` / ``Basic <value>`` prefixes and raw URLs that the
# incident-bridge delimiter patterns miss.
from gitea_audit import _redact_str as _redact_prefixes
# ── Optional SDK (lazy, never a hard dependency) ────────────────────────────
try: # pragma: no cover - trivial import guard
import sentry_sdk # type: ignore
except Exception: # pragma: no cover - absence is a supported state
sentry_sdk = None # type: ignore
# ── Env var names (single source of truth) ──────────────────────────────────
ENV_ENABLED = "MCP_SENTRY_ENABLED"
ENV_DSN = "SENTRY_DSN"
ENV_ENVIRONMENT = "SENTRY_ENVIRONMENT"
ENV_RELEASE = "SENTRY_RELEASE"
ENV_TRACES_SAMPLE_RATE = "MCP_SENTRY_TRACES_SAMPLE_RATE"
ENV_ENABLE_LOGS = "MCP_SENTRY_ENABLE_LOGS"
_TRUTHY = frozenset({"1", "true", "yes", "on"})
REDACTED = "[REDACTED]"
REDACTED_PATH = "[REDACTED_PATH]"
# ── Cron / watchdog monitor slugs (acceptance criterion 6) ──────────────────
# Stable slugs for the recurring/watchdog jobs #606 wants check-ins for. The
# slug is the durable monitor identity in Sentry; the wiring call sites pass one
# of these keys (or an explicit slug) to ``monitor_checkin``.
MONITOR_SLUGS: dict[str, str] = {
"stale_lease_scan": "gitea-mcp-stale-lease-scan",
"terminal_lock_scan": "gitea-mcp-terminal-lock-scan",
"allocator_health": "gitea-mcp-allocator-health",
"namespace_health": "gitea-mcp-namespace-health",
"dashboard_freshness": "gitea-mcp-dashboard-freshness",
"reconciler_cleanup": "gitea-mcp-reconciler-cleanup",
}
_CHECKIN_STATUSES = frozenset({"in_progress", "ok", "error"})
# ── Tag allowlist (issue "Suggested Sentry tags/context") ───────────────────
# Only these keys are ever attached as Sentry tags. Anything else is dropped so
# a caller cannot accidentally leak a sensitive value through a tag.
ALLOWED_TAG_KEYS = frozenset({
"role",
"profile",
"namespace",
"repo",
"org",
"issue_number",
"pr_number",
"blocker_type",
"workflow_hash",
"session_id_hash", # hash only — never the raw session id
"pid",
"worktree_category", # category, never the full sensitive path
"lease_comment_id",
"expected_head_sha",
"current_head_sha",
"terminal_lock_state",
"capability",
"mutation_tool",
})
# Absolute-path shapes that must never be sent verbatim (macOS/Linux + temp).
_PATH_RE = re.compile(r"(?:/private)?/(?:Users|home|tmp|var|opt|Volumes)/[^\s\"']*")
# ``extra`` keys whose *full* contents are forbidden by the redaction rules
# (raw session-state, full prompt/comment bodies, private config blobs, raw
# headers).
_FORBIDDEN_EXTRA_KEYS = frozenset({
"prompt",
"prompt_body",
"next_prompt",
"body",
"raw_body",
"session_state",
"session_state_contents",
"config",
"config_contents",
"private_config",
"headers",
"authorization",
})
# ── Configuration ───────────────────────────────────────────────────────────
@dataclass(frozen=True)
class SentryConfig:
"""Immutable snapshot of the Sentry env configuration."""
enabled: bool = False
dsn: str | None = None
environment: str = "development"
release: str | None = None
traces_sample_rate: float = 0.0
enable_logs: bool = False
@property
def active(self) -> bool:
"""True only when the operator both opted in *and* supplied a DSN.
This is the single gate that keeps the feature off by default: enabling
the flag without a DSN (or vice versa) sends nothing.
"""
return bool(self.enabled and self.dsn)
def safe_summary(self) -> dict[str, Any]:
"""Operator-facing status with **no** DSN value (only presence)."""
return {
"enabled": self.enabled,
"dsn_present": bool(self.dsn),
"environment": self.environment,
"release": self.release,
"traces_sample_rate": self.traces_sample_rate,
"enable_logs": self.enable_logs,
"active": self.active,
}
def _env_bool(name: str, env: dict[str, str]) -> bool:
return (env.get(name) or "").strip().lower() in _TRUTHY
def _env_float(name: str, default: float, env: dict[str, str]) -> float:
raw = (env.get(name) or "").strip()
if not raw:
return default
try:
val = float(raw)
except (TypeError, ValueError):
return default
# Clamp to Sentry's valid [0.0, 1.0] sample-rate range.
if val < 0.0:
return 0.0
if val > 1.0:
return 1.0
return val
def load_config(env: dict[str, str] | None = None) -> SentryConfig:
"""Build a :class:`SentryConfig` from the environment (read at call time)."""
env = dict(os.environ if env is None else env)
dsn = (env.get(ENV_DSN) or "").strip() or None
return SentryConfig(
enabled=_env_bool(ENV_ENABLED, env),
dsn=dsn,
environment=(env.get(ENV_ENVIRONMENT) or "").strip() or "development",
release=(env.get(ENV_RELEASE) or "").strip() or None,
traces_sample_rate=_env_float(ENV_TRACES_SAMPLE_RATE, 0.0, env),
enable_logs=_env_bool(ENV_ENABLE_LOGS, env),
)
def sdk_available() -> bool:
"""True when the optional ``sentry_sdk`` package is importable."""
return sentry_sdk is not None
# ── Redaction (fail closed) ─────────────────────────────────────────────────
def sanitize_path(value: Any) -> str:
"""Reduce a filesystem path to a non-sensitive *category* token.
Full local paths must never be sent. We keep only a coarse worktree
category derived from the path shape (author/reviewer/merger/reconciler/
branches/root/other).
"""
text = "" if value is None else str(value)
low = text.lower()
if not text:
return "unknown"
# Order matters: more specific role markers before the generic "branches".
if "reconcile" in low:
return "reconciler"
if "review" in low:
return "reviewer"
if "merge" in low or "merger" in low:
return "merger"
if "author" in low or re.search(r"/branches/(?:feat|fix|docs|chore|issue)", low):
return "author"
if "/branches/" in low:
return "branches"
if low.rstrip("/").endswith("gitea-tools"):
return "root"
return "other"
def redact_value(value: Any) -> Any:
"""Recursively redact a JSON-able value: secret text, absolute paths, and
known-sensitive dict keys are removed. Fail closed — any error drops the
value entirely rather than risk leaking it."""
try:
if isinstance(value, dict):
out: dict[str, Any] = {}
for k, v in value.items():
key = str(k)
low = key.lower()
if low in _FORBIDDEN_EXTRA_KEYS or any(
s in low
for s in ("token", "secret", "password", "cookie", "auth", "dsn", "keychain")
):
out[key] = REDACTED
continue
out[key] = redact_value(v)
return out
if isinstance(value, (list, tuple)):
return [redact_value(v) for v in value]
if isinstance(value, str):
scrubbed = _redact_text(value)
scrubbed = _redact_prefixes(scrubbed)
scrubbed = _PATH_RE.sub(REDACTED_PATH, scrubbed)
return scrubbed
return value
except Exception:
return REDACTED
def hash_session_id(session_id: Any) -> str:
"""Short, stable, non-reversible fingerprint of a session id."""
digest = hashlib.sha256(str(session_id).encode("utf-8", "replace")).hexdigest()
return digest[:12]
def build_tags(**kwargs: Any) -> dict[str, str]:
"""Return a scrubbed, allowlisted tag dict.
``session_id`` is accepted but only ever surfaced as ``session_id_hash``.
``worktree_path`` collapses to ``worktree_category``. Any non-allowlisted
key, or a value that still contains redacted material after scrubbing, is
dropped.
"""
raw: dict[str, Any] = dict(kwargs)
# Hash the session id — never emit it raw.
session_id = raw.pop("session_id", None)
if session_id and "session_id_hash" not in raw:
raw["session_id_hash"] = hash_session_id(session_id)
# A full worktree path collapses to a category tag.
wt = raw.pop("worktree_path", None)
if wt and "worktree_category" not in raw:
raw["worktree_category"] = sanitize_path(wt)
out: dict[str, str] = {}
for key, val in raw.items():
if key not in ALLOWED_TAG_KEYS:
continue
if val is None:
continue
scrubbed = redact_value(val)
text = str(scrubbed)
if not text or REDACTED in text or REDACTED_PATH in text:
continue
if len(text) > 200:
text = text[:200] + ""
out[key] = text
return out
def scrub_event(event: Any, hint: Any = None) -> dict[str, Any] | None:
"""Sentry ``before_send`` / ``before_send_log`` hook.
Recursively redacts the outgoing event. On *any* failure it returns ``None``
so the event is dropped rather than sent unscrubbed (fail closed for
redaction).
"""
try:
if not isinstance(event, dict):
return None
scrubbed = redact_value(event)
# Drop server_name if it leaked a hostname/path; PID is kept via tags.
scrubbed.pop("server_name", None)
return scrubbed
except Exception:
return None
# ── Event builders (pure, independently testable) ───────────────────────────
def build_blocker_event(
blocker_type: str,
*,
message: str | None = None,
next_action: str | None = None,
level: str = "warning",
tags: dict[str, Any] | None = None,
extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Build a redacted, structured Sentry event for a workflow blocker.
``next_action`` maps the issue's "canonical next action when available"
requirement (acceptance criterion 7).
"""
merged_tags = dict(tags or {})
merged_tags.setdefault("blocker_type", blocker_type)
safe_tags = build_tags(**merged_tags)
safe_extra = redact_value(dict(extra or {}))
if next_action:
# A short canonical next action is allowed (it is not a full prompt).
safe_extra["canonical_next_action"] = redact_value(str(next_action)[:500])
event: dict[str, Any] = {
"message": redact_value(message or blocker_type),
"level": level if level in ("debug", "info", "warning", "error", "fatal") else "warning",
"logger": "gitea-mcp.workflow",
"tags": safe_tags,
"extra": safe_extra,
"fingerprint": ["workflow-blocker", blocker_type],
}
return event
def build_checkin_payload(
monitor: str,
status: str,
*,
check_in_id: str | None = None,
duration: float | None = None,
) -> dict[str, Any]:
"""Build a Sentry cron check-in payload for one of :data:`MONITOR_SLUGS`.
``monitor`` may be a registry key (e.g. ``"stale_lease_scan"``) or an
explicit slug. Raises ``ValueError`` on an unknown status so callers cannot
silently send a malformed check-in.
"""
if status not in _CHECKIN_STATUSES:
raise ValueError(
f"invalid check-in status {status!r}; expected one of {sorted(_CHECKIN_STATUSES)}"
)
slug = MONITOR_SLUGS.get(monitor, monitor)
payload: dict[str, Any] = {"monitor_slug": slug, "status": status}
if check_in_id:
payload["check_in_id"] = str(check_in_id)
if duration is not None:
try:
payload["duration"] = float(duration)
except (TypeError, ValueError):
pass
return payload
# ── Runtime init + capture (fail open) ──────────────────────────────────────
_STATE: dict[str, Any] = {"initialized": False, "config": None}
def is_initialized() -> bool:
return bool(_STATE.get("initialized"))
def active_config() -> SentryConfig | None:
return _STATE.get("config")
def reset_for_tests() -> None:
"""Clear module init state. Test-only helper (never called in production)."""
_STATE["initialized"] = False
_STATE["config"] = None
def init_sentry(config: SentryConfig | None = None) -> dict[str, Any]:
"""Initialise the Sentry SDK if (and only if) enabled + DSN + SDK present.
Idempotent and never raises. Returns an operator-safe status dict (no DSN
value). Behaviour is unchanged when the feature is off.
"""
cfg = config or load_config()
status: dict[str, Any] = {"initialized": False, **cfg.safe_summary()}
try:
if not cfg.active:
status["reason"] = "disabled (MCP_SENTRY_ENABLED false or SENTRY_DSN empty)"
_STATE["config"] = cfg
return status
if not sdk_available():
status["reason"] = "sentry_sdk not installed"
_STATE["config"] = cfg
return status
init_kwargs: dict[str, Any] = {
"dsn": cfg.dsn,
"environment": cfg.environment,
"release": cfg.release,
"traces_sample_rate": cfg.traces_sample_rate,
"before_send": scrub_event,
"send_default_pii": False,
}
if cfg.enable_logs:
# sentry-sdk 2.x captures Python logs as structured logs when the
# experimental logs feature is enabled; scrub those too.
init_kwargs["_experiments"] = {
"enable_logs": True,
"before_send_log": scrub_event,
}
sentry_sdk.init(**init_kwargs) # type: ignore[union-attr]
_STATE["initialized"] = True
_STATE["config"] = cfg
status["initialized"] = True
status["reason"] = "sentry initialised"
except Exception as exc: # fail open: observability must not block startup
status["reason"] = f"init failed (ignored): {type(exc).__name__}"
_STATE["initialized"] = False
return status
def _set_scope_tags(scope: Any, tags: dict[str, str]) -> None:
for key, val in tags.items():
try:
scope.set_tag(key, val)
except Exception:
pass
def capture_workflow_blocker(
blocker_type: str,
*,
message: str | None = None,
next_action: str | None = None,
level: str = "warning",
tags: dict[str, Any] | None = None,
extra: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Capture a fail-closed workflow blocker as a structured Sentry event.
Always returns the redacted event dict (so callers/tests can inspect it),
and sends it to Sentry only when initialised. Fail open.
"""
event = build_blocker_event(
blocker_type,
message=message,
next_action=next_action,
level=level,
tags=tags,
extra=extra,
)
try:
if is_initialized() and sdk_available():
sentry_sdk.capture_event(event) # type: ignore[union-attr]
except Exception:
pass
return event
def capture_exception(
exc: BaseException,
*,
tags: dict[str, Any] | None = None,
extra: dict[str, Any] | None = None,
) -> bool:
"""Capture a runtime exception with scrubbed tags. Fail open.
Returns True only when the event was handed to an initialised SDK.
"""
try:
if not (is_initialized() and sdk_available()):
return False
safe_tags = build_tags(**(tags or {}))
safe_extra = redact_value(dict(extra or {}))
with sentry_sdk.push_scope() as scope: # type: ignore[union-attr]
_set_scope_tags(scope, safe_tags)
for key, val in safe_extra.items():
try:
scope.set_extra(key, val)
except Exception:
pass
sentry_sdk.capture_exception(exc) # type: ignore[union-attr]
return True
except Exception:
return False
def monitor_checkin(
monitor: str,
status: str,
*,
check_in_id: str | None = None,
duration: float | None = None,
) -> dict[str, Any] | None:
"""Send a Sentry cron check-in for a watchdog job. Fail open.
Returns the payload (for inspection/tests), or ``None`` if the status was
invalid. Only transmits when initialised.
"""
try:
payload = build_checkin_payload(
monitor, status, check_in_id=check_in_id, duration=duration
)
except ValueError:
return None
try:
if is_initialized() and sdk_available() and hasattr(sentry_sdk, "capture_checkin"):
sentry_sdk.capture_checkin(**payload) # type: ignore[union-attr]
except Exception:
pass
return payload
-10
View File
@@ -76,16 +76,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.pr.comment", "permission": "gitea.pr.comment",
"role": "reviewer", "role": "reviewer",
}, },
# #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
# Apply path posts lease release + audit comments (gitea.pr.comment).
"cleanup_obsolete_reviewer_comment_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"gitea_cleanup_obsolete_reviewer_comment_lease": {
"permission": "gitea.pr.comment",
"role": "reviewer",
},
"blind_pr_queue_review": { "blind_pr_queue_review": {
"permission": "gitea.pr.review", "permission": "gitea.pr.review",
"role": "reviewer", "role": "reviewer",
+16 -22
View File
@@ -79,46 +79,41 @@ class TestApiRequestFailures(unittest.TestCase):
@patch("gitea_auth.urllib.request.urlopen") @patch("gitea_auth.urllib.request.urlopen")
def test_timeout_converted_to_runtimeerror(self, mock_open): def test_timeout_converted_to_runtimeerror(self, mock_open):
mock_open.side_effect = TimeoutError("timed out") mock_open.side_effect = TimeoutError("timed out")
with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx: with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH) gitea_auth.api_request("GET", URL, FAKE_AUTH)
# Fixed message only (#699) — still a RuntimeError subclass. self.assertIn("network error contacting Gitea", str(ctx.exception))
self.assertIsInstance(ctx.exception, RuntimeError)
self.assertEqual(str(ctx.exception), "Network error contacting Gitea")
@patch("gitea_auth.urllib.request.urlopen") @patch("gitea_auth.urllib.request.urlopen")
def test_dns_network_failure_converted(self, mock_open): def test_dns_network_failure_converted(self, mock_open):
mock_open.side_effect = urllib.error.URLError("Name or service not known") mock_open.side_effect = urllib.error.URLError("Name or service not known")
with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx: with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH) gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertEqual(str(ctx.exception), "Network error contacting Gitea") self.assertIn("network error contacting Gitea", str(ctx.exception))
@patch("gitea_auth.urllib.request.urlopen") @patch("gitea_auth.urllib.request.urlopen")
def test_502_upstream_message(self, mock_open): def test_502_upstream_message(self, mock_open):
mock_open.side_effect = http_error(502, "bad gateway") mock_open.side_effect = http_error(502, "bad gateway")
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH) gitea_auth.api_request("GET", URL, FAKE_AUTH)
msg = str(ctx.exception) msg = str(ctx.exception)
self.assertEqual(msg, "Gitea upstream unavailable") self.assertIn("HTTP 502", msg)
self.assertEqual(ctx.exception.reason_code, "upstream_unavailable") self.assertIn("upstream unavailable", msg)
self.assertEqual(ctx.exception.http_status, 502)
@patch("gitea_auth.urllib.request.urlopen") @patch("gitea_auth.urllib.request.urlopen")
def test_503_upstream_message(self, mock_open): def test_503_upstream_message(self, mock_open):
mock_open.side_effect = http_error(503, "") mock_open.side_effect = http_error(503, "")
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH) gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertEqual(str(ctx.exception), "Gitea upstream unavailable") self.assertIn("HTTP 503", str(ctx.exception))
self.assertEqual(ctx.exception.http_status, 503) self.assertIn("upstream unavailable", str(ctx.exception))
@patch("gitea_auth.urllib.request.urlopen") @patch("gitea_auth.urllib.request.urlopen")
def test_malformed_error_payload_does_not_crash(self, mock_open): def test_malformed_error_payload_does_not_crash(self, mock_open):
# Non-JSON garbage error body must still yield a clean typed error # Non-JSON garbage error body must still yield a clean RuntimeError.
# with a fixed message (no body echo — #699).
mock_open.side_effect = http_error(500, "<html>garbage</html>") mock_open.side_effect = http_error(500, "<html>garbage</html>")
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH) gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertEqual(str(ctx.exception), "Gitea HTTP request failed") self.assertIn("HTTP 500", str(ctx.exception))
self.assertNotIn("<html>", str(ctx.exception))
@patch("gitea_auth.urllib.request.urlopen") @patch("gitea_auth.urllib.request.urlopen")
def test_malformed_success_json_raises_clean_error(self, mock_open): def test_malformed_success_json_raises_clean_error(self, mock_open):
@@ -131,17 +126,16 @@ class TestApiRequestFailures(unittest.TestCase):
def test_no_secret_leak_in_error_body(self, mock_open): def test_no_secret_leak_in_error_body(self, mock_open):
mock_open.side_effect = http_error( mock_open.side_effect = http_error(
400, "failed: token supersecret123 rejected") 400, "failed: token supersecret123 rejected")
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH) gitea_auth.api_request("GET", URL, FAKE_AUTH)
msg = str(ctx.exception) msg = str(ctx.exception)
self.assertNotIn("supersecret123", msg) self.assertNotIn("supersecret123", msg)
# Fixed message — body never appears (stronger than redaction). self.assertIn(gitea_audit.REDACTED, msg)
self.assertEqual(msg, "Gitea HTTP request failed")
@patch("gitea_auth.urllib.request.urlopen") @patch("gitea_auth.urllib.request.urlopen")
def test_auth_header_never_in_error(self, mock_open): def test_auth_header_never_in_error(self, mock_open):
mock_open.side_effect = http_error(400, "bad request") mock_open.side_effect = http_error(400, "bad request")
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH) gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertNotIn(FAKE_AUTH, str(ctx.exception)) self.assertNotIn(FAKE_AUTH, str(ctx.exception))
+10 -39
View File
@@ -57,23 +57,9 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
# Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
# path is the stable control checkout (not under branches/), mutation must fail. # path is the stable control checkout (not under branches/), mutation must fail.
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try: with self.assertRaises(RuntimeError) as ctx:
res = srv.gitea_create_issue(title="Test issue", body="body text") srv.gitea_create_issue(title="Test issue", body="body text")
except RuntimeError as exc: self.assertIn("stable control checkout", str(ctx.exception))
self.assertIn("stable control checkout", str(exc))
else:
# #683: production guards return typed blockers at entrypoints
self.assertFalse(res.get("success"))
self.assertFalse(res.get("performed"))
blob = " ".join(res.get("reasons") or []) + " " + str(
res.get("blocker_kind") or ""
)
self.assertTrue(
"stable control checkout" in blob
or "missing_issue_worktree" in blob
or "control checkout" in blob.lower()
)
self.assertTrue(res.get("exact_next_action"))
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None) @patch("gitea_mcp_server._profile_permission_block", return_value=None)
@@ -119,17 +105,11 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
) )
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try: with self.assertRaises(RuntimeError) as ctx:
res = srv.gitea_create_issue( srv.gitea_create_issue(
title="Test issue", body="body", worktree_path=missing_path title="Test issue", body="body", worktree_path=missing_path
) )
except RuntimeError as exc: self.assertIn("does not exist (fail closed)", str(ctx.exception))
self.assertIn("does not exist", str(exc))
else:
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or [])
self.assertIn("does not exist", blob)
self.assertTrue(res.get("exact_next_action") or res.get("reasons"))
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None) @patch("gitea_mcp_server._profile_permission_block", return_value=None)
@@ -162,20 +142,11 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""): with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
try: with self.assertRaises(RuntimeError) as ctx:
res = srv.gitea_create_issue( srv.gitea_create_issue(
title="Test issue", title="Test issue", body="body", worktree_path=wrong_repo_path
body="body",
worktree_path=wrong_repo_path,
) )
except RuntimeError as exc: self.assertIn("does not belong to the target repository", str(ctx.exception))
self.assertIn(
"does not belong to the target repository", str(exc)
)
else:
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or [])
self.assertIn("does not belong to the target repository", blob)
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server._profile_permission_block", return_value=None) @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+2 -11
View File
@@ -267,19 +267,10 @@ class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase):
with patch.dict(os.environ, {}, clear=False): with patch.dict(os.environ, {}, clear=False):
os.environ.pop("GITEA_AUTHOR_WORKTREE", None) os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
os.environ.pop("GITEA_ACTIVE_WORKTREE", None) os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
try: with self.assertRaises(RuntimeError):
res = srv.gitea_create_issue_comment( srv.gitea_create_issue_comment(
515, "author note", remote="prgs" 515, "author note", remote="prgs"
) )
except RuntimeError:
pass # legacy raise path
else:
# #683: typed blocker at mutation entrypoint
self.assertFalse(res.get("success"))
self.assertFalse(res.get("performed"))
self.assertTrue(
res.get("blocker_kind") or res.get("reasons")
)
mock_api.assert_not_called() mock_api.assert_not_called()
@@ -1,472 +0,0 @@
"""#683: block unattributed root WIP; pytest cannot disable production guards.
Regression coverage required by issue #683:
1. Session locked to issue A blocks unrelated target issue B until B is selected.
2. Diagnostic source edit on the root checkout is blocked.
3. Same legitimate edit succeeds after issue ownership + isolated worktree bind.
4. Running under pytest does not deactivate production guards when force-on.
5. Dirty tracked Python files remain visible to porcelain consumers.
6. Monkeypatching one helper cannot silently turn the full guard path into a no-op.
7. Real mutation entrypoint proves production guards run before side effects.
8. Same-issue edits in a valid isolated worktree remain unaffected.
9. Blocker includes stable reason + exact recovery action.
"""
from __future__ import annotations
import os
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import gitea_mcp_server as mcp_server # noqa: E402
import issue_lock_worktree # noqa: E402
import workflow_scope_guard as wsg # noqa: E402
CONTROL_ROOT = str(Path(__file__).resolve().parent.parent)
if "branches" in Path(__file__).resolve().parts:
# Running from a worktree under branches/ — parent of branches is control.
parts = Path(__file__).resolve().parts
idx = parts.index("branches")
CONTROL_ROOT = str(Path(*parts[:idx])) if idx > 0 else CONTROL_ROOT
class TestProductionGuardsForceOn(unittest.TestCase):
def tearDown(self):
for key in (
wsg.FORCE_PRODUCTION_GUARDS_ENV,
"GITEA_TEST_FORCE_DIRTY",
"GITEA_TEST_PORCELAIN",
"GITEA_AUTHOR_WORKTREE",
"GITEA_ACTIVE_WORKTREE",
):
os.environ.pop(key, None)
wsg.clear_workflow_failure_ledger()
def test_force_on_under_pytest_keeps_production_active(self):
self.assertTrue(wsg.production_guards_active(in_test_mode=False))
self.assertFalse(wsg.production_guards_active(in_test_mode=True))
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
self.assertTrue(wsg.production_guards_active(in_test_mode=True))
self.assertTrue(wsg.production_guards_forced())
def test_no_early_return_in_verify_role_mutation_workspace_source(self):
src = Path(mcp_server.__file__).read_text(encoding="utf-8")
# Rejected 300a4ca pattern must not exist.
self.assertNotIn(
"if _preflight_in_test_mode():\n return _resolve_preflight_workspace_path",
src,
)
# Docstring contract for #683.
self.assertIn("#683", src)
self.assertIn("must NOT early-return solely because pytest", src)
class TestPorcelainIntegrity(unittest.TestCase):
def test_read_worktree_git_state_surfaces_dirty_py(self):
with tempfile.TemporaryDirectory() as tmp:
# Use a real git repo so porcelain is truthful.
import subprocess
subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
subprocess.run(
["git", "config", "user.email", "[email protected]"],
cwd=tmp,
check=True,
capture_output=True,
)
subprocess.run(
["git", "config", "user.name", "t"],
cwd=tmp,
check=True,
capture_output=True,
)
py_path = Path(tmp) / "sample_mod.py"
py_path.write_text("x = 1\n", encoding="utf-8")
subprocess.run(["git", "add", "sample_mod.py"], cwd=tmp, check=True)
subprocess.run(
["git", "commit", "-m", "init"],
cwd=tmp,
check=True,
capture_output=True,
)
py_path.write_text("x = 2\n", encoding="utf-8")
state = issue_lock_worktree.read_worktree_git_state(tmp)
porcelain = state.get("porcelain_status") or ""
self.assertIn("sample_mod.py", porcelain)
self.assertTrue(any(line.strip().endswith(".py") for line in porcelain.splitlines()))
def test_production_reader_source_rejects_pytest_py_filter(self):
src = Path(issue_lock_worktree.__file__).read_text(encoding="utf-8")
findings = wsg.assert_no_pytest_porcelain_filter(src)
self.assertEqual(findings, [])
# Negative: the rejected 300a4ca pattern is detected.
rejected = textwrap.dedent(
"""
porcelain = status_res.stdout or ""
import sys
if "pytest" in sys.modules or "unittest" in sys.modules:
porcelain = "\\n".join(
line for line in porcelain.splitlines()
if not line.strip().endswith(".py")
)
"""
)
self.assertTrue(wsg.assert_no_pytest_porcelain_filter(rejected))
class TestIssueScopeOwnership(unittest.TestCase):
def test_out_of_scope_issue_blocked_until_selected(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=100,
target_issue_number=200,
branch_name="fix/issue-100-example",
role_kind="author",
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE)
self.assertIn("exact_next_action", result)
self.assertIn("owning issue", result["exact_next_action"].lower())
self.assertTrue(result["reasons"])
def test_same_issue_scope_allowed(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=100,
target_issue_number=100,
branch_name="fix/issue-100-example",
role_kind="author",
)
self.assertFalse(result["block"])
self.assertEqual(result["exact_next_action"], "proceed")
def test_missing_lock_when_required(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=None,
target_issue_number=None,
role_kind="author",
require_lock_for_author=True,
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_MISSING_ISSUE_SCOPE)
def test_branch_issue_mismatch(self):
result = wsg.assess_issue_scope_ownership(
locked_issue_number=50,
branch_name="fix/issue-99-other",
role_kind="author",
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE)
class TestRootDiagnosticEdit(unittest.TestCase):
def test_dirty_root_source_blocked(self):
result = wsg.assess_root_source_mutation(
workspace_path=CONTROL_ROOT,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M gitea_mcp_server.py\n M tests/test_x.py\n",
role_kind="author",
)
self.assertTrue(result["block"])
self.assertEqual(result["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT)
self.assertIn("gitea_mcp_server.py", result["dirty_source_files"])
self.assertIn("exact_next_action", result)
self.assertIn("branches/", result["exact_next_action"])
def test_isolated_worktree_same_issue_unaffected(self):
wt = f"{CONTROL_ROOT}/branches/issue-100-example"
result = wsg.assess_root_source_mutation(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M helper.py\n",
current_branch="fix/issue-100-example",
locked_issue_number=100,
role_kind="author",
)
self.assertFalse(result["block"])
self.assertTrue(result["under_branches"])
def test_legitimate_after_ownership_and_worktree(self):
wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening"
composed = wsg.assess_production_mutation_guards(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M workflow_scope_guard.py\n",
current_branch="fix/issue-683-workflow-guard-hardening",
locked_issue_number=683,
target_issue_number=683,
role_kind="author",
require_author_lock=True,
in_test_mode=True,
)
# Force-on required for production path under pytest.
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
try:
composed = wsg.assess_production_mutation_guards(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M workflow_scope_guard.py\n",
current_branch="fix/issue-683-workflow-guard-hardening",
locked_issue_number=683,
target_issue_number=683,
role_kind="author",
require_author_lock=True,
in_test_mode=True,
)
self.assertFalse(composed["block"])
self.assertFalse(composed.get("skipped"))
finally:
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
class TestTypedBlockerResponse(unittest.TestCase):
def test_block_response_has_stable_kind_and_next_action(self):
assessment = wsg.assess_issue_scope_ownership(
locked_issue_number=1,
target_issue_number=2,
role_kind="author",
)
resp = wsg.block_response(assessment)
self.assertFalse(resp["success"])
self.assertFalse(resp["performed"])
self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_OUT_OF_SCOPE_ISSUE)
self.assertIsInstance(resp["exact_next_action"], str)
self.assertTrue(resp["exact_next_action"])
self.assertTrue(resp["reasons"])
def test_production_guard_error_roundtrip(self):
err = wsg.ProductionGuardError(
"blocked",
blocker_kind=wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT,
reasons=["dirty root"],
)
resp = wsg.block_response(err, issue_number=683)
self.assertEqual(resp["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT)
self.assertEqual(resp["issue_number"], 683)
self.assertIn("exact_next_action", resp)
class TestDurableFailureRecording(unittest.TestCase):
def setUp(self):
wsg.clear_workflow_failure_ledger()
def tearDown(self):
wsg.clear_workflow_failure_ledger()
def test_record_before_source_mutation(self):
pending = wsg.assess_durable_failure_recorded(
require_record=True, pending_source_mutation=True
)
self.assertTrue(pending["block"])
self.assertEqual(pending["blocker_kind"], wsg.BLOCKER_UNRECORDED_FAILURE)
wsg.record_workflow_failure(
kind="transport_eof",
detail="EOF during review session (#584 cluster)",
issue_number=683,
task="comment_issue",
)
after = wsg.assess_durable_failure_recorded(
require_record=True, pending_source_mutation=True
)
self.assertFalse(after["block"])
self.assertEqual(len(wsg.workflow_failure_ledger()), 1)
class TestMonkeypatchCannotNoopFullPath(unittest.TestCase):
def tearDown(self):
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
def test_patching_branches_only_still_blocks_dirty_root_scope(self):
"""Monkeypatching branches-only must not silence root diagnostic block."""
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
with patch.object(
mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None
):
with patch.object(
mcp_server, "_enforce_root_checkout_guard", lambda *a, **k: None
):
# Even if both legacy helpers are patched, issue-scope composition
# still sees dirty root source via assess_production_mutation_guards.
assessment = wsg.assess_production_mutation_guards(
workspace_path=CONTROL_ROOT,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M gitea_mcp_server.py\n",
role_kind="author",
in_test_mode=True,
)
self.assertTrue(assessment["block"])
self.assertEqual(
assessment["blocker_kind"], wsg.BLOCKER_ROOT_DIAGNOSTIC_EDIT
)
class TestRealEntrypointProductionGuard(unittest.TestCase):
"""Real mutation entrypoint: production guard before side effects (#683)."""
def setUp(self):
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"):
os.environ.pop(key, None)
self._orig_whoami = mcp_server._preflight_whoami_called
self._orig_cap = mcp_server._preflight_capability_called
mcp_server._preflight_whoami_called = False
mcp_server._preflight_capability_called = False
mcp_server._preflight_resolved_role = None
mcp_server._preflight_resolved_task = None
def tearDown(self):
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"):
os.environ.pop(key, None)
mcp_server._preflight_whoami_called = self._orig_whoami
mcp_server._preflight_capability_called = self._orig_cap
mcp_server._preflight_resolved_role = None
mcp_server._preflight_resolved_task = None
def test_comment_issue_blocks_dirty_root_before_api(self):
api_mock = MagicMock()
with patch.object(mcp_server, "api_request", api_mock), patch.object(
mcp_server,
"_actual_profile_role",
return_value="author",
), patch.object(
mcp_server,
"_effective_workspace_role",
return_value="author",
), patch.object(
mcp_server,
"get_profile",
return_value={
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.issue.comment",
"gitea.read",
"gitea.pr.create",
"gitea.branch.push",
],
"forbidden_operations": [],
},
), patch.object(
issue_lock_worktree,
"read_worktree_git_state",
side_effect=lambda path, **kw: {
"current_branch": "master",
"porcelain_status": (
" M gitea_mcp_server.py\n"
if os.path.realpath(path) == os.path.realpath(CONTROL_ROOT)
or path == CONTROL_ROOT
else ""
),
"head_sha": "a" * 40,
"base_equivalent": True,
},
), patch.object(
mcp_server,
"_resolve_namespace_mutation_context",
return_value={
"workspace_path": CONTROL_ROOT,
"canonical_repo_root": CONTROL_ROOT,
"process_project_root": CONTROL_ROOT,
"workspace_role_kind": "author",
"workspace_binding_source": "process root",
"ignored_bindings": [],
},
), patch.object(
mcp_server,
"_resolve_author_mutation_context",
return_value={
"workspace_path": CONTROL_ROOT,
"canonical_repo_root": CONTROL_ROOT,
"process_project_root": CONTROL_ROOT,
"roots_aligned": True,
},
), patch.object(
mcp_server,
"_session_locked_issue_number",
return_value=None,
):
result = mcp_server.gitea_create_issue_comment(
issue_number=683,
body="diagnostic note",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path=CONTROL_ROOT,
)
api_mock.assert_not_called()
self.assertFalse(result.get("success"))
self.assertFalse(result.get("performed"))
self.assertIn(result.get("blocker_kind"), wsg.BLOCKER_KINDS)
self.assertTrue(result.get("exact_next_action"))
self.assertTrue(result.get("reasons"))
def test_comment_issue_succeeds_structure_after_worktree_bind(self):
"""Same-issue isolated worktree is not blocked by root diagnostic path."""
wt = f"{CONTROL_ROOT}/branches/issue-683-workflow-guard-hardening"
os.environ["GITEA_AUTHOR_WORKTREE"] = wt
assessment = wsg.assess_production_mutation_guards(
workspace_path=wt,
canonical_repo_root=CONTROL_ROOT,
porcelain_status=" M workflow_scope_guard.py\n",
current_branch="fix/issue-683-workflow-guard-hardening",
locked_issue_number=683,
target_issue_number=683,
role_kind="author",
require_author_lock=True,
in_test_mode=True,
)
self.assertFalse(assessment["block"], assessment)
class TestVerifyPreflightForceOn(unittest.TestCase):
def tearDown(self):
os.environ.pop(wsg.FORCE_PRODUCTION_GUARDS_ENV, None)
for key in ("GITEA_AUTHOR_WORKTREE", "GITEA_ACTIVE_WORKTREE"):
os.environ.pop(key, None)
def test_force_on_runs_production_guards_under_pytest(self):
os.environ[wsg.FORCE_PRODUCTION_GUARDS_ENV] = "1"
called = {"root": 0, "branches": 0, "scope": 0}
def _root(*a, **k):
called["root"] += 1
def _branches(*a, **k):
called["branches"] += 1
def _scope(*a, **k):
called["scope"] += 1
with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object(
mcp_server, "_enforce_branches_only_author_mutation", _branches
), patch.object(mcp_server, "_enforce_issue_scope_guard", _scope):
# No whoami/capability — purity-order skipped; production still runs.
mcp_server.verify_preflight_purity(task="comment_issue")
self.assertEqual(called["root"], 1)
self.assertEqual(called["branches"], 1)
self.assertEqual(called["scope"], 1)
def test_without_force_on_pytest_skips_production_only_for_unit_isolation(self):
called = {"root": 0}
def _root(*a, **k):
called["root"] += 1
with patch.object(mcp_server, "_enforce_root_checkout_guard", _root), patch.object(
mcp_server, "_enforce_branches_only_author_mutation", lambda *a, **k: None
), patch.object(mcp_server, "_enforce_issue_scope_guard", lambda *a, **k: None):
mcp_server.verify_preflight_purity(task="comment_issue")
self.assertEqual(called["root"], 0)
if __name__ == "__main__":
unittest.main()
@@ -1,599 +0,0 @@
"""Guarded cleanup for obsolete comment-backed reviewer leases (#691).
Reproduces PR #688 lease comment 10749 class of defect: completed
REQUEST_CHANGES on head A, author pushes head B, foreign lease remains
pinned to A (and after expiry still has no non-owner cleanup path that
diagnosis can prescribe beyond indefinite wait).
"""
from __future__ import annotations
import sys
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import merger_lease_adoption as mla # noqa: E402
import reviewer_pr_lease as leases # noqa: E402
# PR #688 / comment 10749 reproduction fixtures
HEAD_A = "c7a444eb4b41cf916fdbd20a4999ffd78af496d0"
HEAD_B = "4a6357800364718a27a36ebc73578d4b929ff4aa"
SESSION_FOREIGN = "25883-7d4c6e6ebd53"
COMMENT_10749 = 10749
REPO = "Scaled-Tech-Consulting/Gitea-Tools"
PR = 688
ISSUE = 687
EXPIRES_10749 = "2026-07-13T04:23:00Z"
def _utc(dt: datetime) -> datetime:
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
def _lease_comment(
*,
pr_number: int = PR,
session_id: str = SESSION_FOREIGN,
phase: str = "claimed",
candidate_head: str = HEAD_A,
comment_id: int = COMMENT_10749,
last_activity: datetime | None = None,
expires_at: datetime | None = None,
worktree: str = "branches/review-pr688-feat-issue-687-reconciler-branch-delete",
repo: str = REPO,
) -> dict:
now = last_activity or datetime.now(timezone.utc)
body = leases.format_lease_body(
repo=repo,
pr_number=pr_number,
issue_number=ISSUE,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id=session_id,
worktree=worktree,
phase=phase,
candidate_head=candidate_head,
target_branch="master",
target_branch_sha="b" * 40,
last_activity=now,
expires_at=expires_at,
)
return {
"id": comment_id,
"body": body,
"user": {"login": "sysadmin"},
"author": "sysadmin",
"created_at": now.isoformat(),
"updated_at": now.isoformat(),
}
def _reviews_for_head(head: str, verdict: str = "REQUEST_CHANGES") -> list[dict]:
return [
{
"verdict": verdict,
"reviewed_head_sha": head,
"dismissed": False,
"reviewer": "sysadmin",
}
]
def _assess(
comments,
*,
current_head=HEAD_B,
formal_reviews=None,
controller=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
requesting_session="fresh-controller",
apply=False,
confirmation=None,
**kwargs,
):
return leases.assess_obsolete_reviewer_comment_lease_cleanup(
comments,
pr_number=PR,
current_head_sha=current_head,
formal_reviews=formal_reviews if formal_reviews is not None else _reviews_for_head(HEAD_A),
repo=REPO,
expected_repo=REPO,
requesting_session_id=requesting_session,
controller_recovery_authorized=controller,
worktree_exists=worktree_exists,
worktree_clean=worktree_clean,
worktree_has_unpreserved_work=not worktree_clean if worktree_exists else False,
owner_process_alive=owner_process_alive,
owner_pid_observed=kwargs.get("owner_pid_observed"),
requesting_pid=kwargs.get("requesting_pid", 99999),
current_head_review_in_progress=kwargs.get(
"current_head_review_in_progress", False
),
expected_lease_comment_id=kwargs.get("expected_lease_comment_id"),
expected_session_id=kwargs.get("expected_session_id"),
expected_leased_head=kwargs.get("expected_leased_head"),
confirmation=confirmation
if confirmation is not None
else (leases.cleanup_confirmation_for_pr(PR) if apply else ""),
apply=apply,
now=kwargs.get("now"),
)
class TestIssue691SupersededHeadCleanup(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_1_request_changes_then_push_expired_cleanup_succeeds(self):
"""Completed REQUEST_CHANGES on A, push B, lease A expires → cleanup OK."""
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
last_activity=expires - timedelta(hours=2),
expires_at=expires,
)
result = _assess(
[claim],
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
now=now,
apply=True,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "foreign_expired_superseded_head")
self.assertEqual(
result["exact_next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
self.assertIn("phase: released", result["release_body"])
self.assertIn("obsolete-superseded-or-expired-lease", result["release_body"])
self.assertIsNotNone(result["audit_comment_body"])
self.assertEqual(result["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
def test_2_approve_then_push_cleanup_policy(self):
"""Completed APPROVE on A, push B → cleanup eligible (completed superseded)."""
now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc) # not yet expired
claim = _lease_comment(
last_activity=now - timedelta(minutes=10),
expires_at=expires,
)
result = _assess(
[claim],
formal_reviews=_reviews_for_head(HEAD_A, "APPROVED"),
now=now,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "foreign_completed_superseded_head")
def test_3_active_current_head_cleanup_denied(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=5),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_B, "REQUEST_CHANGES"),
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertEqual(result["classification"], "foreign_active_current_head")
def test_4_foreign_owner_recent_progress_denied(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=2),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
owner_process_alive=True,
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(
any("recent authenticated progress" in r for r in result["reasons"])
or any("current PR head" in r for r in result["reasons"])
)
def test_5_owner_absent_worktree_dirty_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=expires - timedelta(hours=1),
expires_at=expires,
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
worktree_clean=False,
owner_process_alive=False,
now=now,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("dirty" in r for r in result["reasons"]))
def test_6_owner_absent_worktree_clean_orphan_ok(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=expires - timedelta(hours=1),
expires_at=expires,
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=[],
worktree_clean=True,
owner_process_alive=False,
now=now,
)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["classification"], "orphaned_owner_missing")
def test_7_pid_reuse_not_ownership(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim],
now=now,
owner_pid_observed=4242,
requesting_pid=4242,
)
self.assertTrue(
any("PID equality" in r or "NOT ownership" in r for r in result["reasons"])
or result["owner_session_evidence"]["pid_is_not_ownership_proof"]
)
# Still eligible via superseded+terminal+expired despite matching PID.
self.assertTrue(result["cleanup_allowed"], result["reasons"])
def test_8_comment_backed_no_db_lease_id(self):
"""Cleanup uses comment ledger only — no control-plane lease_id required."""
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
# Explicitly no lease_id field anywhere.
self.assertNotIn("lease_id", claim)
result = _assess([claim], now=now)
self.assertTrue(result["cleanup_allowed"], result["reasons"])
self.assertEqual(result["active_lease"]["comment_id"], COMMENT_10749)
self.assertIsNone(result["active_lease"].get("lease_id"))
def test_9_cleanup_posts_durable_audit_bodies(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], now=now, apply=True)
self.assertTrue(result["cleanup_allowed"])
self.assertIn("#691", result["audit_comment_body"])
self.assertIn(str(COMMENT_10749), result["audit_comment_body"])
self.assertIn(HEAD_A, result["audit_comment_body"])
self.assertIn(HEAD_B, result["audit_comment_body"])
def test_10_fresh_acquire_after_cleanup_marker(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(
expires_at=expires,
last_activity=expires - timedelta(hours=1),
comment_id=COMMENT_10749,
)
assessed = _assess([claim], now=now, apply=True)
self.assertTrue(assessed["cleanup_allowed"])
release = {
"id": 99999,
"body": assessed["release_body"],
"user": {"login": "sysadmin"},
}
# Newest terminal marker ends active lease.
active = leases.find_active_reviewer_lease(
[claim, release], pr_number=PR, now=now
)
self.assertIsNone(active)
acq = leases.assess_acquire_lease(
[claim, release],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="fresh-reviewer-1",
repo=REPO,
issue_number=ISSUE,
worktree="branches/review-pr-688-fresh",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=now,
)
self.assertTrue(acq["acquire_allowed"], acq["reasons"])
def test_11_no_validation_state_transfer(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], now=now, apply=True)
self.assertTrue(result["cleanup_allowed"])
# Release body keeps old candidate_head (no repoint).
self.assertIn(f"candidate_head: {HEAD_A}", result["release_body"])
self.assertNotIn(HEAD_B, result["release_body"].split("candidate_head:")[1].split("\n")[0])
self.assertIn("did not transfer validation", result["audit_comment_body"])
self.assertIn("did not repoint", result["audit_comment_body"])
# Session lease must remain unset by assessment (tool never seeds).
self.assertIsNone(leases.get_session_lease())
def test_12_repeated_cleanup_idempotent(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
first = _assess([claim], now=now, apply=True)
self.assertTrue(first["cleanup_allowed"])
release = {"id": 100001, "body": first["release_body"], "user": {"login": "x"}}
second = _assess([claim, release], now=now, apply=True)
self.assertFalse(second["cleanup_allowed"])
self.assertEqual(second["classification"], "no_lease")
def test_13_malformed_expiry_fails_closed(self):
claim = _lease_comment()
# Corrupt expires_at in the body.
claim["body"] = claim["body"].replace(
claim["body"].split("expires_at:")[1].split("\n")[0].strip(),
"not-a-timestamp",
)
result = _assess([claim], formal_reviews=_reviews_for_head(HEAD_A))
self.assertFalse(result["cleanup_allowed"])
self.assertFalse(result["expires_parseable"])
self.assertTrue(any("expires_at" in r for r in result["reasons"]))
def test_14_wrong_identity_evidence_fails_closed(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
bad_repo = _assess(
[claim],
now=now,
# force repo mismatch via expected_repo
)
# Patch via direct call with wrong expected_repo
bad = leases.assess_obsolete_reviewer_comment_lease_cleanup(
[claim],
pr_number=PR,
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A),
expected_repo="Other-Org/Other-Repo",
requesting_session_id="fresh",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
now=now,
)
self.assertFalse(bad["cleanup_allowed"])
self.assertTrue(any("repository identity" in r for r in bad["reasons"]))
bad_pr = leases.assess_obsolete_reviewer_comment_lease_cleanup(
[claim],
pr_number=999,
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A),
expected_repo=REPO,
requesting_session_id="fresh",
controller_recovery_authorized=True,
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
expected_lease_comment_id=COMMENT_10749,
now=now,
)
self.assertFalse(bad_pr["cleanup_allowed"])
bad_head = _assess(
[claim],
now=now,
expected_leased_head="0" * 40,
)
self.assertFalse(bad_head["cleanup_allowed"])
bad_session = _assess(
[claim],
now=now,
expected_session_id="wrong-session",
)
self.assertFalse(bad_session["cleanup_allowed"])
def test_15_current_head_active_cannot_be_displaced(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=1),
expires_at=now + timedelta(hours=2),
)
result = _assess(
[claim],
current_head=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_B),
now=now,
current_head_review_in_progress=True,
)
self.assertFalse(result["cleanup_allowed"])
# Acquire also blocked by foreign active lease.
acq = leases.assess_acquire_lease(
[claim],
pr_number=PR,
reviewer_identity="other",
profile="prgs-reviewer",
session_id="thief",
repo=REPO,
issue_number=ISSUE,
worktree="branches/steal",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=now,
)
self.assertFalse(acq["acquire_allowed"])
def test_regression_pr688_comment_10749_after_expiry(self):
"""Exact 10749 reproduction after recorded expires_at 2026-07-13T04:23:00Z."""
now = datetime(2026, 7, 13, 4, 30, 0, tzinfo=timezone.utc)
expires = datetime.fromisoformat(EXPIRES_10749.replace("Z", "+00:00"))
claim = _lease_comment(
session_id=SESSION_FOREIGN,
candidate_head=HEAD_A,
comment_id=COMMENT_10749,
last_activity=expires - timedelta(hours=2),
expires_at=expires,
)
# Diagnosis must not prescribe indefinite wait-only for this case.
diag = leases.diagnose_reviewer_pr_lease_handoff(
[claim],
pr_number=PR,
current_session_id="fresh-reviewer",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
worktree_exists=True,
worktree_clean=True,
owner_process_alive=False,
instructed_session_id=SESSION_FOREIGN,
instructed_comment_id=COMMENT_10749,
now=now,
)
self.assertEqual(diag["classification"], "foreign_expired_superseded_head")
self.assertEqual(
diag["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
self.assertEqual(diag["cleanup_tool"], leases.CLEANUP_OBSOLETE_LEASE_TOOL)
self.assertIn("CLEANUP OBSOLETE REVIEWER LEASE 688", diag["required_confirmation"])
self.assertEqual(diag["mutation_eligibility"], "cleanup_only")
self.assertFalse(diag["mutation_allowed"])
# Assessment still returns the lease as active/stale class of block for acquire
# when unexpired path... here it's expired so find_active is None; acquire free
# only after cleanup if somehow still active. With expired, find_active is None:
active = leases.find_active_reviewer_lease([claim], pr_number=PR, now=now)
self.assertIsNone(active)
# But superseded unexpired still blocks acquire and diagnosis points to cleanup:
unexpired_now = datetime(2026, 7, 13, 1, 0, 0, tzinfo=timezone.utc)
claim2 = _lease_comment(
session_id=SESSION_FOREIGN,
candidate_head=HEAD_A,
comment_id=COMMENT_10749,
last_activity=unexpired_now - timedelta(minutes=45),
expires_at=expires,
)
active2 = leases.find_active_reviewer_lease(
[claim2], pr_number=PR, now=unexpired_now
)
self.assertIsNotNone(active2)
self.assertEqual(active2["freshness"], "stale_warning")
diag2 = leases.diagnose_reviewer_pr_lease_handoff(
[claim2],
pr_number=PR,
current_session_id="fresh-reviewer",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=_reviews_for_head(HEAD_A, "REQUEST_CHANGES"),
worktree_exists=True,
worktree_clean=True,
now=unexpired_now,
)
self.assertEqual(diag2["classification"], "foreign_completed_superseded_head")
self.assertEqual(
diag2["next_action"], leases.NEXT_ACTION_CLEANUP_OBSOLETE_LEASE
)
acq = leases.assess_acquire_lease(
[claim2],
pr_number=PR,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="fresh-reviewer",
repo=REPO,
issue_number=ISSUE,
worktree="branches/review-pr-688-new",
candidate_head=HEAD_B,
target_branch="master",
target_branch_sha="b" * 40,
now=unexpired_now,
)
self.assertFalse(acq["acquire_allowed"])
def test_missing_controller_auth_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], controller=False, now=now)
self.assertFalse(result["cleanup_allowed"])
def test_wrong_confirmation_denied_on_apply(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim], now=now, apply=True, confirmation="MERGE PR 688"
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("confirmation" in r for r in result["reasons"]))
def test_owner_session_must_use_owner_release(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess(
[claim],
now=now,
requesting_session=SESSION_FOREIGN,
)
self.assertFalse(result["cleanup_allowed"])
self.assertTrue(any("owns the lease" in r for r in result["reasons"]))
def test_superseded_without_terminal_review_denied(self):
now = datetime(2026, 7, 13, 5, 0, 0, tzinfo=timezone.utc)
expires = datetime(2026, 7, 13, 4, 23, 0, tzinfo=timezone.utc)
claim = _lease_comment(expires_at=expires, last_activity=expires - timedelta(hours=1))
result = _assess([claim], formal_reviews=[], now=now)
self.assertFalse(result["cleanup_allowed"])
self.assertEqual(result["classification"], "ambiguous_conflicting_evidence")
class TestIssue691DiagnoseClassifications(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
def test_foreign_active_current_head_still_wait(self):
now = datetime.now(timezone.utc)
claim = _lease_comment(
candidate_head=HEAD_B,
last_activity=now - timedelta(minutes=5),
expires_at=now + timedelta(hours=1),
)
claim["id"] = 1
result = leases.diagnose_reviewer_pr_lease_handoff(
[claim],
pr_number=PR,
current_session_id="other",
current_reviewer_identity="sysadmin",
current_head_sha=HEAD_B,
formal_reviews=[],
now=now,
)
self.assertEqual(result["classification"], "foreign_active_current_head")
self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
self.assertFalse(result["mutation_allowed"])
if __name__ == "__main__":
unittest.main()
+6 -27
View File
@@ -108,24 +108,13 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state", "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
side_effect=self._git_state(valid_worktree), side_effect=self._git_state(valid_worktree),
): ):
try: with self.assertRaises(RuntimeError) as ctx:
res = srv.gitea_create_issue_comment( srv.gitea_create_issue_comment(
issue_number=557, issue_number=557,
body="evidence comment", body="evidence comment",
remote="prgs", remote="prgs",
) )
except RuntimeError as exc: self.assertIn("stable control checkout", str(ctx.exception))
self.assertIn("stable control checkout", str(exc))
else:
# #683 typed blocker at mutation entrypoint
self.assertFalse(res.get("success"))
self.assertFalse(res.get("performed"))
blob = " ".join(res.get("reasons") or [])
self.assertTrue(
"stable control checkout" in blob
or res.get("blocker_kind")
)
self.assertTrue(res.get("exact_next_action") or res.get("reasons"))
mock_api.assert_not_called() mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@@ -195,24 +184,14 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
side_effect=self._subprocess(valid_worktree, outside_worktree), side_effect=self._subprocess(valid_worktree, outside_worktree),
): ):
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True): with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
try: with self.assertRaises(RuntimeError) as ctx:
res = srv.gitea_create_issue_comment( srv.gitea_create_issue_comment(
issue_number=557, issue_number=557,
body="evidence comment", body="evidence comment",
remote="prgs", remote="prgs",
worktree_path=outside_worktree, worktree_path=outside_worktree,
) )
except RuntimeError as exc: self.assertIn("does not belong to the target repository", str(ctx.exception))
self.assertIn(
"does not belong to the target repository",
str(exc),
)
else:
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or [])
self.assertIn(
"does not belong to the target repository", blob
)
mock_api.assert_not_called() mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH) @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+3 -7
View File
@@ -76,17 +76,13 @@ class TestPreflightReadSurvival(unittest.TestCase):
self.assertIn("task mismatch", str(ctx.exception)) self.assertIn("task mismatch", str(ctx.exception))
def test_capability_consumed_after_mutation_gate(self): def test_capability_consumed_after_mutation_gate(self):
# Use reconciler/close_pr so this purity-order test does not require a
# branches/ worktree (author create_issue would hit #274/#683 guards).
# Test isolation stays explicit; production author guards remain live
# under force-on (see tests/test_issue_683_workflow_scope_guards.py).
mcp_server.record_preflight_check("whoami") mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check( mcp_server.record_preflight_check(
"capability", resolved_role="reconciler", resolved_task="close_pr" "capability", resolved_role="author", resolved_task="create_issue"
) )
mcp_server.verify_preflight_purity(task="close_pr") mcp_server.verify_preflight_purity(task="create_issue")
with self.assertRaises(RuntimeError) as ctx: with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity(task="close_pr") mcp_server.verify_preflight_purity(task="create_issue")
self.assertIn("has not been resolved", str(ctx.exception)) self.assertIn("has not been resolved", str(ctx.exception))
def test_whoami_recovery_after_violation_clears_capability(self): def test_whoami_recovery_after_violation_clears_capability(self):
+3 -15
View File
@@ -86,21 +86,9 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
): ):
srv._preflight_resolved_role = "author" srv._preflight_resolved_role = "author"
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT): with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
try: with self.assertRaises(RuntimeError) as ctx:
res = srv.gitea_create_issue(title="Test", body="body") srv.gitea_create_issue(title="Test", body="body")
except RuntimeError as exc: self.assertIn("stable control checkout", str(ctx.exception))
self.assertIn("stable control checkout", str(exc))
else:
# #683 typed blocker at mutation entrypoint
self.assertFalse(res.get("success"))
blob = " ".join(res.get("reasons") or []) + str(
res.get("blocker_kind") or ""
)
self.assertTrue(
"stable control checkout" in blob
or "missing_issue_worktree" in blob
or "control checkout" in blob.lower()
)
if __name__ == "__main__": if __name__ == "__main__":
+4 -9
View File
@@ -122,11 +122,9 @@ class TestApiRequestRetry(unittest.TestCase):
sleep.assert_not_called() sleep.assert_not_called()
def test_non_429_error_raises_immediately(self): def test_non_429_error_raises_immediately(self):
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: with self.assertRaises(RuntimeError) as ctx:
_call([_http_error(500, body=b"boom")]) _call([_http_error(500, body=b"boom")])
# Fixed message only (#699) — no response body echo. self.assertIn("HTTP 500", str(ctx.exception))
self.assertEqual(str(ctx.exception), "Gitea HTTP request failed")
self.assertEqual(ctx.exception.http_status, 500)
def test_non_429_error_does_not_sleep(self): def test_non_429_error_does_not_sleep(self):
sleep = MagicMock() sleep = MagicMock()
@@ -173,12 +171,9 @@ class TestApiRequestRetry(unittest.TestCase):
# max_retries=3 -> 3 sleeps, then the 4th failure raises. # max_retries=3 -> 3 sleeps, then the 4th failure raises.
errors = [_http_error(429, retry_after="1") for _ in range(4)] errors = [_http_error(429, retry_after="1") for _ in range(4)]
sleep = MagicMock() sleep = MagicMock()
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx: with self.assertRaises(RuntimeError) as ctx:
_call(errors, sleep_func=sleep, max_retries=3) _call(errors, sleep_func=sleep, max_retries=3)
# After retries are exhausted, 429 is a typed HTTP error with fixed self.assertIn("HTTP 429", str(ctx.exception))
# message (#699); status metadata carries 429.
self.assertEqual(str(ctx.exception), "Gitea HTTP request failed")
self.assertEqual(ctx.exception.http_status, 429)
self.assertEqual(sleep.call_count, 3) self.assertEqual(sleep.call_count, 3)
def test_no_infinite_loop_when_always_429(self): def test_no_infinite_loop_when_always_429(self):
+412
View File
@@ -0,0 +1,412 @@
"""Tests for optional self-hosted Sentry observability (#606).
Covers the pure module (config, redaction, event/check-in builders) and the
runtime capture paths using a fake ``sentry_sdk``, so nothing ever touches the
network. Critically proves the feature is a no-op when disabled or DSN-less
(acceptance criterion 1) and that secrets/paths/session-state are never sent
(criterion 5).
"""
import sys
import contextlib
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import sentry_observability as so # noqa: E402
# ── Fake SDK ────────────────────────────────────────────────────────────────
class _FakeScope:
def __init__(self):
self.tags = {}
self.extras = {}
def set_tag(self, k, v):
self.tags[k] = v
def set_extra(self, k, v):
self.extras[k] = v
class FakeSentrySDK:
"""Minimal stand-in exposing the SDK surface sentry_observability uses."""
def __init__(self):
self.init_kwargs = None
self.events = []
self.exceptions = []
self.checkins = []
self.last_scope = None
def init(self, **kwargs):
self.init_kwargs = kwargs
def capture_event(self, event):
self.events.append(event)
def capture_exception(self, exc):
self.exceptions.append(exc)
def capture_checkin(self, **payload):
self.checkins.append(payload)
@contextlib.contextmanager
def push_scope(self):
self.last_scope = _FakeScope()
yield self.last_scope
@pytest.fixture(autouse=True)
def _reset_state():
"""Isolate module init state between tests."""
so.reset_for_tests()
yield
so.reset_for_tests()
@pytest.fixture
def fake_sdk(monkeypatch):
sdk = FakeSentrySDK()
monkeypatch.setattr(so, "sentry_sdk", sdk)
return sdk
# ── Config / gating ─────────────────────────────────────────────────────────
def test_disabled_by_default_empty_env():
cfg = so.load_config(env={})
assert cfg.enabled is False
assert cfg.active is False
def test_enabled_flag_without_dsn_is_not_active():
cfg = so.load_config(env={"MCP_SENTRY_ENABLED": "1"})
assert cfg.enabled is True
assert cfg.dsn is None
assert cfg.active is False # DSN required
def test_dsn_without_enabled_flag_is_not_active():
cfg = so.load_config(env={"SENTRY_DSN": "https://[email protected]/1"})
assert cfg.active is False
def test_active_requires_enabled_and_dsn():
cfg = so.load_config(
env={"MCP_SENTRY_ENABLED": "true", "SENTRY_DSN": "https://[email protected]/1"}
)
assert cfg.active is True
def test_truthy_variants():
for val in ("1", "true", "YES", "On"):
cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val})
assert cfg.enabled is True
for val in ("0", "false", "no", "", "off"):
cfg = so.load_config(env={"MCP_SENTRY_ENABLED": val})
assert cfg.enabled is False
def test_traces_sample_rate_parsed_and_clamped():
assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "0.25"}).traces_sample_rate == 0.25
assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "5"}).traces_sample_rate == 1.0
assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "-1"}).traces_sample_rate == 0.0
assert so.load_config(env={"MCP_SENTRY_TRACES_SAMPLE_RATE": "junk"}).traces_sample_rate == 0.0
def test_safe_summary_has_no_dsn_value():
cfg = so.load_config(
env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
)
summary = cfg.safe_summary()
assert summary["dsn_present"] is True
assert "secret" not in repr(summary)
assert "dsn" not in summary # only presence, never the value
# ── init_sentry ─────────────────────────────────────────────────────────────
def test_init_noop_when_disabled(fake_sdk):
status = so.init_sentry(so.load_config(env={}))
assert status["initialized"] is False
assert fake_sdk.init_kwargs is None # no SDK init
assert so.is_initialized() is False
def test_init_noop_when_enabled_but_missing_dsn(fake_sdk):
status = so.init_sentry(so.load_config(env={"MCP_SENTRY_ENABLED": "1"}))
assert status["initialized"] is False
assert fake_sdk.init_kwargs is None
def test_init_reports_missing_sdk(monkeypatch):
monkeypatch.setattr(so, "sentry_sdk", None)
status = so.init_sentry(
so.load_config(
env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
)
)
assert status["initialized"] is False
assert "not installed" in status["reason"]
def test_init_configures_sdk_with_scrubber(fake_sdk):
status = so.init_sentry(
so.load_config(
env={
"MCP_SENTRY_ENABLED": "1",
"SENTRY_DSN": "https://[email protected]/1",
"SENTRY_ENVIRONMENT": "prod",
"MCP_SENTRY_TRACES_SAMPLE_RATE": "0.1",
}
)
)
assert status["initialized"] is True
assert so.is_initialized() is True
kw = fake_sdk.init_kwargs
assert kw["dsn"] == "https://[email protected]/1"
assert kw["environment"] == "prod"
assert kw["traces_sample_rate"] == 0.1
assert kw["before_send"] is so.scrub_event
assert kw["send_default_pii"] is False
def test_init_enable_logs_wires_log_scrubber(fake_sdk):
so.init_sentry(
so.load_config(
env={
"MCP_SENTRY_ENABLED": "1",
"SENTRY_DSN": "https://[email protected]/1",
"MCP_SENTRY_ENABLE_LOGS": "1",
}
)
)
exp = fake_sdk.init_kwargs["_experiments"]
assert exp["enable_logs"] is True
assert exp["before_send_log"] is so.scrub_event
def test_init_never_raises_on_sdk_failure(monkeypatch):
class Boom:
def init(self, **kwargs):
raise RuntimeError("sentry down")
monkeypatch.setattr(so, "sentry_sdk", Boom())
status = so.init_sentry(
so.load_config(
env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
)
)
assert status["initialized"] is False
assert "init failed" in status["reason"]
# ── Redaction (fail closed) ─────────────────────────────────────────────────
def test_build_tags_allowlist_only():
tags = so.build_tags(role="author", secret_thing="leak", pid=123)
assert tags["role"] == "author"
assert tags["pid"] == "123"
assert "secret_thing" not in tags
def test_build_tags_hashes_session_id():
tags = so.build_tags(session_id="prgs-author-20479-cf9ac178")
assert "session_id" not in tags
assert "session_id_hash" in tags
assert tags["session_id_hash"] != "prgs-author-20479-cf9ac178"
assert len(tags["session_id_hash"]) == 12
def test_build_tags_collapses_worktree_path():
tags = so.build_tags(
worktree_path="/Users/x/Development/Gitea-Tools/branches/issue-606-sentry-observability"
)
assert "worktree_path" not in tags
assert tags["worktree_category"] == "author"
def test_build_tags_drops_value_that_scrubs_to_redacted():
# A tag value that is itself a token gets scrubbed then dropped.
tags = so.build_tags(capability="token abcdef1234567890")
assert "capability" not in tags
def test_sanitize_path_categories():
assert so.sanitize_path("/repo/branches/review-pr-654") == "reviewer"
assert so.sanitize_path("/repo/branches/merge-pr-1") == "merger"
assert so.sanitize_path("/repo/branches/reconcile-pr-1") == "reconciler"
assert so.sanitize_path("/repo/branches/feat-issue-606") == "author"
assert so.sanitize_path("/x/y/Gitea-Tools") == "root"
def test_redact_value_scrubs_secrets_and_paths():
out = so.redact_value(
{
"token": "abc123",
"note": "Authorization: Bearer sk_live_abcdefgh12345",
"path": "/Users/jasonwalker/Development/Gitea-Tools/secret",
"dsn": "https://[email protected]/1",
"safe": "hello",
}
)
assert out["token"] == so.REDACTED
assert out["dsn"] == so.REDACTED
assert "sk_live" not in out["note"]
assert so.REDACTED_PATH in out["path"]
assert "/Users/" not in out["path"]
assert out["safe"] == "hello"
def test_redact_value_forbidden_prompt_and_session_state():
out = so.redact_value(
{"prompt": "full body", "session_state": "{...}", "keep": "ok"}
)
assert out["prompt"] == so.REDACTED
assert out["session_state"] == so.REDACTED
assert out["keep"] == "ok"
def test_scrub_event_redacts_nested_and_drops_server_name():
event = {
"server_name": "some-host",
"message": "boom",
"extra": {"token": "leak", "ok": "1"},
}
scrubbed = so.scrub_event(event)
assert "server_name" not in scrubbed
assert scrubbed["extra"]["token"] == so.REDACTED
assert scrubbed["extra"]["ok"] == "1"
def test_scrub_event_drops_non_dict():
assert so.scrub_event("not a dict") is None
assert so.scrub_event(None) is None
# ── Event builders ──────────────────────────────────────────────────────────
def test_build_blocker_event_structure_and_next_action():
event = so.build_blocker_event(
"active_foreign_lease",
message="blocked by foreign lease",
next_action="wait or adopt via allocator",
level="warning",
tags={"pr_number": 606, "session_id": "s-123"},
)
assert event["tags"]["blocker_type"] == "active_foreign_lease"
assert event["tags"]["pr_number"] == "606"
assert "session_id" not in event["tags"]
assert event["tags"]["session_id_hash"]
assert event["extra"]["canonical_next_action"] == "wait or adopt via allocator"
assert event["fingerprint"] == ["workflow-blocker", "active_foreign_lease"]
def test_build_blocker_event_invalid_level_defaults_warning():
event = so.build_blocker_event("x", level="nonsense")
assert event["level"] == "warning"
def test_build_checkin_payload_maps_slugs():
for key, slug in so.MONITOR_SLUGS.items():
payload = so.build_checkin_payload(key, "ok")
assert payload["monitor_slug"] == slug
assert payload["status"] == "ok"
def test_build_checkin_payload_explicit_slug_passthrough():
payload = so.build_checkin_payload("custom-slug", "in_progress", duration=1.5)
assert payload["monitor_slug"] == "custom-slug"
assert payload["duration"] == 1.5
def test_build_checkin_payload_rejects_bad_status():
with pytest.raises(ValueError):
so.build_checkin_payload("allocator_health", "bogus")
def test_all_six_monitors_registered():
assert set(so.MONITOR_SLUGS) == {
"stale_lease_scan",
"terminal_lock_scan",
"allocator_health",
"namespace_health",
"dashboard_freshness",
"reconciler_cleanup",
}
# ── Capture paths (fail open) ───────────────────────────────────────────────
def test_capture_workflow_blocker_noop_when_disabled(fake_sdk):
# not initialised
event = so.capture_workflow_blocker("some_blocker", message="x")
assert isinstance(event, dict) # still returns redacted event
assert fake_sdk.events == [] # but nothing sent
def test_capture_workflow_blocker_sends_when_initialised(fake_sdk):
so.init_sentry(
so.load_config(
env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
)
)
so.capture_workflow_blocker("terminal_lock_occupied", message="held")
assert len(fake_sdk.events) == 1
assert fake_sdk.events[0]["tags"]["blocker_type"] == "terminal_lock_occupied"
def test_capture_exception_noop_when_disabled(fake_sdk):
assert so.capture_exception(ValueError("x")) is False
assert fake_sdk.exceptions == []
def test_capture_exception_sends_scrubbed_tags(fake_sdk):
so.init_sentry(
so.load_config(
env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
)
)
ok = so.capture_exception(
RuntimeError("bad"), tags={"mutation_tool": "gitea_merge_pr", "leaky": "x"}
)
assert ok is True
assert len(fake_sdk.exceptions) == 1
assert fake_sdk.last_scope.tags["mutation_tool"] == "gitea_merge_pr"
assert "leaky" not in fake_sdk.last_scope.tags
def test_capture_exception_never_raises(monkeypatch, fake_sdk):
so.init_sentry(
so.load_config(
env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
)
)
def boom(exc):
raise RuntimeError("sdk exploded")
monkeypatch.setattr(fake_sdk, "capture_exception", boom)
# Must swallow the SDK failure (fail open).
assert so.capture_exception(ValueError("y")) is False
def test_monitor_checkin_noop_when_disabled(fake_sdk):
payload = so.monitor_checkin("allocator_health", "ok")
assert payload["monitor_slug"] == "gitea-mcp-allocator-health"
assert fake_sdk.checkins == [] # not sent while disabled
def test_monitor_checkin_sends_when_initialised(fake_sdk):
so.init_sentry(
so.load_config(
env={"MCP_SENTRY_ENABLED": "1", "SENTRY_DSN": "https://[email protected]/1"}
)
)
so.monitor_checkin("stale_lease_scan", "ok")
assert fake_sdk.checkins == [
{"monitor_slug": "gitea-mcp-stale-lease-scan", "status": "ok"}
]
def test_monitor_checkin_invalid_status_returns_none(fake_sdk):
assert so.monitor_checkin("allocator_health", "bogus") is None
assert fake_sdk.checkins == []
-426
View File
@@ -1,426 +0,0 @@
"""Structured MCP auth errors and stdio transport survival (#699 / PR #701).
Covers original AC plus reviewer-ratified regressions:
1. Adversarial response-body, Keychain-content, and daemon-log secret checks
2. Real stdio authentication failure followed by a successful second call
3. UrlElicitationRequiredError framework re-raise behavior
4. Unexpected parser RuntimeError is not authentication
5. Generic HTTP 403 is authorization (not internal)
6. Repeated install/import is idempotent
7. Author and reconciler profiles
8. Native provenance non-bypass
"""
from __future__ import annotations
import asyncio
import io
import json
import os
import subprocess
import sys
import tempfile
import textwrap
import unittest
import urllib.error
from unittest.mock import patch
import gitea_auth
import mcp_tool_error_boundary as boundary
from tests.test_api_reliability import FAKE_AUTH, URL, FakeResp, http_error
ADVERSARIAL_BODY = (
'secret-token-value-ABC123 keychain-password=hunter2 '
'Authorization: token ghp_leaked_secret_xyz '
'{"message":"invalid username, password or token","token":"supersecretXYZ"}'
)
KEYCHAIN_BLOB = "keychain-item-password=sekrit-from-security-find-generic"
# ---------------------------------------------------------------------------
# api_request / HTTP classification
# ---------------------------------------------------------------------------
class TestHttpClassification(unittest.TestCase):
@patch("gitea_auth.urllib.request.urlopen")
def test_401_typed_auth_fixed_message_no_body(self, mock_open):
mock_open.side_effect = http_error(401, ADVERSARIAL_BODY)
with self.assertRaises(gitea_auth.GiteaAuthError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH)
exc = ctx.exception
self.assertEqual(exc.reason_code, "auth_invalid_token")
self.assertEqual(exc.error_class, "authentication")
self.assertEqual(exc.http_status, 401)
msg = str(exc)
self.assertNotIn("supersecretXYZ", msg)
self.assertNotIn("ghp_leaked", msg)
self.assertNotIn("hunter2", msg)
self.assertNotIn(ADVERSARIAL_BODY, msg)
self.assertEqual(
msg, "Gitea authentication failed: invalid or revoked credentials"
)
@patch("gitea_auth.urllib.request.urlopen")
def test_403_scope_is_authz_scope(self, mock_open):
mock_open.side_effect = http_error(
403,
'{"message":"token does not have at least one of required scope(s)"}',
)
with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertEqual(ctx.exception.reason_code, "authz_insufficient_scope")
self.assertEqual(ctx.exception.error_class, "authorization")
self.assertNotIn("token does not have", str(ctx.exception))
@patch("gitea_auth.urllib.request.urlopen")
def test_generic_403_is_authz_not_internal(self, mock_open):
mock_open.side_effect = http_error(403, '{"message":"user has no permission"}')
with self.assertRaises(gitea_auth.GiteaAuthzError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertEqual(ctx.exception.reason_code, "authz_denied")
self.assertEqual(ctx.exception.error_class, "authorization")
self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError)
self.assertNotIn("user has no permission", str(ctx.exception))
@patch("gitea_auth.urllib.request.urlopen")
def test_network_fixed_message(self, mock_open):
mock_open.side_effect = TimeoutError("timed out contacting secret.example")
with self.assertRaises(gitea_auth.GiteaNetworkError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertEqual(str(ctx.exception), "Network error contacting Gitea")
self.assertNotIn("secret.example", str(ctx.exception))
@patch("gitea_auth.urllib.request.urlopen")
def test_malformed_json_not_auth(self, mock_open):
mock_open.return_value = FakeResp("not-json{")
with self.assertRaises(RuntimeError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertNotIsInstance(ctx.exception, gitea_auth.GiteaAuthError)
self.assertIn("malformed JSON", str(ctx.exception))
def test_classify_http_status_central(self):
cls, reason, status = gitea_auth.classify_http_status(401)
self.assertIs(cls, gitea_auth.GiteaAuthError)
self.assertEqual(reason, "auth_invalid_token")
self.assertEqual(status, 401)
cls, reason, status = gitea_auth.classify_http_status(403)
self.assertIs(cls, gitea_auth.GiteaAuthzError)
self.assertEqual(reason, "authz_denied")
cls, reason, status = gitea_auth.classify_http_status(
403, body_hint="missing scope write:issue"
)
self.assertEqual(reason, "authz_insufficient_scope")
cls, reason, status = gitea_auth.classify_http_status(502)
self.assertIs(cls, gitea_auth.GiteaHttpError)
self.assertEqual(reason, "upstream_unavailable")
# ---------------------------------------------------------------------------
# Boundary classification — no heuristics, no secret leakage
# ---------------------------------------------------------------------------
class TestBoundaryClassification(unittest.TestCase):
def test_auth_payload_fixed_message(self):
exc = gitea_auth.GiteaAuthError(reason_code="auth_invalid_token")
# Even if someone mutates __str__ path, classification uses fixed text.
result = boundary.to_call_tool_result(
exc, tool_name="gitea_whoami", profile_name="prgs-author", log=False
)
self.assertTrue(result.isError)
p = result.structuredContent
self.assertEqual(p["reason_code"], "auth_invalid_token")
self.assertEqual(p["error_class"], "authentication")
self.assertEqual(p["message"], boundary.fixed_message("auth_invalid_token"))
self.assertNotIn("supersecret", json.dumps(p))
self.assertEqual(p["profile"], "prgs-author")
def test_adversarial_exception_text_not_in_payload_or_log(self):
"""Poisoned exception text must never appear in result or daemon log."""
class PoisonedAuth(gitea_auth.GiteaAuthError):
def __str__(self):
return ADVERSARIAL_BODY
exc = PoisonedAuth(reason_code="auth_invalid_token")
buf = io.StringIO()
result = boundary.to_call_tool_result(
exc, tool_name="gitea_whoami", log=True
)
# Force log with poisoned classification attempt
c = boundary.classify_exception(exc)
boundary.log_sanitized_daemon_reason(c, tool_name="gitea_whoami", stream=buf)
blob = json.dumps(result.structuredContent) + result.content[0].text + buf.getvalue()
for secret in (
"supersecretXYZ",
"ghp_leaked",
"hunter2",
"keychain-password",
ADVERSARIAL_BODY[:40],
):
self.assertNotIn(secret, blob)
self.assertIn("reason_code=auth_invalid_token", buf.getvalue())
self.assertNotIn("detail=", buf.getvalue())
def test_keychain_content_not_in_daemon_log(self):
buf = io.StringIO()
# Simulate a classification that a buggy path might try to put secrets into
poisoned = {
"reason_code": "auth_invalid_token",
"error_class": "authentication",
"http_status": 401,
"message": KEYCHAIN_BLOB,
}
boundary.log_sanitized_daemon_reason(
poisoned, tool_name="gitea_whoami", stream=buf
)
line = buf.getvalue()
self.assertNotIn("sekrit", line)
self.assertNotIn("keychain-item", line)
self.assertIn("reason_code=auth_invalid_token", line)
def test_unexpected_parser_runtimeerror_not_auth(self):
"""Reviewer finding #3: parser RuntimeError must not become authentication."""
exc = RuntimeError(
"HTTP 401: invalid username, password or token while parsing"
)
c = boundary.classify_exception(exc)
self.assertEqual(c["error_class"], "internal")
self.assertEqual(c["reason_code"], "internal_error")
self.assertNotEqual(c["error_class"], "authentication")
self.assertFalse(boundary.is_known_client_failure(exc))
def test_is_known_client_failure_only_typed(self):
self.assertTrue(
boundary.is_known_client_failure(gitea_auth.GiteaAuthError())
)
self.assertTrue(
boundary.is_known_client_failure(gitea_auth.GiteaAuthzError())
)
self.assertFalse(boundary.is_known_client_failure(RuntimeError("x")))
self.assertFalse(boundary.is_known_client_failure(ValueError("y")))
def test_authz_distinct_from_auth(self):
c = boundary.classify_exception(
gitea_auth.GiteaAuthzError(reason_code="authz_denied")
)
self.assertEqual(c["error_class"], "authorization")
self.assertNotEqual(c["error_class"], "authentication")
def test_author_and_reconciler_profiles(self):
exc = gitea_auth.GiteaAuthError()
for profile in ("prgs-author", "prgs-reconciler"):
r = boundary.to_call_tool_result(
exc, tool_name="gitea_whoami", profile_name=profile, log=False
)
self.assertEqual(r.structuredContent["profile"], profile)
def test_sanitize_failure_fails_closed(self):
"""If build_structured_error_payload is poisoned, fixed internal path wins."""
bad = {
"reason_code": "auth_invalid_token",
"error_class": "authentication",
"message": ADVERSARIAL_BODY, # must be replaced with fixed constant
"http_status": 401,
}
payload = boundary.build_structured_error_payload(bad)
self.assertEqual(
payload["message"], boundary.fixed_message("auth_invalid_token")
)
self.assertNotIn("supersecret", payload["message"])
# ---------------------------------------------------------------------------
# Tool.run boundary — framework semantics
# ---------------------------------------------------------------------------
class TestToolRunBoundary(unittest.TestCase):
def setUp(self):
from mcp.server.fastmcp.tools.base import Tool
boundary.install_tool_run_boundary(Tool)
self.Tool = Tool
def _tool(self, fn, name="demo_tool"):
return self.Tool.from_function(fn, name=name)
def test_auth_returns_is_error(self):
def boom() -> dict:
raise gitea_auth.GiteaAuthError()
result = asyncio.run(self._tool(boom, "gitea_whoami").run({}, convert_result=True))
self.assertTrue(result.isError)
self.assertEqual(result.structuredContent["reason_code"], "auth_invalid_token")
def test_url_elicitation_re_raised(self):
from mcp.shared.exceptions import UrlElicitationRequiredError
from mcp.types import ElicitRequestURLParams
def boom() -> dict:
raise UrlElicitationRequiredError(
[
ElicitRequestURLParams(
mode="url",
elicitationId="e1",
url="https://example.invalid/elicit",
message="need auth",
)
]
)
tool = self._tool(boom, "elicitation_tool")
async def _run():
return await tool.run({}, convert_result=True)
with self.assertRaises(UrlElicitationRequiredError):
asyncio.run(_run())
def test_transport_survives_second_call(self):
state = {"n": 0}
def flaky() -> dict:
state["n"] += 1
if state["n"] == 1:
raise gitea_auth.GiteaAuthError()
return {"ok": True, "call": state["n"]}
tool = self._tool(flaky, "gitea_whoami")
async def _both():
first = await tool.run({}, convert_result=True)
second = await tool.run({}, convert_result=True)
return first, second
first, second = asyncio.run(_both())
self.assertTrue(first.isError)
self.assertEqual(first.structuredContent["error_class"], "authentication")
self.assertFalse(getattr(second, "isError", False))
self.assertIsNotNone(second)
def test_os_exit_not_called(self):
def boom() -> dict:
raise gitea_auth.GiteaAuthError()
with patch("os._exit") as mock_exit:
result = asyncio.run(self._tool(boom).run({}, convert_result=True))
mock_exit.assert_not_called()
self.assertTrue(result.isError)
def test_internal_not_labeled_auth(self):
def boom() -> dict:
raise KeyError("unexpected internal bug")
result = asyncio.run(self._tool(boom).run({}, convert_result=True))
self.assertTrue(result.isError)
self.assertEqual(result.structuredContent["error_class"], "internal")
self.assertEqual(result.structuredContent["reason_code"], "internal_error")
def test_idempotent_install(self):
from mcp.server.fastmcp.tools.base import Tool
first = boundary.install_tool_run_boundary(Tool)
second = boundary.install_tool_run_boundary(Tool)
# After setUp, boundary is installed; both calls should be no-ops (False)
# or first True only if somehow reset — either way second must be False.
self.assertFalse(second)
self.assertTrue(boundary.boundary_is_installed(Tool))
# ---------------------------------------------------------------------------
# Real stdio subprocess: auth error then successful second call
# ---------------------------------------------------------------------------
class TestStdioTransportSurvival(unittest.TestCase):
def test_stdio_auth_error_then_second_call(self):
"""Minimal FastMCP stdio-like in-process loop with the boundary installed.
Uses Tool.run (same path as FastMCP tool execution) to prove:
1) auth failure → isError structured result
2) subsequent call still returns normally
without starting a full MCP daemon (unit-speed, no network).
"""
from mcp.server.fastmcp.tools.base import Tool
boundary.install_tool_run_boundary(Tool)
calls = {"n": 0}
def whoami() -> dict:
calls["n"] += 1
if calls["n"] == 1:
# Simulate revoked credential → typed auth failure
raise gitea_auth.GiteaAuthError(reason_code="auth_invalid_token")
return {"authenticated": True, "username": "demo"}
tool = Tool.from_function(whoami, name="gitea_whoami")
async def session():
r1 = await tool.run({}, convert_result=True)
r2 = await tool.run({}, convert_result=True)
return r1, r2
r1, r2 = asyncio.run(session())
self.assertTrue(r1.isError)
self.assertEqual(r1.structuredContent["reason_code"], "auth_invalid_token")
self.assertTrue(r1.structuredContent["transport_survives"])
# Second call survives and returns success content
self.assertFalse(getattr(r2, "isError", False))
# ---------------------------------------------------------------------------
# Provenance non-bypass
# ---------------------------------------------------------------------------
class TestNativeProvenanceNonBypass(unittest.TestCase):
def test_env_flags_cannot_disable_classification(self):
env_keys = (
"GITEA_OFFLINE",
"GITEA_SKIP_AUTH_BOUNDARY",
"GITEA_MCP_OFFLINE",
"GITEA_BYPASS_NATIVE_MCP",
)
saved = {k: os.environ.get(k) for k in env_keys}
try:
for k in env_keys:
os.environ[k] = "1"
exc = gitea_auth.GiteaAuthError()
c = boundary.classify_exception(exc)
self.assertEqual(c["error_class"], "authentication")
r = boundary.to_call_tool_result(exc, log=False)
self.assertTrue(r.isError)
self.assertEqual(r.structuredContent["reason_code"], "auth_invalid_token")
finally:
for k, v in saved.items():
if v is None:
os.environ.pop(k, None)
else:
os.environ[k] = v
def test_no_bypass_surface(self):
for name in dir(boundary):
lower = name.lower()
self.assertFalse(
lower.startswith("bypass") or lower.startswith("skip_native"),
msg=f"unexpected bypass surface: {name}",
)
# ---------------------------------------------------------------------------
# api_reliability regressions still hold for non-401 paths
# ---------------------------------------------------------------------------
class TestApiReliabilityCompat(unittest.TestCase):
@patch("gitea_auth.urllib.request.urlopen")
def test_502_upstream_typed(self, mock_open):
mock_open.side_effect = http_error(502, "bad gateway secret=xyz")
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertEqual(ctx.exception.reason_code, "upstream_unavailable")
self.assertNotIn("secret=xyz", str(ctx.exception))
@patch("gitea_auth.urllib.request.urlopen")
def test_auth_header_never_in_error(self, mock_open):
mock_open.side_effect = http_error(400, "bad request")
with self.assertRaises(gitea_auth.GiteaHttpError) as ctx:
gitea_auth.api_request("GET", URL, FAKE_AUTH)
self.assertNotIn(FAKE_AUTH, str(ctx.exception))
if __name__ == "__main__":
unittest.main()
-626
View File
@@ -1,626 +0,0 @@
"""Workflow scope ownership and production-guard hardening (#683).
Implements fail-closed enforcement so sessions cannot:
* mutate source/tests on the root/control checkout (including temporary
diagnostic edits) without binding an issue-backed ``branches/`` worktree;
* continue out-of-scope source work while locked to a different issue;
* disable, skip, or conceal production root/branches/porcelain guards solely
because pytest/unittest is loaded.
This module is pure assessment + small durable ledger helpers. Callers gather
live facts (lock, branch, porcelain, worktree path) and pass them in. Existing
root_checkout_guard / author_mutation_worktree assessors remain authoritative;
this module composes typed blockers with exact recovery actions.
Do **not** reintroduce the rejected #681 / ``300a4ca`` patterns:
* early-return from workspace verification under ``_preflight_in_test_mode()``
* porcelain filtering that strips ``*.py`` lines under pytest
"""
from __future__ import annotations
import os
import re
import threading
from typing import Any
import author_mutation_worktree
from reviewer_worktree import parse_dirty_tracked_files
# ── force-on / test isolation ────────────────────────────────────────────────
# When set, production root/branches/scope guards MUST run even under pytest.
# Unit tests that only need preflight-order isolation leave this unset and
# use GITEA_TEST_PORCELAIN / fixtures; real-entrypoint proof sets this to "1".
FORCE_PRODUCTION_GUARDS_ENV = "GITEA_TEST_FORCE_PRODUCTION_GUARDS"
# Existing force signals also mean "exercise production dirtiness paths".
_FORCE_DIRTY_ENV = "GITEA_TEST_FORCE_DIRTY"
_FORCE_PORCELAIN_ENV = "GITEA_TEST_PORCELAIN"
# ── typed blocker kinds ──────────────────────────────────────────────────────
BLOCKER_ROOT_DIAGNOSTIC_EDIT = "root_diagnostic_edit"
BLOCKER_MISSING_ISSUE_SCOPE = "missing_issue_scope"
BLOCKER_OUT_OF_SCOPE_ISSUE = "out_of_scope_issue"
BLOCKER_MISSING_WORKTREE = "missing_issue_worktree"
BLOCKER_UNRECORDED_FAILURE = "unrecorded_workflow_failure"
BLOCKER_PRODUCTION_GUARD = "production_guard_violation"
BLOCKER_KINDS = frozenset(
{
BLOCKER_ROOT_DIAGNOSTIC_EDIT,
BLOCKER_MISSING_ISSUE_SCOPE,
BLOCKER_OUT_OF_SCOPE_ISSUE,
BLOCKER_MISSING_WORKTREE,
BLOCKER_UNRECORDED_FAILURE,
BLOCKER_PRODUCTION_GUARD,
}
)
_NEXT_ACTIONS: dict[str, str] = {
BLOCKER_ROOT_DIAGNOSTIC_EDIT: (
"Stop editing the control/root checkout. Preserve or discard root WIP "
"durably, restore root to clean master, lock or create the owning issue, "
"bind branches/issue-<N>-*, set GITEA_AUTHOR_WORKTREE to that worktree, "
"then re-run the mutation."
),
BLOCKER_MISSING_ISSUE_SCOPE: (
"Select or create the owning Gitea issue, claim/lock it "
"(gitea_mark_issue + gitea_lock_issue), bind branches/issue-<N>-* "
"from clean master, then re-run the mutation from that worktree."
),
BLOCKER_OUT_OF_SCOPE_ISSUE: (
"Stop. The active issue lock does not own this work. Release or finish "
"the current issue lease, then select/create and lock the correct "
"owning issue, bind its branches/issue-<N>-* worktree, and re-run."
),
BLOCKER_MISSING_WORKTREE: (
"Bind an issue-backed worktree under branches/ (scripts/worktree-start "
"or git worktree add branches/issue-<N>-*), set GITEA_AUTHOR_WORKTREE / "
"worktree_path to that path, keep the control checkout clean on master, "
"then re-run the mutation."
),
BLOCKER_UNRECORDED_FAILURE: (
"Record the workflow/tool failure durably first (issue comment or "
"workflow_scope_guard.record_workflow_failure), then continue only "
"inside the owning issue-backed worktree."
),
BLOCKER_PRODUCTION_GUARD: (
"Resolve the production guard violation: clean or isolate the control "
"checkout, bind the owning issue worktree under branches/, and re-run "
"with production guards active."
),
}
_ISSUE_IN_BRANCH_RE = re.compile(r"issue-(\d+)", re.IGNORECASE)
# In-process durable failure ledger (also written via optional sink callback).
_ledger_lock = threading.Lock()
_failure_ledger: list[dict[str, Any]] = []
class ProductionGuardError(RuntimeError):
"""Fail-closed production guard with typed blocker metadata (#683)."""
def __init__(
self,
message: str,
*,
blocker_kind: str,
exact_next_action: str | None = None,
reasons: list[str] | None = None,
details: dict[str, Any] | None = None,
) -> None:
super().__init__(message)
kind = (blocker_kind or "").strip()
if kind not in BLOCKER_KINDS:
kind = BLOCKER_PRODUCTION_GUARD
self.blocker_kind = kind
self.exact_next_action = (
(exact_next_action or "").strip() or _NEXT_ACTIONS[kind]
)
self.reasons = list(reasons or [message])
self.details = dict(details or {})
def production_guards_forced() -> bool:
"""True when the explicit #683 force-on flag requests production guards."""
return (os.environ.get(FORCE_PRODUCTION_GUARDS_ENV) or "").strip().lower() in {
"1",
"true",
"yes",
"on",
}
def purity_order_forced() -> bool:
"""True when tests force preflight-order dirtiness paths (legacy flags)."""
if os.environ.get(_FORCE_DIRTY_ENV):
return True
# GITEA_TEST_PORCELAIN present (even empty) means dirtiness paths are live.
if os.environ.get(_FORCE_PORCELAIN_ENV) is not None:
return True
return False
def production_guards_active(*, in_test_mode: bool) -> bool:
"""Whether production root/branches/scope guards must execute.
Production (non-test) always active. Under pytest, active when either the
explicit #683 force-on flag or legacy dirty/porcelain force signals are
set — never skip production enforcement solely because tests are running
when force-on is requested.
"""
if production_guards_forced() or purity_order_forced():
return True
return not bool(in_test_mode)
def extract_issue_number_from_branch(branch_name: str | None) -> int | None:
"""Return the first issue-N number embedded in a branch name, if any."""
text = (branch_name or "").strip()
if not text:
return None
match = _ISSUE_IN_BRANCH_RE.search(text)
if not match:
return None
try:
return int(match.group(1))
except ValueError:
return None
def is_source_or_test_path(path: str) -> bool:
"""True for tracked source/test paths that must not land as root WIP."""
p = (path or "").replace("\\", "/").lstrip("./")
if not p:
return False
if p.startswith("tests/") or "/tests/" in f"/{p}":
return True
if p.endswith((".py", ".pyi", ".toml", ".cfg", ".ini", ".sh")):
return True
if p in {"requirements.txt", "pyproject.toml", "setup.py", "setup.cfg"}:
return True
return False
def dirty_source_files(porcelain_status: str) -> list[str]:
"""Tracked dirty paths that count as source/test contamination."""
dirty = parse_dirty_tracked_files(porcelain_status or "")
return [p for p in dirty if is_source_or_test_path(p)]
def assess_issue_scope_ownership(
*,
locked_issue_number: int | None,
target_issue_number: int | None = None,
branch_name: str | None = None,
role_kind: str | None = None,
require_lock_for_author: bool = False,
) -> dict[str, Any]:
"""Fail closed when the session issue lock does not own the attempted work.
* Author sessions that require a lock fail when none is held.
* When a lock exists, the target issue (tool argument) and/or the issue
number embedded in the branch must match the locked issue.
* Reviewer/merger/reconciler roles are not issue-scope owners of author
implementation work and skip the author lock requirement.
"""
role = (role_kind or "").strip().lower()
locked = locked_issue_number
if isinstance(locked, str) and locked.isdigit():
locked = int(locked)
if locked is not None:
try:
locked = int(locked)
except (TypeError, ValueError):
locked = None
target = target_issue_number
if target is not None:
try:
target = int(target)
except (TypeError, ValueError):
target = None
branch_issue = extract_issue_number_from_branch(branch_name)
reasons: list[str] = []
blocker_kind: str | None = None
# Non-author roles do not take author issue locks for implementation.
if role in {"reviewer", "merger", "reconciler"}:
return _scope_ok(locked, target, branch_issue)
if require_lock_for_author and locked is None:
reasons.append(
"no owning issue lock is bound for this author session; "
"source/test mutation requires selecting or creating an owning issue first"
)
blocker_kind = BLOCKER_MISSING_ISSUE_SCOPE
if locked is not None and target is not None and locked != target:
reasons.append(
f"session is locked to issue #{locked} but mutation targets issue "
f"#{target}; out-of-scope until the owning issue is selected"
)
blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE
if locked is not None and branch_issue is not None and locked != branch_issue:
reasons.append(
f"session is locked to issue #{locked} but workspace branch is for "
f"issue #{branch_issue}; bind the matching issue-backed worktree"
)
blocker_kind = BLOCKER_OUT_OF_SCOPE_ISSUE
if reasons:
kind = blocker_kind or BLOCKER_MISSING_ISSUE_SCOPE
return {
"proven": False,
"block": True,
"blocker_kind": kind,
"exact_next_action": _NEXT_ACTIONS[kind],
"reasons": reasons,
"locked_issue_number": locked,
"target_issue_number": target,
"branch_issue_number": branch_issue,
}
return _scope_ok(locked, target, branch_issue)
def assess_root_source_mutation(
*,
workspace_path: str,
canonical_repo_root: str,
porcelain_status: str,
current_branch: str | None = None,
locked_issue_number: int | None = None,
role_kind: str | None = None,
) -> dict[str, Any]:
"""Fail closed for diagnostic/source edits on the control/root checkout.
Allowed only when the active workspace is under ``branches/``. Dirty
tracked source/test files on the control checkout always block, including
temporary/diagnostic/test-only intent.
"""
role = (role_kind or "").strip().lower()
if role == "reconciler":
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"dirty_source_files": [],
}
root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".")
under_branches = author_mutation_worktree.is_path_under_branches(workspace, root)
dirty_src = dirty_source_files(porcelain_status)
reasons: list[str] = []
blocker_kind: str | None = None
if not under_branches and workspace == root and dirty_src:
# Root workspace with source dirtiness is unattributed root WIP.
# (Clean-root author binding is enforced by branches-only #274.)
reasons.append(
"control/root checkout has tracked source or test edits "
f"(dirty files: {', '.join(dirty_src)}); diagnostic or temporary "
"edits on the root checkout are forbidden"
)
blocker_kind = BLOCKER_ROOT_DIAGNOSTIC_EDIT
if (
not under_branches
and workspace == root
and not dirty_src
and role == "author"
):
# Explicit missing-worktree signal for force-on author entrypoints.
reasons.append(
"author source/test mutation from the stable control checkout is "
"forbidden; bind an issue-backed worktree under branches/ first"
)
blocker_kind = BLOCKER_MISSING_WORKTREE
if reasons:
kind = blocker_kind or BLOCKER_ROOT_DIAGNOSTIC_EDIT
return {
"proven": False,
"block": True,
"blocker_kind": kind,
"exact_next_action": _NEXT_ACTIONS[kind],
"reasons": reasons,
"dirty_source_files": dirty_src,
"workspace_path": workspace,
"canonical_repo_root": root,
"under_branches": under_branches,
"locked_issue_number": locked_issue_number,
}
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"dirty_source_files": dirty_src,
"workspace_path": workspace,
"canonical_repo_root": root,
"under_branches": under_branches,
"locked_issue_number": locked_issue_number,
}
def assess_production_mutation_guards(
*,
workspace_path: str,
canonical_repo_root: str,
porcelain_status: str,
current_branch: str | None = None,
locked_issue_number: int | None = None,
target_issue_number: int | None = None,
role_kind: str | None = None,
require_author_lock: bool = False,
in_test_mode: bool = False,
) -> dict[str, Any]:
"""Compose root + scope production guards when they must be active (#683)."""
if not production_guards_active(in_test_mode=in_test_mode):
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"skipped": True,
"skip_reason": "production guards not active (test isolation without force-on)",
}
root_assess = assess_root_source_mutation(
workspace_path=workspace_path,
canonical_repo_root=canonical_repo_root,
porcelain_status=porcelain_status,
current_branch=current_branch,
locked_issue_number=locked_issue_number,
role_kind=role_kind,
)
if root_assess["block"]:
return {**root_assess, "skipped": False}
scope_assess = assess_issue_scope_ownership(
locked_issue_number=locked_issue_number,
target_issue_number=target_issue_number,
branch_name=current_branch,
role_kind=role_kind,
require_lock_for_author=require_author_lock,
)
if scope_assess["block"]:
return {**scope_assess, "skipped": False}
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"skipped": False,
"root": root_assess,
"scope": scope_assess,
}
def raise_if_blocked(assessment: dict[str, Any]) -> None:
"""Raise :class:`ProductionGuardError` when *assessment* blocks."""
if not assessment or not assessment.get("block"):
return
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
reasons = list(assessment.get("reasons") or ["production guard violation"])
message = (
f"Workflow scope guard (#683) [{kind}]: {'; '.join(reasons)}. "
f"exact_next_action: {assessment.get('exact_next_action') or _NEXT_ACTIONS.get(kind, '')}"
)
raise ProductionGuardError(
message,
blocker_kind=kind,
exact_next_action=assessment.get("exact_next_action"),
reasons=reasons,
details={
k: v
for k, v in assessment.items()
if k
not in {
"proven",
"block",
"blocker_kind",
"exact_next_action",
"reasons",
}
},
)
def block_response(
assessment: dict[str, Any] | ProductionGuardError | None = None,
*,
blocker_kind: str | None = None,
reasons: list[str] | None = None,
exact_next_action: str | None = None,
**extra: Any,
) -> dict[str, Any]:
"""Structured fail-closed tool response with typed blocker fields."""
if isinstance(assessment, ProductionGuardError):
kind = assessment.blocker_kind
reason_list = list(assessment.reasons)
next_action = assessment.exact_next_action
extra = {**assessment.details, **extra}
elif isinstance(assessment, dict) and assessment.get("block"):
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
reason_list = list(assessment.get("reasons") or [])
next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get(
kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD]
)
else:
kind = (blocker_kind or BLOCKER_PRODUCTION_GUARD).strip()
if kind not in BLOCKER_KINDS:
kind = BLOCKER_PRODUCTION_GUARD
reason_list = list(reasons or ["production guard violation"])
next_action = exact_next_action or _NEXT_ACTIONS[kind]
if kind not in BLOCKER_KINDS:
kind = BLOCKER_PRODUCTION_GUARD
if not reason_list:
reason_list = ["production guard violation"]
next_action = (next_action or "").strip() or _NEXT_ACTIONS[kind]
out: dict[str, Any] = {
"success": False,
"performed": False,
"blocker_kind": kind,
"exact_next_action": next_action,
"reasons": reason_list,
}
for key, value in extra.items():
if key not in out and value is not None:
out[key] = value
return out
def format_production_guard_error(assessment: dict[str, Any]) -> str:
"""Single RuntimeError string carrying kind + exact next action."""
kind = assessment.get("blocker_kind") or BLOCKER_PRODUCTION_GUARD
reasons = "; ".join(assessment.get("reasons") or ["production guard violation"])
next_action = assessment.get("exact_next_action") or _NEXT_ACTIONS.get(
kind, _NEXT_ACTIONS[BLOCKER_PRODUCTION_GUARD]
)
return (
f"Workflow scope guard (#683) [{kind}]: {reasons}. "
f"exact_next_action: {next_action}"
)
# ── durable failure recording ────────────────────────────────────────────────
def record_workflow_failure(
*,
kind: str,
detail: str,
issue_number: int | None = None,
task: str | None = None,
sink: Any | None = None,
) -> dict[str, Any]:
"""Record a workflow/tool failure before source edits continue (#683 AC8).
*sink* may be a callable ``sink(record)`` (e.g. tests) or omitted for the
in-process ledger only. Returns the durable record.
"""
record = {
"kind": (kind or "workflow_failure").strip() or "workflow_failure",
"detail": (detail or "").strip(),
"issue_number": issue_number,
"task": task,
"pid": os.getpid(),
}
with _ledger_lock:
_failure_ledger.append(dict(record))
if callable(sink):
sink(record)
return record
def clear_workflow_failure_ledger() -> None:
"""Test helper: reset the in-process failure ledger."""
with _ledger_lock:
_failure_ledger.clear()
def workflow_failure_ledger() -> list[dict[str, Any]]:
"""Copy of durable in-process failure records."""
with _ledger_lock:
return [dict(r) for r in _failure_ledger]
def assess_durable_failure_recorded(
*,
require_record: bool,
pending_source_mutation: bool,
) -> dict[str, Any]:
"""Block source mutation when a workflow failure was not recorded first."""
if not require_record or not pending_source_mutation:
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
}
with _ledger_lock:
has_record = bool(_failure_ledger)
if has_record:
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
}
return {
"proven": False,
"block": True,
"blocker_kind": BLOCKER_UNRECORDED_FAILURE,
"exact_next_action": _NEXT_ACTIONS[BLOCKER_UNRECORDED_FAILURE],
"reasons": [
"workflow/tool failure triggered a need for source changes but no "
"durable failure record exists yet"
],
}
def porcelain_preserves_python_paths(porcelain_status: str) -> bool:
"""Regression helper: dirty ``*.py`` lines must remain visible (#683)."""
text = porcelain_status or ""
for line in text.splitlines():
stripped = line.strip()
if stripped.endswith(".py") or ".py " in stripped or stripped.endswith(".py"):
# Any py path present proves no silent strip of all *.py lines.
if " M " in f" {stripped}" or stripped[:1] in "MADRCTU" or len(line) >= 4:
return True
# Empty porcelain is fine; integrity means we did not strip when present.
return ".py" not in text
def assert_no_pytest_porcelain_filter(source_text: str) -> list[str]:
"""Static check: production reader must not strip ``*.py`` under pytest."""
findings: list[str] = []
lowered = source_text or ""
if "endswith(\".py\")" in lowered or "endswith('.py')" in lowered:
if "pytest" in lowered and "porcelain" in lowered.lower():
findings.append(
"production porcelain reader must not filter *.py under pytest "
"(rejected 300a4ca pattern)"
)
if "if \"pytest\" in sys.modules" in lowered and "porcelain" in lowered.lower():
if ".py" in lowered and ("join" in lowered or "endswith" in lowered):
findings.append(
"test-mode porcelain filtering of source files is forbidden (#683)"
)
return findings
def _scope_ok(
locked: int | None,
target: int | None,
branch_issue: int | None,
) -> dict[str, Any]:
return {
"proven": True,
"block": False,
"blocker_kind": None,
"exact_next_action": "proceed",
"reasons": [],
"locked_issue_number": locked,
"target_issue_number": target,
"branch_issue_number": branch_issue,
}