diff --git a/anti_stomp_preflight.py b/anti_stomp_preflight.py index 10cacdd..f025f13 100644 --- a/anti_stomp_preflight.py +++ b/anti_stomp_preflight.py @@ -77,6 +77,7 @@ MUTATION_TASKS = frozenset({ "create_issue", "comment_issue", "close_issue", + "edit_issue", "mark_issue", "lock_issue", "set_issue_labels", diff --git a/docs/mcp-tool-inventory.md b/docs/mcp-tool-inventory.md new file mode 100644 index 0000000..5e5a54f --- /dev/null +++ b/docs/mcp-tool-inventory.md @@ -0,0 +1,169 @@ +# Registered MCP tool inventory + +This is the canonical list of tools the Gitea-Tools MCP server registers. It +exists because documentation and the registered inventory drifted: the workflow +documented a `gitea_edit_issue` tool that no namespace had ever registered, so a +mutation could be planned against a tool that did not exist and only fail at +execution time (#781). + +## The rule + +**Documentation must never name a tool an actor cannot reach.** + +Two guards enforce it, both in `tests/test_issue_781_edit_issue_tool.py`: + +1. The list below must equal the registered tool set exactly — sorted, no + duplicates, nothing missing in either direction. Adding a tool without + documenting it fails, and documenting a tool without registering it fails. +2. Every backticked `gitea_*` / `mcp_*` identifier in `skills/**/*.md` must be a + registered tool. Module and script names that share the prefix are listed + explicitly in `mcp_tool_inventory.NON_TOOL_IDENTIFIERS` rather than being + waved through by a looser pattern. + +## Updating this file + +When you add or remove an `@mcp.tool()`, regenerate the block below: + +```bash +PYTEST_CURRENT_TEST=1 venv/bin/python -c " +import mcp_server, mcp_tool_inventory +print(mcp_tool_inventory.render_inventory_block( + mcp_server.mcp._tool_manager._tools)) +" +``` + +Replace everything between the markers with that output. Do not hand-edit +individual entries — the generator and the guard share one ordering rule. + +## Registered tools + +Namespaces (`gitea-tools`, `gitea-reviewer`, `gitea-merger`, `gitea-reconciler`) +register the same tool set; what differs per namespace is the execution profile +that gates each call, not which tools exist. + + + +- `gitea_abandon_workflow_lease` +- `gitea_acquire_conflict_fix_lease` +- `gitea_acquire_merger_pr_lease` +- `gitea_acquire_reviewer_pr_lease` +- `gitea_activate_profile` +- `gitea_adopt_merger_pr_lease` +- `gitea_adopt_workflow_lease` +- `gitea_allocate_next_work` +- `gitea_assess_already_landed_reconciliation` +- `gitea_assess_conflict_fix_classification` +- `gitea_assess_conflict_fix_push` +- `gitea_assess_gitea_operation_path` +- `gitea_assess_master_parity` +- `gitea_assess_mcp_namespace_health` +- `gitea_assess_pr_sync_status` +- `gitea_assess_review_merge_state_machine` +- `gitea_assess_reviewer_pr_lease` +- `gitea_assess_terminal_label_hygiene` +- `gitea_assess_work_issue_duplicate` +- `gitea_assess_worktree_cleanup_integrity` +- `gitea_audit_config` +- `gitea_audit_stable_branch_contamination` +- `gitea_audit_worktree_cleanup` +- `gitea_authorize_reconciliation_cleanup_phase` +- `gitea_authorize_review_correction` +- `gitea_capability_stop_terminal_report` +- `gitea_capture_branches_worktree_snapshot` +- `gitea_check_pr_eligibility` +- `gitea_cleanup_merged_pr_branch` +- `gitea_cleanup_obsolete_reviewer_comment_lease` +- `gitea_cleanup_post_merge_moot_lease` +- `gitea_cleanup_stale_claims` +- `gitea_cleanup_stale_review_decision_lock` +- `gitea_cleanup_terminal_pr_labels` +- `gitea_close_issue` +- `gitea_commit_files` +- `gitea_consume_irrecoverable_decision_lock_provenance` +- `gitea_create_issue` +- `gitea_create_issue_comment` +- `gitea_create_label` +- `gitea_create_pr` +- `gitea_delete_branch` +- `gitea_diagnose_review_decision_lock` +- `gitea_diagnose_reviewer_pr_lease_handoff` +- `gitea_diagnose_terminal` +- `gitea_dry_run_pr_review` +- `gitea_edit_issue` +- `gitea_edit_pr` +- `gitea_expire_workflow_leases` +- `gitea_get_authenticated_user` +- `gitea_get_current_user` +- `gitea_get_file` +- `gitea_get_pr_review_feedback` +- `gitea_get_profile` +- `gitea_get_runtime_context` +- `gitea_get_shell_health` +- `gitea_heartbeat_reviewer_pr_lease` +- `gitea_inspect_workflow_lease` +- `gitea_issue_irrecoverable_provenance_authorization` +- `gitea_list_issue_comments` +- `gitea_list_issues` +- `gitea_list_labels` +- `gitea_list_profiles` +- `gitea_list_prs` +- `gitea_list_workflow_leases` +- `gitea_load_review_workflow` +- `gitea_lock_issue` +- `gitea_mark_final_review_decision` +- `gitea_mark_issue` +- `gitea_merge_pr` +- `gitea_mirror_refs` +- `gitea_observability_link_issue` +- `gitea_observability_list_projects` +- `gitea_observability_reconcile_incident` +- `gitea_post_heartbeat` +- `gitea_quarantine_contaminated_review` +- `gitea_reclaim_expired_workflow_lease` +- `gitea_reconcile_already_landed_pr` +- `gitea_reconcile_issue_claims` +- `gitea_reconcile_merged_cleanups` +- `gitea_reconcile_superseded_by_merged_pr` +- `gitea_record_irrecoverable_decision_lock_provenance` +- `gitea_record_pre_review_command` +- `gitea_record_shell_spawn_outcome` +- `gitea_record_stable_branch_push_attempt` +- `gitea_release_merger_pr_lease` +- `gitea_release_reviewer_pr_lease` +- `gitea_release_workflow_lease` +- `gitea_resolve_task_capability` +- `gitea_resume_review_draft` +- `gitea_review_pr` +- `gitea_route_task_session` +- `gitea_save_review_draft` +- `gitea_scan_already_landed_open_prs` +- `gitea_sentry_get_issue_events` +- `gitea_sentry_link_gitea_issue` +- `gitea_sentry_list_issues` +- `gitea_sentry_reconcile_issue` +- `gitea_sentry_watchdog` +- `gitea_set_issue_labels` +- `gitea_submit_pr_review` +- `gitea_update_pr_branch_by_merge` +- `gitea_validate_review_final_report` +- `gitea_view_issue` +- `gitea_view_pr` +- `gitea_whoami` +- `gitea_workflow_dashboard` +- `mcp_check_workflow_skill_preflight` +- `mcp_get_control_plane_guide` +- `mcp_get_skill_guide` +- `mcp_list_project_skills` + + + +## Issue-content editing + +`gitea_edit_issue` is the only path that changes an issue's title or body. It +PATCHes the issue endpoint, refuses a pull-request number, sends only the fields +the caller named, and proves the result by read-after-write — including that +state, labels, assignees, and milestone did not move. + +`gitea_edit_pr` remains pull-request-only. The two paths never merge: a single +tool that accepted either kind would make the narrower capability reachable +through the wider one. diff --git a/edit_issue.py b/edit_issue.py new file mode 100644 index 0000000..c853f15 --- /dev/null +++ b/edit_issue.py @@ -0,0 +1,271 @@ +"""Authoritative rule for editing an issue's title and body (#781). + +The workflow documented a ``gitea_edit_issue`` tool that was never registered, +so an authorized body correction on an issue had no sanctioned path at all: the +only edit tool, ``gitea_edit_pr``, PATCHes the pull-request endpoint and cannot +target an issue. This module is the rule that path is built on, kept separate +from the pull-request edit path by construction. + +- :func:`validate_edit_request` rejects structurally invalid requests before any + credential, network, or profile work happens. A request that names no field, + or names one with the wrong type, is a pure input error. +- :func:`assess_issue_target` refuses a pull request. Gitea serves pull requests + from the same ``/issues/{n}`` collection, so without this check the issue edit + path would quietly become a second, ungated PR edit path. +- :func:`plan_issue_edit` decides the exact PATCH payload from the pre-image. It + only ever sends fields the caller named, and it reports a request that would + change nothing as an explicit no-op rather than a silent success. +- :func:`verify_issue_edit` is the read-after-write check. It proves the applied + title/body match what was requested *and* that every field the caller did not + name — state, labels, assignees, milestone — is unchanged. + +This module performs no I/O — callers own the Gitea API calls. +""" + +from __future__ import annotations + +from typing import Any, Mapping + +import issue_workflow_labels + +#: Fields this tool is allowed to change. Anything else must be untouched. +EDITABLE_FIELDS: tuple[str, ...] = ("title", "body") + +#: Fields the caller never names and which must survive an edit verbatim. +PRESERVED_FIELDS: tuple[str, ...] = ("state", "labels", "assignees", "milestone") + + +def validate_edit_request( + title: str | None = None, + body: str | None = None, +) -> dict[str, str]: + """Return the requested field map, failing closed on an invalid request. + + Raises ``ValueError`` when no field is named, when a named field is not a + string, or when a title is blank. An empty *body* is legitimate — clearing + an issue description is a real edit — but an empty title is not, because + Gitea has no issue without one. + """ + requested: dict[str, str] = {} + + if title is not None: + if not isinstance(title, str): + raise ValueError( + f"Invalid title type {type(title).__name__}: title must be a " + "string (fail closed)." + ) + if not title.strip(): + raise ValueError( + "Invalid title: an issue title cannot be blank. Pass the exact " + "replacement title, or omit title= to leave it unchanged " + "(fail closed)." + ) + requested["title"] = title + + if body is not None: + if not isinstance(body, str): + raise ValueError( + f"Invalid body type {type(body).__name__}: body must be a " + "string (fail closed)." + ) + requested["body"] = body + + if not requested: + raise ValueError( + "At least one field to edit (title, body) must be provided. " + "gitea_edit_issue never edits state, labels, assignees, or " + "milestone (fail closed)." + ) + + return requested + + +def assess_issue_target( + issue: Mapping[str, Any], + *, + issue_number: int, +) -> dict[str, Any]: + """Confirm the fetched object is an issue and not a pull request. + + Gitea serves pull requests from ``/issues/{n}`` as well, so a PR number + reaches this path unchallenged. Issue and pull-request edits stay separate + capabilities, so a PR target is refused here rather than silently PATCHed. + """ + is_pull_request = bool(issue.get("pull_request")) + return { + "is_issue": not is_pull_request, + "is_pull_request": is_pull_request, + "reasons": ( + [ + f"#{issue_number} is a pull request, not an issue; " + "gitea_edit_issue never edits pull requests" + ] + if is_pull_request + else [] + ), + "safe_next_action": ( + f"Use gitea_edit_pr for pull request #{issue_number}." + if is_pull_request + else "" + ), + } + + +def preserved_snapshot(issue: Mapping[str, Any]) -> dict[str, Any]: + """Capture the fields an edit must leave alone, in a comparable shape.""" + return { + "state": issue.get("state"), + "labels": issue_workflow_labels.label_names(issue), + "assignees": _assignee_names(issue), + "milestone": _milestone_key(issue), + } + + +def _assignee_names(issue: Mapping[str, Any]) -> list[str]: + names: list[str] = [] + for entry in issue.get("assignees") or []: + if isinstance(entry, Mapping): + login = entry.get("login") or entry.get("username") + else: + login = entry + if login: + names.append(str(login)) + return names + + +def _milestone_key(issue: Mapping[str, Any]) -> str | None: + milestone = issue.get("milestone") + if not milestone: + return None + if isinstance(milestone, Mapping): + key = milestone.get("title") or milestone.get("id") + return None if key is None else str(key) + return str(milestone) + + +def plan_issue_edit( + current: Mapping[str, Any], + *, + title: str | None = None, + body: str | None = None, + issue_number: int | None = None, +) -> dict[str, Any]: + """Plan the PATCH payload for an issue edit against its pre-image. + + Only fields the caller named are ever put in the payload, so unspecified + fields cannot be overwritten with a stale read. A request whose named fields + already hold the requested values is reported as a no-op with an actionable + reason instead of being sent and reported as a success. + """ + requested = validate_edit_request(title=title, body=body) + number = issue_number if issue_number is not None else current.get("number") + + changes: dict[str, dict[str, Any]] = {} + unchanged: list[str] = [] + for field, value in requested.items(): + before = current.get(field) + if field == "body": + before = before or "" + if before == value: + unchanged.append(field) + else: + changes[field] = {"before": before, "after": value} + + no_op = not changes + payload = {field: requested[field] for field in changes} + + return { + "issue_number": number, + "requested_fields": sorted(requested), + "requested": dict(requested), + "payload": payload, + "changes": changes, + "unchanged_fields": sorted(unchanged), + "no_op": no_op, + "preserved_before": preserved_snapshot(current), + "reasons": ( + [ + "requested " + + ", ".join(sorted(unchanged)) + + " already match the issue's current content; no edit was sent" + ] + if no_op + else [] + ), + "safe_next_action": ( + ( + f"Re-read issue #{number} and call gitea_edit_issue only with " + "content that differs, or drop the call if the issue is already " + "correct." + ) + if no_op + else "" + ), + } + + +def verify_issue_edit( + observed: Mapping[str, Any], + *, + plan: Mapping[str, Any], +) -> dict[str, Any]: + """Read-after-write proof for an applied issue edit. + + Fails closed on two distinct defects: an edited field whose stored value is + not what was requested, and an untouched field that moved anyway. + """ + requested = dict(plan.get("requested") or {}) + number = plan.get("issue_number") + + applied: dict[str, Any] = {} + mismatches: list[dict[str, Any]] = [] + for field, expected in requested.items(): + actual = observed.get(field) + if field == "body": + actual = actual or "" + applied[field] = actual + if actual != expected: + mismatches.append( + {"field": field, "expected": expected, "observed": actual} + ) + + before = dict(plan.get("preserved_before") or {}) + after = preserved_snapshot(observed) + preserved_changed: list[dict[str, Any]] = [ + {"field": field, "before": before.get(field), "after": after.get(field)} + for field in PRESERVED_FIELDS + if before.get(field) != after.get(field) + ] + + reasons: list[str] = [] + for entry in mismatches: + reasons.append( + f"{entry['field']} was not applied: requested " + f"{entry['expected']!r} but the issue stores {entry['observed']!r}" + ) + for entry in preserved_changed: + reasons.append( + f"{entry['field']} changed during the edit: {entry['before']!r} " + f"became {entry['after']!r}; gitea_edit_issue must leave it alone" + ) + + verified = not reasons + return { + "verified": verified, + "applied": applied, + "mismatches": mismatches, + "preserved_before": before, + "preserved_after": after, + "preserved_changed": preserved_changed, + "preserved_intact": not preserved_changed, + "reasons": reasons, + "safe_next_action": ( + "" + if verified + else ( + f"Re-read issue #{number} with gitea_view_issue and reconcile it " + "before treating the edit as applied. Do not retry blindly — the " + "stored content does not match what was requested." + ) + ), + } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 7d0a2c6..613c393 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -2123,6 +2123,7 @@ def _seed_session_context( import issue_work_duplicate_gate # noqa: E402 import issue_workflow_labels # noqa: E402 import terminal_pr_label_cleanup # noqa: E402 # #780 status:pr-open terminal rule +import edit_issue # noqa: E402 # #781 issue title/body edit rule import reviewer_pr_lease # noqa: E402 import post_merge_moot_lease_gate # noqa: E402 # #745 reconciler cleanup gate import merger_lease_adoption # noqa: E402 @@ -11764,6 +11765,175 @@ def gitea_close_issue( } +@mcp.tool() +def gitea_edit_issue( + issue_number: int, + title: str | None = None, + body: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + worktree_path: str | None = None, +) -> dict: + """Edit an issue's title, body, or both through the sanctioned path (#781). + + This is the only MCP path that changes issue content. It PATCHes the issue + endpoint and never the pull-request endpoint: ``gitea_edit_pr`` stays + pull-request-only, and a pull-request number passed here is refused rather + than edited, even though Gitea serves pull requests from ``/issues/{n}``. + + Only the fields named in the call are sent, so labels, state, assignee, and + milestone cannot be overwritten by a stale read. The edit is proven by + read-after-write: the tool re-reads the issue and fails closed if the stored + title/body differ from what was requested, or if any unspecified field moved. + + Args: + issue_number: The issue number to edit. + title: New issue title. Omit to leave the title unchanged. + body: New issue body. Omit to leave the body unchanged; pass '' + deliberately to clear it. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + worktree_path: Optional branches worktree to verify before mutation. + + Returns: + dict with 'success', 'performed', the 'applied' content, and a + 'read_after_write' proof carrying the preserved-field comparison. A + no-op, a pull-request target, a permission block, or a failed + verification all return 'performed'/'success' False with 'reasons' and + a 'safe_next_action'. Invalid input raises ValueError before any + credential or network work. + """ + # Validate BEFORE auth/profile/API setup: a malformed or empty request is a + # pure input error and must not depend on credentials or configuration. + edit_issue.validate_edit_request(title=title, body=body) + + blocked = _profile_permission_block( + task_capability_map.required_permission("edit_issue"), + issue_number=issue_number, + remote=remote, + host=host, + org=org, + repo=repo, + org_explicit=org is not None, + repo_explicit=repo is not None, + ) + if blocked: + return blocked + verify_preflight_purity( + remote, + worktree_path=worktree_path, + task="edit_issue", + org=org, + repo=repo, + ) + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + url = f"{repo_api_url(h, o, r)}/issues/{issue_number}" + + try: + current = api_request("GET", url, auth) + except Exception as exc: # noqa: BLE001 — redact before surfacing + return { + "success": False, + "performed": False, + "issue_number": issue_number, + "reasons": [ + f"could not read issue #{issue_number} before editing: " + f"{_redact(str(exc))}" + ], + "safe_next_action": ( + f"Confirm issue #{issue_number} exists in {o}/{r} with " + "gitea_view_issue, then retry the edit." + ), + } + + target = edit_issue.assess_issue_target(current, issue_number=issue_number) + if not target["is_issue"]: + return { + "success": False, + "performed": False, + "blocked": True, + "issue_number": issue_number, + "target": target, + "reasons": target["reasons"], + "safe_next_action": target["safe_next_action"], + } + + plan = edit_issue.plan_issue_edit( + current, title=title, body=body, issue_number=issue_number + ) + if plan["no_op"]: + return { + "success": False, + "performed": False, + "no_op": True, + "issue_number": issue_number, + "requested_fields": plan["requested_fields"], + "unchanged_fields": plan["unchanged_fields"], + "reasons": plan["reasons"], + "safe_next_action": plan["safe_next_action"], + } + + try: + with _audited("edit_issue", host=h, remote=remote, org=o, repo=r, + issue_number=issue_number, + request_metadata={"fields": plan["requested_fields"]}): + api_request("PATCH", url, auth, plan["payload"]) + except Exception as exc: # noqa: BLE001 — redact before surfacing + return { + "success": False, + "performed": False, + "issue_number": issue_number, + "requested_fields": plan["requested_fields"], + "reasons": [f"issue edit failed: {_redact(str(exc))}"], + "safe_next_action": ( + f"Re-read issue #{issue_number} with gitea_view_issue to confirm " + "nothing was applied, then retry the edit." + ), + } + + # Read-after-write (#781): the PATCH response is not proof. Re-read the + # issue so an unapplied edit, or a preserved field that moved, fails closed. + try: + observed = api_request("GET", url, auth) + except Exception as exc: # noqa: BLE001 — redact before surfacing + return { + "success": False, + "performed": True, + "verified": False, + "issue_number": issue_number, + "requested_fields": plan["requested_fields"], + "reasons": [ + f"issue edit read-back failed: {_redact(str(exc))}" + ], + "safe_next_action": ( + f"Re-read issue #{issue_number} with gitea_view_issue and confirm " + "the applied content before relying on this edit." + ), + } + + verification = edit_issue.verify_issue_edit(observed, plan=plan) + result = { + "success": verification["verified"], + "performed": True, + "verified": verification["verified"], + "issue_number": issue_number, + "requested_fields": plan["requested_fields"], + "changed_fields": sorted(plan["changes"]), + "applied": verification["applied"], + "read_after_write": verification, + "preserved_intact": verification["preserved_intact"], + "reasons": verification["reasons"], + "safe_next_action": verification["safe_next_action"], + } + return _with_optional_url(result, observed.get("html_url")) + + @mcp.tool() def gitea_cleanup_terminal_pr_labels( issue_numbers: list[int], @@ -16002,6 +16172,12 @@ def gitea_assess_master_parity( if parity["stale"] and enforced: out["report"] = master_parity_gate.parity_report(parity) return out + + +# #781: documented in the canonical review workflow as a tool reviewers call +# before workflow load, but the registration decorator had been lost, so the +# documented inventory named something no namespace could reach. +@mcp.tool() def gitea_record_pre_review_command( command: str, cwd: str | None = None, diff --git a/mcp_tool_inventory.py b/mcp_tool_inventory.py new file mode 100644 index 0000000..68d7982 --- /dev/null +++ b/mcp_tool_inventory.py @@ -0,0 +1,195 @@ +"""Documented-vs-registered MCP tool inventory guard (#781). + +The workflow documentation named a ``gitea_edit_issue`` tool that no namespace +had ever registered. Nothing compared the two lists, so an actor could plan a +mutation against a tool that did not exist and only discover it at execution +time — after the work was already scoped around it. + +This module is that comparison, in two directions: + +- :func:`assess_inventory_drift` compares the canonical inventory documented in + ``docs/mcp-tool-inventory.md`` against the tools actually registered on the + MCP server. Either list drifting fails the guard, so a new tool must be + documented and a removed tool must be undocumented in the same change. +- :func:`assess_doc_references` catches the original defect directly: any tool + name a workflow/skill document tells an actor to call must be registered. + +Module and script names legitimately appear in the same prose (``gitea_auth``, +``offline_mcp_runner``), so :data:`NON_TOOL_IDENTIFIERS` names the known +non-tool identifiers explicitly rather than loosening the pattern. + +This module performs no I/O — callers own reading the files. +""" + +from __future__ import annotations + +import re +from typing import Any, Iterable, Mapping + +#: Canonical documented inventory, relative to the repository root. +INVENTORY_DOC_PATH = "docs/mcp-tool-inventory.md" + +#: Delimiters around the generated inventory list in the doc. +INVENTORY_BEGIN_MARKER = "" +INVENTORY_END_MARKER = "" + +#: Backticked identifiers that look like tool names but are modules/scripts. +#: Every entry is a real file in this repository, not an MCP tool. +NON_TOOL_IDENTIFIERS: frozenset[str] = frozenset( + { + "gitea_auth", + "gitea_config", + "gitea_mcp_server", + "mcp_server", + "offline_mcp_helper", + "offline_mcp_runner", + } +) + +#: Prefixes that mark an identifier as a candidate MCP tool name. +TOOL_NAME_PREFIXES: tuple[str, ...] = ("gitea_", "mcp_") + +_INVENTORY_ENTRY = re.compile(r"^-\s+`([A-Za-z_][A-Za-z0-9_]*)`") +_BACKTICKED = re.compile(r"`([A-Za-z_][A-Za-z0-9_]*)`") + + +def parse_documented_inventory(text: str) -> list[str]: + """Return the tool names listed between the inventory markers. + + Raises ``ValueError`` when the markers are missing or out of order, so a + mangled document fails the guard instead of silently documenting nothing. + """ + start = text.find(INVENTORY_BEGIN_MARKER) + end = text.find(INVENTORY_END_MARKER) + if start == -1 or end == -1 or end < start: + raise ValueError( + f"{INVENTORY_DOC_PATH} must contain " + f"'{INVENTORY_BEGIN_MARKER}' followed by " + f"'{INVENTORY_END_MARKER}' (fail closed)." + ) + block = text[start + len(INVENTORY_BEGIN_MARKER) : end] + names: list[str] = [] + for line in block.splitlines(): + match = _INVENTORY_ENTRY.match(line.strip()) + if match: + names.append(match.group(1)) + return names + + +def looks_like_tool_name(identifier: str) -> bool: + """Return whether a backticked identifier is a candidate tool name.""" + if identifier in NON_TOOL_IDENTIFIERS: + return False + return identifier.startswith(TOOL_NAME_PREFIXES) + + +def extract_tool_references(text: str) -> set[str]: + """Return candidate tool names a document tells an actor to call.""" + return { + name + for name in _BACKTICKED.findall(text) + if looks_like_tool_name(name) + } + + +def assess_inventory_drift( + documented: Iterable[str], + registered: Iterable[str], +) -> dict[str, Any]: + """Compare the documented inventory against the registered tool set.""" + documented_list = list(documented) + documented_set = set(documented_list) + registered_set = set(registered) + + duplicates = sorted( + {name for name in documented_list if documented_list.count(name) > 1} + ) + documented_not_registered = sorted(documented_set - registered_set) + registered_not_documented = sorted(registered_set - documented_set) + unsorted = documented_list != sorted(documented_list) + + reasons: list[str] = [] + if documented_not_registered: + reasons.append( + "documented but not registered: " + + ", ".join(documented_not_registered) + ) + if registered_not_documented: + reasons.append( + "registered but not documented: " + + ", ".join(registered_not_documented) + ) + if duplicates: + reasons.append("listed more than once: " + ", ".join(duplicates)) + if unsorted: + reasons.append("inventory entries are not in sorted order") + + in_sync = not reasons + return { + "in_sync": in_sync, + "documented_count": len(documented_set), + "registered_count": len(registered_set), + "documented_not_registered": documented_not_registered, + "registered_not_documented": registered_not_documented, + "duplicates": duplicates, + "sorted": not unsorted, + "reasons": reasons, + "safe_next_action": ( + "" + if in_sync + else ( + f"Update {INVENTORY_DOC_PATH} so the block between the " + "inventory markers lists exactly the registered tools, sorted, " + "one '- `tool_name`' entry per line." + ) + ), + } + + +def assess_doc_references( + references: Mapping[str, Iterable[str]], + registered: Iterable[str], +) -> dict[str, Any]: + """Verify every tool a document names is actually registered. + + *references* maps a document path to the candidate tool names it mentions. + """ + registered_set = set(registered) + unregistered: list[dict[str, Any]] = [] + checked = 0 + for path, names in sorted(references.items()): + for name in sorted(set(names)): + checked += 1 + if name not in registered_set: + unregistered.append({"document": path, "tool": name}) + + clean = not unregistered + reasons = [ + f"{entry['document']} documents '{entry['tool']}', " + "which no namespace registers" + for entry in unregistered + ] + return { + "clean": clean, + "checked_count": checked, + "unregistered": unregistered, + "reasons": reasons, + "safe_next_action": ( + "" + if clean + else ( + "Either register the named tool with @mcp.tool() or correct the " + "document. Documentation must never name a tool an actor cannot " + "reach. If the identifier is a module or script rather than a " + "tool, add it to mcp_tool_inventory.NON_TOOL_IDENTIFIERS." + ) + ), + } + + +def render_inventory_block(registered: Iterable[str]) -> str: + """Render the marker-delimited inventory block for the documentation.""" + lines = [INVENTORY_BEGIN_MARKER, ""] + lines.extend(f"- `{name}`" for name in sorted(set(registered))) + lines.extend(["", INVENTORY_END_MARKER]) + return "\n".join(lines) diff --git a/skills/gitea-workflow/SKILL.md b/skills/gitea-workflow/SKILL.md index 0f39091..7d37251 100644 --- a/skills/gitea-workflow/SKILL.md +++ b/skills/gitea-workflow/SKILL.md @@ -35,3 +35,11 @@ Install for Codex: ``` Preflight via MCP: `mcp_check_workflow_skill_preflight`. + +## Tool inventory + +Which tools actually exist is documented in +[`docs/mcp-tool-inventory.md`](../../docs/mcp-tool-inventory.md), which a test +holds equal to the registered set. Never plan a mutation against a tool that is +not listed there — that is the #781 failure mode, where a documented +`gitea_edit_issue` did not exist until execution time. diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 9f14d7a..44cb731 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -216,6 +216,15 @@ Helpers: `scripts/worktree-start`, `scripts/worktree-review`, - Never place raw tokens in LLM/MCP config. - Use `gitea_whoami` and `gitea_resolve_task_capability` before mutating. +## Tool inventory + +[`docs/mcp-tool-inventory.md`](../../docs/mcp-tool-inventory.md) is the canonical +list of registered tools, held equal to the live registry by a test. A tool that +is not listed there does not exist — do not scope work around it (#781). + +Issue content is edited with `gitea_edit_issue` (title/body only, read-after-write +verified). `gitea_edit_pr` is pull-request-only and never accepts an issue number. + ## Controller Handoff Every task must end with a section titled exactly `Controller Handoff`. Compact diff --git a/task_capability_map.py b/task_capability_map.py index 903505a..23409e0 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -36,6 +36,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.issue.comment", "role": "author", }, + # #781: editing an issue title/body is issue authoring, the same authority + # every other non-create/non-close issue mutation gates on. Deliberately not + # a new operation name: introducing one would silently strip the capability + # from every already-configured author profile. + "edit_issue": { + "permission": "gitea.issue.comment", + "role": "author", + }, # #780: retire status:pr-open after a terminal PR transition. Same label # authority as set_issue_labels — it is a strictly narrower operation. "cleanup_terminal_pr_labels": { @@ -510,6 +518,7 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset( ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = { "gitea_create_issue": "create_issue", "gitea_close_issue": "close_issue", + "gitea_edit_issue": "edit_issue", "gitea_create_issue_comment": "comment_issue", "gitea_mark_issue": "mark_issue", "gitea_set_issue_labels": "set_issue_labels", diff --git a/tests/test_issue_781_edit_issue_tool.py b/tests/test_issue_781_edit_issue_tool.py new file mode 100644 index 0000000..f852a21 --- /dev/null +++ b/tests/test_issue_781_edit_issue_tool.py @@ -0,0 +1,635 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from mutation_profile_fixture import install_deterministic_remote_urls # noqa: E402 + +install_deterministic_remote_urls() +"""#781: sanctioned issue title/body editing, and the documentation drift guard. + +Two defects are covered here. The first is that no MCP path could edit an issue +title or body at all, so an authorized correction had to be recorded as a +comment. The second is why nobody noticed: documentation named a tool that was +never registered, and nothing compared the two lists. +""" +import glob +import json +import os +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import anti_stomp_preflight # noqa: E402 +import edit_issue # noqa: E402 +import mcp_server # noqa: E402 +import mcp_tool_inventory # noqa: E402 +import task_capability_map # noqa: E402 + +REPO_ROOT = Path(__file__).resolve().parent.parent + +CONFIG = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, + } + }, + "profiles": { + "edit-author": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "author-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": ["gitea.read", "gitea.issue.comment"], + "forbidden_operations": [], + "allowed_repositories": [ + "Scaled-Tech-Consulting/Gitea-Tools", + "Example-Org/Example-Repo", + "913443/eAgenda", + ], + "execution_profile": "edit-author", + }, + "read-only-author": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "reader-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": ["gitea.read"], + "forbidden_operations": ["gitea.issue.comment"], + "allowed_repositories": [ + "Scaled-Tech-Consulting/Gitea-Tools", + "Example-Org/Example-Repo", + "913443/eAgenda", + ], + "execution_profile": "read-only-author", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + +ISSUE_NUMBER = 9 +ORIGINAL_TITLE = "fix(mcp): original title" +ORIGINAL_BODY = "Original body.\n" +NEW_TITLE = "fix(mcp): corrected title" +NEW_BODY = "Corrected body.\n" + + +def _registered_tool_names() -> set[str]: + manager = mcp_server.mcp._tool_manager + return set((getattr(manager, "_tools", None) or {}).keys()) + + +# --------------------------------------------------------------------------- +# Rule: request validation +# --------------------------------------------------------------------------- +class TestValidateEditRequest(unittest.TestCase): + def test_no_field_is_rejected(self): + with self.assertRaises(ValueError) as ctx: + edit_issue.validate_edit_request() + self.assertIn("At least one field", str(ctx.exception)) + + def test_blank_title_is_rejected(self): + with self.assertRaises(ValueError) as ctx: + edit_issue.validate_edit_request(title=" ") + self.assertIn("cannot be blank", str(ctx.exception)) + + def test_non_string_title_is_rejected(self): + with self.assertRaises(ValueError): + edit_issue.validate_edit_request(title=42) + + def test_non_string_body_is_rejected(self): + with self.assertRaises(ValueError): + edit_issue.validate_edit_request(body=["not", "a", "string"]) + + def test_empty_body_is_a_legitimate_edit(self): + self.assertEqual(edit_issue.validate_edit_request(body=""), {"body": ""}) + + +# --------------------------------------------------------------------------- +# Rule: planning against the pre-image +# --------------------------------------------------------------------------- +class TestPlanIssueEdit(unittest.TestCase): + def _current(self, **overrides): + issue = { + "number": ISSUE_NUMBER, + "title": ORIGINAL_TITLE, + "body": ORIGINAL_BODY, + "state": "open", + "labels": [{"name": "type:bug"}, {"name": "mcp"}], + "assignees": [{"login": "author-user"}], + "milestone": {"title": "v1.2.0"}, + } + issue.update(overrides) + return issue + + def test_title_only_sends_only_the_title(self): + plan = edit_issue.plan_issue_edit(self._current(), title=NEW_TITLE) + self.assertEqual(plan["payload"], {"title": NEW_TITLE}) + self.assertEqual(plan["requested_fields"], ["title"]) + self.assertFalse(plan["no_op"]) + + def test_body_only_sends_only_the_body(self): + plan = edit_issue.plan_issue_edit(self._current(), body=NEW_BODY) + self.assertEqual(plan["payload"], {"body": NEW_BODY}) + + def test_combined_edit_sends_both(self): + plan = edit_issue.plan_issue_edit( + self._current(), title=NEW_TITLE, body=NEW_BODY + ) + self.assertEqual(plan["payload"], {"title": NEW_TITLE, "body": NEW_BODY}) + self.assertEqual(plan["requested_fields"], ["body", "title"]) + + def test_identical_content_is_an_explicit_no_op(self): + plan = edit_issue.plan_issue_edit( + self._current(), title=ORIGINAL_TITLE, body=ORIGINAL_BODY + ) + self.assertTrue(plan["no_op"]) + self.assertEqual(plan["payload"], {}) + self.assertTrue(plan["reasons"]) + self.assertTrue(plan["safe_next_action"]) + + def test_partially_unchanged_request_sends_only_the_difference(self): + plan = edit_issue.plan_issue_edit( + self._current(), title=ORIGINAL_TITLE, body=NEW_BODY + ) + self.assertFalse(plan["no_op"]) + self.assertEqual(plan["payload"], {"body": NEW_BODY}) + self.assertEqual(plan["unchanged_fields"], ["title"]) + + def test_missing_body_is_compared_as_empty(self): + current = self._current() + current.pop("body") + plan = edit_issue.plan_issue_edit(current, body="") + self.assertTrue(plan["no_op"]) + + def test_preserved_snapshot_captures_untouched_fields(self): + plan = edit_issue.plan_issue_edit(self._current(), title=NEW_TITLE) + before = plan["preserved_before"] + self.assertEqual(before["state"], "open") + self.assertEqual(before["labels"], ["type:bug", "mcp"]) + self.assertEqual(before["assignees"], ["author-user"]) + self.assertEqual(before["milestone"], "v1.2.0") + + +# --------------------------------------------------------------------------- +# Rule: pull requests are refused +# --------------------------------------------------------------------------- +class TestAssessIssueTarget(unittest.TestCase): + def test_issue_is_accepted(self): + target = edit_issue.assess_issue_target( + {"number": 9, "title": "t"}, issue_number=9 + ) + self.assertTrue(target["is_issue"]) + self.assertEqual(target["reasons"], []) + + def test_pull_request_is_refused_with_a_next_action(self): + target = edit_issue.assess_issue_target( + {"number": 9, "pull_request": {"merged": False}}, issue_number=9 + ) + self.assertFalse(target["is_issue"]) + self.assertTrue(target["is_pull_request"]) + self.assertIn("gitea_edit_pr", target["safe_next_action"]) + + +# --------------------------------------------------------------------------- +# Rule: read-after-write verification +# --------------------------------------------------------------------------- +class TestVerifyIssueEdit(unittest.TestCase): + def _plan(self, **kwargs): + current = { + "number": ISSUE_NUMBER, + "title": ORIGINAL_TITLE, + "body": ORIGINAL_BODY, + "state": "open", + "labels": [{"name": "type:bug"}], + "assignees": [], + "milestone": None, + } + return edit_issue.plan_issue_edit(current, **kwargs) + + def test_applied_content_verifies(self): + plan = self._plan(title=NEW_TITLE) + observed = { + "title": NEW_TITLE, + "body": ORIGINAL_BODY, + "state": "open", + "labels": [{"name": "type:bug"}], + "assignees": [], + "milestone": None, + } + result = edit_issue.verify_issue_edit(observed, plan=plan) + self.assertTrue(result["verified"]) + self.assertTrue(result["preserved_intact"]) + self.assertEqual(result["applied"], {"title": NEW_TITLE}) + + def test_unapplied_content_fails_closed(self): + plan = self._plan(title=NEW_TITLE) + observed = { + "title": ORIGINAL_TITLE, + "state": "open", + "labels": [{"name": "type:bug"}], + } + result = edit_issue.verify_issue_edit(observed, plan=plan) + self.assertFalse(result["verified"]) + self.assertEqual(result["mismatches"][0]["field"], "title") + self.assertTrue(result["safe_next_action"]) + + def test_dropped_label_fails_closed(self): + plan = self._plan(title=NEW_TITLE) + observed = {"title": NEW_TITLE, "state": "open", "labels": []} + result = edit_issue.verify_issue_edit(observed, plan=plan) + self.assertFalse(result["verified"]) + self.assertFalse(result["preserved_intact"]) + self.assertEqual(result["preserved_changed"][0]["field"], "labels") + + def test_changed_state_fails_closed(self): + plan = self._plan(body=NEW_BODY) + observed = { + "body": NEW_BODY, + "state": "closed", + "labels": [{"name": "type:bug"}], + } + result = edit_issue.verify_issue_edit(observed, plan=plan) + self.assertFalse(result["verified"]) + self.assertEqual(result["preserved_changed"][0]["field"], "state") + + +# --------------------------------------------------------------------------- +# Tool: gitea_edit_issue against a fake Gitea +# --------------------------------------------------------------------------- +class _EditIssueToolHarness(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)) + + self.issue = { + "number": ISSUE_NUMBER, + "title": ORIGINAL_TITLE, + "body": ORIGINAL_BODY, + "state": "open", + "labels": [{"name": "type:bug"}, {"name": "mcp"}], + "assignees": [{"login": "author-user"}], + "milestone": {"title": "v1.2.0"}, + "html_url": "https://gitea.example.com/Example-Org/Example-Repo/issues/9", + } + self.calls: list[tuple[str, str]] = [] + self.patched_payloads: list[dict] = [] + + patch("gitea_audit.audit_enabled", return_value=False).start() + patch("mcp_server.get_auth_header", return_value="token author-pass").start() + patch("mcp_server.api_request", side_effect=self._api).start() + self.addCleanup(patch.stopall) + + def tearDown(self): + self._remotes.stop() + mcp_server._IDENTITY_CACHE.clear() + self._dir.cleanup() + + def _api(self, method, url, auth, payload=None): + self.calls.append((method, url)) + if url.endswith("/user"): + return {"login": "author-user"} + if "/issues/" in url: + if method == "GET": + return dict(self.issue) + if method == "PATCH": + self.patched_payloads.append(dict(payload or {})) + self.issue.update(payload or {}) + return dict(self.issue) + raise AssertionError(f"unexpected API call: {method} {url}") + + def _env(self, profile: str = "edit-author") -> dict: + return { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + "PYTEST_CURRENT_TEST": os.environ.get( + "PYTEST_CURRENT_TEST", "issue_781_edit_issue" + ), + } + + def _edit(self, profile: str = "edit-author", **kwargs): + with patch.dict(os.environ, self._env(profile), clear=True): + return mcp_server.gitea_edit_issue( + issue_number=ISSUE_NUMBER, remote="prgs", **kwargs + ) + + def _patch_methods(self) -> list[str]: + return [method for method, _url in self.calls if method == "PATCH"] + + +class TestEditIssueSucceeds(_EditIssueToolHarness): + def test_title_only_edit(self): + result = self._edit(title=NEW_TITLE) + self.assertTrue(result["success"]) + self.assertTrue(result["verified"]) + self.assertEqual(result["changed_fields"], ["title"]) + self.assertEqual(self.patched_payloads, [{"title": NEW_TITLE}]) + self.assertEqual(self.issue["title"], NEW_TITLE) + self.assertEqual(self.issue["body"], ORIGINAL_BODY) + + def test_body_only_edit(self): + result = self._edit(body=NEW_BODY) + self.assertTrue(result["success"]) + self.assertEqual(self.patched_payloads, [{"body": NEW_BODY}]) + self.assertEqual(self.issue["title"], ORIGINAL_TITLE) + self.assertEqual(self.issue["body"], NEW_BODY) + + def test_combined_edit(self): + result = self._edit(title=NEW_TITLE, body=NEW_BODY) + self.assertTrue(result["success"]) + self.assertEqual( + self.patched_payloads, [{"title": NEW_TITLE, "body": NEW_BODY}] + ) + self.assertEqual(result["applied"], {"title": NEW_TITLE, "body": NEW_BODY}) + + def test_body_can_be_cleared(self): + result = self._edit(body="") + self.assertTrue(result["success"]) + self.assertEqual(self.issue["body"], "") + + def test_targets_the_issue_endpoint_never_the_pull_endpoint(self): + self._edit(title=NEW_TITLE) + patched = [url for method, url in self.calls if method == "PATCH"] + self.assertTrue(patched) + for url in patched: + self.assertIn("/issues/", url) + self.assertNotIn("/pulls/", url) + + def test_labels_state_assignee_and_milestone_are_provably_unchanged(self): + result = self._edit(title=NEW_TITLE) + proof = result["read_after_write"] + self.assertTrue(proof["preserved_intact"]) + self.assertEqual(proof["preserved_before"], proof["preserved_after"]) + self.assertEqual(proof["preserved_after"]["labels"], ["type:bug", "mcp"]) + self.assertEqual(proof["preserved_after"]["state"], "open") + self.assertEqual(proof["preserved_after"]["assignees"], ["author-user"]) + self.assertEqual(proof["preserved_after"]["milestone"], "v1.2.0") + + def test_read_after_write_re_reads_the_issue(self): + self._edit(title=NEW_TITLE) + issue_calls = [method for method, url in self.calls if "/issues/" in url] + self.assertEqual(issue_calls, ["GET", "PATCH", "GET"]) + + +class TestEditIssueFailsClosed(_EditIssueToolHarness): + def test_no_op_request_is_rejected_without_a_patch(self): + result = self._edit(title=ORIGINAL_TITLE) + self.assertFalse(result["success"]) + self.assertFalse(result["performed"]) + self.assertTrue(result["no_op"]) + self.assertEqual(self._patch_methods(), []) + self.assertTrue(result["reasons"]) + self.assertTrue(result["safe_next_action"]) + + def test_invalid_request_raises_before_any_api_call(self): + with self.assertRaises(ValueError): + self._edit() + self.assertEqual(self.calls, []) + + def test_blank_title_raises_before_any_api_call(self): + with self.assertRaises(ValueError): + self._edit(title=" ") + self.assertEqual(self.calls, []) + + def test_authorization_failure_blocks_before_any_api_call(self): + result = self._edit(profile="read-only-author", title=NEW_TITLE) + self.assertFalse(result["success"]) + self.assertFalse(result["performed"]) + self.assertIn("permission_report", result) + self.assertEqual( + result["permission_report"]["missing_permission"], + task_capability_map.required_permission("edit_issue"), + ) + self.assertEqual(self.calls, []) + + def test_pull_request_target_is_refused_without_a_patch(self): + self.issue["pull_request"] = {"merged": False} + result = self._edit(title=NEW_TITLE) + self.assertFalse(result["success"]) + self.assertFalse(result["performed"]) + self.assertEqual(self._patch_methods(), []) + self.assertIn("gitea_edit_pr", result["safe_next_action"]) + + def test_pre_read_transport_error_is_reported(self): + def boom(method, url, auth, payload=None): + raise RuntimeError("connection reset by peer") + + with patch("mcp_server.api_request", side_effect=boom): + result = self._edit(title=NEW_TITLE) + self.assertFalse(result["success"]) + self.assertFalse(result["performed"]) + self.assertTrue(result["reasons"]) + self.assertTrue(result["safe_next_action"]) + + def test_patch_transport_error_is_reported_not_swallowed(self): + def flaky(method, url, auth, payload=None): + if method == "PATCH": + raise RuntimeError("gitea exploded") + return self._api(method, url, auth, payload) + + with patch("mcp_server.api_request", side_effect=flaky): + result = self._edit(title=NEW_TITLE) + self.assertFalse(result["success"]) + self.assertFalse(result["performed"]) + self.assertIn("issue edit failed", result["reasons"][0]) + self.assertEqual(self.issue["title"], ORIGINAL_TITLE) + + def test_read_back_transport_error_reports_an_unverified_edit(self): + state = {"gets": 0} + + def flaky(method, url, auth, payload=None): + if method == "GET" and "/issues/" in url: + state["gets"] += 1 + if state["gets"] > 1: + raise RuntimeError("read timed out") + return self._api(method, url, auth, payload) + + with patch("mcp_server.api_request", side_effect=flaky): + result = self._edit(title=NEW_TITLE) + self.assertFalse(result["success"]) + self.assertTrue(result["performed"]) + self.assertFalse(result["verified"]) + self.assertTrue(result["safe_next_action"]) + + def test_unapplied_edit_fails_verification(self): + def sticky(method, url, auth, payload=None): + if method == "PATCH": + self.calls.append((method, url)) + # Report success but store nothing. + return dict(self.issue) + return self._api(method, url, auth, payload) + + with patch("mcp_server.api_request", side_effect=sticky): + result = self._edit(title=NEW_TITLE) + self.assertFalse(result["success"]) + self.assertTrue(result["performed"]) + self.assertFalse(result["verified"]) + self.assertEqual( + result["read_after_write"]["mismatches"][0]["field"], "title" + ) + + def test_edit_that_drops_a_label_fails_verification(self): + def label_eating(method, url, auth, payload=None): + if method == "PATCH": + self.calls.append((method, url)) + self.issue.update(payload or {}) + self.issue["labels"] = [] + return dict(self.issue) + return self._api(method, url, auth, payload) + + with patch("mcp_server.api_request", side_effect=label_eating): + result = self._edit(title=NEW_TITLE) + self.assertFalse(result["success"]) + self.assertFalse(result["read_after_write"]["preserved_intact"]) + + +# --------------------------------------------------------------------------- +# Registration and gate wiring +# --------------------------------------------------------------------------- +class TestEditIssueRegistrationAndGates(unittest.TestCase): + def test_tool_is_registered(self): + self.assertIn("gitea_edit_issue", _registered_tool_names()) + + def test_resolver_task_exists_with_author_role(self): + self.assertEqual( + task_capability_map.required_permission("edit_issue"), + "gitea.issue.comment", + ) + self.assertEqual(task_capability_map.required_role("edit_issue"), "author") + + def test_tool_gate_matches_the_resolver_task(self): + self.assertEqual( + task_capability_map.ISSUE_MUTATION_TOOL_TASKS["gitea_edit_issue"], + "edit_issue", + ) + self.assertEqual( + task_capability_map.tool_required_permission("gitea_edit_issue"), + task_capability_map.required_permission("edit_issue"), + ) + + def test_declared_as_an_anti_stomp_mutation_task(self): + self.assertIn("edit_issue", anti_stomp_preflight.MUTATION_TASKS) + + def test_edit_pr_remains_pull_request_only(self): + import inspect + + params = inspect.signature(mcp_server.gitea_edit_pr).parameters + self.assertIn("pr_number", params) + self.assertNotIn("issue_number", params) + + def test_edit_issue_cannot_change_state_or_labels(self): + import inspect + + params = inspect.signature(mcp_server.gitea_edit_issue).parameters + self.assertEqual( + [name for name in params if name in ("title", "body")], + ["title", "body"], + ) + for forbidden in ("state", "labels", "assignee", "assignees", "milestone"): + self.assertNotIn(forbidden, params) + + +# --------------------------------------------------------------------------- +# The drift guard itself +# --------------------------------------------------------------------------- +class TestInventoryDriftRule(unittest.TestCase): + def test_missing_markers_fail_closed(self): + with self.assertRaises(ValueError): + mcp_tool_inventory.parse_documented_inventory("no markers here") + + def test_documented_but_unregistered_is_drift(self): + result = mcp_tool_inventory.assess_inventory_drift( + ["gitea_edit_issue", "gitea_view_issue"], ["gitea_view_issue"] + ) + self.assertFalse(result["in_sync"]) + self.assertEqual(result["documented_not_registered"], ["gitea_edit_issue"]) + self.assertTrue(result["safe_next_action"]) + + def test_registered_but_undocumented_is_drift(self): + result = mcp_tool_inventory.assess_inventory_drift( + ["gitea_view_issue"], ["gitea_view_issue", "gitea_edit_issue"] + ) + self.assertFalse(result["in_sync"]) + self.assertEqual(result["registered_not_documented"], ["gitea_edit_issue"]) + + def test_unsorted_inventory_is_drift(self): + result = mcp_tool_inventory.assess_inventory_drift( + ["gitea_view_issue", "gitea_edit_issue"], + ["gitea_view_issue", "gitea_edit_issue"], + ) + self.assertFalse(result["in_sync"]) + self.assertFalse(result["sorted"]) + + def test_module_names_are_not_treated_as_tools(self): + self.assertFalse(mcp_tool_inventory.looks_like_tool_name("gitea_auth")) + self.assertTrue(mcp_tool_inventory.looks_like_tool_name("gitea_view_issue")) + + def test_unregistered_doc_reference_is_reported(self): + result = mcp_tool_inventory.assess_doc_references( + {"skills/example.md": {"gitea_edit_issue"}}, ["gitea_view_issue"] + ) + self.assertFalse(result["clean"]) + self.assertEqual(result["unregistered"][0]["tool"], "gitea_edit_issue") + + def test_rendered_block_round_trips(self): + block = mcp_tool_inventory.render_inventory_block( + ["gitea_view_issue", "gitea_edit_issue"] + ) + self.assertEqual( + mcp_tool_inventory.parse_documented_inventory(block), + ["gitea_edit_issue", "gitea_view_issue"], + ) + + +class TestDocumentationMatchesRegistry(unittest.TestCase): + """The live guard: docs and the registry must not drift apart.""" + + def test_documented_inventory_equals_registered_tools(self): + doc = REPO_ROOT / mcp_tool_inventory.INVENTORY_DOC_PATH + self.assertTrue(doc.exists(), f"{doc} is missing") + documented = mcp_tool_inventory.parse_documented_inventory( + doc.read_text(encoding="utf-8") + ) + result = mcp_tool_inventory.assess_inventory_drift( + documented, _registered_tool_names() + ) + self.assertTrue(result["in_sync"], "; ".join(result["reasons"])) + + def test_every_tool_named_in_the_skills_is_registered(self): + references: dict[str, set[str]] = {} + pattern = str(REPO_ROOT / "skills" / "**" / "*.md") + paths = glob.glob(pattern, recursive=True) + self.assertTrue(paths, "no skill documents found to check") + for path in paths: + text = Path(path).read_text(encoding="utf-8") + names = mcp_tool_inventory.extract_tool_references(text) + if names: + references[str(Path(path).relative_to(REPO_ROOT))] = names + result = mcp_tool_inventory.assess_doc_references( + references, _registered_tool_names() + ) + self.assertTrue(result["clean"], "; ".join(result["reasons"])) + + +if __name__ == "__main__": + unittest.main()