Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 67 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'fix(mcp): add sanctioned gitea_edit_issue and a doc/registry drift guard (Closes #781)' (#783) from fix/issue-781-mcp-edit-issue-tool into master

Browse Source
This commit was merged in pull request #783.
This commit is contained in:
sysadmin
2026-07-21 16:15:01 -05:00
parent 8e149e6cfa a002864a06
commit 300e8acd13
9 changed files with 1473 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
anti_stomp_preflight.py
+1
View File
@@ -77,6 +77,7 @@ MUTATION_TASKS = frozenset({
"create_issue",
"comment_issue",
"close_issue",
"edit_issue",
"mark_issue",
"lock_issue",
"set_issue_labels",
docs/mcp-tool-inventory.md
+169
View File
@@ -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.
<!-- BEGIN REGISTERED TOOL INVENTORY -->
- `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`
<!-- END REGISTERED TOOL INVENTORY -->
## 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.
edit_issue.py
+271
View File
@@ -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."
)
),
}
gitea_mcp_server.py
+176
View File
@@ -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,
mcp_tool_inventory.py
+195
View File
@@ -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 = "<!-- BEGIN REGISTERED TOOL INVENTORY -->"
INVENTORY_END_MARKER = "<!-- END REGISTERED TOOL INVENTORY -->"
#: 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)
skills/gitea-workflow/SKILL.md
+8
View File
@@ -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.
skills/llm-project-workflow/SKILL.md
+9
View File
@@ -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
task_capability_map.py
+9
View File
@@ -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",
tests/test_issue_781_edit_issue_tool.py
+635
View File
@@ -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()