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

Merge prgs/master into feat/issue-269-merged-pr-cleanup-reconcile

Browse Source 
Resolve task_capability_map.py conflict by keeping all three capability
entries: commit_files and gitea_commit_files (gitea.repo.commit, #262,
from master) and reconcile_merged_cleanups (gitea.read, #269, this
branch). Capability boundaries unchanged — reconciliation reporting stays
read-only; no close/merge capability broadened.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 04:37:11 -04:00
co-authored by Claude Opus 4.8
parent 4cc31a42f9 fd396df334
commit 3a490521e4
39 changed files with 6471 additions and 690 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+285 -13
View File
@@ -171,6 +171,9 @@ _preflight_whoami_violation_files: list[str] = []
_preflight_capability_violation_files: list[str] = []
_preflight_reviewer_violation_files: list[str] = []
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
def _preflight_in_test_mode() -> bool:
return "pytest" in sys.modules or "unittest" in sys.modules
@@ -184,19 +187,47 @@ def _ensure_process_start_porcelain() -> str:
return _process_start_porcelain
def _get_workspace_porcelain() -> str:
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the workspace root inspected by pre-flight guards."""
path = (worktree_path or "").strip()
if not path:
path = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip()
if not path:
path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
if not path:
path = PROJECT_ROOT
return os.path.realpath(os.path.abspath(path))
def _get_git_root(path: str) -> str | None:
try:
res = subprocess.run(
["git", "-C", path, "rev-parse", "--show-toplevel"],
capture_output=True,
text=True,
check=False,
)
except Exception:
return None
if res.returncode != 0:
return None
return (res.stdout or "").strip() or None
def _get_workspace_porcelain(worktree_path: str | None = None) -> str:
"""Return tracked-workspace porcelain for pre-flight attribution."""
if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
return " M __gitea_test_force_dirty__.py\n"
override = os.environ.get("GITEA_TEST_PORCELAIN")
if override is not None:
return override
workspace = _resolve_preflight_workspace_path(worktree_path)
try:
res = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True,
text=True,
cwd=PROJECT_ROOT,
cwd=workspace,
)
return res.stdout or ""
except Exception:
@@ -234,7 +265,35 @@ def _format_preflight_files(files: list[str]) -> str:
return ", ".join(files)
def assess_preflight_status() -> dict:
def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
workspace = _resolve_preflight_workspace_path(worktree_path)
inspected_root = _get_git_root(workspace)
control_root = os.path.realpath(PROJECT_ROOT)
active_root = os.path.realpath(inspected_root or workspace)
if active_root == control_root:
dirty_scope = "control checkout"
else:
dirty_scope = "active task workspace"
return {
"mcp_server_process_root": control_root,
"active_task_workspace_root": active_root,
"inspected_git_root": inspected_root,
"dirty_files": list(dirty_files),
"dirty_scope": dirty_scope,
}
def _format_preflight_workspace_details(details: dict) -> str:
return (
f"MCP server process root: {details.get('mcp_server_process_root')}; "
f"active task workspace root: {details.get('active_task_workspace_root')}; "
f"inspected git root: {details.get('inspected_git_root')}; "
f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
f"dirty scope: {details.get('dirty_scope')}"
)
def assess_preflight_status(worktree_path: str | None = None) -> dict:
"""Non-throwing pre-flight readiness for runtime-context alignment (#252)."""
reasons: list[str] = []
if not _preflight_whoami_called:
@@ -245,6 +304,25 @@ def assess_preflight_status() -> dict:
reasons.append(
"Task capability (gitea_resolve_task_capability) has not been resolved"
)
workspace_details = None
if worktree_path:
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
workspace_details = _preflight_workspace_details(worktree_path, dirty_files)
if dirty_files:
reasons.append(
"Active task workspace has tracked file edits before mutation "
f"({_format_preflight_workspace_details(workspace_details)})"
)
return {
"preflight_ready": not reasons,
"preflight_block_reasons": reasons,
"preflight_whoami_verified": _preflight_whoami_called,
"preflight_capability_resolved": _preflight_capability_called,
"preflight_whoami_violation_files": [],
"preflight_capability_violation_files": [],
"preflight_reviewer_violation_files": [],
"preflight_workspace": workspace_details,
}
if _preflight_whoami_violation:
reasons.append(
"Workspace file edits occurred before gitea_whoami verification "
@@ -260,14 +338,25 @@ def assess_preflight_status() -> dict:
"Reviewer profile modified tracked workspace files after capability resolution "
f"(offending files: {_format_preflight_files(_preflight_reviewer_violation_files)})"
)
warnings: list[str] = []
agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
_get_workspace_porcelain()
)
if agent_artifacts:
warnings.append(
"Agent temp artifacts detected at repo root (delete before mutations): "
f"{_format_preflight_files(agent_artifacts)}"
)
return {
"preflight_ready": not reasons,
"preflight_block_reasons": reasons,
"preflight_warnings": warnings,
"preflight_whoami_verified": _preflight_whoami_called,
"preflight_capability_resolved": _preflight_capability_called,
"preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
"preflight_capability_violation_files": list(_preflight_capability_violation_files),
"preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files),
"preflight_workspace": _preflight_workspace_details(None, []),
}
@@ -308,7 +397,7 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
_preflight_resolved_role = resolved_role
def verify_preflight_purity(remote: str | None = None):
def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
"""Verify that identity and capability were verified prior to session edits."""
global _preflight_reviewer_violation_files
@@ -328,6 +417,17 @@ def verify_preflight_purity(remote: str | None = None):
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
)
if worktree_path:
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
if dirty_files:
details = _preflight_workspace_details(worktree_path, dirty_files)
raise RuntimeError(
"Pre-flight order violation: Active task workspace has tracked "
"file edits before mutation (fail closed). "
f"{_format_preflight_workspace_details(details)}"
)
return
if _preflight_whoami_violation:
raise RuntimeError(
"Pre-flight order violation: Workspace file edits occurred before "
@@ -373,6 +473,7 @@ import role_session_router # noqa: E402
import role_namespace_gate # noqa: E402
import task_capability_map # noqa: E402
import review_proofs # noqa: E402
import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import merged_cleanup_reconcile # noqa: E402
@@ -729,6 +830,8 @@ def gitea_create_issue(
Returns:
dict with 'number' of the created issue ('url' only with the reveal opt-in).
"""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
"create_issue"
)
@@ -750,8 +853,6 @@ def gitea_create_issue(
if blocked:
return blocked
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
closed_issues = api_get_all(
@@ -819,14 +920,23 @@ def gitea_lock_issue(
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
)
blocked = _profile_permission_block(
task_capability_map.required_permission("lock_issue"))
if blocked:
return blocked
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
worktree_path, PROJECT_ROOT
)
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
verify_preflight_purity(remote, worktree_path=resolved_worktree)
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
worktree_path=resolved_worktree,
current_branch=git_state.get("current_branch"),
porcelain_status=git_state.get("porcelain_status") or "",
base_equivalent=git_state.get("base_equivalent"),
inspected_git_root=git_state.get("inspected_git_root"),
base_branch=git_state.get("base_branch"),
)
if lock_assessment["block"]:
raise RuntimeError(
@@ -878,7 +988,10 @@ def gitea_lock_issue(
except Exception as e:
raise RuntimeError(f"Could not write issue lock file: {e}")
return {
agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
git_state.get("porcelain_status") or ""
)
result = {
"success": True,
"message": (
f"Successfully locked issue #{issue_number} to branch '{branch_name}' "
@@ -888,6 +1001,12 @@ def gitea_lock_issue(
"branch_name": branch_name,
"worktree_path": resolved_worktree,
}
if agent_artifacts:
result["warnings"] = [
"Agent temp artifacts at repo root (delete before implementation): "
+ ", ".join(agent_artifacts)
]
return result
@mcp.tool()
@@ -2276,6 +2395,106 @@ def gitea_get_file(
}
def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[dict]]:
import base64
import json
import os
processed_files = []
source_proofs = []
lock_data = {}
if os.path.exists(ISSUE_LOCK_FILE):
try:
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
lock_data = json.load(f)
except Exception:
pass
locked_worktree = lock_data.get("worktree_path")
if locked_worktree:
locked_worktree = os.path.realpath(locked_worktree)
for f in files:
f_copy = dict(f)
path = f_copy.get("path", "")
content_keys = [k for k in ["content", "content_plain", "workspace_path", "local_path"] if k in f_copy]
if len(content_keys) > 1:
raise ValueError(
f"Multiple content sources specified for file '{path}'; provide exactly one of "
f"'content', 'content_plain', 'workspace_path', or 'local_path' (fail closed)"
)
source = "none"
if "content_plain" in f_copy:
content_plain = f_copy.pop("content_plain")
if content_plain is not None:
content_bytes = content_plain.encode("utf-8")
f_copy["content"] = base64.b64encode(content_bytes).decode("utf-8")
source = "inline_plain"
elif "workspace_path" in f_copy:
workspace_path = f_copy.pop("workspace_path")
if not locked_worktree:
raise RuntimeError(
f"Issue lock is missing or does not define a worktree path. "
f"Cannot resolve workspace_path '{workspace_path}' (fail closed)."
)
target_path = os.path.realpath(os.path.abspath(os.path.join(locked_worktree, workspace_path)))
is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
if not (is_inside or is_tmp):
raise ValueError(
f"workspace_path '{workspace_path}' resolves to '{target_path}' which falls outside "
f"of locked worktree '{locked_worktree}' (fail closed)"
)
if not os.path.exists(target_path):
raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
with open(target_path, "rb") as fh:
file_bytes = fh.read()
f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
source = "workspace_path"
elif "local_path" in f_copy:
local_path = f_copy.pop("local_path")
if not locked_worktree:
raise RuntimeError(
f"Issue lock is missing or does not define a worktree path. "
f"Cannot resolve local_path '{local_path}' (fail closed)."
)
target_path = os.path.realpath(os.path.abspath(local_path))
is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
if not (is_inside or is_tmp):
raise ValueError(
f"local_path '{local_path}' resolves to '{target_path}' which falls outside "
f"of locked worktree '{locked_worktree}' (fail closed)"
)
if not os.path.exists(target_path):
raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
with open(target_path, "rb") as fh:
file_bytes = fh.read()
f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
source = "local_path"
elif "content" in f_copy:
source = "inline_base64"
processed_files.append(f_copy)
source_proofs.append({
"path": path,
"source": source
})
return processed_files, source_proofs
@mcp.tool()
def gitea_commit_files(
files: list[dict],
@@ -2290,7 +2509,7 @@ def gitea_commit_files(
"""Commit changes to multiple files in a Gitea repository in a single atomic commit.
Args:
files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and 'content' (base64 encoded for create/update), and optionally 'sha' (required for update/delete) or 'from_path' (for rename).
files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and one of content payload sources: 'content' (base64), 'content_plain', 'workspace_path', or 'local_path'.
message: The commit message.
branch: Optional existing branch to start/commit from.
new_branch: Optional new branch name to create for this commit.
@@ -2302,13 +2521,38 @@ def gitea_commit_files(
Returns:
dict with success status and commit/branch information.
"""
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
"commit_files"
)
if not ok:
return {
"success": False,
"performed": False,
"commit": "",
"branch": "",
"reasons": block_reasons,
}
blocked = _namespace_mutation_block(
"commit_files", commit="", branch="", remote=remote
)
if blocked:
return blocked
blocked = _profile_permission_block(
task_capability_map.required_permission("commit_files"),
commit="", branch="",
)
if blocked:
return blocked
verify_preflight_purity(remote)
processed_files, source_proofs = _prepare_commit_payload_files(files)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/contents"
payload = {
"files": files,
"files": processed_files,
"message": message,
}
if branch is not None:
@@ -2319,13 +2563,14 @@ def gitea_commit_files(
with _audited("commit_files", host=h, remote=remote, org=o, repo=r,
target_branch=(new_branch or branch),
request_metadata={"message": message,
"paths": [f.get("path") for f in files],
"operations": [f.get("operation") for f in files]}):
"paths": [f.get("path") for f in processed_files],
"operations": [f.get("operation") for f in processed_files]}):
data = api_request("POST", url, auth, payload)
return {
"success": True,
"commit": data.get("commit", {}).get("sha", ""),
"branch": data.get("branch", {}).get("name", ""),
"content_source_proof": source_proofs,
}
@@ -3619,6 +3864,29 @@ _GUIDE_RULES = {
"small fixes, review fixes, conflicts, emergencies). Main checkout: "
"read-only inspect, fetch, create worktrees, post-merge stable "
"update, explicit repair only."),
"shell_spawn_hard_stop": (
"A shell result of exit_code: -1 with empty stdout/stderr is an "
"executor spawn failure, not a command failure. Probe once (echo/pwd); "
"if the probe fails, mark shell unavailable for the session. After "
"two consecutive spawn failures, hard-stop all shell use — never "
"retry the same failing spawn — and emit a recovery report: restart "
"the session, kill hung background terminals, prefer MCP-native "
"paths (e.g. gitea_commit_files) for remaining mutations. Shell "
"unavailability never authorizes WebFetch/browser/manual-encoding "
"fallbacks (#258)."),
"subagent_delegation": (
"Deterministic write workflows (issue claim, branch creation, code "
"edits, commits, PR creation, review, merge, cleanup) run inline in "
"the parent session — subagents are blocked for them unless "
"explicitly allowed with a recorded justification. An authorized "
"write subagent must inherit the full gate context: issue lock, "
"branch/worktree path, identity/profile, allowed tool class, "
"command deny list, validation ledger requirement, and final report "
"schema. Subagent output is accepted only with the same proof "
"fields as the parent workflow; read-only delegation (search, "
"inventory, summarize) needs no explicit authorization. Enforced "
"fail closed via subagent_gate.assess_subagent_delegation and "
"validate_subagent_report (#266)."),
}
_COMMON_WORKFLOWS = [
@@ -4327,6 +4595,7 @@ def _build_runtime_task_capabilities(
def gitea_get_runtime_context(
remote: str = "dadeschools",
host: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Read-only: explicit visibility into active profile, configuration model, and eligibility.
@@ -4414,7 +4683,7 @@ def gitea_get_runtime_context(
allowed, forbidden, config
)
preflight = assess_preflight_status()
preflight = assess_preflight_status(worktree_path)
if not preflight["preflight_ready"]:
safe_next_action = (
"Complete pre-flight verification before mutating: call gitea_whoami, then "
@@ -4438,6 +4707,7 @@ def gitea_get_runtime_context(
"safe_next_action": safe_next_action,
"preflight_ready": preflight["preflight_ready"],
"preflight_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"),
"session_capabilities": session_capabilities,
}
@@ -4689,6 +4959,7 @@ def gitea_mark_issue(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Claim or release an issue via the status:in-progress label.
@@ -4701,6 +4972,7 @@ def gitea_mark_issue(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Active task worktree to inspect for pre-flight purity.
Returns:
dict with 'success' boolean and 'message'.
@@ -4712,7 +4984,7 @@ def gitea_mark_issue(
task_capability_map.required_permission("mark_issue"))
if blocked:
return blocked
verify_preflight_purity(remote)
verify_preflight_purity(remote, worktree_path=worktree_path)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)