From e09df4a34811c9b004c284f2f87fad8137b1d47b Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 00:28:32 -0400 Subject: [PATCH] feat: add fail-closed subagent delegation gates (#266) Adds subagent_gate.py following the author_proofs/review_proofs pure-policy pattern: - assess_subagent_delegation: deterministic write tasks (issue claim, branch creation, code edits, commits, PR creation, review, merge, cleanup) are blocked for subagents unless explicitly allowed with a recorded justification and the full inherited gate context (issue lock, branch/worktree path, identity/profile, allowed tool class, command deny list, validation ledger requirement, final report schema); read-only delegation stays allowed; unknown task types fail closed. - validate_subagent_report: subagent output is accepted only when it carries the same proof fields as a parent workflow final report. - Operator guide: new subagent_delegation rule in _GUIDE_RULES so MCP clients discover the policy. - tests/test_subagent_gate.py: 13 tests covering blocked write delegation, justification requirement, missing/blank inherited context, allowed read-only delegation, unknown tasks, and invalid subagent final reports. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 13 +++ subagent_gate.py | 166 ++++++++++++++++++++++++++++++++++++ tests/test_subagent_gate.py | 152 +++++++++++++++++++++++++++++++++ 3 files changed, 331 insertions(+) create mode 100644 subagent_gate.py create mode 100644 tests/test_subagent_gate.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index dd1480c..81fc8a7 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -3483,6 +3483,19 @@ _GUIDE_RULES = { "small fixes, review fixes, conflicts, emergencies). Main checkout: " "read-only inspect, fetch, create worktrees, post-merge stable " "update, explicit repair only."), + "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 = [ diff --git a/subagent_gate.py b/subagent_gate.py new file mode 100644 index 0000000..f3fc18e --- /dev/null +++ b/subagent_gate.py @@ -0,0 +1,166 @@ +"""Fail-closed subagent delegation gates (#266). + +Subagents can bypass or lose context for worktree, capability, mutation, +retry, and reporting rules. These helpers make delegation an explicit, +provable decision instead of a default: deterministic write tasks stay +inline unless the parent session records why a subagent is needed and +hands the subagent the full gate context it must operate under. + +Like ``author_proofs``/``review_proofs``, the helpers are pure (no git, +no API calls): the parent workflow gathers the facts and passes them in, +so the same logic works from prompts, harness assertions, and tests. +Nothing here weakens the review/merge/permission gates — a delegated +subagent is subject to the same gates as its parent. +""" + +# AC1: deterministic write workflows a subagent must never run by default. +DETERMINISTIC_WRITE_TASKS = frozenset({ + "claim_issue", + "create_branch", + "edit_code", + "commit", + "push_branch", + "create_pr", + "review_pr", + "merge_pr", + "cleanup_branch", + "close_issue", +}) + +# Read-only delegation that needs no explicit authorization. +READ_ONLY_TASKS = frozenset({ + "read_files", + "code_search", + "inventory_prs", + "summarize_issue", + "explore_codebase", +}) + +# AC3: context a subagent must inherit from the parent session before any +# authorized write delegation may proceed. +REQUIRED_INHERITED_CONTEXT = ( + "issue_lock", + "branch", + "worktree_path", + "identity_profile", + "allowed_tool_class", + "command_deny_list", + "validation_ledger_requirement", + "final_report_schema", +) + +# AC4: proof fields a subagent final report must carry — the same fields a +# parent workflow's final report requires. +REQUIRED_SUBAGENT_REPORT_FIELDS = ( + "identity_profile", + "worktree_path", + "branch", + "changed_files", + "validation_results", + "workspace_mutations", +) + + +def _clean(value): + return (value or "").strip() if isinstance(value, str) else value + + +def _classify_task(task_type): + task = _clean(task_type) + if not task: + return "", "unknown" + if task in DETERMINISTIC_WRITE_TASKS: + return task, "deterministic_write" + if task in READ_ONLY_TASKS: + return task, "read_only" + return task, "unknown" + + +def _missing_context_fields(inherited_context): + context = inherited_context or {} + missing = [] + for field in REQUIRED_INHERITED_CONTEXT: + value = context.get(field) + if not isinstance(value, str) or not value.strip(): + missing.append(field) + return missing + + +def assess_subagent_delegation(task_type, *, explicitly_allowed=False, + justification=None, inherited_context=None): + """Decide whether delegating *task_type* to a subagent may proceed. + + Fail closed: unknown tasks are blocked; deterministic write tasks are + blocked unless explicitly allowed (AC1) with a recorded justification + (AC2) and the full inherited gate context (AC3). Read-only delegation + is allowed without explicit authorization. + + Returns {'block', 'allowed', 'task_type', 'task_class', 'reasons', + 'missing_context'}. + """ + task, task_class = _classify_task(task_type) + reasons = [] + missing_context = [] + + if task_class == "unknown": + reasons.append( + f"task type '{task}' is not a recognized delegation class; " + "run it inline in the parent session (fail closed)" + ) + elif task_class == "deterministic_write": + if not explicitly_allowed: + reasons.append( + f"deterministic write task '{task}' must run inline unless " + "subagent use is explicitly allowed by the parent session" + ) + if not _clean(justification): + reasons.append( + "no recorded justification for why a subagent is needed; " + "the parent session must record one before delegating" + ) + missing_context = _missing_context_fields(inherited_context) + if missing_context: + reasons.append( + "subagent would not inherit required gate context: " + + ", ".join(missing_context) + ) + + block = bool(reasons) + return { + "block": block, + "allowed": not block, + "task_type": task, + "task_class": task_class, + "reasons": reasons, + "missing_context": missing_context, + } + + +def validate_subagent_report(report_fields): + """AC4: accept subagent output only with the parent-grade proof fields. + + *report_fields* maps field name -> reported value. Missing or blank + proof fields make the report invalid (fail closed). + + Returns {'valid', 'block', 'reasons', 'missing_fields'}. + """ + reasons = [] + report = report_fields or {} + missing = [] + for field in REQUIRED_SUBAGENT_REPORT_FIELDS: + value = report.get(field) + if not isinstance(value, str) or not value.strip(): + missing.append(field) + if missing: + reasons.append( + "subagent final report missing required proof fields: " + + ", ".join(missing) + ) + + valid = not reasons + return { + "valid": valid, + "block": not valid, + "reasons": reasons, + "missing_fields": missing, + } diff --git a/tests/test_subagent_gate.py b/tests/test_subagent_gate.py new file mode 100644 index 0000000..a4d08fd --- /dev/null +++ b/tests/test_subagent_gate.py @@ -0,0 +1,152 @@ +"""Tests for fail-closed subagent delegation gates (#266). + +Covers the acceptance criteria: blocked write delegation, allowed read-only +delegation, missing inherited context, and invalid subagent final reports. +""" +import unittest + +import subagent_gate + + +def _full_context(): + return { + "issue_lock": "issue #266 locked to feat/issue-266-subagent-gate-inheritance", + "branch": "feat/issue-266-subagent-gate-inheritance", + "worktree_path": "branches/feat-issue-266-subagent-gate-inheritance", + "identity_profile": "jcwalker3/prgs-author", + "allowed_tool_class": "read_write_files", + "command_deny_list": "git push --force; rm -rf", + "validation_ledger_requirement": "record command/exit/output for every validation claim", + "final_report_schema": "controller-handoff-v1", + } + + +def _full_report(): + return { + "identity_profile": "jcwalker3/prgs-author", + "worktree_path": "branches/feat-issue-266-subagent-gate-inheritance", + "branch": "feat/issue-266-subagent-gate-inheritance", + "changed_files": "subagent_gate.py, tests/test_subagent_gate.py", + "validation_results": "pytest tests/test_subagent_gate.py -q: exit 0, 14 passed", + "workspace_mutations": "edited subagent_gate.py", + } + + +class TestAssessSubagentDelegation(unittest.TestCase): + # AC1: deterministic write tasks are blocked by default + def test_write_delegation_blocked_by_default(self): + for task in ("claim_issue", "create_branch", "edit_code", "commit", + "push_branch", "create_pr", "review_pr", "merge_pr", + "cleanup_branch"): + res = subagent_gate.assess_subagent_delegation(task) + self.assertTrue(res["block"], task) + self.assertFalse(res["allowed"], task) + self.assertEqual(res["task_class"], "deterministic_write", task) + self.assertTrue( + any("explicitly allowed" in r for r in res["reasons"]), task) + + # AC2: explicit allowance without a recorded justification still blocks + def test_write_delegation_requires_recorded_justification(self): + res = subagent_gate.assess_subagent_delegation( + "create_pr", explicitly_allowed=True, + inherited_context=_full_context()) + self.assertTrue(res["block"]) + self.assertTrue( + any("justification" in r for r in res["reasons"])) + + # AC3: explicit allowance + justification but missing inherited context + def test_write_delegation_blocks_on_missing_inherited_context(self): + context = _full_context() + del context["issue_lock"] + del context["command_deny_list"] + res = subagent_gate.assess_subagent_delegation( + "commit", explicitly_allowed=True, + justification="parent session proved batch commit needs isolation", + inherited_context=context) + self.assertTrue(res["block"]) + self.assertIn("issue_lock", res["missing_context"]) + self.assertIn("command_deny_list", res["missing_context"]) + + def test_write_delegation_blocks_on_empty_context_values(self): + context = _full_context() + context["worktree_path"] = " " + res = subagent_gate.assess_subagent_delegation( + "commit", explicitly_allowed=True, + justification="isolation required", + inherited_context=context) + self.assertTrue(res["block"]) + self.assertIn("worktree_path", res["missing_context"]) + + def test_write_delegation_allowed_with_authorization_and_full_context(self): + res = subagent_gate.assess_subagent_delegation( + "edit_code", explicitly_allowed=True, + justification="parallel mechanical rename across many files", + inherited_context=_full_context()) + self.assertFalse(res["block"]) + self.assertTrue(res["allowed"]) + self.assertEqual(res["missing_context"], []) + + # Allowed read-only delegation needs no explicit authorization + def test_read_only_delegation_allowed(self): + for task in ("read_files", "code_search", "inventory_prs", + "summarize_issue"): + res = subagent_gate.assess_subagent_delegation(task) + self.assertFalse(res["block"], task) + self.assertTrue(res["allowed"], task) + self.assertEqual(res["task_class"], "read_only", task) + + # Unknown task types fail closed + def test_unknown_task_fails_closed(self): + res = subagent_gate.assess_subagent_delegation("launch_missiles") + self.assertTrue(res["block"]) + self.assertEqual(res["task_class"], "unknown") + + def test_blank_task_fails_closed(self): + res = subagent_gate.assess_subagent_delegation(" ") + self.assertTrue(res["block"]) + self.assertEqual(res["task_class"], "unknown") + + +class TestValidateSubagentReport(unittest.TestCase): + # AC4: subagent output must carry the same proof fields as the parent + def test_full_report_valid(self): + res = subagent_gate.validate_subagent_report(_full_report()) + self.assertTrue(res["valid"]) + self.assertFalse(res["block"]) + self.assertEqual(res["missing_fields"], []) + + def test_missing_proof_fields_invalid(self): + report = _full_report() + del report["validation_results"] + del report["worktree_path"] + res = subagent_gate.validate_subagent_report(report) + self.assertFalse(res["valid"]) + self.assertTrue(res["block"]) + self.assertIn("validation_results", res["missing_fields"]) + self.assertIn("worktree_path", res["missing_fields"]) + + def test_empty_field_values_invalid(self): + report = _full_report() + report["changed_files"] = "" + res = subagent_gate.validate_subagent_report(report) + self.assertFalse(res["valid"]) + self.assertIn("changed_files", res["missing_fields"]) + + def test_none_report_invalid(self): + res = subagent_gate.validate_subagent_report(None) + self.assertFalse(res["valid"]) + self.assertTrue(res["block"]) + + +class TestOperatorGuideRule(unittest.TestCase): + def test_operator_guide_declares_subagent_rule(self): + import gitea_mcp_server + + rule = gitea_mcp_server._GUIDE_RULES["subagent_delegation"].lower() + for phrase in ("deterministic write", "inherit", "read-only", + "fail closed"): + self.assertIn(phrase, rule) + + +if __name__ == "__main__": + unittest.main()