Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 b36eb201d4 feat: enforce capability stop terminal mode after reviewer denial (Issue #197)
Enter terminal mode when review_pr/merge_pr capability is denied. Block
reviewer queue tools (list_prs, eligibility checks) and add report-purity
validators for forbidden PR selection, fallback, and empty-queue claims.

Closes #197

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-05 20:30:15 -04:00
55 changed files with 4475 additions and 11273 deletions
-20
View File
@@ -1,20 +0,0 @@
## Description
[Summary of changes and issue number closed.]
Closes #[Issue Number]
## Checklist
- [ ] I have verified my identity matches the required role.
- [ ] No secrets, tokens, keychain IDs, or raw service URLs are committed.
- [ ] All tests pass for touched code.
- [ ] `git diff --check` is clean.
## Documentation and Wiki
- [ ] Does this change require a wiki update (workflows, tools, profiles, runbooks)?
- [ ] If yes, has `docs/wiki/` been updated accordingly?
- [ ] If wiki pages changed, plan the Gitea Wiki sync after merge (`scripts/sync-gitea-wiki.sh`, see Runbooks).
- [ ] Readiness gate (#224): the Gitea Wiki is populated and current for this repo — verify the repo **Wiki tab**, not `docs/wiki/`. If stale or empty, record the required sync as a follow-up before approval.
- [ ] If this PR closes a wiki-related issue: closure requires live Gitea Wiki proof links (Wiki Home plus page listing or wiki git log). Markdown in `docs/wiki/`, sync-helper code, or policy docs alone are not sufficient to close a wiki issue.
-8
View File
@@ -172,14 +172,6 @@ Recognized environment fields (see [`.env.example`](.env.example) for placeholde
| `GITEA_MCP_CONFIG` | Optional path to a JSON file defining multiple named runtime profiles. Unset ⇒ pure env behaviour. |
| `GITEA_MCP_PROFILE` | Name of the profile (from `GITEA_MCP_CONFIG`) to activate for this runtime. |
#### External MCP Control Plane servers
Jenkins and GlitchTip are separate MCP trust boundaries, not tools inside this
Gitea MCP runtime. Register them as `jenkins-mcp` and `glitchtip-mcp` in the
client that will use them, then reconnect or reload the client and verify the
expected tools are visible before claiming readiness. See
[`docs/mcp-client-registration.md`](docs/mcp-client-registration.md).
Notes:
- This provides **one token + one profile per process**. It does not implement
+4 -34
View File
@@ -53,28 +53,6 @@ def enter_from_capability_result(capability: dict) -> dict | None:
return dict(record)
def _is_reviewer_denial(capability: dict) -> bool:
task = (capability or {}).get("requested_task", "")
required_role = (capability or {}).get("required_role_kind")
return (
required_role == "reviewer"
or task in REVIEWER_CAPABILITY_TASKS
)
def sync_from_capability_result(capability: dict) -> dict | None:
"""Enter or clear terminal mode from a capability resolution (#238).
Reviewer denials activate terminal mode for the denied operation only.
A later allowed task route clears stale denial state so author read-only
tools (e.g. ``list_prs``) are not permanently blocked.
"""
if (capability or {}).get("stop_required") and _is_reviewer_denial(capability):
return enter_from_capability_result(capability)
clear()
return None
def enter_from_route_result(route: dict) -> dict | None:
"""Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
if (route or {}).get("route_result") != "wrong_role_stop":
@@ -112,13 +90,11 @@ def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
return True, []
name = (tool_name or "").strip().lower().removeprefix("gitea_")
if name in BLOCKED_QUEUE_TOOLS:
denied_task = (_session_terminal or {}).get("requested_task") or "unknown"
return False, [
TERMINAL_REPORT_HEADING,
f"Reviewer queue tool '{tool_name}' is blocked by the current "
f"capability denial for task '{denied_task}' (fail closed).",
"Resolve or route an allowed author task to clear stale denial "
"state, or relaunch a reviewer MCP namespace for reviewer work.",
f"Reviewer queue tool '{tool_name}' is blocked after "
"capability denial (fail closed).",
"Relaunch a reviewer MCP namespace to perform reviewer work.",
]
return True, []
@@ -172,14 +148,8 @@ def assess_capability_stop_report(
r"inventory empty",
re.I,
)
parsed_status = None
for line in text.splitlines():
if "pr_inventory_trust_gate.status:" in line.lower():
parsed_status = line.split(":", 1)[1].strip()
break
effective_status = trust_gate_status or parsed_status
if empty_queue_patterns.search(text):
if effective_status != "trusted_empty":
if trust_gate_status != "trusted_empty":
violations.append(
"empty-queue claim after capability stop without "
"pr_inventory_trust_gate.status == trusted_empty"
@@ -22,13 +22,13 @@ Strictly read-only, per ADR-0001:
tools — never one dual-credential server).
- **This server never holds Gitea write credentials.**
## 2. Boundary placement
## 2. Boundary placement (namespace pending)
These tools belong to the GlitchTip observability boundary of the MCP Control
Plane. The canonical MCP server name is `glitchtip-mcp`; client registration
and reload instructions live in
[`../mcp-client-registration.md`](../mcp-client-registration.md). Tool names
below use the `glitchtip_` prefix.
Plane`glitchtip-mcp` (ADR-0001's recommendation), `observability-mcp`, or
folded into `ops-mcp`. **ADR-0001 open owner decision #2 picks the name; this
design does not assume it.** Tool names below use the `glitchtip_` prefix for
readability and rename mechanically with the decision.
Fixed regardless of the name (per `tool-boundaries.md`,
`credential-isolation.md`):
@@ -160,13 +160,11 @@ Mocked-GlitchTip unit tests only, per `docs/developer-testing-guidelines.md`:
## 10. Implementation-readiness checklist
Ready to operate once:
Ready to implement once:
1. The MCP client registers the external server under the exact `glitchtip-mcp`
name and is reconnected/reloaded.
2. Tool discoverability proves the expected `glitchtip_*` read tools are
visible; if the server is enabled but exposes no usable tools, report
`SKIPPED` and stop.
1. ADR-0001 owner decision #2 (namespace/placement) is made — mechanical
rename of the `glitchtip_` prefix if needed.
2. ADR-0001 owner decision #1 (repo home) is made.
3. #76 profile schema exists (or a minimal `glitchtip-readonly` profile is
hand-rolled to the same rules).
4. A pinned GlitchTip version is chosen for API-subset testing (§3).
@@ -5,9 +5,7 @@
- **Related:** #77 (repo/branch/PR → job mapping, designed separately)
- **Date:** 2026-07-02
Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The read server boundary remains read-only; gated triggers live on the separate `jenkins-write-mcp` / `jenkins_mcp.write_server` boundary (see #56 / #152).
Client registration and reload instructions live in
[`../mcp-client-registration.md`](../mcp-client-registration.md).
Note on naming: This design used historical `jenkins-readonly` skill name in Gitea-Tools. Actual package/server is `jenkins-mcp` (see mcp-control-plane registration in #55). The server boundary now contains gated trigger (see #56), but read tools remain as designed.
## 1. Purpose and scope
@@ -18,9 +16,7 @@ detail (build URL, number, timing, result) to report or investigate.
Phase 1 is **primarily read-only**, per ADR-0001
([`adr-0001-mcp-control-plane-boundaries.md`](adr-0001-mcp-control-plane-boundaries.md)):
- Build triggers are outside this read-only surface and require the separate
`jenkins-write-mcp` boundary, a dedicated profile, exact confirmation, and
fail-closed mutation audit (landed in #4, boundary correction in #56 / #152).
- Build triggers are gated behind dedicated profile + exact confirmation (landed in #4, boundary correction in #56).
- **Excluded: deploy triggers.**
- **Excluded: parameterized job launches.**
- Excluded: job creation/deletion/config changes, queue manipulation, node
@@ -111,7 +107,7 @@ by #76):
`forbidden_operations: ["jenkins.build.trigger", "jenkins.deploy", "jenkins.job.configure"]`
as belt-and-braces even though no mutating tool exists.
- Missing URL/user/token/profile ⇒ **fail closed** with a clear message.
- Since every tool on `jenkins-mcp` is read-only, no confirmation gates are needed — but
- Since every tool is read-only, no confirmation gates are needed — but
identity (`jenkins_whoami`) must still work so workflows can prove which
Jenkins account they act as.
@@ -145,16 +141,12 @@ repo's conventions (`docs/developer-testing-guidelines.md`):
## 10. Implementation-readiness checklist
Ready to operate through `jenkins-mcp` once:
Ready to implement in `jenkins-mcp` once:
1. The MCP client registers the external server under the exact `jenkins-mcp`
name and is reconnected/reloaded.
2. Tool discoverability proves the expected `jenkins_*` read tools are visible;
if the server is enabled but exposes no usable tools, report `SKIPPED` and
stop.
3. #76 profile schema exists (or a minimal `jenkins-readonly` profile is
1. ADR-0001 owner decision #1 (where `jenkins-mcp` lives) is made.
2. #76 profile schema exists (or a minimal `jenkins-readonly` profile is
hand-rolled to the same rules).
4. #77 mapping design is accepted (or tools ship path-addressed only, mapping
3. #77 mapping design is accepted (or tools ship path-addressed only, mapping
deferred).
Explicitly **not** unlocked by this document: build triggers, deploys,
-11
View File
@@ -298,17 +298,6 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
This guarantees the active profile operations align with the actual Gitea authenticated user credential.
## Gitea MCP Runtime Isolation and Worktree Safety
To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
- **Startup Conflict Check:** The MCP server (`mcp_server.py`) acts as a conflict-free loader. On startup, it scans all Python files in the directory for unresolved git merge conflicts (`<<<<<<<`, `=======`, `>>>>>>>`). If any are found, it prints an `infra_stop` message and exits immediately.
- **Workflow Guard:** Before starting reviewer work, task routing checks if the MCP runtime source is mid-merge (by checking for `.git/MERGE_HEAD` or conflict markers). If dirty, it returns `infra_stop` (never `wrong_role_stop` or empty queue) to prevent unsafe mutations.
- **Recovery Instructions:** To recover from an `infra_stop` state:
1. Resolve all merge conflicts in the local repository or abort the merge (`git merge --abort` / `git rebase --abort`).
2. Restart the Gitea MCP server process.
3. Retry the task.
## Relationship to roadmap issues
This document defines the **model only**. Related work is tracked separately
-61
View File
@@ -30,11 +30,6 @@ audit logging). See [Related documents](#related-documents).
> the standard rules; operator prompts still control task-specific scope.
> See issue #129 for the skill registry design.
Jenkins and GlitchTip workflows use separate MCP servers, not this Gitea MCP
runtime. Register them as `jenkins-mcp` and `glitchtip-mcp`, reconnect or
reload the client, and verify visible tools before claiming either integration
is usable. See [`mcp-client-registration.md`](mcp-client-registration.md).
For cross-project use, copy the portable workflow skill at
[`../skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
It extracts the issue-first, isolated-worktree, no-self-review, profile-safety,
@@ -229,62 +224,6 @@ Legacy environment-only setups keep working unchanged until migrated.
Each runbook names the **profile role** it runs under, the steps, and a safe
prompt. Confirm the active profile first (`gitea_get_profile` / `gitea_whoami`).
## Work Selection Rule for LLMs
Before starting any issue or PR work, acquire or verify a work lease. Do not
begin coding, reviewing, fixing, branching, committing, pushing, commenting,
or creating a PR until you prove the target is not already being worked.
Required checks:
1. List open PRs.
2. Search for PRs linked to the target issue.
3. Search local and remote branches for the issue number.
4. Search registered worktrees for the issue branch.
5. Check dirty worktrees.
6. Check active leases or recent handoffs.
7. Check whether the issue was already completed by a merged PR.
If another active LLM/session owns the lease, stop. Allowed responses:
continue as the lease owner; review the existing PR if reviewer capability
allows; produce a handoff; request takeover after lease expiry; stop with
"work already claimed."
Never create a parallel branch or PR for the same issue unless the old branch
is proven abandoned and the takeover is recorded.
Gitea-Tools lease gates: `gitea_lock_issue` (fail-closed before author
mutations), `status:in-progress`, and claim comments. Full portable wording:
[`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
## Global LLM Worktree Rule
The main project checkout is a stable control checkout. It must stay on the
configured stable branch: `master`, `main`, or `dev`.
All LLM task work must happen inside the project's `branches/` directory.
Before any mutation, prove:
1. current project root
2. current working directory
3. current branch
4. stable branch for the main checkout
5. session-owned worktree path under `branches/`
If `cwd` is not inside `branches/`, stop. Do not edit, create, delete, format,
test-write, commit, merge, rebase, checkout task branches, resolve conflicts,
or run cleanup.
There are no exceptions for small fixes, docs, tests, cleanup, PR review fixes,
conflict resolution, or emergencies.
The main checkout may only be used for read-only inspection, fetching,
stable-branch update after merged PRs, creating `branches/` worktrees, or
explicit control-checkout repair.
Portable wording: [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md).
## Branch worktree isolation
All LLM implementation and review work happens in an isolated branch worktree
-101
View File
@@ -1,101 +0,0 @@
# MCP Client Registration for External Control Plane Servers
Issue #151 fixes the registration and naming contract for the Jenkins and
GlitchTip MCP servers that live outside this Gitea MCP runtime.
## Canonical Server Names
Use these exact MCP server names in clients:
| Server name | Boundary | Default capability |
|---|---|---|
| `jenkins-mcp` | Jenkins CI inspection (read) | Read-only build/job inspection |
| `jenkins-write-mcp` | Jenkins build trigger (write) | Gated `jenkins_trigger_build` only |
| `glitchtip-mcp` | GlitchTip observability inspection | Read-only issue/event inspection |
The write boundary (`jenkins-write-mcp`) is **not** registered by default (#152).
It exposes a single mutating tool and requires operator approval of a dedicated
trigger profile before any client config references it.
Historical names such as `jenkins-readonly` and `glitchtip-readonly` are
descriptive profile labels only. They are not the canonical MCP server names
unless an operator intentionally creates aliases and documents them.
## Registration Pattern
Register each external server as its own MCP entry. Do not add Jenkins or
GlitchTip credentials to the Gitea MCP server. Also, do not add Gitea write
credentials to the GlitchTip server.
```json
{
"mcpServers": {
"jenkins-mcp": {
"command": "/path/to/mcp-control-plane/venv/bin/python3",
"args": ["-m", "jenkins_mcp"],
"env": {
"JENKINS_MCP_CONFIG": "/path/to/mcp-control-plane/profiles.json",
"JENKINS_MCP_PROFILE": "jenkins-readonly"
}
},
"glitchtip-mcp": {
"command": "/path/to/mcp-control-plane/venv/bin/python3",
"args": ["-m", "glitchtip_mcp"],
"env": {
"GLITCHTIP_MCP_CONFIG": "/path/to/mcp-control-plane/profiles.json",
"GLITCHTIP_MCP_PROFILE": "glitchtip-readonly"
}
}
}
}
```
Client-specific wrappers may differ, but the server names, trust boundaries,
and profile separation must remain the same. After adding or changing either
entry, reconnect or reload the MCP client before claiming the tools are usable.
## Discoverability Validation
Before using either server in a task, prove the expected tools are visible in
the client. It is not enough for the config entry to exist.
Expected Jenkins read tools (`jenkins-mcp` only):
- `jenkins_whoami`
- `jenkins_list_jobs`
- `jenkins_latest_build`
- `jenkins_build_status`
- `jenkins_get_build`
`jenkins_trigger_build` must **not** appear on `jenkins-mcp`. When an operator
explicitly enables the write boundary, the only expected tool on
`jenkins-write-mcp` is `jenkins_trigger_build`.
Expected GlitchTip tools:
- `glitchtip_whoami`
- `glitchtip_list_projects`
- `glitchtip_list_unresolved`
- `glitchtip_get_issue`
- `glitchtip_recent_events`
- `glitchtip_search`
If a client reports the server as enabled but exposes no usable tools, report
`SKIPPED: server enabled but no usable tools visible`, then stop. Do not fall
back to shell commands, raw service APIs, or unrelated MCP servers.
## Boundary Rules
- `jenkins-mcp` read profiles must not expose build trigger tools.
- Build triggers live on the separate `jenkins-write-mcp` server
(`jenkins_mcp.write_server`), not on `jenkins-mcp`.
- Jenkins build triggers require a dedicated trigger profile with
`jenkins.build.trigger` allowed, exact confirmation
(`TRIGGER BUILD <job-path>`), and fail-closed mutation audit.
- Do not register `jenkins-write-mcp` until an operator approves a trigger
profile; no shipped profile carries trigger capability by default.
- `glitchtip-mcp` remains read-only. It must not file or mutate Gitea issues.
- GlitchTip-to-Gitea filing is a separate orchestrator that composes GlitchTip
read tools with Gitea issue-write tools.
- Gitea credentials never enter Jenkins or GlitchTip runtimes.
- Jenkins and GlitchTip credentials never enter the Gitea MCP runtime.
+2 -7
View File
@@ -21,10 +21,5 @@ Note on naming: Historical design docs used `jenkins-readonly` / `glitchtip-read
## 5. Mutation Gating
Any mutating action (e.g., Gitea issue creation from GlitchTip, or Jenkins builds) must be explicitly allowed by the execution profile.
- **Jenkins build triggers** are gated on a separate write boundary
(`jenkins-write-mcp` / `jenkins_mcp.write_server`), not on the read-only
`jenkins-mcp` surface. Triggers require a dedicated profile with
`jenkins.build.trigger`, exact confirmation, and fail-closed mutation audit.
No default profile carries trigger capability (#152 / mcp-control-plane #56).
- **GlitchTip to Gitea issue filing** is a library-only orchestrator in
mcp-control-plane (not on `glitchtip-mcp`). See #153 / mcp-control-plane #57.
- **Jenkins build triggers** exist in jenkins-mcp (landed #4) but require dedicated profile/identity and exact confirmation; not on standard reader profiles. See #56 for boundary correction.
- **GlitchTip to Gitea issue filing** is documented as a gated, orchestrated workflow (not in glitchtip-mcp), currently partial (mocked, dedup not wired, audit missing). See #57.
-15
View File
@@ -1,15 +0,0 @@
# Project History
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
## 2026-07-06
- **Wiki bootstrap (#224)** — Added repo-tracked `docs/wiki/` (10 pages), `scripts/sync-gitea-wiki.sh`, PR template gate, and sync safety tests for Gitea-Tools publication.
## Prior milestones
- Gated review/merge path (#16), task capability resolver (#69), issue-write tool gates.
- Canonical JSON execution profiles (#19) and thin MCP launchers.
- Role session router (#206) and review decision lock (#211).
-35
View File
@@ -1,35 +0,0 @@
# Gitea-Tools Project Wiki
Welcome to the version-controlled wiki for the Gitea-Tools MCP server and CLI tooling.
## Navigation
- [Home](Home.md)
- [Operator Guide](Operator-Guide.md)
- [Repositories Map](Repositories.md)
- [Identity and Profiles](Identity-and-Profiles.md)
- [Workflow Model](Workflow.md)
- [Safety and Gates](Safety-and-Gates.md)
- [MCP Tools Reference](MCP-Tools.md)
- [Operator Runbooks](Runbooks.md)
- [Open Decisions](Open-Decisions.md)
- [Project History](History.md)
## Gitea Wiki mirror
These repo-tracked pages are the **source of truth**. The Gitea native Wiki
for this repository is a read-only convenience mirror generated from this
directory with `scripts/sync-gitea-wiki.sh` (dry-run by default; see
[Runbooks](Runbooks.md)). Never edit the Gitea Wiki directly — change the
pages here through a PR, then sync.
## Project overview
Gitea-Tools provides the Gitea MCP server, execution-profile configuration,
identity handling, and CLI helpers used across MCP Control Plane workflows.
It enforces author/reviewer separation, gated review/merge, task-capability
resolution, and fail-closed safety rails in code.
Canonical long-form docs also live under `docs/` in the repository (for example
`docs/gitea-execution-profiles.md` and `docs/llm-workflow-runbooks.md`). The
wiki summarizes operator-facing rules; the repo docs carry implementation detail.
-39
View File
@@ -1,39 +0,0 @@
# Identity and Profiles
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
The LLM is not the role — the **MCP execution profile** is the role. Profiles
bind an authenticated Gitea identity to an allowed operation set.
## Reference profiles (prgs)
### Author / implementer
- **Profile:** `prgs-author`
- **Typical identity:** `jcwalker3`
- **Allowed:** branch create/push, PR create, issue comment/create/close, repo commit, read.
- **Forbidden:** PR approve, merge, request_changes.
### Reviewer / merger
- **Profile:** `prgs-reviewer`
- **Typical identity:** `sysadmin`
- **Allowed:** PR review/approve/merge/request_changes, issue comment, read.
- **Forbidden:** branch push, PR create, repo commit.
## Configuration
Profiles are defined in the canonical JSON config (`GITEA_MCP_CONFIG`, typically
`~/.config/gitea-tools/profiles.json`). Launchers are thin: they set
`GITEA_MCP_PROFILE` and point at the config file. Credentials resolve from
keychain or env references — never inline in client configs.
See `docs/gitea-execution-profiles.md` in the repository for the full model.
## Profile switching
Use separate MCP server namespaces (`gitea-tools` author vs `gitea-reviewer`)
or distinct launcher entries. Runtime in-place profile switching is disabled by
default (fail closed).
-34
View File
@@ -1,34 +0,0 @@
# MCP Tools Reference
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
## Identity and capability
- `gitea_whoami` — authenticated user and active profile metadata.
- `gitea_get_profile` / `gitea_get_runtime_context` — allowed and forbidden operations.
- `gitea_resolve_task_capability` — required pre-flight for gated mutations.
- `gitea_route_task_session` — role/session router before task execution.
## Author tools
- `gitea_create_issue`, `gitea_create_issue_comment`, `gitea_close_issue`
- `gitea_mark_issue`, `gitea_set_issue_labels`, `gitea_lock_issue`
- `gitea_create_pr`, `gitea_edit_pr`, `gitea_commit_files`
- `gitea_delete_branch`
## Reviewer tools
- `gitea_check_pr_eligibility` — read-only eligibility check.
- `gitea_dry_run_pr_review` — validation-phase review mechanics.
- `gitea_mark_final_review_decision` — mark validation complete.
- `gitea_submit_pr_review` / `gitea_review_pr` — gated live review.
- `gitea_merge_pr` — gated merge (only merge path).
## Read tools
- `gitea_list_prs`, `gitea_view_pr`, `gitea_list_issues`, `gitea_view_issue`
- `gitea_get_file`, `gitea_list_labels`, `gitea_mirror_refs`
See the repository `README.md` for the full tool table and client setup.
-11
View File
@@ -1,11 +0,0 @@
# Open Decisions
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
Pending architectural and workflow decisions:
- **Operation-scoped role selection (#228)** — Move from launcher-profile-dependent routing to per-operation profile resolution.
- **Gitea-Tools wiki for dadeschools remote** — This bootstrap covers `prgs`; dadeschools instance wiki parity is undecided.
- **Server-side self-merge block** — Complement tool gates with Gitea branch protection where available.
-32
View File
@@ -1,32 +0,0 @@
# Operator Guide
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
Handbook for LLM operators and human developers using the Gitea-Tools MCP server.
## Key rules
1. **Verify identity first** — Run `gitea_whoami` and confirm the active profile before any mutation.
2. **Resolve task capability** — Call `gitea_resolve_task_capability` for the intended task before gated tools.
3. **One unit of work per session** — Implement one claimed issue *or* review/merge one PR; do not mix author and reviewer mutations in one session.
4. **No self-review / no self-merge** — The authenticated Gitea user must not approve or merge a PR they authored.
5. **Follow the gates** — Prompts express intent; MCP tools enforce safety. Never bypass gates via prompt instructions.
6. **Global LLM Worktree Rule** — Main checkout stays on `master`/`main`/`dev`; all mutations happen under `branches/`. Prove project root, `cwd`, branch, stable main-checkout branch, and session worktree path before editing. No exceptions.
## Supported Gitea instances
| Remote | Host | Default org/repo |
|--------|------|------------------|
| `dadeschools` | `gitea.dadeschools.net` | `Contractor / Timesheet` |
| `prgs` | `gitea.prgs.cc` | `Scaled-Tech-Consulting / Timesheet` |
Always pass `remote` explicitly on tool calls. The server default is `dadeschools`; forgetting `remote` on a `prgs` task hits the wrong host.
## New session checklist
1. `gitea_whoami` — confirm authenticated user and profile.
2. `gitea_get_runtime_context` — allowed/forbidden operations for this session.
3. `gitea_resolve_task_capability` — prove the session may perform the planned task.
4. For reviewer work: dry-run validation (`gitea_dry_run_pr_review`) before live review mutations.
-38
View File
@@ -1,38 +0,0 @@
# Repositories Map
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
Active MCP Control Plane repositories (authoritative inventory for wiki publication gate #224 / #87):
| Repository | Purpose | Wiki required |
|------------|---------|---------------|
| `Scaled-Tech-Consulting/Gitea-Tools` | Gitea MCP server, execution profiles, workflow tooling | Yes — live Gitea Wiki on Wiki tab |
| `Scaled-Tech-Consulting/mcp-control-plane` | Orchestrators, audit trails, multi-service controller workflows | Yes — live Gitea Wiki on Wiki tab |
Repo-tracked `docs/wiki/` is the source of truth; the Gitea Wiki is a mirror.
Wiki-related issues cannot be closed until the live Wiki is verified — see
[Safety and Gates](Safety-and-Gates.md) and [Runbooks](Runbooks.md).
### Gitea-Tools
- **Repository:** `Scaled-Tech-Consulting/Gitea-Tools`
- **Purpose:** Gitea MCP server, profile configuration, CLI scripts, safety gates.
- **Base branch:** `master`
### mcp-control-plane
- **Repository:** `Scaled-Tech-Consulting/mcp-control-plane`
- **Purpose:** Controller workflows, Jenkins/GlitchTip integrations, audit orchestration.
- **Base branch:** `master`
## Wiki publication status (#224 readiness gate)
| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
|---|---|---|---|
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home); 10 pages (History, Home, Identity-and-Profiles, MCP-Tools, Open-Decisions, Operator-Guide, Repositories, Runbooks, Safety-and-Gates, Workflow); wiki git log head `ef3dec2` |
Update this table whenever a wiki is published, re-synced, or found stale.
-40
View File
@@ -1,40 +0,0 @@
# Operator Runbooks
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
## PR review and merge
1. `gitea_resolve_task_capability(task="review_pr")`
2. `gitea_check_pr_eligibility` for review and merge actions.
3. Validate locally: tests, `py_compile`, `git diff --check`.
4. `gitea_mark_final_review_decision` → approve via `gitea_review_pr`.
5. `gitea_merge_pr` with pinned head SHA and `confirmation="MERGE PR <n>"`.
## Gitea Wiki sync
The Gitea Wiki mirrors `docs/wiki/` (source of truth). After merging wiki changes:
1. Preview (no network):
```bash
scripts/sync-gitea-wiki.sh
```
2. Operator-confirmed push:
```bash
GITEA_WIKI_SYNC_CONFIRM="SYNC WIKI Gitea-Tools" scripts/sync-gitea-wiki.sh --push
```
3. If clone fails on first run, bootstrap Home via Gitea API or UI, then re-run.
The script mirrors only `docs/wiki/*.md`, never deletes wiki pages, and never
prints credentials.
## Wiki Publication Readiness Gate (#224)
Actual Gitea Wiki publication is a **required repo readiness gate**.
1. **Closure prevention** — No wiki issue may close on markdown/helper work alone.
Closing requires live-Wiki proof: Wiki Home link plus page listing or wiki git log.
2. **Reviewer checklist** — PR template requires verifying the repo **Wiki tab**.
3. **Publication authority** — `--push` requires exact `GITEA_WIKI_SYNC_CONFIRM` phrase.
4. **Per-repo status** — Update [Repositories Map](Repositories.md) when published.
-19
View File
@@ -1,19 +0,0 @@
# Safety and Gates
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
*Prompts express intent; MCP tools enforce safety.*
## Primary gates
1. **Fail closed** — Unknown tasks, missing capability resolution, or profile mismatches block mutations.
2. **Task capability map**`gitea_resolve_task_capability` must precede gated issue/PR mutations.
3. **Issue lock**`gitea_lock_issue` required before author implementation on a tracked issue.
4. **Head SHA pinning** — Reviews and merges refuse when the PR head moved.
5. **Explicit merge confirmation**`gitea_merge_pr` requires `confirmation="MERGE PR <n>"`.
6. **No self-review / self-merge** — Authenticated user must differ from PR author for approve/merge.
7. **Review decision lock** — Live review mutations require validation-phase dry-run and `gitea_mark_final_review_decision`.
8. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
9. **Wiki publication (#224)**`docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).
-34
View File
@@ -1,34 +0,0 @@
# Workflow Model
## Navigation
- [Home](Home.md) | [Operator Guide](Operator-Guide.md) | [Repositories Map](Repositories.md) | [Identity and Profiles](Identity-and-Profiles.md) | [Workflow Model](Workflow.md) | [Safety and Gates](Safety-and-Gates.md) | [MCP Tools Reference](MCP-Tools.md) | [Operator Runbooks](Runbooks.md) | [Open Decisions](Open-Decisions.md) | [Project History](History.md)
## Step 1: Review queue first (reviewer profile)
1. `gitea_resolve_task_capability(task="review_pr")`
2. List open PRs; pick the oldest eligible PR (not self-authored, mergeable).
3. Pin head SHA; validate diff scope and run tests locally.
4. `gitea_mark_final_review_decision``gitea_review_pr` / `gitea_submit_pr_review`
5. `gitea_merge_pr` only with `confirmation="MERGE PR <n>"` and pinned head SHA.
## Step 2: Implement issues (author profile)
0. Work Selection Rule — verify a work lease before any mutations (open PRs,
issue-linked PRs, branches, worktrees, dirty worktrees, active leases/
handoffs, merged-PR completion). Stop if another session owns the lease.
0b. Global LLM Worktree Rule — main checkout on `master`/`main`/`dev` only;
mutate only from a `branches/` worktree after proving root, cwd, branch,
stable main-checkout branch, and session worktree path (no exceptions).
1. `gitea_resolve_task_capability` for the author task.
2. `gitea_lock_issue` before implementation mutations.
3. Claim with `gitea_mark_issue` / `status:in-progress` label.
4. Branch, implement, test, push, `gitea_create_pr`.
5. Release claim when done; never self-review the PR.
## Step 3: Operator follow-ups
Wiki publication, credential provisioning, and cross-repo mirror operations are
operator-confirmed actions — see [Runbooks](Runbooks.md).
Full Gitea-specific runbooks: `docs/llm-workflow-runbooks.md` in the repository.
+2 -10
View File
@@ -163,13 +163,12 @@ def audit_enabled():
def build_event(*, action, result, remote=None, server=None, repository=None,
issue_number=None, pr_number=None, profile_name=None,
audit_label=None, authenticated_username=None, target_branch=None,
head_sha=None, reason=None, request_metadata=None, now=None,
mcp_namespace=None, task_role=None, operation=None):
head_sha=None, reason=None, request_metadata=None, now=None):
"""Build a redacted, JSON-able audit record for a mutating action."""
ts = now or datetime.datetime.now(datetime.timezone.utc)
if isinstance(ts, datetime.datetime):
ts = ts.isoformat()
event = {
return {
"timestamp": ts,
"action": action,
"action_type": "mutating",
@@ -187,13 +186,6 @@ def build_event(*, action, result, remote=None, server=None, repository=None,
"reason": _redact_str(reason) if reason else reason,
"request_metadata": redact(request_metadata) if request_metadata is not None else None,
}
if mcp_namespace is not None:
event["mcp_namespace"] = mcp_namespace
if task_role is not None:
event["task_role"] = task_role
if operation is not None:
event["operation"] = operation
return event
def write_event(event, path=None):
+1 -9
View File
@@ -205,7 +205,7 @@ def selected_profile_name():
def is_runtime_switching_enabled(path=None):
"""Check if runtime profile switching is enabled in config."""
"""Check if runtime profile switching is explicitly enabled in config."""
try:
config = load_config(path)
except Exception:
@@ -213,18 +213,10 @@ def is_runtime_switching_enabled(path=None):
if not config:
return False
rules = config.get("rules") or {}
if rules.get("allow_runtime_switching") is False:
return False
if config.get("allow_runtime_switching") is False:
return False
if rules.get("allow_runtime_switching") is True:
return True
if config.get("allow_runtime_switching") is True:
return True
# Default to True if multiple profiles exist in the config
profiles = config.get("profiles") or {}
if len(profiles) > 1:
return True
return False
-5147
View File
File diff suppressed because it is too large Load Diff
-156
View File
@@ -1,156 +0,0 @@
"""Issue-lock worktree validation (#249).
Author issue locks must validate the caller's own scratch clone (or declared
worktree path), not the shared MCP server working directory. A clean scratch at
``master``/``main`` must remain lockable while an unrelated session leaves the
shared dev worktree dirty or on a feature branch.
"""
from __future__ import annotations
import os
import subprocess
from reviewer_worktree import parse_dirty_tracked_files
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
BASE_BRANCHES = frozenset({"master", "main"})
def resolve_author_worktree_path(
explicit: str | None,
project_root: str,
) -> str:
"""Resolve the author worktree path for lock/PR gates."""
path = (explicit or "").strip()
if not path:
path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
if not path:
path = project_root
return os.path.realpath(os.path.abspath(path))
def read_worktree_git_state(worktree_path: str) -> dict:
"""Read branch name and porcelain status from a git worktree."""
path = (worktree_path or "").strip()
if not path:
return {"current_branch": None, "porcelain_status": ""}
branch_res = subprocess.run(
["git", "-C", path, "branch", "--show-current"],
capture_output=True,
text=True,
check=False,
)
current_branch = (branch_res.stdout or "").strip() or None
status_res = subprocess.run(
["git", "-C", path, "status", "--porcelain"],
capture_output=True,
text=True,
check=False,
)
return {
"current_branch": current_branch,
"porcelain_status": status_res.stdout or "",
}
def assess_issue_lock_worktree(
*,
worktree_path: str,
current_branch: str | None,
porcelain_status: str,
base_branches: frozenset[str] | None = None,
) -> dict:
"""Fail closed when lock preconditions are not met on the declared worktree."""
bases = base_branches or BASE_BRANCHES
reasons: list[str] = []
path = (worktree_path or "").strip()
if not path:
reasons.append("worktree path not declared for issue lock; fail closed")
return _assessment(False, reasons, path, None, [])
branch = (current_branch or "").strip()
dirty_files = parse_dirty_tracked_files(porcelain_status)
if dirty_files:
reasons.append(
"tracked file edits exist before issue lock; "
"lock must precede implementation work"
)
if not branch:
reasons.append(
"current branch unknown (detached HEAD?); issue lock must be taken "
f"from base branch ({_base_list(bases)})"
)
elif branch not in bases:
reasons.append(
f"issue lock must be taken from base branch ({_base_list(bases)}), "
f"not '{branch}'"
)
proven = not reasons
return _assessment(proven, reasons, path, branch or None, dirty_files)
def format_issue_lock_worktree_error(assessment: dict) -> str:
"""Format a single fail-closed error for ``gitea_lock_issue``."""
reasons = list(assessment.get("reasons") or [])
if not reasons:
reasons = ["issue lock worktree validation failed"]
return "; ".join(reasons) + " (fail closed)"
def verify_pr_worktree_matches_lock(
locked_worktree_path: str | None,
declared_worktree_path: str | None,
project_root: str,
) -> dict:
"""PR creation must use the same worktree the lock was validated against."""
locked = (locked_worktree_path or "").strip()
if not locked:
return {"proven": True, "block": False, "reasons": []}
declared = resolve_author_worktree_path(declared_worktree_path, project_root)
locked_real = os.path.realpath(locked)
declared_real = os.path.realpath(declared)
if locked_real != declared_real:
return {
"proven": False,
"block": True,
"reasons": [
f"PR worktree '{declared_real}' does not match locked worktree "
f"'{locked_real}' (fail closed)"
],
"locked_worktree_path": locked_real,
"declared_worktree_path": declared_real,
}
return {
"proven": True,
"block": False,
"reasons": [],
"locked_worktree_path": locked_real,
"declared_worktree_path": declared_real,
}
def _base_list(bases: frozenset[str]) -> str:
return "/".join(sorted(bases))
def _assessment(
proven: bool,
reasons: list[str],
worktree_path: str,
current_branch: str | None,
dirty_files: list[str],
) -> dict:
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"worktree_path": worktree_path or None,
"current_branch": current_branch,
"dirty_files": dirty_files,
}
-259
View File
@@ -1,259 +0,0 @@
#!/usr/bin/env python3
"""MCP Discoverability Validation Tool for external servers (Issue #155)."""
import os
import sys
import json
import argparse
import subprocess
EXPECTED_JENKINS_TOOLS = {
"jenkins_whoami",
"jenkins_list_jobs",
"jenkins_latest_build",
"jenkins_build_status",
"jenkins_get_build",
}
EXPECTED_GLITCHTIP_TOOLS = {
"glitchtip_whoami",
"glitchtip_list_projects",
"glitchtip_list_unresolved",
"glitchtip_get_issue",
"glitchtip_recent_events",
"glitchtip_search",
}
RELOAD_INSTRUCTIONS = """
=== MCP CLIENT RELOAD/RECONNECT RUNBOOK ===
After registering or changing external MCP servers, reload your client to discover the new tools:
- Codex: Click 'Reload Developer Tools' or restart the editor.
- Gemini / Grok / ChatGPT Desktop: Restart the client or run the reload slash command if available.
- Claude Desktop: Use 'Developer -> Reload' or restart the app.
- General MCP Clients: Restart the process or reload the server config.
===========================================
"""
def parse_gitea_mcp_config(path):
if not path or not os.path.exists(path):
return {}
with open(path, "r", encoding="utf-8") as fh:
try:
return json.load(fh)
except Exception:
return {}
def read_json_rpc_response(proc, req_id):
import time
start_time = time.time()
while time.time() - start_time < 5.0:
line = proc.stdout.readline()
if not line:
break
try:
data = json.loads(line)
if data.get("id") == req_id:
return data
except json.JSONDecodeError:
continue
return None
def query_live_tools(command, args, env):
run_env = os.environ.copy()
if env:
run_env.update(env)
proc = subprocess.Popen(
[command] + args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=run_env,
text=True,
bufsize=1
)
tools = []
try:
# 1. Send initialize
init_req = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "mcp-discoverability-check", "version": "1.0.0"}
}
}
proc.stdin.write(json.dumps(init_req) + "\n")
proc.stdin.flush()
# Read init response
init_resp = read_json_rpc_response(proc, 1)
if init_resp:
# Send initialized notification
init_notif = {
"jsonrpc": "2.0",
"method": "notifications/initialized"
}
proc.stdin.write(json.dumps(init_notif) + "\n")
proc.stdin.flush()
# 2. Send tools/list
tools_req = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}
proc.stdin.write(json.dumps(tools_req) + "\n")
proc.stdin.flush()
tools_resp = read_json_rpc_response(proc, 2)
if tools_resp and "result" in tools_resp and "tools" in tools_resp["result"]:
for t in tools_resp["result"]["tools"]:
tools.append(t["name"])
except Exception as e:
print(f"Error querying live tools: {e}", file=sys.stderr)
finally:
proc.terminate()
try:
proc.wait(timeout=2)
except Exception:
proc.kill()
return set(tools)
def validate_mcp_client_config(client_config_path, gitea_config_path=None, live=False):
if not client_config_path or not os.path.exists(client_config_path):
print(f"SKIPPED: MCP client config not found at '{client_config_path}'", file=sys.stderr)
return True
with open(client_config_path, "r", encoding="utf-8") as fh:
try:
config_data = json.load(fh)
except Exception as e:
print(f"Error parsing client config: {e}", file=sys.stderr)
return False
mcp_servers = config_data.get("mcpServers", {})
# Check for stale server names
stale_names = {"jenkins-readonly", "glitchtip-readonly"}
for name in mcp_servers:
if name in stale_names:
print(f"ERROR: Stale server name '{name}' configured. Use canonical names 'jenkins-mcp' or 'glitchtip-mcp'.", file=sys.stderr)
return False
gitea_data = parse_gitea_mcp_config(gitea_config_path)
enabled_services = set()
contexts = gitea_data.get("contexts", {})
profile_name = os.environ.get("GITEA_MCP_PROFILE")
if profile_name and "profiles" in gitea_data:
profile = gitea_data["profiles"].get(profile_name)
if profile and "context" in profile:
ctx_name = profile["context"]
ctx = contexts.get(ctx_name, {})
if ctx.get("enabled"):
services = ctx.get("services", {})
for s_name, s_data in services.items():
if s_data.get("enabled"):
enabled_services.add(s_name)
else:
for ctx_name, ctx in contexts.items():
if ctx.get("enabled"):
services = ctx.get("services", {})
for s_name, s_data in services.items():
if s_data.get("enabled"):
enabled_services.add(s_name)
if not enabled_services:
print("No external services enabled in Gitea contexts. Discoverability check complete.", file=sys.stderr)
return True
success = True
for service in enabled_services:
canonical_name = f"{service}-mcp"
if canonical_name not in mcp_servers:
stale_match = f"{service}-readonly"
if stale_match in mcp_servers:
print(f"ERROR: Server '{canonical_name}' references stale name '{stale_match}' (fail closed).", file=sys.stderr)
success = False
continue
print(f"ERROR: Enabled service '{service}' is not registered under canonical name '{canonical_name}' in client config.", file=sys.stderr)
success = False
continue
server_conf = mcp_servers[canonical_name]
command = server_conf.get("command")
args = server_conf.get("args") or []
env = server_conf.get("env") or {}
if not command:
print(f"ERROR: Server '{canonical_name}' has no command configured.", file=sys.stderr)
success = False
continue
expected_module = f"{service}_mcp"
if "-m" not in args or expected_module not in args:
print(f"ERROR: Server '{canonical_name}' args do not point to expected module '{expected_module}' (args: {args}).", file=sys.stderr)
success = False
continue
profile_var = f"{service.upper()}_MCP_PROFILE"
config_var = f"{service.upper()}_MCP_CONFIG"
if profile_var not in env or config_var not in env:
print(f"ERROR: Server '{canonical_name}' env is missing required variables '{profile_var}' or '{config_var}'.", file=sys.stderr)
success = False
continue
if live:
tools = query_live_tools(command, args, env)
if not tools:
print("SKIPPED: server enabled but no usable tools visible", file=sys.stdout)
success = False
continue
expected = EXPECTED_JENKINS_TOOLS if service == "jenkins" else EXPECTED_GLITCHTIP_TOOLS
missing = expected - tools
if missing:
print(f"ERROR: Server '{canonical_name}' is missing expected tools: {', '.join(missing)}", file=sys.stderr)
success = False
else:
print(f"SUCCESS: Server '{canonical_name}' discoverability verified.", file=sys.stderr)
else:
print(f"SUCCESS: Server '{canonical_name}' static registration verified.", file=sys.stderr)
return success
def main():
parser = argparse.ArgumentParser(description="MCP client registration discoverability checks.")
parser.add_argument("--client-config", help="Path to MCP client config JSON file.")
parser.add_argument("--gitea-config", help="Path to Gitea MCP config JSON file.")
parser.add_argument("--live", action="store_true", help="Perform live stdio checks on configured servers.")
parser.add_argument("--runbook", action="store_true", help="Print reload/reconnect guide runbook instructions.")
args = parser.parse_args()
if args.runbook:
print(RELOAD_INSTRUCTIONS)
return 0
if not args.client_config:
print("ERROR: --client-config must be specified.", file=sys.stderr)
return 2
ok = validate_mcp_client_config(
client_config_path=args.client_config,
gitea_config_path=args.gitea_config,
live=args.live
)
if not ok:
print(RELOAD_INSTRUCTIONS, file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
+4323 -41
View File
File diff suppressed because it is too large Load Diff
+11 -1043
View File
File diff suppressed because it is too large Load Diff
-210
View File
@@ -1,210 +0,0 @@
"""Fail-closed reviewer worktree and local-git safety proofs (#233).
Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
local changes from another session. When the active worktree has dirty tracked
files outside the PR scope, the workflow must stop or switch to a disposable
scratch worktree (``scripts/worktree-review``).
Git command policy (#243): reviewers use an allowlist, not a blocklist.
Any ``git`` invocation that does not match ``_READONLY_REVIEWER_GIT`` is
forbidden including ``checkout HEAD --``, ``checkout .``, ``switch``,
and uncommon ``stash`` subcommands that older blocklists missed.
"""
from __future__ import annotations
import re
import shlex
# Read-only git operations reviewers may use for validation (#243 allowlist).
_READONLY_REVIEWER_GIT = re.compile(
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
r"worktree\s+list|worktree\s+add)",
re.IGNORECASE,
)
_GIT_INVOCATION = re.compile(r"\bgit\b", re.IGNORECASE)
def parse_dirty_tracked_files(porcelain: str) -> list[str]:
"""Return tracked paths with local modifications from ``git status --porcelain``.
Untracked entries (``??``) are ignored they do not block reviewer work
when a scratch worktree is used, and authors may have unrelated untracked
files without implying reviewer interference.
"""
paths: list[str] = []
for line in (porcelain or "").splitlines():
if not line or len(line) < 4:
continue
if line.startswith("??"):
continue
path = line[3:].strip()
if " -> " in path:
path = path.split(" -> ", 1)[1].strip()
if path:
paths.append(path)
return paths
def files_outside_pr_scope(
dirty_files: list[str] | None,
pr_scope_files: list[str] | None,
) -> list[str]:
"""Dirty tracked files not explained by the PR diff file set."""
dirty = [p for p in (dirty_files or []) if p]
scope = {p for p in (pr_scope_files or []) if p}
if not dirty:
return []
if not scope:
return list(dirty)
return [path for path in dirty if path not in scope]
def _is_git_command(command: str) -> bool:
return bool(_GIT_INVOCATION.search((command or "").strip()))
def is_readonly_reviewer_git_command(command: str) -> bool:
"""True when the command is an explicitly allowed read-only git operation."""
text = (command or "").strip()
if not text or not _is_git_command(text):
return False
return bool(_READONLY_REVIEWER_GIT.search(text))
def is_forbidden_reviewer_git_command(command: str) -> bool:
"""True when a git command is not on the reviewer readonly allowlist."""
text = (command or "").strip()
if not text or not _is_git_command(text):
return False
return not is_readonly_reviewer_git_command(text)
def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
"""Fail closed when reviewer shell history includes forbidden git mutations."""
forbidden = [
cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
]
if forbidden:
return {
"proven": False,
"block": True,
"forbidden_commands": forbidden,
"reasons": [
"reviewer workflow executed forbidden local git mutation: "
f"{cmd!r}"
for cmd in forbidden
],
"safe_next_action": (
"stop; report worktree interference; do not stash/reset/checkout "
"unrelated files — use scripts/worktree-review instead"
),
}
return {
"proven": True,
"block": False,
"forbidden_commands": [],
"reasons": [],
"safe_next_action": "proceed",
}
def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
"""Evaluate reviewer worktree safety before checkout/diff/validation/review.
*proof* keys:
- ``worktree_path`` (required)
- ``porcelain_status`` or ``dirty_files``
- ``pr_scope_files`` (paths in the PR diff)
- ``scratch_used`` (bool)
- ``scratch_path`` (when scratch_used)
- ``git_commands`` (shell commands executed this session)
- ``unrelated_mutations_claimed`` (bool) stash/reset/drop reported
"""
proof = dict(proof or {})
reasons: list[str] = []
worktree_path = (proof.get("worktree_path") or "").strip()
if not worktree_path:
reasons.append("reviewer worktree path not reported; fail closed")
if proof.get("dirty_files") is not None:
dirty_files = list(proof.get("dirty_files") or [])
else:
dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")
pr_scope = list(proof.get("pr_scope_files") or [])
unrelated = files_outside_pr_scope(dirty_files, pr_scope)
scratch_used = bool(proof.get("scratch_used"))
scratch_path = (proof.get("scratch_path") or "").strip()
is_dirty = bool(dirty_files)
unrelated_dirty = bool(unrelated)
if unrelated_dirty and not scratch_used:
reasons.append(
"worktree has dirty tracked files outside PR scope "
f"({', '.join(unrelated)}); stop or use a scratch worktree"
)
if scratch_used and not scratch_path:
reasons.append(
"scratch worktree was used but scratch_path was not reported"
)
if proof.get("unrelated_mutations_claimed"):
reasons.append(
"reviewer reported stash/reset/checkout cleanup of unrelated "
"local changes; this is forbidden"
)
command_assessment = assess_reviewer_git_command_log(
list(proof.get("git_commands") or [])
)
if command_assessment["block"]:
reasons.extend(command_assessment["reasons"])
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"worktree_path": worktree_path or None,
"is_dirty": is_dirty,
"dirty_files": dirty_files,
"unrelated_dirty_files": unrelated,
"scratch_used": scratch_used,
"scratch_path": scratch_path or None,
"unrelated_mutations_avoided": not bool(
proof.get("unrelated_mutations_claimed")
or command_assessment.get("forbidden_commands")
),
"safe_next_action": (
"proceed"
if proven
else command_assessment.get("safe_next_action")
or "stop; use scripts/worktree-review or report dirty worktree"
),
"forbidden_commands": command_assessment.get("forbidden_commands", []),
}
def assess_author_worktree_continuity(proof: dict | None) -> dict:
"""Authors may keep dirty feature worktrees; reviewers may not manipulate them.
This helper only proves the task role is author when dirty unrelated files
exist it does not grant reviewers an exception.
"""
proof = dict(proof or {})
role = (proof.get("task_role") or "").strip().lower()
dirty_files = list(proof.get("dirty_files") or [])
if role == "author" and dirty_files:
return {
"allowed": True,
"reasons": [
"author task may continue with dirty tracked files in its own "
"worktree; reviewer interference rules do not apply"
],
}
if role == "reviewer" and dirty_files:
return assess_reviewer_worktree_proof(proof)
return {"allowed": True, "reasons": []}
-89
View File
@@ -1,89 +0,0 @@
"""Block author mutations from reviewer-bound MCP namespaces (#209).
Every mutation must prove that the active profile role, inferred MCP
namespace, and declared task role align. Reviewer sessions cannot silently
perform author-side work (especially PR creation).
"""
from __future__ import annotations
import gitea_config
import role_session_router
def infer_mcp_namespace(profile_name: str | None) -> str:
"""Map a runtime profile name to its MCP namespace label."""
lower = (profile_name or "").strip().lower()
if "reviewer" in lower and "author" not in lower:
return "gitea-reviewer"
if "author" in lower:
return "gitea-author"
return profile_name or "gitea-default"
def derive_role_kind(allowed, forbidden=()) -> str:
"""Classify the active profile the same way as ``mcp_server._role_kind``."""
def can(op):
return gitea_config.check_operation(op, allowed, forbidden)[0]
review = can("gitea.pr.approve") or can("gitea.pr.merge")
author = can("gitea.pr.create") or can("gitea.branch.push")
if review and author:
return "mixed"
if review:
return "reviewer"
if author:
return "author"
return "limited"
def check_author_mutation_namespace(
mutation_task: str,
profile: dict,
) -> tuple[bool, list[str]]:
"""Return (allowed, reasons). Fail closed on reviewer/author mismatch."""
required_role = role_session_router.required_role_for_task(mutation_task)
if required_role != "author":
return True, []
allowed = profile.get("allowed_operations") or []
forbidden = profile.get("forbidden_operations") or []
active_role = derive_role_kind(allowed, forbidden)
profile_name = profile.get("profile_name") or ""
namespace = infer_mcp_namespace(profile_name)
if mutation_task == "create_pr":
if active_role == "reviewer" or namespace == "gitea-reviewer":
return False, [
"author mutation 'create_pr' blocked in reviewer MCP namespace "
f"({namespace}); launch gitea-author",
]
if active_role == "reviewer" or namespace == "gitea-reviewer":
if mutation_task == "create_issue":
ok, _ = gitea_config.check_operation(
"gitea.issue.create", allowed, forbidden)
if ok:
return True, []
return False, [
f"author mutation '{mutation_task}' blocked: active session is "
f"reviewer-bound ({profile_name} / {namespace})",
]
return True, []
def mutation_audit_context(mutation_task: str, profile: dict, *,
remote=None, repository=None) -> dict:
"""Structured mutation metadata for audit records (#209)."""
allowed = profile.get("allowed_operations") or []
forbidden = profile.get("forbidden_operations") or []
return {
"mcp_namespace": infer_mcp_namespace(profile.get("profile_name")),
"profile_name": profile.get("profile_name"),
"task_role": role_session_router.required_role_for_task(mutation_task),
"operation": mutation_task,
"remote": remote,
"repository": repository,
}
+4 -139
View File
@@ -5,53 +5,12 @@ returns a route result before any downstream mutation tools run.
"""
from __future__ import annotations
import os
ROUTE_ALLOWED = "allowed_current_session"
ROUTE_WRONG_ROLE = "wrong_role_stop"
ROUTE_TO_AUTHOR = "route_to_author_session"
ROUTE_TO_REVIEWER = "route_to_reviewer_session"
ROUTE_AMBIGUOUS = "ambiguous_task_stop"
ROUTE_INFRA_STOP = "infra_stop"
_CONFLICT_HEAD = b"<" * 7 + b" "
_CONFLICT_TAIL = b">" * 7 + b" "
_CONFLICT_SEPARATOR = b"=" * 7
def python_bytes_have_conflict_markers(content: bytes) -> bool:
"""Return True when *content* contains git merge-conflict marker lines."""
for line in content.splitlines():
stripped = line.rstrip(b"\r\n")
if stripped.startswith(_CONFLICT_HEAD):
return True
if stripped.startswith(_CONFLICT_TAIL):
return True
if stripped == _CONFLICT_SEPARATOR:
return True
return False
def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
"""Skip venv/git/cache and sibling worktrees under orchestration checkout.
When *project_root* is itself a worktree inside ``branches/``, still scan
that tree do not treat the ``branches`` path segment as a skip signal.
"""
rel = os.path.relpath(walk_root, project_root)
if rel == ".":
return False
head = rel.split(os.sep, 1)[0]
if head in ("venv", ".git", ".pytest_cache"):
return True
if head == "branches":
nested = os.path.join(project_root, "branches")
if os.path.isdir(nested) and (
walk_root == nested or walk_root.startswith(nested + os.sep)
):
return True
return False
REVIEWER_TASKS = frozenset({
"review_pr",
@@ -134,23 +93,6 @@ def route_task_session(
_record_route(result)
return result
if required_role == "reviewer" and check_mid_merge():
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_INFRA_STOP,
"downstream_allowed": False,
"reasons": [
"infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
"Please resolve all conflicts manually, finish/abort the merge, restart the MCP server, and retry."
],
"message": "infra_stop: Unresolved merge conflict or mid-merge state detected in MCP runtime source.",
}
_record_route(result)
return result
if allowed_in_current_session:
result = {
"task_type": task_type,
@@ -233,56 +175,11 @@ def _record_route(result: dict):
_session_last_route = dict(result)
def sync_route_from_capability(capability: dict) -> None:
"""Align sticky route state with operation-scoped capability resolution (#228)."""
capability = capability or {}
task = (capability.get("requested_task") or "").strip()
required_role = capability.get("required_role_kind")
if not task or not required_role:
return
if capability.get("allowed_in_current_session"):
_record_route({
"task_type": task,
"required_role": required_role,
"active_role": required_role,
"active_profile": capability.get("active_profile"),
"route_result": ROUTE_ALLOWED,
"downstream_allowed": True,
"reasons": [],
"message": (
f"Operation-scoped task '{task}' resolved for current session; "
"proceed."
),
})
return
if required_role == "reviewer" and capability.get("stop_required"):
_record_route({
"task_type": task,
"required_role": required_role,
"active_role": capability.get("required_role_kind"),
"active_profile": capability.get("active_profile"),
"route_result": ROUTE_WRONG_ROLE,
"downstream_allowed": False,
"reasons": [WRONG_ROLE_REVIEWER_MSG],
"message": WRONG_ROLE_REVIEWER_MSG,
})
def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]:
"""Block author-side fallback after a reviewer wrong_role_stop (#206).
An explicit operation-scoped author capability resolution for the same
*mutation_task* clears the sticky reviewer denial (#228).
"""
"""Block author-side fallback after a reviewer wrong_role_stop (#206)."""
last = _session_last_route
if not last:
return True, []
if (
last.get("route_result") == ROUTE_ALLOWED
and last.get("task_type") == mutation_task
and last.get("required_role") == "author"
):
return True, []
if last.get("route_result") != ROUTE_WRONG_ROLE:
return True, []
if last.get("required_role") != "reviewer":
@@ -291,40 +188,8 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
return False, [
WRONG_ROLE_REVIEWER_MSG,
"Author-side mutations are blocked after a reviewer-task "
"wrong_role_stop unless the operator explicitly resolves the "
"author task via gitea_resolve_task_capability.",
"wrong_role_stop unless the operator explicitly changes the "
"task and relaunches an author MCP session.",
f"Attempted fallback mutation: {mutation_task}",
]
return True, []
def first_conflict_marker_path(project_root: str | None = None) -> str | None:
"""Return the first .py path containing a git conflict marker, or None."""
root_dir = project_root or os.path.dirname(os.path.abspath(__file__))
for root, dirs, files in os.walk(root_dir):
if skip_python_scan_walk_root(root_dir, root):
continue
for file in files:
if not file.endswith(".py"):
continue
file_path = os.path.join(root, file)
try:
with open(file_path, "rb") as f:
if python_bytes_have_conflict_markers(f.read()):
return file_path
except OSError:
pass
return None
def check_mid_merge() -> bool:
"""Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
project_root = os.path.dirname(os.path.abspath(__file__))
git_dir = os.path.join(project_root, ".git")
if os.path.exists(git_dir):
if (os.path.exists(os.path.join(git_dir, "MERGE_HEAD"))
or os.path.exists(os.path.join(git_dir, "rebase-merge"))
or os.path.exists(os.path.join(git_dir, "rebase-apply"))):
return True
return first_conflict_marker_path(project_root) is not None
return True, []
-86
View File
@@ -1,86 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
# sync-gitea-wiki.sh — mirror the repo-tracked docs/wiki/ pages into the
# Gitea native wiki. docs/wiki/ remains the source of truth; the Gitea Wiki
# is a read convenience mirrored FROM it, never edited directly.
#
# scripts/sync-gitea-wiki.sh dry-run (default): print the plan
# scripts/sync-gitea-wiki.sh --push actually sync, ONLY with the exact
# confirmation below
#
# Safety contract:
# - Dry-run is the default and performs no network or repository-mutating
# operation (only a local read of the origin remote URL).
# - A push requires GITEA_WIKI_SYNC_CONFIRM to equal exactly
# "SYNC WIKI <repo-name>" (repo name derived from the origin remote).
# Anything else refuses before any clone happens.
# - The wiki remote is derived from the local origin remote; nothing is
# hardcoded here and no credential material is read, printed, or stored
# by this script — git's own configured auth is used as-is.
# - Only *.md pages from docs/wiki/ are mirrored; nothing else is touched
# and no page is deleted from the wiki by this script.
here="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
repo_root="$(cd "$here/.." && pwd)"
wiki_src="$repo_root/docs/wiki"
[ -d "$wiki_src" ] || { echo "error: $wiki_src not found" >&2; exit 1; }
if git -C "$repo_root" remote get-url origin >/dev/null 2>&1; then
origin_url="$(git -C "$repo_root" remote get-url origin)"
elif git -C "$repo_root" remote get-url prgs >/dev/null 2>&1; then
origin_url="$(git -C "$repo_root" remote get-url prgs)"
else
echo "error: configure an origin or prgs git remote" >&2
exit 1
fi
repo_name="$(basename "$origin_url" .git)"
wiki_remote="${origin_url%.git}.wiki.git"
expected_confirm="SYNC WIKI $repo_name"
pages=()
while IFS= read -r -d '' f; do
pages+=("$(basename "$f")")
done < <(find "$wiki_src" -maxdepth 1 -name '*.md' -print0 | sort -z)
mode="${1:-}"
if [ "$mode" != "--push" ]; then
echo "dry-run: would sync ${#pages[@]} pages from docs/wiki/ to the '$repo_name' Gitea Wiki:"
for p in "${pages[@]}"; do
echo " $p"
done
echo "dry-run: no git or network operation performed."
echo "To sync for real: GITEA_WIKI_SYNC_CONFIRM=\"$expected_confirm\" $0 --push"
exit 0
fi
if [ "${GITEA_WIKI_SYNC_CONFIRM:-}" != "$expected_confirm" ]; then
echo "refused: --push requires GITEA_WIKI_SYNC_CONFIRM to equal exactly:" >&2
echo " $expected_confirm" >&2
exit 1
fi
workdir="$(mktemp -d)"
trap 'rm -rf "$workdir"' EXIT
echo "cloning wiki repository for '$repo_name'"
if ! git clone --quiet -- "$wiki_remote" "$workdir/wiki" 2>/dev/null; then
echo "error: could not clone the wiki repository. If this repo's wiki has" >&2
echo "never been initialized, create its first page once in the Gitea UI" >&2
echo "(Wiki tab -> New Page), then re-run this script." >&2
exit 1
fi
cp "$wiki_src"/*.md "$workdir/wiki/"
if git -C "$workdir/wiki" status --porcelain | grep -q .; then
git -C "$workdir/wiki" add -A
git -C "$workdir/wiki" commit --quiet -m "docs: sync from repo docs/wiki (source of truth)"
echo "pushing wiki update for '$repo_name'"
git -C "$workdir/wiki" push --quiet
echo "done: ${#pages[@]} pages synced."
else
echo "done: wiki already up to date; nothing pushed."
fi
+6 -99
View File
@@ -53,73 +53,10 @@ Additional issue-first rules:
owner decision.** Do not create a new repository or a new tracker unless
explicitly approved by the owner.
## Work Selection Rule for LLMs
Before starting any issue or PR work, acquire or verify a work lease.
Do not begin coding, reviewing, fixing, branching, committing, pushing,
commenting, or creating a PR until you prove the target is not already being
worked.
Required checks:
1. List open PRs.
2. Search for PRs linked to the target issue.
3. Search local and remote branches for the issue number.
4. Search registered worktrees for the issue branch.
5. Check dirty worktrees.
6. Check active leases or recent handoffs.
7. Check whether the issue was already completed by a merged PR.
If another active LLM/session owns the lease, stop.
Allowed responses:
- continue as the lease owner,
- review the existing PR if reviewer capability allows,
- produce a handoff,
- request takeover after lease expiry,
- stop with "work already claimed."
Never create a parallel branch or PR for the same issue unless the old branch
is proven abandoned and the takeover is recorded.
For Gitea-Tools: `gitea_lock_issue` is the fail-closed lease gate before author
mutations; `status:in-progress` and claim comments are supporting lease signals.
Use `review_proofs.classify_issue_for_selection` when reporting fresh issue
selection (#188).
## Global LLM Worktree Rule
The main project checkout is a stable control checkout. It must stay on the
configured stable branch: `master`, `main`, or `dev`.
All LLM task work must happen inside the project's `branches/` directory.
Before any mutation, prove:
1. current project root
2. current working directory
3. current branch
4. stable branch for the main checkout
5. session-owned worktree path under `branches/`
If `cwd` is not inside `branches/`, stop. Do not edit, create, delete, format,
test-write, commit, merge, rebase, checkout task branches, resolve conflicts,
or run cleanup.
There are no exceptions for small fixes, docs, tests, cleanup, PR review fixes,
conflict resolution, or emergencies.
The main checkout may only be used for read-only inspection, fetching,
stable-branch update after merged PRs, creating `branches/` worktrees, or
explicit control-checkout repair.
## B. Isolated worktree rule
**Never implement or review in the main checkout** (Global LLM Worktree Rule).
The main checkout is for orchestration and status only (issue creation,
`git status`, creating worktrees) and must remain on the stable branch.
**Never implement or review in the main checkout.** The main checkout is for
orchestration and status only (issue creation, `git status`, creating worktrees).
- Each issue gets its own branch worktree under an ignored `branches/` directory.
- Review work uses a **separate** review worktree, never the author's folder.
@@ -205,24 +142,10 @@ Worktree folder = branch with `/` replaced by `-`
## E. Start-work workflow
0. Acquire or verify a work lease (Work Selection Rule) — complete all seven
checks before any claim, branch, or PR work.
0b. Global LLM Worktree Rule — prove project root, `cwd`, branch, main-checkout
stable branch, and session-owned `branches/` worktree path. If `cwd` is not
under `branches/`, stop before any mutation (no exceptions).
1. Verify the orchestration checkout (right repo, clean tree, on stable branch).
1. Verify the orchestration checkout (right repo, clean tree).
2. Fetch/prune: `git fetch <remote> --prune`.
3. Confirm local `master` equals remote `master` (`git rev-list --left-right --count <remote>/master...master``0 0`).
4. Create/claim the issue (§A).
4b. **Issue lock from your scratch clone (#249):** when using
`gitea_lock_issue`, pass `worktree_path` pointing at your own clean
scratch clone (or set `GITEA_AUTHOR_WORKTREE`). The lock gate validates
*that* path — clean tree on `master`/`main`, no tracked edits yet —
not the shared MCP/orchestration checkout. Another session's dirty
feature branch in the shared dev worktree must not block your lock.
Never stash, reset, or checkout files in the shared worktree to satisfy
the gate. Pass the same `worktree_path` to `gitea_create_pr` so the PR
gate matches the lock record.
5. Create the isolated worktree (§B) from latest remote `master`.
6. Implement the narrow scope only — no unrelated refactors or formatting churn.
7. Add/update focused tests when behavior changes.
@@ -293,10 +216,7 @@ Worktree folder = branch with `/` replaced by `-`
the other.
Both configured repos must be reported with state filter, pagination proof,
and open-PR count (`review_proofs.assess_inventory_completeness` and
`resolve_repos_from_user_reference`). Before inventory, reconcile the
operator-supplied PR backlog against the target repo
(`review_proofs.reconcile_queue_target`); never report `trusted_empty`
for one repo while ignoring contradictory supplied PR numbers in another.
`resolve_repos_from_user_reference`).
7. **Role-boundary proof (#175):** a reviewer queue task must not silently
become author implementation. If no eligible PR exists, stop with the
queue report. Do not claim issues, create branches, commit, push, or open
@@ -317,9 +237,6 @@ Worktree folder = branch with `/` replaced by `-`
validation completes, call `gitea_mark_final_review_decision`, then submit
exactly one live review via
`gitea_submit_pr_review(..., final_review_decision_ready=True)`.
After submitting, re-read `gitea_get_pr_review_feedback` and confirm the
verdict is visible (`approval_visible` true for APPROVE; PENDING drafts do
not count — #244). Do not merge until a visible APPROVED review exists.
Final reports must list exactly one review mutation
(`review_proofs.assess_review_mutation_final_report`) unless an
operator-approved correction flow was invoked and explained.
@@ -391,9 +308,7 @@ When in doubt, stop and surface the discrepancy; do not guess or work around a g
## I. Recovery patterns
- **Dirty worktree from another issue:** do not touch it. Start your issue in its
own new worktree; unrelated dirty work must not block you. For Gitea-Tools
author flows, lock the issue from your scratch clone (`worktree_path` on
`gitea_lock_issue`) — do not manipulate the shared dev checkout.
own new worktree; unrelated dirty work must not block you.
- **Local `master` ahead of remote unexpectedly:** do not push `master`. Confirm
the commits are preserved on a feature branch (local + remote) first, then
`git reset --hard <remote>/master` to realign. Never discard commits that are
@@ -426,13 +341,11 @@ Ready-to-copy templates live in [`templates/`](templates/):
## K. Controller Handoff (required, every task)
Every LLM task **must end with a `Controller Handoff`** (exact title) — whether the
Every LLM task **must end with a `Controller Handoff`** — whether the
task was implementation, review, merge, issue triage, documentation,
discussion-only, or blocked planning. It lets a controller LLM understand the
current state immediately, without rereading the conversation.
The section title must be exactly "Controller Handoff" (or "Controller Handoff Summary" for long form). Reports without it are downgraded (see review_proofs.assess_controller_handoff).
**The compact format is the default.** It is written for controller-LLM
readability, not as a full human status report. PR bodies still carry the
full review detail — the handoff never replaces PR documentation.
@@ -464,12 +377,6 @@ Role-specific fields (append to the compact block):
`Linked issue status:`, `Cleanup status:`
- author tasks: `Selected issue:`, `Claim/comment status:`,
`PR number opened:`, `No review/merge:` (explicit confirmation)
- continuation tasks (#188): `Continuation mode:`, `Existing PR:`,
`PR author:`, `Branch:`, `Old PR head:`, `New PR head:`,
`Session authored PR:`, `Why continuation allowed:` — issues with open
PRs are excluded from fresh selection unless operator explicitly requests
continuation (`review_proofs.classify_issue_for_selection`,
`assess_issue_selection_final_report`)
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
`Selected PR or reason none selected:`, `Inventory completeness:`
@@ -17,32 +17,9 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
configured repos were not checked. This is not a complete queue inventory."
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
if the other configured repo was not inventoried.
- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
`review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
sufficient proof.
- Empty-queue report wall (#198): if the final report claims "no open PRs",
"queue empty", or "nothing to review", it must include verbatim:
`pr_inventory_trust_gate.status`, trust-gate reasons, corroboration,
remote/owner/repo/state filter, and the inventory MCP profile. A recent merge
commit is not valid corroboration. Author-bound sessions must not present
reviewer queue inventory as a reviewer decision.
Rules (llm-project-workflow):
- Review in a SEPARATE detached review worktree, never the author's folder.
- Worktree safety (#233): before checkout, diff, validation, review, or merge,
report the starting worktree path and whether it was dirty. If unrelated
tracked files exist outside the PR scope, STOP or run
`scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
Scratch-clone validation is the norm; tests must not assume the shared
development worktree or a repo-local ``venv/`` (#245).
NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
or `git clean` to manage another session's dirty files.
- Final report must state: Worktree path, Worktree dirty (yes/no),
Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
files were modified, stashed, reset, or dropped.
- You must NOT be the PR author. If the authenticated user == PR author, stop.
A different LLM-Agent-SHA does NOT make you a different actor — only a
different authenticated Gitea user does (docs/llm-agent-sha.md).
@@ -13,21 +13,6 @@ Rules (llm-project-workflow):
- Do not self-review or self-merge.
Steps:
0. Work Selection Rule — before any claim, branch, or file edits, acquire or
verify a work lease. Required checks: list open PRs; search PRs linked to
the target issue; search local/remote branches for the issue number; search
registered worktrees for the issue branch; check dirty worktrees; check
active leases or recent handoffs; check whether a merged PR already
completed the issue. If another session owns the lease, stop (continue only
as lease owner, review the existing PR, hand off, request takeover after
expiry, or report "work already claimed"). Never open a parallel branch/PR
unless the old branch is proven abandoned and takeover is recorded.
0b. Global LLM Worktree Rule — before any mutation, prove and state: project
root; cwd; current branch; stable branch for the main checkout (master/main/dev);
session-owned worktree path under branches/. If cwd is not inside branches/,
STOP (no exceptions — not for docs, tests, small fixes, review fixes, conflicts,
or cleanup). Main checkout is control-only: read-only inspect, fetch, create
worktrees, stable-branch update after merge, explicit repair.
1. Identity Checklist: Before claiming work, verify and state:
- Required identity/profile for this task: author (allowed to push branches / create PRs)
- Current authenticated identity (from whoami): <username>
-122
View File
@@ -1,122 +0,0 @@
"""Shared task→permission map for resolver and tool gates (#69).
``gitea_resolve_task_capability`` and issue-mutating MCP tools must agree on
which profile operation each task requires. This module is the single source
of truth; regression tests assert tool gates cannot drift from it.
"""
from __future__ import annotations
TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"create_issue": {
"permission": "gitea.issue.create",
"role": "author",
},
"comment_issue": {
"permission": "gitea.issue.comment",
"role": "author",
},
"close_issue": {
"permission": "gitea.issue.close",
"role": "author",
},
"claim_issue": {
"permission": "gitea.issue.comment",
"role": "author",
},
"mark_issue": {
"permission": "gitea.issue.comment",
"role": "author",
},
"set_issue_labels": {
"permission": "gitea.issue.comment",
"role": "author",
},
"create_branch": {
"permission": "gitea.branch.create",
"role": "author",
},
"push_branch": {
"permission": "gitea.branch.push",
"role": "author",
},
"create_pr": {
"permission": "gitea.pr.create",
"role": "author",
},
"comment_pr": {
"permission": "gitea.pr.comment",
"role": "author",
},
"close_pr": {
"permission": "gitea.pr.close",
"role": "author",
},
"address_pr_change_requests": {
"permission": "gitea.branch.push",
"role": "author",
},
"review_pr": {
"permission": "gitea.pr.review",
"role": "reviewer",
},
"merge_pr": {
"permission": "gitea.pr.merge",
"role": "reviewer",
},
"blind_pr_queue_review": {
"permission": "gitea.pr.review",
"role": "reviewer",
},
"request_changes_pr": {
"permission": "gitea.pr.request_changes",
"role": "reviewer",
},
"approve_pr": {
"permission": "gitea.pr.approve",
"role": "reviewer",
},
"delete_branch": {
"permission": "gitea.branch.delete",
"role": "author",
},
"commit_files": {
"permission": "gitea.repo.commit",
"role": "author",
},
"gitea_commit_files": {
"permission": "gitea.repo.commit",
"role": "author",
},
}
# Issue-mutating MCP tools and their resolver task keys.
ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_create_issue": "create_issue",
"gitea_close_issue": "close_issue",
"gitea_create_issue_comment": "comment_issue",
"gitea_mark_issue": "mark_issue",
"gitea_set_issue_labels": "set_issue_labels",
"gitea_commit_files": "commit_files",
}
def required_permission(task: str) -> str:
"""Return the canonical operation a *task* requires (fail closed)."""
try:
return TASK_CAPABILITY_MAP[task]["permission"]
except KeyError as exc:
raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc
def required_role(task: str) -> str:
"""Return author/reviewer role kind for *task* (fail closed)."""
try:
return TASK_CAPABILITY_MAP[task]["role"]
except KeyError as exc:
raise KeyError(f"Unknown task/action: {task!r} (fail closed)") from exc
def tool_required_permission(tool_name: str) -> str:
"""Return the operation an issue-mutating tool must gate on."""
return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])
+4 -14
View File
@@ -162,8 +162,7 @@ class _AuditWiringBase(unittest.TestCase):
def _env(self, **extra):
env = {"GITEA_AUDIT_LOG": self.audit_path,
"GITEA_PROFILE_NAME": "gitea-author",
"GITEA_ALLOWED_OPERATIONS": (
"read,merge,gitea.issue.create,gitea.issue.close")}
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
env.update(extra)
return env
@@ -238,10 +237,7 @@ class TestSimpleToolAudit(_AuditWiringBase):
def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
# No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
with patch.dict(os.environ, {
"GITEA_PROFILE_NAME": "gitea-author",
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
}, clear=True):
with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
gitea_create_issue(title="x", remote="prgs")
issue_posts = [
c for c in mock_api.call_args_list if c.args[0] == "POST"
@@ -293,12 +289,9 @@ class TestGatedToolAudit(_AuditWiringBase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_merge_success_audited(self, _auth, mock_api):
# user, pr, feedback pr+reviews, merge POST, readback.
# user, pr, merge POST, readback — no extra identity call (uses result).
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
self._pr("author-bot"),
[{"id": 1, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
"submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
{}, {"merged_commit_sha": "c1"},
]
env = self._env(GITEA_PROFILE_NAME="gitea-merger",
@@ -335,10 +328,7 @@ class TestGatedToolAudit(_AuditWiringBase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_submit_review_success_audited(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "APPROVED"},
[{"id": 7, "user": {"login": "reviewer-bot"}, "state": "APPROVED",
"submitted_at": "2026-07-06T10:00:00Z", "dismissed": False}],
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
]
env = self._env(GITEA_PROFILE_NAME="gitea-reviewer",
GITEA_ALLOWED_OPERATIONS="read,review,approve")
+2 -39
View File
@@ -29,8 +29,8 @@ CONFIG = {
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
"gitea.pr.create", "gitea.branch.push",
"gitea.read", "gitea.issue.create", "gitea.pr.create",
"gitea.branch.push",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
@@ -97,30 +97,6 @@ class TestCapabilityStopTerminal(unittest.TestCase):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_list_prs(remote="prgs")
self.assertIn("Cannot perform reviewer task", str(ctx.exception))
self.assertIn("review_pr", str(ctx.exception))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_list_prs_allowed_after_author_task_clears_stale_denial(
self, _auth, _api, _get_all,
):
with patch.dict(os.environ, self._env()):
denied = mcp_server.gitea_resolve_task_capability(
task="review_pr", remote="prgs"
)
self.assertTrue(denied["stop_required"])
self.assertTrue(capability_stop_terminal.is_active())
cleared = mcp_server.gitea_resolve_task_capability(
task="claim_issue", remote="prgs"
)
self.assertTrue(cleared["allowed_in_current_session"])
self.assertTrue(cleared.get("cleared_stale_denial"))
self.assertFalse(capability_stop_terminal.is_active())
prs = mcp_server.gitea_list_prs(remote="prgs")
self.assertEqual(prs, [])
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
@@ -172,19 +148,6 @@ class TestCapabilityStopTerminal(unittest.TestCase):
)
self.assertFalse(result["pure"])
def test_empty_queue_with_parsed_trusted_status_passes_gate_check(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Repository: Scaled-Tech-Consulting/Gitea-Tools\n"
"pr_inventory_trust_gate.status: trusted_empty\n"
"pr_inventory_trust_gate.corroborated: true\n"
"Inventory profile: prgs-reviewer\n"
"No open PRs in queue."
)
result = assess_capability_stop_terminal_report(report)
self.assertTrue(result["pure"])
def test_pure_terminal_report_passes(self):
report = (
"Cannot perform reviewer task under current profile. "
-251
View File
@@ -1,251 +0,0 @@
"""Regression tests: gitea_commit_files tool gates match resolver and whoami verification.
Covers Issue #262 requirements.
"""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import task_capability_map
import gitea_config
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"full-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.repo.commit"
],
"forbidden_operations": [],
"execution_profile": "full-author",
},
"reviewer-no-commit": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "reviewer-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": [
"gitea.read", "gitea.pr.review", "gitea.pr.approve"
],
"forbidden_operations": [
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
],
"execution_profile": "reviewer-no-commit",
},
},
"rules": {"allow_runtime_switching": False},
}
class TestCommitFilesGate(unittest.TestCase):
def setUp(self):
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
"repo": "Example-Repo"},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir.cleanup()
def _env(self, profile: str) -> dict:
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
}
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_allowed_author_proceeds(self, _auth, mock_api, _role):
mock_api.return_value = {
"commit": {"sha": "abc123commit"},
"branch": {"name": "some-branch"},
}
with patch.dict(os.environ, self._env("full-author"), clear=True):
resolve = mcp_server.gitea_resolve_task_capability(
task="commit_files", remote="prgs"
)
self.assertTrue(resolve["allowed_in_current_session"])
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertTrue(res["success"])
self.assertEqual(res["commit"], "abc123commit")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_denied_reviewer_blocked(self, _auth, mock_api, _role):
mock_api.return_value = {"login": "reviewer-user"}
with patch.dict(os.environ, self._env("reviewer-no-commit"), clear=True):
resolve = mcp_server.gitea_resolve_task_capability(
task="commit_files", remote="prgs"
)
self.assertFalse(resolve["allowed_in_current_session"])
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertFalse(res["success"])
self.assertFalse(res.get("performed", True))
self.assertIn("permission_report", res)
self.assertEqual(
res["permission_report"]["missing_permission"], "gitea.repo.commit"
)
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
self.assertFalse(post_calls)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_unknown_profile_fails_closed(self, _auth, mock_api, _role):
mock_api.return_value = {"login": "reviewer-user"}
with patch.dict(os.environ, self._env("non-existent"), clear=True):
res = mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertFalse(res["success"])
self.assertFalse(res.get("performed", True))
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
self.assertFalse(post_calls)
class TestPreflightCommitFilesGate(unittest.TestCase):
def setUp(self):
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
"repo": "Example-Repo"},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
self.orig_whoami_called = mcp_server._preflight_whoami_called
self.orig_capability_called = mcp_server._preflight_capability_called
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
self.orig_capability_violation = mcp_server._preflight_capability_violation
self.orig_resolved_role = mcp_server._preflight_resolved_role
self.orig_process_start = mcp_server._process_start_porcelain
self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
self.orig_capability_files = mcp_server._preflight_capability_violation_files
self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
mcp_server._preflight_whoami_called = False
mcp_server._preflight_capability_called = False
mcp_server._preflight_whoami_violation = False
mcp_server._preflight_capability_violation = False
mcp_server._preflight_resolved_role = None
mcp_server._process_start_porcelain = ""
mcp_server._preflight_whoami_baseline_porcelain = None
mcp_server._preflight_capability_baseline_porcelain = None
mcp_server._preflight_whoami_violation_files = []
mcp_server._preflight_capability_violation_files = []
mcp_server._preflight_reviewer_violation_files = []
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
mcp_server._preflight_whoami_called = self.orig_whoami_called
mcp_server._preflight_capability_called = self.orig_capability_called
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
mcp_server._preflight_capability_violation = self.orig_capability_violation
mcp_server._preflight_resolved_role = self.orig_resolved_role
mcp_server._process_start_porcelain = self.orig_process_start
mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
mcp_server._preflight_capability_violation_files = self.orig_capability_files
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
self._dir.cleanup()
os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
os.environ.pop("GITEA_TEST_PORCELAIN", None)
def _env(self, profile: str) -> dict:
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
}
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_preflight_not_called_blocks_commit(self, _auth, mock_api):
mock_api.return_value = {"login": "author-user"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
os.environ["GITEA_TEST_PORCELAIN"] = ""
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_dirty_workspace_before_whoami_blocks_commit(self, _auth, mock_api):
mock_api.return_value = {"login": "author-user"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_commit_files(
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
message="Add x",
remote="prgs",
)
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
if __name__ == "__main__":
unittest.main()
-18
View File
@@ -107,24 +107,6 @@ class TestGetCredentials(unittest.TestCase):
class TestGetAuthHeader(unittest.TestCase):
"""Test the get_auth_header function."""
def setUp(self):
self._saved_env = os.environ.copy()
self._saved_configs = gitea_auth.DYNAMIC_CONFIGS.copy()
for key in (
"GITEA_MCP_CONFIG",
"GITEA_MCP_PROFILE",
"GITEA_TOKEN",
"GITEA_TOKEN_PRGS",
):
os.environ.pop(key, None)
gitea_auth.DYNAMIC_CONFIGS.clear()
def tearDown(self):
os.environ.clear()
os.environ.update(self._saved_env)
gitea_auth.DYNAMIC_CONFIGS.clear()
gitea_auth.DYNAMIC_CONFIGS.update(self._saved_configs)
@patch("gitea_auth.get_credentials", return_value=("user", "pass"))
def test_returns_basic_header(self, _cred):
header = gitea_auth.get_auth_header("example.com")
@@ -1,102 +0,0 @@
"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151/#152)."""
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
DOC = REPO_ROOT / "docs" / "mcp-client-registration.md"
def _doc_text():
assert DOC.is_file(), "missing docs/mcp-client-registration.md"
return DOC.read_text(encoding="utf-8")
def test_registration_doc_names_canonical_servers():
text = _doc_text()
assert "jenkins-mcp" in text
assert "glitchtip-mcp" in text
assert "Historical names" in text
assert "jenkins-readonly" in text
assert "glitchtip-readonly" in text
def test_registration_doc_lists_expected_tool_visibility():
text = _doc_text()
for tool in (
"jenkins_whoami",
"jenkins_list_jobs",
"jenkins_latest_build",
"jenkins_build_status",
"jenkins_get_build",
"glitchtip_whoami",
"glitchtip_list_projects",
"glitchtip_list_unresolved",
"glitchtip_get_issue",
"glitchtip_recent_events",
"glitchtip_search",
):
assert tool in text
def test_registration_doc_requires_reload_and_negative_case():
text = _doc_text()
lower = text.lower()
assert "reconnect" in lower
assert "reload" in lower
assert "enabled but no usable tools" in lower
assert "SKIPPED" in text
def test_registration_doc_preserves_boundaries():
text = " ".join(_doc_text().split())
for phrase in (
"Do not add Jenkins or GlitchTip credentials to the Gitea MCP server",
"do not add Gitea write credentials to the GlitchTip server",
"must not expose build trigger tools",
"jenkins-write-mcp",
"jenkins_mcp.write_server",
"remains read-only",
):
assert phrase in text
def test_registration_doc_separates_jenkins_trigger_from_read_surface():
text = _doc_text()
assert "jenkins_trigger_build" in text
assert "must **not** appear on `jenkins-mcp`" in text
assert "not" in text.lower() and "registered by default" in text.lower()
def test_registration_doc_has_no_secret_material_or_live_urls():
text = _doc_text()
for marker in (
"https://",
"http://",
"Authorization:",
"BEGIN PRIVATE KEY",
"ghp_",
"token-value",
):
assert marker not in text
def test_related_docs_link_registration_doc():
for name in (
"README.md",
"docs/llm-workflow-runbooks.md",
"docs/architecture/jenkins-readonly-build-status-design.md",
"docs/architecture/glitchtip-readonly-tools-design.md",
):
text = (REPO_ROOT / name).read_text(encoding="utf-8")
assert "mcp-client-registration.md" in text, (
f"{name} does not link docs/mcp-client-registration.md")
def test_related_docs_keep_jenkins_trigger_off_read_surface():
for name in (
"docs/safety-model.md",
"docs/architecture/jenkins-readonly-build-status-design.md",
):
text = (REPO_ROOT / name).read_text(encoding="utf-8")
assert "jenkins-write-mcp" in text
assert "jenkins_mcp.write_server" in text
assert "jenkins-mcp" in text
-147
View File
@@ -1,147 +0,0 @@
import os
import subprocess
import sys
import unittest
from unittest.mock import patch
import role_session_router
from role_session_router import python_bytes_have_conflict_markers
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
_HEALTH_SUBPROCESS_TIMEOUT_SEC = 30
def _health_test_python():
"""Portable interpreter for subprocess health checks (#245).
Prefer the active pytest interpreter, then ``GITEA_TOOLS_TEST_PYTHON``,
else skip never hard-code ``<repo>/venv/bin/python``.
"""
if (
sys.executable
and os.path.isfile(sys.executable)
and os.access(sys.executable, os.X_OK)
):
return sys.executable
override = (os.environ.get("GITEA_TOOLS_TEST_PYTHON") or "").strip()
if override and os.path.isfile(override) and os.access(override, os.X_OK):
return override
raise unittest.SkipTest(
"repo venv not available; run under a python interpreter or set "
"GITEA_TOOLS_TEST_PYTHON"
)
def _run_python_script(cwd, script_path, *, timeout=_HEALTH_SUBPROCESS_TIMEOUT_SEC):
"""Run a script in a child process with timeout and guaranteed teardown."""
python = _health_test_python()
proc = subprocess.Popen(
[python, script_path],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=cwd,
)
try:
stdout, stderr = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
proc.kill()
stdout, stderr = proc.communicate()
raise
finally:
if proc.poll() is None:
proc.terminate()
try:
proc.wait(timeout=5)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
return proc.returncode, stdout, stderr, proc
class TestMCPHealth(unittest.TestCase):
def setUp(self):
self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
self.temp_file = os.path.join(self.project_root, "test_temp_conflict.py")
if os.path.exists(self.temp_file):
os.remove(self.temp_file)
def tearDown(self):
if os.path.exists(self.temp_file):
os.remove(self.temp_file)
def test_health_test_python_prefers_sys_executable(self):
resolved = _health_test_python()
self.assertEqual(resolved, sys.executable)
def test_health_test_python_skips_when_no_interpreter(self):
with patch.object(sys, "executable", ""), patch.dict(
os.environ, {}, clear=True
):
with self.assertRaises(unittest.SkipTest):
_health_test_python()
def test_conflict_marker_helper_ignores_decorative_equals_border(self):
sample = b'banner = """\n===========================================\n"""\n'
self.assertFalse(python_bytes_have_conflict_markers(sample))
def test_conflict_marker_helper_detects_real_markers(self):
sample = (
b"<" * 7 + b" HEAD\n"
b"print('hello')\n"
b"=" * 7 + b"\n"
b"print('world')\n"
b">" * 7 + b" main\n"
)
self.assertTrue(python_bytes_have_conflict_markers(sample))
def test_startup_conflict_detection(self):
# Create a Python file with conflict markers constructed dynamically
with open(self.temp_file, "w") as f:
f.write("<" * 7 + " HEAD\n")
f.write("print('hello')\n")
f.write("=" * 7 + "\n")
f.write("print('world')\n")
f.write(">" * 7 + " main\n")
script_path = os.path.join(self.project_root, "mcp_server.py")
returncode, _stdout, stderr, _proc = _run_python_script(
self.project_root, script_path
)
self.assertEqual(returncode, 1)
stderr_text = stderr.decode()
self.assertIn("infra_stop", stderr_text)
self.assertIn(
"Unresolved merge conflict detected in test_temp_conflict.py",
stderr_text,
)
@patch("role_session_router.check_mid_merge", return_value=True)
@patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-reviewer",
"allowed_operations": ["gitea.pr.review"],
},
)
def test_route_task_session_blocks_during_merge(self, mock_profile, mock_check):
res = gitea_route_task_session("review_pr")
self.assertEqual(res["route_result"], "infra_stop")
self.assertIn("infra_stop", res["message"])
@patch("role_session_router.check_mid_merge", return_value=True)
@patch(
"mcp_server.get_profile",
return_value={
"profile_name": "prgs-reviewer",
"allowed_operations": ["gitea.pr.review"],
},
)
def test_resolve_task_capability_blocks_during_merge(
self, mock_profile, mock_check
):
res = gitea_resolve_task_capability("review_pr")
self.assertTrue(res["infra_stop"])
self.assertFalse(res["allowed_in_current_session"])
self.assertIn("infra_stop", res["exact_safe_next_action"])
+9 -15
View File
@@ -19,9 +19,6 @@ from mcp_server import gitea_create_issue # noqa: E402
from review_proofs import assess_duplicate_search_proof as proof_assess # noqa: E402
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
}
CANONICAL_TITLE = (
"Add hard queue-target resolution wall before PR inventory "
"or empty-queue claims"
@@ -121,8 +118,7 @@ class TestCreateIssueMCPGate(unittest.TestCase):
mock_get_all.return_value = [
{"number": 201, "title": CANONICAL_TITLE, "state": "open"},
]
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
self.assertFalse(result["performed"])
self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE)
mock_api.assert_not_called()
@@ -139,14 +135,13 @@ class TestCreateIssueMCPGate(unittest.TestCase):
{"number": 201, "title": CANONICAL_TITLE, "state": "open"},
]
mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"}
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_create_issue(
title=CANONICAL_TITLE,
body="Follow-up scope",
remote="prgs",
allow_duplicate_override=True,
split_from_issue=201,
)
result = gitea_create_issue(
title=CANONICAL_TITLE,
body="Follow-up scope",
remote="prgs",
allow_duplicate_override=True,
split_from_issue=201,
)
self.assertEqual(result["number"], 210)
payload = mock_api.call_args[0][3]
self.assertIn("Operator-approved split from #201.", payload["body"])
@@ -161,8 +156,7 @@ class TestCreateIssueMCPGate(unittest.TestCase):
):
mock_role_check.return_value = (True, [])
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_create_issue(title="Unique new issue", body="body text")
result = gitea_create_issue(title="Unique new issue", body="body text")
self.assertEqual(result["number"], 1)
-99
View File
@@ -1,99 +0,0 @@
"""Tests for issue-lock worktree validation (#249)."""
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import issue_lock_worktree # noqa: E402
class TestIssueLockWorktreeAssessment(unittest.TestCase):
def test_clean_base_branch_passes(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path="/scratch/wt",
current_branch="master",
porcelain_status="",
)
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_dirty_tracked_files_fail(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path="/scratch/wt",
current_branch="master",
porcelain_status=" M gitea_mcp_server.py\n",
)
self.assertFalse(result["proven"])
self.assertIn("tracked file edits exist before issue lock", result["reasons"][0])
def test_feature_branch_fails(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path="/scratch/wt",
current_branch="feat/issue-243-forbidden-git-gaps",
porcelain_status="",
)
self.assertFalse(result["proven"])
self.assertIn("issue lock must be taken from base branch", result["reasons"][0])
def test_untracked_files_do_not_block(self):
result = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path="/scratch/wt",
current_branch="main",
porcelain_status="?? notes.txt\n",
)
self.assertTrue(result["proven"])
class TestIssueLockWorktreeResolution(unittest.TestCase):
def test_explicit_path_wins(self):
resolved = issue_lock_worktree.resolve_author_worktree_path(
"/tmp/scratch/wt", "/shared/dev"
)
self.assertEqual(resolved, os.path.realpath("/tmp/scratch/wt"))
def test_env_var_when_explicit_missing(self):
with patch.dict(os.environ, {"GITEA_AUTHOR_WORKTREE": "/tmp/from-env"}):
resolved = issue_lock_worktree.resolve_author_worktree_path(
None, "/shared/dev"
)
self.assertEqual(resolved, os.path.realpath("/tmp/from-env"))
def test_project_root_fallback(self):
resolved = issue_lock_worktree.resolve_author_worktree_path(
None, "/shared/dev"
)
self.assertEqual(resolved, os.path.realpath("/shared/dev"))
class TestPrWorktreeMatch(unittest.TestCase):
def test_matching_paths_pass(self):
result = issue_lock_worktree.verify_pr_worktree_matches_lock(
"/tmp/scratch/wt",
"/tmp/scratch/wt",
"/shared/dev",
)
self.assertTrue(result["proven"])
def test_mismatch_fails(self):
result = issue_lock_worktree.verify_pr_worktree_matches_lock(
"/tmp/scratch/wt",
"/shared/dev",
"/shared/dev",
)
self.assertFalse(result["proven"])
self.assertIn("does not match locked worktree", result["reasons"][0])
def test_missing_locked_path_skips_check(self):
result = issue_lock_worktree.verify_pr_worktree_matches_lock(
None,
"/any/path",
"/shared/dev",
)
self.assertTrue(result["proven"])
if __name__ == "__main__":
unittest.main()
-210
View File
@@ -1,210 +0,0 @@
"""Regression tests: issue-write tool gates match gitea_resolve_task_capability (#69)."""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
import task_capability_map
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"comment-only": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "commenter",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.issue.comment"],
"forbidden_operations": [
"gitea.issue.create", "gitea.issue.close"],
"execution_profile": "comment-only",
},
"full-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.issue.close",
"gitea.issue.comment"],
"forbidden_operations": [],
"execution_profile": "full-author",
},
},
"rules": {"allow_runtime_switching": False},
}
ISSUE_WRITE_TASKS = (
"create_issue", "close_issue", "comment_issue", "mark_issue",
"set_issue_labels",
)
TOOL_CALLS = {
"create_issue": lambda: mcp_server.gitea_create_issue(
title="New issue", remote="prgs"),
"close_issue": lambda: mcp_server.gitea_close_issue(
issue_number=9, remote="prgs"),
"comment_issue": lambda: mcp_server.gitea_create_issue_comment(
issue_number=9, body="hello", remote="prgs"),
"mark_issue": lambda: mcp_server.gitea_mark_issue(
issue_number=9, action="start", remote="prgs"),
"set_issue_labels": lambda: mcp_server.gitea_set_issue_labels(
issue_number=9, labels=["bug"], remote="prgs"),
}
class IssueWriteGateBase(unittest.TestCase):
def setUp(self):
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
"repo": "Example-Repo"},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
self._dir.cleanup()
def _env(self, profile: str) -> dict:
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
}
class TestResolverToolAlignment(unittest.TestCase):
def test_issue_mutation_tools_match_resolver_permissions(self):
for tool_name, task_key in (
task_capability_map.ISSUE_MUTATION_TOOL_TASKS.items()
):
self.assertEqual(
task_capability_map.tool_required_permission(tool_name),
task_capability_map.required_permission(task_key),
msg=f"{tool_name} gate drift from resolver task {task_key}",
)
class TestDeniedIssueWritesFailClosed(IssueWriteGateBase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolver_denied_tasks_block_raw_tools(
self, _auth, mock_api, _get_all, _role
):
mock_api.return_value = {"number": 1}
with patch.dict(os.environ, self._env("comment-only"), clear=True):
for task in ISSUE_WRITE_TASKS:
with self.subTest(task=task):
resolve = mcp_server.gitea_resolve_task_capability(
task=task, remote="prgs")
if resolve["allowed_in_current_session"]:
continue
result = TOOL_CALLS[task]()
self.assertFalse(result.get("success", True), task)
self.assertFalse(result.get("performed", True), task)
self.assertTrue(result.get("reasons"), task)
self.assertIn("permission_report", result, task)
self.assertEqual(
result["permission_report"]["missing_permission"],
resolve["required_operation_permission"],
)
class TestAllowedIssueWritesProceed(IssueWriteGateBase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolver_allowed_create_issue_proceeds(
self, _auth, mock_api, _get_all, _role
):
mock_api.return_value = {"number": 42, "html_url": "https://x/42"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
resolve = mcp_server.gitea_resolve_task_capability(
task="create_issue", remote="prgs")
self.assertTrue(resolve["allowed_in_current_session"])
result = mcp_server.gitea_create_issue(title="Allowed", remote="prgs")
self.assertEqual(result.get("number"), 42)
post_calls = [
c for c in mock_api.call_args_list if c.args[0] == "POST"
]
self.assertTrue(post_calls)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolver_allowed_close_issue_proceeds(self, _auth, mock_api):
mock_api.return_value = {"state": "closed"}
with patch.dict(os.environ, self._env("full-author"), clear=True):
resolve = mcp_server.gitea_resolve_task_capability(
task="close_issue", remote="prgs")
self.assertTrue(resolve["allowed_in_current_session"])
result = mcp_server.gitea_close_issue(issue_number=9, remote="prgs")
self.assertTrue(result.get("success"))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolver_allowed_mark_issue_proceeds(self, _auth, mock_api):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and url.endswith("/labels?limit=100"):
return [{"id": 10, "name": "status:in-progress"}]
if method == "POST" and "/issues/9/labels" in url:
return [{"name": "status:in-progress"}]
if method == "GET" and url.endswith("/user"):
return {"login": "author-user"}
return {}
mock_api.side_effect = api_side_effect
with patch.dict(os.environ, self._env("full-author"), clear=True):
resolve = mcp_server.gitea_resolve_task_capability(
task="mark_issue", remote="prgs")
self.assertTrue(resolve["allowed_in_current_session"])
result = mcp_server.gitea_mark_issue(
issue_number=9, action="start", remote="prgs")
self.assertTrue(result.get("success"))
class TestCreateIssueMissingPermissionReport(IssueWriteGateBase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_create_issue_without_create_permission_blocked(
self, _auth, mock_api, _get_all, _role
):
with patch.dict(os.environ, self._env("comment-only"), clear=True):
result = mcp_server.gitea_create_issue(title="Blocked", remote="prgs")
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertIsNone(result["number"])
self.assertEqual(
result["permission_report"]["missing_permission"],
"gitea.issue.create",
)
mock_api.assert_not_called()
if __name__ == "__main__":
unittest.main()
-260
View File
@@ -1,260 +0,0 @@
"""Tests for MCP discoverability validation (Issue #155)."""
import os
import sys
import json
import tempfile
import unittest
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from mcp_discoverability import (
validate_mcp_client_config,
RELOAD_INSTRUCTIONS,
EXPECTED_JENKINS_TOOLS,
EXPECTED_GLITCHTIP_TOOLS,
)
class TestMcpDiscoverability(unittest.TestCase):
def setUp(self):
self.tmp_dir = tempfile.TemporaryDirectory()
self.gitea_config_path = os.path.join(self.tmp_dir.name, "gitea-mcp.json")
self.client_config_path = os.path.join(self.tmp_dir.name, "claude_desktop_config.json")
def tearDown(self):
self.tmp_dir.cleanup()
def _write_json(self, path, data):
with open(path, "w", encoding="utf-8") as fh:
json.dump(data, fh)
def _v2_gitea_config(self, jenkins_enabled=True, glitchtip_enabled=False):
return {
"version": 2,
"contexts": {
"prod": {
"enabled": True,
"services": {
"jenkins": {
"enabled": jenkins_enabled,
"kind": "jenkins",
"capabilities": ["read"]
},
"glitchtip": {
"enabled": glitchtip_enabled,
"kind": "glitchtip",
"capabilities": ["read"]
}
}
}
}
}
def test_static_validation_success(self):
g_cfg = self._v2_gitea_config(jenkins_enabled=True, glitchtip_enabled=True)
self._write_json(self.gitea_config_path, g_cfg)
c_cfg = {
"mcpServers": {
"jenkins-mcp": {
"command": "python3",
"args": ["-m", "jenkins_mcp"],
"env": {
"JENKINS_MCP_PROFILE": "jenkins-readonly",
"JENKINS_MCP_CONFIG": "/path/to/config.json"
}
},
"glitchtip-mcp": {
"command": "python3",
"args": ["-m", "glitchtip_mcp"],
"env": {
"GLITCHTIP_MCP_PROFILE": "glitchtip-readonly",
"GLITCHTIP_MCP_CONFIG": "/path/to/config.json"
}
}
}
}
self._write_json(self.client_config_path, c_cfg)
res = validate_mcp_client_config(
client_config_path=self.client_config_path,
gitea_config_path=self.gitea_config_path,
live=False
)
self.assertTrue(res)
def test_stale_server_name_rejected(self):
g_cfg = self._v2_gitea_config(jenkins_enabled=True)
self._write_json(self.gitea_config_path, g_cfg)
# Uses stale key in client config
c_cfg = {
"mcpServers": {
"jenkins-readonly": {
"command": "python3",
"args": ["-m", "jenkins_mcp"],
"env": {
"JENKINS_MCP_PROFILE": "jenkins-readonly",
"JENKINS_MCP_CONFIG": "/path/to/config.json"
}
}
}
}
self._write_json(self.client_config_path, c_cfg)
res = validate_mcp_client_config(
client_config_path=self.client_config_path,
gitea_config_path=self.gitea_config_path,
live=False
)
self.assertFalse(res)
def test_incorrect_arguments_rejected(self):
g_cfg = self._v2_gitea_config(jenkins_enabled=True)
self._write_json(self.gitea_config_path, g_cfg)
# Missing -m or wrong module
c_cfg = {
"mcpServers": {
"jenkins-mcp": {
"command": "python3",
"args": ["wrong_runner.py"],
"env": {
"JENKINS_MCP_PROFILE": "jenkins-readonly",
"JENKINS_MCP_CONFIG": "/path/to/config.json"
}
}
}
}
self._write_json(self.client_config_path, c_cfg)
res = validate_mcp_client_config(
client_config_path=self.client_config_path,
gitea_config_path=self.gitea_config_path,
live=False
)
self.assertFalse(res)
def test_missing_required_env_rejected(self):
g_cfg = self._v2_gitea_config(jenkins_enabled=True)
self._write_json(self.gitea_config_path, g_cfg)
# Missing JENKINS_MCP_PROFILE env variable
c_cfg = {
"mcpServers": {
"jenkins-mcp": {
"command": "python3",
"args": ["-m", "jenkins_mcp"],
"env": {
"JENKINS_MCP_CONFIG": "/path/to/config.json"
}
}
}
}
self._write_json(self.client_config_path, c_cfg)
res = validate_mcp_client_config(
client_config_path=self.client_config_path,
gitea_config_path=self.gitea_config_path,
live=False
)
self.assertFalse(res)
@patch("subprocess.Popen")
def test_live_check_verification_success(self, mock_popen):
g_cfg = self._v2_gitea_config(jenkins_enabled=True)
self._write_json(self.gitea_config_path, g_cfg)
c_cfg = {
"mcpServers": {
"jenkins-mcp": {
"command": "python3",
"args": ["-m", "jenkins_mcp"],
"env": {
"JENKINS_MCP_PROFILE": "jenkins-readonly",
"JENKINS_MCP_CONFIG": "/path/to/config.json"
}
}
}
}
self._write_json(self.client_config_path, c_cfg)
# Mock stdout lines for JSON-RPC
mock_proc = MagicMock()
mock_popen.return_value = mock_proc
init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2024-11-05"}})
tools_resp = json.dumps({
"jsonrpc": "2.0",
"id": 2,
"result": {
"tools": [{"name": tool} for tool in EXPECTED_JENKINS_TOOLS]
}
})
mock_proc.stdout.readline.side_effect = [
init_resp + "\n",
tools_resp + "\n",
""
]
res = validate_mcp_client_config(
client_config_path=self.client_config_path,
gitea_config_path=self.gitea_config_path,
live=True
)
self.assertTrue(res)
@patch("subprocess.Popen")
def test_live_check_empty_toolset_rejected(self, mock_popen):
g_cfg = self._v2_gitea_config(jenkins_enabled=True)
self._write_json(self.gitea_config_path, g_cfg)
c_cfg = {
"mcpServers": {
"jenkins-mcp": {
"command": "python3",
"args": ["-m", "jenkins_mcp"],
"env": {
"JENKINS_MCP_PROFILE": "jenkins-readonly",
"JENKINS_MCP_CONFIG": "/path/to/config.json"
}
}
}
}
self._write_json(self.client_config_path, c_cfg)
mock_proc = MagicMock()
mock_popen.return_value = mock_proc
init_resp = json.dumps({"jsonrpc": "2.0", "id": 1, "result": {}})
tools_resp = json.dumps({
"jsonrpc": "2.0",
"id": 2,
"result": {
"tools": [] # empty tool list
}
})
mock_proc.stdout.readline.side_effect = [
init_resp + "\n",
tools_resp + "\n",
""
]
# Should return False (coverage fails) when no tools are exposed
res = validate_mcp_client_config(
client_config_path=self.client_config_path,
gitea_config_path=self.gitea_config_path,
live=True
)
self.assertFalse(res)
def test_runbook_instructions_content(self):
self.assertIn("MCP CLIENT RELOAD/RECONNECT RUNBOOK", RELOAD_INSTRUCTIONS)
self.assertIn("Codex", RELOAD_INSTRUCTIONS)
self.assertIn("Gemini", RELOAD_INSTRUCTIONS)
self.assertIn("Claude Desktop", RELOAD_INSTRUCTIONS)
if __name__ == "__main__":
unittest.main()
+70 -546
View File
@@ -48,54 +48,6 @@ import mcp_server
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
return {
"id": review_id,
"user": {"login": reviewer},
"state": verdict,
"commit_id": sha,
"submitted_at": "2026-07-06T10:00:00Z",
"dismissed": False,
}
def _visible_approval_reviews(reviewer="reviewer-bot", sha="abc123"):
return [_formal_review(reviewer, "APPROVED", sha=sha)]
# Issue-write tools are profile-gated (#69).
ISSUE_WRITE_ENV = {
"GITEA_ALLOWED_OPERATIONS": (
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
),
}
# create_pr is gated on author profile + namespace (#69, #209).
CREATE_PR_ENV = {
"GITEA_PROFILE_NAME": "author-test",
"GITEA_ALLOWED_OPERATIONS": (
"gitea.read,gitea.pr.create,gitea.branch.push"
),
"GITEA_FORBIDDEN_OPERATIONS": (
"gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
),
}
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
record = {
"issue_number": issue_number,
"branch_name": branch_name,
"remote": "dadeschools",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
}
record.update(overrides)
return record
# ---------------------------------------------------------------------------
# Create Issue
# ---------------------------------------------------------------------------
@@ -108,7 +60,7 @@ class TestCreateIssue(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_issue(self, _auth, _get_all, mock_api, _role):
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with patch.dict(os.environ, {}, clear=True):
result = gitea_create_issue(title="Test issue", body="body text")
self.assertEqual(result["number"], 1)
self.assertNotIn("url", result)
@@ -125,8 +77,7 @@ class TestCreateIssue(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
result = gitea_create_issue(title="Test issue", body="body text")
self.assertIn("issues/1", result["url"])
@@ -137,8 +88,7 @@ class TestCreateIssue(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_create_issue(title="Test", remote="prgs")
result = gitea_create_issue(title="Test", remote="prgs")
self.assertEqual(result["number"], 5)
url = mock_api.call_args[0][1]
self.assertIn("gitea.prgs.cc", url)
@@ -150,56 +100,32 @@ class TestCreateIssue(unittest.TestCase):
# ---------------------------------------------------------------------------
class TestCreatePR(unittest.TestCase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with patch.dict(os.environ, {}, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
payload = mock_api.call_args[0][3]
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
self.assertIn("Closes #123", payload["title"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertIn("pulls/3", result["url"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
self.assertIn("Closes #123", str(ctx.exception))
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
# ---------------------------------------------------------------------------
# Close Issue
@@ -210,8 +136,7 @@ class TestCloseIssue(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_closes_issue(self, _auth, mock_api):
mock_api.return_value = {"state": "closed"}
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_close_issue(issue_number=42)
result = gitea_close_issue(issue_number=42)
self.assertTrue(result["success"])
self.assertIn("42", result["message"])
patch_call = next(call for call in mock_api.call_args_list if call[0][0] == "PATCH")
@@ -310,8 +235,7 @@ class TestMarkIssue(unittest.TestCase):
[{"id": 10, "name": "status:in-progress"}],
[{"name": "status:in-progress"}],
]
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_mark_issue(issue_number=5, action="start")
result = gitea_mark_issue(issue_number=5, action="start")
self.assertTrue(result["success"])
self.assertIn("claimed", result["message"])
@@ -322,8 +246,7 @@ class TestMarkIssue(unittest.TestCase):
[{"id": 10, "name": "status:in-progress"}],
None,
]
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_mark_issue(issue_number=5, action="done")
result = gitea_mark_issue(issue_number=5, action="done")
self.assertTrue(result["success"])
self.assertIn("released", result["message"])
@@ -335,9 +258,8 @@ class TestMarkIssue(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_missing_label_raises(self, _auth, mock_api):
mock_api.return_value = [] # no labels exist
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with self.assertRaises(RuntimeError):
gitea_mark_issue(issue_number=5, action="start")
with self.assertRaises(RuntimeError):
gitea_mark_issue(issue_number=5, action="start")
# ---------------------------------------------------------------------------
@@ -347,14 +269,12 @@ class TestAuthErrors(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=None)
def test_no_credentials_raises(self, _auth):
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with self.assertRaises(RuntimeError):
gitea_create_issue(title="test")
with self.assertRaises(RuntimeError):
gitea_create_issue(title="test")
def test_unknown_remote_raises(self):
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with self.assertRaises(ValueError):
gitea_create_issue(title="test", remote="nonexistent")
with self.assertRaises(ValueError):
gitea_create_issue(title="test", remote="nonexistent")
# ---------------------------------------------------------------------------
@@ -504,10 +424,6 @@ class TestMergePR(unittest.TestCase):
f"unexpected merge mutation: {method} {url}",
)
def _feedback_reads(self, author="author-bot", sha="abc123"):
"""PR + reviews GETs for gitea_get_pr_review_feedback during merge."""
return [self._pr(author, sha=sha), _visible_approval_reviews(sha=sha)]
# -- success --------------------------------------------------------------
@patch("mcp_server.api_request")
@@ -515,7 +431,6 @@ class TestMergePR(unittest.TestCase):
def test_merge_succeeds_when_all_gates_pass(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, # merge POST
{"merged_commit_sha": "mergecommit99"}, # read-back
]
@@ -532,8 +447,8 @@ class TestMergePR(unittest.TestCase):
self.assertEqual(r["head_sha"], "abc123")
self.assertEqual(r["merge_method"], "squash")
self.assertEqual(r["merge_commit"], "mergecommit99")
# 5th call is the merge POST with the requested method/title/message.
merge_call = mock_api.call_args_list[4]
# 3rd call is the merge POST with the requested method/title/message.
merge_call = mock_api.call_args_list[2]
self.assertEqual(merge_call.args[0], "POST")
self.assertTrue(merge_call.args[1].endswith("/pulls/8/merge"))
payload = merge_call.args[3]
@@ -548,7 +463,6 @@ class TestMergePR(unittest.TestCase):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
[{"filename": "a.py"}, {"filename": "b.py"}], # files
*self._feedback_reads(),
{}, # merge POST
{"merged_commit_sha": "c1"}, # read-back
]
@@ -568,7 +482,6 @@ class TestMergePR(unittest.TestCase):
"""Merge OK + read-back GET failure => explicit cleanup skip, not silence."""
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, # merge POST
RuntimeError("HTTP 502: Gitea upstream unavailable"), # read-back fails
]
@@ -584,8 +497,8 @@ class TestMergePR(unittest.TestCase):
# The skip is explicit, not silent.
self.assertEqual(r["cleanup_status"], "skipped (merge read-back failed)")
# No tracker-cleanup API traffic after the failed read-back:
# user, PR (eligibility), feedback PR+reviews, merge POST, read-back.
self.assertEqual(mock_api.call_count, 6)
# user, PR (eligibility), merge POST, read-back — and nothing more.
self.assertEqual(mock_api.call_count, 4)
for c in mock_api.call_args_list:
self.assertNotEqual(c.args[0], "DELETE")
@@ -597,7 +510,6 @@ class TestMergePR(unittest.TestCase):
"""Unexpected cleanup exception => merge still succeeds; error surfaced redacted."""
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, # merge POST
{"merged_commit_sha": "c9"}, # read-back OK
]
@@ -785,7 +697,6 @@ class TestMergePR(unittest.TestCase):
def test_output_redacts_secrets(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
{}, {"merged_commit_sha": "c1"},
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -803,7 +714,6 @@ class TestMergePR(unittest.TestCase):
def test_merge_error_message_redacts_credential(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
*self._feedback_reads(),
RuntimeError("HTTP 500: token abc-secret-xyz rejected"),
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
@@ -816,45 +726,6 @@ class TestMergePR(unittest.TestCase):
self.assertIn("[REDACTED]", blob)
self.assertNotIn("abc-secret-xyz", blob)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_merge_blocked_without_visible_approval(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
self._pr("author-bot"),
[_formal_review("sysadmin", "PENDING")],
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
self.assertFalse(r["performed"])
self.assertFalse(r.get("approval_visible"))
self.assertTrue(any("no visible APPROVED review" in x for x in r["reasons"]))
self._assert_no_merge_call(mock_api)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_merge_blocked_on_request_changes(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "merger-bot"}, self._pr("author-bot"),
self._pr("author-bot"),
[
_formal_review("reviewer-bot", "APPROVED"),
_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=2),
],
]
env = {"GITEA_PROFILE_NAME": "gitea-merger",
"GITEA_ALLOWED_OPERATIONS": "read,merge"}
with patch.dict(os.environ, env, clear=True):
r = gitea_merge_pr(
pr_number=8, confirmation=self._confirm(8), remote="prgs")
self.assertFalse(r["performed"])
self.assertTrue(r.get("has_blocking_change_requests"))
self.assertTrue(any("REQUEST_CHANGES" in x for x in r["reasons"]))
self._assert_no_merge_call(mock_api)
class TestNoUngatedMergePath(unittest.TestCase):
"""Prove no other exposed tool can merge (#16 surface audit)."""
@@ -1071,11 +942,9 @@ class TestGetFile(unittest.TestCase):
# ---------------------------------------------------------------------------
class TestCommitFiles(unittest.TestCase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_commit_files_success(self, _auth, mock_api, _role):
def test_commit_files_success(self, _auth, mock_api):
mock_api.return_value = {
"commit": {"sha": "commit-sha-123"},
"branch": {"name": "test-branch"}
@@ -1083,15 +952,11 @@ class TestCommitFiles(unittest.TestCase):
files = [
{"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
]
env = {
"GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit",
}
with patch.dict(os.environ, env, clear=True):
result = gitea_commit_files(
files=files,
message="Initial commit",
new_branch="test-branch"
)
result = gitea_commit_files(
files=files,
message="Initial commit",
new_branch="test-branch"
)
self.assertTrue(result["success"])
self.assertEqual(result["commit"], "commit-sha-123")
self.assertEqual(result["branch"], "test-branch")
@@ -1616,9 +1481,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_duplicate_terminal_decision_blocked(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 1, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=1)],
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 1},
{"login": "reviewer-bot"}, self._pr("author-bot"),
]
gitea_mark_final_review_decision(
@@ -1701,9 +1564,7 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_approve_succeeds_when_eligible(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=7)],
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 7},
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -1713,63 +1574,16 @@ class TestSubmitPrReview(unittest.TestCase):
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
self.assertTrue(r.get("review_verdict_visible"))
self.assertEqual(r["authenticated_user"], "reviewer-bot")
self.assertEqual(r["pr_author"], "author-bot")
self.assertEqual(r["head_sha"], "abc123")
post_calls = [
c for c in mock_api.call_args_list
if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
]
self.assertEqual(len(post_calls), 1)
payload = post_calls[0].args[3]
self.assertEqual(payload["event"], "APPROVED")
method, url = mock_api.call_args.args[0], mock_api.call_args.args[1]
self.assertEqual(method, "POST")
self.assertTrue(url.endswith("/pulls/8/reviews"))
payload = mock_api.call_args.args[3]
self.assertEqual(payload["event"], "APPROVE")
self.assertEqual(payload["commit_id"], "abc123")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_approve_fails_when_verdict_stays_pending(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "PENDING"},
{"id": 7, "state": "PENDING"},
[_formal_review("reviewer-bot", "PENDING", review_id=7)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve", body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
self.assertFalse(r["performed"])
self.assertFalse(r.get("review_verdict_visible"))
self.assertTrue(any("expected visible 'APPROVED'" in x for x in r["reasons"]))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_approve_submits_pending_draft_when_api_returns_pending(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 7, "state": "PENDING"},
{"id": 7, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=7)],
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
with patch.dict(os.environ, env, clear=True):
r = gitea_submit_pr_review(
pr_number=8, action="approve", body="LGTM", remote="prgs",
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
submit_calls = [
c for c in mock_api.call_args_list
if c.args[0] == "POST" and "/reviews/7" in c.args[1]
]
self.assertEqual(len(submit_calls), 1)
self.assertEqual(submit_calls[0].args[3]["event"], "APPROVED")
# -- request_changes ------------------------------------------------------
@patch("mcp_server.api_request")
@@ -1777,9 +1591,7 @@ class TestSubmitPrReview(unittest.TestCase):
def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 9, "state": "REQUEST_CHANGES"},
[_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=9)],
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 9},
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,request_changes"}
@@ -1790,12 +1602,7 @@ class TestSubmitPrReview(unittest.TestCase):
final_review_decision_ready=True,
)
self.assertTrue(r["performed"])
post_calls = [
c for c in mock_api.call_args_list
if c.args[0] == "POST" and c.args[1].endswith("/pulls/8/reviews")
]
self.assertEqual(len(post_calls), 1)
self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
self.assertEqual(mock_api.call_args.args[3]["event"], "REQUEST_CHANGES")
def test_request_changes_blocked_without_eligibility(self):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
@@ -1906,8 +1713,7 @@ class TestSubmitPrReview(unittest.TestCase):
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot", sha="abc123"),
{"id": 5, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=5)],
{"id": 5},
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve"}
@@ -2059,12 +1865,8 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_correction_flow_allows_second_terminal_review(self, _auth, mock_api):
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 42, "state": "APPROVED"},
[_formal_review("reviewer-bot", "APPROVED", review_id=42)],
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 43, "state": "REQUEST_CHANGES"},
[_formal_review("reviewer-bot", "REQUEST_CHANGES", review_id=43)],
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 42},
{"login": "reviewer-bot"}, self._pr("author-bot"), {"id": 43},
]
env = {"GITEA_PROFILE_NAME": "gitea-reviewer",
"GITEA_ALLOWED_OPERATIONS": "read,review,approve,request_changes"}
@@ -2137,7 +1939,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
patch("gitea_audit.audit_enabled", return_value=True).start()
self.mock_audit = patch("gitea_audit.write_event").start()
# gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["read", "merge", "edit", "close", "gitea.pr.close", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
def tearDown(self):
patch.stopall()
@@ -2185,8 +1987,6 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
if method == "GET" and url.endswith("/reviews"):
return [_formal_review("reviewer", "APPROVED", sha="sha123")]
if method == "GET" and "pulls/1" in url and "/files" not in url:
return {
"user": {"login": "author"},
@@ -2219,8 +2019,6 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "merger"}
if method == "GET" and url.endswith("/reviews"):
return [_formal_review("reviewer", "APPROVED", sha="sha123")]
if method == "GET" and "pulls/1" in url and "/files" not in url:
return {
"user": {"login": "author"},
@@ -2940,41 +2738,26 @@ class TestVerifyMutationAuthority(unittest.TestCase):
class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
@staticmethod
def _clean_master_git_state():
return {"current_branch": "master", "porcelain_status": ""}
def tearDown(self):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_success(self, _auth, mock_api, _git_state):
def test_lock_issue_success(self, _auth, mock_api):
mock_api.return_value = [] # no open PRs
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
lock = json.load(f)
self.assertIn("worktree_path", lock)
self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
def test_lock_issue_mismatch_branch_fails(self):
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
self.assertIn("must contain locked issue pattern", str(ctx.exception))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api, _git_state):
def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
mock_api.return_value = [{
"number": 200,
"head": {"ref": "feat/issue-196-boundary"},
@@ -2985,13 +2768,9 @@ class TestIssueLocking(unittest.TestCase):
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already tied to an open PR", str(ctx.exception))
@patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
)
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api, _git_state):
def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
mock_api.return_value = [{
"number": 200,
"head": {"ref": "feat/other-branch"},
@@ -3002,290 +2781,35 @@ class TestIssueLocking(unittest.TestCase):
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already tied to an open PR", str(ctx.exception))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_from_clean_scratch_worktree(self, _auth, _api):
scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
with patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={"current_branch": "master", "porcelain_status": ""},
) as mock_git:
res = gitea_lock_issue(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
remote="prgs",
worktree_path=scratch,
)
self.assertTrue(res["success"])
self.assertEqual(res["worktree_path"], os.path.realpath(scratch))
mock_git.assert_called_once_with(os.path.realpath(scratch))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_fails_when_declared_worktree_dirty(self, _auth, _api):
with patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "master",
"porcelain_status": " M gitea_mcp_server.py\n",
},
):
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
remote="prgs",
worktree_path="/tmp/scratch/wt",
)
self.assertIn("tracked file edits exist before issue lock", str(ctx.exception))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_fails_when_declared_worktree_not_on_base(self, _auth, _api):
with patch(
"mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "feat/issue-243-forbidden-git-gaps",
"porcelain_status": "",
},
):
with self.assertRaises(RuntimeError) as ctx:
gitea_lock_issue(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
remote="prgs",
worktree_path="/tmp/scratch/wt",
)
self.assertIn("issue lock must be taken from base branch", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_lock_fails(self, _auth, _role):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
def test_create_pr_missing_lock_fails(self, _auth):
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("Issue lock is missing", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
def test_create_pr_branch_mismatch_fails(self, _auth):
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
self.assertIn("does not match locked branch", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
for term in ("equivalent to #196", "related to #196", "same as #196"):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("contains forbidden term", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
def test_create_pr_forbidden_terms_fails(self, _auth):
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
for term in ("equivalent to #196", "related to #196", "same as #196"):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("contains forbidden term", str(ctx.exception))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_closes_ref_fails(self, _auth):
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_worktree_mismatch_fails(self, _auth, _role):
scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-pr")
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(
title="feat: lock scratch worktree Closes #249",
head="feat/issue-249-issue-lock-scratch-worktree",
remote="prgs",
worktree_path="/tmp/other-scratch",
)
self.assertIn("does not match locked worktree", str(ctx.exception))
@patch("mcp_server.api_request")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_honors_scratch_worktree_lock(self, _auth, _role, mock_api):
scratch = os.path.realpath("/tmp/gitea-tools-author-scratch/issue-249-e2e")
mock_api.return_value = {"number": 250, "html_url": "https://example/pr/250"}
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=249,
branch_name="feat/issue-249-issue-lock-scratch-worktree",
worktree_path=scratch,
), f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
res = gitea_create_pr(
title="feat: issue-lock scratch worktree Closes #249",
head="feat/issue-249-issue-lock-scratch-worktree",
remote="prgs",
worktree_path=scratch,
)
self.assertEqual(res["number"], 250)
# ---------------------------------------------------------------------------
# Pre-flight ordering and workspace edit block (#210)
# ---------------------------------------------------------------------------
class TestPreflightVerification(unittest.TestCase):
"""Assert that pre-flight verification gates order correctly."""
def setUp(self):
# Save real global variables
import mcp_server
self.orig_whoami_called = mcp_server._preflight_whoami_called
self.orig_capability_called = mcp_server._preflight_capability_called
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
self.orig_capability_violation = mcp_server._preflight_capability_violation
self.orig_resolved_role = mcp_server._preflight_resolved_role
self.orig_process_start = mcp_server._process_start_porcelain
self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
self.orig_capability_files = mcp_server._preflight_capability_violation_files
self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
# Reset state for each test
mcp_server._preflight_whoami_called = False
mcp_server._preflight_capability_called = False
mcp_server._preflight_whoami_violation = False
mcp_server._preflight_capability_violation = False
mcp_server._preflight_resolved_role = None
mcp_server._process_start_porcelain = ""
mcp_server._preflight_whoami_baseline_porcelain = None
mcp_server._preflight_capability_baseline_porcelain = None
mcp_server._preflight_whoami_violation_files = []
mcp_server._preflight_capability_violation_files = []
mcp_server._preflight_reviewer_violation_files = []
os.environ["GITEA_TEST_PORCELAIN"] = ""
def tearDown(self):
# Restore real global variables
import mcp_server
mcp_server._preflight_whoami_called = self.orig_whoami_called
mcp_server._preflight_capability_called = self.orig_capability_called
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
mcp_server._preflight_capability_violation = self.orig_capability_violation
mcp_server._preflight_resolved_role = self.orig_resolved_role
mcp_server._process_start_porcelain = self.orig_process_start
mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
mcp_server._preflight_capability_violation_files = self.orig_capability_files
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
if key in os.environ:
del os.environ[key]
def test_preflight_whoami_violation(self):
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server.record_preflight_check("whoami")
self.assertTrue(mcp_server._preflight_whoami_violation)
mcp_server.record_preflight_check("capability", resolved_role="author")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
self.assertIn("Offending files:", str(ctx.exception))
def test_preflight_capability_violation(self):
import mcp_server
mcp_server.record_preflight_check("whoami")
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server.record_preflight_check("capability", resolved_role="author")
self.assertTrue(mcp_server._preflight_capability_violation)
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn(
"Workspace file edits occurred before gitea_resolve_task_capability verification",
str(ctx.exception),
)
self.assertIn("Offending files:", str(ctx.exception))
def test_preflight_not_called_fails_closed(self):
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
mcp_server._preflight_whoami_called = True
with self.assertRaises(RuntimeError) as ctx2:
mcp_server.verify_preflight_purity()
self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))
def test_preflight_reviewer_mutation_violation(self):
import mcp_server
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="reviewer")
# Session-owned reviewer edits after capability are blocked.
os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
self.assertIn("reviewer_edit.py", str(ctx.exception))
# Foreign pre-existing dirty state does not block when unchanged.
os.environ["GITEA_TEST_PORCELAIN"] = ""
mcp_server.verify_preflight_purity()
def test_foreign_workspace_edits_do_not_block_clean_reviewer(self):
"""#252: concurrent-session dirt in the shared worktree is not attributed."""
import mcp_server
mcp_server._process_start_porcelain = " M foreign_author.py\n"
os.environ["GITEA_TEST_PORCELAIN"] = " M foreign_author.py\n"
mcp_server.record_preflight_check("whoami")
self.assertFalse(mcp_server._preflight_whoami_violation)
mcp_server.record_preflight_check("capability", resolved_role="reviewer")
self.assertFalse(mcp_server._preflight_capability_violation)
mcp_server.verify_preflight_purity()
def test_fresh_whoami_clears_sticky_violation(self):
"""#252: re-running whoami re-evaluates instead of replaying sticky state."""
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server.record_preflight_check("whoami")
self.assertTrue(mcp_server._preflight_whoami_violation)
del os.environ["GITEA_TEST_FORCE_DIRTY"]
os.environ["GITEA_TEST_PORCELAIN"] = ""
mcp_server.record_preflight_check("whoami")
self.assertFalse(mcp_server._preflight_whoami_violation)
mcp_server.record_preflight_check("capability", resolved_role="reviewer")
mcp_server.verify_preflight_purity()
def test_runtime_context_matches_preflight_block(self):
"""#252: safe_next_action must not claim ready while pre-flight blocks."""
import mcp_server
status = mcp_server.assess_preflight_status()
self.assertFalse(status["preflight_ready"])
self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])
-181
View File
@@ -1,181 +0,0 @@
"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
import os
import sys
import json
import tempfile
import unittest
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_config
import gitea_auth
import mcp_server
from reviewer_worktree import assess_author_worktree_continuity
CONFIG_TEST = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {
"enabled": True,
"base_url": "https://gitea.example.com"
}
}
},
"profiles": {
"author-profile": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
"execution_profile": "author-profile"
},
"reviewer-profile": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "reviewer-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"],
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
"execution_profile": "reviewer-profile"
}
}
}
class TestOperationScopedRoles(unittest.TestCase):
def setUp(self):
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
"dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
"prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
})
self._remotes_patch.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
self._write_config(CONFIG_TEST)
def tearDown(self):
self._remotes_patch.stop()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
self._dir.cleanup()
def _write_config(self, obj):
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(obj))
def _env(self, profile="author-profile", with_reviewer_token=True):
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
}
if with_reviewer_token:
env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass"
return env
@patch("mcp_server.api_request")
def test_auto_switch_to_reviewer(self, mock_api):
# mock identity resolution to return username matching profile
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("author-profile")):
# initially we are author-profile
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
# resolve a reviewer task (review_pr)
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
# verify it automatically switched to reviewer-profile
self.assertTrue(res["allowed_in_current_session"])
self.assertTrue(res["available_in_session"])
self.assertEqual(res["active_profile"], "reviewer-profile")
self.assertEqual(res["active_identity"], "reviewer-user")
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
@patch("mcp_server.api_request")
def test_auto_switch_to_author(self, mock_api):
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("reviewer-profile")):
# initially we are reviewer-profile
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
# resolve an author task (create_issue)
res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
# verify it automatically switched to author-profile
self.assertTrue(res["allowed_in_current_session"])
self.assertTrue(res["available_in_session"])
self.assertEqual(res["active_profile"], "author-profile")
self.assertEqual(res["active_identity"], "author-user")
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
@patch("mcp_server.api_request")
def test_restart_required_when_unattached(self, mock_api):
mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
# launch without reviewer token in env
with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
# resolve a reviewer task (review_pr)
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
# verify it did NOT switch and reports restart_required
self.assertFalse(res["allowed_in_current_session"])
self.assertFalse(res["available_in_session"])
self.assertTrue(res["configured"])
self.assertTrue(res["restart_required"])
self.assertTrue(res["stop_required"])
self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
def test_author_continuity_dirty_worktree(self):
# author is allowed to keep dirty worktree
res = assess_author_worktree_continuity({
"task_role": "author",
"dirty_files": ["review_proofs.py"],
})
self.assertTrue(res["allowed"])
# reviewer is blocked by reviewer worktree proof
res2 = assess_author_worktree_continuity({
"task_role": "reviewer",
"worktree_path": "/repo",
"dirty_files": ["review_proofs.py"],
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": False,
})
self.assertFalse(res2["proven"])
@patch("mcp_server.api_request")
def test_mutating_actions_auto_switch(self, mock_api):
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("author-profile")):
# verify we are author-profile
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
# call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs")
self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block)
# verify we dynamically switched to reviewer-profile
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
if __name__ == "__main__":
unittest.main()
+7 -21
View File
@@ -132,8 +132,7 @@ class TestControlPlaneGuide(GuideTestBase):
rules = g["rules"]
for key in ("hard_stops", "fail_closed", "head_sha_pinning",
"merge_confirmation", "redaction", "separation",
"profile_switching", "identity_verification",
"work_selection", "global_worktree"):
"profile_switching", "identity_verification"):
self.assertIn(key, rules)
self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"]))
self.assertTrue(rules["hard_stops"])
@@ -235,8 +234,8 @@ class TestProjectSkills(GuideTestBase):
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
r = mcp_list_project_skills()
by_name = {s["name"]: s for s in r["skills"]}
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
self.assertEqual(by_name["glitchtip-mcp"]["status"], "external-mcp")
self.assertNotEqual(by_name["jenkins-mcp"]["status"], "available")
self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available")
def test_no_urls_in_registry(self):
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
@@ -247,29 +246,16 @@ class TestProjectSkills(GuideTestBase):
self.assertNotIn("keychain:", blob)
def test_enabled_but_no_usable_tools_negative_assertion(self):
"""Negative assertion for 'enabled but no usable tools' (per issue #151)."""
"""Negative assertion for 'enabled but no usable tools' (per issue #146)."""
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
r = mcp_list_project_skills()
by_name = {s["name"]: s for s in r["skills"]}
# jenkins-mcp is an external boundary; Gitea profile discovery must not
# treat a local config entry as a usable Jenkins tool surface.
# jenkins-mcp is designed-not-implemented; even if "enabled" in config,
# it should not be usable/available to current profile without tools.
self.assertIn("jenkins-mcp", by_name)
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
self.assertEqual(by_name["jenkins-mcp"]["status"], "designed-not-implemented")
self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False))
def test_external_mcp_skill_guides_name_expected_tools(self):
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
jenkins = mcp_get_skill_guide("jenkins-mcp")
glitchtip = mcp_get_skill_guide("glitchtip-mcp")
self.assertEqual(jenkins["skill"]["status"], "external-mcp")
self.assertIn("jenkins_whoami", " ".join(jenkins["steps"]))
self.assertIn("jenkins_get_build", " ".join(jenkins["steps"]))
self.assertIn("SKIPPED", " ".join(jenkins["steps"]))
self.assertEqual(glitchtip["skill"]["status"], "external-mcp")
self.assertIn("glitchtip_whoami", " ".join(glitchtip["steps"]))
self.assertIn("glitchtip_search", " ".join(glitchtip["steps"]))
self.assertIn("SKIPPED", " ".join(glitchtip["steps"]))
# ---------------------------------------------------------------------------
# mcp_get_skill_guide
+6 -89
View File
@@ -119,28 +119,12 @@ class TestPRQueueInventory(unittest.TestCase):
mock_get_all.return_value = [
{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
]
# mock_api: inventory whoami, eligibility whoami, eligibility PR,
# POST review (#244: state + visible-verdict GET reviews).
# mock_api: 1) /user (inventory), 2) /user (eligibility), 3) /pulls/1 (eligibility), 4) /pulls/1/reviews (POST review)
mock_api.side_effect = [
{"login": "reviewer1"},
{"login": "reviewer1"},
{
"user": {"login": "other_user"},
"state": "open",
"head": {"sha": "abc1"},
"mergeable": True,
},
{"id": 100, "state": "APPROVED"},
[
{
"id": 100,
"user": {"login": "reviewer1"},
"state": "APPROVED",
"commit_id": "abc1",
"submitted_at": "2026-07-06T10:00:00Z",
"dismissed": False,
}
],
{"login": "reviewer1"}, # inventory whoami
{"login": "reviewer1"}, # submit eligibility whoami
{"user": {"login": "other_user"}, "state": "open", "head": {"sha": "abc1"}, "mergeable": True}, # submit eligibility PR
{"id": 100}, # POST review
]
from mcp_server import init_review_decision_lock, gitea_mark_final_review_decision
@@ -277,79 +261,12 @@ class TestPRQueueInventory(unittest.TestCase):
for call in mock_api.call_args_list:
self.assertEqual(call.args[0], "GET")
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = None
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertFalse(result["success"])
msg = result["message"]
self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
self.assertIn("Empty-queue claim blocked", msg)
self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_empty_inventory_trusted_empty_when_gate_passes(
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
)
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertFalse(result["success"])
msg = result["message"]
self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
"""Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
mock_local_url.return_value = (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
)
for event in ["APPROVE", "REQUEST_CHANGES"]:
mock_get_profile.return_value = {
"profile_name": "gitea-author",
-11
View File
@@ -130,17 +130,6 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertIn("author-profile", res["matching_configured_profile"])
self.assertIn("reviewer-profile", res["matching_configured_profile"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_mark_issue_author_profile_allowed(self, _auth, _api):
# #183: mark_issue must map to gitea.issue.comment (prgs-author allowed op).
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_resolve_task_capability(task="mark_issue", remote="prgs")
self.assertEqual(res["requested_task"], "mark_issue")
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
self.assertTrue(res["allowed_in_current_session"])
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
def test_resolve_unknown_task_fails_closed(self):
with patch.dict(os.environ, self._env("author-profile")):
with self.assertRaises(ValueError):
+6 -653
View File
@@ -21,32 +21,17 @@ import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from review_proofs import ( # noqa: E402
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_capability_evidence,
assess_capability_proof,
assess_contradictory_no_pr_claim,
assess_continuation_mode_report,
assess_controller_handoff,
assess_edited_pr_inventory_coverage,
assess_empty_queue_report,
assess_fresh_issue_selection,
assess_inventory_completeness,
assess_issue_selection_final_report,
assess_queue_target_final_report,
assess_reviewer_queue_inventory,
assess_live_state_recheck,
assess_review_mutation_final_report,
assess_role_boundary,
assess_secret_sweep,
assess_self_review_contamination,
assess_sweep_evidence,
assess_validation_report,
build_final_report,
classify_issue_for_selection,
pr_inventory_trust_gate,
reconcile_queue_target,
resolve_repos_from_user_reference,
verify_pinned_head_checkout,
)
@@ -195,24 +180,6 @@ def _good_review_mutation():
return assess_review_mutation_final_report(report, lock)
def _good_worktree(**overrides):
proof = {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
"git_commands": [
"git fetch prgs master feat/issue-224-wiki-proof-refresh",
"git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
],
}
proof.update(overrides)
from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402
return assess_reviewer_worktree_proof(proof)
def _good_role_boundary_179(**overrides):
kwargs = {
"task_role": "reviewer",
@@ -223,50 +190,6 @@ def _good_role_boundary_179(**overrides):
return assess_role_boundary(**kwargs)
GOOD_HANDOFF = "\n".join([
"## Controller Handoff",
"",
"- Task: review PR #999",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: reviewer",
"- Identity: sysadmin / prgs-reviewer",
"- Issue/PR: #182 / PR #999",
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: review only",
"- Workspace mutations: none",
"- Current status: PR open",
"- Blockers: none",
"- Next: merge if approved",
"- Safety: no self-review; no self-merge; no secrets",
])
def _good_handoff():
return assess_controller_handoff(GOOD_HANDOFF)
def _good_capability_proof():
return assess_capability_proof({
"comment_issue": {
"requested_task": "comment_issue",
"required_operation_permission": "gitea.issue.comment",
"allowed_in_current_session": True,
}
})
def _good_secret_sweep(**overrides):
sweep = {
"method": "git diff prgs/master...HEAD | grep -iE 'token|secret'",
"scope": "full PR diff relative to prgs/master",
"clean": True,
}
sweep.update(overrides)
return assess_secret_sweep(sweep)
class TestCheckoutProof(unittest.TestCase):
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
@@ -622,16 +545,6 @@ class TestFinalReport(unittest.TestCase):
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
"review_mutation": _good_review_mutation(),
"controller_handoff": _good_handoff(),
"capability_proof": _good_capability_proof(),
"sweep_proof": _good_secret_sweep(),
"worktree_proof": {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
},
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -749,28 +662,6 @@ class TestFinalReport(unittest.TestCase):
self.assertEqual(report["grade"], "blocked")
self.assertFalse(report["merge_allowed"])
def test_failed_capability_proof_downgrades(self):
report = self._report(
capability_proof={
"proven": False,
"reasons": ["task mark_issue not allowed"],
}
)
self.assertNotEqual(report["grade"], "A")
self.assertTrue(
any("capability" in r for r in report["downgrade_reasons"])
)
def test_failed_sweep_proof_downgrades(self):
report = self._report(
sweep_proof={
"proven": False,
"reasons": ["method missing"],
}
)
self.assertNotEqual(report["grade"], "A")
self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
class TestStdoutIsolation(unittest.TestCase):
"""Regression test for #178: tests must not close or corrupt stdout/stderr
@@ -898,7 +789,6 @@ class TestControllerHandoff(unittest.TestCase):
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: one PR opened",
"- Workspace mutations: none",
"- Current status: PR open",
"- Blockers: none",
"- Next: review PR #999",
@@ -936,17 +826,12 @@ class TestControllerHandoff(unittest.TestCase):
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Pinned reviewed head", result["missing_fields"])
self.assertIn("Worktree path", result["missing_fields"])
self.assertIn("Merge result", result["missing_fields"])
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Worktree path: /repo/branches/review-pr-999",
"- Worktree dirty: no",
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
"- Unrelated local mutations: none",
"- Review decision: approve",
"- Merge result: merged",
"- Linked issue status: closed",
@@ -955,46 +840,24 @@ class TestControllerHandoff(unittest.TestCase):
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
def _author_role_fields(self, issue_number=182, pr_number=999):
return [
f"- Selected issue: #{issue_number}",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
f"- PR number opened: #{pr_number}",
"- No review/merge: confirmed",
]
def test_handoff_role_fields_author_includes_issue_lock_proof(self):
from review_proofs import HANDOFF_ROLE_FIELDS
names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
self.assertIn("Issue lock proof", names)
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_requires_issue_lock_proof(self):
without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #182",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
result = assess_controller_handoff(without_lock, role="author")
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_rejects_equivalent_or_multiple_issues(self):
# 1. equivalent reference blocked
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: Issue #194 / #196 equivalent",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -1006,7 +869,6 @@ class TestControllerHandoff(unittest.TestCase):
# 2. multiple issues blocked
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #194, #196",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -1018,7 +880,6 @@ class TestControllerHandoff(unittest.TestCase):
def test_author_role_rejects_fuzzy_pr_number(self):
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: PR #203 / #204 equivalent",
"- No review/merge: confirmed",
@@ -1031,10 +892,6 @@ class TestControllerHandoff(unittest.TestCase):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Repositories checked: Gitea-Tools, mcp-control-plane",
"- Open PR counts: 2 / 0",
"- PR inventory trust gate: trusted_nonempty / trusted_empty",
"- Trust gate reasons: none",
"- Trust gate corroborated: true",
"- Inventory profile: prgs-reviewer",
"- Selected PR or reason: none eligible (self-authored)",
"- Inventory completeness: complete, no pagination needed",
])
@@ -1051,27 +908,6 @@ class TestControllerHandoff(unittest.TestCase):
self.assertIn("## Controller Handoff", skill)
self.assertIn("assess_controller_handoff", skill)
self.assertIn("issue #182", skill)
self.assertIn("## Work Selection Rule for LLMs", skill)
self.assertIn("work already claimed", skill)
self.assertIn("## Global LLM Worktree Rule", skill)
self.assertIn("branches/", skill)
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
# 1. Workspace mutations: none is rejected when local_edits is True
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
self._author_role_fields(issue_number=196, pr_number=203))
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Workspace mutations", res["missing_fields"])
# 2. Workspace mutations: edited files is allowed when local_edits is True
complete_eq = self.BASE_HANDOFF.replace(
"- Workspace mutations: none",
"- Workspace mutations: edited review_proofs.py",
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
class TestReviewMutationFinalReport(unittest.TestCase):
@@ -1149,144 +985,6 @@ class TestReviewMutationFinalReport(unittest.TestCase):
self.assertTrue(final["review_mutation_complete"])
class TestQueueTargetReconciliation(unittest.TestCase):
"""Queue target lock: reconcile operator-supplied backlog before inventory (#200)."""
CONFIGURED = [
"Scaled-Tech-Consulting/Gitea-Tools",
"Scaled-Tech-Consulting/mcp-control-plane",
]
GITEA_TOOLS = "Scaled-Tech-Consulting/Gitea-Tools"
MCP = "Scaled-Tech-Consulting/mcp-control-plane"
OPERATOR_CONTEXT = (
"six open PRs in Scaled-Tech-Consulting/Gitea-Tools including "
"#195, #193, #192, #190, #187, and #181"
)
PROFILE = {
"profile_name": "prgs-reviewer",
"allowed_operations": ["read", "gitea.read"],
}
def test_supplied_gitea_tools_prs_but_inventoried_mcp_is_mismatch(self):
lock = reconcile_queue_target(
operator_context=self.OPERATOR_CONTEXT,
inventoried_repo=self.MCP,
configured_repos=self.CONFIGURED,
)
self.assertEqual(lock["status"], "target_repo_mismatch")
self.assertEqual(lock["resolved_repo"], self.GITEA_TOOLS)
self.assertEqual(lock["resolution_source"], "operator_context")
self.assertIn(195, lock["supplied_pr_numbers"])
self.assertFalse(lock["allow_clean_stop"])
self.assertFalse(lock["allow_trusted_empty"])
def test_wrong_repo_zero_open_cannot_stop_cleanly(self):
lock = reconcile_queue_target(
operator_context=self.OPERATOR_CONTEXT,
inventoried_repo=self.MCP,
configured_repos=self.CONFIGURED,
)
gate = pr_inventory_trust_gate(
[],
remote="prgs",
org="Scaled-Tech-Consulting",
repo="mcp-control-plane",
state="open",
authenticated_profile=self.PROFILE,
local_remote_url=(
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
"mcp-control-plane.git"
),
corroboration_open_pr_counter=0,
queue_target_lock=lock,
)
self.assertEqual(gate["status"], "target_repo_mismatch")
self.assertNotEqual(gate["status"], "trusted_empty")
def test_supplied_pr_numbers_reconciled_before_empty_stop(self):
lock = reconcile_queue_target(
operator_context=(
"currently we have 6 open PRs: #195, #193, #192, "
"#190, #187, #181 in Gitea-Tools"
),
inventoried_repo=self.GITEA_TOOLS,
configured_repos=self.CONFIGURED,
)
self.assertEqual(lock["status"], "resolved")
self.assertEqual(len(lock["supplied_pr_numbers"]), 6)
self.assertTrue(all(item["matches_inventoried_repo"]
for item in lock["reconciliation"]))
def test_ambiguous_repo_context_fails_closed(self):
lock = reconcile_queue_target(
operator_context=(
"open PRs in mcp-control-plane and gitea-tools including PR #195"
),
inventoried_repo=self.MCP,
configured_repos=self.CONFIGURED,
)
self.assertEqual(lock["status"], "unresolved")
self.assertFalse(lock["allow_trusted_empty"])
self.assertTrue(
any("could not be resolved" in r for r in lock["reasons"])
)
def test_conflicting_directive_vs_backlog_fails_closed(self):
lock = reconcile_queue_target(
operator_context="Repository: Scaled-Tech-Consulting/mcp-control-plane",
supplied_pr_backlog=[
{"number": 195, "repo": self.GITEA_TOOLS},
],
inventoried_repo=self.MCP,
configured_repos=self.CONFIGURED,
)
self.assertEqual(lock["status"], "unresolved")
self.assertIn("conflicts", " ".join(lock["reasons"]).lower())
def test_final_report_must_document_reconciliation(self):
lock = reconcile_queue_target(
operator_context=self.OPERATOR_CONTEXT,
inventoried_repo=self.GITEA_TOOLS,
configured_repos=self.CONFIGURED,
)
incomplete = assess_queue_target_final_report(
"Open PRs: 0. Stopping.", lock
)
self.assertFalse(incomplete["complete"])
self.assertTrue(incomplete["downgraded"])
complete_report = "\n".join([
"Queue inventory complete.",
f"Resolved repo: {self.GITEA_TOOLS}",
"Resolution source: operator_context",
"queue_target_lock.status: resolved",
"Supplied PR reconciliation: #195, #193, #192, #190, #187, #181",
])
complete = assess_queue_target_final_report(complete_report, lock)
self.assertTrue(complete["complete"])
self.assertFalse(complete["downgraded"])
def test_assess_reviewer_queue_inventory_blocks_mismatch(self):
result = assess_reviewer_queue_inventory(
[{
"repo": self.MCP,
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 0,
"list_prs_response": [],
"remote": "prgs",
"authenticated_profile": self.PROFILE,
"local_remote_url": (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
"mcp-control-plane.git"
),
}],
operator_context=self.OPERATOR_CONTEXT,
)
self.assertFalse(result["can_claim_empty_queue"])
self.assertIn("target_repo_mismatch", str(result["trust_gates"]))
class TestPRInventoryTrustGate(unittest.TestCase):
"""Issue #194: unit tests for the PR inventory trust gate."""
@@ -1369,157 +1067,6 @@ class TestPRInventoryTrustGate(unittest.TestCase):
self.assertTrue(res["corroborated"])
class TestAssessReviewerQueueInventory(unittest.TestCase):
"""Issue #196: trust gate wired into canonical queue inventory."""
def _repo_report(self, **overrides):
report = {
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 0,
"list_prs_response": [],
"remote": "prgs",
"authenticated_profile": {
"profile_name": "prgs-reviewer",
"allowed_operations": ["read", "gitea.read"],
},
"local_remote_url": (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
),
}
report.update(overrides)
return report
def test_empty_without_corroboration_blocks_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(pagination_complete=False),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=0,
pagination_complete=False,
),
])
self.assertFalse(result["can_claim_empty_queue"])
self.assertIn("untrusted_empty", str(result["trust_gates"]))
def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
local_remote_url=(
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
"mcp-control-plane.git"
),
),
])
self.assertTrue(result["can_claim_empty_queue"])
self.assertTrue(result["can_claim_exhaustive"])
def test_user_context_indicating_open_prs_blocks_empty_claim(self):
result = assess_reviewer_queue_inventory(
[self._repo_report()],
user_context="please review open PR #195 in the queue",
)
self.assertFalse(result["can_claim_empty_queue"])
self.assertTrue(result["blockers"])
def test_nonempty_inventory_skips_empty_trust_gate_block(self):
result = assess_reviewer_queue_inventory([
self._repo_report(open_pr_count=2),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=1,
),
])
self.assertTrue(result["complete"])
self.assertEqual(result["trust_gates"], {})
class TestAssessEmptyQueueReport(unittest.TestCase):
"""Issue #198: empty-queue reports require formal trust-gate proof."""
def _trusted_report(self, **extra):
lines = [
"Queue inventory complete.",
"Repository: Scaled-Tech-Consulting/Gitea-Tools",
"Open PR count: 0",
"pr_inventory_trust_gate.status: trusted_empty",
"pr_inventory_trust_gate.corroborated: true",
"Inventory profile: prgs-reviewer",
"Workflow correctly stops with nothing to review.",
]
lines.extend(extra)
return "\n".join(lines)
def test_non_empty_report_not_claimed(self):
result = assess_empty_queue_report("Reviewed PR #236 and merged.")
self.assertFalse(result["claimed"])
self.assertTrue(result["proven"])
def test_empty_claim_without_trust_gate_blocked(self):
report = (
"Open PR count: 0\n"
"Pagination complete: yes\n"
"Queue cleared."
)
result = assess_empty_queue_report(report)
self.assertTrue(result["claimed"])
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_trusted_empty_report_with_required_fields_passes(self):
result = assess_empty_queue_report(self._trusted_report())
self.assertTrue(result["proven"])
def test_weak_merge_commit_corroboration_blocked(self):
report = (
"Open PR count: 0\n"
"Master latest commit is merge of PR #79 so queue is empty."
)
result = assess_empty_queue_report(report)
self.assertFalse(result["proven"])
self.assertTrue(
any("weak corroboration" in r for r in result["reasons"])
)
def test_author_session_reviewer_queue_wording_blocked(self):
result = assess_empty_queue_report(
self._trusted_report(),
task_role="author",
)
self.assertFalse(result["proven"])
def test_build_final_report_downgrades_weak_empty_queue(self):
final = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
review_mutation=_good_review_mutation(),
controller_handoff=_good_handoff(),
capability_proof=_good_capability_proof(),
sweep_proof=_good_secret_sweep(),
worktree_proof={
"worktree_path": "/repo/branches/review-pr-1",
"porcelain_status": "",
"scratch_used": True,
"scratch_path": "/repo/branches/review-pr-1",
},
report_text="Open PR count: 0. Queue cleared.",
)
self.assertNotEqual(final["grade"], "A")
self.assertFalse(final["empty_queue_trust_gate_proven"])
class TestCapabilityEvidence(unittest.TestCase):
"""#179 gap 1: capability claims need exact evidence."""
@@ -1660,16 +1207,6 @@ class TestFinalReport179Bar(unittest.TestCase):
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
"review_mutation": _good_review_mutation(),
"controller_handoff": _good_handoff(),
"capability_proof": _good_capability_proof(),
"sweep_proof": _good_secret_sweep(),
"worktree_proof": {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
},
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -1745,189 +1282,5 @@ class TestFinalReport179Bar(unittest.TestCase):
self.assertFalse(report["review_mutation_complete"])
class TestAuthorReporting(unittest.TestCase):
"""Harness assertions for author reporting and capability/sweep proofs."""
def test_good_capability_proof_passes(self):
result = _good_capability_proof()
self.assertTrue(result["proven"])
def test_missing_capability_proof_fails_closed(self):
result = assess_capability_proof({})
self.assertFalse(result["proven"])
self.assertTrue(any("fail closed" in r for r in result["reasons"]))
def test_unresolved_capability_proof_fails_closed(self):
result = assess_capability_proof({
"mark_issue": {
"requested_task": "unknown_task",
"required_operation_permission": None,
"allowed_in_current_session": None,
}
})
self.assertFalse(result["proven"])
self.assertTrue(
any("could not be resolved" in r for r in result["reasons"])
)
def test_good_secret_sweep_passes(self):
result = _good_secret_sweep()
self.assertTrue(result["proven"])
self.assertEqual(result["verdict"], "strong")
def test_vague_secret_sweep_without_method_is_weak(self):
result = _good_secret_sweep(method="")
self.assertFalse(result["proven"])
self.assertEqual(result["verdict"], "weak")
self.assertTrue(
any("method" in r or "scan" in r for r in result["reasons"])
)
def test_unconfirmed_secret_sweep_is_weak(self):
result = _good_secret_sweep(clean=False)
self.assertFalse(result["proven"])
self.assertEqual(result["verdict"], "weak")
def test_good_author_pr_report_passes(self):
result = assess_author_pr_report({
"pr_number": 185,
"branch": "feat/issue-184-repo-name-disambiguation",
"head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae",
})
self.assertTrue(result["complete"])
def test_author_pr_report_without_head_sha_is_incomplete(self):
result = assess_author_pr_report({
"pr_number": 185,
"branch": "feat/issue-184-repo-name-disambiguation",
"head_sha": "",
})
self.assertFalse(result["complete"])
self.assertTrue(any("head SHA" in r for r in result["reasons"]))
def test_author_pr_report_rejects_short_sha(self):
result = assess_author_pr_report({
"pr_number": 185,
"branch": "feat/issue-184-repo-name-disambiguation",
"head_sha": "e2bccba",
})
self.assertFalse(result["complete"])
class TestIssueSelectionContinuation(unittest.TestCase):
"""Issue #188: continuation mode wall for issues with open PRs."""
OLD_SHA = PINNED
NEW_SHA = OTHER
OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
def test_open_pr_issue_excluded_from_fresh_selection(self):
classified = classify_issue_for_selection(
183, open_prs=self.OPEN_PR,
)
self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
self.assertFalse(classified["selectable_for_fresh_work"])
blocked = assess_fresh_issue_selection([classified])
self.assertTrue(blocked["downgraded"])
def test_explicit_continuation_allows_represented_issue(self):
classified = classify_issue_for_selection(
183,
open_prs=self.OPEN_PR,
operator_continuation_requested=True,
)
self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
blocked = assess_fresh_issue_selection([classified])
self.assertFalse(blocked["downgraded"])
def test_contradictory_no_pr_claim_downgrades(self):
report = (
"Selected issue #183; no duplicate PR open. "
"Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
)
result = assess_contradictory_no_pr_claim(
report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
)
self.assertTrue(result["downgraded"])
def test_edited_pr_must_appear_in_inventory(self):
report = "Open PR inventory: PR #195 only."
result = assess_edited_pr_inventory_coverage(
report,
edited_pr_numbers=[187],
inventoried_pr_numbers=[195],
)
self.assertTrue(result["downgraded"])
def test_continuation_report_requires_old_and_new_head(self):
report = (
"Issue #182 continuation mode. PR #186. "
f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
"Session authored PR: yes. Continuation allowed: operator requested."
)
result = assess_continuation_mode_report(
report,
pr_number=186,
pr_author="jcwalker3",
branch="feat/issue-182-controller-handoff",
old_head_sha=self.OLD_SHA,
new_head_sha=self.NEW_SHA,
session_authored_pr=True,
continuation_allowed_reason="operator requested continuation",
)
self.assertTrue(result["complete"])
def test_issue_selection_final_report_continuation_earns_a(self):
report = "\n".join([
"Issue #182 continuation mode — no new issue claimed.",
f"PR #186 updated: old head {self.OLD_SHA}, "
f"new head {self.NEW_SHA}.",
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
"Session authored PR: yes.",
"Continuation allowed: operator requested rebase.",
"Open PR inventory included PR #186.",
"## Controller Handoff",
"- Task: continuation",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: prgs-author",
"- Issue/PR: #182 / PR #186",
"- Branch/SHA: feat/issue-182-controller-handoff",
"- Files changed: review_proofs.py",
"- Validation: tests passed",
"- Mutations: push_branch",
"- Workspace mutations: none",
"- Current status: PR mergeable",
"- Blockers: none",
"- Next: review",
"- Safety: no review/merge",
"- Continuation mode: issue #182 continuation",
"- Existing PR: #186",
"- PR author: jcwalker3",
"- Branch: feat/issue-182-controller-handoff",
f"- Old PR head: {self.OLD_SHA}",
f"- New PR head: {self.NEW_SHA}",
"- Session authored PR: yes",
"- Why continuation allowed: operator requested rebase",
])
result = assess_issue_selection_final_report(
report,
mode="continuation",
continuation_proof={
"pr_number": 186,
"pr_author": "jcwalker3",
"branch": "feat/issue-182-controller-handoff",
"old_head_sha": self.OLD_SHA,
"new_head_sha": self.NEW_SHA,
"session_authored_pr": True,
"continuation_allowed_reason": "operator requested rebase",
},
edited_pr_numbers=[186],
inventoried_pr_numbers=[186],
)
self.assertEqual(result["grade"], "A")
if __name__ == "__main__":
unittest.main()
-183
View File
@@ -1,183 +0,0 @@
"""Tests for reviewer worktree safety proofs (Issue #233)."""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from reviewer_worktree import ( # noqa: E402
assess_author_worktree_continuity,
assess_reviewer_git_command_log,
assess_reviewer_worktree_proof,
files_outside_pr_scope,
is_forbidden_reviewer_git_command,
is_readonly_reviewer_git_command,
parse_dirty_tracked_files,
)
class TestParseDirtyTrackedFiles(unittest.TestCase):
def test_ignores_untracked_files(self):
porcelain = "?? untracked.txt\n M tracked.py\n"
self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"])
def test_parses_renamed_paths(self):
porcelain = "R old.py -> new.py\n"
self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"])
class TestFilesOutsidePrScope(unittest.TestCase):
def test_all_dirty_in_scope_is_clean(self):
self.assertEqual(
files_outside_pr_scope(
["docs/wiki/Repositories.md"],
["docs/wiki/Repositories.md"],
),
[],
)
def test_unrelated_dirty_files_detected(self):
self.assertEqual(
files_outside_pr_scope(
["review_proofs.py", "docs/wiki/Repositories.md"],
["docs/wiki/Repositories.md"],
),
["review_proofs.py"],
)
class TestForbiddenGitCommands(unittest.TestCase):
def test_blocks_stash_operations(self):
self.assertTrue(is_forbidden_reviewer_git_command("git stash"))
self.assertTrue(
is_forbidden_reviewer_git_command(
'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py'
)
)
self.assertTrue(is_forbidden_reviewer_git_command("git stash pop"))
self.assertTrue(is_forbidden_reviewer_git_command("git stash drop"))
def test_blocks_checkout_reset_and_clean(self):
self.assertTrue(
is_forbidden_reviewer_git_command(
"git checkout -- review_proofs.py tests/test_mcp_server.py"
)
)
self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard"))
self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd"))
def test_blocks_checkout_restore_and_switch_bypasses(self):
"""Issue #243: blocklist gaps closed via readonly allowlist model."""
blocked = (
"git checkout HEAD -- review_proofs.py",
"git checkout prgs/master -- review_proofs.py",
"git checkout .",
"git switch -",
"git switch -C feat/other-branch",
"git switch master",
"git stash store",
"git stash branch wip-stash",
)
for cmd in blocked:
with self.subTest(cmd=cmd):
self.assertTrue(is_forbidden_reviewer_git_command(cmd))
self.assertFalse(is_readonly_reviewer_git_command(cmd))
def test_allows_readonly_commands(self):
for cmd in (
"git fetch prgs master",
"git status --porcelain",
"git diff prgs/master...HEAD",
"git rev-parse HEAD",
"git -C /repo log -1",
):
with self.subTest(cmd=cmd):
self.assertFalse(is_forbidden_reviewer_git_command(cmd))
self.assertTrue(is_readonly_reviewer_git_command(cmd))
class TestAssessReviewerWorktreeProof(unittest.TestCase):
def test_clean_worktree_proceeds(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo/branches/review-pr-231",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": False,
"git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"],
})
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_dirty_unrelated_without_scratch_blocks(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo",
"dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"],
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": False,
})
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0])
def test_scratch_worktree_allows_dirty_main_repo(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo",
"dirty_files": ["review_proofs.py"],
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
})
self.assertTrue(result["proven"])
def test_forbidden_command_history_blocks(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo/branches/review-pr-231",
"porcelain_status": "",
"git_commands": ["git stash push -m temp -- review_proofs.py"],
})
self.assertFalse(result["proven"])
self.assertTrue(result["forbidden_commands"])
def test_unrelated_mutations_claimed_blocks(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo",
"porcelain_status": "",
"unrelated_mutations_claimed": True,
})
self.assertFalse(result["proven"])
class TestAuthorContinuity(unittest.TestCase):
def test_author_may_keep_dirty_worktree(self):
result = assess_author_worktree_continuity({
"task_role": "author",
"dirty_files": ["feat.py"],
})
self.assertTrue(result["allowed"])
def test_reviewer_dirty_worktree_uses_reviewer_gate(self):
result = assess_author_worktree_continuity({
"task_role": "reviewer",
"worktree_path": "/repo",
"dirty_files": ["other.py"],
"pr_scope_files": ["docs/a.md"],
"scratch_used": False,
})
self.assertFalse(result["proven"])
class TestAssessReviewerGitCommandLog(unittest.TestCase):
def test_empty_log_is_clean(self):
result = assess_reviewer_git_command_log([])
self.assertTrue(result["proven"])
def test_mixed_log_blocks_on_forbidden(self):
result = assess_reviewer_git_command_log([
"git fetch prgs",
"git stash",
])
self.assertFalse(result["proven"])
self.assertEqual(len(result["forbidden_commands"]), 1)
if __name__ == "__main__":
unittest.main()
-215
View File
@@ -1,215 +0,0 @@
"""Tests for reviewer/author namespace mismatch walls (#209)."""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_audit
import mcp_server
import role_namespace_gate
from review_proofs import assess_mutation_namespace_handoff
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"prgs-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.pr.create",
"gitea.branch.push", "gitea.issue.comment",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
],
"execution_profile": "prgs-author",
},
"prgs-reviewer": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "sysadmin",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": [
"gitea.read", "gitea.pr.review", "gitea.pr.approve",
"gitea.pr.merge", "gitea.issue.comment",
],
"forbidden_operations": [
"gitea.pr.create", "gitea.branch.push", "gitea.issue.create",
],
"execution_profile": "prgs-reviewer",
},
"prgs-reviewer-issue-writer": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "sysadmin",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": [
"gitea.read", "gitea.pr.review", "gitea.issue.create",
],
"forbidden_operations": ["gitea.pr.create"],
"execution_profile": "prgs-reviewer-issue-writer",
},
},
"rules": {"allow_runtime_switching": False},
}
ISSUE_WRITE_ENV = {"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create"}
class NamespaceGateBase(unittest.TestCase):
def setUp(self):
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {
"host": "gitea.example.com",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
self.audit_path = os.path.join(self._dir.name, "audit.log")
def tearDown(self):
self._remotes.stop()
mcp_server._IDENTITY_CACHE.clear()
self._dir.cleanup()
def _env(self, profile, *, audit=False):
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
}
if audit:
env["GITEA_AUDIT_LOG"] = self.audit_path
return env
class TestNamespaceInference(unittest.TestCase):
def test_reviewer_and_author_namespaces(self):
self.assertEqual(
role_namespace_gate.infer_mcp_namespace("prgs-reviewer"),
"gitea-reviewer",
)
self.assertEqual(
role_namespace_gate.infer_mcp_namespace("prgs-author"),
"gitea-author",
)
class TestReviewerAuthorMutationBlocks(NamespaceGateBase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_reviewer_namespace_create_issue_blocked(
self, _auth, mock_api, _get_all, _role
):
with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
result = mcp_server.gitea_create_issue(
title="should not post", remote="prgs")
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertTrue(result.get("namespace_block"))
mock_api.assert_not_called()
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_reviewer_namespace_create_pr_blocked(
self, _auth, mock_api, _get_all, _role
):
with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
result = mcp_server.gitea_create_pr(
title="feat: x Closes #1",
head="feat/issue-1-x",
body="Closes #1",
remote="prgs",
)
self.assertFalse(result.get("success", True))
self.assertFalse(result.get("performed", True))
self.assertTrue(result.get("namespace_block"))
mock_api.assert_not_called()
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request", return_value={"number": 12})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_reviewer_with_explicit_issue_create_allowed(
self, _auth, mock_api, _get_all, _role
):
with patch.dict(os.environ, self._env("prgs-reviewer-issue-writer"),
clear=True):
result = mcp_server.gitea_create_issue(
title="escalation issue", remote="prgs")
self.assertEqual(result.get("number"), 12)
mock_api.assert_called()
class TestHandoffNamespaceParity(unittest.TestCase):
def test_author_handoff_with_reviewer_namespace_invalid(self):
result = assess_mutation_namespace_handoff(
declared_role="author",
namespaces_used=["gitea-reviewer"],
)
self.assertFalse(result["valid"])
self.assertTrue(result["blocked"])
def test_capability_author_profile_with_reviewer_namespace_blocked(self):
result = assess_mutation_namespace_handoff(
declared_role="author",
namespaces_used=["gitea-reviewer"],
capability_profile="prgs-author",
)
self.assertFalse(result["valid"])
blob = " ".join(result["reasons"])
self.assertIn("capability proof profile", blob)
class TestMutationAuditFields(NamespaceGateBase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request", return_value={"number": 3})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_successful_create_issue_audit_includes_namespace_fields(
self, _auth, mock_api, _get_all, _role
):
env = {**self._env("prgs-author", audit=True), **ISSUE_WRITE_ENV}
with patch.dict(os.environ, env, clear=True):
mcp_server.gitea_create_issue(title="audited", remote="prgs")
with open(self.audit_path, encoding="utf-8") as fh:
record = json.loads(fh.readline())
self.assertEqual(record["mcp_namespace"], "gitea-author")
self.assertEqual(record["task_role"], "author")
self.assertEqual(record["operation"], "create_issue")
self.assertEqual(record["profile_name"], "prgs-author")
self.assertEqual(record["remote"], "prgs")
if __name__ == "__main__":
unittest.main()
-63
View File
@@ -119,25 +119,6 @@ class TestRoleSessionRouter(unittest.TestCase):
]
self.assertEqual(issue_posts, [])
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request", return_value={"number": 888})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_author_task_allowed_after_capability_resolve(
self, _auth, _api, _get_all
):
"""#228: explicit create_issue capability clears reviewer wrong_role_stop."""
with patch.dict(os.environ, self._env("prgs-author")):
mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
cap = mcp_server.gitea_resolve_task_capability(
task="create_issue", remote="prgs"
)
self.assertTrue(cap["allowed_in_current_session"])
result = mcp_server.gitea_create_issue(
title="operation-scoped author recovery",
remote="prgs",
)
self.assertEqual(result.get("number"), 888)
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request", return_value={"number": 999})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
@@ -204,49 +185,5 @@ class TestRoleSessionRouter(unittest.TestCase):
self.assertTrue(complete["complete"])
class TestCheckMidMerge(unittest.TestCase):
def test_skip_scan_walk_root_skips_sibling_worktrees_only(self):
worktree_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
main_root = os.path.dirname(os.path.dirname(worktree_root))
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
main_root, main_root
)
)
self.assertTrue(
role_session_router.skip_python_scan_walk_root(
main_root, os.path.join(main_root, "branches", "fix-issue-1")
)
)
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
worktree_root, worktree_root
)
)
self.assertFalse(
role_session_router.skip_python_scan_walk_root(
worktree_root, os.path.join(worktree_root, "tests")
)
)
def test_decorative_equals_banner_is_not_mid_merge(self):
self.assertFalse(role_session_router.check_mid_merge())
def test_python_bytes_have_conflict_markers_rejects_decorative_equals(self):
banner = b"===========================================\n"
self.assertFalse(role_session_router.python_bytes_have_conflict_markers(banner))
def test_python_bytes_have_conflict_markers_detects_real_markers(self):
self.assertTrue(
role_session_router.python_bytes_have_conflict_markers(b"<<<<<<< HEAD\n")
)
self.assertTrue(
role_session_router.python_bytes_have_conflict_markers(b"=======\n")
)
self.assertTrue(
role_session_router.python_bytes_have_conflict_markers(b">>>>>>> topic\n")
)
if __name__ == "__main__":
unittest.main()
+2 -31
View File
@@ -95,13 +95,9 @@ class TestRuntimeClarity(unittest.TestCase):
# -------------------------------------------------------------------------
# gitea_get_runtime_context
# -------------------------------------------------------------------------
@patch(
"mcp_server.assess_preflight_status",
return_value={"preflight_ready": True, "preflight_block_reasons": []},
)
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_get_runtime_context_author(self, _auth, _api, _preflight):
def test_get_runtime_context_author(self, _auth, _api):
with patch.dict(os.environ, self._env("author-profile"), clear=True):
ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
self.assertEqual(ctx["active_profile"], "author-profile")
@@ -114,26 +110,10 @@ class TestRuntimeClarity(unittest.TestCase):
self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
self.assertIn("does not permit review or merge", ctx["review_merge_blocked_reasons"][0])
self.assertIn("Switch to the reviewer MCP session", ctx["safe_next_action"])
caps = ctx["session_capabilities"]
self.assertTrue(caps["can_author_prs"])
self.assertFalse(caps["can_create_issues"])
self.assertFalse(caps["can_comment_on_issues"])
self.assertFalse(caps["can_review_prs"])
self.assertFalse(caps["can_merge_prs"])
self.assertTrue(caps["issue_comment_not_implied_by_pr_comment"])
merge_entry = next(
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
)
self.assertFalse(merge_entry["allowed_in_current_session"])
self.assertIn("reviewer-profile", merge_entry["matching_configured_profiles"])
@patch(
"mcp_server.assess_preflight_status",
return_value={"preflight_ready": True, "preflight_block_reasons": []},
)
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_get_runtime_context_reviewer(self, _auth, _api, _preflight):
def test_get_runtime_context_reviewer(self, _auth, _api):
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
self.assertEqual(ctx["active_profile"], "reviewer-profile")
@@ -141,15 +121,6 @@ class TestRuntimeClarity(unittest.TestCase):
self.assertTrue(ctx["review_merge_allowed"])
self.assertEqual(ctx["suggested_fix"], "none")
self.assertEqual(ctx["safe_next_action"], "None; ready for operations.")
caps = ctx["session_capabilities"]
self.assertFalse(caps["can_author_prs"])
self.assertFalse(caps["can_create_issues"])
self.assertFalse(caps["can_review_prs"])
self.assertTrue(caps["can_merge_prs"])
merge_entry = next(
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
)
self.assertTrue(merge_entry["allowed_in_current_session"])
# -------------------------------------------------------------------------
# gitea_list_profiles
-112
View File
@@ -1,112 +0,0 @@
"""Checks for the Gitea Wiki sync helper (issue #224).
The repo-tracked ``docs/wiki/`` stays the source of truth; the helper
mirrors it into the Gitea native wiki only when explicitly invoked.
"""
import os
import subprocess
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
SCRIPT = REPO_ROOT / "scripts" / "sync-gitea-wiki.sh"
REQUIRED_PAGES = [
"Home.md",
"Operator-Guide.md",
"Repositories.md",
"Identity-and-Profiles.md",
"Workflow.md",
"Safety-and-Gates.md",
"MCP-Tools.md",
"Runbooks.md",
"Open-Decisions.md",
"History.md",
]
def _run(*args, env_extra=None):
env = dict(os.environ)
env.pop("GITEA_WIKI_SYNC_CONFIRM", None)
if env_extra:
env.update(env_extra)
return subprocess.run(
[str(SCRIPT), *args], capture_output=True, text=True, env=env,
cwd=str(REPO_ROOT), timeout=30)
def test_script_exists_and_is_executable():
assert SCRIPT.is_file(), "missing scripts/sync-gitea-wiki.sh"
assert os.access(SCRIPT, os.X_OK), "sync script is not executable"
def test_dry_run_is_default_and_lists_every_page():
result = _run()
assert result.returncode == 0, result.stderr
out = result.stdout
assert "dry-run" in out.lower()
for page in REQUIRED_PAGES:
assert page in out, f"dry-run plan does not list {page}"
def test_dry_run_does_not_push_or_clone():
result = _run()
blob = (result.stdout + result.stderr).lower()
assert "pushing" not in blob
assert "cloning" not in blob
def test_push_refused_without_exact_confirmation():
result = _run("--push")
assert result.returncode != 0
blob = result.stdout + result.stderr
assert "GITEA_WIKI_SYNC_CONFIRM" in blob
assert "cloning" not in blob.lower()
def test_push_refused_with_wrong_confirmation():
result = _run("--push", env_extra={"GITEA_WIKI_SYNC_CONFIRM": "yes please"})
assert result.returncode != 0
assert "cloning" not in (result.stdout + result.stderr).lower()
def test_script_embeds_no_service_urls_or_secrets():
text = SCRIPT.read_text(encoding="utf-8")
for marker in ("http://", "https://", "token ", "Authorization", "ghp_", "keychain"):
assert marker not in text, f"script embeds {marker!r}"
def test_runbook_documents_wiki_sync():
runbooks = (REPO_ROOT / "docs" / "wiki" / "Runbooks.md").read_text(encoding="utf-8")
assert "sync-gitea-wiki" in runbooks
home = (REPO_ROOT / "docs" / "wiki" / "Home.md").read_text(encoding="utf-8")
assert "source of truth" in home
template = (REPO_ROOT / ".gitea" / "pull_request_template.md").read_text(encoding="utf-8")
assert "sync-gitea-wiki" in template
def test_wiki_pages_have_no_local_file_links():
for path in (REPO_ROOT / "docs" / "wiki").glob("*.md"):
text = path.read_text(encoding="utf-8")
assert "file://" not in text, f"{path.name} contains a file:// link"
def test_pr_template_carries_wiki_publication_gate():
template = (REPO_ROOT / ".gitea" / "pull_request_template.md").read_text(encoding="utf-8")
assert "Wiki tab" in template
assert "#224" in template
assert "not sufficient to close a wiki issue" in template
def test_runbooks_document_publication_gate_and_closure_rule():
runbooks = (REPO_ROOT / "docs" / "wiki" / "Runbooks.md").read_text(encoding="utf-8")
assert "Wiki Publication Readiness Gate (#224)" in runbooks or "Wiki publication readiness gate (#224)" in runbooks.lower()
assert "Closure prevention" in runbooks
assert "GITEA_WIKI_SYNC_CONFIRM" in runbooks
def test_repositories_map_tracks_wiki_publication_status():
repositories = (REPO_ROOT / "docs" / "wiki" / "Repositories.md").read_text(encoding="utf-8")
assert "Wiki publication status" in repositories or "Wiki Publication Status" in repositories
for repo in ("mcp-control-plane", "Gitea-Tools"):
assert repo in repositories
assert "Gitea Wiki published" in repositories