sysadmin
749e480baf
Expose read-only assess/scan tools and capability planning so already-landed open PRs can be reconciled without review or merge. Register the gitea-reconcile-landed-pr skill and workflow-source verification.
366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""Pre-task role/session router (#206).
|
|
|
|
Classifies a declared task type against the active MCP profile/session and
|
|
returns a route result before any downstream mutation tools run.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import os
|
|
|
|
ROUTE_ALLOWED = "allowed_current_session"
|
|
ROUTE_WRONG_ROLE = "wrong_role_stop"
|
|
ROUTE_TO_AUTHOR = "route_to_author_session"
|
|
ROUTE_TO_REVIEWER = "route_to_reviewer_session"
|
|
ROUTE_AMBIGUOUS = "ambiguous_task_stop"
|
|
ROUTE_INFRA_STOP = "infra_stop"
|
|
|
|
_CONFLICT_HEAD = b"<" * 7 + b" "
|
|
_CONFLICT_TAIL = b">" * 7 + b" "
|
|
_CONFLICT_SEPARATOR = b"=" * 7
|
|
|
|
|
|
def python_bytes_have_conflict_markers(content: bytes) -> bool:
|
|
"""Return True when *content* contains git merge-conflict marker lines."""
|
|
for line in content.splitlines():
|
|
stripped = line.rstrip(b"\r\n")
|
|
if stripped.startswith(_CONFLICT_HEAD):
|
|
return True
|
|
if stripped.startswith(_CONFLICT_TAIL):
|
|
return True
|
|
if stripped == _CONFLICT_SEPARATOR:
|
|
return True
|
|
return False
|
|
|
|
|
|
def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
|
|
"""Skip venv/git/cache and sibling worktrees under orchestration checkout.
|
|
|
|
When *project_root* is itself a worktree inside ``branches/``, still scan
|
|
that tree — do not treat the ``branches`` path segment as a skip signal.
|
|
"""
|
|
rel = os.path.relpath(walk_root, project_root)
|
|
if rel == ".":
|
|
return False
|
|
head = rel.split(os.sep, 1)[0]
|
|
if head in ("venv", ".git", ".pytest_cache"):
|
|
return True
|
|
if head == "branches":
|
|
nested = os.path.join(project_root, "branches")
|
|
if os.path.isdir(nested) and (
|
|
walk_root == nested or walk_root.startswith(nested + os.sep)
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
REVIEWER_TASKS = frozenset({
|
|
"review_pr",
|
|
"merge_pr",
|
|
"blind_pr_queue_review",
|
|
"request_changes_pr",
|
|
"approve_pr",
|
|
})
|
|
|
|
AUTHOR_TASKS = frozenset({
|
|
"create_issue",
|
|
"comment_issue",
|
|
"close_issue",
|
|
"claim_issue",
|
|
"create_branch",
|
|
"push_branch",
|
|
"create_pr",
|
|
"comment_pr",
|
|
"address_pr_change_requests",
|
|
"delete_branch",
|
|
"reconcile_landed_pr",
|
|
"reconcile-landed-pr",
|
|
})
|
|
|
|
TASK_REQUIRED_ROLE = {
|
|
"create_issue": "author",
|
|
"comment_issue": "author",
|
|
"close_issue": "author",
|
|
"claim_issue": "author",
|
|
"create_branch": "author",
|
|
"push_branch": "author",
|
|
"create_pr": "author",
|
|
"comment_pr": "author",
|
|
"address_pr_change_requests": "author",
|
|
"delete_branch": "author",
|
|
"review_pr": "reviewer",
|
|
"merge_pr": "reviewer",
|
|
"blind_pr_queue_review": "reviewer",
|
|
"request_changes_pr": "reviewer",
|
|
"approve_pr": "reviewer",
|
|
"reconcile_landed_pr": "author",
|
|
"reconcile-landed-pr": "author",
|
|
}
|
|
|
|
WRONG_ROLE_REVIEWER_MSG = (
|
|
"Wrong role/session for reviewer task. Launch reviewer MCP namespace."
|
|
)
|
|
|
|
_session_last_route: dict | None = None
|
|
|
|
|
|
def required_role_for_task(task_type: str) -> str | None:
|
|
return TASK_REQUIRED_ROLE.get((task_type or "").strip())
|
|
|
|
|
|
def route_task_session(
|
|
task_type: str,
|
|
*,
|
|
active_profile: str,
|
|
active_role_kind: str,
|
|
allowed_in_current_session: bool,
|
|
runtime_switching_supported: bool = False,
|
|
) -> dict:
|
|
"""Return routing verdict for *task_type* under the active session."""
|
|
task_type = (task_type or "").strip()
|
|
required_role = required_role_for_task(task_type)
|
|
if required_role is None:
|
|
result = {
|
|
"task_type": task_type,
|
|
"required_role": None,
|
|
"active_role": active_role_kind,
|
|
"active_profile": active_profile,
|
|
"route_result": ROUTE_AMBIGUOUS,
|
|
"downstream_allowed": False,
|
|
"reasons": [
|
|
f"unknown task type '{task_type}'; cannot route session "
|
|
"(fail closed)"
|
|
],
|
|
"message": (
|
|
"Ambiguous task type; relaunch with an explicit task before "
|
|
"any tool use."
|
|
),
|
|
}
|
|
_record_route(result)
|
|
return result
|
|
|
|
if required_role == "reviewer":
|
|
infra = assess_infra_stop()
|
|
if infra["infra_stop"]:
|
|
detail = "; ".join(infra.get("infra_stop_reasons") or [])
|
|
message = (
|
|
"infra_stop: Unresolved merge conflict or mid-merge state detected "
|
|
f"in MCP runtime source ({detail}). Please resolve all conflicts "
|
|
"manually, finish/abort the merge, and retry."
|
|
)
|
|
result = {
|
|
"task_type": task_type,
|
|
"required_role": required_role,
|
|
"active_role": active_role_kind,
|
|
"active_profile": active_profile,
|
|
"route_result": ROUTE_INFRA_STOP,
|
|
"downstream_allowed": False,
|
|
"reasons": [message],
|
|
"message": message,
|
|
"infra_stop_assessment": infra,
|
|
}
|
|
_record_route(result)
|
|
return result
|
|
|
|
if allowed_in_current_session:
|
|
result = {
|
|
"task_type": task_type,
|
|
"required_role": required_role,
|
|
"active_role": active_role_kind,
|
|
"active_profile": active_profile,
|
|
"route_result": ROUTE_ALLOWED,
|
|
"downstream_allowed": True,
|
|
"reasons": [],
|
|
"message": "Task role matches active session; proceed.",
|
|
}
|
|
_record_route(result)
|
|
return result
|
|
|
|
if required_role == "reviewer":
|
|
result = {
|
|
"task_type": task_type,
|
|
"required_role": required_role,
|
|
"active_role": active_role_kind,
|
|
"active_profile": active_profile,
|
|
"route_result": ROUTE_WRONG_ROLE,
|
|
"downstream_allowed": False,
|
|
"reasons": [
|
|
WRONG_ROLE_REVIEWER_MSG,
|
|
"Reviewer tasks cannot run in author-bound sessions.",
|
|
"Static-profile mode does not permit in-place role switching.",
|
|
],
|
|
"message": WRONG_ROLE_REVIEWER_MSG,
|
|
"runtime_switching_supported": runtime_switching_supported,
|
|
"profile_switch_blocked": not runtime_switching_supported,
|
|
}
|
|
_record_route(result)
|
|
return result
|
|
|
|
if required_role == "author":
|
|
route = ROUTE_TO_AUTHOR
|
|
message = (
|
|
"Wrong role/session for author task. Launch author MCP namespace."
|
|
)
|
|
result = {
|
|
"task_type": task_type,
|
|
"required_role": required_role,
|
|
"active_role": active_role_kind,
|
|
"active_profile": active_profile,
|
|
"route_result": route,
|
|
"downstream_allowed": False,
|
|
"reasons": [message],
|
|
"message": message,
|
|
"runtime_switching_supported": runtime_switching_supported,
|
|
"profile_switch_blocked": not runtime_switching_supported,
|
|
}
|
|
_record_route(result)
|
|
return result
|
|
|
|
result = {
|
|
"task_type": task_type,
|
|
"required_role": required_role,
|
|
"active_role": active_role_kind,
|
|
"active_profile": active_profile,
|
|
"route_result": ROUTE_AMBIGUOUS,
|
|
"downstream_allowed": False,
|
|
"reasons": ["unable to classify task role (fail closed)"],
|
|
"message": "Ambiguous task type; stop before any mutation.",
|
|
}
|
|
_record_route(result)
|
|
return result
|
|
|
|
|
|
def last_route() -> dict | None:
|
|
return _session_last_route
|
|
|
|
|
|
def clear_route_state():
|
|
global _session_last_route
|
|
_session_last_route = None
|
|
|
|
|
|
def _record_route(result: dict):
|
|
global _session_last_route
|
|
_session_last_route = dict(result)
|
|
|
|
|
|
def sync_route_from_capability(capability: dict) -> None:
|
|
"""Align sticky route state with operation-scoped capability resolution (#228)."""
|
|
capability = capability or {}
|
|
task = (capability.get("requested_task") or "").strip()
|
|
required_role = capability.get("required_role_kind")
|
|
if not task or not required_role:
|
|
return
|
|
if capability.get("allowed_in_current_session"):
|
|
_record_route({
|
|
"task_type": task,
|
|
"required_role": required_role,
|
|
"active_role": required_role,
|
|
"active_profile": capability.get("active_profile"),
|
|
"route_result": ROUTE_ALLOWED,
|
|
"downstream_allowed": True,
|
|
"reasons": [],
|
|
"message": (
|
|
f"Operation-scoped task '{task}' resolved for current session; "
|
|
"proceed."
|
|
),
|
|
})
|
|
return
|
|
if required_role == "reviewer" and capability.get("stop_required"):
|
|
_record_route({
|
|
"task_type": task,
|
|
"required_role": required_role,
|
|
"active_role": capability.get("required_role_kind"),
|
|
"active_profile": capability.get("active_profile"),
|
|
"route_result": ROUTE_WRONG_ROLE,
|
|
"downstream_allowed": False,
|
|
"reasons": [WRONG_ROLE_REVIEWER_MSG],
|
|
"message": WRONG_ROLE_REVIEWER_MSG,
|
|
})
|
|
|
|
|
|
def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]:
|
|
"""Block author-side fallback after a reviewer wrong_role_stop (#206).
|
|
|
|
An explicit operation-scoped author capability resolution for the same
|
|
*mutation_task* clears the sticky reviewer denial (#228).
|
|
"""
|
|
last = _session_last_route
|
|
if not last:
|
|
return True, []
|
|
if (
|
|
last.get("route_result") == ROUTE_ALLOWED
|
|
and last.get("task_type") == mutation_task
|
|
and last.get("required_role") == "author"
|
|
):
|
|
return True, []
|
|
if last.get("route_result") != ROUTE_WRONG_ROLE:
|
|
return True, []
|
|
if last.get("required_role") != "reviewer":
|
|
return True, []
|
|
if mutation_task in AUTHOR_TASKS:
|
|
return False, [
|
|
WRONG_ROLE_REVIEWER_MSG,
|
|
"Author-side mutations are blocked after a reviewer-task "
|
|
"wrong_role_stop unless the operator explicitly resolves the "
|
|
"author task via gitea_resolve_task_capability.",
|
|
f"Attempted fallback mutation: {mutation_task}",
|
|
]
|
|
return True, []
|
|
|
|
|
|
def first_conflict_marker_path(project_root: str | None = None) -> str | None:
|
|
"""Return the first .py path containing a git conflict marker, or None."""
|
|
root_dir = project_root or os.path.dirname(os.path.abspath(__file__))
|
|
for root, dirs, files in os.walk(root_dir):
|
|
if skip_python_scan_walk_root(root_dir, root):
|
|
continue
|
|
for file in files:
|
|
if not file.endswith(".py"):
|
|
continue
|
|
file_path = os.path.join(root, file)
|
|
try:
|
|
with open(file_path, "rb") as f:
|
|
if python_bytes_have_conflict_markers(f.read()):
|
|
return file_path
|
|
except OSError:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _default_project_root() -> str:
|
|
override = (os.environ.get("GITEA_MCP_PROJECT_ROOT") or "").strip()
|
|
if override:
|
|
return os.path.realpath(override)
|
|
return os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def assess_infra_stop(project_root: str | None = None) -> dict:
|
|
"""Recompute infra_stop from live git and source scan state (#285)."""
|
|
root_dir = os.path.realpath(project_root or _default_project_root())
|
|
reasons: list[str] = []
|
|
mid_merge = False
|
|
git_dir = os.path.join(root_dir, ".git")
|
|
if os.path.isdir(git_dir):
|
|
for marker in ("MERGE_HEAD", "rebase-merge", "rebase-apply"):
|
|
marker_path = os.path.join(git_dir, marker)
|
|
if os.path.exists(marker_path):
|
|
mid_merge = True
|
|
reasons.append(f"git state: {marker} present under {git_dir}")
|
|
conflict_file = first_conflict_marker_path(root_dir)
|
|
if conflict_file:
|
|
reasons.append(
|
|
f"conflict markers detected in {conflict_file} (project_root={root_dir})"
|
|
)
|
|
infra_stop = mid_merge or bool(conflict_file)
|
|
return {
|
|
"infra_stop": infra_stop,
|
|
"project_root": root_dir,
|
|
"conflict_file": conflict_file,
|
|
"mid_merge": mid_merge,
|
|
"infra_stop_reasons": reasons,
|
|
}
|
|
|
|
|
|
def check_mid_merge(project_root: str | None = None) -> bool:
|
|
"""Return True if the repository is mid-merge, mid-rebase, or has conflict markers."""
|
|
return assess_infra_stop(project_root)["infra_stop"] |