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

Merge branch 'master' into feat/issue-285-live-infra-stop-recompute

Browse Source
This commit is contained in:
sysadmin
2026-07-07 01:00:48 -05:00
parent 5af1b796ef 042783076f
commit 988a5cd975
10 changed files with 1014 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+151 -5
View File
@@ -260,9 +260,19 @@ 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),
@@ -373,6 +383,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
@@ -877,7 +888,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}' "
@@ -887,6 +901,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()
@@ -2275,6 +2295,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],
@@ -2289,7 +2409,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.
@@ -2301,13 +2421,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:
@@ -2318,13 +2463,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,
}