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) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -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()
|
||||
Reference in New Issue
Block a user