Compare commits

...
Author SHA1 Message Date
sysadminandClaude Opus 4.8 eb97060867 docs(wiki): refresh Gitea-Tools wiki proof head to d1f0693 (#232)
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-06 01:56:10 -04:00
sysadmin 09a766be5d Merge pull request 'Bootstrap Gitea-Tools docs/wiki and wiki sync helper (#224)' (#230) from feat/issue-224-gitea-tools-wiki-bootstrap into master 2026-07-06 00:40:44 -05:00
sysadminandClaude Opus 4.8 342974f840 docs(wiki): record Gitea-Tools live wiki publication proof (#224)
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-06 01:36:49 -04:00
sysadminandClaude Opus 4.8 f58317d079 Add docs/wiki source tree and Gitea Wiki sync helper (#224)
Bootstrap repo-tracked wiki pages (10 required pages), the operator-
confirmed sync script, PR template publication gate, and safety tests so
Gitea-Tools can publish to the live Gitea Wiki per issue #224.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-06 01:36:10 -04:00
sysadmin 125f52a059 Merge pull request 'Fix test env isolation for credentials and create_pr gate tests' (#229) from feat/issue-227-fix-test-env-isolation into master 2026-07-06 00:29:21 -05:00
sysadmin 3320a5b3e6 test: isolate credentials and create_pr tests from MCP env leakage
Fix 8 suite failures triggered when GITEA_MCP_CONFIG/GITEA_MCP_PROFILE
are set in the operator environment.

- TestGetAuthHeader: clear MCP config env and DYNAMIC_CONFIGS in setUp
- TestCreatePR/TestIssueLocking: add CREATE_PR_ENV author fixture and
  patch reviewer-stop gate so tests reach intended code paths

Closes #227
2026-07-06 01:26:33 -04:00
sysadmin 8436276991 Merge pull request 'Fix MCP registration and naming for Jenkins/GlitchTip servers (Issue #151)' (#226) from feat/issue-151-mcp-registration-naming into master 2026-07-06 00:20:14 -05:00
sysadmin 6971a75818 Merge pull request 'Block role/tool namespace mismatch for author and reviewer mutations (Issue #209)' (#223) from feat/issue-209-namespace-mismatch into master 2026-07-06 00:14:48 -05:00
sysadmin 25c19b26d3 Merge pull request 'Enforce issue-write tool gates to match task capability resolver (mcp-control-plane #69)' (#222) from feat/issue-69-issue-write-tool-gates into master 2026-07-06 00:14:40 -05:00
sysadmin d747e0a2c9 Document external MCP registration for Jenkins and GlitchTip 2026-07-06 01:07:46 -04:00
sysadmin e727210aad Merge pull request 'Isolate Gitea MCP runtime from mutable development worktrees (Issue #221)' (#225) from feat/issue-221-runtime-isolation into master 2026-07-05 23:56:36 -05:00
sysadmin 2f5e0408fd test(review-proofs): add required Workspace mutations field to GOOD_HANDOFF mock 2026-07-05 21:14:37 -04:00
sysadmin 2cd61a90fe feat(runtime-isolation): loader, health checks, tests, and documentation for Issue #221 2026-07-05 21:13:40 -04:00
sysadmin cb974d659f feat(workflow-guard): check mid-merge in gitea_resolve_task_capability 2026-07-05 21:13:38 -04:00
sysadmin 1948eae184 feat(workflow-guard): verify MCP runtime clean/not mid-merge before starting reviewer work 2026-07-05 21:13:15 -04:00
sysadminandClaude Opus 4.8 cbce6cb2e7 Block author mutations from reviewer MCP namespace (Issue #209)
Add role_namespace_gate to fail closed when reviewer-bound sessions attempt
author mutations (create_pr always blocked; create_issue only when profile
explicitly grants gitea.issue.create). Extend mutation audit records with
mcp_namespace, task_role, and operation fields. Add handoff parity checks in
review_proofs for author role vs reviewer namespace mismatches.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-05 21:12:33 -04:00
sysadminandClaude Opus 4.8 00394d43b7 Enforce issue-write tool gates to match task capability resolver (Issue #69)
Extract TASK_CAPABILITY_MAP into task_capability_map.py as the single source
of truth shared by gitea_resolve_task_capability and issue-mutating tools.
Gate gitea_create_issue, gitea_close_issue, gitea_mark_issue, and
gitea_set_issue_labels with structured #142 permission reports. Add regression
tests proving resolver-denied tasks fail closed at the raw tool layer.

Implements mcp-control-plane #69.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-05 21:09:34 -04:00
sysadmin c89ae2eb5c Rename mcp_server.py to gitea_mcp_server.py 2026-07-05 21:07:42 -04:00
sysadmin 380ed5fde1 Merge pull request 'Harden author-run reporting and mutation capability proof per #183' (#187) from feat/issue-183-harden-author-run-reporting into master 2026-07-05 19:44:41 -05:00
sysadmin 5229dedb03 feat(author): harden reporting and claim capability per #183
- Add 'mark_issue' to TASK_MAP in resolve_task_capability for exact mutation proof (closes gap on unknown treated as allowed).
- Add assess_controller_handoff and integrate into build_final_report (downgrades missing handoff).
- Add tests for handoff and update existing.
- Update SKILL.md and review-pr.md to mandate exact 'Controller Handoff' section, exact sweep evidence, PR head SHA in reports.
- Author profile used throughout; no review/merge.

Refs #183
2026-07-05 20:41:34 -04:00
sysadmin 57ac56fcb3 Merge pull request 'feat(workflow): capability stop terminal mode after reviewer denial (Issue #197)' (#218) from feat/issue-197-capability-stop-terminal into master 2026-07-05 19:38:21 -05:00
sysadminandClaude Opus 4.8 7a7e9c0cbc 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:36:23 -04:00
sysadmin 392266c553 Merge pull request 'feat: block workspace edits before preflight verification Closes #210' (#215) from feat/issue-210-block-workspace-edits-clean into master 2026-07-05 19:34:35 -05:00
sysadmin ab95f1b253 feat(preflight): block workspace edits before identity/capability verification (Closes #210) 2026-07-05 20:31:51 -04:00
41 changed files with 7027 additions and 4357 deletions
+20
View File
@@ -0,0 +1,20 @@
## 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,6 +172,14 @@ 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
+202
View File
@@ -0,0 +1,202 @@
"""Hard-stop terminal mode after reviewer capability denial (#197)."""
from __future__ import annotations
import re
TERMINAL_REPORT_HEADING = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed."
)
REVIEWER_CAPABILITY_TASKS = frozenset({
"review_pr",
"merge_pr",
"blind_pr_queue_review",
"request_changes_pr",
"approve_pr",
})
BLOCKED_QUEUE_TOOLS = frozenset({
"list_prs",
"check_pr_eligibility",
"view_pr",
"submit_pr_review",
"dry_run_pr_review",
"merge_pr",
"review_pr",
})
_session_terminal: dict | None = None
def enter_from_capability_result(capability: dict) -> dict | None:
"""Enter terminal mode when a reviewer/merge task is denied."""
global _session_terminal
task = (capability or {}).get("requested_task", "")
required_role = (capability or {}).get("required_role_kind")
if not capability.get("stop_required"):
return None
if required_role != "reviewer" and task not in REVIEWER_CAPABILITY_TASKS:
return None
record = {
"active": True,
"requested_task": task,
"required_role_kind": required_role,
"active_profile": capability.get("active_profile"),
"active_identity": capability.get("active_identity"),
"stop_required": True,
"exact_safe_next_action": capability.get("exact_safe_next_action"),
"terminal_message": TERMINAL_REPORT_HEADING,
}
_session_terminal = record
return dict(record)
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":
return None
if route.get("required_role") != "reviewer":
return None
return enter_from_capability_result({
"requested_task": route.get("task_type"),
"required_role_kind": "reviewer",
"stop_required": True,
"active_profile": route.get("active_profile"),
"active_identity": None,
"exact_safe_next_action": route.get("message"),
})
def is_active() -> bool:
return bool(_session_terminal and _session_terminal.get("active"))
def active_record() -> dict | None:
if not is_active():
return None
return dict(_session_terminal)
def clear():
global _session_terminal
_session_terminal = None
def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
"""Return (allowed, reasons). False when terminal mode blocks queue work."""
if not is_active():
return True, []
name = (tool_name or "").strip().lower().removeprefix("gitea_")
if name in BLOCKED_QUEUE_TOOLS:
return False, [
TERMINAL_REPORT_HEADING,
f"Reviewer queue tool '{tool_name}' is blocked after "
"capability denial (fail closed).",
"Relaunch a reviewer MCP namespace to perform reviewer work.",
]
return True, []
def validate_eligibility_wording(text: str) -> tuple[bool, list[str]]:
"""Reject session-based eligibility reasoning (#197)."""
lower = (text or "").lower()
violations = []
if "not authored by this session" in lower:
violations.append(
"eligibility must use authenticated account identity, not "
"'this session' wording"
)
if re.search(r"not (?:self-)?authored by (?:the )?session", lower):
violations.append("session-based eligibility reasoning is invalid")
return (len(violations) == 0), violations
def assess_capability_stop_report(
report_text: str,
*,
trust_gate_status: str | None = None,
capability_denied: bool = True,
) -> dict:
"""Validate final report purity after reviewer capability denial."""
text = report_text or ""
lower = text.lower()
violations = []
if capability_denied and TERMINAL_REPORT_HEADING.lower() not in lower:
violations.append("missing required terminal report heading")
forbidden_patterns = [
("pr selection", re.compile(
r"selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
r"next pr to review", re.I)),
("sibling repo inventory", re.compile(
r"sibling repo|other repo|mcp-control-plane|gitea-tools and", re.I)),
("author fallback", re.compile(
r"rebase conflicted|author-side fallback|have me rebase|"
r"implement the fix|push a branch|open a pr for", re.I)),
("invalid session eligibility", re.compile(
r"not authored by this session", re.I)),
]
for label, pattern in forbidden_patterns:
if pattern.search(text):
violations.append(f"forbidden after hard stop: {label}")
empty_queue_patterns = re.compile(
r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
r"inventory empty",
re.I,
)
if empty_queue_patterns.search(text):
if trust_gate_status != "trusted_empty":
violations.append(
"empty-queue claim after capability stop without "
"pr_inventory_trust_gate.status == trusted_empty"
)
ok, elig_violations = validate_eligibility_wording(text)
violations.extend(elig_violations)
if violations:
return {
"pure": False,
"downgraded": True,
"violations": violations,
"reasons": violations,
}
return {
"pure": True,
"downgraded": False,
"violations": [],
"reasons": [],
}
def build_terminal_report(capability: dict) -> dict:
"""Minimal allowed report fields after hard stop."""
return {
"terminal_mode": True,
"heading": TERMINAL_REPORT_HEADING,
"authenticated_profile": capability.get("active_profile"),
"authenticated_identity": capability.get("active_identity"),
"denied_task": capability.get("requested_task"),
"required_role_kind": capability.get("required_role_kind"),
"stop_required": capability.get("stop_required"),
"required_action": capability.get("exact_safe_next_action"),
"mutations_performed": False,
"allowed_sections": [
"authenticated identity/profile",
"denied capability result",
"reason task cannot proceed",
"required reviewer profile/identity",
"mutation confirmation (none)",
],
"forbidden_sections": [
"PR selection",
"sibling-repo queue recommendations",
"author-side fallback suggestions",
"empty-queue claims without trusted_empty",
"session-based eligibility wording",
],
}
@@ -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 (namespace pending)
## 2. Boundary placement
These tools belong to the GlitchTip observability boundary of the MCP Control
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.
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.
Fixed regardless of the name (per `tool-boundaries.md`,
`credential-isolation.md`):
@@ -160,11 +160,13 @@ Mocked-GlitchTip unit tests only, per `docs/developer-testing-guidelines.md`:
## 10. Implementation-readiness checklist
Ready to implement once:
Ready to operate once:
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.
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.
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).
@@ -6,6 +6,8 @@
- **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 server boundary now contains gated trigger (see #56), but read tools remain as designed.
Client registration and reload instructions live in
[`../mcp-client-registration.md`](../mcp-client-registration.md).
## 1. Purpose and scope
@@ -141,12 +143,16 @@ repo's conventions (`docs/developer-testing-guidelines.md`):
## 10. Implementation-readiness checklist
Ready to implement in `jenkins-mcp` once:
Ready to operate through `jenkins-mcp` once:
1. ADR-0001 owner decision #1 (where `jenkins-mcp` lives) is made.
2. #76 profile schema exists (or a minimal `jenkins-readonly` profile is
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
hand-rolled to the same rules).
3. #77 mapping design is accepted (or tools ship path-addressed only, mapping
4. #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,6 +298,17 @@ 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
+5
View File
@@ -30,6 +30,11 @@ 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,
+87
View File
@@ -0,0 +1,87 @@
# 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-only build/job inspection |
| `glitchtip-mcp` | GlitchTip observability inspection | Read-only issue/event inspection |
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 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`
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.
- Jenkins build triggers require a separately named write profile, exact
confirmation, and fail-closed mutation audit.
- `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.
+15
View File
@@ -0,0 +1,15 @@
# 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
@@ -0,0 +1,35 @@
# 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
@@ -0,0 +1,39 @@
# 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
@@ -0,0 +1,34 @@
# 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
@@ -0,0 +1,11 @@
# 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.
+31
View File
@@ -0,0 +1,31 @@
# 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.
## 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.
+37
View File
@@ -0,0 +1,37 @@
# 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) |
Update this table whenever a wiki is published, re-synced, or found stale.
+40
View File
@@ -0,0 +1,40 @@
# 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
@@ -0,0 +1,19 @@
# 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).
+28
View File
@@ -0,0 +1,28 @@
# 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)
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.
+10 -2
View File
@@ -163,12 +163,13 @@ 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):
head_sha=None, reason=None, request_metadata=None, now=None,
mcp_namespace=None, task_role=None, operation=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()
return {
event = {
"timestamp": ts,
"action": action,
"action_type": "mutating",
@@ -186,6 +187,13 @@ 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):
+4531
View File
File diff suppressed because it is too large Load Diff
+45 -4281
View File
File diff suppressed because it is too large Load Diff
+208 -11
View File
@@ -635,7 +635,9 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
issue_status_verified,
capability_evidence=None, sweep=None, live_state=None,
role_boundary=None, review_mutation=None,
report_text=None, review_decision_lock=None):
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -671,6 +673,16 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"violations": [],
}
role_status = role_boundary.get("status", "warning")
handoff = _normalize_controller_handoff_proof(controller_handoff)
capability_proof = capability_proof or {
"proven": False,
"reasons": ["capability proof not provided"],
}
sweep_proof = sweep_proof or {
"proven": False,
"verdict": "invalid",
"reasons": ["secret/provenance sweep proof not provided"],
}
capability_evidence = capability_evidence or {
"proven": False,
@@ -731,6 +743,15 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
downgrade_reasons.extend(role_boundary.get("reasons", []))
if not issue_status_verified:
downgrade_reasons.append("linked issue status not verified")
if not handoff["complete"]:
downgrade_reasons.append("Controller Handoff missing or incomplete")
downgrade_reasons.extend(handoff.get("reasons", []))
if not capability_proof.get("proven"):
downgrade_reasons.append("exact mutation capability proof missing")
downgrade_reasons.extend(capability_proof.get("reasons", []))
if not sweep_proof.get("proven"):
downgrade_reasons.append("exact secret/provenance sweep proof missing")
downgrade_reasons.extend(sweep_proof.get("reasons", []))
if not capability_proven:
downgrade_reasons.append("exact capability evidence missing (#179)")
downgrade_reasons.extend(capability_evidence.get("reasons", []))
@@ -795,6 +816,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"merge_allowed": merge_allowed,
"merge_performed": bool(merge_performed),
"issue_status_verified": bool(issue_status_verified),
"controller_handoff_present": handoff["present"],
"controller_handoff_complete": handoff["complete"],
"capability_proof": bool(capability_proof.get("proven")),
"secret_sweep_proof": bool(sweep_proof.get("proven")),
"capability_evidence_proven": capability_proven,
"sweep_verdict": sweep.get("verdict"),
"live_state_recheck_proven": live_state_proven,
@@ -803,6 +828,118 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
}
def _normalize_controller_handoff_proof(controller_handoff):
"""Normalize old boolean handoff proof and rich #182 handoff verdicts."""
if controller_handoff is None:
return {
"present": False,
"complete": False,
"reasons": ["Controller Handoff proof not provided"],
}
if isinstance(controller_handoff, str):
controller_handoff = assess_controller_handoff(controller_handoff)
if not isinstance(controller_handoff, dict):
return {
"present": False,
"complete": False,
"reasons": ["Controller Handoff proof has invalid type"],
}
if "verdict" in controller_handoff:
verdict = controller_handoff.get("verdict")
return {
"present": verdict in {"complete", "incomplete"},
"complete": verdict == "complete",
"reasons": list(controller_handoff.get("reasons") or []),
}
present = bool(controller_handoff.get("present"))
return {
"present": present,
"complete": present,
"reasons": list(controller_handoff.get("reasons") or []),
}
def assess_capability_proof(resolved_capabilities: dict) -> dict:
"""Every mutation must have exact capability proof (#183)."""
reasons = []
if not resolved_capabilities:
return {
"proven": False,
"reasons": ["no capability proof resolved; fail closed"],
}
for task, cap in resolved_capabilities.items():
allowed = cap.get("allowed_in_current_session")
op = cap.get("required_operation_permission")
if not allowed:
reasons.append(
f"task '{task}' requires permission '{op}' which is not "
"allowed in current session"
)
if (
"unknown" in str(cap.get("requested_task", "")).lower()
or cap.get("allowed_in_current_session") is None
):
reasons.append(
f"task '{task}' could not be resolved to a known capability"
)
return {"proven": not reasons, "reasons": reasons}
def assess_secret_sweep(sweep_report: dict) -> dict:
"""Secret/provenance sweeps must state exact command/method and scope."""
if not sweep_report:
return {
"proven": False,
"verdict": "invalid",
"reasons": ["no secret/provenance sweep report provided"],
}
reasons = []
method = (sweep_report.get("method") or "").strip()
scope = (sweep_report.get("scope") or "").strip()
if not method:
reasons.append(
"secret sweep report missing exact scan command, pattern, or method"
)
if not scope:
reasons.append("secret sweep report missing scanned scope")
if sweep_report.get("clean") is not True:
reasons.append("secret sweep report did not confirm diff is clean")
verdict = "weak" if reasons else "strong"
return {
"proven": not reasons and sweep_report.get("clean") is True,
"verdict": verdict,
"reasons": reasons,
}
def assess_author_pr_report(pr_report: dict) -> dict:
"""Author PR reports must include PR number, branch, and exact head SHA."""
if not pr_report:
return {
"complete": False,
"reasons": ["no PR report provided"],
}
reasons = []
pr_number = pr_report.get("pr_number")
branch = (pr_report.get("branch") or "").strip()
head_sha = (pr_report.get("head_sha") or "").strip()
if not isinstance(pr_number, int) or pr_number <= 0:
reasons.append("PR number is missing or invalid")
if not branch:
reasons.append("PR branch name is missing")
if not head_sha:
reasons.append("PR head SHA is missing")
elif not _FULL_SHA.match(head_sha):
reasons.append("PR head SHA is not a full 40-hex commit SHA")
return {
"complete": not reasons,
"reasons": reasons,
}
# ── Controller Handoff validation (Issue #182) ────────────────────────────────
#
# Every final report must end with a compact section titled exactly
@@ -822,6 +959,7 @@ HANDOFF_BASE_FIELDS = (
("Files changed", ("files changed", "changed", "files")),
("Validation", ("validation",)),
("Mutations", ("mutations",)),
("Workspace mutations", ("workspace mutations",)),
("Current status", ("current status", "status")),
("Blockers", ("blockers",)),
("Next", ("next",)),
@@ -872,7 +1010,7 @@ def _handoff_section_lines(report_text):
return lines[start:]
def assess_controller_handoff(report_text, role=None):
def assess_controller_handoff(report_text, role=None, local_edits=False):
"""Issue #182: final reports without a Controller Handoff downgrade.
Verdicts:
@@ -897,10 +1035,14 @@ def assess_controller_handoff(report_text, role=None):
}
labels = []
fields_dict = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
labels.append(stripped.split(":", 1)[0].strip().lower())
k, v = stripped.split(":", 1)
label = k.strip().lower()
labels.append(label)
fields_dict[label] = v.strip()
required = list(HANDOFF_BASE_FIELDS)
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
@@ -919,15 +1061,7 @@ def assess_controller_handoff(report_text, role=None):
"reasons": [f"handoff missing required field: {m}"
for m in missing],
}
# Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
fields_dict = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
k, v = stripped.split(":", 1)
fields_dict[k.strip().lower()] = v.strip()
for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
val = fields_dict.get(alias)
if val:
@@ -948,6 +1082,18 @@ def assess_controller_handoff(report_text, role=None):
],
}
if local_edits:
workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
if not workspace_mutations_val or workspace_mutations_val == "none":
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": ["Workspace mutations"],
"reasons": [
"Workspace mutations cannot be 'none' or empty when local edits exist"
],
}
return {
"verdict": "complete",
"downgraded": False,
@@ -993,6 +1139,57 @@ def assess_role_route_handoff(report_text, route_result=None):
}
def assess_mutation_namespace_handoff(
*,
declared_role: str | None = None,
namespaces_used: list[str] | None = None,
capability_profile: str | None = None,
):
"""Issue #209: handoff role and capability proof must match mutation namespace."""
reasons = []
role = (declared_role or "").strip().lower()
namespaces = list(namespaces_used or [])
reviewer_namespaces = [
ns for ns in namespaces if "reviewer" in (ns or "").lower()
]
if role == "author" and reviewer_namespaces:
reasons.append(
"author handoff reported reviewer MCP namespace for mutations")
cap = (capability_profile or "").strip().lower()
if cap and "author" in cap and reviewer_namespaces:
reasons.append(
"capability proof profile does not match mutation namespace "
f"(profile={capability_profile}, namespaces={reviewer_namespaces})")
blocked = bool(reasons)
return {
"valid": not blocked,
"blocked": blocked,
"complete": not blocked,
"downgraded": blocked,
"reasons": reasons,
"missing_fields": [] if not blocked else ["namespace parity"],
}
def assess_capability_stop_terminal_report(
report_text,
*,
trust_gate_status=None,
capability_denied=True,
):
"""Issue #197: reports after reviewer capability denial must stay pure."""
from capability_stop_terminal import assess_capability_stop_report
return assess_capability_stop_report(
report_text,
trust_gate_status=trust_gate_status,
capability_denied=capability_denied,
)
# ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
#
# A reviewer agent may not convert an empty PR list response into a definitive
+89
View File
@@ -0,0 +1,89 @@
"""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,
}
+53 -1
View File
@@ -5,12 +5,14 @@ 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"
REVIEWER_TASKS = frozenset({
"review_pr",
@@ -93,6 +95,23 @@ 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,
@@ -192,4 +211,37 @@ def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool,
"task and relaunches an author MCP session.",
f"Attempted fallback mutation: {mutation_task}",
]
return True, []
return True, []
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
# Scan python files for conflict markers
conflict_patterns = [
b"<" * 7 + b" ",
b"=" * 7 + b"\n",
b"=" * 7 + b"\r\n",
b">" * 7 + b" "
]
for root, dirs, files in os.walk(project_root):
if any(p in root for p in ("venv", ".git", ".pytest_cache", "branches")):
continue
for file in files:
if file.endswith(".py"):
file_path = os.path.join(root, file)
try:
with open(file_path, "rb") as f:
content = f.read()
if any(pattern in content for pattern in conflict_patterns):
return True
except Exception:
pass
return False
+86
View File
@@ -0,0 +1,86 @@
#!/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
+3 -1
View File
@@ -341,11 +341,13 @@ Ready-to-copy templates live in [`templates/`](templates/):
## K. Controller Handoff (required, every task)
Every LLM task **must end with a `Controller Handoff`** — whether the
Every LLM task **must end with a `Controller Handoff`** (exact title) — 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.
+113
View File
@@ -0,0 +1,113 @@
"""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",
},
}
# 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",
}
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])
+10
View File
@@ -26,4 +26,14 @@ def _reset_mutation_authority(monkeypatch):
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
try:
import capability_stop_terminal
capability_stop_terminal.clear()
except Exception:
pass
yield
try:
import capability_stop_terminal
capability_stop_terminal.clear()
except Exception:
pass
+6 -2
View File
@@ -162,7 +162,8 @@ 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_ALLOWED_OPERATIONS": (
"read,merge,gitea.issue.create,gitea.issue.close")}
env.update(extra)
return env
@@ -237,7 +238,10 @@ 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"}, clear=True):
with patch.dict(os.environ, {
"GITEA_PROFILE_NAME": "gitea-author",
"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create",
}, 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"
+164
View File
@@ -0,0 +1,164 @@
"""Tests for capability stop terminal mode (#197)."""
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 capability_stop_terminal
import gitea_config
import mcp_server
from review_proofs import assess_capability_stop_terminal_report
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",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
],
"execution_profile": "prgs-author",
},
},
"rules": {"allow_runtime_switching": False},
}
class TestCapabilityStopTerminal(unittest.TestCase):
def setUp(self):
capability_stop_terminal.clear()
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()
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()
capability_stop_terminal.clear()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir.cleanup()
def _env(self):
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": "prgs-author",
"GITEA_TOKEN_AUTHOR": "author-pass",
}
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
with patch.dict(os.environ, self._env()):
res = mcp_server.gitea_resolve_task_capability(
task="review_pr", remote="prgs"
)
self.assertTrue(res["stop_required"])
self.assertTrue(res.get("terminal_mode"))
self.assertTrue(capability_stop_terminal.is_active())
self.assertIn(
capability_stop_terminal.TERMINAL_REPORT_HEADING,
res["terminal_report"]["heading"],
)
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
with patch.dict(os.environ, self._env()):
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_list_prs(remote="prgs")
self.assertIn("Cannot perform reviewer task", str(ctx.exception))
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_eligibility_check_blocked_after_stop(self, _auth, _api):
with patch.dict(os.environ, self._env()):
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
res = mcp_server.gitea_check_pr_eligibility(
pr_number=193, action="review", remote="prgs"
)
self.assertFalse(res["eligible"])
self.assertTrue(res.get("terminal_mode"))
def test_report_with_pr_selection_impure(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Selected PR #193 for review anyway."
)
result = assess_capability_stop_terminal_report(report)
self.assertFalse(result["pure"])
def test_session_eligibility_wording_blocked(self):
ok, violations = capability_stop_terminal.validate_eligibility_wording(
"PR 193 is not authored by this session so it is eligible."
)
self.assertFalse(ok)
self.assertTrue(violations)
def test_rebase_fallback_blocked_in_report(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Or have me rebase conflicted PR 193."
)
result = assess_capability_stop_terminal_report(report)
self.assertFalse(result["pure"])
self.assertTrue(
any("author fallback" in v for v in result["violations"])
)
def test_empty_queue_without_trusted_empty_blocked(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"The repo has 0 open PRs."
)
result = assess_capability_stop_terminal_report(
report, trust_gate_status="untrusted_empty"
)
self.assertFalse(result["pure"])
def test_pure_terminal_report_passes(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Identity: jcwalker3 / prgs-author.\n"
"Required: prgs-reviewer.\n"
"No mutations performed."
)
result = assess_capability_stop_terminal_report(report)
self.assertTrue(result["pure"])
if __name__ == "__main__":
unittest.main()
+18
View File
@@ -107,6 +107,24 @@ 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")
@@ -0,0 +1,82 @@
"""Documentation checks for external Jenkins/GlitchTip MCP registration (#151)."""
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",
"remains read-only",
):
assert phrase in text
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")
+58
View File
@@ -0,0 +1,58 @@
import os
import subprocess
import sys
import unittest
from unittest.mock import patch
import role_session_router
from mcp_server import gitea_route_task_session, gitea_resolve_task_capability
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_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")
# Run mcp_server.py
venv_python = os.path.join(self.project_root, "venv", "bin", "python")
script_path = os.path.join(self.project_root, "mcp_server.py")
res = subprocess.run(
[venv_python, script_path],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=self.project_root
)
self.assertEqual(res.returncode, 1)
stderr = res.stderr.decode()
self.assertIn("infra_stop", stderr)
self.assertIn("Unresolved merge conflict detected in test_temp_conflict.py", stderr)
@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"])
+15 -9
View File
@@ -19,6 +19,9 @@ 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"
@@ -118,7 +121,8 @@ class TestCreateIssueMCPGate(unittest.TestCase):
mock_get_all.return_value = [
{"number": 201, "title": CANONICAL_TITLE, "state": "open"},
]
result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
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()
@@ -135,13 +139,14 @@ 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"}
result = gitea_create_issue(
title=CANONICAL_TITLE,
body="Follow-up scope",
remote="prgs",
allow_duplicate_override=True,
split_from_issue=201,
)
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,
)
self.assertEqual(result["number"], 210)
payload = mock_api.call_args[0][3]
self.assertIn("Operator-approved split from #201.", payload["body"])
@@ -156,7 +161,8 @@ class TestCreateIssueMCPGate(unittest.TestCase):
):
mock_role_check.return_value = (True, [])
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
result = gitea_create_issue(title="Unique new issue", body="body text")
with patch.dict(__import__("os").environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_create_issue(title="Unique new issue", body="body text")
self.assertEqual(result["number"], 1)
+210
View File
@@ -0,0 +1,210 @@
"""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()
+158 -31
View File
@@ -47,6 +47,24 @@ import mcp_server
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
# 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"
),
}
# ---------------------------------------------------------------------------
# Create Issue
@@ -60,7 +78,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, {}, clear=True):
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_create_issue(title="Test issue", body="body text")
self.assertEqual(result["number"], 1)
self.assertNotIn("url", result)
@@ -77,7 +95,8 @@ 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"}
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
env = {**ISSUE_WRITE_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
result = gitea_create_issue(title="Test issue", body="body text")
self.assertIn("issues/1", result["url"])
@@ -88,7 +107,8 @@ 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"}
result = gitea_create_issue(title="Test", remote="prgs")
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
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)
@@ -100,14 +120,16 @@ 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):
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
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, {}, clear=True):
with patch.dict(os.environ, CREATE_PR_ENV, 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)
@@ -115,14 +137,17 @@ class TestCreatePR(unittest.TestCase):
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
@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):
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
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, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertIn("pulls/3", result["url"])
@@ -136,7 +161,8 @@ 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"}
result = gitea_close_issue(issue_number=42)
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
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")
@@ -235,7 +261,8 @@ class TestMarkIssue(unittest.TestCase):
[{"id": 10, "name": "status:in-progress"}],
[{"name": "status:in-progress"}],
]
result = gitea_mark_issue(issue_number=5, action="start")
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_mark_issue(issue_number=5, action="start")
self.assertTrue(result["success"])
self.assertIn("claimed", result["message"])
@@ -246,7 +273,8 @@ class TestMarkIssue(unittest.TestCase):
[{"id": 10, "name": "status:in-progress"}],
None,
]
result = gitea_mark_issue(issue_number=5, action="done")
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
result = gitea_mark_issue(issue_number=5, action="done")
self.assertTrue(result["success"])
self.assertIn("released", result["message"])
@@ -258,8 +286,9 @@ 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 self.assertRaises(RuntimeError):
gitea_mark_issue(issue_number=5, action="start")
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with self.assertRaises(RuntimeError):
gitea_mark_issue(issue_number=5, action="start")
# ---------------------------------------------------------------------------
@@ -269,12 +298,14 @@ class TestAuthErrors(unittest.TestCase):
@patch("mcp_server.get_auth_header", return_value=None)
def test_no_credentials_raises(self, _auth):
with self.assertRaises(RuntimeError):
gitea_create_issue(title="test")
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with self.assertRaises(RuntimeError):
gitea_create_issue(title="test")
def test_unknown_remote_raises(self):
with self.assertRaises(ValueError):
gitea_create_issue(title="test", remote="nonexistent")
with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True):
with self.assertRaises(ValueError):
gitea_create_issue(title="test", remote="nonexistent")
# ---------------------------------------------------------------------------
@@ -1939,7 +1970,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": ["merge", "edit", "close", "gitea.pr.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", "gitea.issue.close"], "audit_label": "test", "forbidden_operations": []}).start()
def tearDown(self):
patch.stopall()
@@ -2781,35 +2812,131 @@ 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.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):
def test_create_pr_missing_lock_fails(self, _auth, _role):
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")
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")
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):
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
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")
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")
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):
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
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=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("contains forbidden term", str(ctx.exception))
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):
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
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")
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
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))
# ---------------------------------------------------------------------------
# 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
# 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
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
if "GITEA_TEST_FORCE_DIRTY" in os.environ:
del os.environ["GITEA_TEST_FORCE_DIRTY"]
def test_preflight_whoami_violation(self):
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server._preflight_capability_called = True
mcp_server.record_preflight_check("whoami")
self.assertTrue(mcp_server._preflight_whoami_violation)
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
def test_preflight_capability_violation(self):
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server._preflight_whoami_called = True
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))
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._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "reviewer"
# When dirty, reviewer edits are blocked
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
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))
# When clean, reviewer is allowed
del os.environ["GITEA_TEST_FORCE_DIRTY"]
mcp_server.verify_preflight_purity()
+19 -6
View File
@@ -234,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.assertNotEqual(by_name["jenkins-mcp"]["status"], "available")
self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available")
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
self.assertEqual(by_name["glitchtip-mcp"]["status"], "external-mcp")
def test_no_urls_in_registry(self):
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
@@ -246,16 +246,29 @@ 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 #146)."""
"""Negative assertion for 'enabled but no usable tools' (per issue #151)."""
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 designed-not-implemented; even if "enabled" in config,
# it should not be usable/available to current profile without tools.
# jenkins-mcp is an external boundary; Gitea profile discovery must not
# treat a local config entry as a usable Jenkins tool surface.
self.assertIn("jenkins-mcp", by_name)
self.assertEqual(by_name["jenkins-mcp"]["status"], "designed-not-implemented")
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
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
+11
View File
@@ -130,6 +130,17 @@ 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):
+168
View File
@@ -21,12 +21,15 @@ import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from review_proofs import ( # noqa: E402
assess_author_pr_report,
assess_capability_evidence,
assess_capability_proof,
assess_controller_handoff,
assess_inventory_completeness,
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,
@@ -190,6 +193,50 @@ 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."""
@@ -545,6 +592,9 @@ 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(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -662,6 +712,28 @@ 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
@@ -789,6 +861,7 @@ 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",
@@ -909,6 +982,29 @@ class TestControllerHandoff(unittest.TestCase):
self.assertIn("assess_controller_handoff", skill)
self.assertIn("issue #182", 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([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
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([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
class TestReviewMutationFinalReport(unittest.TestCase):
"""Final reports must list exactly one live review mutation."""
@@ -1207,6 +1303,9 @@ 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(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -1282,5 +1381,74 @@ 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"])
if __name__ == "__main__":
unittest.main()
+215
View File
@@ -0,0 +1,215 @@
"""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()
+112
View File
@@ -0,0 +1,112 @@
"""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