Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 324b4b3e93 feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610)
Master-parity previously compared only the daemon's startup commit against the
local on-disk HEAD. When the checkout was not pulled, parity reported green even
though the live remote master had advanced, so a stale daemon could claim a
mutation-safe result while running outdated capability gates (observed during
PR #592 recovery, where the resolver correctly required restart but parity said
in_parity=true).

Changes:
- master_parity_gate.assess_master_parity() gains an optional live_remote_head
  and reports the three commits distinctly (daemon_start_head, local_head,
  live_remote_head) plus live_known / live_stale / mutation_safe. A result is
  mutation_safe only when daemon, local checkout, and live remote all agree.
- parity_block_reasons() now blocks mutations on live-staleness too; read-only
  operations remain unblocked (non-goal: never block diagnostics offline).
- New parity_resolver_disagreement(): typed fail-closed blocker naming the
  capability resolver as authoritative when it requires restart but parity
  looks locally green.
- New read_remote_master_head(): best-effort `git ls-remote` for the live
  target, cached with a 60s TTL (bounded offline latency, no network probe per
  gate call); env override GITEA_TEST_LIVE_REMOTE_HEAD keeps tests hermetic.
- Server: _current_master_parity() reads the live remote head;
  gitea_assess_master_parity and runtime_context surface the distinguished
  SHAs, mutation_safe, and resolver-authoritative guidance.

