Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
78cc37a977 |
@@ -39,6 +39,27 @@ GITEA_AUDIT_LOG=/path/to/gitea-mcp-audit.log
|
||||
# only — never the token value. Surfaced by gitea_get_profile.
|
||||
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.0–1.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,
|
||||
# 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
|
||||
|
||||
@@ -386,27 +386,6 @@ def assess_canonical_comment(
|
||||
if not related or related.lower() in {"none", "n/a", "-"}:
|
||||
missing.append("RELATED_PRS")
|
||||
|
||||
# #695 AC7: untrusted / offline approval claims cannot certify merger handoff.
|
||||
try:
|
||||
import review_quarantine as _rq
|
||||
|
||||
for claim_reason in _rq.assess_untrusted_canonical_approval_claim(text):
|
||||
extra.append(claim_reason)
|
||||
except Exception:
|
||||
# Fail closed on import/runtime errors for approval-shaped claims only.
|
||||
lower = text.lower()
|
||||
if (
|
||||
"state:\napproved" in lower
|
||||
or "who_is_next:\nmerger" in lower
|
||||
or "merge_ready: true" in lower
|
||||
or "merge_ready:\ntrue" in lower
|
||||
or "ready-to-merge" in lower
|
||||
):
|
||||
extra.append(
|
||||
"canonical approval/merge-ready claim could not be verified "
|
||||
"for native review proof (#695 AC7; fail closed)"
|
||||
)
|
||||
|
||||
allowed = not missing and not vague and not extra
|
||||
correction = ""
|
||||
if not allowed:
|
||||
|
||||
@@ -1,77 +1,23 @@
|
||||
# MCP daemon import and native-transport guard (#558 / #695)
|
||||
# MCP daemon import and keychain guard (#558)
|
||||
|
||||
## Problem
|
||||
|
||||
During deadlock debugging and the PR #694 incident (#695), agents imported
|
||||
`gitea_mcp_server` or ran credential helpers from a raw shell / offline helper,
|
||||
bypassing native MCP transport, preflight purity, and role gates. Contaminated
|
||||
formal reviews then looked identical to native approvals.
|
||||
During deadlock debugging, agents imported `gitea_mcp_server` / ran credential
|
||||
helpers from a raw shell, bypassing preflight purity and role gates.
|
||||
|
||||
## Rule
|
||||
|
||||
Mutation auth, keychain fill, and controller quarantine require a **native MCP
|
||||
transport runtime** established only by the official entrypoint.
|
||||
Mutation auth and keychain fill require a **sanctioned MCP daemon** process.
|
||||
|
||||
| Context | Allowed |
|
||||
|---------|---------|
|
||||
| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) calls `mark_sanctioned_daemon()` and holds a process-local runtime token | yes |
|
||||
| pytest (hermetic unit tests) | yes |
|
||||
| `GITEA_MCP_SANCTIONED_DAEMON=1` alone (no process-local native runtime) | **no** (#695) |
|
||||
| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set in agent sessions |
|
||||
| `GITEA_ALLOW_KEYCHAIN_CLI=1` in LLM sessions | **no** — human operator only |
|
||||
| bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** |
|
||||
| keychain fill outside native/pytest | **no** |
|
||||
|
||||
Native runtime is **process-local**: a random token bound to the daemon PID.
|
||||
It is never reconstructed from environment variables, session-state files, or
|
||||
importing internals in a fresh Python process.
|
||||
|
||||
## Contaminated review quarantine (#695 AC8)
|
||||
|
||||
Controller/reconciler/merger profiles may call
|
||||
`gitea_quarantine_contaminated_review` with explicit confirmation:
|
||||
|
||||
```text
|
||||
QUARANTINE CONTAMINATED REVIEW <review_id> PR <pr_number>
|
||||
```
|
||||
|
||||
Quarantine records are durable under the MCP session-state root and are
|
||||
**honored** by:
|
||||
|
||||
- `gitea_get_pr_review_feedback` (quarantined approvals do not authorize merge)
|
||||
- `gitea_check_pr_eligibility` action=`merge`
|
||||
- `gitea_merge_pr` (mutation)
|
||||
- merger lease adoption paths that read `approval_at_current_head`
|
||||
|
||||
Forensic Gitea reviews and historical comments are **never deleted**.
|
||||
|
||||
## STOP after native MCP failure (AC10)
|
||||
|
||||
If the native MCP namespace dies (EOF, capability disconnect, session death):
|
||||
|
||||
1. **STOP.** State BLOCKED + DIAGNOSE.
|
||||
2. Do **not** import `gitea_mcp_server` from a standalone process.
|
||||
3. Do **not** run `offline_mcp_helper.py`, `offline_mcp_runner.py`,
|
||||
`run_quarantine.py`, or any offline mutation helper.
|
||||
4. Do **not** set direct-import, keychain-bypass, or raw-token environment
|
||||
variables.
|
||||
5. Reconnect / restart the official MCP daemon; resume only via native tools.
|
||||
|
||||
Any further native MCP failure is a hard stop. Do not construct another fallback.
|
||||
|
||||
## Canonical approval claims (AC7)
|
||||
|
||||
Comments that claim `approved` / `ready-to-merge` / `WHO_IS_NEXT: merger` /
|
||||
`MERGE_READY: true` must include:
|
||||
|
||||
```text
|
||||
NATIVE_REVIEW_PROOF: transport=native_mcp; …
|
||||
```
|
||||
|
||||
Claims that cite offline/import helpers are rejected even if a proof line is
|
||||
present.
|
||||
| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) sets `GITEA_MCP_SANCTIONED_DAEMON=1` | yes |
|
||||
| pytest | yes |
|
||||
| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` (operator/tests only) | yes |
|
||||
| bare `python -c 'import gitea_auth; get_auth_header(...)'` | **no** |
|
||||
| keychain fill without daemon | **no** unless `GITEA_ALLOW_KEYCHAIN_CLI=1` |
|
||||
|
||||
## Operator note
|
||||
|
||||
LLM sessions must never set allow-direct-import, allow-keychain-cli, or raw
|
||||
token overrides. Those are human-only escape hatches outside agent workflows.
|
||||
LLM sessions must never set the allow-direct-import or allow-keychain-cli
|
||||
overrides. Those are human-only escape hatches.
|
||||
|
||||
@@ -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.0–1.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.
|
||||
+150
-954
File diff suppressed because it is too large
Load Diff
+33
-156
@@ -1,153 +1,67 @@
|
||||
"""Sanctioned MCP daemon guards for imports and credential access (#558 / #695).
|
||||
"""Sanctioned MCP daemon guards for imports and credential access (#558).
|
||||
|
||||
Direct ``import gitea_mcp_server`` from a shell bypasses native MCP transport.
|
||||
#558 introduced a daemon marker; #695 hardens it so:
|
||||
Direct ``import gitea_mcp_server`` / ``import gitea_auth`` from a shell, plus
|
||||
raw keychain dumps, bypass preflight purity and role gates. Mutation helpers
|
||||
and keychain fallbacks therefore require an explicit sanctioned runtime.
|
||||
|
||||
- Environment variables alone cannot reconstruct a native session
|
||||
(``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are
|
||||
insufficient for mutation gates).
|
||||
- A process-local runtime record is created only by the official entrypoint
|
||||
(``mcp_server.py`` calling ``mark_sanctioned_daemon``), bound to PID and a
|
||||
random secret that never leaves process memory.
|
||||
- Offline scripts that import internals fail closed on mutations.
|
||||
- Pytest remains allowed (hermetic tests); optional force flags exist for
|
||||
provenance regression tests.
|
||||
Sanctioned contexts (any one):
|
||||
- ``GITEA_MCP_SANCTIONED_DAEMON=1`` (set by the official MCP entrypoint)
|
||||
- pytest (``PYTEST_CURRENT_TEST`` present)
|
||||
- explicit operator opt-in ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` (tests/tools only)
|
||||
|
||||
Manual deletion of session-state files is never a recovery path.
|
||||
Credential keychain fill additionally allows:
|
||||
- ``GITEA_ALLOW_KEYCHAIN_CLI=1`` for operator-only non-MCP scripts that must
|
||||
use git-credential (never the default for LLM shells).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import inspect
|
||||
import os
|
||||
import secrets
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON"
|
||||
ALLOW_DIRECT_IMPORT_ENV = "GITEA_ALLOW_DIRECT_MCP_IMPORT"
|
||||
ALLOW_KEYCHAIN_CLI_ENV = "GITEA_ALLOW_KEYCHAIN_CLI"
|
||||
# Test-only: force provenance failure even under pytest (#695 regressions).
|
||||
FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED"
|
||||
|
||||
# Process-local native runtime (never persisted, never read from env alone).
|
||||
_NATIVE_RUNTIME: dict[str, Any] | None = None
|
||||
|
||||
|
||||
class UnsanctionedRuntimeError(RuntimeError):
|
||||
"""Raised when mutation/credential code runs outside a native MCP daemon."""
|
||||
"""Raised when mutation/credential code runs outside a sanctioned MCP daemon."""
|
||||
|
||||
|
||||
def is_pytest_runtime() -> bool:
|
||||
if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}:
|
||||
if os.environ.get("GITEA_TEST_FORCE_UNSANCTIONED") == "1":
|
||||
return False
|
||||
import sys
|
||||
|
||||
if "pytest" in sys.modules:
|
||||
return True
|
||||
return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())
|
||||
|
||||
|
||||
def _caller_is_official_entrypoint() -> bool:
|
||||
"""True when mark_sanctioned_daemon is invoked from mcp_server.py."""
|
||||
for frame in inspect.stack()[1:12]:
|
||||
path = (frame.filename or "").replace("\\", "/")
|
||||
base = path.rsplit("/", 1)[-1]
|
||||
if base == "mcp_server.py":
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def mark_sanctioned_daemon(*, allow_test_bootstrap: bool = False) -> dict[str, Any]:
|
||||
"""Mark this process as the official native MCP daemon (#695).
|
||||
|
||||
Only the official ``mcp_server.py`` entrypoint (or pytest test bootstrap)
|
||||
may establish native transport. Setting env vars alone is insufficient.
|
||||
"""
|
||||
global _NATIVE_RUNTIME
|
||||
if not is_pytest_runtime() and not allow_test_bootstrap:
|
||||
if not _caller_is_official_entrypoint():
|
||||
raise UnsanctionedRuntimeError(
|
||||
"mark_sanctioned_daemon rejected: not called from official "
|
||||
"mcp_server.py entrypoint (#695). Offline import / standalone "
|
||||
"scripts cannot reconstruct native transport. Stop after native "
|
||||
"MCP failure; do not run offline mutation helpers."
|
||||
)
|
||||
token = secrets.token_hex(32)
|
||||
_NATIVE_RUNTIME = {
|
||||
"token": token,
|
||||
"token_fingerprint": hashlib.sha256(token.encode()).hexdigest()[:16],
|
||||
"pid": os.getpid(),
|
||||
"started_at": time.time(),
|
||||
"entrypoint": "mcp_server",
|
||||
}
|
||||
# Legacy signal for older probes; alone does not authorize mutations.
|
||||
os.environ[SANCTIONED_DAEMON_ENV] = "1"
|
||||
return native_runtime_status()
|
||||
|
||||
|
||||
def clear_native_runtime_for_tests() -> None:
|
||||
"""Test helper: drop native runtime (does not clear env)."""
|
||||
global _NATIVE_RUNTIME
|
||||
_NATIVE_RUNTIME = None
|
||||
|
||||
|
||||
def is_native_mcp_transport() -> bool:
|
||||
"""True when this process holds a live native MCP runtime record (#695)."""
|
||||
if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}:
|
||||
return False
|
||||
if _NATIVE_RUNTIME is None:
|
||||
return False
|
||||
if int(_NATIVE_RUNTIME.get("pid") or -1) != os.getpid():
|
||||
return False
|
||||
if not (_NATIVE_RUNTIME.get("token") or "").strip():
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def is_sanctioned_mcp_daemon() -> bool:
|
||||
"""Backward-compatible name; #695 requires native transport, not env alone."""
|
||||
if is_native_mcp_transport():
|
||||
if (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {"1", "true", "yes"}:
|
||||
return True
|
||||
if (os.environ.get(ALLOW_DIRECT_IMPORT_ENV) or "").strip() in {"1", "true", "yes"}:
|
||||
return True
|
||||
if is_pytest_runtime():
|
||||
return True
|
||||
# Explicit direct-import override is for non-LLM operator/test tools only.
|
||||
# It is deliberately ignored when a native runtime is expected for mutations
|
||||
# under LLM sessions (tests use pytest path). Env alone never grants native.
|
||||
return False
|
||||
|
||||
|
||||
def mark_sanctioned_daemon() -> None:
|
||||
"""Call from the official MCP server entrypoint before serving tools."""
|
||||
os.environ[SANCTIONED_DAEMON_ENV] = "1"
|
||||
|
||||
|
||||
def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
|
||||
"""Fail closed when mutation code runs outside native MCP transport (#695)."""
|
||||
"""Fail closed when server mutation code is used outside the MCP daemon."""
|
||||
if is_sanctioned_mcp_daemon():
|
||||
return
|
||||
env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}
|
||||
extra = ""
|
||||
if env_spoof:
|
||||
extra = (
|
||||
f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); "
|
||||
"native transport requires the official MCP entrypoint."
|
||||
)
|
||||
raise UnsanctionedRuntimeError(
|
||||
f"Unsanctioned / non-native runtime blocked {context} (#695). "
|
||||
"Do not import gitea_mcp_server or call mutation helpers from a raw "
|
||||
"shell, offline runner, or ad-hoc script after native MCP failure. "
|
||||
"Stop and reconnect the official MCP daemon (mcp_server.py). "
|
||||
f"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM "
|
||||
f"sessions.{extra}"
|
||||
f"Unsanctioned runtime blocked {context} (#558). "
|
||||
"Do not import gitea_mcp_server / call mutation helpers from a raw "
|
||||
"shell or ad-hoc script. Use the official MCP daemon entrypoint "
|
||||
f"(sets {SANCTIONED_DAEMON_ENV}=1), or run under pytest. "
|
||||
f"Operator-only override: {ALLOW_DIRECT_IMPORT_ENV}=1 (not for LLM sessions)."
|
||||
)
|
||||
|
||||
|
||||
@@ -155,58 +69,21 @@ def assert_keychain_access_allowed() -> None:
|
||||
"""Fail closed for git-credential keychain fill outside sanctioned contexts."""
|
||||
if is_sanctioned_mcp_daemon():
|
||||
return
|
||||
# Operator-only keychain CLI remains available outside LLM mutation path.
|
||||
if (os.environ.get(ALLOW_KEYCHAIN_CLI_ENV) or "").strip() in {"1", "true", "yes"}:
|
||||
if not is_pytest_runtime():
|
||||
# Still block pure env spoof of SANCTIONED_DAEMON for keychain when
|
||||
# FORCE is set for tests.
|
||||
if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
|
||||
"1",
|
||||
"true",
|
||||
"yes",
|
||||
}:
|
||||
pass
|
||||
else:
|
||||
return
|
||||
return
|
||||
raise UnsanctionedRuntimeError(
|
||||
"Unsanctioned keychain/credential fill blocked (#558/#695). "
|
||||
"Unsanctioned keychain/credential fill blocked (#558). "
|
||||
"Token extraction via git-credential is only allowed inside the "
|
||||
f"official native MCP daemon or with explicit operator opt-in "
|
||||
f"{ALLOW_KEYCHAIN_CLI_ENV}=1 (never for offline mutation runners)."
|
||||
f"official MCP daemon ({SANCTIONED_DAEMON_ENV}=1), pytest, or with "
|
||||
f"explicit operator opt-in {ALLOW_KEYCHAIN_CLI_ENV}=1."
|
||||
)
|
||||
|
||||
|
||||
def native_runtime_status() -> dict[str, Any]:
|
||||
"""LLM-safe native runtime status (no raw token)."""
|
||||
rt = _NATIVE_RUNTIME or {}
|
||||
def runtime_status() -> dict[str, Any]:
|
||||
return {
|
||||
"native_mcp_transport": is_native_mcp_transport(),
|
||||
"sanctioned_daemon": is_sanctioned_mcp_daemon(),
|
||||
"pytest": is_pytest_runtime(),
|
||||
"pid": rt.get("pid"),
|
||||
"token_fingerprint": rt.get("token_fingerprint"),
|
||||
"started_at": rt.get("started_at"),
|
||||
"entrypoint": rt.get("entrypoint"),
|
||||
"env_sanctioned_alone_insufficient": True,
|
||||
"sanctioned_env": SANCTIONED_DAEMON_ENV,
|
||||
"allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV,
|
||||
"allow_keychain_cli_env": ALLOW_KEYCHAIN_CLI_ENV,
|
||||
}
|
||||
|
||||
|
||||
def runtime_status() -> dict[str, Any]:
|
||||
"""Backward-compatible status payload."""
|
||||
status = native_runtime_status()
|
||||
status["sanctioned_daemon"] = is_sanctioned_mcp_daemon()
|
||||
return status
|
||||
|
||||
|
||||
def mutation_provenance_fields() -> dict[str, Any]:
|
||||
"""Fields to attach to live mutation / review audit records (#695 AC6)."""
|
||||
st = native_runtime_status()
|
||||
return {
|
||||
"transport": "native_mcp" if st["native_mcp_transport"] else "untrusted",
|
||||
"native_mcp_transport": bool(st["native_mcp_transport"]),
|
||||
"native_runtime_pid": st.get("pid"),
|
||||
"native_token_fingerprint": st.get("token_fingerprint"),
|
||||
"entrypoint": st.get("entrypoint"),
|
||||
}
|
||||
|
||||
@@ -13,7 +13,6 @@ from typing import Any
|
||||
|
||||
AUTHORIZED_CLEANUP_TOOLS = frozenset({
|
||||
"gitea_cleanup_post_merge_moot_lease",
|
||||
"gitea_cleanup_obsolete_reviewer_comment_lease",
|
||||
"gitea_reconcile_merged_cleanups",
|
||||
"gitea_delete_branch",
|
||||
"gitea_cleanup_merged_pr_branch",
|
||||
|
||||
+9
-43
@@ -1,8 +1,7 @@
|
||||
"""Merge approval must pin the current PR head SHA (#471 / #695).
|
||||
"""Merge approval must pin the current PR head SHA (#471).
|
||||
|
||||
Formal APPROVED reviews that predate the live PR head must not satisfy
|
||||
``gitea_merge_pr`` eligibility. Contaminated / quarantined approvals (#695)
|
||||
must not authorize merge either. Pure assessment helpers are isolated here
|
||||
``gitea_merge_pr`` eligibility. Pure assessment helpers are isolated here
|
||||
for hermetic unit tests apart from MCP HTTP calls.
|
||||
"""
|
||||
|
||||
@@ -13,7 +12,6 @@ def assess_merge_approval_head(
|
||||
*,
|
||||
current_head_sha: str | None,
|
||||
latest_by_reviewer: dict,
|
||||
quarantined_review_ids: set | None = None,
|
||||
) -> dict:
|
||||
"""Return whether a visible approval applies to the live PR head.
|
||||
|
||||
@@ -21,43 +19,18 @@ def assess_merge_approval_head(
|
||||
current_head_sha: Current PR head commit SHA.
|
||||
latest_by_reviewer: Map of reviewer login → review entry dicts with
|
||||
``verdict``, ``dismissed``, and ``reviewed_head_sha`` keys.
|
||||
Optional ``review_id`` / ``id`` used for quarantine checks (#695).
|
||||
quarantined_review_ids: Optional set of review IDs that must not count
|
||||
toward merge authorization (#695).
|
||||
|
||||
Returns:
|
||||
dict with ``approval_at_current_head``, ``latest_approved_head_sha``,
|
||||
and ``stale_approval_block_reason`` (set when merge must fail closed).
|
||||
"""
|
||||
current = (current_head_sha or "").strip()
|
||||
blocked_ids = {int(x) for x in (quarantined_review_ids or set()) if x is not None}
|
||||
|
||||
def _rid(entry: dict):
|
||||
raw = entry.get("review_id", entry.get("id"))
|
||||
try:
|
||||
return int(raw) if raw is not None else None
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
approved_entries = []
|
||||
quarantined_at_head = []
|
||||
for entry in (latest_by_reviewer or {}).values():
|
||||
if (entry.get("verdict") or "").upper() != "APPROVED":
|
||||
continue
|
||||
if entry.get("dismissed"):
|
||||
continue
|
||||
rid = _rid(entry)
|
||||
head = (entry.get("reviewed_head_sha") or "").strip()
|
||||
if rid is not None and rid in blocked_ids:
|
||||
if current and head == current:
|
||||
quarantined_at_head.append(entry)
|
||||
continue
|
||||
if entry.get("quarantined"):
|
||||
if current and head == current:
|
||||
quarantined_at_head.append(entry)
|
||||
continue
|
||||
approved_entries.append(entry)
|
||||
|
||||
approved_entries = [
|
||||
entry
|
||||
for entry in (latest_by_reviewer or {}).values()
|
||||
if (entry.get("verdict") or "").upper() == "APPROVED"
|
||||
and not entry.get("dismissed")
|
||||
]
|
||||
at_current = any(
|
||||
(entry.get("reviewed_head_sha") or "").strip() == current
|
||||
for entry in approved_entries
|
||||
@@ -74,13 +47,7 @@ def assess_merge_approval_head(
|
||||
)[-1]
|
||||
latest_approved = (latest_entry.get("reviewed_head_sha") or "").strip() or None
|
||||
reason = None
|
||||
if quarantined_at_head and not at_current:
|
||||
reason = (
|
||||
"contaminated/quarantined approval at current head is void for "
|
||||
"merge authorization (#695); required next action: fresh native "
|
||||
"MCP re-review after controller quarantine evidence is recorded"
|
||||
)
|
||||
elif approved_entries and not at_current:
|
||||
if approved_entries and not at_current:
|
||||
reason = (
|
||||
f"stale approval: approved SHA '{latest_approved}' does not match "
|
||||
f"current live PR head SHA '{current or '(unknown)'}' (fail closed); "
|
||||
@@ -91,5 +58,4 @@ def assess_merge_approval_head(
|
||||
"approval_at_current_head": at_current,
|
||||
"latest_approved_head_sha": latest_approved,
|
||||
"stale_approval_block_reason": reason,
|
||||
"quarantined_approvals_at_current_head": len(quarantined_at_head),
|
||||
}
|
||||
@@ -31,6 +31,7 @@ python-multipart==0.0.32
|
||||
referencing==0.37.0
|
||||
rich==15.0.0
|
||||
rpds-py==2026.5.1
|
||||
sentry-sdk==2.20.0
|
||||
shellingham==1.5.4
|
||||
sse-starlette==3.4.5
|
||||
starlette==1.3.1
|
||||
|
||||
@@ -1,351 +0,0 @@
|
||||
"""Controller quarantine of contaminated formal reviews (#695).
|
||||
|
||||
Quarantine records are durable under the MCP session-state root and are only
|
||||
writable when the caller holds a **native** MCP transport runtime. Untrusted
|
||||
local scripts that import this module cannot establish a valid quarantine
|
||||
(writes fail closed). Live server-side gates (eligibility, merge, feedback)
|
||||
honor quarantine by review_id + PR + head.
|
||||
|
||||
Forensic evidence (Gitea review objects, lease comments) is never deleted.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
import mcp_daemon_guard
|
||||
import mcp_session_state
|
||||
|
||||
KIND_QUARANTINE = "review_quarantine"
|
||||
CONFIRMATION_PREFIX = "QUARANTINE CONTAMINATED REVIEW"
|
||||
|
||||
|
||||
def _now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def quarantine_state_path(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
pr_number: int,
|
||||
review_id: int,
|
||||
) -> str:
|
||||
# Dedicated subdir under session state root (mode 0o700).
|
||||
root = os.path.join(mcp_session_state.default_state_dir(), "quarantine")
|
||||
os.makedirs(root, mode=0o700, exist_ok=True)
|
||||
# Explicit multi-segment key so remote/org/repo/pr/review cannot collide.
|
||||
segs = [
|
||||
KIND_QUARANTINE,
|
||||
mcp_session_state._sanitize_segment(remote),
|
||||
mcp_session_state._sanitize_segment(org),
|
||||
mcp_session_state._sanitize_segment(repo),
|
||||
f"pr{int(pr_number)}",
|
||||
f"rev{int(review_id)}",
|
||||
]
|
||||
return os.path.join(root, "-".join(segs) + ".json")
|
||||
|
||||
|
||||
def build_quarantine_record(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
pr_number: int,
|
||||
review_id: int,
|
||||
reviewed_head_sha: str,
|
||||
reason: str,
|
||||
actor_username: str | None,
|
||||
profile_name: str | None,
|
||||
incident_issue: int | None = None,
|
||||
forensic_comment_ids: list[int] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
provenance = mcp_daemon_guard.mutation_provenance_fields()
|
||||
return {
|
||||
"kind": KIND_QUARANTINE,
|
||||
"issue_ref": "#695",
|
||||
"remote": remote,
|
||||
"org": org,
|
||||
"repo": repo,
|
||||
"pr_number": int(pr_number),
|
||||
"review_id": int(review_id),
|
||||
"reviewed_head_sha": (reviewed_head_sha or "").strip().lower(),
|
||||
"reason": (reason or "").strip(),
|
||||
"actor_username": actor_username,
|
||||
"profile_name": profile_name,
|
||||
"incident_issue": incident_issue,
|
||||
"forensic_comment_ids": list(forensic_comment_ids or []),
|
||||
"created_at": _now(),
|
||||
"native_provenance": provenance,
|
||||
"merge_authorization": "void",
|
||||
"retain_forensic_evidence": True,
|
||||
}
|
||||
|
||||
|
||||
def assess_quarantine_write(
|
||||
*,
|
||||
confirmation: str,
|
||||
pr_number: int,
|
||||
review_id: int,
|
||||
reason: str,
|
||||
native_required: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Pure assessment of whether a quarantine write may proceed."""
|
||||
reasons: list[str] = []
|
||||
expected = f"{CONFIRMATION_PREFIX} {int(review_id)} PR {int(pr_number)}"
|
||||
if (confirmation or "").strip() != expected:
|
||||
reasons.append(
|
||||
f"confirmation must equal exactly '{expected}' (fail closed, #695)"
|
||||
)
|
||||
if not (reason or "").strip():
|
||||
reasons.append("quarantine reason is required (fail closed, #695)")
|
||||
if native_required and not mcp_daemon_guard.is_native_mcp_transport():
|
||||
if not mcp_daemon_guard.is_pytest_runtime():
|
||||
reasons.append(
|
||||
"quarantine write requires native MCP transport; untrusted "
|
||||
"local code cannot create quarantine records (#695)"
|
||||
)
|
||||
return {
|
||||
"allowed": not reasons,
|
||||
"expected_confirmation": expected,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def write_quarantine_record(record: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Persist quarantine record; fail closed outside native/pytest runtime."""
|
||||
if not mcp_daemon_guard.is_native_mcp_transport():
|
||||
if not mcp_daemon_guard.is_pytest_runtime():
|
||||
raise mcp_daemon_guard.UnsanctionedRuntimeError(
|
||||
"quarantine write blocked: non-native runtime (#695)"
|
||||
)
|
||||
path = quarantine_state_path(
|
||||
remote=str(record["remote"]),
|
||||
org=str(record["org"]),
|
||||
repo=str(record["repo"]),
|
||||
pr_number=int(record["pr_number"]),
|
||||
review_id=int(record["review_id"]),
|
||||
)
|
||||
# Refuse overwriting with weaker provenance from untrusted caller by
|
||||
# requiring native fields present.
|
||||
if not (record.get("native_provenance") or {}).get("native_mcp_transport"):
|
||||
if not mcp_daemon_guard.is_pytest_runtime():
|
||||
raise mcp_daemon_guard.UnsanctionedRuntimeError(
|
||||
"quarantine record missing native provenance (#695)"
|
||||
)
|
||||
tmp = path + ".tmp"
|
||||
data = json.dumps(record, indent=2, sort_keys=True)
|
||||
with open(tmp, "w", encoding="utf-8") as fh:
|
||||
fh.write(data)
|
||||
fh.flush()
|
||||
os.fsync(fh.fileno())
|
||||
os.replace(tmp, path)
|
||||
os.chmod(path, 0o600)
|
||||
return {"path": path, "written": True, "record": record}
|
||||
|
||||
|
||||
def load_quarantine_record(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
pr_number: int,
|
||||
review_id: int,
|
||||
) -> dict[str, Any] | None:
|
||||
path = quarantine_state_path(
|
||||
remote=remote,
|
||||
org=org,
|
||||
repo=repo,
|
||||
pr_number=pr_number,
|
||||
review_id=review_id,
|
||||
)
|
||||
if not os.path.isfile(path):
|
||||
return None
|
||||
try:
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
data = json.load(fh)
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
return data
|
||||
|
||||
|
||||
def is_review_quarantined(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
pr_number: int,
|
||||
review_id: int | None,
|
||||
reviewed_head_sha: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Whether a formal review is quarantined for merge authorization (#695)."""
|
||||
if review_id is None:
|
||||
return {"quarantined": False, "record": None, "reasons": []}
|
||||
rec = load_quarantine_record(
|
||||
remote=remote,
|
||||
org=org,
|
||||
repo=repo,
|
||||
pr_number=pr_number,
|
||||
review_id=int(review_id),
|
||||
)
|
||||
if not rec:
|
||||
return {"quarantined": False, "record": None, "reasons": []}
|
||||
reasons = [
|
||||
f"formal review_id {review_id} on PR #{pr_number} is quarantined "
|
||||
f"(#695; incident issue {rec.get('incident_issue')}); "
|
||||
"merge authorization is void; forensic evidence retained"
|
||||
]
|
||||
want_head = (reviewed_head_sha or "").strip().lower()
|
||||
rec_head = (rec.get("reviewed_head_sha") or "").strip().lower()
|
||||
if want_head and rec_head and want_head != rec_head:
|
||||
# Still quarantined by review_id; note head mismatch for operators.
|
||||
reasons.append(
|
||||
f"quarantine head {rec_head[:12]}… differs from assessed head "
|
||||
f"{want_head[:12]}… (still void by review_id)"
|
||||
)
|
||||
return {"quarantined": True, "record": rec, "reasons": reasons}
|
||||
|
||||
|
||||
def filter_approvals_for_merge(
|
||||
*,
|
||||
remote: str,
|
||||
org: str,
|
||||
repo: str,
|
||||
pr_number: int,
|
||||
current_head_sha: str | None,
|
||||
reviews: list[dict],
|
||||
) -> dict[str, Any]:
|
||||
"""Split reviews into merge-usable vs quarantined contaminated approvals."""
|
||||
current = (current_head_sha or "").strip().lower()
|
||||
usable: list[dict] = []
|
||||
quarantined: list[dict] = []
|
||||
for rev in reviews or []:
|
||||
if not isinstance(rev, dict):
|
||||
continue
|
||||
verdict = (rev.get("verdict") or rev.get("state") or "").upper()
|
||||
if verdict not in {"APPROVED", "APPROVE"}:
|
||||
continue
|
||||
if rev.get("dismissed"):
|
||||
continue
|
||||
rid = rev.get("review_id") or rev.get("id")
|
||||
head = (
|
||||
rev.get("reviewed_head_sha")
|
||||
or rev.get("commit_id")
|
||||
or rev.get("head_sha")
|
||||
or ""
|
||||
).strip().lower()
|
||||
q = is_review_quarantined(
|
||||
remote=remote,
|
||||
org=org,
|
||||
repo=repo,
|
||||
pr_number=pr_number,
|
||||
review_id=int(rid) if rid is not None else None,
|
||||
reviewed_head_sha=head or current,
|
||||
)
|
||||
entry = {**rev, "review_id": rid, "reviewed_head_sha": head}
|
||||
if q["quarantined"]:
|
||||
entry["quarantined"] = True
|
||||
entry["quarantine_reasons"] = q["reasons"]
|
||||
quarantined.append(entry)
|
||||
else:
|
||||
entry["quarantined"] = False
|
||||
usable.append(entry)
|
||||
usable_at_head = [
|
||||
e
|
||||
for e in usable
|
||||
if current and (e.get("reviewed_head_sha") or "") == current
|
||||
]
|
||||
return {
|
||||
"usable_approvals": usable,
|
||||
"usable_approvals_at_current_head": usable_at_head,
|
||||
"quarantined_approvals": quarantined,
|
||||
"approval_visible_for_merge": bool(usable_at_head),
|
||||
"has_quarantined_approval_at_head": any(
|
||||
current
|
||||
and (e.get("reviewed_head_sha") or "") == current
|
||||
for e in quarantined
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def format_quarantine_audit_comment(record: dict[str, Any]) -> str:
|
||||
"""Append-only forensic audit comment body (does not delete evidence)."""
|
||||
lines = [
|
||||
"## Contaminated formal review quarantine (#695)",
|
||||
"",
|
||||
"Status: **QUARANTINED — merge authorization VOID**",
|
||||
"",
|
||||
f"- review_id: `{record.get('review_id')}`",
|
||||
f"- pr: `#{record.get('pr_number')}`",
|
||||
f"- reviewed_head_sha: `{record.get('reviewed_head_sha')}`",
|
||||
f"- actor: `{record.get('actor_username')}`",
|
||||
f"- profile: `{record.get('profile_name')}`",
|
||||
f"- incident_issue: `#{record.get('incident_issue')}`",
|
||||
f"- reason: {record.get('reason')}",
|
||||
f"- created_at: `{record.get('created_at')}`",
|
||||
f"- native_transport: "
|
||||
f"`{(record.get('native_provenance') or {}).get('native_mcp_transport')}`",
|
||||
f"- native_token_fingerprint: "
|
||||
f"`{(record.get('native_provenance') or {}).get('native_token_fingerprint')}`",
|
||||
f"- forensic_comment_ids retained: "
|
||||
f"`{record.get('forensic_comment_ids')}`",
|
||||
"",
|
||||
"This record does **not** delete Gitea reviews or historical comments. "
|
||||
"Fresh native-MCP re-review is required before merge.",
|
||||
]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def assess_untrusted_canonical_approval_claim(body: str) -> list[str]:
|
||||
"""Reasons to reject canonical comments that claim approved/merge-ready.
|
||||
|
||||
Used when the comment asserts approval without native review proof (#695 AC7).
|
||||
False "official workflow" claims that cite offline/import paths are always
|
||||
rejected even when a NATIVE_REVIEW_PROOF line is present.
|
||||
"""
|
||||
text = body or ""
|
||||
lower = text.lower()
|
||||
claims_approved = (
|
||||
"state:\napproved" in lower
|
||||
or "state: approved" in lower
|
||||
or "who_is_next:\nmerger" in lower
|
||||
or "who_is_next: merger" in lower
|
||||
or "merge_ready: true" in lower
|
||||
or "merge_ready:\ntrue" in lower
|
||||
or "ready-to-merge" in lower
|
||||
)
|
||||
if not claims_approved:
|
||||
return []
|
||||
# Reject explicit offline/import "official workflow" spoofing (#695 AC9).
|
||||
offline_spoof = (
|
||||
"offline_mcp" in lower
|
||||
or "offline import" in lower
|
||||
or "direct import" in lower
|
||||
or "import gitea_mcp_server" in lower
|
||||
or "run_quarantine.py" in lower
|
||||
or "offline_mcp_helper" in lower
|
||||
or "offline_mcp_runner" in lower
|
||||
)
|
||||
has_proof = (
|
||||
"NATIVE_REVIEW_PROOF:" in text
|
||||
or "native_review_proof:" in lower
|
||||
)
|
||||
if offline_spoof:
|
||||
return [
|
||||
"canonical approval claim cites offline/import/helper path; "
|
||||
"NATIVE_REVIEW_PROOF rejected (#695 AC7/AC9); stop after native "
|
||||
"MCP failure — do not construct offline mutation fallbacks"
|
||||
]
|
||||
if has_proof:
|
||||
return []
|
||||
return [
|
||||
"canonical comment claims approved/merge-ready state without "
|
||||
"NATIVE_REVIEW_PROOF from a native MCP review mutation (#695 AC7); "
|
||||
"contaminated or untrusted approvals cannot certify merger handoff"
|
||||
]
|
||||
+43
-657
@@ -524,26 +524,13 @@ def assess_lease_inventory(
|
||||
"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_WAIT = "wait"
|
||||
NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
|
||||
NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
|
||||
NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
|
||||
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({
|
||||
"no_lease",
|
||||
@@ -552,12 +539,6 @@ _HANDOFF_CLASSIFICATIONS = frozenset({
|
||||
"foreign_active",
|
||||
"foreign_reclaimable",
|
||||
"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",
|
||||
"worktree_binding_mismatch",
|
||||
})
|
||||
@@ -570,476 +551,6 @@ def _norm_path(value: str | None) -> str:
|
||||
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(
|
||||
comments: list[dict],
|
||||
*,
|
||||
@@ -1050,50 +561,40 @@ def diagnose_reviewer_pr_lease_handoff(
|
||||
env_bound_worktree: str | None = None,
|
||||
instructed_session_id: str | 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,
|
||||
) -> 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.
|
||||
Fail-closed acquisition rules for active foreign leases remain intact.
|
||||
|
||||
Returns a structured diagnosis with:
|
||||
- classification (including #691 superseded/expired distinctions)
|
||||
- next_action (including cleanup_obsolete_reviewer_comment_lease)
|
||||
- classification
|
||||
- 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
|
||||
- worktree_binding match result
|
||||
- instructed-lease mismatch flags
|
||||
- cleanup tool/confirmation when cleanup-eligible
|
||||
"""
|
||||
now = now or datetime.now(timezone.utc)
|
||||
session_id = (current_session_id or "").strip()
|
||||
identity = (current_reviewer_identity or "").strip()
|
||||
instructed_sid = (instructed_session_id or "").strip()
|
||||
reasons: list[str] = []
|
||||
current_head = _normalize_sha(current_head_sha)
|
||||
|
||||
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()
|
||||
|
||||
# Worktree binding: env-bound vs proposed vs active lease worktree.
|
||||
env_wt = _norm_path(env_bound_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_details: dict[str, Any] = {
|
||||
"env_bound_worktree": env_bound_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,
|
||||
}
|
||||
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_missing_with_replacement = False
|
||||
if instructed_sid or instructed_comment_id is not None:
|
||||
if not lease_for_class:
|
||||
if not active:
|
||||
reasons.append(
|
||||
"instructed lease is gone and no active replacement lease remains"
|
||||
)
|
||||
else:
|
||||
owner = (lease_for_class.get("session_id") or "").strip()
|
||||
cid = lease_for_class.get("comment_id")
|
||||
owner = (active.get("session_id") or "").strip()
|
||||
cid = active.get("comment_id")
|
||||
sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
|
||||
cid_mismatch = (
|
||||
instructed_comment_id is not None
|
||||
@@ -1130,94 +631,35 @@ def diagnose_reviewer_pr_lease_handoff(
|
||||
# Classification + next_action.
|
||||
classification = "no_lease"
|
||||
next_action = NEXT_ACTION_ACQUIRE
|
||||
cleanup_hint: dict[str, Any] | None = None
|
||||
|
||||
if lease_for_class:
|
||||
owner = (lease_for_class.get("session_id") or "").strip()
|
||||
freshness = lease_for_class.get("freshness") or classify_lease_freshness(
|
||||
lease_for_class, now=now
|
||||
if active:
|
||||
owner = (active.get("session_id") or "").strip()
|
||||
freshness = active.get("freshness") or classify_lease_freshness(
|
||||
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)
|
||||
# Same identity alone is NOT ownership for resume; session_id must match.
|
||||
same_identity = bool(
|
||||
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"
|
||||
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"
|
||||
next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
|
||||
reasons.append(
|
||||
f"own lease freshness is '{freshness}'; release via "
|
||||
"gitea_release_reviewer_pr_lease then re-acquire"
|
||||
)
|
||||
elif not is_own and head_superseded and terminal and past_expiry:
|
||||
classification = "foreign_expired_superseded_head"
|
||||
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"
|
||||
)
|
||||
elif not is_own and freshness in {"active", "stale_warning"}:
|
||||
classification = "foreign_active"
|
||||
next_action = NEXT_ACTION_WAIT
|
||||
reasons.append(
|
||||
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"
|
||||
)
|
||||
if same_identity:
|
||||
@@ -1232,12 +674,11 @@ def diagnose_reviewer_pr_lease_handoff(
|
||||
f"foreign reclaimable lease (session_id={owner}); clear only via "
|
||||
"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"
|
||||
next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
|
||||
reasons.append(
|
||||
f"foreign expired lease (session_id={owner}); use sanctioned release "
|
||||
f"or {CLEANUP_OBSOLETE_LEASE_TOOL}"
|
||||
f"foreign expired lease (session_id={owner}); use sanctioned release"
|
||||
)
|
||||
else:
|
||||
classification = "foreign_active"
|
||||
@@ -1250,26 +691,13 @@ def diagnose_reviewer_pr_lease_handoff(
|
||||
if instructed_missing_with_replacement and classification.startswith(
|
||||
"foreign"
|
||||
):
|
||||
# Preserve #691 cleanup path when superseded/expired; otherwise
|
||||
# keep the PR #592 instructed-missing label.
|
||||
if next_action != NEXT_ACTION_CLEANUP_OBSOLETE_LEASE:
|
||||
classification = "instructed_lease_missing_with_replacement"
|
||||
if next_action == NEXT_ACTION_WAIT:
|
||||
reasons.append(
|
||||
"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"),
|
||||
}
|
||||
classification = "instructed_lease_missing_with_replacement"
|
||||
# Foreign active still means wait; reclaimable still release.
|
||||
if next_action == NEXT_ACTION_WAIT:
|
||||
reasons.append(
|
||||
"replacement foreign lease is active — wait; "
|
||||
"operator_authorized_cleanup only with explicit operator authority"
|
||||
)
|
||||
else:
|
||||
classification = "no_lease"
|
||||
next_action = NEXT_ACTION_ACQUIRE
|
||||
@@ -1292,55 +720,29 @@ def diagnose_reviewer_pr_lease_handoff(
|
||||
)
|
||||
|
||||
lease_summary = None
|
||||
if lease_for_class:
|
||||
if active:
|
||||
lease_summary = {
|
||||
"comment_id": lease_for_class.get("comment_id"),
|
||||
"session_id": lease_for_class.get("session_id"),
|
||||
"phase": lease_for_class.get("phase"),
|
||||
"candidate_head": lease_for_class.get("candidate_head"),
|
||||
"expires_at": lease_for_class.get("expires_at"),
|
||||
"last_activity": lease_for_class.get("last_activity"),
|
||||
"freshness": lease_for_class.get("freshness")
|
||||
or classify_lease_freshness(lease_for_class, now=now),
|
||||
"reviewer_identity": lease_for_class.get("reviewer_identity"),
|
||||
"profile": lease_for_class.get("profile"),
|
||||
"worktree": lease_for_class.get("worktree"),
|
||||
"blocker": lease_for_class.get("blocker"),
|
||||
"comment_id": active.get("comment_id"),
|
||||
"session_id": active.get("session_id"),
|
||||
"phase": active.get("phase"),
|
||||
"candidate_head": active.get("candidate_head"),
|
||||
"expires_at": active.get("expires_at"),
|
||||
"last_activity": active.get("last_activity"),
|
||||
"freshness": active.get("freshness")
|
||||
or classify_lease_freshness(active, now=now),
|
||||
"reviewer_identity": active.get("reviewer_identity"),
|
||||
"profile": active.get("profile"),
|
||||
"worktree": active.get("worktree"),
|
||||
"blocker": active.get("blocker"),
|
||||
}
|
||||
|
||||
return {
|
||||
"pr_number": pr_number,
|
||||
"classification": classification,
|
||||
"blocker_kind": classification,
|
||||
"next_action": next_action,
|
||||
"exact_next_action": next_action,
|
||||
"active_lease": lease_summary,
|
||||
"session_lease": session_lease,
|
||||
"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_comment_id": instructed_comment_id,
|
||||
"instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
|
||||
@@ -1349,19 +751,6 @@ def diagnose_reviewer_pr_lease_handoff(
|
||||
and not binding_mismatch
|
||||
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,
|
||||
"forbidden": [
|
||||
"manual lock deletion",
|
||||
@@ -1369,8 +758,5 @@ def diagnose_reviewer_pr_lease_handoff(
|
||||
"mtime manipulation",
|
||||
"direct _SESSION_LEASE seeding",
|
||||
"silent foreign lease steal",
|
||||
"PID-based ownership",
|
||||
"repointing old lease to new head",
|
||||
"transfer of validation or decision state",
|
||||
],
|
||||
}
|
||||
|
||||
@@ -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
|
||||
@@ -14,8 +14,6 @@ before any PR mutation. Final report schema:
|
||||
|
||||
**BLOCKED + DIAGNOSE (universal default):** If at any point you cannot perform a required step (skill not available, terminal broken, capability missing, wrong profile, guard blocks, worktree misbound, instructions unavailable, preflight fails, etc.), STOP. Clearly state BLOCKED. Use the standard [`../templates/blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template. Only safe non-mutating recovery. Report using the template. Do not continue or use any fallback (temp scripts, direct API, etc.) unless controller authorizes. All blocker classes listed in llm-project-workflow/SKILL.md must trigger this.
|
||||
|
||||
**Native MCP failure is a hard stop (#695):** If the native MCP namespace dies (EOF, capability disconnect, session death), STOP. Do **not** import `gitea_mcp_server` from a standalone process, do **not** run offline helpers (`offline_mcp_helper.py`, `offline_mcp_runner.py`, `run_quarantine.py`), and do **not** set direct-import / keychain-bypass / raw-token environment variables. Reconnect the official MCP daemon only. Contaminated approvals must be controller-quarantined; they never authorize merge.
|
||||
|
||||
**Default task prompt:**
|
||||
|
||||
> Review the next eligible open PR in this project. Merge it only if every
|
||||
|
||||
@@ -14,8 +14,6 @@ before any issue implementation mutation. Final report schema:
|
||||
|
||||
**BLOCKED + DIAGNOSE (universal default):** If at any point you cannot perform a required step (skill not available, terminal broken, capability missing, wrong profile, guard blocks, worktree misbound, instructions unavailable, preflight fails, etc.), STOP. Clearly state BLOCKED. Use the standard [`../templates/blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template. Only safe non-mutating recovery. Report using the template. Do not continue or use any fallback (temp scripts, direct API, etc.) unless controller authorizes. All blocker classes listed in llm-project-workflow/SKILL.md must trigger this.
|
||||
|
||||
**Native MCP failure is a hard stop (#695):** Do not import MCP server internals, run offline mutation helpers, or set credential-bypass env vars after a native transport failure. Reconnect the official daemon only.
|
||||
|
||||
**Default task prompt:**
|
||||
|
||||
> Find the next eligible issue in this project, work on it only if all gates
|
||||
|
||||
@@ -72,30 +72,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
"permission": "gitea.pr.merge",
|
||||
"role": "merger",
|
||||
},
|
||||
# #695 AC8: controller quarantine of contaminated formal reviews.
|
||||
# Apply path posts an append-only forensic audit comment (pr.comment).
|
||||
"quarantine_contaminated_review": {
|
||||
"permission": "gitea.pr.comment",
|
||||
"role": "reconciler",
|
||||
},
|
||||
"gitea_quarantine_contaminated_review": {
|
||||
"permission": "gitea.pr.comment",
|
||||
"role": "reconciler",
|
||||
},
|
||||
"adopt_merger_pr_lease": {
|
||||
"permission": "gitea.pr.comment",
|
||||
"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": {
|
||||
"permission": "gitea.pr.review",
|
||||
"role": "reviewer",
|
||||
|
||||
+5
-9
@@ -329,17 +329,13 @@ class TestGatedToolAudit(_AuditWiringBase):
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_merge_success_audited(self, _auth, mock_api):
|
||||
# user, pr, eligibility feedback pr+reviews (#695), gate-7 feedback
|
||||
# pr+reviews, merge POST, readback.
|
||||
approval = [{
|
||||
"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
|
||||
"commit_id": "abc123", "submitted_at": "2026-07-06T10:00:00Z",
|
||||
"dismissed": False,
|
||||
}]
|
||||
# user, pr, feedback pr+reviews, merge POST, readback.
|
||||
mock_api.side_effect = [
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
self._pr("author-bot"), approval, # eligibility merge feedback
|
||||
self._pr("author-bot"), approval, # gate 7 feedback
|
||||
self._pr("author-bot"),
|
||||
[{"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
|
||||
"commit_id": "abc123", "submitted_at": "2026-07-06T10:00:00Z",
|
||||
"dismissed": False}],
|
||||
{}, {"merged_commit_sha": "c1"},
|
||||
]
|
||||
env = self._env(GITEA_PROFILE_NAME="gitea-merger",
|
||||
|
||||
@@ -94,7 +94,6 @@ REVIEW_STATUS: approved / approval_at_current_head
|
||||
MERGE_READY: true
|
||||
BLOCKERS: none
|
||||
VALIDATION: pytest passed; reviewer approved at head {FULL_SHA}
|
||||
NATIVE_REVIEW_PROOF: transport=native_mcp; entrypoint=mcp_server; token_fingerprint=testharmless
|
||||
LAST_UPDATED_BY: prgs-reviewer
|
||||
"""
|
||||
|
||||
|
||||
@@ -57,23 +57,9 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
|
||||
# Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
|
||||
# path is the stable control checkout (not under branches/), mutation must fail.
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
try:
|
||||
res = srv.gitea_create_issue(title="Test issue", body="body text")
|
||||
except RuntimeError as exc:
|
||||
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"))
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.gitea_create_issue(title="Test issue", body="body text")
|
||||
self.assertIn("stable control checkout", str(ctx.exception))
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@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):
|
||||
try:
|
||||
res = srv.gitea_create_issue(
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.gitea_create_issue(
|
||||
title="Test issue", body="body", worktree_path=missing_path
|
||||
)
|
||||
except RuntimeError as exc:
|
||||
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"))
|
||||
self.assertIn("does not exist (fail closed)", str(ctx.exception))
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@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("gitea_mcp_server._get_workspace_porcelain", return_value=""):
|
||||
try:
|
||||
res = srv.gitea_create_issue(
|
||||
title="Test issue",
|
||||
body="body",
|
||||
worktree_path=wrong_repo_path,
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.gitea_create_issue(
|
||||
title="Test issue", body="body", worktree_path=wrong_repo_path
|
||||
)
|
||||
except RuntimeError as exc:
|
||||
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)
|
||||
self.assertIn("does not belong to the target repository", str(ctx.exception))
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
||||
|
||||
@@ -267,19 +267,10 @@ class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase):
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
|
||||
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
|
||||
try:
|
||||
res = srv.gitea_create_issue_comment(
|
||||
with self.assertRaises(RuntimeError):
|
||||
srv.gitea_create_issue_comment(
|
||||
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()
|
||||
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -1,400 +0,0 @@
|
||||
"""Regression tests for Issue #695 — second incident (PR #694 / review 427).
|
||||
|
||||
Reproduces offline import, env-only runtime spoof, exposed-token invocation,
|
||||
direct imports, locally generated runtime keys, standalone quarantine attempts,
|
||||
and false “official workflow” canonical claims. Gates must fail closed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import mcp_daemon_guard
|
||||
import merge_approval_gate
|
||||
import review_quarantine
|
||||
import canonical_comment_validator as ccv
|
||||
|
||||
HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd"
|
||||
HEAD_OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
|
||||
|
||||
class TestNativeTransportBinding(unittest.TestCase):
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
|
||||
os.environ.pop(mcp_daemon_guard.SANCTIONED_DAEMON_ENV, None)
|
||||
os.environ.pop(mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV, None)
|
||||
|
||||
def test_env_alone_does_not_establish_native_transport(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
os.environ[mcp_daemon_guard.SANCTIONED_DAEMON_ENV] = "1"
|
||||
os.environ[mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV] = "1"
|
||||
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
||||
self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("offline_import")
|
||||
msg = str(ctx.exception)
|
||||
self.assertIn("#695", msg)
|
||||
self.assertIn("not sufficient", msg.lower() + " " + msg)
|
||||
|
||||
def test_direct_import_mark_rejected_outside_entrypoint(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
||||
mcp_daemon_guard.mark_sanctioned_daemon()
|
||||
self.assertIn("mcp_server.py", str(ctx.exception))
|
||||
self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
|
||||
|
||||
def test_locally_generated_runtime_key_without_entrypoint_rejected(self):
|
||||
"""Spoofing process-local fields via mark outside entrypoint fails."""
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
||||
# Even if a caller tries allow_test_bootstrap under force-unsanctioned
|
||||
# pytest path is also forced off — only real entrypoint may mark.
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=False)
|
||||
|
||||
def test_test_bootstrap_establishes_native_for_hermetic_tests(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
st = mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
||||
self.assertTrue(st["native_mcp_transport"])
|
||||
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("test-bootstrap")
|
||||
fields = mcp_daemon_guard.mutation_provenance_fields()
|
||||
self.assertEqual(fields["transport"], "native_mcp")
|
||||
self.assertTrue(fields["native_mcp_transport"])
|
||||
self.assertIsNotNone(fields["native_token_fingerprint"])
|
||||
|
||||
def test_exposed_token_env_never_grants_native(self):
|
||||
"""Raw / exposed token env vars must never reconstruct native transport."""
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
||||
for key in (
|
||||
"GITEA_TOKEN",
|
||||
"GITEA_ACCESS_TOKEN",
|
||||
"GITHUB_TOKEN",
|
||||
"GITEA_MCP_TOKEN",
|
||||
"GITEA_RAW_TOKEN",
|
||||
):
|
||||
os.environ[key] = "exposed-token-value-must-not-authorize"
|
||||
try:
|
||||
self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("exposed-token")
|
||||
finally:
|
||||
for key in (
|
||||
"GITEA_TOKEN",
|
||||
"GITEA_ACCESS_TOKEN",
|
||||
"GITHUB_TOKEN",
|
||||
"GITEA_MCP_TOKEN",
|
||||
"GITEA_RAW_TOKEN",
|
||||
):
|
||||
os.environ.pop(key, None)
|
||||
|
||||
|
||||
class TestQuarantineWriteNativeOnly(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
|
||||
|
||||
def test_standalone_quarantine_write_blocked_when_unsanctioned(self):
|
||||
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
||||
assessment = review_quarantine.assess_quarantine_write(
|
||||
confirmation="QUARANTINE CONTAMINATED REVIEW 427 PR 694",
|
||||
pr_number=694,
|
||||
review_id=427,
|
||||
reason="contaminated offline approval",
|
||||
native_required=True,
|
||||
)
|
||||
self.assertFalse(assessment["allowed"])
|
||||
self.assertTrue(
|
||||
any("native MCP transport" in r for r in assessment["reasons"])
|
||||
)
|
||||
record = review_quarantine.build_quarantine_record(
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
pr_number=694,
|
||||
review_id=427,
|
||||
reviewed_head_sha=HEAD_694,
|
||||
reason="contaminated",
|
||||
actor_username="sysadmin",
|
||||
profile_name="prgs-merger",
|
||||
incident_issue=695,
|
||||
forensic_comment_ids=[10883, 10886],
|
||||
)
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||
review_quarantine.write_quarantine_record(record)
|
||||
|
||||
def test_confirmation_must_match_exactly(self):
|
||||
assessment = review_quarantine.assess_quarantine_write(
|
||||
confirmation="quarantine 427",
|
||||
pr_number=694,
|
||||
review_id=427,
|
||||
reason="x",
|
||||
native_required=False,
|
||||
)
|
||||
self.assertFalse(assessment["allowed"])
|
||||
self.assertIn("confirmation must equal exactly", assessment["reasons"][0])
|
||||
|
||||
def test_quarantine_honored_by_merge_approval_gate(self):
|
||||
"""Contaminated review 427 at head must not authorize merge (#695)."""
|
||||
result = merge_approval_gate.assess_merge_approval_head(
|
||||
current_head_sha=HEAD_694,
|
||||
latest_by_reviewer={
|
||||
"sysadmin": {
|
||||
"verdict": "APPROVED",
|
||||
"dismissed": False,
|
||||
"reviewed_head_sha": HEAD_694,
|
||||
"review_id": 427,
|
||||
"submitted_at": "2026-07-13T07:20:00Z",
|
||||
}
|
||||
},
|
||||
quarantined_review_ids={427},
|
||||
)
|
||||
self.assertFalse(result["approval_at_current_head"])
|
||||
self.assertIn("quarantined", result["stale_approval_block_reason"])
|
||||
self.assertEqual(result["quarantined_approvals_at_current_head"], 1)
|
||||
|
||||
def test_filter_approvals_for_merge_splits_quarantined(self):
|
||||
with patch(
|
||||
"review_quarantine.mcp_session_state.default_state_dir",
|
||||
return_value=self._tmp.name,
|
||||
):
|
||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
||||
record = review_quarantine.build_quarantine_record(
|
||||
remote="prgs",
|
||||
org="org",
|
||||
repo="repo",
|
||||
pr_number=694,
|
||||
review_id=427,
|
||||
reviewed_head_sha=HEAD_694,
|
||||
reason="offline import contaminated approval",
|
||||
actor_username="sysadmin",
|
||||
profile_name="prgs-merger",
|
||||
incident_issue=695,
|
||||
)
|
||||
# Force native provenance bit for write path under test bootstrap.
|
||||
record["native_provenance"] = {
|
||||
**record["native_provenance"],
|
||||
"native_mcp_transport": True,
|
||||
}
|
||||
review_quarantine.write_quarantine_record(record)
|
||||
filtered = review_quarantine.filter_approvals_for_merge(
|
||||
remote="prgs",
|
||||
org="org",
|
||||
repo="repo",
|
||||
pr_number=694,
|
||||
current_head_sha=HEAD_694,
|
||||
reviews=[
|
||||
{
|
||||
"verdict": "APPROVED",
|
||||
"review_id": 427,
|
||||
"reviewed_head_sha": HEAD_694,
|
||||
"reviewer": "sysadmin",
|
||||
},
|
||||
{
|
||||
"verdict": "APPROVED",
|
||||
"review_id": 999,
|
||||
"reviewed_head_sha": HEAD_694,
|
||||
"reviewer": "fresh-reviewer",
|
||||
},
|
||||
],
|
||||
)
|
||||
self.assertTrue(filtered["has_quarantined_approval_at_head"])
|
||||
self.assertEqual(len(filtered["quarantined_approvals"]), 1)
|
||||
self.assertEqual(filtered["quarantined_approvals"][0]["review_id"], 427)
|
||||
self.assertEqual(len(filtered["usable_approvals_at_current_head"]), 1)
|
||||
self.assertEqual(
|
||||
filtered["usable_approvals_at_current_head"][0]["review_id"], 999
|
||||
)
|
||||
self.assertTrue(filtered["approval_visible_for_merge"])
|
||||
|
||||
|
||||
class TestCanonicalHandoffValidation(unittest.TestCase):
|
||||
def test_false_official_workflow_offline_claim_rejected(self):
|
||||
body = f"""## Canonical PR State
|
||||
|
||||
STATE: approved
|
||||
WHO_IS_NEXT: merger
|
||||
NEXT_ACTION: Merge PR #694 immediately after offline helper success
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Merger: land PR #694; offline_mcp_runner completed official workflow.
|
||||
```
|
||||
WHAT_HAPPENED: offline import of gitea_mcp_server submitted APPROVED review 427
|
||||
WHY: claimed official workflow via direct import after native EOF
|
||||
ISSUE: #693
|
||||
HEAD_SHA: {HEAD_694}
|
||||
REVIEW_STATUS: approved / approval_at_current_head
|
||||
MERGE_READY: true
|
||||
BLOCKERS: none
|
||||
VALIDATION: offline_mcp_helper.py + offline_mcp_runner.py; import gitea_mcp_server
|
||||
NATIVE_REVIEW_PROOF: transport=offline_mcp; spoofed
|
||||
LAST_UPDATED_BY: contaminated-session
|
||||
"""
|
||||
result = ccv.assess_canonical_comment(body, context="pr_comment")
|
||||
self.assertFalse(result["allowed"])
|
||||
joined = " ".join(result.get("extra_reasons") or [])
|
||||
self.assertIn("#695", joined)
|
||||
self.assertIn("offline", joined.lower())
|
||||
|
||||
def test_merge_ready_without_native_proof_rejected(self):
|
||||
body = f"""## Canonical PR State
|
||||
|
||||
STATE: ready-to-merge
|
||||
WHO_IS_NEXT: merger
|
||||
NEXT_ACTION: Merge PR after confirming approval_at_current_head
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Merge PR #694 for issue #693 after live mergeable check passes.
|
||||
```
|
||||
WHAT_HAPPENED: Reviewer approved at current head
|
||||
WHY: All gates passed and head SHA is current
|
||||
ISSUE: #693
|
||||
HEAD_SHA: {HEAD_694}
|
||||
REVIEW_STATUS: approved / approval_at_current_head
|
||||
MERGE_READY: true
|
||||
BLOCKERS: none
|
||||
VALIDATION: pytest passed; reviewer approved at head {HEAD_694}
|
||||
LAST_UPDATED_BY: prgs-reviewer
|
||||
"""
|
||||
result = ccv.assess_canonical_comment(body, context="pr_comment")
|
||||
self.assertFalse(result["allowed"])
|
||||
joined = " ".join(result.get("extra_reasons") or [])
|
||||
self.assertIn("NATIVE_REVIEW_PROOF", joined)
|
||||
|
||||
def test_merge_ready_with_native_proof_allowed(self):
|
||||
body = f"""## Canonical PR State
|
||||
|
||||
STATE: ready-to-merge
|
||||
WHO_IS_NEXT: merger
|
||||
NEXT_ACTION: Merge PR after confirming approval_at_current_head
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Merge PR #500 for issue #496 after live mergeable check passes.
|
||||
```
|
||||
WHAT_HAPPENED: Reviewer approved at current head via native MCP
|
||||
WHY: All gates passed and head SHA is current
|
||||
ISSUE: #496
|
||||
HEAD_SHA: {HEAD_694}
|
||||
REVIEW_STATUS: approved / approval_at_current_head
|
||||
MERGE_READY: true
|
||||
BLOCKERS: none
|
||||
VALIDATION: pytest passed; reviewer approved at head {HEAD_694}
|
||||
NATIVE_REVIEW_PROOF: transport=native_mcp; entrypoint=mcp_server; token_fingerprint=abc123
|
||||
LAST_UPDATED_BY: prgs-reviewer
|
||||
"""
|
||||
result = ccv.assess_canonical_comment(body, context="pr_comment")
|
||||
self.assertTrue(result["allowed"], result)
|
||||
|
||||
|
||||
class TestFeedbackQuarantineIntegration(unittest.TestCase):
|
||||
"""gitea_get_pr_review_feedback must void quarantined review 427 at head."""
|
||||
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
|
||||
def test_feedback_excludes_quarantined_approval_from_merge_auth(self):
|
||||
import mcp_server
|
||||
|
||||
with patch(
|
||||
"review_quarantine.mcp_session_state.default_state_dir",
|
||||
return_value=self._tmp.name,
|
||||
):
|
||||
record = review_quarantine.build_quarantine_record(
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
pr_number=694,
|
||||
review_id=427,
|
||||
reviewed_head_sha=HEAD_694,
|
||||
reason="contaminated offline approval (incident #695)",
|
||||
actor_username="controller",
|
||||
profile_name="prgs-merger",
|
||||
incident_issue=695,
|
||||
forensic_comment_ids=[10883, 10886],
|
||||
)
|
||||
record["native_provenance"]["native_mcp_transport"] = True
|
||||
review_quarantine.write_quarantine_record(record)
|
||||
|
||||
def _api(method, url, auth=None, payload=None):
|
||||
if url.endswith("/pulls/694") and method == "GET":
|
||||
return {
|
||||
"number": 694,
|
||||
"state": "open",
|
||||
"head": {"sha": HEAD_694},
|
||||
"user": {"login": "jcwalker3"},
|
||||
}
|
||||
if url.endswith("/reviews"):
|
||||
return [
|
||||
{
|
||||
"id": 427,
|
||||
"user": {"login": "sysadmin"},
|
||||
"state": "APPROVED",
|
||||
"body": "contaminated",
|
||||
"submitted_at": "2026-07-13T07:20:00Z",
|
||||
"commit_id": HEAD_694,
|
||||
"dismissed": False,
|
||||
"stale": False,
|
||||
}
|
||||
]
|
||||
return {}
|
||||
|
||||
with patch("mcp_server.api_request", side_effect=_api):
|
||||
with patch(
|
||||
"mcp_server.get_auth_header", return_value="Basic dGVzdA=="
|
||||
):
|
||||
with patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "prgs-merger",
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.merge"],
|
||||
"forbidden_operations": [],
|
||||
"base_url": None,
|
||||
},
|
||||
):
|
||||
result = mcp_server.gitea_get_pr_review_feedback(
|
||||
pr_number=694,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
)
|
||||
self.assertTrue(result.get("success"), result)
|
||||
self.assertFalse(result.get("approval_at_current_head"))
|
||||
self.assertFalse(result.get("approval_visible"))
|
||||
self.assertIn(427, result.get("quarantined_review_ids") or [])
|
||||
self.assertGreaterEqual(
|
||||
result.get("quarantined_approvals_at_current_head") or 0, 1
|
||||
)
|
||||
self.assertIn("quarantined", (result.get("stale_approval_block_reason") or ""))
|
||||
|
||||
|
||||
class TestDocsStopAfterNativeFailure(unittest.TestCase):
|
||||
def test_daemon_guard_doc_requires_stop(self):
|
||||
root = Path(__file__).resolve().parent.parent
|
||||
doc = (root / "docs" / "mcp-daemon-import-guard.md").read_text(encoding="utf-8")
|
||||
self.assertIn("#695", doc)
|
||||
self.assertIn("STOP after native MCP failure", doc)
|
||||
self.assertIn("offline_mcp_runner", doc)
|
||||
self.assertIn("run_quarantine.py", doc)
|
||||
self.assertIn("GITEA_ALLOW_DIRECT_MCP_IMPORT", doc)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -108,24 +108,13 @@ class TestIssueCommentWorkspaceGuard(unittest.TestCase):
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
side_effect=self._git_state(valid_worktree),
|
||||
):
|
||||
try:
|
||||
res = srv.gitea_create_issue_comment(
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.gitea_create_issue_comment(
|
||||
issue_number=557,
|
||||
body="evidence comment",
|
||||
remote="prgs",
|
||||
)
|
||||
except RuntimeError as exc:
|
||||
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"))
|
||||
self.assertIn("stable control checkout", str(ctx.exception))
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@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),
|
||||
):
|
||||
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
|
||||
try:
|
||||
res = srv.gitea_create_issue_comment(
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.gitea_create_issue_comment(
|
||||
issue_number=557,
|
||||
body="evidence comment",
|
||||
remote="prgs",
|
||||
worktree_path=outside_worktree,
|
||||
)
|
||||
except RuntimeError as exc:
|
||||
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
|
||||
)
|
||||
self.assertIn("does not belong to the target repository", str(ctx.exception))
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Tests for sanctioned MCP daemon guards (#558 / #695)."""
|
||||
"""Tests for sanctioned MCP daemon guards (#558)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -11,10 +11,6 @@ import gitea_auth
|
||||
|
||||
|
||||
class TestMcpDaemonGuard(unittest.TestCase):
|
||||
def tearDown(self) -> None:
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
|
||||
|
||||
def test_unsanctioned_blocks_mutation_runtime(self):
|
||||
env = {k: v for k, v in os.environ.items() if k not in {
|
||||
mcp_daemon_guard.SANCTIONED_DAEMON_ENV,
|
||||
@@ -30,29 +26,11 @@ class TestMcpDaemonGuard(unittest.TestCase):
|
||||
# Running under pytest already sets PYTEST_CURRENT_TEST.
|
||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest")
|
||||
|
||||
def test_mark_sanctioned_bootstrap_allows(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
|
||||
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
||||
|
||||
def test_env_alone_insufficient_when_force_unsanctioned(self):
|
||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||
env = {
|
||||
k: v
|
||||
for k, v in os.environ.items()
|
||||
if k not in {
|
||||
mcp_daemon_guard.SANCTIONED_DAEMON_ENV,
|
||||
mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV,
|
||||
"PYTEST_CURRENT_TEST",
|
||||
}
|
||||
}
|
||||
def test_mark_sanctioned_allows(self):
|
||||
env = {k: v for k, v in os.environ.items() if k != "PYTEST_CURRENT_TEST"}
|
||||
env[mcp_daemon_guard.SANCTIONED_DAEMON_ENV] = "1"
|
||||
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("env-spoof")
|
||||
self.assertIn("not sufficient", str(ctx.exception))
|
||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
|
||||
|
||||
def test_keychain_blocked_without_sanction(self):
|
||||
env = {
|
||||
|
||||
+12
-33
@@ -836,24 +836,16 @@ class TestMergePR(unittest.TestCase):
|
||||
)
|
||||
|
||||
def _feedback_reads(self, author="author-bot", sha="abc123"):
|
||||
"""PR + reviews GETs for one gitea_get_pr_review_feedback call."""
|
||||
"""PR + reviews GETs for gitea_get_pr_review_feedback during merge."""
|
||||
return [self._pr(author, sha=sha), _visible_approval_reviews(sha=sha)]
|
||||
|
||||
def _eligibility_merge_reads(self, author="author-bot", sha="abc123"):
|
||||
"""Eligibility user/PR plus #695 merge-approval feedback PR+reviews."""
|
||||
return [
|
||||
{"login": "merger-bot"},
|
||||
self._pr(author, sha=sha),
|
||||
*self._feedback_reads(author=author, sha=sha),
|
||||
]
|
||||
|
||||
# -- success --------------------------------------------------------------
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(),
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
*self._feedback_reads(),
|
||||
{}, # merge POST
|
||||
{"merged_commit_sha": "mergecommit99"}, # read-back
|
||||
@@ -872,7 +864,7 @@ class TestMergePR(unittest.TestCase):
|
||||
self.assertEqual(r["merge_method"], "squash")
|
||||
self.assertEqual(r["merge_commit"], "mergecommit99")
|
||||
# 5th call is the merge POST with the requested method/title/message.
|
||||
merge_call = mock_api.call_args_list[6]
|
||||
merge_call = mock_api.call_args_list[4]
|
||||
self.assertEqual(merge_call.args[0], "POST")
|
||||
self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge"))
|
||||
payload = merge_call.args[3]
|
||||
@@ -885,7 +877,7 @@ class TestMergePR(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_expected_changed_files_match_allows(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(),
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
[{"filename": "a.py"}, {"filename": "b.py"}], # files
|
||||
*self._feedback_reads(),
|
||||
{}, # merge POST
|
||||
@@ -907,7 +899,7 @@ class TestMergePR(unittest.TestCase):
|
||||
def test_readback_failure_reports_skipped_cleanup(self, _auth, mock_api):
|
||||
"""Merge OK + read-back GET failure => explicit cleanup skip, not silence."""
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(),
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
*self._feedback_reads(),
|
||||
{}, # merge POST
|
||||
RuntimeError("HTTP 502: Gitea upstream unavailable"), # read-back fails
|
||||
@@ -927,7 +919,7 @@ class TestMergePR(unittest.TestCase):
|
||||
self.assertEqual(r["cleanup_status"], "skipped (merge read-back failed)")
|
||||
# No tracker-cleanup API traffic after the failed read-back:
|
||||
# user, PR (eligibility), feedback PR+reviews, merge POST, read-back.
|
||||
self.assertEqual(mock_api.call_count, 8)
|
||||
self.assertEqual(mock_api.call_count, 6)
|
||||
for c in mock_api.call_args_list:
|
||||
self.assertNotEqual(c.args[0], "DELETE")
|
||||
|
||||
@@ -938,7 +930,7 @@ class TestMergePR(unittest.TestCase):
|
||||
def test_cleanup_exception_surfaced_and_redacted(self, _auth, mock_api, _cleanup):
|
||||
"""Unexpected cleanup exception => merge still succeeds; error surfaced redacted."""
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(),
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
*self._feedback_reads(),
|
||||
{}, # merge POST
|
||||
{"merged_commit_sha": "c9"}, # read-back OK
|
||||
@@ -1150,10 +1142,8 @@ class TestMergePR(unittest.TestCase):
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_head_sha_mismatch_blocks(self, _auth, mock_api):
|
||||
# Eligibility (#695) needs approval feedback before head-gate runs.
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(sha="abc123"),
|
||||
]
|
||||
{"login": "merger-bot"}, self._pr("author-bot", sha="abc123")]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-merger",
|
||||
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
@@ -1172,7 +1162,7 @@ class TestMergePR(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_changed_files_mismatch_blocks(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(),
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
[{"filename": "a.py"}, {"filename": "c.py"}], # actual files
|
||||
]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-merger",
|
||||
@@ -1203,7 +1193,7 @@ class TestMergePR(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_output_redacts_secrets(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(),
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
*self._feedback_reads(),
|
||||
{}, {"merged_commit_sha": "c1"},
|
||||
]
|
||||
@@ -1223,7 +1213,7 @@ class TestMergePR(unittest.TestCase):
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_merge_error_message_redacts_credential(self, _auth, mock_api):
|
||||
mock_api.side_effect = [
|
||||
*self._eligibility_merge_reads(),
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
*self._feedback_reads(),
|
||||
RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
|
||||
]
|
||||
@@ -1244,7 +1234,6 @@ class TestMergePR(unittest.TestCase):
|
||||
def test_merge_blocked_on_stale_approval_head(self, _auth, mock_api):
|
||||
old_sha = "8b61c4b41f1b49b271ed3b99657431cf06eeda3e"
|
||||
new_sha = "3e4b721d60e97147ba0704773cf57cd0d42cbe31"
|
||||
# #695: eligibility itself now fails closed on stale approval head.
|
||||
mock_api.side_effect = [
|
||||
{"login": "merger-bot"}, self._pr("author-bot", sha=new_sha),
|
||||
self._pr("author-bot", sha=new_sha),
|
||||
@@ -1267,7 +1256,6 @@ class TestMergePR(unittest.TestCase):
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_merge_blocked_without_visible_approval(self, _auth, mock_api):
|
||||
# #695: eligibility denies merge when no non-quarantined APPROVED review.
|
||||
mock_api.side_effect = [
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
self._pr("author-bot"),
|
||||
@@ -1282,21 +1270,12 @@ class TestMergePR(unittest.TestCase):
|
||||
)
|
||||
self.assertFalse(r["performed"])
|
||||
self.assertFalse(r.get("approval_visible"))
|
||||
self.assertTrue(
|
||||
any(
|
||||
"no non-quarantined APPROVED review" in x
|
||||
or "no visible APPROVED review" in x
|
||||
or "merge eligibility denied" in x
|
||||
for x in r["reasons"]
|
||||
),
|
||||
msg=r["reasons"],
|
||||
)
|
||||
self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
|
||||
self._assert_no_merge_call(mock_api)
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_merge_blocked_on_request_changes(self, _auth, mock_api):
|
||||
# #695: eligibility denies merge on undismissed REQUEST_CHANGES.
|
||||
mock_api.side_effect = [
|
||||
{"login": "merger-bot"}, self._pr("author-bot"),
|
||||
self._pr("author-bot"),
|
||||
|
||||
@@ -54,26 +54,6 @@ class TestMergeApprovalGate(unittest.TestCase):
|
||||
self.assertFalse(result["approval_at_current_head"])
|
||||
self.assertIsNone(result["latest_approved_head_sha"])
|
||||
|
||||
def test_quarantined_approval_void_for_merge(self):
|
||||
"""#695: contaminated formal review at head must not authorize merge."""
|
||||
result = assess_merge_approval_head(
|
||||
current_head_sha=HEAD_NEW,
|
||||
latest_by_reviewer={
|
||||
"sysadmin": {
|
||||
"verdict": "APPROVED",
|
||||
"dismissed": False,
|
||||
"reviewed_head_sha": HEAD_NEW,
|
||||
"review_id": 427,
|
||||
"submitted_at": "2026-07-13T07:20:00Z",
|
||||
}
|
||||
},
|
||||
quarantined_review_ids={427},
|
||||
)
|
||||
self.assertFalse(result["approval_at_current_head"])
|
||||
self.assertEqual(result["quarantined_approvals_at_current_head"], 1)
|
||||
self.assertIn("quarantined", result["stale_approval_block_reason"])
|
||||
self.assertIn("#695", result["stale_approval_block_reason"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -206,20 +206,7 @@ class TestEligibilityNormalizesOperations(unittest.TestCase):
|
||||
def test_namespaced_profile_ops_allow_legacy_action(self, _auth, mock_api):
|
||||
# JSON-config profiles carry canonical namespaced ops; the raw action
|
||||
# "merge" must still match them after normalization.
|
||||
# #695: merge eligibility also loads quarantine-aware review feedback.
|
||||
mock_api.side_effect = [
|
||||
{"login": "merger-bot"},
|
||||
self._pr("author-bot"),
|
||||
self._pr("author-bot"),
|
||||
[{
|
||||
"id": 1,
|
||||
"user": {"login": "reviewer-bot"},
|
||||
"state": "APPROVED",
|
||||
"commit_id": "abc123",
|
||||
"submitted_at": "2026-07-06T10:00:00Z",
|
||||
"dismissed": False,
|
||||
}],
|
||||
]
|
||||
mock_api.side_effect = [{"login": "merger-bot"}, self._pr("author-bot")]
|
||||
env = {"GITEA_PROFILE_NAME": "gitea-merger",
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.merge"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
|
||||
@@ -76,17 +76,13 @@ class TestPreflightReadSurvival(unittest.TestCase):
|
||||
self.assertIn("task mismatch", str(ctx.exception))
|
||||
|
||||
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(
|
||||
"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:
|
||||
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))
|
||||
|
||||
def test_whoami_recovery_after_violation_clears_capability(self):
|
||||
|
||||
@@ -86,21 +86,9 @@ class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
|
||||
):
|
||||
srv._preflight_resolved_role = "author"
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
try:
|
||||
res = srv.gitea_create_issue(title="Test", body="body")
|
||||
except RuntimeError as exc:
|
||||
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()
|
||||
)
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.gitea_create_issue(title="Test", body="body")
|
||||
self.assertIn("stable control checkout", str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -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 == []
|
||||
@@ -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,
|
||||
}
|
||||
Reference in New Issue
Block a user