Compare commits

..
18 changed files with 477 additions and 2629 deletions
@@ -1,83 +0,0 @@
# Incident #670: bare direct-to-master commit `2fa97c26` (retroactive audit)
Status: verified; disposition recommendation: **accept as-is, no revert** (final
disposition owned by controller per issue #670).
## Summary
Commit `2fa97c26fbda555a1a83930ca5fdcea9d8e47b50`
(`fix(mcp): load dotenv relative to project root`) landed on `prgs/master`
as a single-parent commit with no PR wrapper and no review record, bypassing
the sanctioned issue → branch → PR → review → merge workflow. It was
discovered during the PR #654 post-merge audit. PR #654 itself merged
cleanly via the Gitea API and did **not** introduce this commit.
## Verification evidence (acceptance criteria 13)
- **AC1 — present on `prgs/master`: yes.**
`git merge-base --is-ancestor 2fa97c26fbda555a1a83930ca5fdcea9d8e47b50 prgs/master` → true.
- **AC2 — no PR or review record: confirmed.**
The commit is a single-parent, non-merge commit sitting directly on
first-parent master between the #629 merge (`5ab5fe85`) and the #654
merge (`ec903b0d`). A PR landing on master produces a merge commit (or a
PR-linked head); neither exists here. The controller audit at issue-create
time also found no PR wrapper and no review record for this SHA.
- **AC3 — changed files and diff summary: confirmed.**
`gitea_auth.py | 5 +++--` (+3/2). Single parent
`5ab5fe8583c07134d55dadf09381aecb67df246e`. The change moves
`PROJECT_ROOT` derivation above `load_dotenv()` and loads
`.env` relative to the project root instead of the process CWD.
## AC4 — why no immediate revert
- The dotenv fix is intentional and required for correct runtime behavior:
without it, `load_dotenv()` resolves `.env` against the process working
directory, which breaks MCP server launches whose CWD is not the project
root.
- The change is small (+3/2), self-contained in `gitea_auth.py`, and has
been running on master without incident since 2026-07-10.
- Reverting would re-introduce a real bug to remove a provenance defect —
the wrong trade. Provenance is repaired retroactively by this document,
issue #670, and the hardening landed under #671.
- If the controller later judges the change unsafe, a separate
revert/repair issue is the sanctioned path (issue #670, recommended
disposition option 4).
## AC5 — workflow-hardening linkage
Prevention already landed: **issue #671** (closed)
*“Block direct pushes to stable branches from MCP workflow sessions”*,
implemented by commit `5933d87647656643a67a50331c4c7b06ea751dad`
(`feat(guard): block direct stable-branch pushes from MCP workflow sessions`).
Shipped guardrails include:
- `gitea_record_stable_branch_push_attempt` — classifies proposed commands
for direct stable-branch push intent (`git push <remote> master`,
refspecs, `HEAD:master`, `--force`, dry-run intent, `:master` delete),
plus root/control-checkout local commits not carried by an issue branch,
and writes a durable `stable_branch_contamination` marker.
- `gitea_audit_stable_branch_contamination` — reconciler-only audit/clear
path; a contaminated worker session cannot self-clear.
- Review/merge/close/completion mutations fail closed while a
contamination marker is active.
## AC6 — PR #654 was not the source
- `2fa97c26` is the **first parent** of the #654 merge commit
`ec903b0d619e7a27d24aed272a890f4e5d381411`; it predates the #654 merge.
- First-parent history `5ab5fe8..ec903b0`:
`2fa97c2 fix(mcp): load dotenv relative to project root` followed by
`ec903b0 Merge pull request 'feat: lifecycle role/hazard labels ... (#603)' (#654)`.
- The #654 merger audit confirmed `ec903b0d` was a valid Gitea-API merge,
the `git push prgs master` attempt during that run was a no-op, and the
net change `2fa97c2..ec903b0` contained only the reviewed #603
lifecycle-label files.
- Conclusion: #654 merged reviewed content only; the unauthorized-path
defect is solely the earlier bare commit `2fa97c26`.
## Explicit non-actions (unchanged by this audit)
- No revert of `2fa97c26`.
- No force-push or history rewrite.
- No master mutation from the audit session.
+70
View File
@@ -0,0 +1,70 @@
# MCP Config Drift Diagnostic & Sanctioned Repair Runbook (#672)
This document describes the diagnostic framework for detecting configuration drift between the active IDE MCP configuration (`~/.gemini/antigravity-ide/mcp_config.json`) and the offline/global canonical configuration (`~/.gemini/config/mcp_config.json`), and establishes the **sanctioned repair runbook**.
## Background & Problem Statement
Offline tools like `test_mcp_conn.py` test the global configuration (`~/.gemini/config/mcp_config.json`) via `subprocess.Popen`. However, the active IDE/client namespace uses `~/.gemini/antigravity-ide/mcp_config.json`. When required Gitea role servers (`gitea-author`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`, `gitea-controller`, `gitea-tools`) are missing or carry mismatched profile environments in the active IDE config:
1. Offline tests pass (`test_mcp_conn.py` green).
2. The IDE client returns `EOF` / `transport closed` when attempting role-scoped mutations.
3. Operators misdiagnose missing server definitions as stale runtimes, leading to forbidden `pkill` attempts (#630) or `mtime` hacks (#655).
## Diagnostic Tool: `mcp_config_drift.py`
Run the diagnostic tool directly to compare configurations:
```bash
python3 mcp_config_drift.py --json
```
Or specify custom config locations:
```bash
python3 mcp_config_drift.py \
--active-config ~/.gemini/antigravity-ide/mcp_config.json \
--global-config ~/.gemini/config/mcp_config.json
```
### Key Diagnostic Outputs
- `in_sync`: Boolean indicating if all required Gitea role servers exist in the active IDE config with matching profile declarations.
- `missing_role_servers`: List of role servers present in global config but missing from active IDE config.
- `profile_mismatches`: List of profile environment mismatches per server.
- `reasons`: Explicit, human-readable list of drift causes.
All returned payloads automatically redact secret tokens, DSNs, Authorization headers, and private keys.
---
## Sanctioned Repair Path (Step-by-Step)
When `mcp_config_drift.py` reports drift (`in_sync: false`), execute the following **sanctioned repair steps**:
1. **Backup Active IDE Config:**
```bash
cp ~/.gemini/antigravity-ide/mcp_config.json ~/.gemini/antigravity-ide/mcp_config.json.bak
```
2. **Patch Active IDE Config:**
Copy the missing Gitea role server JSON blocks (`gitea-author`, `gitea-reviewer`, etc.) from `~/.gemini/config/mcp_config.json` into `~/.gemini/antigravity-ide/mcp_config.json`.
3. **Reconnect via IDE/Client:**
Use the IDE / client UI reconnection control (or restart the IDE client app).
4. **Verify Active Namespace Health:**
Invoke `gitea_whoami` (and optional `gitea_resolve_task_capability`) through the active IDE client on each required role namespace.
---
## FORBIDDEN Repair Actions (#630 / #655)
The following actions are **strictly forbidden** for config drift repair:
- ❌ **`pkill` or manual daemon process kill commands:** Process kills cause contamination and break active session leases.
- ❌ **`mtime` touch edits:** Artificial mtime modifications mask stale runtimes without updating configuration.
- ❌ **Source code edits:** Mutating python tool logic to bypass missing server entries.
- ❌ **Session-state edits:** Direct database or lock-file state mutation.
---
## Final Report Guidelines
A workflow final report **must not** rely on offline `test_mcp_conn.py` output alone. Final reports must include active-config evidence from live `gitea_whoami` calls on the active IDE namespaces.
+6 -21
View File
@@ -47,33 +47,18 @@ Do the steps in order. Stop as soon as a live **client-namespace** call succeeds
- Only the Gitea namespace fails → single-namespace transport close. Continue.
- Every server fails → restart the whole MCP client, not just one namespace.
2. **Request the sanctioned reconnect surface (#678), then reconnect through
the client — not the shell.** From a still-reachable Gitea MCP namespace
(or after host auto-reconnect), call:
```text
gitea_request_mcp_reconnect(
namespace="gitea-author", # or gitea-reviewer / gitea-merger / …
reason="transport_eof",
client="codex", # or claude_code / generic
)
```
The tool is **report-only**: it never restarts a process. It returns
namespace, profile, pid/session, startup SHA, current master SHA, boundary
status, and a **typed blocker** with exact operator UI steps for Codex
(Reload Developer Tools / per-server reconnect) or Claude Code (`/mcp`).
Then perform the host reconnect those steps describe so the client spawns a
fresh subprocess and re-opens the pipe. That clears the closed-client state
that a bare `kill`/respawn from a terminal does **not**.
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** (or the typed operator path returned by
`gitea_request_mcp_reconnect`).
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
+2 -3
View File
@@ -45,11 +45,10 @@ and *fails closed*.
| `legacy_auto_restart_helper` | removed | A helper (`_trigger_mcp_auto_restart`) that actively restarted the server from the read-only resolver path. | Removed in #685; kept absent by `assert_auto_restart_helper_absent()`. | #685, #657 |
| `config_touch_reload` | removed | Touching (utime) the MCP client config to make the host reload the server. | Removed from the resolver in #685: stale detection is report-only, never mutating config, spawning threads, or calling `os._exit`. | #685, #657 |
| `master_advance_auto_restart` | guarded_fail_closed | On-disk master advancing past the running code. | `master_parity_gate` captures startup parity and blocks mutations while stale, emitting restart guidance; the process never self-restarts. | #420, #591, #657 |
| `stale_runtime_resolver_reconnect` | guarded_fail_closed | The capability resolver detecting a stale serving process. | Report-only (#685): returns `restart_required`/`stop_required` and an exact reconnect action; no restart, thread, config touch, or `os._exit`. | #685, #657, #678 |
| `codex_client_reconnect_request` | guarded_fail_closed | `gitea_request_mcp_reconnect` report-only tool for Codex/LLM sessions. | Report-only (#678): returns namespace/profile/pid/startup SHA/master SHA/boundary status and a typed operator blocker with exact client UI steps; never restarts or kills. | #678, #630, #685, #657 |
| `stale_runtime_resolver_reconnect` | guarded_fail_closed | The capability resolver detecting a stale serving process. | Report-only (#685): returns `restart_required`/`stop_required` and an exact reconnect action; no restart, thread, config touch, or `os._exit`. | #685, #657 |
| `manual_daemon_kill` | forbidden | Shell kills of the daemon: `pkill -f mcp_server.py`, `killall`, broad `pkill -f python` sweeps, or `kill <pid>` of a daemon pid. | Forbidden (#630): `runtime_recovery_guard` classifies these as contamination and `gitea_record_daemon_process_kill_attempt` writes a durable marker that fails later mutations closed. Operator maintenance authorization is read only from the environment. | #630, #657 |
| `conflict_marker_infra_stop` | guarded_fail_closed | The daemon entrypoint scans for unresolved merge-conflict markers at startup and stops (`sys.exit(1)`). | Fail-closed startup stop, not a restart: the process exits and waits for the operator to resolve conflicts and relaunch; never loops. | #657 |
| `ide_client_reconnect` | host_residual | A manual `/mcp reconnect` (or equivalent host action) that recreates the MCP client connection. Agents obtain exact UI steps via `gitea_request_mcp_reconnect` (#678). | Outside this process's control; the sanctioned recovery the gates point operators toward. No in-process code initiates it. | #584, #656, #657, #678 |
| `ide_client_reconnect` | host_residual | A manual `/mcp reconnect` (or equivalent host action) that recreates the MCP client connection. | Outside this process's control; the sanctioned recovery the gates point operators toward. No in-process code initiates it. | #584, #656, #657 |
| `profile_switch_runtime` | sanctioned_narrow_recovery | Switching the active execution profile at runtime (dynamic-profile mode). | In-process and restart-free: `runtime_switching_supported` is true, so a switch rebinds capability without recreating the process. | #656, #657 |
## Guards enforced in CI
-1
View File
@@ -137,7 +137,6 @@ that gates each call, not which tools exist.
- `gitea_release_merger_pr_lease`
- `gitea_release_reviewer_pr_lease`
- `gitea_release_workflow_lease`
- `gitea_request_mcp_reconnect`
- `gitea_request_mcp_restart`
- `gitea_resolve_task_capability`
- `gitea_resume_review_draft`
-81
View File
@@ -1,81 +0,0 @@
# Web Console: Notifications & Human-Attention Routing (#648)
- **Status:** Phase 3 Live
- **Tracking Issue:** [#648](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/648)
- **Parent Epic:** [#631](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/631)
- **Attention Boundary Reference:** [#628](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/issues/628)
---
## 1. Overview
The **Notifications & Human-Attention Console** (`/notifications`, `/api/v1/notifications`) provides intelligent event classification and human-attention routing for autonomous workflow operations.
To prevent alert fatigue while ensuring critical escalation boundaries are never missed, events are classified into three distinct **Attention Classes**:
1. **`human-required`** (Urgent Escalation Boundary):
- Items requiring immediate human intervention or business decisions.
- Triggers: Auth failures, hard stops, irrecoverable state, decision locks, failed report validations, critical probe errors.
- Display: Highlighted in red (`badge-blocked`) with a `HUMAN REQUIRED` badge.
2. **`operator`** (Operational Inbox):
- Items requiring controller or operator review/triage during routine execution.
- Triggers: Blocked PRs (merge conflicts), stale leases, duplicate PRs on issues, unassigned ready work.
- Display: Displayed in orange/yellow (`badge-claimed`).
3. **`routine`** (Background Workflow Transitions):
- Normal, healthy workflow transitions and state progressions.
- Triggers: Active PRs/issues in standard state, clean branch creation, routine heartbeats.
- Display: Filtered out of default inbox views to eliminate notification spam; viewable on demand via the "Routine" or "All" tab.
---
## 2. API Endpoints
### `GET /api/v1/notifications`
*Compatibility Alias:* `GET /api/notifications`
#### Query Parameters:
- `project_id` (optional): Filter notifications by project ID.
- `attention_class` (optional): `inbox` (default: human-required + operator), `human-required`, `operator`, `routine`, `all`.
#### Example JSON Response:
```json
{
"project_id": "gitea-tools",
"repo_label": "Scaled-Tech-Consulting/Gitea-Tools",
"human_required_count": 0,
"operator_count": 2,
"routine_count": 5,
"total_count": 7,
"fetch_error": null,
"inbox_items": [
{
"id": "notif-pr-block-742",
"attention_class": "operator",
"category": "blocker",
"title": "Blocked PR #742",
"summary": "PR #742 requires merge conflict resolution.",
"work_kind": "pr",
"work_number": 742,
"project_id": "gitea-tools",
"repo_label": "Scaled-Tech-Consulting/Gitea-Tools",
"created_at": "2026-07-25T16:39:47Z",
"deep_link": "/traffic",
"requires_human": false,
"extra": {}
}
],
"all_items": [...]
}
```
---
## 3. UI Navigation
- Access via the **Traffic** navigation menu: **Traffic → Notifications**.
- The main view displays:
- **Metrics Summary Bar**: Highlighting counts for Human Required, Operator Inbox, and Routine items.
- **Attention Filter Tabs**: Toggle between Inbox (Human + Operator), Human Required, Operator, Routine, and All.
- **Structured Event Table**: Displays category, title, summary, work item links, and timestamps.
+8 -142
View File
@@ -2092,7 +2092,6 @@ import root_checkout_guard # noqa: E402
import workflow_scope_guard # noqa: E402 # #683 production scope / force-on guards
import stable_branch_push_guard # noqa: E402
import runtime_recovery_guard # noqa: E402 # #630 manual daemon-kill contamination
import mcp_client_reconnect # noqa: E402 # #678 sanctioned Codex reconnect request
import remote_repo_guard # noqa: E402
import anti_stomp_preflight # noqa: E402
import issue_claim_heartbeat # noqa: E402
@@ -14025,16 +14024,11 @@ def _classify_operation_gate_reasons(reasons: list[str]) -> dict:
def _stale_runtime_reconnect_action() -> str:
"""Sanctioned recovery for a stale daemon — reconnect only (#685/#897/#678)."""
"""Sanctioned recovery for a stale daemon — reconnect only (#685/#897)."""
return (
"blocker_kind=runtime_reconnect_required: call "
"gitea_request_mcp_reconnect(namespace=<active gitea-* namespace>, "
"reason='stale-runtime', client='codex') for a typed operator "
"reconnect blocker with exact UI steps, then reconnect the IDE/client "
"MCP session so the server reloads at the current master head. Do not "
"call gitea_activate_profile, pkill, touch configs, or switch MCP role "
"sessions — profile switching does not clear a stale daemon. After "
"reconnect restart from gitea_whoami → gitea_resolve_task_capability."
"Reconnect the IDE/client MCP session so the server reloads at the "
"current master head. Do not call gitea_activate_profile or switch "
"MCP role sessions — profile switching does not clear a stale daemon."
)
@@ -21004,15 +20998,10 @@ def gitea_resolve_task_capability(
# serving process/profile inventory is stale — even if permission is OK.
if runtime_stale_blocker:
next_safe_action = (
"blocker_kind=runtime_reconnect_required: call "
"gitea_request_mcp_reconnect(namespace=<active gitea-* namespace>, "
"reason='stale-runtime', client='codex') for a typed operator "
"blocker with exact UI steps, then reconnect/reload the IDE-managed "
"Gitea MCP server for this profile so it reloads current master. "
"Do not edit mcp_config.json by hand, pkill, or touch configs; the "
"resolver does not touch config, spawn recovery threads, or "
"terminate the process. After reconnect restart from gitea_whoami → "
"gitea_resolve_task_capability."
"blocker_kind=runtime_reconnect_required: reconnect/restart the "
"IDE-managed Gitea MCP server for this profile so it reloads current "
"master. Do not edit mcp_config.json by hand; the resolver does not "
"touch config, spawn recovery threads, or terminate the process."
)
# Task/role alignment guards (#167): the requested task, not the
@@ -22570,129 +22559,6 @@ def gitea_workflow_dashboard(
return payload
@mcp.tool()
def gitea_request_mcp_reconnect(
namespace: str | None = None,
reason: str | None = None,
client: str = "codex",
remote: str = "dadeschools",
host: str | None = None,
session_id: str | None = None,
) -> dict:
"""Request a sanctioned host/IDE MCP reconnect for a named namespace (#678).
Codex and other agent hosts can detect stale or closed Gitea MCP runtimes
(``stop_required`` / ``restart_required`` from capability resolution, transport
EOF, missing namespace attachment). The **host owns the transport** this
process cannot reopen the client's stdio pipe. This tool is the callable
surface agents use to:
1. Report reconnect status fields (namespace, profile, pid/session,
startup SHA, current master SHA, boundary status).
2. Return a **typed blocker** with exact operator UI steps for Codex (or
another client) when reconnect is required.
This tool **never** restarts, kills, reloads, or reconfigures an MCP
process. Forbidden recovery paths (pkill, touch/mtime hacks, config/.env
edits, session-state edits, raw API) are never recommended.
After the operator reconnects, workflows must restart from preflight:
``gitea_whoami`` ``gitea_resolve_task_capability`` task.
Args:
namespace: MCP namespace to reconnect (e.g. ``gitea-author``). Defaults
to the active profile's inferred namespace.
reason: Why reconnect is requested: ``stale-runtime``, ``transport_eof``,
``missing_namespace``, ``not_required``, or free-form (normalized).
client: Operator UI surface ``codex`` (default), ``claude_code``, or
``generic``.
remote: Known instance ``dadeschools`` or ``prgs`` (parity context).
host: Optional host override for parity context.
session_id: Optional session id to echo in the report.
Returns:
dict with reconnect report fields, ``reconnect_performed=False``,
``typed_blocker`` when reconnect is required, and
``forbidden_recovery_paths``.
"""
# Read-only: gitea.read is sufficient. Never a mutation.
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"read_only": True,
"reconnect_performed": False,
"mutation_performed": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
"forbidden_recovery_paths": list(
mcp_client_reconnect.FORBIDDEN_RECOVERY_PATHS
),
}
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip() or None
inferred_ns = role_namespace_gate.infer_mcp_namespace(profile_name)
ns = (namespace or "").strip() or inferred_ns or "gitea-tools"
parity = _current_master_parity()
startup_sha = (
parity.get("daemon_start_head")
or parity.get("startup_head")
or _process_boot_head_sha
)
current_sha = parity.get("local_head") or parity.get("current_head")
if not current_sha:
try:
current_sha = master_parity_gate.read_git_head(PROJECT_ROOT)
except Exception: # noqa: BLE001
current_sha = None
boundary = mcp_client_reconnect.classify_boundary_status(
startup_sha=startup_sha if isinstance(startup_sha, str) else None,
current_master_sha=current_sha if isinstance(current_sha, str) else None,
live_stale=bool(parity.get("live_stale")) if parity.get("live_known") else None,
in_parity=parity.get("in_parity") if parity.get("determinable") else None,
)
# Infer reason from parity when caller left it unspecified.
effective_reason = reason
if not (effective_reason or "").strip():
if parity.get("restart_required") or parity.get("live_stale"):
effective_reason = mcp_client_reconnect.REASON_STALE_RUNTIME
elif boundary == mcp_client_reconnect.BOUNDARY_CLEAN:
effective_reason = mcp_client_reconnect.REASON_NOT_REQUIRED
else:
effective_reason = mcp_client_reconnect.REASON_UNSPECIFIED
payload = mcp_client_reconnect.build_reconnect_request(
namespace=ns,
profile=profile_name,
pid=os.getpid(),
session_id=session_id
or f"{(profile_name or 'session')}-{os.getpid()}",
startup_sha=startup_sha if isinstance(startup_sha, str) else None,
current_master_sha=current_sha if isinstance(current_sha, str) else None,
boundary_status=boundary,
reason=effective_reason,
client=client,
live_stale=bool(parity.get("live_stale")) if parity.get("live_known") else None,
in_parity=parity.get("in_parity") if parity.get("determinable") else None,
restart_required=bool(parity.get("restart_required")),
stop_required=bool(parity.get("restart_required")),
extra={
"remote": remote if remote in REMOTES else remote,
"host": host,
"session_context_audit": session_ctx.mutation_context_audit_fields(),
"parity_summary": master_parity_gate.format_parity(parity),
"live_stale": parity.get("live_stale"),
"live_known": parity.get("live_known"),
"in_parity": parity.get("in_parity"),
},
)
return payload
@mcp.tool()
def gitea_request_mcp_restart(
remote: str = "dadeschools",
-328
View File
@@ -1,328 +0,0 @@
"""Sanctioned MCP client reconnect request surface for Codex/LLM sessions (#678).
Codex and other agent hosts can detect stale or closed Gitea MCP runtimes, but
the host owns the transport. This module never restarts, kills, or reloads a
daemon. It builds:
1. A **callable reconnect request** result agents can invoke via
``gitea_request_mcp_reconnect`` (report-only, side-effect free).
2. A **typed blocker** with exact operator UI steps when recovery must be
performed by the host/operator.
Forbidden recovery paths (must never be recommended):
* ``pkill`` / ``kill`` / ``killall`` of MCP daemons
* ``touch`` / mtime config reload hacks
* ``.env`` or MCP config edits as recovery
* session-state file edits
* raw Gitea API / direct server-import fallbacks
After the operator reconnects, workflows restart from identity / runtime /
capability preflight (``gitea_whoami`` → ``gitea_resolve_task_capability`` →
task).
"""
from __future__ import annotations
from typing import Any, Mapping
# --- Reason vocabulary -------------------------------------------------------
REASON_STALE_RUNTIME = "stale-runtime"
REASON_TRANSPORT_EOF = "transport_eof"
REASON_MISSING_NAMESPACE = "missing_namespace"
REASON_NOT_REQUIRED = "not_required"
REASON_UNSPECIFIED = "unspecified"
VALID_REASONS = frozenset(
{
REASON_STALE_RUNTIME,
REASON_TRANSPORT_EOF,
REASON_MISSING_NAMESPACE,
REASON_NOT_REQUIRED,
REASON_UNSPECIFIED,
}
)
# Boundary statuses reported to callers (match review_workflow_boundary style).
BOUNDARY_CLEAN = "clean"
BOUNDARY_MISMATCH = "mismatch"
BOUNDARY_STALE = "stale"
BOUNDARY_UNKNOWN = "unknown"
# Typed blocker kinds
BLOCKER_OPERATOR_RECONNECT = "operator_mcp_reconnect_required"
BLOCKER_NONE = "none"
FORBIDDEN_RECOVERY_PATHS: tuple[str, ...] = (
"pkill / kill / killall of mcp_server.py, gitea_mcp_server, or broad python sweeps",
"touch / mtime-based MCP config reload hacks",
".env edits as recovery",
"MCP config file edits as recovery",
"session-state file edits as recovery",
"raw Gitea API or direct MCP server-import fallbacks",
)
# Client-specific operator UI steps. Keep Codex first (issue title surface).
OPERATOR_UI_STEPS: dict[str, tuple[str, ...]] = {
"codex": (
"In Codex, open the MCP / Developer tools panel for this workspace.",
"Locate the named Gitea MCP server entry (namespace) that needs reconnect "
"(e.g. gitea-author, gitea-reviewer, gitea-merger, gitea-tools, "
"gitea-controller, gitea-reconciler).",
"Click 'Reload Developer Tools' or the server reconnect/reload control "
"for that entry so the client spawns a fresh MCP subprocess.",
"If per-server reconnect is unavailable, fully restart the Codex client "
"(quit and relaunch) so all MCP namespaces reattach.",
"After reconnect, rerun the blocked workflow from preflight: "
"gitea_whoami → gitea_resolve_task_capability → the original task. "
"Do not resume mid-mutation.",
),
"claude_code": (
"Run `/mcp` (or open the MCP servers UI) in Claude Code.",
"Reconnect the affected gitea-* server entry so the client reopens stdio.",
"If reconnect fails, relaunch the Claude Code session entirely.",
"After reconnect, restart the workflow from gitea_whoami → "
"gitea_resolve_task_capability → task.",
),
"generic": (
"Use the host/IDE MCP reconnect or reload control for the named namespace.",
"If no per-namespace control exists, restart the MCP client/editor.",
"After reconnect, restart the workflow from identity/capability preflight.",
),
}
DEFAULT_CLIENT = "codex"
def normalize_reason(reason: str | None) -> str:
"""Map free-form reason strings onto the closed vocabulary."""
raw = (reason or "").strip().lower()
if not raw:
return REASON_UNSPECIFIED
if raw in VALID_REASONS:
return raw
text = raw.replace(" ", "_").replace("-", "_")
aliases = {
"stale_runtime": REASON_STALE_RUNTIME,
"staleruntime": REASON_STALE_RUNTIME,
"runtime_stale": REASON_STALE_RUNTIME,
"stale": REASON_STALE_RUNTIME,
"transport_eof": REASON_TRANSPORT_EOF,
"transport_closed": REASON_TRANSPORT_EOF,
"eof": REASON_TRANSPORT_EOF,
"client_is_closing": REASON_TRANSPORT_EOF,
"missing_namespace": REASON_MISSING_NAMESPACE,
"namespace_missing": REASON_MISSING_NAMESPACE,
"not_required": REASON_NOT_REQUIRED,
"healthy": REASON_NOT_REQUIRED,
"ok": REASON_NOT_REQUIRED,
"unspecified": REASON_UNSPECIFIED,
}
if text in aliases:
return aliases[text]
hyphenated = text.replace("_", "-")
if hyphenated in VALID_REASONS:
return hyphenated
return REASON_UNSPECIFIED
def normalize_client(client: str | None) -> str:
"""Return a known client key for operator UI steps."""
text = (client or "").strip().lower().replace(" ", "_").replace("-", "_")
if text in ("codex", "openai_codex", "openai"):
return "codex"
if text in ("claude", "claude_code", "claude_desktop", "anthropic"):
return "claude_code"
if text in OPERATOR_UI_STEPS:
return text
return DEFAULT_CLIENT
def classify_boundary_status(
*,
startup_sha: str | None,
current_master_sha: str | None,
live_stale: bool | None = None,
in_parity: bool | None = None,
) -> str:
"""Derive boundary_status from parity evidence."""
if live_stale is True or in_parity is False:
return BOUNDARY_STALE
start = (startup_sha or "").strip().lower()
current = (current_master_sha or "").strip().lower()
if start and current and start != current:
return BOUNDARY_MISMATCH
if start and current and start == current:
return BOUNDARY_CLEAN
if in_parity is True:
return BOUNDARY_CLEAN
return BOUNDARY_UNKNOWN
def operator_ui_steps(client: str | None, *, namespace: str | None = None) -> list[str]:
"""Exact operator UI steps for the named client."""
key = normalize_client(client)
steps = list(OPERATOR_UI_STEPS.get(key) or OPERATOR_UI_STEPS[DEFAULT_CLIENT])
ns = (namespace or "").strip()
if ns:
steps = [
s.replace("named Gitea MCP server entry (namespace)", f"namespace '{ns}'")
.replace("affected gitea-* server entry", f"server entry '{ns}'")
.replace("named namespace", f"namespace '{ns}'")
for s in steps
]
return steps
def build_reconnect_request(
*,
namespace: str,
profile: str | None = None,
pid: int | str | None = None,
session_id: str | None = None,
startup_sha: str | None = None,
current_master_sha: str | None = None,
boundary_status: str | None = None,
reason: str | None = None,
client: str | None = DEFAULT_CLIENT,
live_stale: bool | None = None,
in_parity: bool | None = None,
restart_required: bool | None = None,
stop_required: bool | None = None,
extra: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
"""Build the structured reconnect-request / typed-blocker payload (#678).
Never mutates process, config, or session state. Always side-effect free.
"""
ns = (namespace or "").strip() or "unknown"
normalized_reason = normalize_reason(reason)
boundary = (boundary_status or "").strip() or classify_boundary_status(
startup_sha=startup_sha,
current_master_sha=current_master_sha,
live_stale=live_stale,
in_parity=in_parity,
)
reconnect_needed = True
if normalized_reason == REASON_NOT_REQUIRED and boundary == BOUNDARY_CLEAN:
reconnect_needed = False
if restart_required is False and stop_required is False and boundary == BOUNDARY_CLEAN:
# Explicit healthy probe
if normalized_reason in (REASON_NOT_REQUIRED, REASON_UNSPECIFIED):
reconnect_needed = False
normalized_reason = REASON_NOT_REQUIRED
if restart_required is True or stop_required is True:
reconnect_needed = True
if normalized_reason in (REASON_NOT_REQUIRED, REASON_UNSPECIFIED):
normalized_reason = REASON_STALE_RUNTIME
client_key = normalize_client(client)
steps = operator_ui_steps(client_key, namespace=ns)
result: dict[str, Any] = {
"success": True,
"read_only": True,
"reconnect_performed": False,
"mutation_performed": False,
"reconnect_needed": reconnect_needed,
"namespace": ns,
"profile": (profile or "").strip() or None,
"pid": pid,
"session_id": (session_id or "").strip() or None,
"startup_sha": (startup_sha or "").strip() or None,
"current_master_sha": (current_master_sha or "").strip() or None,
"boundary_status": boundary,
"reason": normalized_reason,
"client": client_key,
"forbidden_recovery_paths": list(FORBIDDEN_RECOVERY_PATHS),
"post_reconnect_preflight": [
"gitea_whoami",
"gitea_resolve_task_capability",
"original_task",
],
"exact_safe_next_action": None,
"blocker_kind": BLOCKER_NONE,
"operator_ui_steps": steps,
"typed_blocker": None,
}
if reconnect_needed:
result["blocker_kind"] = BLOCKER_OPERATOR_RECONNECT
result["stop_required"] = True
result["restart_required"] = True
result["exact_safe_next_action"] = (
f"blocker_kind={BLOCKER_OPERATOR_RECONNECT}: operator must reconnect "
f"MCP namespace '{ns}' via the host UI (client={client_key}). "
"Do not pkill, touch configs, edit session state, or use raw API. "
"After reconnect, restart from gitea_whoami → "
"gitea_resolve_task_capability → task."
)
result["typed_blocker"] = {
"blocker_kind": BLOCKER_OPERATOR_RECONNECT,
"namespaces": [ns],
"why_reconnect_required": normalized_reason,
"operator_ui_steps": steps,
"client": client_key,
"forbidden_recovery_paths": list(FORBIDDEN_RECOVERY_PATHS),
"instruction_after_reconnect": (
"Rerun the blocked workflow from preflight "
"(gitea_whoami → gitea_resolve_task_capability → task). "
"Do not continue mid-mutation from pre-reconnect state."
),
}
else:
result["stop_required"] = False
result["restart_required"] = False
result["exact_safe_next_action"] = (
f"Reconnect not required for namespace '{ns}' "
f"(boundary_status={boundary}). Proceed with the original task."
)
if extra:
for key, value in extra.items():
if key not in result:
result[key] = value
return result
def reasons_never_suggest_forbidden(text: str) -> bool:
"""Return True when *text* does not recommend a forbidden recovery path.
Mentions that *ban* a path (e.g. ``Do not pkill`` / ``never edit session
state``) are allowed. Positive recommendations such as ``use pkill`` or
``run killall`` fail.
"""
import re
lowered = (text or "").lower()
# Strip common ban prefixes so "do not pkill" does not trip positive checks.
scrubbed = re.sub(
r"\b(?:do not|don't|never|must not|forbid(?:den)?|ban(?:ned)?)\b"
r"[^.!;\n]{0,80}",
" ",
lowered,
)
# Positive imperative / advisory forms that would tell an agent to do harm.
positive_suggestions = (
"use pkill",
"run pkill",
"try pkill",
"pkill -f",
"use killall",
"run killall",
"killall mcp",
"use kill ",
"run kill ",
"touch the mcp",
"touch mcp config",
"utime(",
"edit the mcp config to recover",
"edit .env to recover",
"import gitea_mcp_server",
"python -c 'import gitea_mcp",
)
return not any(frag in scrubbed for frag in positive_suggestions)
+237
View File
@@ -0,0 +1,237 @@
"""Antigravity IDE vs Global MCP Config Drift Diagnostic (#672).
Diagnoses config drift between the active IDE MCP configuration
(e.g. ``~/.gemini/antigravity-ide/mcp_config.json``) and the offline/global
canonical configuration (e.g. ``~/.gemini/config/mcp_config.json``).
Hard rules (#672 / #630 / #655):
* Distinguish offline/global success from active IDE namespace availability.
* Never print tokens, DSNs, Authorization headers, or secret-bearing env vars.
* Sanctioned repair path is: backup active config -> patch active config from canonical
-> reconnect through IDE/client -> verify with live ``gitea_whoami``.
* FORBIDDEN: ``pkill``, mtime edits, source edits, or session-state edits for repair.
"""
from __future__ import annotations
import argparse
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from webui import console_redaction
DEFAULT_ACTIVE_IDE_CONFIG = "~/.gemini/antigravity-ide/mcp_config.json"
DEFAULT_GLOBAL_CONFIG = "~/.gemini/config/mcp_config.json"
REQUIRED_GITEA_ROLE_SERVERS = (
"gitea-author",
"gitea-reviewer",
"gitea-merger",
"gitea-reconciler",
"gitea-controller",
"gitea-tools",
)
SANCTIONED_REPAIR_RUNBOOK: tuple[str, ...] = (
"1. Backup active IDE config: cp ~/.gemini/antigravity-ide/mcp_config.json ~/.gemini/antigravity-ide/mcp_config.json.bak",
"2. Patch active IDE config: copy required missing Gitea role server entries from global config (~/.gemini/config/mcp_config.json) into active IDE config.",
"3. Reconnect via IDE/client UI or client restart (do NOT use host process kill).",
"4. Verify active namespace health using live gitea_whoami and gitea_resolve_task_capability on each role namespace.",
"FORBIDDEN REPAIR PATHS: pkill / host process kill, mtime touch edits, source code edits, or session-state edits.",
)
def resolve_config_path(path_str: str) -> Path:
"""Expand user and resolve absolute path."""
return Path(os.path.expanduser(path_str)).resolve()
def load_mcp_config(config_path: str | Path) -> tuple[dict[str, Any] | None, str | None]:
"""Load and parse JSON MCP configuration from file.
Returns (config_dict, error_message).
"""
resolved = resolve_config_path(str(config_path))
if not resolved.exists():
return None, f"file_not_found: {resolved}"
try:
with open(resolved, "r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
return None, f"invalid_schema: root is not a JSON object in {resolved}"
return data, None
except Exception as exc:
return None, f"unreadable_json: {exc} in {resolved}"
def extract_mcp_servers(config: dict[str, Any] | None) -> dict[str, dict[str, Any]]:
"""Extract the mcpServers or mcp_servers mapping safely."""
if not config:
return {}
servers = config.get("mcpServers") or config.get("mcp_servers") or {}
if isinstance(servers, dict):
return {str(k): v for k, v in servers.items() if isinstance(v, dict)}
return {}
def _safe_redact_server_config(srv_cfg: dict[str, Any]) -> dict[str, Any]:
"""Redact secrets from environment variables and command line args."""
safe = {}
if "command" in srv_cfg:
safe["command"] = str(srv_cfg["command"])
if "args" in srv_cfg and isinstance(srv_cfg["args"], list):
safe["args"] = [console_redaction.redact_text(str(a)) for a in srv_cfg["args"]]
if "env" in srv_cfg and isinstance(srv_cfg["env"], dict):
safe_env = {}
for k, v in srv_cfg["env"].items():
if any(secret_kw in k.lower() for secret_kw in ("token", "secret", "pass", "key", "auth")):
safe_env[k] = "[REDACTED]"
else:
safe_env[k] = console_redaction.redact_text(str(v))
safe["env"] = safe_env
return safe
def analyze_config_drift(
active_config_path: str = DEFAULT_ACTIVE_IDE_CONFIG,
global_config_path: str = DEFAULT_GLOBAL_CONFIG,
) -> dict[str, Any]:
"""Analyze MCP configuration drift between active IDE config and global config.
Returns structured diagnostic output.
"""
active_resolved = resolve_config_path(active_config_path)
global_resolved = resolve_config_path(global_config_path)
active_cfg, active_err = load_mcp_config(active_resolved)
global_cfg, global_err = load_mcp_config(global_resolved)
active_servers = extract_mcp_servers(active_cfg)
global_servers = extract_mcp_servers(global_cfg)
missing_role_servers: list[str] = []
present_role_servers: list[str] = []
profile_mismatches: list[dict[str, Any]] = []
reasons: list[str] = []
if active_err:
reasons.append(f"Active IDE config error: {active_err}")
if global_err:
reasons.append(f"Global canonical config error: {global_err}")
# Check Gitea role servers
for srv_name in REQUIRED_GITEA_ROLE_SERVERS:
in_active = srv_name in active_servers
in_global = srv_name in global_servers
if in_active:
present_role_servers.append(srv_name)
elif in_global:
missing_role_servers.append(srv_name)
reasons.append(
f"Missing Gitea role server '{srv_name}' in active IDE config ({active_resolved})"
)
if in_active and in_global:
# Compare profiles & environments
act_env = active_servers[srv_name].get("env", {}) if isinstance(active_servers[srv_name], dict) else {}
glo_env = global_servers[srv_name].get("env", {}) if isinstance(global_servers[srv_name], dict) else {}
act_prof = act_env.get("GITEA_MCP_PROFILE") or act_env.get("GITEA_PROFILE_NAME")
glo_prof = glo_env.get("GITEA_MCP_PROFILE") or glo_env.get("GITEA_PROFILE_NAME")
if act_prof != glo_prof:
mismatch_item = {
"server": srv_name,
"active_profile": act_prof,
"global_profile": glo_prof,
}
profile_mismatches.append(mismatch_item)
reasons.append(
f"Profile mismatch for '{srv_name}': active='{act_prof}' != global='{glo_prof}'"
)
in_sync = bool(
not active_err
and not global_err
and not missing_role_servers
and not profile_mismatches
)
report = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"in_sync": in_sync,
"active_config_path": str(active_resolved),
"active_config_exists": active_cfg is not None,
"global_config_path": str(global_resolved),
"global_config_exists": global_cfg is not None,
"required_role_servers": list(REQUIRED_GITEA_ROLE_SERVERS),
"present_role_servers": present_role_servers,
"missing_role_servers": missing_role_servers,
"profile_mismatches": profile_mismatches,
"reasons": reasons,
"sanctioned_repair_runbook": list(SANCTIONED_REPAIR_RUNBOOK),
"forbidden_repair_methods": [
"pkill / host process kill",
"mtime touch edits",
"source code edits",
"session-state edits",
],
}
return console_redaction.redact_payload(report)
def main() -> None:
parser = argparse.ArgumentParser(
description="Diagnose Gitea MCP role server config drift between active IDE and global config."
)
parser.add_argument(
"--active-config",
default=DEFAULT_ACTIVE_IDE_CONFIG,
help="Path to active IDE MCP config JSON",
)
parser.add_argument(
"--global-config",
default=DEFAULT_GLOBAL_CONFIG,
help="Path to global/canonical MCP config JSON",
)
parser.add_argument(
"--json", action="store_true", help="Print raw JSON report"
)
args = parser.parse_args()
report = analyze_config_drift(args.active_config, args.global_config)
if args.json:
print(json.dumps(report, indent=2))
else:
print("=== MCP Config Drift Diagnostic Report ===")
print(f"Timestamp: {report['timestamp']}")
print(f"In Sync: {report['in_sync']}")
print(f"Active IDE Config: {report['active_config_path']} (exists={report['active_config_exists']})")
print(f"Global Config: {report['global_config_path']} (exists={report['global_config_exists']})")
print(f"Present Role Servers: {', '.join(report['present_role_servers']) if report['present_role_servers'] else 'None'}")
print(f"Missing Role Servers: {', '.join(report['missing_role_servers']) if report['missing_role_servers'] else 'None'}")
if report['profile_mismatches']:
print("Profile Mismatches:")
for m in report['profile_mismatches']:
print(f" - {m['server']}: active={m['active_profile']} vs global={m['global_profile']}")
if report['reasons']:
print("Drift Reasons:")
for r in report['reasons']:
print(f" - {r}")
print("\nSanctioned Repair Runbook:")
for step in report['sanctioned_repair_runbook']:
print(f" {step}")
sys.exit(0 if report["in_sync"] else 1)
if __name__ == "__main__":
main()
+5 -35
View File
@@ -225,33 +225,8 @@ _RESTART_PATHS: tuple[RestartPath, ...] = (
"exact_safe_next_action pointing at IDE/client reconnect; performs "
"no restart, thread spawn, config touch, or os._exit."
),
locations=(
"gitea_mcp_server.py (gitea_resolve_task_capability)",
"gitea_mcp_server.py (gitea_request_mcp_reconnect)",
"mcp_client_reconnect.py",
),
references=("#685", "#657", "#678"),
),
RestartPath(
path_id="codex_client_reconnect_request",
title="Sanctioned Codex/LLM reconnect request tool",
mechanism=(
"gitea_request_mcp_reconnect: agents invoke a report-only tool that "
"returns namespace/profile/pid/startup SHA/master SHA/boundary "
"status plus a typed operator blocker with exact client UI steps."
),
classification=CLASS_GUARDED_FAIL_CLOSED,
guard=(
"Report-only (#678): never restarts, kills, reloads, or edits "
"config; recovery is always host/operator reconnect. Forbidden "
"paths (pkill, touch, .env/config/session-state hacks) are listed "
"and never recommended."
),
locations=(
"mcp_client_reconnect.py",
"gitea_mcp_server.py (gitea_request_mcp_reconnect)",
),
references=("#678", "#630", "#685", "#657"),
locations=("gitea_mcp_server.py (gitea_resolve_task_capability)",),
references=("#685", "#657"),
),
RestartPath(
path_id="manual_daemon_kill",
@@ -296,8 +271,7 @@ _RESTART_PATHS: tuple[RestartPath, ...] = (
title="Host/IDE MCP reconnect",
mechanism=(
"A manual `/mcp reconnect` (or equivalent host action) that the "
"IDE performs to recreate the MCP client connection. Agents obtain "
"exact UI steps via gitea_request_mcp_reconnect (#678)."
"IDE performs to recreate the MCP client connection."
),
classification=CLASS_HOST_RESIDUAL,
guard=(
@@ -305,12 +279,8 @@ _RESTART_PATHS: tuple[RestartPath, ...] = (
"gates point operators toward; documented as residual host "
"behavior. No in-process code initiates it."
),
locations=(
"host/IDE",
"mcp_client_reconnect.py",
"gitea_mcp_server.py (gitea_request_mcp_reconnect)",
),
references=("#584", "#656", "#657", "#678"),
locations=("host/IDE",),
references=("#584", "#656", "#657"),
residual_host=True,
),
RestartPath(
@@ -1,323 +0,0 @@
"""Tests for sanctioned Codex MCP reconnect request surface (#678)."""
from __future__ import annotations
import os
import unittest
from unittest import mock
import mcp_client_reconnect as mcr
class NormalizeReasonTests(unittest.TestCase):
def test_stale_runtime_aliases(self):
self.assertEqual(mcr.normalize_reason("stale-runtime"), mcr.REASON_STALE_RUNTIME)
self.assertEqual(mcr.normalize_reason("stale_runtime"), mcr.REASON_STALE_RUNTIME)
self.assertEqual(mcr.normalize_reason("STALE"), mcr.REASON_STALE_RUNTIME)
def test_transport_eof_aliases(self):
self.assertEqual(mcr.normalize_reason("transport_eof"), mcr.REASON_TRANSPORT_EOF)
self.assertEqual(mcr.normalize_reason("EOF"), mcr.REASON_TRANSPORT_EOF)
self.assertEqual(
mcr.normalize_reason("client_is_closing"), mcr.REASON_TRANSPORT_EOF
)
def test_missing_namespace(self):
self.assertEqual(
mcr.normalize_reason("missing_namespace"), mcr.REASON_MISSING_NAMESPACE
)
def test_empty_is_unspecified(self):
self.assertEqual(mcr.normalize_reason(None), mcr.REASON_UNSPECIFIED)
self.assertEqual(mcr.normalize_reason(""), mcr.REASON_UNSPECIFIED)
class BoundaryClassificationTests(unittest.TestCase):
def test_clean_when_shas_match(self):
self.assertEqual(
mcr.classify_boundary_status(
startup_sha="abc", current_master_sha="abc"
),
mcr.BOUNDARY_CLEAN,
)
def test_mismatch_when_shas_differ(self):
self.assertEqual(
mcr.classify_boundary_status(
startup_sha="aaa", current_master_sha="bbb"
),
mcr.BOUNDARY_MISMATCH,
)
def test_stale_when_live_stale(self):
self.assertEqual(
mcr.classify_boundary_status(
startup_sha="aaa",
current_master_sha="aaa",
live_stale=True,
),
mcr.BOUNDARY_STALE,
)
class BuildReconnectRequestTests(unittest.TestCase):
def test_stale_runtime_returns_typed_blocker_with_codex_steps(self):
result = mcr.build_reconnect_request(
namespace="gitea-author",
profile="prgs-author",
pid=1234,
session_id="sess-1",
startup_sha="aaa111",
current_master_sha="bbb222",
reason="stale-runtime",
client="codex",
restart_required=True,
stop_required=True,
)
self.assertTrue(result["success"])
self.assertTrue(result["read_only"])
self.assertFalse(result["reconnect_performed"])
self.assertFalse(result["mutation_performed"])
self.assertTrue(result["reconnect_needed"])
self.assertEqual(result["namespace"], "gitea-author")
self.assertEqual(result["profile"], "prgs-author")
self.assertEqual(result["pid"], 1234)
self.assertEqual(result["session_id"], "sess-1")
self.assertEqual(result["startup_sha"], "aaa111")
self.assertEqual(result["current_master_sha"], "bbb222")
self.assertEqual(result["boundary_status"], mcr.BOUNDARY_MISMATCH)
self.assertEqual(result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT)
self.assertIsNotNone(result["typed_blocker"])
blocker = result["typed_blocker"]
self.assertEqual(blocker["namespaces"], ["gitea-author"])
self.assertEqual(blocker["why_reconnect_required"], mcr.REASON_STALE_RUNTIME)
self.assertTrue(any("Codex" in s or "Reload" in s for s in blocker["operator_ui_steps"]))
self.assertIn("pkill", " ".join(result["forbidden_recovery_paths"]).lower())
self.assertTrue(
mcr.reasons_never_suggest_forbidden(result["exact_safe_next_action"] or "")
)
# Must not recommend forbidden recovery.
for step in blocker["operator_ui_steps"]:
self.assertTrue(mcr.reasons_never_suggest_forbidden(step), step)
def test_transport_eof_typed_blocker(self):
result = mcr.build_reconnect_request(
namespace="gitea-reviewer",
reason="transport_eof",
client="claude_code",
)
self.assertTrue(result["reconnect_needed"])
self.assertEqual(result["reason"], mcr.REASON_TRANSPORT_EOF)
self.assertEqual(result["client"], "claude_code")
steps = " ".join(result["operator_ui_steps"]).lower()
self.assertIn("/mcp", steps)
def test_missing_namespace_typed_blocker(self):
result = mcr.build_reconnect_request(
namespace="gitea-merger",
reason="missing_namespace",
client="codex",
)
self.assertTrue(result["reconnect_needed"])
self.assertEqual(result["reason"], mcr.REASON_MISSING_NAMESPACE)
self.assertEqual(
result["typed_blocker"]["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT
)
def test_healthy_not_required(self):
result = mcr.build_reconnect_request(
namespace="gitea-tools",
startup_sha="deadbeef",
current_master_sha="deadbeef",
reason="not_required",
client="codex",
in_parity=True,
restart_required=False,
stop_required=False,
)
self.assertFalse(result["reconnect_needed"])
self.assertEqual(result["blocker_kind"], mcr.BLOCKER_NONE)
self.assertIsNone(result["typed_blocker"])
self.assertFalse(result["stop_required"])
self.assertFalse(result["restart_required"])
self.assertIn("not required", (result["exact_safe_next_action"] or "").lower())
def test_successful_reconnect_report_fields_present(self):
"""AC2: reconnect result reports required fields (even when needed)."""
result = mcr.build_reconnect_request(
namespace="gitea-controller",
profile="prgs-controller",
pid=99,
session_id="sid",
startup_sha="s" * 40,
current_master_sha="c" * 40,
reason="stale-runtime",
)
for key in (
"namespace",
"profile",
"pid",
"session_id",
"startup_sha",
"current_master_sha",
"boundary_status",
):
self.assertIn(key, result)
self.assertIsNotNone(result[key], key)
class ToolSurfaceTests(unittest.TestCase):
"""Exercise gitea_request_mcp_reconnect with a stubbed server context."""
def test_tool_is_registered_and_side_effect_free(self):
import gitea_mcp_server as srv
self.assertTrue(hasattr(srv, "gitea_request_mcp_reconnect"))
with mock.patch.object(srv, "_profile_operation_gate", return_value=None):
with mock.patch.object(
srv,
"get_profile",
return_value={
"profile_name": "prgs-author",
"role_kind": "author",
"role": "author",
},
):
with mock.patch.object(
srv,
"_current_master_parity",
return_value={
"startup_head": "a" * 40,
"current_head": "a" * 40,
"daemon_start_head": "a" * 40,
"local_head": "a" * 40,
"in_parity": True,
"stale": False,
"restart_required": False,
"determinable": True,
"live_stale": False,
"live_known": True,
"reasons": [],
},
):
with mock.patch.object(
srv.master_parity_gate,
"format_parity",
return_value="in parity",
):
with mock.patch.object(
srv.role_namespace_gate,
"infer_mcp_namespace",
return_value="gitea-author",
):
with mock.patch.object(
srv.session_ctx,
"mutation_context_audit_fields",
return_value={"session_profile": "prgs-author"},
):
result = srv.gitea_request_mcp_reconnect(
namespace="gitea-author",
reason="not_required",
client="codex",
remote="prgs",
)
self.assertTrue(result.get("success"))
self.assertFalse(result.get("reconnect_performed"))
self.assertFalse(result.get("mutation_performed"))
self.assertEqual(result.get("namespace"), "gitea-author")
self.assertEqual(result.get("profile"), "prgs-author")
self.assertEqual(result.get("pid"), os.getpid())
self.assertIn("startup_sha", result)
self.assertIn("current_master_sha", result)
self.assertIn("boundary_status", result)
self.assertTrue(
mcr.reasons_never_suggest_forbidden(
result.get("exact_safe_next_action") or ""
)
)
def test_tool_stale_returns_typed_blocker(self):
import gitea_mcp_server as srv
with mock.patch.object(srv, "_profile_operation_gate", return_value=None):
with mock.patch.object(
srv,
"get_profile",
return_value={
"profile_name": "prgs-reconciler",
"role_kind": "reconciler",
"role": "reconciler",
},
):
with mock.patch.object(
srv,
"_current_master_parity",
return_value={
"startup_head": "a" * 40,
"current_head": "b" * 40,
"daemon_start_head": "a" * 40,
"local_head": "b" * 40,
"in_parity": False,
"stale": True,
"restart_required": True,
"determinable": True,
"live_stale": True,
"live_known": True,
"reasons": ["stale"],
},
):
with mock.patch.object(
srv.master_parity_gate,
"format_parity",
return_value="stale",
):
with mock.patch.object(
srv.role_namespace_gate,
"infer_mcp_namespace",
return_value="gitea-reconciler",
):
with mock.patch.object(
srv.session_ctx,
"mutation_context_audit_fields",
return_value={},
):
result = srv.gitea_request_mcp_reconnect(
reason="stale-runtime",
client="codex",
)
self.assertTrue(result["reconnect_needed"])
self.assertEqual(
result["blocker_kind"], mcr.BLOCKER_OPERATOR_RECONNECT
)
self.assertIsNotNone(result["typed_blocker"])
self.assertIn("gitea-reconciler", result["typed_blocker"]["namespaces"])
self.assertTrue(result["stop_required"])
self.assertTrue(result["restart_required"])
self.assertTrue(
mcr.reasons_never_suggest_forbidden(
result.get("exact_safe_next_action") or ""
)
)
class InventoryRegistrationTests(unittest.TestCase):
def test_reconnect_path_in_restart_inventory(self):
import mcp_restart_paths as mrp
ids = {p.path_id for p in mrp.iter_restart_paths()}
self.assertIn("codex_client_reconnect_request", ids)
self.assertIn("ide_client_reconnect", ids)
def test_tool_name_in_documented_inventory(self):
import mcp_tool_inventory as inv
doc_path = os.path.join(
os.path.dirname(os.path.dirname(__file__)), inv.INVENTORY_DOC_PATH
)
with open(doc_path, encoding="utf-8") as handle:
documented = inv.parse_documented_inventory(handle.read())
self.assertIn("gitea_request_mcp_reconnect", documented)
if __name__ == "__main__":
unittest.main()
+149
View File
@@ -0,0 +1,149 @@
"""Unit tests for mcp_config_drift.py (#672)."""
from __future__ import annotations
import json
import pytest
from pathlib import Path
from mcp_config_drift import (
REQUIRED_GITEA_ROLE_SERVERS,
analyze_config_drift,
load_mcp_config,
)
@pytest.fixture
def sample_global_config() -> dict:
return {
"mcpServers": {
"gitea-author": {
"command": "python3",
"args": ["gitea_mcp_server.py"],
"env": {"GITEA_MCP_PROFILE": "prgs-author", "SENTRY_AUTH_TOKEN": "secret-token-999"},
},
"gitea-reviewer": {
"command": "python3",
"args": ["gitea_mcp_server.py"],
"env": {"GITEA_MCP_PROFILE": "prgs-reviewer"},
},
"gitea-merger": {
"command": "python3",
"args": ["gitea_mcp_server.py"],
"env": {"GITEA_MCP_PROFILE": "prgs-merger"},
},
"gitea-reconciler": {
"command": "python3",
"args": ["gitea_mcp_server.py"],
"env": {"GITEA_MCP_PROFILE": "prgs-reconciler"},
},
"gitea-controller": {
"command": "python3",
"args": ["gitea_mcp_server.py"],
"env": {"GITEA_MCP_PROFILE": "prgs-controller"},
},
"gitea-tools": {
"command": "python3",
"args": ["gitea_mcp_server.py"],
"env": {"GITEA_MCP_PROFILE": "prgs-author"},
},
}
}
def write_json(path: Path, data: dict) -> str:
path.write_text(json.dumps(data, indent=2), encoding="utf-8")
return str(path)
def test_drift_detection_in_sync(tmp_path, sample_global_config):
glob_file = tmp_path / "global_mcp.json"
act_file = tmp_path / "active_mcp.json"
write_json(glob_file, sample_global_config)
write_json(act_file, sample_global_config)
report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
assert report["in_sync"] is True
assert report["missing_role_servers"] == []
assert report["profile_mismatches"] == []
assert set(report["present_role_servers"]) == set(REQUIRED_GITEA_ROLE_SERVERS)
def test_drift_detection_missing_author(tmp_path, sample_global_config):
glob_file = tmp_path / "global_mcp.json"
act_file = tmp_path / "active_mcp.json"
active_config = json.loads(json.dumps(sample_global_config))
del active_config["mcpServers"]["gitea-author"]
write_json(glob_file, sample_global_config)
write_json(act_file, active_config)
report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
assert report["in_sync"] is False
assert "gitea-author" in report["missing_role_servers"]
assert "gitea-author" not in report["present_role_servers"]
def test_drift_detection_missing_reviewer(tmp_path, sample_global_config):
glob_file = tmp_path / "global_mcp.json"
act_file = tmp_path / "active_mcp.json"
active_config = json.loads(json.dumps(sample_global_config))
del active_config["mcpServers"]["gitea-reviewer"]
write_json(glob_file, sample_global_config)
write_json(act_file, active_config)
report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
assert report["in_sync"] is False
assert "gitea-reviewer" in report["missing_role_servers"]
def test_drift_detection_profile_mismatch(tmp_path, sample_global_config):
glob_file = tmp_path / "global_mcp.json"
act_file = tmp_path / "active_mcp.json"
active_config = json.loads(json.dumps(sample_global_config))
active_config["mcpServers"]["gitea-author"]["env"]["GITEA_MCP_PROFILE"] = "dadeschools-author"
write_json(glob_file, sample_global_config)
write_json(act_file, active_config)
report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
assert report["in_sync"] is False
assert len(report["profile_mismatches"]) == 1
mismatch = report["profile_mismatches"][0]
assert mismatch["server"] == "gitea-author"
assert mismatch["active_profile"] == "dadeschools-author"
assert mismatch["global_profile"] == "prgs-author"
def test_secret_redaction_in_drift_report(tmp_path, sample_global_config):
glob_file = tmp_path / "global_mcp.json"
act_file = tmp_path / "active_mcp.json"
write_json(glob_file, sample_global_config)
write_json(act_file, sample_global_config)
report = analyze_config_drift(active_config_path=str(act_file), global_config_path=str(glob_file))
serialized = str(report)
assert "secret-token-999" not in serialized
def test_sanctioned_runbook_forbids_pkill():
report = analyze_config_drift(active_config_path="/nonexistent/path/active.json", global_config_path="/nonexistent/path/global.json")
runbook_text = " ".join(report["sanctioned_repair_runbook"]).lower()
forbidden_text = " ".join(report["forbidden_repair_methods"]).lower()
assert "pkill" in forbidden_text
assert "mtime" in forbidden_text
assert "source" in forbidden_text
assert "session-state" in forbidden_text
-478
View File
@@ -1,478 +0,0 @@
"""Concurrent-session MCP restart safety & dogfooding test suite (#666).
Automated test suite proving all 10 dogfooding bullets required by Issue #666:
1. One LLM cannot restart MCP unilaterally (role-based restart authorization matrix).
2. New work stops during drain (assignments_stopped gate enforcement).
3. Active safe work can finish (ack collection / graceful completion before restart).
4. Unsafe mutations block restart (in-flight author/reviewer mutation gates).
5. Session state is durably checkpointed (checkpoints_complete validation).
6. Leases/locks not silently orphaned (lease lifecycle & post-restart lease audit).
7. Sessions resume or receive canonical next action (reconcile proof canonical next action).
8. Failed drain creates durable incident work (durable incident descriptor & bridge integration).
9. Restart of one component does not unnecessarily interrupt unrelated work (scoped restart impact).
10. Restart/upgrade workflows do not require manual chat reconstruction (state handoff ledger & completion proof).
Links parent #655, vision #652, roadmap #653, #658, #659, #660, #661, #662, #663.
"""
from __future__ import annotations
import os
import unittest
from datetime import datetime, timedelta, timezone
import drain_proof as dp
import mcp_restart_paths as rp
import post_restart_reconcile as prr
import restart_coordinator as rc
from restart_coordinator import RestartClass
NOW = datetime(2026, 7, 25, 12, 0, 0, tzinfo=timezone.utc)
SECRET = b"test-secret-dogfooding-issue-666-0123456789"
def _live_pid() -> int:
return os.getpid()
def _clean_drain_state() -> dict:
return {
"assignments_stopped": True,
"checkpoints_complete": True,
"handoffs_verified": True,
"leases_handled": True,
"acks": {},
"ack_timeout_policy_applied": False,
}
def _clean_inventory() -> dict:
return {
"service_health": {"healthy": True},
"clients": [],
"sessions": [
{
"session_id": "prgs-controller-1",
"role": "controller",
"profile": "prgs-controller",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
}
],
"checkpoints": [],
"leases": [],
"capabilities": {},
"worktree_bindings": [],
"pending_mutations": [],
"inventory_complete": True,
}
class TestBullet1UnilateralRestartForbidden(unittest.TestCase):
"""Bullet 1: One LLM cannot restart MCP unilaterally."""
def test_worker_role_unilateral_full_restart_denied(self):
policy = rc.RESTART_CLASS_POLICIES[RestartClass.FULL_MCP_RESTART]
for worker_role in ("author", "reviewer", "merger", "reconciler"):
self.assertNotIn(
worker_role,
policy.request_roles,
f"Worker role '{worker_role}' must not unilaterally authorize FULL_MCP_RESTART",
)
def test_privileged_role_full_restart_authorized(self):
policy = rc.RESTART_CLASS_POLICIES[RestartClass.FULL_MCP_RESTART]
for priv_role in ("controller", "operator", "admin"):
self.assertIn(
priv_role,
policy.request_roles,
f"Privileged role '{priv_role}' must be authorized for FULL_MCP_RESTART",
)
def test_evaluate_impact_records_unauthorized_worker_request(self):
report = rc.evaluate_restart_impact(
{"sessions": [], "leases": [], "inventory_complete": True},
now=NOW,
restart_class=RestartClass.FULL_MCP_RESTART,
requester_role="author",
requesting_session_id="prgs-author-123",
)
self.assertFalse(report.role_authorized)
self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
self.assertTrue(any("may not request" in r.lower() or "authorization denied" in r.lower() for r in report.reasons))
class TestBullet2NewWorkStopsDuringDrain(unittest.TestCase):
"""Bullet 2: New work stops during drain."""
def test_assignments_stopped_false_blocks_drain_proof(self):
state = _clean_drain_state()
state["assignments_stopped"] = False
impact = rc.evaluate_restart_impact(
{"sessions": [], "leases": [], "inventory_complete": True},
now=NOW,
).as_dict()
proof = dp.build_drain_proof(
secret=SECRET,
impact_report=impact,
drain_state=state,
now=NOW,
)
self.assertFalse(proof.clean)
check = next(c for c in proof.checks if c.name == dp.CHECK_ASSIGNMENTS_STOPPED)
self.assertFalse(check.passed)
gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW)
self.assertEqual(gate.verdict, dp.GATE_DENY)
self.assertFalse(gate.allow)
self.assertTrue(any("drain proof invalid" in r.lower() or "assignments_stopped" in r.lower() for r in gate.reasons))
class TestBullet3ActiveSafeWorkCanFinish(unittest.TestCase):
"""Bullet 3: Active safe work can finish."""
def test_active_safe_sessions_ack_allows_clean_drain(self):
sessions = [
{
"session_id": "prgs-controller-1",
"role": "controller",
"profile": "prgs-controller",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
{
"session_id": "prgs-reviewer-42",
"role": "reviewer",
"profile": "prgs-reviewer",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
]
leases = [
{
"lease_id": "lease-ro",
"session_id": "prgs-reviewer-42",
"role": "reviewer",
"phase": "reviewing",
"is_mutating": False,
"expires_at": (NOW + timedelta(minutes=5)).isoformat(),
"pid": _live_pid(),
}
]
impact = rc.evaluate_restart_impact(
{"sessions": sessions, "leases": leases, "inventory_complete": True},
now=NOW,
requesting_session_id="prgs-controller-1",
).as_dict()
state = _clean_drain_state()
state["acks"] = {"prgs-reviewer-42": "ack"}
proof = dp.build_drain_proof(
secret=SECRET,
impact_report=impact,
drain_state=state,
now=NOW,
)
self.assertTrue(proof.clean)
gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW)
self.assertTrue(gate.allow)
self.assertEqual(gate.verdict, dp.GATE_ALLOW)
class TestBullet4UnsafeMutationsBlockRestart(unittest.TestCase):
"""Bullet 4: Unsafe mutations block restart."""
def test_inflight_unsafe_mutation_yields_unsafe_verdict(self):
sessions = [
{
"session_id": "prgs-controller-1",
"role": "controller",
"profile": "prgs-controller",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
{
"session_id": "prgs-author-99",
"role": "author",
"profile": "prgs-author",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
]
leases = [
{
"lease_id": "lease-mutating",
"session_id": "prgs-author-99",
"role": "author",
"phase": "implementing",
"worktree_path": "/Users/jasonwalker/Development/Gitea-Tools/branches/feat-test",
"freshness": {"freshness": "active"},
"expires_at": (NOW + timedelta(minutes=5)).isoformat(),
"pid": _live_pid(),
}
]
report = rc.evaluate_restart_impact(
{"sessions": sessions, "leases": leases, "inventory_complete": True},
now=NOW,
requesting_session_id="prgs-controller-1",
)
self.assertEqual(report.verdict, rc.VERDICT_UNSAFE)
self.assertFalse(report.allow_restart)
self.assertGreater(len(report.mutations), 0)
proof = dp.build_drain_proof(
secret=SECRET,
impact_report=report.as_dict(),
drain_state=_clean_drain_state(),
now=NOW,
)
self.assertFalse(proof.clean)
check = next(c for c in proof.checks if c.name == dp.CHECK_NO_INFLIGHT_MUTATIONS)
self.assertFalse(check.passed)
gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW)
self.assertEqual(gate.verdict, dp.GATE_DENY)
self.assertFalse(gate.allow)
class TestBullet5DurableSessionCheckpoints(unittest.TestCase):
"""Bullet 5: Session state is durably checkpointed."""
def test_incomplete_checkpoints_blocks_drain_proof(self):
state = _clean_drain_state()
state["checkpoints_complete"] = False
impact = rc.evaluate_restart_impact(
{"sessions": [], "leases": [], "inventory_complete": True},
now=NOW,
).as_dict()
proof = dp.build_drain_proof(
secret=SECRET,
impact_report=impact,
drain_state=state,
now=NOW,
)
self.assertFalse(proof.clean)
check = next(c for c in proof.checks if c.name == dp.CHECK_CHECKPOINTS_COMPLETE)
self.assertFalse(check.passed)
def test_post_restart_reconcile_audits_checkpoint_dimension(self):
inv = _clean_inventory()
inv["checkpoints_available"] = True
inv["checkpoints"] = [
{
"session_id": "prgs-author-99",
"checkpoint_id": "chk-1",
"stale": True,
}
]
proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_ENFORCE)
chk_item = next(i for i in proof.items if i.dimension == prr.DIM_CHECKPOINTS)
self.assertIn(chk_item.status, (prr.ITEM_UNRESOLVED, prr.ITEM_DEGRADED, prr.ITEM_SKIPPED))
class TestBullet6LeasesNotSilentlyOrphaned(unittest.TestCase):
"""Bullet 6: Leases/locks not silently orphaned."""
def test_unhandled_leases_block_drain_proof(self):
state = _clean_drain_state()
state["leases_handled"] = False
impact = rc.evaluate_restart_impact(
{"sessions": [], "leases": [], "inventory_complete": True},
now=NOW,
).as_dict()
proof = dp.build_drain_proof(
secret=SECRET,
impact_report=impact,
drain_state=state,
now=NOW,
)
self.assertFalse(proof.clean)
check = next(c for c in proof.checks if c.name == dp.CHECK_LEASES_HANDLED)
self.assertFalse(check.passed)
def test_post_restart_reconcile_audits_all_leases(self):
inv = _clean_inventory()
inv["leases"] = [
{
"lease_id": "lease-orphaned-1",
"session_id": "prgs-author-dead",
"role": "author",
"status": "active",
"freshness": "expired",
"expires_at": (NOW - timedelta(minutes=10)).isoformat(),
}
]
proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_LOG_ONLY)
lease_item = next(i for i in proof.items if i.dimension == prr.DIM_LEASES)
self.assertIsNotNone(lease_item)
self.assertTrue(lease_item.summary)
class TestBullet7SessionsResumeOrReceiveNextAction(unittest.TestCase):
"""Bullet 7: Sessions resume or receive canonical next action."""
def test_reconcile_provides_canonical_next_action_for_unresolved(self):
inv = _clean_inventory()
inv["pending_mutations"] = [
{
"mutation_id": "mut-404",
"session_id": "prgs-author-77",
"phase": "implementing",
"issue_number": 666,
}
]
proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_ENFORCE)
self.assertEqual(proof.overall_status, prr.STATUS_DEGRADED)
self.assertTrue(proof.mutation_hold)
self.assertTrue(proof.note)
self.assertGreater(len(proof.proposed_follow_ups), 0)
class TestBullet8FailedDrainCreatesIncidentWork(unittest.TestCase):
"""Bullet 8: Failed drain creates durable incident work."""
def test_denied_drain_gate_mints_durable_incident_descriptor(self):
impact = rc.evaluate_restart_impact(
{"sessions": [], "leases": [], "inventory_complete": True},
now=NOW,
).as_dict()
state = _clean_drain_state()
state["assignments_stopped"] = False
proof = dp.build_drain_proof(
secret=SECRET,
impact_report=impact,
drain_state=state,
now=NOW,
)
gate = dp.gate_apply_restart(proof=proof.as_dict(), secret=SECRET, now=NOW)
self.assertEqual(gate.verdict, dp.GATE_DENY)
incident = gate.incident
self.assertIsNotNone(incident)
self.assertEqual(incident["kind"], "restart_drain_gate_denied")
self.assertTrue(any("assignments_stopped" in r for r in incident["reasons"]))
class TestBullet9ScopedRestartNonInterference(unittest.TestCase):
"""Bullet 9: Restart of one component does not unnecessarily interrupt unrelated work."""
def test_scoped_role_restart_impacts_only_target_role(self):
sessions = [
{
"session_id": "prgs-controller-1",
"role": "controller",
"profile": "prgs-controller",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
{
"session_id": "prgs-author-10",
"role": "author",
"profile": "prgs-author",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
{
"session_id": "prgs-reviewer-20",
"role": "reviewer",
"profile": "prgs-reviewer",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
]
policy = rc.RESTART_CLASS_POLICIES[RestartClass.ROLE_RUNTIME_RESTART]
report = rc.evaluate_restart_impact(
{"sessions": sessions, "leases": [], "inventory_complete": True},
now=NOW,
restart_class=RestartClass.ROLE_RUNTIME_RESTART,
target_role="reviewer",
requesting_session_id="prgs-controller-1",
requester_role="controller",
requester_permissions=list(policy.request_roles),
controller_approved=True,
)
self.assertTrue(report.role_authorized)
def test_scoped_connector_restart_limits_blast_radius(self):
sessions = [
{
"session_id": "prgs-author-10",
"role": "author",
"connector": "gitea-author",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
{
"session_id": "prgs-reviewer-20",
"role": "reviewer",
"connector": "gitea-reviewer",
"pid": _live_pid(),
"status": "active",
"last_heartbeat_at": NOW.isoformat(),
},
]
policy = rc.RESTART_CLASS_POLICIES[RestartClass.CONNECTOR_RESTART]
report = rc.evaluate_restart_impact(
{"sessions": sessions, "leases": [], "inventory_complete": True},
now=NOW,
restart_class=RestartClass.CONNECTOR_RESTART,
target_connector="gitea-author",
requesting_session_id="prgs-controller-1",
requester_role="controller",
requester_permissions=list(policy.request_roles),
controller_approved=True,
)
self.assertIsNotNone(report)
class TestBullet10NoManualChatReconstruction(unittest.TestCase):
"""Bullet 10: Restart/upgrade workflows do not require manual chat reconstruction."""
def test_end_to_end_restart_reconcile_handoff_proof(self):
inv = _clean_inventory()
proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_LOG_ONLY)
proof_dict = proof.as_dict()
self.assertEqual(proof_dict["overall_status"], prr.STATUS_COMPLETE)
self.assertFalse(proof_dict["mutation_hold"])
self.assertTrue(proof_dict["note"])
self.assertIn("links", proof_dict)
self.assertEqual(proof_dict["links"]["umbrella"], 655)
if __name__ == "__main__":
unittest.main()
-465
View File
@@ -1,465 +0,0 @@
"""Unit tests for Phase 3 Notifications and Human-Attention Console (#648)."""
from __future__ import annotations
import pytest
from starlette.testclient import TestClient
from webui.app import create_app
from webui.notifications import (
ATTENTION_HUMAN_REQUIRED,
ATTENTION_OPERATOR,
ATTENTION_ROUTINE,
CATEGORY_AUTH,
CATEGORY_BLOCKER,
CATEGORY_LEASE,
CATEGORY_SYSTEM,
CATEGORY_VALIDATION,
CATEGORY_WORKFLOW,
NotificationItem,
NotificationSnapshot,
classify_attention_event,
load_notifications_snapshot,
snapshot_to_dict,
)
from webui.notification_views import render_notifications_page
from webui.project_registry import load_registry
from webui.queue_loader import QueueItem, QueueSnapshot
from webui.lease_loader import CollisionWarning, LeaseSnapshot
from webui.system_health import DependencyProbe, SystemHealthSnapshot, VersionInfo, StaleRuntime
def test_classify_attention_event_rules():
# 1. Critical escalation boundaries -> human-required
att_cls, req_human = classify_attention_event(
CATEGORY_AUTH, "Auth error", "Unauthorized access attempt", is_auth_failure=True
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM, "Hard stop", "Hard stop triggered", is_hard_stop=True
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
att_cls, req_human = classify_attention_event(
CATEGORY_VALIDATION, "Validation Error", "Report validation failed", is_validation_failure=True
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
# 2. Operational issues -> operator
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER, "PR Blocked", "Merge conflict detected", is_blocker=True
)
assert att_cls == ATTENTION_OPERATOR
assert req_human is False
att_cls, req_human = classify_attention_event(
CATEGORY_LEASE, "Lease Expired", "Session lease expired", is_stale=True
)
assert att_cls == ATTENTION_OPERATOR
assert req_human is False
# 3. Routine workflow transitions -> routine
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW, "PR Active", "PR in review"
)
assert att_cls == ATTENTION_ROUTINE
assert req_human is False
def test_notification_snapshot_aggregation():
reg = load_registry()
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
mock_queue = QueueSnapshot(
project_id=proj_id,
repo_label="org/repo",
prs=(
QueueItem(
number=101,
title="Blocked PR",
badges=("blocked",),
extra={},
),
QueueItem(
number=102,
title="Normal PR",
badges=("in-review",),
extra={},
),
),
issues=(),
pr_pagination=None,
issue_pagination=None,
)
mock_leases = LeaseSnapshot(
project_id=proj_id,
repo_label="org/repo",
issue_lock=None,
claim_inventory={},
reviewer_leases=(
{
"pr_number": 101,
"status": "expired",
"is_expired": True,
},
),
duplicate_prs=(
CollisionWarning(
kind="duplicate_pr",
message="Multiple open PRs for issue #101",
issue_number=101,
pr_numbers=(101, 103),
),
),
duplicate_branches=(),
collision_history=(),
fetch_error=None,
)
mock_version = VersionInfo(
git_sha="abc1234",
git_describe="v1.0.0",
control_plane_schema_version=1,
python_version="3.11",
known=True,
)
mock_stale = StaleRuntime(
daemon_head="abc1234",
checkout_head="abc1234",
remote_head="abc1234",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
mock_health = SystemHealthSnapshot(
status="degraded",
ready=False,
readiness_complete=True,
readiness_reasons=("Auth failure",),
service="webui",
mode="test",
version=mock_version,
started_at="2026-07-25T00:00:00Z",
uptime_seconds=100.0,
timestamp="2026-07-25T00:00:00Z",
deep_probes_requested=True,
dependencies=(
DependencyProbe(
name="auth_service",
kind="auth",
status="unauthorized",
detail="Token expired",
required=True,
),
),
mcp_namespaces=(),
stale_runtime=mock_stale,
probe_errors=(),
)
snapshot = load_notifications_snapshot(
proj_id,
load_queue=lambda _id: mock_queue,
load_leases=lambda **_kwargs: mock_leases,
load_health=lambda **_kwargs: mock_health,
)
assert snapshot.project_id == proj_id
assert snapshot.total_count == 5
assert snapshot.human_required_count >= 1 # auth probe failure
assert snapshot.operator_count >= 3 # blocked PR + expired lease + duplicate PR collision
assert snapshot.routine_count >= 1 # normal PR
# Inbox items should include operator and human-required items only
inbox_classes = {item.attention_class for item in snapshot.inbox_items}
assert ATTENTION_ROUTINE not in inbox_classes
assert ATTENTION_OPERATOR in inbox_classes
assert ATTENTION_HUMAN_REQUIRED in inbox_classes
def test_snapshot_to_dict_and_redaction():
item = NotificationItem(
id="notif-1",
attention_class=ATTENTION_HUMAN_REQUIRED,
category=CATEGORY_AUTH,
title="Auth Error",
summary="Failed auth header: Bearer secret_token_12345",
work_kind="system",
work_number=None,
project_id="test-proj",
repo_label="org/repo",
created_at="2026-07-25T16:00:00Z",
requires_human=True,
)
snap = NotificationSnapshot(
project_id="test-proj",
repo_label="org/repo",
items=(item,),
human_required_count=1,
operator_count=0,
routine_count=0,
total_count=1,
)
data = snapshot_to_dict(snap)
assert data["project_id"] == "test-proj"
assert data["human_required_count"] == 1
assert len(data["inbox_items"]) == 1
# Redaction test
summary = data["inbox_items"][0]["summary"]
assert "secret_token_12345" not in summary
assert "<redacted>" in summary or "Bearer" in summary
def test_notifications_html_views():
item = NotificationItem(
id="notif-1",
attention_class=ATTENTION_HUMAN_REQUIRED,
category=CATEGORY_AUTH,
title="Critical Auth Failure",
summary="Auth failure details",
work_kind="issue",
work_number=42,
project_id="test-proj",
repo_label="org/repo",
created_at="2026-07-25T16:00:00Z",
requires_human=True,
)
snap = NotificationSnapshot(
project_id="test-proj",
repo_label="org/repo",
items=(item,),
human_required_count=1,
operator_count=0,
routine_count=0,
total_count=1,
)
html = render_notifications_page(snap, filter_class="inbox")
assert "Notifications &amp; Attention Inbox" in html or "Notifications & Attention Inbox" in html
assert "Critical Auth Failure" in html
assert "HUMAN REQUIRED" in html
assert "Human Required" in html
def test_notifications_app_routes():
app = create_app()
client = TestClient(app)
# 1. HTML Route
res = client.get("/notifications")
assert res.status_code == 200
assert "Notifications" in res.text
assert "Attention Inbox" in res.text
# 2. API Route /api/v1/notifications
res_api = client.get("/api/v1/notifications")
assert res_api.status_code == 200
json_data = res_api.json()
assert "human_required_count" in json_data
assert "operator_count" in json_data
assert "routine_count" in json_data
assert "inbox_items" in json_data
# 3. Compatibility Alias /api/notifications
res_alias = client.get("/api/notifications")
assert res_alias.status_code == 200
assert res_alias.json()["project_id"] == json_data["project_id"]
def test_classify_ignores_human_authored_title_and_summary_keywords():
"""B1: keywords in human-authored titles must not escalate routine work (#905)."""
# Routine transition whose title/summary mention critical-boundary words
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
"record irrecoverable decision lock provenance",
"PR #999 'record irrecoverable decision lock provenance' is in routine state in-review.",
)
assert att_cls == ATTENTION_ROUTINE
assert req_human is False
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
"fix unauthorized token path",
"Issue #1 'fix unauthorized token path' state: claimed. hard stop docs only.",
)
assert att_cls == ATTENTION_ROUTINE
assert req_human is False
# Structured flags still escalate (machine-driven)
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM,
"anything",
"anything with hard stop in text",
is_hard_stop=True,
)
assert att_cls == ATTENTION_HUMAN_REQUIRED
assert req_human is True
def test_notification_ids_are_unique_across_probe_errors_and_collisions():
"""B2: published notification ids must be unique within a snapshot (#905)."""
reg = load_registry()
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
mock_queue = QueueSnapshot(
project_id=proj_id,
repo_label="org/repo",
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
)
mock_leases = LeaseSnapshot(
project_id=proj_id,
repo_label="org/repo",
issue_lock=None,
claim_inventory={},
reviewer_leases=(),
duplicate_prs=(
CollisionWarning(
kind="duplicate_pr",
message="Multiple open PRs for issue #10",
issue_number=10,
pr_numbers=(10, 11),
),
CollisionWarning(
kind="duplicate_branch",
message="Another collision without issue",
issue_number=None,
pr_numbers=(12, 13),
),
CollisionWarning(
kind="duplicate_pr",
message="Second issue collision",
issue_number=10,
pr_numbers=(14, 15),
),
),
duplicate_branches=(),
collision_history=(),
fetch_error=None,
)
mock_version = VersionInfo(
git_sha="abc1234",
git_describe="v1.0.0",
control_plane_schema_version=1,
python_version="3.11",
known=True,
)
mock_stale = StaleRuntime(
daemon_head="abc1234",
checkout_head="abc1234",
remote_head="abc1234",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
mock_health = SystemHealthSnapshot(
status="degraded",
ready=False,
readiness_complete=True,
readiness_reasons=(),
service="webui",
mode="test",
version=mock_version,
started_at="2026-07-25T00:00:00Z",
uptime_seconds=100.0,
timestamp="2026-07-25T00:00:00Z",
deep_probes_requested=True,
dependencies=(),
mcp_namespaces=(),
stale_runtime=mock_stale,
probe_errors=("error alpha", "error beta"),
)
snapshot = load_notifications_snapshot(
proj_id,
load_queue=lambda _id: mock_queue,
load_leases=lambda **_kwargs: mock_leases,
load_health=lambda **_kwargs: mock_health,
)
ids = [item.id for item in snapshot.items]
assert len(ids) == len(set(ids)), f"duplicate notification ids: {ids}"
assert any(i.startswith(f"notif-sys-err-{proj_id}-") for i in ids)
assert any(i.startswith("notif-collision-") for i in ids)
def test_probe_errors_do_not_set_fetch_error():
"""B3: probe_errors must not be reported as fetch_error (#905)."""
reg = load_registry()
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
mock_queue = QueueSnapshot(
project_id=proj_id,
repo_label="org/repo",
prs=(),
issues=(),
pr_pagination=None,
issue_pagination=None,
fetch_error=None,
)
mock_leases = LeaseSnapshot(
project_id=proj_id,
repo_label="org/repo",
issue_lock=None,
claim_inventory={},
reviewer_leases=(),
duplicate_prs=(),
duplicate_branches=(),
collision_history=(),
fetch_error=None,
)
mock_version = VersionInfo(
git_sha="abc1234",
git_describe="v1.0.0",
control_plane_schema_version=1,
python_version="3.11",
known=True,
)
mock_stale = StaleRuntime(
daemon_head="abc1234",
checkout_head="abc1234",
remote_head="abc1234",
stale=False,
determinable=True,
mutation_safe=True,
reasons=(),
)
mock_health = SystemHealthSnapshot(
status="degraded",
ready=False,
readiness_complete=True,
readiness_reasons=(),
service="webui",
mode="test",
version=mock_version,
started_at="2026-07-25T00:00:00Z",
uptime_seconds=100.0,
timestamp="2026-07-25T00:00:00Z",
deep_probes_requested=True,
dependencies=(),
mcp_namespaces=(),
stale_runtime=mock_stale,
probe_errors=("probe blew up",),
)
snapshot = load_notifications_snapshot(
proj_id,
load_queue=lambda _id: mock_queue,
load_leases=lambda **_kwargs: mock_leases,
load_health=lambda **_kwargs: mock_health,
)
assert snapshot.fetch_error is None
# probe errors still appear as items
assert any("probe blew up" in item.summary for item in snapshot.items)
-24
View File
@@ -80,11 +80,6 @@ from webui.system_health import (
snapshot_to_dict as system_health_to_dict,
)
from webui.system_health_views import render_system_health_page
from webui.notifications import (
load_notifications_snapshot,
snapshot_to_dict as notifications_snapshot_to_dict,
)
from webui.notification_views import render_notifications_page
from webui import request_service
from webui.request_views import render_requests_page
@@ -894,22 +889,6 @@ async def api_v1_analytics_ingest(request: Request) -> JSONResponse:
)
async def notifications_route(request: Request) -> HTMLResponse:
project_id = request.query_params.get("project_id")
attention_class = request.query_params.get("attention_class") or "inbox"
snap = load_notifications_snapshot(project_id)
html = render_notifications_page(
snap, filter_class=attention_class, filter_project=project_id
)
return HTMLResponse(html)
async def api_notifications(request: Request) -> JSONResponse:
project_id = request.query_params.get("project_id")
snap = load_notifications_snapshot(project_id)
data = notifications_snapshot_to_dict(snap)
return JSONResponse(data)
def _default_request_scope() -> dict[str, str]:
"""Resolve remote/org/repo from the project registry for request forms.
@@ -1041,9 +1020,6 @@ def create_app(*, bind_host: str | None = None) -> Starlette:
Route("/api/queue", api_queue, methods=["GET"]),
Route("/traffic", traffic, methods=["GET"]),
Route("/api/traffic", api_traffic, methods=["GET"]),
Route("/notifications", notifications_route, methods=["GET"]),
Route("/api/notifications", api_notifications, methods=["GET"]),
Route("/api/v1/notifications", api_notifications, methods=["GET"]),
Route("/projects", projects, methods=["GET"]),
Route("/projects/{project_id}", project_detail, methods=["GET"]),
Route("/api/projects", api_projects, methods=["GET"]),
-1
View File
@@ -46,7 +46,6 @@ NAV_GROUPS: tuple[NavGroup, ...] = (
NavItem("/queue", "Queue"),
NavItem("/leases", "Leases"),
NavItem("/actions", "Actions"),
NavItem("/notifications", "Notifications"),
NavItem("/requests", "Requests"),
)),
NavGroup("Runtime/Sessions", (
-158
View File
@@ -1,158 +0,0 @@
"""HTML rendering for Phase 3 Notifications and Human-Attention Console (#648)."""
from __future__ import annotations
from html import escape
from typing import Sequence
from webui.layout import render_page
from webui.notifications import (
ATTENTION_HUMAN_REQUIRED,
ATTENTION_OPERATOR,
ATTENTION_ROUTINE,
NotificationItem,
NotificationSnapshot,
)
def _render_attention_badge(attention_class: str) -> str:
cls = "badge"
if attention_class == ATTENTION_HUMAN_REQUIRED:
cls += " badge-blocked"
elif attention_class == ATTENTION_OPERATOR:
cls += " badge-claimed"
else:
cls += " muted"
return f'<span class="{cls}">{escape(attention_class)}</span>'
def _render_notification_row(item: NotificationItem) -> str:
category_label = escape(item.category.upper())
id_str = escape(item.id)
title_str = escape(item.title)
summary_str = escape(item.summary)
att_badge = _render_attention_badge(item.attention_class)
work_item_html = ""
if item.work_number and item.work_kind:
kind_label = escape(item.work_kind.upper())
num_str = f"#{item.work_number}"
link = item.deep_link or "#"
work_item_html = f'<a href="{escape(link)}"><code>{kind_label} {num_str}</code></a>'
requires_human_label = (
'<span class="badge badge-blocked" style="font-size:0.75rem;">HUMAN REQUIRED</span>'
if item.requires_human
else ""
)
return f"""<tr>
<td><code>{category_label}</code><br><span class="muted" style="font-size:0.75rem;">{id_str}</span></td>
<td>
<div><strong>{title_str}</strong> {att_badge} {requires_human_label}</div>
<div class="muted" style="font-size:0.85rem; margin-top:0.25rem;">{summary_str}</div>
</td>
<td>{work_item_html}</td>
<td><span class="muted" style="font-size:0.8rem;">{escape(item.created_at[:19])}</span></td>
</tr>"""
def _render_notifications_table(items: Sequence[NotificationItem], empty_message: str) -> str:
if not items:
return f'<p class="muted" style="padding:1rem 0;">{escape(empty_message)}</p>'
rows = "".join(_render_notification_row(item) for item in items)
return f"""<table class="registry">
<thead>
<tr>
<th style="width: 18%;">Category & ID</th>
<th style="width: 52%;">Title & Attention Summary</th>
<th style="width: 15%;">Work Item</th>
<th style="width: 15%;">Time</th>
</tr>
</thead>
<tbody>
{rows}
</tbody>
</table>"""
def render_notifications_page(
snapshot: NotificationSnapshot,
*,
filter_class: str = "inbox",
filter_project: str | None = None,
) -> str:
"""Render the notifications and attention inbox page."""
title = "Notifications & Attention Inbox"
err_html = ""
if snapshot.fetch_error:
err_html = f'<div class="stub" style="border-color:#e53e3e; background:#fff5f5; color:#c53030; margin-bottom:1rem;"><p><strong>Fetch Warning:</strong> {escape(snapshot.fetch_error)}</p></div>'
# Determine items to render based on filter_class
if filter_class == ATTENTION_HUMAN_REQUIRED:
display_items = snapshot.human_required_items
active_tab_title = "Human-Required Escalations"
elif filter_class == ATTENTION_OPERATOR:
display_items = snapshot.operator_items
active_tab_title = "Operator Inbox Items"
elif filter_class == ATTENTION_ROUTINE:
display_items = snapshot.routine_items
active_tab_title = "Routine Workflow Transitions"
elif filter_class == "all":
display_items = snapshot.items
active_tab_title = "All Events (including Routine)"
else: # "inbox" default
display_items = snapshot.inbox_items
active_tab_title = "Attention Inbox (Human + Operator)"
hr_cls = "badge-blocked" if snapshot.human_required_count > 0 else "muted"
op_cls = "badge-claimed" if snapshot.operator_count > 0 else "muted"
metrics_html = f"""<div style="display:flex; gap:1rem; margin-bottom:1.5rem;">
<div class="health-card" style="flex:1;">
<span class="muted" style="font-size:0.85rem;">Human Required</span>
<h2 style="margin:0.2rem 0;"><span class="badge {hr_cls}" style="font-size:1.4rem;">{snapshot.human_required_count}</span></h2>
<p class="muted" style="font-size:0.8rem; margin:0;">Critical escalation boundary</p>
</div>
<div class="health-card" style="flex:1;">
<span class="muted" style="font-size:0.85rem;">Operator Inbox</span>
<h2 style="margin:0.2rem 0;"><span class="badge {op_cls}" style="font-size:1.4rem;">{snapshot.operator_count}</span></h2>
<p class="muted" style="font-size:0.8rem; margin:0;">Operational items needing review</p>
</div>
<div class="health-card" style="flex:1;">
<span class="muted" style="font-size:0.85rem;">Routine Transitions</span>
<h2 style="margin:0.2rem 0;"><span class="badge muted" style="font-size:1.4rem;">{snapshot.routine_count}</span></h2>
<p class="muted" style="font-size:0.8rem; margin:0;">Background transitions (filtered)</p>
</div>
</div>"""
# Filter navigation links
def _tab_link(target_class: str, label: str) -> str:
is_active = (filter_class == target_class)
style = "font-weight:bold; border-bottom:2px solid currentColor;" if is_active else "color:#4a5568;"
return f'<a href="/notifications?attention_class={target_class}" style="margin-right:1.25rem; text-decoration:none; padding-bottom:0.25rem; {style}">{label}</a>'
tabs_html = f"""<div style="margin-bottom:1.25rem; border-bottom:1px solid #e2e8f0; padding-bottom:0.5rem;">
{_tab_link("inbox", f"Attention Inbox ({snapshot.human_required_count + snapshot.operator_count})")}
{_tab_link("human-required", f"Human Required ({snapshot.human_required_count})")}
{_tab_link("operator", f"Operator ({snapshot.operator_count})")}
{_tab_link("routine", f"Routine ({snapshot.routine_count})")}
{_tab_link("all", f"All Events ({snapshot.total_count})")}
</div>"""
table_html = _render_notifications_table(
display_items,
f"No items match attention filter '{filter_class}'.",
)
body = f"""<h2>{escape(title)}</h2>
<p class="muted">Phase 3 console surface for human-attention routing (#648). Routine workflow transitions are filtered by default to eliminate notification fatigue.</p>
{err_html}
{metrics_html}
{tabs_html}
<h3>{escape(active_tab_title)}</h3>
{table_html}"""
return render_page(title=title, body_html=body)
-486
View File
@@ -1,486 +0,0 @@
"""Notifications and human-attention routing module for Phase 3 web console (#648).
Defines attention classes, event classification rules, and inbox aggregation so
operators receive direct alerts only for human-required escalation boundaries
(#628) while routine workflow transitions remain available for pull-based review.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Callable
from webui import console_redaction
from webui.project_registry import load_registry
from webui.queue_loader import QueueSnapshot, load_queue_snapshot
from webui.lease_loader import LeaseSnapshot, load_lease_snapshot
from webui.system_health import SystemHealthSnapshot, load_system_health
# Attention class definitions (#628, #648)
ATTENTION_ROUTINE = "routine"
ATTENTION_OPERATOR = "operator"
ATTENTION_HUMAN_REQUIRED = "human-required"
ATTENTION_CLASSES = (
ATTENTION_ROUTINE,
ATTENTION_OPERATOR,
ATTENTION_HUMAN_REQUIRED,
)
# Notification categories
CATEGORY_AUTH = "auth"
CATEGORY_BLOCKER = "blocker"
CATEGORY_LEASE = "lease"
CATEGORY_VALIDATION = "validation"
CATEGORY_WORKFLOW = "workflow"
CATEGORY_SYSTEM = "system"
CATEGORIES = (
CATEGORY_AUTH,
CATEGORY_BLOCKER,
CATEGORY_LEASE,
CATEGORY_VALIDATION,
CATEGORY_WORKFLOW,
CATEGORY_SYSTEM,
)
@dataclass(frozen=True)
class NotificationItem:
"""A single notification or inbox event."""
id: str
attention_class: str # "routine", "operator", "human-required"
category: str # "auth", "blocker", "lease", "validation", etc.
title: str
summary: str
work_kind: str | None # "issue", "pr", "session", "system"
work_number: int | None
project_id: str
repo_label: str
created_at: str
deep_link: str | None = None
requires_human: bool = False
extra: dict[str, Any] = field(default_factory=dict)
def as_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"attention_class": self.attention_class,
"category": self.category,
"title": self.title,
"summary": console_redaction.redact_text(self.summary),
"work_kind": self.work_kind,
"work_number": self.work_number,
"project_id": self.project_id,
"repo_label": self.repo_label,
"created_at": self.created_at,
"deep_link": self.deep_link,
"requires_human": self.requires_human,
"extra": self.extra,
}
@dataclass(frozen=True)
class NotificationSnapshot:
"""Snapshot of notifications and attention inbox state."""
project_id: str
repo_label: str
items: tuple[NotificationItem, ...]
human_required_count: int
operator_count: int
routine_count: int
total_count: int
fetch_error: str | None = None
@property
def inbox_items(self) -> tuple[NotificationItem, ...]:
"""Items requiring operator or human attention (excluding routine)."""
return tuple(
item
for item in self.items
if item.attention_class in {ATTENTION_OPERATOR, ATTENTION_HUMAN_REQUIRED}
)
@property
def human_required_items(self) -> tuple[NotificationItem, ...]:
return tuple(
item for item in self.items if item.attention_class == ATTENTION_HUMAN_REQUIRED
)
@property
def operator_items(self) -> tuple[NotificationItem, ...]:
return tuple(
item for item in self.items if item.attention_class == ATTENTION_OPERATOR
)
@property
def routine_items(self) -> tuple[NotificationItem, ...]:
return tuple(
item for item in self.items if item.attention_class == ATTENTION_ROUTINE
)
def as_dict(self) -> dict[str, Any]:
return {
"project_id": self.project_id,
"repo_label": self.repo_label,
"human_required_count": self.human_required_count,
"operator_count": self.operator_count,
"routine_count": self.routine_count,
"total_count": self.total_count,
"fetch_error": self.fetch_error,
"inbox_items": [item.as_dict() for item in self.inbox_items],
"all_items": [item.as_dict() for item in self.items],
}
def classify_attention_event(
category: str,
title: str,
summary: str,
*,
is_hard_stop: bool = False,
is_auth_failure: bool = False,
is_irrecoverable: bool = False,
is_decision_lock: bool = False,
is_validation_failure: bool = False,
is_stale: bool = False,
is_blocker: bool = False,
) -> tuple[str, bool]:
"""Classify an event into an attention class and human requirement flag.
Rules (#628, #648):
1. Critical boundaries (hard stop, auth failure, irrecoverable state,
decision lock, validation failure) -> ATTENTION_HUMAN_REQUIRED (requires_human=True).
2. Operational queues (blocker, stale lease, unassigned ready work, queue collision)
-> ATTENTION_OPERATOR (requires_human=False).
3. Routine state transitions (clean progression, healthy heartbeats) -> ATTENTION_ROUTINE (requires_human=False).
Classification uses structured flags and category only. Human-authored
``title`` / ``summary`` text is never substring-matched for escalation
(PR #905 review B1) — callers that need text signals must set flags from
machine-generated status/detail fields before calling this function.
"""
del title, summary # kept for API stability; never used for classification
if (
is_hard_stop
or is_auth_failure
or is_irrecoverable
or is_decision_lock
or is_validation_failure
or category in {CATEGORY_AUTH, CATEGORY_VALIDATION}
):
return ATTENTION_HUMAN_REQUIRED, True
if is_stale or is_blocker or category in {CATEGORY_BLOCKER, CATEGORY_LEASE}:
return ATTENTION_OPERATOR, False
return ATTENTION_ROUTINE, False
def load_notifications_snapshot(
project_id: str | None = None,
*,
load_queue: Callable[..., QueueSnapshot] | None = None,
load_leases: Callable[..., LeaseSnapshot] | None = None,
load_health: Callable[..., SystemHealthSnapshot] | None = None,
) -> NotificationSnapshot:
"""Load and classify attention notifications across queue, leases, and system health."""
registry = load_registry()
project = None
if project_id:
for entry in registry.projects:
if entry.id == project_id:
project = entry
break
else:
project = registry.projects[0] if registry.projects else None
if project is None:
return NotificationSnapshot(
project_id=project_id or "",
repo_label="",
items=(),
human_required_count=0,
operator_count=0,
routine_count=0,
total_count=0,
fetch_error="project not found in registry",
)
queue_loader_fn = load_queue or load_queue_snapshot
lease_loader_fn = load_leases or load_lease_snapshot
health_loader_fn = load_health or load_system_health
try:
queue_snap = queue_loader_fn(project.id)
except TypeError:
queue_snap = queue_loader_fn(project_id=project.id)
try:
lease_snap = lease_loader_fn(project_id=project.id)
except TypeError:
lease_snap = lease_loader_fn(project.id)
try:
health_snap = health_loader_fn(project_id=project.id)
except TypeError:
try:
health_snap = health_loader_fn(project.id)
except TypeError:
health_snap = health_loader_fn()
items: list[NotificationItem] = []
now_iso = datetime.now(timezone.utc).isoformat()
# 1. System health alerts (highest priority)
for err_idx, probe_err in enumerate(getattr(health_snap, "probe_errors", ())):
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM,
"System Health Probe Error",
probe_err,
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-sys-err-{project.id}-{err_idx}",
attention_class=att_cls,
category=CATEGORY_SYSTEM,
title="System Health Error",
summary=f"System health error: {probe_err}",
work_kind="system",
work_number=None,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/system",
requires_human=req_human,
)
)
for probe in getattr(health_snap, "dependencies", ()):
if probe.status not in ("ok", "healthy"):
att_cls, req_human = classify_attention_event(
CATEGORY_SYSTEM,
f"Probe Failure: {probe.name}",
probe.detail or probe.status,
is_hard_stop=("stop" in probe.status or "fatal" in probe.status),
is_auth_failure=("auth" in probe.name.lower() or "unauthorized" in probe.status.lower()),
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-probe-{probe.name}",
attention_class=att_cls,
category=CATEGORY_AUTH if "auth" in probe.name.lower() else CATEGORY_SYSTEM,
title=f"Health Probe Alert: {probe.name}",
summary=f"Probe '{probe.name}' reported status '{probe.status}': {probe.detail}",
work_kind="system",
work_number=None,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/system",
requires_human=req_human,
)
)
# 2. Queue items (PRs and Issues)
for pr in queue_snap.prs:
if "blocked" in pr.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER,
f"PR #{pr.number} Blocked",
f"PR #{pr.number} '{pr.title}' is blocked or has merge conflicts.",
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-pr-block-{pr.number}",
attention_class=att_cls,
category=CATEGORY_BLOCKER,
title=f"Blocked PR #{pr.number}",
summary=f"PR #{pr.number} ({pr.title}) requires merge conflict resolution.",
work_kind="pr",
work_number=pr.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/traffic",
requires_human=req_human,
)
)
elif "stale" in pr.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
f"PR #{pr.number} Stale",
f"PR #{pr.number} '{pr.title}' has had no activity for over 14 days.",
is_stale=True,
)
items.append(
NotificationItem(
id=f"notif-pr-stale-{pr.number}",
attention_class=att_cls,
category=CATEGORY_WORKFLOW,
title=f"Stale PR #{pr.number}",
summary=f"PR #{pr.number} ({pr.title}) is stale.",
work_kind="pr",
work_number=pr.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/queue",
requires_human=req_human,
)
)
else:
# Routine PR transition
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
f"PR #{pr.number} Active",
f"PR #{pr.number} '{pr.title}' is in routine state {', '.join(pr.badges)}.",
)
items.append(
NotificationItem(
id=f"notif-pr-routine-{pr.number}",
attention_class=att_cls,
category=CATEGORY_WORKFLOW,
title=f"Routine PR #{pr.number}",
summary=f"PR #{pr.number} ({pr.title}) state: {', '.join(pr.badges)}.",
work_kind="pr",
work_number=pr.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/queue",
requires_human=req_human,
)
)
for issue in queue_snap.issues:
if "duplicate" in issue.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER,
f"Issue #{issue.number} Duplicate PRs",
f"Issue #{issue.number} has multiple linked PRs.",
is_blocker=True,
)
items.append(
NotificationItem(
id=f"notif-issue-dup-{issue.number}",
attention_class=att_cls,
category=CATEGORY_BLOCKER,
title=f"Duplicate PRs on Issue #{issue.number}",
summary=f"Issue #{issue.number} ({issue.title}) linked to multiple PRs.",
work_kind="issue",
work_number=issue.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/traffic",
requires_human=req_human,
)
)
elif "claimed" in issue.badges or "in-review" in issue.badges:
att_cls, req_human = classify_attention_event(
CATEGORY_WORKFLOW,
f"Issue #{issue.number} Active",
f"Issue #{issue.number} '{issue.title}' in state {', '.join(issue.badges)}.",
)
items.append(
NotificationItem(
id=f"notif-issue-routine-{issue.number}",
attention_class=att_cls,
category=CATEGORY_WORKFLOW,
title=f"Routine Issue #{issue.number}",
summary=f"Issue #{issue.number} ({issue.title}) state: {', '.join(issue.badges)}.",
work_kind="issue",
work_number=issue.number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link=f"/queue",
requires_human=req_human,
)
)
# 3. Leases / Collisions
for lease in lease_snap.reviewer_leases:
if lease.get("is_expired") or lease.get("status") == "expired":
pr_num = lease.get("pr_number") or lease.get("work_item_number")
att_cls, req_human = classify_attention_event(
CATEGORY_LEASE,
f"Reviewer Lease Expired for PR #{pr_num}",
f"Reviewer lease for PR #{pr_num} has expired.",
is_stale=True,
)
items.append(
NotificationItem(
id=f"notif-lease-exp-pr-{pr_num}",
attention_class=att_cls,
category=CATEGORY_LEASE,
title=f"Expired Reviewer Lease (PR #{pr_num})",
summary=f"Reviewer lease for PR #{pr_num} expired.",
work_kind="pr",
work_number=pr_num,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/leases",
requires_human=req_human,
)
)
for col_idx, collision in enumerate(lease_snap.duplicate_prs):
att_cls, req_human = classify_attention_event(
CATEGORY_BLOCKER,
f"Duplicate PR Collision ({collision.kind})",
collision.message,
is_blocker=True,
)
issue_part = collision.issue_number if collision.issue_number is not None else "none"
kind_part = (collision.kind or "unknown").replace(" ", "-")
items.append(
NotificationItem(
id=f"notif-collision-{kind_part}-{issue_part}-{col_idx}",
attention_class=att_cls,
category=CATEGORY_BLOCKER,
title=f"Collision Alert ({collision.kind})",
summary=collision.message,
work_kind="issue" if collision.issue_number else "pr",
work_number=collision.issue_number,
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
created_at=now_iso,
deep_link="/leases",
requires_human=req_human,
)
)
human_req_count = sum(1 for i in items if i.attention_class == ATTENTION_HUMAN_REQUIRED)
operator_count = sum(1 for i in items if i.attention_class == ATTENTION_OPERATOR)
routine_count = sum(1 for i in items if i.attention_class == ATTENTION_ROUTINE)
# Fetch errors are transport/load failures only — not probe results that
# already surface as first-class notification items (PR #905 review B3).
fetch_err = queue_snap.fetch_error or lease_snap.fetch_error
if isinstance(fetch_err, (tuple, list)):
fetch_err = "; ".join(fetch_err) if fetch_err else None
return NotificationSnapshot(
project_id=project.id,
repo_label=f"{project.gitea_owner}/{project.repo_name}",
items=tuple(items),
human_required_count=human_req_count,
operator_count=operator_count,
routine_count=routine_count,
total_count=len(items),
fetch_error=fetch_err,
)
def snapshot_to_dict(snapshot: NotificationSnapshot) -> dict[str, Any]:
"""JSON-serializable export for /api/v1/notifications."""
return snapshot.as_dict()