Tests: 13 new cases (live-remote parity, live-stale blocking + typed blocker,
remote-head reader + TTL cache, server wiring). Full suite: 2423 passed; the 8
remaining failures (test_config TestAuthIntegration, test_credentials
TestGetCredentials) are baseline-proven keychain/env failures identical on the
unmodified base.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-09 18:52:59 -04:00
17 changed files with 481 additions and 4028 deletions
-1151
View File
File diff suppressed because it is too large Load Diff
@@ -1,53 +0,0 @@
# Control-plane DB substrate (#613)
**Status:** Implemented (SQLite single-writer MVP)
**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md)
**Module:** `control_plane_db.py`
## Architecture statement
> **DB coordinates, Gitea records, Sentry/GlitchTip observe; the bridge is the only path that turns observations into Gitea work.**
## What this ships
| Capability | Notes |
|------------|--------|
| Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links` |
| Atomic assign+lease | `ControlPlaneDB.assign_and_lease` — one `BEGIN IMMEDIATE` transaction |
| Mutation gate | `require_valid_assignment` — live lease + allowed action + non-terminal work + non-stale head |
| Heartbeat / release / expire | Lease lifecycle helpers |
| Terminal-lock index | Routing signal for #600 (terminal path first) |
| `incident_links` | Provider-neutral link model for #612**not** assignable work; scope keys NULL-safe |
## Hard rules (enforced in code)
1. Assignable `work_items.kind` ∈ {`issue`, `pr`} only — **never** raw Sentry/GlitchTip incidents.
2. Two concurrent sessions cannot both receive an active assignment on the same open work item (second gets `wait`).
3. Merged/closed work items return `no_safe_work` at assign time, and `require_valid_assignment` fails closed if the work item becomes terminal later.
4. Assignments pin `expected_head_sha`; mutations fail closed if the work item head drifts.
5. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6).
6. `incident_links` optional scope fields are stored as empty strings (never NULL) so UNIQUE is canonical across minimal upserts.
7. Legacy `incident_links` migration collapses NULL-scope duplicates **only** when Gitea targets **and** all meaningful observation metadata agree (fingerprint, status, event_count, permalink, timestamps, linked PRs, etc.). Conflicting metadata fails closed — no silent discard.
## Dependency chain
```text
#613 control-plane DB (this) → #600 allocator API → #612 incident bridge
```
- **#600** must call this substrate (not file locks / comment-only leases alone) for completion.
- **#612** must write `incident_links` here and create **Gitea issues**; the allocator assigns those issues, not raw incidents.
## Non-goals (intentionally deferred)
- Full `gitea_allocate_next_work` routing policy (REQUEST_CHANGES → author, etc.) — **#600**
- Sentry/GlitchTip provider adapters and auto-watchdog — **#612**
- Gitea comment/label mirror writers (may be added by allocator tooling later)
## Tests
```bash
python3 -m pytest tests/test_control_plane_db.py -q
```
@@ -1,301 +0,0 @@
# ADR: MCP allocator, control-plane DB, and Sentry/GlitchTip incident bridge architecture
- **Status:** Accepted (implementation pending; blocks code for #600 / #612 / #613)
- **Date:** 2026-07-09
- **Tracking issues:**
- [#613](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/613) — control-plane DB (first)
- [#600](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/600) — allocator API (second)
- [#612](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/612) — Sentry/GlitchTip incident bridge (third)
- **Related:** existing GlitchTip contracts (`glitchtip-to-gitea-workflow-design.md`, `glitchtip-gitea-deduplication-linking-design.md`), trust boundaries (`tool-boundaries.md`, `safety-model.md`), current leases (`pr_work_lease.py`, `issue_lock_store.py`)
## 1. Context
Multiple LLM sessions can independently inspect the Gitea queue and start the same issue or PR. Existing coordination (Gitea comment leases, local issue-lock files, labels) often detects collisions **after** work has started. Parallel issues #600, #612, and #613 each describe part of a fix; without one ADR they risk overlapping stores, wrong dependency order, and trust-boundary violations.
This ADR is the canonical architecture decision **before** implementing those issues.
## 2. Decision summary (core)
| Layer | Owns | Must not |
|-------|------|----------|
| **Control-plane DB** | Sessions, atomic assignment, leases, heartbeats, terminal-lock index, events, incident links | Bypass Gitea workflow gates or replace issue/PR history |
| **Gitea** | Durable work record: issues, PRs, comments, labels, reviews, merges | Be the only concurrency lock under multi-session load |
| **Sentry / GlitchTip** | Incidents, events, provider UI | Assign work, approve/merge/close, or mutate Gitea outside the bridge |
| **Incident bridge** | Provider adapters, reconcile/create Gitea issues, link storage upsert, optional provider writeback | Hand raw provider incidents to the allocator as work items |
**One-liner:** **DB coordinates. Gitea records. Sentry/GlitchTip observe. The bridge is the only path that turns observations into Gitea work. The allocator assigns only Gitea issues/PRs, never raw monitoring incidents.**
## 3. Dependency order
Implementation **must** follow:
```text
#613 control-plane DB → #600 allocator API → #612 incident bridge
(atomic substrate) (routing policy) (feeds Gitea work)
```
| Issue | Role | Depends on |
|-------|------|------------|
| **#613** | Durable coordination DB + lease/assignment transactions | — |
| **#600** | `gitea_allocate_next_work` policy and tool surface | **#613** (hard) |
| **#612** | Multi-project Sentry/GlitchTip → Gitea bridge | **#613** for `incident_links` index; **#600** before treating bridge-created issues as allocator feed in multi-worker prod |
**Hard rules:**
1. Do **not** implement #600 on file locks / comment-only leases and call it done.
2. Do **not** ship #612 as a second assignment system for raw incidents.
3. Partial previews (read-only list tools, schema stubs) may land earlier if they do not claim “allocator complete” or “bridge complete.”
## 4. Authority boundaries
### 4.1 Gitea (durable source of truth for work)
Gitea remains authoritative for:
- Issue and PR identity, titles, bodies, state (open/closed/merged)
- Comments, reviews, approvals, REQUEST_CHANGES
- Labels and workflow status labels (`status:ready`, `status:in-progress`, …)
- Merges, closes, branch refs as recorded by Gitea/git hosting
Operators and auditors read **history** from Gitea.
### 4.2 Control-plane DB (coordination / index)
The control-plane DB is authoritative for **live multi-session coordination**:
- Which session holds which assignment/lease
- Heartbeat freshness and expiry
- Terminal-lock index for routing
- Event log of allocation/lease transitions
- `incident_links` index (see §8)
It is **not** a substitute for Gitea history. When DB and Gitea disagree on durable work state (e.g. PR already merged), **Gitea wins**; the DB is reconciled.
### 4.3 Sentry / GlitchTip (observe)
Providers own incident lifecycle and raw event data. They:
- Must **not** bypass Gitea gates
- Must **not** assign LLM work
- May receive **writeback** of resolution status only through the bridge, when configured and supported
### 4.4 Trust boundaries for credentials
Aligned with `docs/tool-boundaries.md` and `docs/safety-model.md`:
- **One MCP server process per trust boundary** remains the default.
- Monitor API tokens (Sentry/GlitchTip) live in the **bridge/orchestrator boundary** (or a dedicated observability write/read profile), **not** blindly inside every Gitea MCP author/reviewer/merger process.
- Gitea tokens stay on Gitea MCP profiles only.
- Orchestrators compose services; they must not become a single credential pool that silently mixes Gitea write + monitor tokens into every worker.
Tool names such as `gitea_observability_*` may exist as a **namespace façade** only if the runtime still enforces separate credential scopes (e.g. bridge process vs pure Gitea mutation process).
## 5. Allocator rules (#600 on top of #613)
### 5.1 Worker contract
- Workers **do not self-select** work under the standard multi-LLM workflow.
- Workers call **`gitea_allocate_next_work`** (with `apply=true` only when taking work).
- Mutations that claim exclusive work require a **valid assignment and lease** from the control-plane DB.
- Return values include at least: role, issue/PR number, expected head SHA (when applicable), lease ID, expiry, allowed actions, forbidden actions — or a non-assignment outcome (`WAIT`, terminal-path block, no safe work, needs controller, etc.).
### 5.2 Atomic reserve
- **Assignment creation and lease creation happen in one DB transaction.**
- Two concurrent sessions **must not** receive the same issue/PR as assigned work.
- On contention: second session gets **WAIT** or **owner-resume**, never a duplicate lease.
### 5.3 Routing policy
| Condition | Route |
|-----------|--------|
| Current-head **REQUEST_CHANGES** | **Author** (not reviewer/merger) |
| **Stale** approval (approval not on current head) | **Reviewer** |
| **Clean** approval on current head, mergeable | **Merger** |
| Contested / contaminated approval | **Diagnosis / reconciler** (controller path) |
| Active **foreign** lease | **WAIT** or **owner-resume** |
| **Terminal-review** lock present | **Terminal-path resolution first** before downstream review work |
| Merged / closed PR | **Never assign** |
| Issue locked by sanctioned lease | **Never assign** to another worker unless lease cleanup/adoption is sanctioned |
### 5.4 Relationship to Gitea mirrors
After a successful assignment, the allocator (or sanctioned tooling) may update Gitea comments/labels so humans and audits see state. Those mirrors **are not** the sole lock source.
## 6. Control-plane DB topology (#613)
### 6.1 Preferred architecture (multi-daemon / Proxmox)
For true multi-session concurrency (multiple MCP daemons, multiple hosts, or Proxmox VMs):
**Prefer either:**
1. **Shared Postgres** used by all Gitea MCP profiles / allocator callers, **or**
2. A **single allocator daemon** (sole writer to the coordination store) that all workers call over MCP/RPC.
Both satisfy: one transactional authority for “who has this work item.”
### 6.2 SQLite MVP
SQLite is acceptable **only** for a **single-writer MVP**:
- One process owns writes (allocator daemon or single co-located MCP server), **or**
- Documented single-host single-daemon experiments with file locking and no multi-host claims.
SQLite is **not** sufficient to declare multi-session production readiness when four independent LLM sessions talk to four MCP processes without a shared writer.
### 6.3 Suggested entities (normative shape)
Minimum logical entities (names may vary; semantics must not):
1. **sessions** — session_id, role/profile, namespace, pid, started_at, last_heartbeat_at, status
2. **work_items** — remote/org/repo, kind ∈ {`issue`, `pr`}, number, state, priority, current_head_sha, updated_at
3. **leases** — lease_id, work_item_id, session_id, role, phase, expires_at, heartbeat_at, status
4. **assignments** — assignment_id, work_item_id, session_id, allowed_action(s), expected_head_sha, status
5. **terminal_locks** — repo/org, terminal_pr, review_id, decision, status, cleanup_state
6. **events** — event_id, work_item_id, type, message, created_at
7. **incident_links** — provider-neutral link model (see §8)
**Explicit non-kind:** do **not** use `work_items.kind = sentry_incident` (or glitchtip_incident) as an assignable work unit. Incidents become work only after the bridge creates/links a **Gitea issue**.
## 7. Lease migration (avoid split-brain)
### 7.1 Target state
- **Primary** for live coordination: **control-plane DB** leases and assignments.
- **Gitea comment leases** (e.g. `<!-- mcp-review-lease:v1 -->`) and **local issue-lock files** become:
- **mirrors** of DB state for human visibility / recovery, and/or
- written **only** through the allocator (or sanctioned lease tools that update DB + mirror in one workflow).
### 7.2 Migration rules
1. New exclusive claims go through DB assignment+lease first.
2. Comment lease / file lock writers that skip the DB are **deprecated** once #613+#600 land.
3. Reconciler tooling heals: expired DB lease, orphan Gitea comment, orphan local file — fail closed when ambiguous.
4. **Split-brain is a defect:** “DB free + Gitea comment leased” or “DB leased + Gitea free” must be detectable and reconcilable; production paths must not rely on two independent writers.
### 7.3 Heartbeats
- Heartbeats update the **DB**.
- Optional Gitea summaries must avoid comment spam (periodic summary or edit-in-place policy).
## 8. Observability bridge (#612)
### 8.1 Role
The bridge:
1. Scans configured Sentry and/or GlitchTip projects (multi-project mappings).
2. Deduplicates against existing links / Gitea issues.
3. Creates or updates **normal Gitea issues** (labels, sanitized body, provider URL).
4. Upserts **one** canonical `incident_links` row.
5. Optionally writes resolution status back to the provider when supported.
6. Never approves, merges, closes, or otherwise bypasses Gitea workflow gates.
### 8.2 Allocator interaction
- The allocator sees observability work **only after** a Gitea issue exists (typically `status:ready` + observability labels).
- **Do not assign raw Sentry/GlitchTip incidents** as work items.
- Bridge AC “allocator can see linked issues” means: **as ordinary Gitea issues**, not a special incident queue.
### 8.3 Provider adapters and trust
- Separate **Sentry** and **GlitchTip** adapters; document API differences; do not assume writeback parity.
- Self-hosted base URLs supported (e.g. `https://sentry.prgs.cc`).
- Tokens from environment / secret store only.
- Prefer phase-1 **explicit reconcile / dry-run** before unsupervised watchdog auto-filing, consistent with existing GlitchTip filing safety contracts.
### 8.4 Home of implementation
Filing/orchestration may live in:
- a control-plane / bridge package or profile, and/or
- Gitea-Tools as operator-facing tools that **delegate** to that boundary,
but must not violate §4.4 credential isolation.
## 9. Incident link storage (single source of truth)
### 9.1 Canonical model: `incident_links` (provider-neutral)
Prefer a **provider-neutral** model over Sentry-only `sentry_links`:
| Field (logical) | Purpose |
|-----------------|---------|
| provider | `sentry` \| `glitchtip` \| … |
| provider_base_url | Self-hosted base |
| provider_org / provider_project | Mapping key |
| provider_issue_id / provider_short_id | Provider identity |
| provider_permalink | Human URL |
| fingerprint | Stable dedupe key when available |
| gitea_org / gitea_repo / gitea_issue_number | Linked work |
| linked_pr_numbers | Optional PR association |
| first_seen / last_seen / event_count | Summary metrics (safe) |
| status | link lifecycle |
| release_resolved_at / last_sync_at | Sync bookkeeping |
### 9.2 Gitea as human mirror
- Issue body markers / structured comments (e.g. HTML comment metadata) remain the **human- and audit-visible mirror**.
- They must stay consistent with `incident_links` but are **not** a second independent write path for inventing links without the bridge.
### 9.3 One truth
- **Do not** maintain two independent systems of record (e.g. full mapping only in #612 storage **and** a separate #613 `sentry_links` with different semantics).
- #612 implementation **must** use the same `incident_links` model introduced under #613 (or a single agreed store both reference).
## 10. Security and redaction
Mandatory for bridge payloads, Gitea issue bodies/comments, DB-stored event summaries, and logs:
- No API tokens, DSNs, passwords, cookies, auth headers
- No keychain IDs, private config contents, raw session-state files, full prompt bodies
- Sanitize stack locals and request data; store only safe tags/context
- Logs must never print provider tokens or DSNs
- Redaction tests are required for #612
## 11. Non-goals
- Replace Gitea as the durable workflow record
- Let Sentry/GlitchTip assign work or mutate Gitea workflow state outside sanctioned tools
- Let the bridge approve, merge, close, release, or bypass Gitea gates
- Implement #600 on file locks / comments alone and call allocator complete
- Treat raw monitoring incidents as `work_items` for the allocator
- Put monitor tokens into every Gitea MCP worker process by default
## 12. Consequences
### Positive
- Clear implementation order and ownership
- Multi-session safety becomes testable at the DB layer
- Observability becomes assignable work without special-casing the allocator
- Trust boundaries stay compatible with existing control-plane docs
### Costs / risks
- Migration from comment/file leases requires reconciler work
- Shared Postgres or single allocator daemon is operational cost
- Bridge must be careful not to spam Gitea issues (dedupe, caps, policy modes)
### Follow-ups (documentation only until implemented)
1. Update #600, #612, and #613 bodies to **link this ADR** and restate hard dependencies.
2. Implement #613 schema + atomic assign/lease.
3. Implement #600 against the DB.
4. Implement #612 against `incident_links` + Gitea create/update.
5. Deprecate dual writers for leases.
## 13. Acceptance criteria for *this* ADR
This document is accepted when:
1. It is merged into `docs/architecture/` on the default branch.
2. #600 / #612 / #613 (or their PRs) reference this path as the architecture source of truth.
3. No implementation PR for those issues claims completion without conforming to §§211.
## 14. Document history
| Date | Change |
|------|--------|
| 2026-07-09 | Initial ADR: dependency order, authority model, topology, lease migration, bridge, `incident_links`, security, non-goals |
-118
View File
@@ -1,118 +0,0 @@
# Recovering from `client is closing: EOF` on a Gitea MCP namespace (#543)
## Symptom
A tool call through a Gitea MCP namespace — `gitea-author`, `gitea-reviewer`,
`gitea-merger`, or the shared `gitea-tools` namespace — fails immediately with:
```
client is closing: EOF
```
Every subsequent call to that same namespace returns the same error, including
cheap read tools such as `gitea_whoami` and `gitea_list_profiles`. Other MCP
servers registered with the same client (for example `context7`) keep working,
so this is **not** a global MCP-client outage.
## Why this is not a code defect
This failure is a **transport-level** condition in the IDE / MCP client manager,
not a missing or broken tool:
- The tool can be present and registered in the Python `FastMCP` tool manager.
- Direct Python inspection of the server confirms the tool exists.
- Running the server manually and sending JSON-RPC over stdio works fine
(offline spawn) — that path does **not** prove the IDE namespace is healthy.
The client manager entered a closed state after the backing subprocess for that
namespace terminated (or was killed) behind its back. Once closed, the client
does **not** re-spawn the child on the next tool call — it just replays
`client is closing: EOF`. The OS process may even still be alive if a parent
language-server process is holding the stdio pipes open.
This is the canonical "registered in FastMCP ≠ callable through the namespace"
false-ready state. It is distinct from the **stale-runtime** family in #531 /
#544, where the process is reachable but running behind `master`; that case is
detected by the `ps`-based `_check_mcp_runtimes_diagnostics` in
`gitea_mcp_server.py`. The EOF case is a dead/closed transport, not a stale one,
so the `ps` check alone will not surface it.
## Recovery path (canonical — client reconnect only)
Do the steps in order. Stop as soon as a live **client-namespace** call succeeds.
1. **Confirm the blast radius.** Call a cheap read tool on the failing namespace
(`gitea_whoami` or `gitea_list_profiles`). Then call the same tool on a
different MCP server (e.g. `context7`).
- Only the Gitea namespace fails → single-namespace transport close. Continue.
- Every server fails → restart the whole MCP client, not just one namespace.
2. **Reconnect the namespace through the client, not the shell.** Use the IDE /
client MCP-reconnect action for that server entry (in Claude Code:
`/mcp` → reconnect the affected `gitea-*` server). Reconnecting forces the
client to spawn a fresh subprocess and re-open the pipe. This clears the
closed-client state that a bare `kill`/respawn from a terminal does **not**.
3. **Do not "fix" it by importing the server or poking the process.** Reaching
for `python -c 'import gitea_mcp_server ...'`, raw JSON-RPC from a shell,
killing PIDs to force a respawn, or touching MCP config mtimes does **not**
restore the *client's* view of the namespace and violates the daemon-import
guard (#558, `docs/mcp-daemon-import-guard.md`). The only sanctioned repair
is a **client reconnect / relaunch**.
4. **Verify through the same path the workflow will use.** After reconnect, call
the specific tool the blocked workflow needs — not just any tool — through
the target namespace. For a merge that means calling the merger-authorized
adoption/merge tool through `gitea-merger`. A green `gitea_whoami` on one
namespace does **not** prove another namespace or another tool is callable.
Record success with:
```text
gitea_assess_mcp_namespace_health(..., probe_source="client_namespace")
```
5. **If reconnect does not clear it,** relaunch the client entirely, then repeat
step 4. If EOF persists after a full relaunch, the backing subprocess is
failing to start — inspect its stderr / launch config (command path, venv,
`*_MCP_CONFIG`, `*_MCP_PROFILE` env) rather than retrying the call. Still
do not use PID kill or config-touch as the primary recovery.
## Diagnostics to capture when reporting EOF
Include all of these so the failure is actionable and reproducible:
- **Namespace name** that returned EOF (`gitea-author` / `gitea-reviewer` /
`gitea-merger` / `gitea-tools`).
- **Tool** that was called and the **exact** error string.
- **PID** of the backing process (if any) and whether it was still alive
(informational only — not a recovery action).
- **Profile / env** for that namespace (execution profile, `*_MCP_PROFILE`,
worktree binding such as `GITEA_AUTHOR_WORKTREE`).
- **Config path** the client launched the server from.
- Result of the **cross-server control** call (did `context7` succeed?).
## Offline spawn probe (non-authoritative)
`test_mcp_conn.py` performs a full JSON-RPC handshake against a **fresh
subprocess** (`initialize` → `initialized` → `tools/list` → `tools/call`) and
classifies with `probe_source=offline_spawn`. That is useful for offline
launch/registration debugging. It is **not** proof the IDE-managed namespace is
healthy. See `docs/mcp-namespace-health.md`.
## Do-not list during EOF recovery
- Do **not** retry a blocked merge/adoption until the required tool is confirmed
callable through the merger-authorized **client** namespace (see #543).
- Do **not** clean, reset, or rebind a **foreign** worktree to work around the
error.
- Do **not** bypass the namespace with direct imports, raw API/curl, or
in-memory state restoration.
- Do **not** kill MCP PIDs or touch config mtimes as a substitute for client
reconnect.
## Related
- #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
- #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
- `docs/mcp-client-registration.md` — per-server registration contract.
- `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.
-88
View File
@@ -1,88 +0,0 @@
# MCP namespace health diagnostics (#543)
Gitea MCP tools can be registered in the Python FastMCP server while the IDE's
live MCP namespace is still unusable. The failure usually appears as
`client is closing: EOF`, `transport closed`, or an empty response when calling
a tool such as `gitea_whoami`.
Do not treat static tool registration as proof that review or merge workflows
can proceed. A reviewer or merger flow must have **client-namespace** evidence
that the required tool is callable through the configured IDE MCP namespace.
## Probe sources (do not confuse them)
| Source | How obtained | Proves IDE namespace? |
| --- | --- | --- |
| `client_namespace` | Tool call through the IDE-managed MCP client | **Yes** |
| `offline_spawn` | `test_mcp_conn.py` subprocess JSON-RPC handshake | **No** (offline only) |
`gitea_assess_mcp_namespace_health` accepts `probe_source` and only sets
`ide_namespace_proven=true` for `client_namespace` success. Offline spawn
success never clears review/merge mutation gates.
## Client-namespace health check (canonical)
1. Through the IDE client, call a cheap tool on the target namespace
(`gitea_whoami` or `gitea_list_profiles`).
2. Feed the live result into:
```text
gitea_assess_mcp_namespace_health(
namespace="gitea-merger", # or reviewer / author / tools
registered_tools=[...], # optional static list
probe_result={"success": true, "result": {...}},
probe_source="client_namespace",
)
```
3. A healthy client-namespace assessment is recorded in the MCP session and
consulted by **live** `gitea_submit_pr_review` / `gitea_merge_pr` gates.
4. If the probe fails with EOF, recover via **client reconnect only** — see
`docs/mcp-namespace-eof-recovery.md`. Do **not** kill PIDs or touch MCP
config mtimes as a recovery procedure.
## Offline spawn probe (non-authoritative)
```bash
python3 test_mcp_conn.py --config ~/.gemini/config/mcp_config.json
```
This script spawns a **separate** server process from config, performs
JSON-RPC `initialize``tools/list``tools/call`, and classifies the
result with `probe_source=offline_spawn`. Use it for offline debugging of
launch command / registration. It does **not** prove the IDE-managed
namespace is healthy.
By default the script checks:
| Namespace | Required tool |
| --- | --- |
| `gitea-author` | `gitea_whoami` |
| `gitea-reviewer` | `gitea_whoami` |
| `gitea-merger` | `gitea_whoami` |
| `gitea-tools` | `gitea_list_profiles` |
## Recovery (canonical)
When a namespace returns EOF, follow
`docs/mcp-namespace-eof-recovery.md` in order:
1. Confirm blast radius (Gitea namespace vs all MCP servers).
2. **Reconnect the namespace through the client** (IDE reconnect / relaunch).
3. Do not repair via shell imports, raw JSON-RPC, PID kills, or config mtime
touches — those do not restore the client's closed transport.
4. Re-verify the **specific** required tool through the target namespace.
5. Resume review/merge only after a successful `client_namespace` assessment.
## Enforcement
1. **State machine (read-only):** feed `blocks_merge_workflow` from a
`client_namespace` assessment into
`gitea_assess_review_merge_state_machine(live_namespace_broken=...)`.
2. **Live mutations:** `gitea_submit_pr_review` and `gitea_merge_pr` call
`_live_namespace_health_gate` and fail closed when the session has a
recorded unhealthy or non-client probe for the required namespace
(`gitea-reviewer` for review, `gitea-merger` for merge).
When blocked, repair the IDE namespace and re-record a healthy
`client_namespace` assessment before retrying the mutation.
+79 -458
View File
@@ -191,7 +191,6 @@ MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
import namespace_workspace_binding as nwb # noqa: E402
import mcp_namespace_health # noqa: E402
def _preflight_in_test_mode() -> bool:
@@ -1367,26 +1366,8 @@ def cleanup_in_progress_for_pr(pr_payload: dict, remote: str, host: str | None,
# ── Helpers ───────────────────────────────────────────────────────────────────
def _effective_remote(remote: str) -> str:
"""If remote is the default ('dadeschools') but the active profile base_url maps to a known remote, use that remote instead."""
try:
profile = get_profile()
base_url = profile.get("base_url")
if remote == "dadeschools" and base_url:
import urllib.parse
url = urllib.parse.urlparse(base_url)
host = (url.netloc or url.path or "").strip().lower()
for k, v in REMOTES.items():
if v.get("host") == host:
return k
except Exception:
pass
return remote
def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
"""Resolve remote + overrides to (host, org, repo)."""
remote = _effective_remote(remote)
if remote not in REMOTES:
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
profile = REMOTES[remote]
@@ -1420,7 +1401,6 @@ def _resolve_control_plane_guide_target(
if org is not None and repo is not None:
return _resolve(remote, host, org, repo)
remote = _effective_remote(remote)
if remote not in REMOTES:
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
@@ -2844,35 +2824,6 @@ _TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
# remote + profile identity with TTL.
_REVIEW_DECISION_LOCK: dict | None = None
# Session-scoped live MCP namespace health assessments (#543).
# Keyed by namespace name. Only client_namespace entries authorize mutations.
_LIVE_NAMESPACE_HEALTH: dict[str, dict] = {}
def _record_live_namespace_health(assessment: dict | None) -> None:
"""Store a namespace health assessment for mutation gates."""
if not isinstance(assessment, dict):
return
ns = str(assessment.get("namespace") or "").strip()
if not ns:
return
_LIVE_NAMESPACE_HEALTH[ns] = {
"namespace": ns,
"healthy": bool(assessment.get("healthy")),
"ide_namespace_proven": bool(assessment.get("ide_namespace_proven")),
"probe_source": assessment.get("probe_source"),
"blocks_merge_workflow": bool(assessment.get("blocks_merge_workflow")),
"error_type": assessment.get("error_type"),
"required_tool": assessment.get("required_tool"),
}
def _live_namespace_health_gate(task: str) -> list[str]:
"""Fail closed on recorded broken/non-client IDE namespace health (#543)."""
return mcp_namespace_health.mutation_gate_from_session(
task, _LIVE_NAMESPACE_HEALTH
)
def _decision_lock_binding(lock: dict | None = None) -> dict:
"""Resolve key fields for durable decision-lock storage."""
@@ -3563,11 +3514,6 @@ def _evaluate_pr_review_submission(
reasons.extend(workflow_blockers)
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
return result
if live:
ns_gate = _live_namespace_health_gate("review_pr")
if ns_gate:
reasons.extend(ns_gate)
return result
if action not in _REVIEW_ACTIONS:
reasons.append(
@@ -4216,335 +4162,6 @@ def gitea_submit_pr_review(
)
@mcp.tool()
def gitea_save_review_draft(
pr_number: int,
action: str,
body: str = "",
expected_head_sha: str | None = None,
validation_commands: str | None = None,
validation_results: str | None = None,
blocker_reason: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Save a prepared review verdict as non-terminal draft evidence without submitting a formal review.
This permits resuming the review later (e.g. after a stale runtime restart or connection drop).
"""
action = (action or "").strip().lower()
if action not in _REVIEW_ACTIONS:
return {
"success": False,
"reasons": [
f"unknown review action '{action}'; expected one of "
f"{sorted(_REVIEW_ACTIONS)}"
],
}
# 1. Resolve remote context
try:
h, resolved_org, resolved_repo = _resolve(remote, host, org, repo)
except ValueError as exc:
return {"success": False, "reasons": [str(exc)]}
auth = _auth(h)
# 2. Fetch current PR details to get target base branch and base branch SHA
try:
pr_url = f"{repo_api_url(h, resolved_org, resolved_repo)}/pulls/{pr_number}"
pr = api_request("GET", pr_url, auth) or {}
except Exception as exc:
return {
"success": False,
"reasons": [f"failed to fetch PR #{pr_number} from Gitea: {str(exc)}"],
}
current_head = (pr.get("head") or {}).get("sha")
target_branch = (pr.get("base") or {}).get("ref")
target_branch_sha = (pr.get("base") or {}).get("sha")
if expected_head_sha and current_head and expected_head_sha != current_head:
return {
"success": False,
"reasons": [
f"expected_head_sha '{expected_head_sha}' does not match current PR head SHA '{current_head}'"
],
}
# 3. Resolve worktree path
try:
resolved_worktree = _resolve_preflight_workspace_path(worktree_path)
except Exception as exc:
return {"success": False, "reasons": [f"failed to resolve worktree path: {str(exc)}"]}
# 4. Save payload to durable session state
profile = get_profile()
payload = {
"pr_number": pr_number,
"action": action,
"body": body,
"head_sha": expected_head_sha or current_head,
"target_branch": target_branch,
"target_branch_sha": target_branch_sha,
"worktree_path": resolved_worktree,
"validation_commands": validation_commands,
"validation_results": validation_results,
"blocker_reason": blocker_reason,
"saved_by_identity": _authenticated_username(h) or "",
"saved_by_profile": profile.get("profile_name"),
"remote": remote,
"org": resolved_org,
"repo": resolved_repo,
}
try:
mcp_session_state.save_state(
kind=mcp_session_state.KIND_REVIEW_DRAFT,
payload=payload,
remote=remote,
org=resolved_org,
repo=resolved_repo,
profile_identity=profile.get("profile_name"),
)
except Exception as exc:
return {"success": False, "reasons": [f"failed to save draft state: {str(exc)}"]}
return {
"success": True,
"pr_number": pr_number,
"action": action,
"head_sha": payload["head_sha"],
"target_branch": target_branch,
"target_branch_sha": target_branch_sha,
"worktree_path": resolved_worktree,
}
@mcp.tool()
def gitea_resume_review_draft(
pr_number: int,
submit: bool = False,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Resume a previously saved review draft for the given PR number.
Performs live-state checks (PR head SHA, target branch SHA, terminal lock, active leases,
runtime/master parity, reviewer identity, worktree binding, validation freshness).
Refuses to submit or mark ready if any state has changed unsafely.
"""
profile = get_profile()
active_profile = profile.get("profile_name")
# 1. Load draft state
try:
draft = mcp_session_state.load_state(
kind=mcp_session_state.KIND_REVIEW_DRAFT,
remote=remote,
org=org,
repo=repo,
profile_identity=active_profile,
)
except Exception as exc:
return {"success": False, "reasons": [f"failed to load draft state: {str(exc)}"]}
if not draft:
return {
"success": False,
"reasons": [f"no review draft found for PR #{pr_number} under profile '{active_profile}'"],
}
if draft.get("pr_number") != pr_number:
return {
"success": False,
"reasons": [
f"loaded draft is for PR #{draft.get('pr_number')}, but requested PR #{pr_number}"
],
}
# 2. Resolve remote context
try:
h, resolved_org, resolved_repo = _resolve(remote, host, org, repo)
except ValueError as exc:
return {"success": False, "reasons": [str(exc)]}
auth = _auth(h)
# 3. Fetch live PR state
try:
pr_url = f"{repo_api_url(h, resolved_org, resolved_repo)}/pulls/{pr_number}"
pr = api_request("GET", pr_url, auth) or {}
except Exception as exc:
return {
"success": False,
"reasons": [f"failed to fetch PR #{pr_number} from Gitea: {str(exc)}"],
}
# 4. Perform live-state checks
reasons = []
# PR must be open
if pr.get("state") != "open":
reasons.append(f"PR #{pr_number} is not open (state: '{pr.get('state')}')")
# Live head SHA must match saved head_sha
live_head = (pr.get("head") or {}).get("sha")
saved_head = draft.get("head_sha")
if live_head != saved_head:
reasons.append(
f"PR head SHA changed (live={live_head!r}, saved={saved_head!r})"
)
# Live target branch SHA must match saved target_branch_sha
live_target_sha = (pr.get("base") or {}).get("sha")
saved_target_sha = draft.get("target_branch_sha")
if live_target_sha != saved_target_sha:
reasons.append(
f"target branch SHA changed (live={live_target_sha!r}, saved={saved_target_sha!r})"
)
# Check #332 terminal lock
hard_stop = terminal_review_hard_stop_reasons(pr_number, "resume")
if hard_stop:
reasons.extend(hard_stop)
# Check active reviewer lease (must be claimed by the current session)
try:
comments = _fetch_pr_comments(pr_number, remote=remote, host=host, org=resolved_org, repo=resolved_repo)
except Exception as exc:
reasons.append(f"failed to fetch PR comments: {str(exc)}")
comments = []
active_lease = reviewer_pr_lease.find_active_reviewer_lease(comments, pr_number=pr_number)
if not active_lease:
reasons.append(f"no active reviewer lease found on PR #{pr_number}")
else:
# Check that lease owner is our session_id
session = reviewer_pr_lease.get_session_lease()
session_id = (session or {}).get("session_id")
owner_session_id = active_lease.get("session_id")
if owner_session_id != session_id:
reasons.append(
f"active reviewer lease belongs to session_id={owner_session_id!r}, "
f"but current session has session_id={session_id!r}"
)
# Check runtime/master parity
if "PYTEST_CURRENT_TEST" not in os.environ or "GITEA_FORCE_MCP_RUNTIME_CHECK" in os.environ:
config = gitea_config.load_config()
matching_profiles = []
if config and "profiles" in config:
for p_name, p_data in config["profiles"].items():
p_allowed = p_data.get("allowed_operations") or []
p_forbidden = p_data.get("forbidden_operations") or []
p_allowed_n = []
for op in p_allowed:
try:
p_allowed_n.append(gitea_config.normalize_operation(op))
except Exception:
pass
p_forbidden_n = []
for op in p_forbidden:
try:
p_forbidden_n.append(gitea_config.normalize_operation(op))
except Exception:
pass
# Check for "review_pr" or similar capability
ok, _ = gitea_config.check_operation("gitea.pr.review", p_allowed_n, p_forbidden_n)
if ok:
matching_profiles.append(p_name)
runtime_reasons = _check_mcp_runtimes_diagnostics("review_pr", matching_profiles)
if runtime_reasons:
reasons.extend(runtime_reasons)
# Check reviewer identity
identity = _authenticated_username(h)
if identity != draft.get("saved_by_identity"):
reasons.append(
f"reviewer identity mismatch (active={identity!r}, saved={draft.get('saved_by_identity')!r})"
)
# Check worktree binding
try:
resolved_worktree = _resolve_preflight_workspace_path(worktree_path)
except Exception as exc:
reasons.append(f"failed to resolve current worktree: {str(exc)}")
resolved_worktree = None
if resolved_worktree != draft.get("worktree_path"):
reasons.append(
f"worktree binding mismatch (current={resolved_worktree!r}, saved={draft.get('worktree_path')!r})"
)
if reasons:
return {"success": False, "reasons": reasons}
# 5. All checks pass! Mark decision lock as ready.
action = draft.get("action")
body = draft.get("body")
saved_head = draft.get("head_sha")
lock = _load_review_decision_lock() or {}
lock["final_review_decision_ready"] = True
lock["ready_pr_number"] = pr_number
lock["ready_action"] = action
lock["ready_expected_head_sha"] = saved_head
lock["ready_remote"] = remote
lock["ready_org"] = resolved_org
lock["ready_repo"] = resolved_repo
_save_review_decision_lock(lock)
submitted = False
submission_res = None
if submit:
# Submit the review!
submission_res = gitea_submit_pr_review(
pr_number=pr_number,
action=action,
body=body,
expected_head_sha=saved_head,
remote=remote,
host=host,
org=resolved_org,
repo=resolved_repo,
final_review_decision_ready=True,
worktree_path=resolved_worktree,
)
if submission_res.get("performed") or submission_res.get("success"):
submitted = True
# Delete the draft payload on success
mcp_session_state.save_state(
kind=mcp_session_state.KIND_REVIEW_DRAFT,
payload=None,
remote=remote,
org=resolved_org,
repo=resolved_repo,
profile_identity=active_profile,
)
else:
# If submit fails, return the submit failure reasons
return {
"success": False,
"reasons": submission_res.get("reasons") or ["submission failed"],
"submission_result": submission_res,
}
return {
"success": True,
"pr_number": pr_number,
"action": action,
"marked_ready": True,
"submitted": submitted,
"submission_result": submission_res,
}
@mcp.tool()
def gitea_edit_pr(
pr_number: int,
@@ -4998,12 +4615,6 @@ def gitea_merge_pr(
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
return result
# Gate 0b — recorded live MCP namespace health must not be broken (#543).
ns_gate = _live_namespace_health_gate("merge_pr")
if ns_gate:
reasons.extend(ns_gate)
return result
# Gate 1 — valid merge method (no API call on a bad method).
if do not in _MERGE_METHODS:
reasons.append(
@@ -6998,10 +6609,39 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
return False
def _git_default_remote_name(root: str) -> str:
"""First configured git remote name for *root*, defaulting to 'origin'.
Used to resolve the live remote master target for parity (#610). Best
effort: any failure falls back to 'origin' so callers never raise.
"""
try:
res = subprocess.run(
["git", "-C", root, "remote"],
capture_output=True, text=True, check=False,
)
except Exception:
return "origin"
if res.returncode != 0:
return "origin"
names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
return names[0] if names else "origin"
def _current_master_parity() -> dict:
"""Assess this process's code against the on-disk master HEAD (#420)."""
"""Assess this process's code against local and live remote master (#420/#610).
Compares the daemon's startup commit, the on-disk checkout HEAD, and the
live remote master target. A stale daemon relative to live master fails
closed for mutations even when the local checkout HEAD still matches the
startup commit. The live-remote read is best effort: an unresolved live
head leaves read-only diagnostics unblocked but is never mutation-safe.
"""
current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
live_head = master_parity_gate.read_remote_master_head(
PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
return master_parity_gate.assess_master_parity(
_STARTUP_PARITY, current_head, live_remote_head=live_head)
def _master_parity_block(op: str) -> list[str]:
@@ -8769,7 +8409,6 @@ def gitea_whoami(
metadata: profile_name + allowed_operations; never the token).
"""
record_preflight_check("whoami")
remote = _effective_remote(remote)
if remote not in REMOTES:
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
h = host or REMOTES[remote]["host"]
@@ -8878,6 +8517,8 @@ def gitea_get_profile(
"execution_profile": profile.get("execution_profile"),
"auth_source_type": profile.get("auth_source_type"),
# Auth is reported as a status only (#120): the token source *name*
# (env var name / keychain id) joins endpoint URLs behind the
# GITEA_MCP_REVEAL_ENDPOINTS admin opt-in. Token values never appear.
"auth_status": ("configured" if profile["token_source_name"]
else "unconfigured"),
"remote": remote if remote in REMOTES else None,
@@ -8889,7 +8530,6 @@ def gitea_get_profile(
result["base_url"] = profile["base_url"]
result["server"] = None
remote = _effective_remote(remote)
if remote not in REMOTES:
# Mark ambiguity rather than raising: the tool stays inspectable.
result["identity_status"] = "unknown"
@@ -9013,45 +8653,40 @@ def gitea_assess_review_merge_state_machine(
pre_merge_gates: dict[str, bool] | None = None,
infra_stop: bool = False,
capability_blocked: bool = False,
live_namespace_broken: bool = False,
recovery_handoff_text: str | None = None,
final_report_text: str | None = None,
) -> dict:
"""Read-only: assess enforced PR review/merge workflow state (#290).
``live_namespace_broken`` fails review/merge closed when the live MCP
namespace call path is unusable even though the tool is registered in
FastMCP (#543 AC5). Supply the ``blocks_merge_workflow`` verdict from
``gitea_assess_mcp_namespace_health``.
"""
"""Read-only: assess enforced PR review/merge workflow state (#290)."""
completion = state_completion or {}
blocker_kwargs = {
"infra_stop": infra_stop,
"capability_blocked": capability_blocked,
"live_namespace_broken": live_namespace_broken,
}
blockers = review_merge_state_machine.assess_workflow_blockers(**blocker_kwargs)
blockers = review_merge_state_machine.assess_workflow_blockers(
infra_stop=infra_stop,
capability_blocked=capability_blocked,
)
result = {
"workflow": review_merge_state_machine.workflow_status(
completion,
**blocker_kwargs,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
),
"blockers": blockers,
"approve": review_merge_state_machine.can_approve(
completion,
**blocker_kwargs,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
),
"merge": review_merge_state_machine.can_merge(
completion,
pre_merge_gates=pre_merge_gates,
**blocker_kwargs,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
),
}
if target_state:
result["advancement"] = review_merge_state_machine.assess_state_advancement(
completion,
target_state=target_state,
**blocker_kwargs,
infra_stop=infra_stop,
capability_blocked=capability_blocked,
)
if recovery_handoff_text is not None:
result["recovery_handoff"] = (
@@ -9336,15 +8971,34 @@ def gitea_get_runtime_context(
"restart_required": parity["restart_required"],
"startup_head": parity["startup_head"],
"current_head": parity["current_head"],
# #610 distinguished mutation-safety signals:
"daemon_start_head": parity["daemon_start_head"],
"local_head": parity["local_head"],
"live_remote_head": parity["live_remote_head"],
"live_known": parity["live_known"],
"live_stale": parity["live_stale"],
"mutation_safe": parity["mutation_safe"],
"summary": master_parity_gate.format_parity(parity),
"mutation_gate_enforced": not master_parity_gate.gate_disabled(),
# #610: the capability resolver is authoritative for mutation safety;
# local parity alone must never authorize a mutation.
"resolver_authoritative_for_mutation_safety": True,
}
if parity["stale"] and not master_parity_gate.gate_disabled():
safe_next_action = (
"Server code is stale relative to master; restart the Gitea MCP "
"server to load current capability gates before mutating. "
f"({master_parity_gate.format_parity(parity)})"
)
if parity["restart_required"] and not master_parity_gate.gate_disabled():
if parity["live_stale"]:
safe_next_action = (
"Daemon is stale relative to LIVE remote master "
f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
"restart/reconnect the Gitea MCP server before mutating. The "
"capability resolver is authoritative for mutation safety."
)
else:
safe_next_action = (
"Server code is stale relative to master; restart the Gitea MCP "
"server to load current capability gates before mutating. "
f"({master_parity_gate.format_parity(parity)})"
)
result["safe_next_action"] = safe_next_action
if reveal and h:
@@ -9383,12 +9037,19 @@ def gitea_assess_master_parity(
"determinable": parity["determinable"],
"startup_head": parity["startup_head"],
"current_head": parity["current_head"],
# #610 distinguished mutation-safety signals:
"daemon_start_head": parity["daemon_start_head"],
"local_head": parity["local_head"],
"live_remote_head": parity["live_remote_head"],
"live_known": parity["live_known"],
"live_stale": parity["live_stale"],
"mutation_safe": parity["mutation_safe"],
"mutation_gate_enforced": enforced,
"summary": master_parity_gate.format_parity(parity),
"reasons": parity["reasons"],
"process_root": PROJECT_ROOT,
}
if parity["stale"] and enforced:
if parity["restart_required"] and enforced:
out["report"] = master_parity_gate.parity_report(parity)
return out
def gitea_record_pre_review_command(
@@ -9581,46 +9242,6 @@ def gitea_list_profiles() -> dict:
return {"profiles": profiles_out}
@mcp.tool()
def gitea_assess_mcp_namespace_health(
namespace: str,
required_tool: str | None = None,
registered_tools: list[str] | None = None,
probe_result: dict | None = None,
process: dict | None = None,
config_path: str | None = None,
profile: str | None = None,
configured: bool = True,
probe_source: str | None = None,
) -> dict:
"""Classify MCP namespace health for required Gitea tools (#543).
Static FastMCP registration is not enough to prove a namespace works: IDE
clients can keep a registered tool list while live calls fail with
``client is closing: EOF``. Pass live IDE invocation evidence with
``probe_source='client_namespace'``. Offline subprocess probes
(``test_mcp_conn.py``) must use ``probe_source='offline_spawn'`` and never
count as IDE proof.
Assessments are recorded in the session so live
``gitea_submit_pr_review`` / ``gitea_merge_pr`` can fail closed when a
client-namespace probe reported unhealthy.
"""
result = mcp_namespace_health.classify_namespace_probe(
namespace,
required_tool=required_tool,
registered_tools=registered_tools,
probe_result=probe_result,
process=process,
config_path=config_path,
profile=profile,
configured=configured,
probe_source=probe_source,
)
_record_live_namespace_health(result)
return result
@mcp.tool()
def gitea_activate_profile(
profile_name: str,
+171 -17
View File
@@ -24,12 +24,26 @@ from __future__ import annotations
import os
import subprocess
import time
# Live-remote head cache: the parity gate runs on every mutation and every
# runtime-context read, so the ``git ls-remote`` result is cached briefly to
# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
_REMOTE_HEAD_TTL = 60.0
def _clear_remote_head_cache() -> None:
"""Reset the live-remote head cache (test isolation / forced refresh)."""
_REMOTE_HEAD_CACHE.clear()
# Environment escape hatches (ops + tests):
# GITEA_MCP_DISABLE_PARITY_GATE -> disable enforcement entirely (fail open).
# GITEA_TEST_CURRENT_HEAD -> force the "current" HEAD read, for tests.
ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
def read_git_head(root: str) -> str | None:
@@ -58,6 +72,57 @@ def read_git_head(root: str) -> str | None:
return (res.stdout or "").strip() or None
def read_remote_master_head(
root: str,
remote: str = "origin",
branch: str = "master",
ttl: float = _REMOTE_HEAD_TTL,
) -> str | None:
"""Return the live remote ``branch`` commit SHA, or ``None`` (#610).
Resolves the *live* target commit via ``git ls-remote`` so parity can tell
a daemon that is behind the live remote master apart from one whose local
checkout simply hasn't been pulled. ``None`` means the live head could not
be resolved (offline, no such remote, git unavailable, error) -- callers
must treat unknown live state as *not mutation-safe* while never blocking
read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
precedence so the wiring can be exercised deterministically and offline.
The result is cached for *ttl* seconds per (root, remote, branch) so the
gate does not run a network probe on every mutation/read (``ttl=0`` forces
a live probe). Both hits and ``None`` misses are cached to bound offline
latency; the env override bypasses the cache and the subprocess entirely.
"""
forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
if forced is not None:
return forced.strip() or None
if not root:
return None
key = (root, remote, branch)
now = time.monotonic()
if ttl > 0:
cached = _REMOTE_HEAD_CACHE.get(key)
if cached is not None and (now - cached[0]) < ttl:
return cached[1]
sha: str | None = None
try:
res = subprocess.run(
["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
capture_output=True,
text=True,
check=False,
timeout=5,
)
if res.returncode == 0:
lines = (res.stdout or "").strip().splitlines()
if lines:
sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
except Exception:
sha = None
_REMOTE_HEAD_CACHE[key] = (now, sha)
return sha
def capture_startup_parity(root: str, head: str | None = None) -> dict:
"""Capture the process source-tree baseline once at server startup.
@@ -72,18 +137,38 @@ def _short(sha: str | None) -> str:
return sha[:12] if sha else "unknown"
def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
def assess_master_parity(
startup: dict | None,
current_head: str | None,
live_remote_head: str | None = None,
) -> dict:
"""Compare the startup baseline against the current on-disk ``HEAD``.
Pure: both HEADs are supplied by the caller. Returns a structured result:
Pure: all HEADs are supplied by the caller. Returns a structured result:
- ``in_parity`` -- server code matches the on-disk master (or parity
could not be determined, which is not treated as stale).
- ``stale`` -- the on-disk master has definitively advanced past the
running process.
- ``restart_required`` -- alias of ``stale``; the recovery action.
- ``determinable`` -- whether both HEADs were known well enough to compare.
- ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
- ``determinable`` -- whether both local HEADs were known well enough to
compare.
- ``startup_head`` / ``current_head`` / ``reasons``.
#610 adds live-remote awareness so a daemon that is stale relative to the
*live* remote master cannot report a mutation-safe result even when the
local checkout HEAD still matches the daemon's startup commit:
- ``daemon_start_head`` -- the commit the running process started at
(alias of ``startup_head``, named for clarity in reports).
- ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
- ``live_remote_head`` -- the live remote target commit, or ``None`` when it
could not be fetched.
- ``live_known`` -- whether the live remote target was resolved.
- ``live_stale`` -- the live remote master has advanced past the running
process (daemon is behind live master) even if local parity is green.
- ``mutation_safe`` -- the daemon code, local checkout, and live remote
target all agree; the only state in which a mutation may rely on parity.
"""
startup_head = (startup or {}).get("startup_head")
reasons: list[str] = []
@@ -91,32 +176,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
if startup_head is None:
reasons.append(
"startup commit was not captured; code parity cannot be enforced")
return _result(True, False, False, startup_head, current_head, reasons)
return _result(True, False, False, startup_head, current_head,
live_remote_head, False, reasons)
if current_head is None:
reasons.append(
"current workspace HEAD could not be read; code parity cannot be "
"enforced")
return _result(True, False, False, startup_head, current_head, reasons)
return _result(True, False, False, startup_head, current_head,
live_remote_head, False, reasons)
if startup_head == current_head:
return _result(True, False, True, startup_head, current_head, reasons)
local_in_parity = startup_head == current_head
local_stale = not local_in_parity
if local_stale:
reasons.append(
f"MCP server started at commit {_short(startup_head)} but the "
f"workspace master is now {_short(current_head)}; restart the "
f"server to load the current capability gates")
reasons.append(
f"MCP server started at commit {_short(startup_head)} but the workspace "
f"master is now {_short(current_head)}; restart the server to load the "
f"current capability gates")
return _result(False, True, True, startup_head, current_head, reasons)
live_known = live_remote_head is not None
live_stale = live_known and live_remote_head != startup_head
if live_stale:
reasons.append(
f"live remote master is {_short(live_remote_head)} but the MCP "
f"server started at {_short(startup_head)}; the daemon is stale "
f"relative to live master -- restart/reconnect before mutating")
return _result(
local_in_parity, local_stale, True, startup_head, current_head,
live_remote_head, live_stale, reasons)
def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
def _result(in_parity, stale, determinable, startup_head, current_head,
live_remote_head, live_stale, reasons):
live_known = live_remote_head is not None
mutation_safe = (
determinable and in_parity and live_known and not live_stale)
return {
"in_parity": in_parity,
"stale": stale,
"restart_required": stale,
"restart_required": stale or live_stale,
"determinable": determinable,
"startup_head": startup_head,
"current_head": current_head,
# #610 distinguished signals:
"daemon_start_head": startup_head,
"local_head": current_head,
"live_remote_head": live_remote_head,
"live_known": live_known,
"live_stale": live_stale,
"mutation_safe": mutation_safe,
"reasons": list(reasons),
}
@@ -130,11 +239,13 @@ def parity_block_reasons(assessment: dict) -> list[str]:
"""Block reasons for a mutation gate (empty when the mutation may proceed).
A disabled gate or an in-parity / non-determinable assessment yields no
reasons; only a definitively stale server blocks.
reasons. A definitively stale server blocks, and (#610) a daemon that is
stale relative to the *live* remote master blocks even when the local
checkout HEAD still matches the daemon's startup commit.
"""
if gate_disabled():
return []
if assessment.get("stale"):
if assessment.get("stale") or assessment.get("live_stale"):
return list(assessment.get("reasons") or
["server code is stale relative to master (fail closed)"])
return []
@@ -147,6 +258,10 @@ def parity_report(assessment: dict) -> dict:
"restart_required": True,
"startup_head": assessment.get("startup_head"),
"current_head": assessment.get("current_head"),
# #610: name the live remote target so the report distinguishes a
# local-code stale from a daemon-behind-live-master stale.
"live_remote_head": assessment.get("live_remote_head"),
"live_stale": bool(assessment.get("live_stale")),
"reasons": list(assessment.get("reasons") or []),
"recovery": [
"The running MCP server is executing code older than the current "
@@ -157,6 +272,45 @@ def parity_report(assessment: dict) -> dict:
}
def parity_resolver_disagreement(
assessment: dict,
resolver_restart_required: bool,
) -> dict | None:
"""Typed blocker when the resolver requires restart but parity looks green.
The capability resolver (``gitea_resolve_task_capability``) detects stale
runtime authoritatively for mutation safety (#610). When it requires a
restart, local-only parity must never override it: this returns a typed,
fail-closed blocker that names the resolver as authoritative. Returns
``None`` when the resolver does not require a restart.
"""
if not resolver_restart_required:
return None
parity_optimistic = bool(assessment.get("in_parity")) and not (
assessment.get("stale") or assessment.get("live_stale"))
return {
"kind": "parity_resolver_disagreement",
"restart_required": True,
"resolver_authoritative": True,
"parity_optimistic": parity_optimistic,
"daemon_start_head": assessment.get("daemon_start_head"),
"local_head": assessment.get("local_head"),
"live_remote_head": assessment.get("live_remote_head"),
"reasons": [
"The capability resolver requires a restart/reconnect (stale "
"runtime) but master-parity reported local code as in-parity. "
"The resolver is authoritative for mutation safety; do not mutate "
"on local parity alone. Restart/reconnect the Gitea MCP server "
"and re-verify before mutating.",
],
"recovery": [
"Trust the resolver: treat this session as stale.",
"Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
"current master and live target state, then re-run preflight.",
],
}
def format_parity(assessment: dict) -> str:
"""One-line human summary for logs / runtime context."""
if assessment.get("stale"):
-306
View File
@@ -1,306 +0,0 @@
"""Assess live MCP namespace health without trusting static registration.
The IDE/client namespace can fail with EOF even when this Python process still
registers the Gitea tools with FastMCP. These helpers keep that distinction
explicit so reviewer/merger flows can fail closed on the live path.
Probe sources
-------------
* ``client_namespace`` — evidence from the IDE-managed MCP client path (the
only source that can prove the workflow namespace is healthy).
* ``offline_spawn`` — a separate ``subprocess.Popen`` JSON-RPC handshake
(e.g. ``test_mcp_conn.py``). Useful offline, but **never** proves the
IDE-managed namespace is callable.
* ``unknown`` — legacy/unspecified; treated as not IDE-proven.
"""
from __future__ import annotations
from typing import Any
REQUIRED_NAMESPACE_TOOLS = {
"gitea-author": "gitea_whoami",
"gitea-reviewer": "gitea_whoami",
"gitea-merger": "gitea_whoami",
"gitea-tools": "gitea_list_profiles",
}
DEFAULT_NAMESPACES = tuple(REQUIRED_NAMESPACE_TOOLS)
# Namespaces that must be healthy for a given mutation task.
TASK_REQUIRED_NAMESPACES = {
"review_pr": "gitea-reviewer",
"submit_review": "gitea-reviewer",
"merge_pr": "gitea-merger",
}
PROBE_SOURCE_CLIENT = "client_namespace"
PROBE_SOURCE_OFFLINE = "offline_spawn"
PROBE_SOURCE_UNKNOWN = "unknown"
VALID_PROBE_SOURCES = frozenset(
{PROBE_SOURCE_CLIENT, PROBE_SOURCE_OFFLINE, PROBE_SOURCE_UNKNOWN}
)
EOF_PATTERNS = (
"client is closing: eof",
"transport closed",
"connection closed",
"broken pipe",
"end of file",
"eof",
)
SAFE_ENV_KEYS = (
"GITEA_MCP_PROFILE",
"GITEA_PROFILE_NAME",
"GITEA_SERVICE",
"GITEA_EXECUTION_ROLE",
"GITEA_MCP_CONFIG",
)
def _as_list(value: Any) -> list[str] | None:
if value is None:
return None
if isinstance(value, (list, tuple, set)):
return [str(v) for v in value]
return [str(value)]
def _contains_eof(text: str | None) -> bool:
lowered = (text or "").lower()
return any(pattern in lowered for pattern in EOF_PATTERNS)
def _normalize_probe_source(probe_source: str | None) -> str:
raw = (probe_source or PROBE_SOURCE_UNKNOWN).strip().lower()
if raw in VALID_PROBE_SOURCES:
return raw
return PROBE_SOURCE_UNKNOWN
def _safe_env_summary(process: dict[str, Any] | None) -> dict[str, str]:
if not process:
return {}
env = process.get("env") or process.get("environment") or {}
if not isinstance(env, dict):
return {}
return {
key: str(env[key])
for key in SAFE_ENV_KEYS
if key in env and env[key] not in (None, "")
}
def classify_namespace_probe(
namespace: str,
*,
required_tool: str | None = None,
registered_tools: list[str] | tuple[str, ...] | set[str] | None = None,
probe_result: dict[str, Any] | None = None,
process: dict[str, Any] | None = None,
config_path: str | None = None,
profile: str | None = None,
configured: bool = True,
probe_source: str | None = None,
) -> dict[str, Any]:
"""Classify whether a required tool is callable through a live namespace.
``registered_tools`` is static/server-side evidence. ``probe_result`` is
live invocation evidence. Only ``probe_source=client_namespace`` proves the
IDE-managed path; ``offline_spawn`` is an offline subprocess check only.
"""
ns = (namespace or "").strip()
tool = required_tool or REQUIRED_NAMESPACE_TOOLS.get(ns) or "gitea_whoami"
source = _normalize_probe_source(probe_source)
registered_list = _as_list(registered_tools)
registered = None if registered_list is None else tool in registered_list
probe = probe_result or {}
probe_success = bool(probe.get("success"))
error_message = str(
probe.get("error")
or probe.get("message")
or probe.get("stderr")
or probe.get("exception")
or ""
)
error_type = str(probe.get("error_type") or "").strip()
if not error_type and error_message:
if _contains_eof(error_message):
error_type = "namespace_eof"
elif "timeout" in error_message.lower():
error_type = "namespace_timeout"
else:
error_type = "namespace_call_failed"
if not configured:
error_type = "namespace_not_configured"
elif registered is False:
error_type = "tool_missing"
elif not probe_result:
error_type = "live_probe_missing"
elif not probe_success and not error_type:
error_type = "namespace_call_failed"
callable_live = bool(configured and probe_result and probe_success)
# Probe-path health (spawn or client). IDE-proven only for client path.
healthy = bool(configured and registered is not False and callable_live)
ide_namespace_proven = bool(healthy and source == PROBE_SOURCE_CLIENT)
process_pid = process.get("pid") if isinstance(process, dict) else None
profile_name = profile or (
process.get("profile") if isinstance(process, dict) else None
)
env_summary = _safe_env_summary(process)
reasons: list[str] = []
if not configured:
reasons.append(f"MCP namespace '{ns}' is not configured.")
if registered is False:
reasons.append(
f"Required tool '{tool}' is not registered in namespace '{ns}'."
)
if error_type == "live_probe_missing":
reasons.append(
f"No live client invocation proof was supplied for '{ns}.{tool}'."
)
elif error_type == "namespace_eof":
reasons.append(
f"Live MCP namespace '{ns}' returned EOF while invoking '{tool}'."
)
elif error_type == "namespace_timeout":
reasons.append(
f"Live MCP namespace '{ns}' timed out while invoking '{tool}'."
)
elif error_type == "namespace_call_failed":
reasons.append(
f"Live MCP namespace '{ns}' failed while invoking '{tool}'."
)
if source == PROBE_SOURCE_OFFLINE:
reasons.append(
"Probe source is offline_spawn (subprocess JSON-RPC); this does "
"not prove the IDE-managed MCP namespace is healthy."
)
elif source == PROBE_SOURCE_UNKNOWN and probe_result:
reasons.append(
"Probe source unspecified; treat as not IDE-namespace proof unless "
"re-supplied with probe_source='client_namespace'."
)
remediation = []
if not healthy or not ide_namespace_proven:
remediation.append(
"Reconnect the IDE MCP client namespace (client reconnect / "
f"relaunch), then invoke '{tool}' through namespace '{ns}' and "
"record the result with probe_source='client_namespace'."
)
remediation.append(
"Do not treat offline subprocess probes (test_mcp_conn.py) or "
"shell kill/PID respawn as proof the IDE namespace is repaired."
)
if process_pid and source != PROBE_SOURCE_OFFLINE:
remediation.append(
f"Diagnostics may include PID {process_pid}; process details "
"are informational only — recovery is client-layer reconnect."
)
else:
remediation.append(
f"IDE-managed namespace '{ns}' can invoke '{tool}' "
f"(probe_source={source})."
)
# Client-namespace broken health blocks review/merge. Offline probes never
# authorize mutations and only block when they report unhealthy (still
# fail-closed for known bad spawn evidence).
blocks = False
if source == PROBE_SOURCE_CLIENT:
blocks = namespace_health_blocks_task("merge_pr", healthy)
elif source == PROBE_SOURCE_OFFLINE:
# Offline never proves IDE health; never unblock. Unhealthy offline
# still surfaces as a soft diagnostic, not a mutation-ledger block.
blocks = False
else:
# Unknown source: only block when evidence is unhealthy (fail closed
# on bad data without treating success as IDE proof).
blocks = namespace_health_blocks_task("merge_pr", healthy)
return {
"success": healthy,
"healthy": healthy,
"namespace": ns,
"required_tool": tool,
"configured": configured,
"registered_tools_checked": registered_list is not None,
"required_tool_registered": registered,
"required_tool_callable": callable_live,
"probe_source": source,
"ide_namespace_proven": ide_namespace_proven,
"error_type": None if healthy else error_type,
"error_message": error_message or None,
"reasons": reasons,
"remediation": remediation,
"diagnostics": {
"namespace": ns,
"required_tool": tool,
"process_pid": process_pid,
"profile": profile_name,
"env": env_summary,
"config_path": config_path,
"probe_source": source,
},
"blocks_merge_workflow": blocks,
}
def namespace_health_blocks_task(task: str, healthy: bool) -> bool:
"""Return whether a broken namespace must block a workflow task."""
if healthy:
return False
return (task or "").strip() in {"merge_pr", "review_pr", "submit_review"}
def required_namespace_for_task(task: str) -> str | None:
"""Map a mutation task to the MCP namespace that must be healthy."""
return TASK_REQUIRED_NAMESPACES.get((task or "").strip())
def mutation_gate_from_session(
task: str,
session_health: dict[str, dict[str, Any]] | None,
) -> list[str]:
"""Fail-closed gate using recorded client-namespace health assessments.
* Missing session entry → no gate (caller has not assessed yet).
* Client-namespace unhealthy / not IDE-proven → block mutation.
* Offline-only session entries never authorize mutations.
"""
ns = required_namespace_for_task(task)
if not ns:
return []
store = session_health or {}
entry = store.get(ns)
if not entry:
return []
source = _normalize_probe_source(entry.get("probe_source"))
if source != PROBE_SOURCE_CLIENT:
return [
f"recorded namespace health for '{ns}' is probe_source={source}, "
"not client_namespace; re-probe through the IDE-managed path "
f"before {(task or 'mutation')}"
]
if entry.get("ide_namespace_proven") and entry.get("healthy"):
return []
if entry.get("blocks_merge_workflow") or not entry.get("healthy"):
detail = entry.get("error_type") or "unhealthy"
return [
f"live MCP namespace '{ns}' is recorded {detail} "
f"(probe_source={source}); repair the IDE namespace before "
f"{(task or 'mutation')} (fail closed, #543)"
]
if not entry.get("ide_namespace_proven"):
return [
f"live MCP namespace '{ns}' is not IDE-proven; supply a "
"client_namespace probe before mutation (fail closed, #543)"
]
return []
-3
View File
@@ -6,9 +6,6 @@ Runs over stdio. All tools authenticate via macOS keychain (git credential fill)
import os
import sys
sys.stderr = open("/tmp/mcp_server_stderr.log", "a", buffering=1)
sys.stderr.write(f"\n--- MCP SERVER STARTUP (PID {os.getpid()}) ---\n")
from role_session_router import (
python_bytes_have_conflict_markers,
skip_python_scan_walk_root,
-3
View File
@@ -30,9 +30,6 @@ DEFAULT_TTL_HOURS = 4.0
KIND_WORKFLOW_LOAD = "review_workflow_load"
KIND_DECISION_LOCK = "review_decision_lock"
KIND_REVIEW_DRAFT = "review_draft"
_SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
+8 -25
View File
@@ -93,16 +93,8 @@ def assess_workflow_blockers(
capability_blocked: bool = False,
mcp_reconnect_failed: bool = False,
stale_capability_state: bool = False,
live_namespace_broken: bool = False,
) -> dict[str, Any]:
"""Return hard blockers that forbid all PR queue work (#290 AC3).
``live_namespace_broken`` fails the merge/review path closed when the live
MCP namespace call path is unusable (e.g. ``client is closing: EOF``) even
though the tool is registered in FastMCP (#543 AC5). Feed it the
``blocks_merge_workflow`` verdict from
``mcp_namespace_health.classify_namespace_probe``.
"""
"""Return hard blockers that forbid all PR queue work (#290 AC3)."""
reasons: list[str] = []
if infra_stop:
reasons.append("infra_stop is active; PR selection/review/merge is forbidden")
@@ -112,12 +104,6 @@ def assess_workflow_blockers(
reasons.append("MCP reconnect failed; stale session state cannot be reused")
if stale_capability_state:
reasons.append("stale MCP capability state detected after reconnect failure")
if live_namespace_broken:
reasons.append(
"live MCP namespace call path is broken (registered in FastMCP but "
"not callable through the namespace); repair the namespace before "
"review/merge"
)
return {
"block": bool(reasons),
"reasons": reasons,
@@ -132,19 +118,16 @@ def assess_state_advancement(
state_completion: dict[str, bool] | None,
*,
target_state: str,
**blocker_kwargs,
infra_stop: bool = False,
capability_blocked: bool = False,
) -> dict[str, Any]:
"""Fail closed when *target_state* is requested before upstream gates pass.
Forwards every blocker flag (``infra_stop``, ``capability_blocked``,
``mcp_reconnect_failed``, ``stale_capability_state``,
``live_namespace_broken``) to :func:`assess_workflow_blockers` so
``can_approve``/``can_merge`` honor the full blocker set, not just
infra_stop/capability_blocked (#543 AC5).
"""
"""Fail closed when *target_state* is requested before upstream gates pass."""
completion = dict(state_completion or {})
target = _clean(target_state).upper()
blockers = assess_workflow_blockers(**blocker_kwargs)
blockers = assess_workflow_blockers(
infra_stop=infra_stop,
capability_blocked=capability_blocked,
)
reasons = list(blockers["reasons"])
try:
+25 -185
View File
@@ -1,45 +1,20 @@
#!/usr/bin/env python3
"""Offline-only MCP namespace spawn probe (NOT IDE-namespace proof).
"""Live health-check script to verify Gitea MCP namespace connections.
Spawns a *separate* MCP server process from config via subprocess.Popen,
performs the JSON-RPC handshake, verifies the required tool is registered,
and invokes that tool. Results are classified with
``probe_source=offline_spawn``.
This path is useful for offline launch/registration debugging. It does
**not** prove the IDE-managed MCP client namespace is healthy (#543). For
workflow gates, pass live IDE call evidence with
``probe_source=client_namespace`` to ``gitea_assess_mcp_namespace_health``.
Spawns the MCP server processes as defined in the IDE's global config,
performs the JSON-RPC handshake, and queries the tools list to verify
that the connection is fully operational and doesn't return EOF.
"""
import argparse
import json
import os
import subprocess
import sys
from mcp_namespace_health import REQUIRED_NAMESPACE_TOOLS, classify_namespace_probe
def _read_json_line(proc):
line = proc.stdout.readline()
if not line:
stderr_content = proc.stderr.read()
return None, stderr_content
return json.loads(line), None
def _write_message(proc, payload):
proc.stdin.write(json.dumps(payload) + "\n")
proc.stdin.flush()
def run_connection_test(name, config, *, required_tool=None, config_path=None):
def run_connection_test(name, config):
print(f"Testing MCP connection for '{name}'...")
command = config.get("command")
args = config.get("args", [])
env = config.get("env", {})
tool_name = required_tool or REQUIRED_NAMESPACE_TOOLS.get(name) or "gitea_whoami"
# Merge current environment
run_env = os.environ.copy()
@@ -56,17 +31,8 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
bufsize=1,
env=run_env
)
except Exception as exc:
print(f" [FAIL] Failed to spawn process: {exc}")
assessment = classify_namespace_probe(
name,
required_tool=tool_name,
probe_result={"success": False, "error": str(exc)},
process={"profile": env.get("GITEA_MCP_PROFILE"), "env": env},
config_path=config_path,
probe_source="offline_spawn",
)
print(f" diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
except Exception as e:
print(f" [FAIL] Failed to spawn process: {e}")
return False
# Send initialize request
@@ -82,36 +48,26 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
}
try:
_write_message(proc, init_req)
proc.stdin.write(json.dumps(init_req) + "\n")
proc.stdin.flush()
# Read response
res, stderr_content = _read_json_line(proc)
if res is None:
line = proc.stdout.readline()
if not line:
stderr_content = proc.stderr.read()
print(f" [FAIL] Received EOF from process. Stderr:\n{stderr_content}")
assessment = classify_namespace_probe(
name,
required_tool=tool_name,
probe_result={"success": False, "error": stderr_content or "EOF"},
process={
"pid": proc.pid,
"profile": env.get("GITEA_MCP_PROFILE"),
"env": env,
},
config_path=config_path,
probe_source="offline_spawn",
)
print(f" remediation: {' '.join(assessment['remediation'])}")
proc.terminate()
return False
print(f" [OK] Received initialize response: {str(res)[:150]}...")
print(f" [OK] Received initialize response: {line.strip()[:150]}...")
# Send initialized notification
init_notif = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
_write_message(proc, init_notif)
proc.stdin.write(json.dumps(init_notif) + "\n")
proc.stdin.flush()
# Send tools/list request
list_req = {
@@ -120,27 +76,16 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
"params": {},
"id": 2
}
_write_message(proc, list_req)
proc.stdin.write(json.dumps(list_req) + "\n")
proc.stdin.flush()
res, stderr_content = _read_json_line(proc)
if res is None:
line = proc.stdout.readline()
if not line:
print(" [FAIL] Received EOF on tools/list request.")
assessment = classify_namespace_probe(
name,
required_tool=tool_name,
probe_result={"success": False, "error": stderr_content or "EOF"},
process={
"pid": proc.pid,
"profile": env.get("GITEA_MCP_PROFILE"),
"env": env,
},
config_path=config_path,
probe_source="offline_spawn",
)
print(f" remediation: {' '.join(assessment['remediation'])}")
proc.terminate()
return False
res = json.loads(line)
if "error" in res:
print(f" [FAIL] Server returned error: {res['error']}")
proc.terminate()
@@ -149,115 +94,16 @@ def run_connection_test(name, config, *, required_tool=None, config_path=None):
tools = res.get("result", {}).get("tools", [])
tool_names = [t.get("name") for t in tools]
print(f" [OK] Successfully retrieved {len(tool_names)} tools: {tool_names[:5]}...")
if tool_name not in tool_names:
assessment = classify_namespace_probe(
name,
required_tool=tool_name,
registered_tools=tool_names,
probe_result={"success": False, "error": "required tool missing"},
process={
"pid": proc.pid,
"profile": env.get("GITEA_MCP_PROFILE"),
"env": env,
},
config_path=config_path,
probe_source="offline_spawn",
)
print(f" [FAIL] Required tool '{tool_name}' is not registered.")
print(f" remediation: {' '.join(assessment['remediation'])}")
proc.terminate()
return False
call_req = {
"jsonrpc": "2.0",
"method": "tools/call",
"params": {"name": tool_name, "arguments": {}},
"id": 3,
}
_write_message(proc, call_req)
call_res, stderr_content = _read_json_line(proc)
if call_res is None:
assessment = classify_namespace_probe(
name,
required_tool=tool_name,
registered_tools=tool_names,
probe_result={"success": False, "error": stderr_content or "EOF"},
process={
"pid": proc.pid,
"profile": env.get("GITEA_MCP_PROFILE"),
"env": env,
},
config_path=config_path,
probe_source="offline_spawn",
)
print(f" [FAIL] Received EOF on {tool_name} invocation.")
print(f" diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
print(f" remediation: {' '.join(assessment['remediation'])}")
proc.terminate()
return False
if "error" in call_res:
assessment = classify_namespace_probe(
name,
required_tool=tool_name,
registered_tools=tool_names,
probe_result={"success": False, "error": call_res["error"]},
process={
"pid": proc.pid,
"profile": env.get("GITEA_MCP_PROFILE"),
"env": env,
},
config_path=config_path,
probe_source="offline_spawn",
)
print(f" [FAIL] {tool_name} invocation returned error: {call_res['error']}")
print(f" remediation: {' '.join(assessment['remediation'])}")
proc.terminate()
return False
assessment = classify_namespace_probe(
name,
required_tool=tool_name,
registered_tools=tool_names,
probe_result={"success": True, "result": call_res.get("result")},
process={
"pid": proc.pid,
"profile": env.get("GITEA_MCP_PROFILE"),
"env": env,
},
config_path=config_path,
probe_source="offline_spawn",
)
print(f" [OK] Successfully invoked required tool '{tool_name}'.")
print(f" diagnostics: {json.dumps(assessment['diagnostics'], sort_keys=True)}")
proc.terminate()
return True
except Exception as exc:
print(f" [FAIL] Error during handshake: {exc}")
except Exception as e:
print(f" [FAIL] Error during handshake: {e}")
proc.terminate()
return False
def main():
parser = argparse.ArgumentParser()
parser.add_argument(
"--config",
default=os.environ.get(
"MCP_CONFIG_PATH",
os.path.expanduser("~/.gemini/config/mcp_config.json"),
),
help="Path to MCP config JSON.",
)
parser.add_argument(
"--namespace",
action="append",
dest="namespaces",
help="Namespace to test. May be repeated.",
)
args = parser.parse_args()
config_path = args.config
config_path = "/Users/jasonwalker/.gemini/config/mcp_config.json"
try:
with open(config_path) as f:
mcp_config = json.load(f)
@@ -266,16 +112,10 @@ def main():
sys.exit(1)
servers = mcp_config.get("mcpServers", {})
namespaces = args.namespaces or list(REQUIRED_NAMESPACE_TOOLS)
failed = False
for name in namespaces:
for name in ["gitea-author", "gitea-reviewer"]:
if name in servers:
if not run_connection_test(
name,
servers[name],
required_tool=REQUIRED_NAMESPACE_TOOLS.get(name),
config_path=config_path,
):
if not run_connection_test(name, servers[name]):
failed = True
else:
print(f"Server '{name}' not found in mcp_config.json")
-1
View File
@@ -66,7 +66,6 @@ def _reset_mutation_authority(monkeypatch):
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)
-824
View File
@@ -1,824 +0,0 @@
"""Tests for control-plane DB substrate (#613)."""
from __future__ import annotations
import os
import tempfile
import threading
import unittest
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import timedelta
from control_plane_db import (
ControlPlaneDB,
InvalidWorkKindError,
LeaseRequiredError,
WORK_KINDS,
_ts,
_utc_now,
)
class ControlPlaneDBTest(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
self.db = ControlPlaneDB(self.db_path)
def tearDown(self) -> None:
self._tmp.cleanup()
def test_schema_and_architecture_meta(self) -> None:
import sqlite3
conn = sqlite3.connect(self.db_path)
try:
rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
finally:
conn.close()
self.assertEqual(rows["schema_version"], "2")
self.assertIn("DB coordinates", rows["architecture"])
self.assertIn("bridge", rows["architecture"].lower())
def test_rejects_raw_incident_as_work_kind(self) -> None:
with self.assertRaises(InvalidWorkKindError):
self.db.upsert_work_item(
remote="prgs",
org="org",
repo="repo",
kind="sentry_incident",
number=1,
)
with self.assertRaises(InvalidWorkKindError):
self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="org",
repo="repo",
kind="glitchtip_incident",
number=9,
)
self.assertEqual(WORK_KINDS, frozenset({"issue", "pr"}))
def test_atomic_assign_and_lease_fields(self) -> None:
self.db.upsert_session(session_id="s-a", role="author", profile="prgs-author")
result = self.db.assign_and_lease(
session_id="s-a",
role="author",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
kind="issue",
number=613,
expected_head_sha="abc123",
allowed_actions=("implement", "comment"),
forbidden_actions=("approve", "merge"),
)
self.assertEqual(result.outcome, "assigned")
self.assertIsNotNone(result.assignment_id)
self.assertIsNotNone(result.lease_id)
self.assertEqual(result.role, "author")
self.assertEqual(result.work_kind, "issue")
self.assertEqual(result.work_number, 613)
self.assertEqual(result.expected_head_sha, "abc123")
self.assertIn("implement", result.allowed_actions)
self.assertIn("merge", result.forbidden_actions)
self.assertIsNotNone(result.expires_at)
def test_second_session_waits_on_foreign_lease(self) -> None:
self.db.upsert_session(session_id="s1", role="author")
self.db.upsert_session(session_id="s2", role="author")
first = self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=100,
expected_head_sha="deadbeef",
)
self.assertEqual(first.outcome, "assigned")
second = self.db.assign_and_lease(
session_id="s2",
role="author",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=100,
expected_head_sha="deadbeef",
)
self.assertEqual(second.outcome, "wait")
self.assertEqual(second.owner_session_id, "s1")
def test_owner_resume_refreshes_lease(self) -> None:
self.db.upsert_session(session_id="s1", role="reviewer")
a = self.db.assign_and_lease(
session_id="s1",
role="reviewer",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=50,
expected_head_sha="head-50",
)
b = self.db.assign_and_lease(
session_id="s1",
role="reviewer",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=50,
expected_head_sha="head-50",
)
self.assertEqual(b.outcome, "assigned")
self.assertEqual(b.lease_id, a.lease_id)
self.assertIn("owner-resume", b.reason)
def test_require_valid_assignment_gates_mutations(self) -> None:
self.db.upsert_session(session_id="s1", role="author")
self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=7,
allowed_actions=("implement",),
forbidden_actions=("merge",),
)
proof = self.db.require_valid_assignment(
session_id="s1",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=7,
action="implement",
)
self.assertEqual(proof["session_id"], "s1")
with self.assertRaises(LeaseRequiredError):
self.db.require_valid_assignment(
session_id="s1",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=7,
action="merge",
)
with self.assertRaises(LeaseRequiredError):
self.db.require_valid_assignment(
session_id="s-other",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=7,
action="implement",
)
def test_expired_lease_allows_reassign(self) -> None:
self.db.upsert_session(session_id="s1", role="author")
self.db.upsert_session(session_id="s2", role="author")
past = _utc_now() - timedelta(hours=1)
# Create lease already expired by using negative TTL edge via direct assign then expire
assigned = self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=3,
lease_ttl_seconds=1,
)
self.assertEqual(assigned.outcome, "assigned")
# Force expiry in DB
import sqlite3
conn = sqlite3.connect(self.db_path)
try:
conn.execute(
"UPDATE leases SET expires_at = ? WHERE lease_id = ?",
(_ts(past), assigned.lease_id),
)
conn.commit()
finally:
conn.close()
n = self.db.expire_stale_leases()
self.assertGreaterEqual(n, 1)
second = self.db.assign_and_lease(
session_id="s2",
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=3,
)
self.assertEqual(second.outcome, "assigned")
self.assertEqual(second.session_id, "s2")
def test_merged_work_never_assigned(self) -> None:
self.db.upsert_session(session_id="s1", role="merger")
self.db.upsert_work_item(
remote="prgs",
org="o",
repo="r",
kind="pr",
number=99,
state="merged",
)
result = self.db.assign_and_lease(
session_id="s1",
role="merger",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=99,
expected_head_sha="merged-head",
)
self.assertEqual(result.outcome, "no_safe_work")
def test_four_concurrent_sessions_unique_assignments(self) -> None:
"""Four concurrent assigners on four different issues — all succeed uniquely.
Also two concurrent assigners on the *same* issue: at most one assigned.
"""
for i in range(4):
self.db.upsert_session(session_id=f"sess-{i}", role="author")
def claim_unique(i: int):
return self.db.assign_and_lease(
session_id=f"sess-{i}",
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=1000 + i,
)
with ThreadPoolExecutor(max_workers=4) as pool:
results = [f.result() for f in as_completed([pool.submit(claim_unique, i) for i in range(4)])]
self.assertEqual({r.outcome for r in results}, {"assigned"})
numbers = sorted(r.work_number for r in results)
self.assertEqual(numbers, [1000, 1001, 1002, 1003])
# Contention on one item
self.db.upsert_session(session_id="c1", role="author")
self.db.upsert_session(session_id="c2", role="author")
self.db.upsert_session(session_id="c3", role="author")
self.db.upsert_session(session_id="c4", role="author")
barrier = threading.Barrier(4)
outcomes: list[str] = []
lock = threading.Lock()
def contend(sid: str) -> None:
barrier.wait()
res = self.db.assign_and_lease(
session_id=sid,
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=7777,
)
with lock:
outcomes.append(res.outcome)
threads = [threading.Thread(target=contend, args=(f"c{i}",)) for i in range(1, 5)]
for t in threads:
t.start()
for t in threads:
t.join()
self.assertEqual(outcomes.count("assigned"), 1)
self.assertEqual(outcomes.count("wait"), 3)
def test_terminal_lock_index(self) -> None:
self.db.set_terminal_lock(
remote="prgs",
org="o",
repo="r",
terminal_pr=332,
review_id="rev-1",
decision="approve",
)
row = self.db.get_active_terminal_lock(remote="prgs", org="o", repo="r")
self.assertIsNotNone(row)
assert row is not None
self.assertEqual(row["terminal_pr"], 332)
self.assertEqual(row["status"], "active")
def test_incident_links_not_work_items(self) -> None:
link = self.db.upsert_incident_link(
provider="sentry",
provider_base_url="https://sentry.prgs.cc",
provider_org="prgs",
provider_project="gitea-tools-mcp",
provider_issue_id="12345",
gitea_org="Scaled-Tech-Consulting",
gitea_repo="Gitea-Tools",
gitea_issue_number=9001,
fingerprint="fp-1",
)
self.assertEqual(link["gitea_issue_number"], 9001)
found = self.db.get_incident_link_for_gitea_issue(
gitea_org="Scaled-Tech-Consulting",
gitea_repo="Gitea-Tools",
gitea_issue_number=9001,
)
self.assertIsNotNone(found)
# Linking does not create a work_item of incident kind
with self.assertRaises(InvalidWorkKindError):
self.db.upsert_work_item(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
kind="sentry",
number=12345,
)
def test_heartbeat_and_release(self) -> None:
self.db.upsert_session(session_id="s1", role="author")
a = self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=1,
)
hb = self.db.heartbeat_lease(a.lease_id, session_id="s1")
self.assertEqual(hb["lease_id"], a.lease_id)
self.db.release_lease(a.lease_id, session_id="s1")
# After release another session can claim
self.db.upsert_session(session_id="s2", role="author")
b = self.db.assign_and_lease(
session_id="s2",
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=1,
)
self.assertEqual(b.outcome, "assigned")
self.assertEqual(b.session_id, "s2")
def test_require_valid_assignment_rejects_stale_head(self) -> None:
"""Assignment must not authorize mutations after work-item head drifts."""
self.db.upsert_session(session_id="s1", role="author")
self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=42,
expected_head_sha="head-v1",
allowed_actions=("implement",),
)
# Head drifts after assignment
self.db.upsert_work_item(
remote="prgs",
org="o",
repo="r",
kind="pr",
number=42,
current_head_sha="head-v2",
)
with self.assertRaises(LeaseRequiredError) as ctx:
self.db.require_valid_assignment(
session_id="s1",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=42,
action="implement",
)
self.assertIn("stale head", str(ctx.exception).lower())
def test_require_valid_assignment_rejects_terminal_state(self) -> None:
"""Assignment must not authorize mutations after work item is merged/closed."""
self.db.upsert_session(session_id="s1", role="author")
self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=55,
expected_head_sha="abc",
allowed_actions=("implement",),
)
self.db.upsert_work_item(
remote="prgs",
org="o",
repo="r",
kind="pr",
number=55,
state="merged",
current_head_sha="abc",
)
with self.assertRaises(LeaseRequiredError) as ctx:
self.db.require_valid_assignment(
session_id="s1",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=55,
action="implement",
)
self.assertIn("terminal", str(ctx.exception).lower())
self.db.upsert_work_item(
remote="prgs",
org="o",
repo="r",
kind="issue",
number=56,
state="open",
)
self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=56,
allowed_actions=("implement",),
)
self.db.upsert_work_item(
remote="prgs",
org="o",
repo="r",
kind="issue",
number=56,
state="closed",
)
with self.assertRaises(LeaseRequiredError):
self.db.require_valid_assignment(
session_id="s1",
remote="prgs",
org="o",
repo="r",
kind="issue",
number=56,
action="implement",
)
def test_pr_assignment_requires_expected_head_sha(self) -> None:
"""PR assign and mutation must fail closed without a head pin."""
self.db.upsert_session(session_id="s1", role="author")
with self.assertRaises(LeaseRequiredError) as ctx:
self.db.assign_and_lease(
session_id="s1",
role="author",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=88,
expected_head_sha=None,
allowed_actions=("implement",),
)
self.assertIn("expected_head_sha", str(ctx.exception))
# Legacy path: force an unpinned PR assignment into the DB, then
# populate head and prove mutation is still rejected.
import sqlite3
self.db.upsert_work_item(
remote="prgs",
org="o",
repo="r",
kind="pr",
number=89,
current_head_sha=None,
)
conn = sqlite3.connect(self.db_path)
try:
wid = conn.execute(
"SELECT work_item_id FROM work_items WHERE kind='pr' AND number=89"
).fetchone()[0]
conn.execute(
"""
INSERT INTO sessions(session_id, role, started_at, last_heartbeat_at, status)
VALUES ('legacy', 'author', '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active')
"""
)
conn.execute(
"""
INSERT INTO leases(
lease_id, work_item_id, session_id, role, phase,
expires_at, heartbeat_at, status
) VALUES (
'lease-legacy', ?, 'legacy', 'author', 'claimed',
'2099-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active'
)
""",
(wid,),
)
conn.execute(
"""
INSERT INTO assignments(
assignment_id, work_item_id, session_id, lease_id,
allowed_actions, forbidden_actions, expected_head_sha,
role, status, created_at
) VALUES (
'asn-legacy', ?, 'legacy', 'lease-legacy',
'["implement"]', '["merge"]', NULL,
'author', 'active', '2020-01-01T00:00:00Z'
)
""",
(wid,),
)
conn.commit()
finally:
conn.close()
self.db.upsert_work_item(
remote="prgs",
org="o",
repo="r",
kind="pr",
number=89,
current_head_sha="populated-head",
)
with self.assertRaises(LeaseRequiredError) as ctx2:
self.db.require_valid_assignment(
session_id="legacy",
remote="prgs",
org="o",
repo="r",
kind="pr",
number=89,
action="implement",
)
self.assertIn("expected_head_sha pin", str(ctx2.exception))
def test_migrate_duplicate_null_scope_incident_links(self) -> None:
"""Legacy NULL-scope duplicates must migrate without UNIQUE crash."""
import sqlite3
from control_plane_db import ControlPlaneDB, ControlPlaneError
# Build a v1-like table with nullable scope columns and insert dups
# that collapse under normalization, then open ControlPlaneDB on it.
path = os.path.join(self._tmp.name, "legacy_dups.sqlite3")
conn = sqlite3.connect(path)
try:
conn.executescript(
"""
CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE incident_links (
link_id INTEGER PRIMARY KEY AUTOINCREMENT,
provider TEXT NOT NULL,
provider_base_url TEXT,
provider_org TEXT,
provider_project TEXT,
provider_issue_id TEXT NOT NULL,
provider_short_id TEXT,
provider_permalink TEXT,
fingerprint TEXT,
gitea_org TEXT NOT NULL,
gitea_repo TEXT NOT NULL,
gitea_issue_number INTEGER NOT NULL,
linked_pr_numbers TEXT,
first_seen TEXT,
last_seen TEXT,
event_count INTEGER,
status TEXT NOT NULL DEFAULT 'open',
release_resolved_at TEXT,
last_sync_at TEXT,
UNIQUE (
provider, provider_base_url, provider_org,
provider_project, provider_issue_id
)
);
INSERT INTO incident_links(
provider, provider_base_url, provider_org, provider_project,
provider_issue_id, gitea_org, gitea_repo, gitea_issue_number, status
) VALUES
('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'),
('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open');
"""
)
# SQLite allows two NULL-scope rows with same provider/issue under UNIQUE.
n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
self.assertEqual(n, 2)
conn.commit()
finally:
conn.close()
db = ControlPlaneDB(path)
conn2 = sqlite3.connect(path)
try:
n2 = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
rows = conn2.execute(
"SELECT provider_base_url, provider_org, provider_project, gitea_issue_number "
"FROM incident_links"
).fetchall()
finally:
conn2.close()
self.assertEqual(n2, 1)
self.assertEqual(rows[0][0], "")
self.assertEqual(rows[0][1], "")
self.assertEqual(rows[0][2], "")
self.assertEqual(rows[0][3], 10)
# Touch to silence unused import in type checkers if needed
self.assertTrue(issubclass(ControlPlaneError, Exception))
del db
def test_migrate_conflicting_duplicate_incident_links_fails_closed(self) -> None:
"""Conflicting Gitea targets for the same provider key must fail closed."""
import sqlite3
from control_plane_db import ControlPlaneDB, ControlPlaneError
path = os.path.join(self._tmp.name, "legacy_conflict.sqlite3")
conn = sqlite3.connect(path)
try:
conn.executescript(
"""
CREATE TABLE incident_links (
link_id INTEGER PRIMARY KEY AUTOINCREMENT,
provider TEXT NOT NULL,
provider_base_url TEXT,
provider_org TEXT,
provider_project TEXT,
provider_issue_id TEXT NOT NULL,
gitea_org TEXT NOT NULL,
gitea_repo TEXT NOT NULL,
gitea_issue_number INTEGER NOT NULL,
status TEXT NOT NULL DEFAULT 'open',
UNIQUE (
provider, provider_base_url, provider_org,
provider_project, provider_issue_id
)
);
INSERT INTO incident_links(
provider, provider_base_url, provider_org, provider_project,
provider_issue_id, gitea_org, gitea_repo, gitea_issue_number
) VALUES
('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 1),
('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 2);
"""
)
conn.commit()
finally:
conn.close()
with self.assertRaises(ControlPlaneError) as ctx:
ControlPlaneDB(path)
self.assertIn("conflicting", str(ctx.exception).lower())
def test_migrate_conflicting_observation_metadata_fails_closed(self) -> None:
"""Same provider key + same Gitea target but differing obs metadata must fail closed.
Regression for silent data loss: migration used to keep lowest link_id and
delete peers after comparing only Gitea targets (#619 RC3).
"""
import sqlite3
from control_plane_db import ControlPlaneDB, ControlPlaneError
path = os.path.join(self._tmp.name, "legacy_meta_conflict.sqlite3")
conn = sqlite3.connect(path)
try:
conn.executescript(
"""
CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE incident_links (
link_id INTEGER PRIMARY KEY AUTOINCREMENT,
provider TEXT NOT NULL,
provider_base_url TEXT,
provider_org TEXT,
provider_project TEXT,
provider_issue_id TEXT NOT NULL,
provider_short_id TEXT,
provider_permalink TEXT,
fingerprint TEXT,
gitea_org TEXT NOT NULL,
gitea_repo TEXT NOT NULL,
gitea_issue_number INTEGER NOT NULL,
linked_pr_numbers TEXT,
first_seen TEXT,
last_seen TEXT,
event_count INTEGER,
status TEXT NOT NULL DEFAULT 'open',
release_resolved_at TEXT,
last_sync_at TEXT,
UNIQUE (
provider, provider_base_url, provider_org,
provider_project, provider_issue_id
)
);
INSERT INTO incident_links(
provider, provider_base_url, provider_org, provider_project,
provider_issue_id, provider_permalink, fingerprint,
gitea_org, gitea_repo, gitea_issue_number,
event_count, status, first_seen, last_seen
) VALUES
(
'sentry', NULL, NULL, NULL, 'dup-meta',
'https://sentry.example/issues/1', 'fingerprint-A',
'org', 'repo', 42,
1, 'open', '2026-01-01T00:00:00Z', '2026-01-01T01:00:00Z'
),
(
'sentry', NULL, NULL, NULL, 'dup-meta',
'https://sentry.example/issues/1', 'fingerprint-B',
'org', 'repo', 42,
99, 'resolved', '2026-01-01T00:00:00Z', '2026-01-02T00:00:00Z'
);
"""
)
n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
self.assertEqual(n, 2)
conn.commit()
finally:
conn.close()
with self.assertRaises(ControlPlaneError) as ctx:
ControlPlaneDB(path)
msg = str(ctx.exception).lower()
self.assertIn("conflicting", msg)
self.assertIn("observation metadata", msg)
# Rows must still be present — migration must not delete before failing.
conn2 = sqlite3.connect(path)
try:
remaining = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0]
fps = {
r[0]
for r in conn2.execute(
"SELECT fingerprint FROM incident_links ORDER BY link_id"
).fetchall()
}
finally:
conn2.close()
self.assertEqual(remaining, 2)
self.assertEqual(fps, {"fingerprint-A", "fingerprint-B"})
def test_incident_links_minimal_upsert_is_canonical(self) -> None:
"""Repeated minimal upserts must update one row (NULL-safe uniqueness)."""
a = self.db.upsert_incident_link(
provider="sentry",
provider_issue_id="inc-1",
gitea_org="org",
gitea_repo="repo",
gitea_issue_number=1,
)
b = self.db.upsert_incident_link(
provider="sentry",
provider_issue_id="inc-1",
gitea_org="org",
gitea_repo="repo",
gitea_issue_number=2,
# omit optional scope fields again
)
c = self.db.upsert_incident_link(
provider="sentry",
provider_issue_id="inc-1",
gitea_org="org",
gitea_repo="repo",
gitea_issue_number=3,
provider_base_url=None,
provider_org="",
provider_project=" ",
)
self.assertEqual(a["link_id"], b["link_id"])
self.assertEqual(b["link_id"], c["link_id"])
self.assertEqual(c["gitea_issue_number"], 3)
self.assertEqual(c["provider_base_url"], "")
self.assertEqual(c["provider_org"], "")
self.assertEqual(c["provider_project"], "")
import sqlite3
conn = sqlite3.connect(self.db_path)
try:
n = conn.execute(
"SELECT COUNT(*) FROM incident_links WHERE provider_issue_id = ?",
("inc-1",),
).fetchone()[0]
finally:
conn.close()
self.assertEqual(n, 1)
if __name__ == "__main__":
unittest.main()
+198
View File
@@ -78,6 +78,95 @@ class TestBlockReasonsAndReport(unittest.TestCase):
self.assertTrue(report["recovery"])
class TestLiveRemoteParity(unittest.TestCase):
"""#610: parity must account for the live remote master, not just local.
The daemon can be stale relative to the live remote target while the local
checkout HEAD still matches the daemon's startup commit, so local parity
reports green even though a mutation would run against outdated code.
"""
SHA_C = "c" * 40
def test_distinguishes_three_shas(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertEqual(res["daemon_start_head"], SHA_A)
self.assertEqual(res["local_head"], SHA_A)
self.assertEqual(res["live_remote_head"], SHA_B)
def test_mutation_safe_only_when_all_three_match(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
self.assertTrue(res["mutation_safe"])
self.assertTrue(res["live_known"])
self.assertFalse(res["live_stale"])
def test_live_stale_when_remote_advanced_past_daemon(self):
# Local checkout still matches the daemon start (local parity green),
# but the live remote master has advanced -> daemon is live-stale.
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertTrue(res["in_parity"]) # local parity still green
self.assertTrue(res["live_stale"])
self.assertFalse(res["mutation_safe"])
self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
# Non-goal: unfetchable live remote must not be treated as stale for
# read-only, but a mutation-safe claim fails closed.
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=None)
self.assertFalse(res["live_known"])
self.assertFalse(res["mutation_safe"])
self.assertFalse(res["live_stale"])
self.assertTrue(res["in_parity"])
def test_default_live_remote_preserves_legacy_shape(self):
# Callers that do not supply a live head keep the pre-#610 behavior:
# in-parity, not live-stale, no live-derived block.
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
self.assertFalse(res["live_stale"])
self.assertEqual(mp.parity_block_reasons(res), [])
class TestLiveStaleBlockAndReport(unittest.TestCase):
"""#610: live-staleness must block mutations and surface a typed blocker."""
def test_live_stale_produces_block_reasons(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
self.assertTrue(mp.parity_block_reasons(res))
def test_disable_env_suppresses_live_stale_block(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
self.assertEqual(mp.parity_block_reasons(res), [])
def test_resolver_disagreement_returns_typed_blocker(self):
# Parity says local-green, resolver says restart required -> disagreement
# is a typed, fail-closed blocker naming the resolver as authoritative.
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
self.assertIsNotNone(blocker)
self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
self.assertTrue(blocker["restart_required"])
self.assertTrue(blocker["resolver_authoritative"])
def test_no_disagreement_when_resolver_agrees(self):
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
self.assertIsNone(
mp.parity_resolver_disagreement(res, resolver_restart_required=False))
def test_live_stale_report_names_live_remote(self):
res = mp.assess_master_parity(
{"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
report = mp.parity_report(res)
self.assertEqual(report["live_remote_head"], SHA_B)
self.assertTrue(report["restart_required"])
class TestReadGitHead(unittest.TestCase):
def test_test_override_takes_precedence(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -95,6 +184,78 @@ class TestReadGitHead(unittest.TestCase):
self.assertIsNone(mp.read_git_head(""))
class TestReadRemoteMasterHead(unittest.TestCase):
"""#610: live remote master head reader (env-overridable, fails to None)."""
def test_test_override_takes_precedence(self):
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
def test_blank_override_is_none(self):
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: " "}):
self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
def test_unfetchable_remote_is_none(self):
# No override; a bogus root/remote must fail closed to None, never raise.
env = {k: v for k, v in os.environ.items()
if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
with patch.dict(os.environ, env, clear=True):
self.assertIsNone(
mp.read_remote_master_head("/nonexistent", remote="nope"))
class TestRemoteHeadCache(unittest.TestCase):
"""#610: live remote reads are cached with a TTL to stay off the network.
The parity gate runs on every mutation and every runtime-context read, so an
unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
"""
def setUp(self):
mp._clear_remote_head_cache()
env = {k: v for k, v in os.environ.items()
if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
self._env = patch.dict(os.environ, env, clear=True)
self._env.start()
self.addCleanup(self._env.stop)
self.addCleanup(mp._clear_remote_head_cache)
def _fake_run(self, sha):
class _R:
returncode = 0
stdout = f"{sha}\trefs/heads/master\n"
calls = {"n": 0}
def run(*args, **kwargs):
calls["n"] += 1
return _R()
return run, calls
def test_second_call_within_ttl_uses_cache(self):
run, calls = self._fake_run(SHA_B)
with patch.object(mp.subprocess, "run", run):
a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
self.assertEqual(a, SHA_B)
self.assertEqual(b, SHA_B)
self.assertEqual(calls["n"], 1)
def test_zero_ttl_bypasses_cache(self):
run, calls = self._fake_run(SHA_B)
with patch.object(mp.subprocess, "run", run):
mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
self.assertEqual(calls["n"], 2)
def test_env_override_never_touches_subprocess(self):
run, calls = self._fake_run(SHA_B)
with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
with patch.object(mp.subprocess, "run", run):
self.assertEqual(
mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
self.assertEqual(calls["n"], 0)
class TestServerWiring(unittest.TestCase):
"""Integration with the gate choke point in the server namespace."""
@@ -105,6 +266,13 @@ class TestServerWiring(unittest.TestCase):
self._saved = self.srv._STARTUP_PARITY
self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
"startup_head": SHA_A}
# Keep the live-remote read hermetic (no real ls-remote network call):
# default the live master to the daemon start so parity is fully green
# unless a test overrides the live head explicitly (#610).
self._live_patch = patch.dict(
os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
self._live_patch.start()
self.addCleanup(self._live_patch.stop)
def tearDown(self):
self.srv._STARTUP_PARITY = self._saved
@@ -147,6 +315,36 @@ class TestServerWiring(unittest.TestCase):
self.assertTrue(out["in_parity"])
self.assertNotIn("report", out)
# --- #610: live-remote wiring -------------------------------------------
def test_live_stale_blocks_mutation_though_local_green(self):
# Local checkout matches the daemon start (local parity green) but the
# live remote master has advanced -> mutations must fail closed.
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
self.assertTrue(
self.srv._master_parity_block("gitea.pr.create"))
def test_assess_tool_exposes_three_distinct_shas(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
out = self.srv.gitea_assess_master_parity(remote="prgs")
self.assertEqual(out["daemon_start_head"], SHA_A)
self.assertEqual(out["local_head"], SHA_A)
self.assertEqual(out["live_remote_head"], SHA_B)
self.assertTrue(out["live_stale"])
self.assertFalse(out["mutation_safe"])
self.assertIn("report", out)
def test_assess_tool_mutation_safe_when_all_three_match(self):
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
out = self.srv.gitea_assess_master_parity(remote="prgs")
self.assertTrue(out["mutation_safe"])
self.assertFalse(out["live_stale"])
self.assertNotIn("report", out)
if __name__ == "__main__":
unittest.main()
-208
View File
@@ -1,208 +0,0 @@
import unittest
import gitea_mcp_server
import mcp_namespace_health
import review_merge_state_machine as rmsm
class TestMcpNamespaceHealth(unittest.TestCase):
def setUp(self):
gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
def test_registered_tool_but_live_eof_blocks_merge(self):
result = mcp_namespace_health.classify_namespace_probe(
"gitea-reviewer",
required_tool="gitea_whoami",
registered_tools=["gitea_whoami", "gitea_list_profiles"],
probe_result={
"success": False,
"error": "client is closing: EOF",
},
process={
"pid": 4321,
"profile": "prgs-reviewer",
"env": {
"GITEA_MCP_PROFILE": "prgs-reviewer",
"GITEA_TOKEN": "must-not-leak",
},
},
config_path="/Users/jasonwalker/.gemini/config/mcp_config.json",
probe_source="client_namespace",
)
self.assertFalse(result["healthy"])
self.assertEqual(result["error_type"], "namespace_eof")
self.assertTrue(result["required_tool_registered"])
self.assertFalse(result["required_tool_callable"])
self.assertTrue(result["blocks_merge_workflow"])
self.assertFalse(result["ide_namespace_proven"])
self.assertEqual(result["probe_source"], "client_namespace")
self.assertEqual(result["diagnostics"]["process_pid"], 4321)
self.assertEqual(result["diagnostics"]["profile"], "prgs-reviewer")
self.assertEqual(
result["diagnostics"]["env"]["GITEA_MCP_PROFILE"],
"prgs-reviewer",
)
self.assertNotIn("GITEA_TOKEN", result["diagnostics"]["env"])
self.assertIn("gitea-reviewer", result["reasons"][0])
self.assertIn("gitea_whoami", result["reasons"][0])
def test_successful_client_namespace_invocation_is_ide_proven(self):
result = mcp_namespace_health.classify_namespace_probe(
"gitea-author",
registered_tools=["gitea_whoami"],
probe_result={"success": True, "result": {"authenticated": True}},
process={"pid": 1234, "profile": "prgs-author"},
config_path="/tmp/mcp_config.json",
probe_source="client_namespace",
)
self.assertTrue(result["healthy"])
self.assertTrue(result["required_tool_registered"])
self.assertTrue(result["required_tool_callable"])
self.assertTrue(result["ide_namespace_proven"])
self.assertFalse(result["blocks_merge_workflow"])
self.assertIsNone(result["error_type"])
def test_offline_spawn_success_is_not_ide_proof(self):
result = mcp_namespace_health.classify_namespace_probe(
"gitea-merger",
registered_tools=["gitea_whoami"],
probe_result={"success": True, "result": {}},
process={"pid": 99, "profile": "prgs-merger"},
probe_source="offline_spawn",
)
self.assertTrue(result["healthy"])
self.assertFalse(result["ide_namespace_proven"])
self.assertFalse(result["blocks_merge_workflow"])
self.assertTrue(
any("offline_spawn" in r for r in result["reasons"])
)
def test_registered_missing_required_tool_fails_before_probe_success(self):
result = mcp_namespace_health.classify_namespace_probe(
"gitea-tools",
registered_tools=["gitea_whoami"],
probe_result={"success": True, "result": {}},
probe_source="client_namespace",
)
self.assertFalse(result["healthy"])
self.assertEqual(result["required_tool"], "gitea_list_profiles")
self.assertFalse(result["required_tool_registered"])
self.assertEqual(result["error_type"], "tool_missing")
def test_server_tool_exposes_same_assessment_and_records_session(self):
result = gitea_mcp_server.gitea_assess_mcp_namespace_health(
"gitea-merger",
registered_tools=["gitea_whoami"],
probe_result={"success": False, "error": "transport closed"},
process={"pid": 9876, "profile": "prgs-merger"},
probe_source="client_namespace",
)
self.assertFalse(result["success"])
self.assertEqual(result["error_type"], "namespace_eof")
self.assertEqual(result["namespace"], "gitea-merger")
self.assertEqual(result["diagnostics"]["process_pid"], 9876)
self.assertIn("gitea-merger", gitea_mcp_server._LIVE_NAMESPACE_HEALTH)
gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
self.assertTrue(gate)
self.assertTrue(any("gitea-merger" in r for r in gate))
class TestLiveNamespaceBlocksMerge(unittest.TestCase):
"""AC5: a broken live namespace must hard-block the review/merge state machine."""
def setUp(self):
gitea_mcp_server._LIVE_NAMESPACE_HEALTH.clear()
def _merge_ready_completion(self):
# Completion through PRE_MERGE_RECHECK is the state where a clean merge
# is otherwise allowed (see test_merge_allowed_with_pre_merge_gates).
idx = rmsm.REVIEW_MERGE_STATES.index("PRE_MERGE_RECHECK")
return {state: True for state in rmsm.REVIEW_MERGE_STATES[: idx + 1]}
def _all_gates(self):
return {gate: True for gate in rmsm._PRE_MERGE_REQUIRED_GATES}
def test_assess_workflow_blockers_flags_live_namespace_broken(self):
clean = rmsm.assess_workflow_blockers()
self.assertFalse(clean["block"])
broken = rmsm.assess_workflow_blockers(live_namespace_broken=True)
self.assertTrue(broken["block"])
self.assertTrue(any("namespace" in r.lower() for r in broken["reasons"]))
def test_can_merge_blocks_even_when_all_gates_pass(self):
# Without the namespace blocker a fully-complete workflow can merge...
allowed = rmsm.can_merge(
self._merge_ready_completion(), pre_merge_gates=self._all_gates()
)
self.assertTrue(allowed["allowed"])
# ...but a broken live namespace overrides every satisfied gate.
blocked = rmsm.can_merge(
self._merge_ready_completion(),
pre_merge_gates=self._all_gates(),
live_namespace_broken=True,
)
self.assertTrue(blocked["block"])
self.assertFalse(blocked["allowed"])
def test_workflow_status_forwards_namespace_blocker(self):
status = rmsm.workflow_status(
self._merge_ready_completion(), live_namespace_broken=True
)
self.assertFalse(status["merge_allowed"])
self.assertFalse(status["approve_allowed"])
def test_server_tool_blocks_merge_on_live_namespace_broken(self):
result = gitea_mcp_server.gitea_assess_review_merge_state_machine(
state_completion=self._merge_ready_completion(),
pre_merge_gates=self._all_gates(),
live_namespace_broken=True,
)
self.assertTrue(result["blockers"]["block"])
self.assertFalse(result["merge"]["allowed"])
def test_classify_verdict_bridges_into_merge_block(self):
# The classify verdict is the intended feed for live_namespace_broken.
verdict = mcp_namespace_health.classify_namespace_probe(
"gitea-merger",
registered_tools=["gitea_whoami", "gitea_adopt_merger_pr_lease"],
probe_result={"success": False, "error": "client is closing: EOF"},
process={"pid": 555, "profile": "prgs-merger"},
probe_source="client_namespace",
)
self.assertTrue(verdict["blocks_merge_workflow"])
blocked = rmsm.can_merge(
self._merge_ready_completion(),
pre_merge_gates=self._all_gates(),
live_namespace_broken=verdict["blocks_merge_workflow"],
)
self.assertFalse(blocked["allowed"])
def test_offline_spawn_does_not_authorize_mutation_gate(self):
gitea_mcp_server.gitea_assess_mcp_namespace_health(
"gitea-merger",
registered_tools=["gitea_whoami"],
probe_result={"success": True, "result": {}},
probe_source="offline_spawn",
)
gate = gitea_mcp_server._live_namespace_health_gate("merge_pr")
self.assertTrue(gate)
self.assertTrue(any("offline_spawn" in r or "client_namespace" in r for r in gate))
def test_client_namespace_healthy_clears_mutation_gate(self):
gitea_mcp_server.gitea_assess_mcp_namespace_health(
"gitea-merger",
registered_tools=["gitea_whoami"],
probe_result={"success": True, "result": {}},
probe_source="client_namespace",
)
self.assertEqual(gitea_mcp_server._live_namespace_health_gate("merge_pr"), [])
if __name__ == "__main__":
unittest.main()
-287
View File
@@ -1,287 +0,0 @@
"""Tests for preserving and resuming prepared Gitea review drafts (#609)."""
from __future__ import annotations
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch, MagicMock
sys_path_root = str(Path(__file__).resolve().parent.parent)
if sys_path_root not in sys.path:
sys.path.insert(0, sys_path_root)
import mcp_session_state
import gitea_mcp_server
import reviewer_pr_lease
class TestReviewDrafts(unittest.TestCase):
def setUp(self):
self._tmpdir = tempfile.TemporaryDirectory()
self.state_dir = self._tmpdir.name
self._env = patch.dict(
os.environ,
{
mcp_session_state.STATE_DIR_ENV: self.state_dir,
mcp_session_state.SESSION_PROFILE_LOCK_ENV: "prgs-reviewer",
"GITEA_MCP_PROFILE": "prgs-reviewer",
"GITEA_PROFILE_NAME": "prgs-reviewer",
"PYTEST_CURRENT_TEST": "1", # skip runtime stale checks
},
clear=False,
)
self._env.start()
# Mock workspace binding to pass in test context
self._nwb_patch = patch(
"gitea_mcp_server.nwb.assess_namespace_mutation_workspace",
return_value={"block": False, "mutation_workspace": "/tmp/test-worktree"},
)
self._nwb_patch.start()
# Mock authentication headers
self._auth_patch = patch(
"gitea_mcp_server.get_auth_header",
return_value="Bearer mock-token",
)
self._auth_patch.start()
self._username_patch = patch(
"gitea_mcp_server._authenticated_username",
return_value="sysadmin",
)
self._username_patch.start()
# Clear/Mock local session lease in-memory state
reviewer_pr_lease.clear_session_lease()
def tearDown(self):
self._username_patch.stop()
self._auth_patch.stop()
self._nwb_patch.stop()
self._env.stop()
self._tmpdir.cleanup()
reviewer_pr_lease.clear_session_lease()
@patch("gitea_mcp_server.api_request")
def test_save_draft_success(self, mock_api):
print("GITEA_MCP_SERVER FILE:", gitea_mcp_server.__file__)
print("DIR OF GITEA_MCP_SERVER:", dir(gitea_mcp_server))
mock_api.return_value = {
"head": {"sha": "headsha1234567890"},
"base": {"ref": "master", "sha": "basesha0987654321"},
}
res = gitea_mcp_server.gitea_save_review_draft(
pr_number=587,
action="approve",
body="Good changes",
expected_head_sha="headsha1234567890",
validation_commands="pytest tests/",
validation_results="all pass",
blocker_reason="stale daemon",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/test-worktree",
)
self.assertTrue(res.get("success"))
self.assertEqual(res.get("head_sha"), "headsha1234567890")
# Load draft and verify fields
draft = mcp_session_state.load_state(
kind=mcp_session_state.KIND_REVIEW_DRAFT,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
profile_identity="prgs-reviewer",
)
self.assertIsNotNone(draft)
self.assertEqual(draft.get("pr_number"), 587)
self.assertEqual(draft.get("action"), "approve")
self.assertEqual(draft.get("body"), "Good changes")
self.assertEqual(draft.get("validation_commands"), "pytest tests/")
self.assertEqual(draft.get("validation_results"), "all pass")
self.assertEqual(draft.get("blocker_reason"), "stale daemon")
self.assertEqual(os.path.realpath(draft.get("worktree_path")), os.path.realpath("/tmp/test-worktree"))
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server._fetch_pr_comments")
@patch("gitea_mcp_server.gitea_submit_pr_review")
def test_resume_draft_and_submit_success(self, mock_submit, mock_comments, mock_api):
# 1. Save draft
mock_api.return_value = {
"state": "open",
"head": {"sha": "headsha1234567890"},
"base": {"ref": "master", "sha": "basesha0987654321"},
}
gitea_mcp_server.gitea_save_review_draft(
pr_number=587,
action="approve",
body="Good changes",
expected_head_sha="headsha1234567890",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/test-worktree",
)
# Seed local session lease
reviewer_pr_lease.record_session_lease({
"pr_number": 587,
"session_id": "session-1234",
})
# Mock Gitea comments to return active reviewer lease owned by us
mock_comments.return_value = [
{
"id": 1,
"user": {"username": "sysadmin"},
"body": (
"<!-- mcp-review-lease:v1 -->\n"
"repo: Scaled-Tech-Consulting/Gitea-Tools\n"
"pr: #587\n"
"reviewer_identity: sysadmin\n"
"profile: prgs-reviewer\n"
"session_id: session-1234\n"
"phase: claimed\n"
),
}
]
mock_submit.return_value = {"performed": True, "success": True}
# 2. Resume draft
res = gitea_mcp_server.gitea_resume_review_draft(
pr_number=587,
submit=True,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/test-worktree",
)
self.assertTrue(res.get("success"))
self.assertTrue(res.get("marked_ready"))
self.assertTrue(res.get("submitted"))
mock_submit.assert_called_once()
# Verify draft was deleted after successful submit
draft = mcp_session_state.load_state(
kind=mcp_session_state.KIND_REVIEW_DRAFT,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
profile_identity="prgs-reviewer",
)
self.assertIsNone(draft)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server._fetch_pr_comments")
def test_resume_draft_fails_checks(self, mock_comments, mock_api):
# Save a draft
mock_api.return_value = {
"state": "open",
"head": {"sha": "headsha1234567890"},
"base": {"ref": "master", "sha": "basesha0987654321"},
}
gitea_mcp_server.gitea_save_review_draft(
pr_number=587,
action="approve",
body="Good changes",
expected_head_sha="headsha1234567890",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/test-worktree",
)
# Seed local session lease
reviewer_pr_lease.record_session_lease({
"pr_number": 587,
"session_id": "session-1234",
})
# Mock Gitea comments to return active reviewer lease owned by us
mock_comments.return_value = [
{
"id": 1,
"user": {"username": "sysadmin"},
"body": (
"<!-- mcp-review-lease:v1 -->\n"
"repo: Scaled-Tech-Consulting/Gitea-Tools\n"
"pr: #587\n"
"reviewer_identity: sysadmin\n"
"profile: prgs-reviewer\n"
"session_id: session-1234\n"
"phase: claimed\n"
),
}
]
# Case A: PR closed
mock_api.return_value = {
"state": "closed",
"head": {"sha": "headsha1234567890"},
"base": {"ref": "master", "sha": "basesha0987654321"},
}
res = gitea_mcp_server.gitea_resume_review_draft(
pr_number=587,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/test-worktree",
)
self.assertFalse(res.get("success"))
self.assertIn("PR #587 is not open", res.get("reasons")[0])
# Case B: Head changed
mock_api.return_value = {
"state": "open",
"head": {"sha": "headsha_NEW"},
"base": {"ref": "master", "sha": "basesha0987654321"},
}
res = gitea_mcp_server.gitea_resume_review_draft(
pr_number=587,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/test-worktree",
)
self.assertFalse(res.get("success"))
self.assertIn("PR head SHA changed", res.get("reasons")[0])
# Case C: Target branch changed
mock_api.return_value = {
"state": "open",
"head": {"sha": "headsha1234567890"},
"base": {"ref": "master", "sha": "basesha_NEW"},
}
res = gitea_mcp_server.gitea_resume_review_draft(
pr_number=587,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/test-worktree",
)
self.assertFalse(res.get("success"))
self.assertIn("target branch SHA changed", res.get("reasons")[0])
# Case D: Worktree mismatch
mock_api.return_value = {
"state": "open",
"head": {"sha": "headsha1234567890"},
"base": {"ref": "master", "sha": "basesha0987654321"},
}
res = gitea_mcp_server.gitea_resume_review_draft(
pr_number=587,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path="/tmp/different-worktree",
)
self.assertFalse(res.get("success"))
self.assertIn("worktree binding mismatch", res.get("reasons")[0])