Keep master router-style SKILL.md (full #333 split) and merge #334 review-merge test coverage into test_llm_workflow_split.py. Conflicts resolved in: - skills/llm-project-workflow/SKILL.md - tests/test_llm_workflow_split.py
This commit is contained in:
@@ -0,0 +1,102 @@
|
||||
"""Identity-disclosure checks for workflow reports (#305).
|
||||
|
||||
Workflow reports sometimes included the authenticated user's personal
|
||||
email even though username/profile identity is sufficient (observed:
|
||||
``Identity: jcwalker3 / [email protected]`` in a reconciliation
|
||||
handoff). These tests pin the no-email identity summary helper and the
|
||||
final-report validator that flags unnecessary email disclosure.
|
||||
"""
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import review_proofs
|
||||
|
||||
|
||||
class TestIdentitySummary(unittest.TestCase):
|
||||
def test_author_summary_uses_username_and_profile_only(self):
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"jcwalker3", "prgs-author")
|
||||
self.assertEqual(summary, "jcwalker3 / prgs-author")
|
||||
self.assertNotIn("@", summary)
|
||||
|
||||
def test_reviewer_summary_uses_username_and_profile_only(self):
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"sysadmin", "prgs-reviewer")
|
||||
self.assertEqual(summary, "sysadmin / prgs-reviewer")
|
||||
|
||||
def test_summary_appends_role_and_remote_without_email(self):
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"jcwalker3", "prgs-author", role="author", remote="prgs")
|
||||
self.assertIn("jcwalker3 / prgs-author", summary)
|
||||
self.assertIn("author", summary)
|
||||
self.assertIn("prgs", summary)
|
||||
self.assertNotIn("@", summary)
|
||||
|
||||
def test_summary_never_leaks_email_passed_as_username(self):
|
||||
"""Defense in depth: an email in the username slot is reduced."""
|
||||
summary = review_proofs.format_identity_summary(
|
||||
"[email protected]", "prgs-author")
|
||||
self.assertNotIn("@", summary)
|
||||
self.assertIn("jcwalker3", summary)
|
||||
|
||||
|
||||
class TestEmailDisclosureAssessment(unittest.TestCase):
|
||||
def test_report_without_email_passes(self):
|
||||
report = "\n".join([
|
||||
"Controller Handoff",
|
||||
"Identity: jcwalker3 / prgs-author",
|
||||
"Role: author",
|
||||
])
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertTrue(res["proven"])
|
||||
self.assertFalse(res["flagged"])
|
||||
self.assertEqual(res["emails"], [])
|
||||
self.assertEqual(res["reasons"], [])
|
||||
|
||||
def test_unnecessary_email_is_flagged(self):
|
||||
report = "\n".join([
|
||||
"Controller Handoff",
|
||||
"Identity: jcwalker3 / [email protected]",
|
||||
])
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertFalse(res["proven"])
|
||||
self.assertTrue(res["flagged"])
|
||||
self.assertIn("[email protected]", res["emails"])
|
||||
self.assertTrue(res["reasons"])
|
||||
self.assertFalse(res["justified"])
|
||||
|
||||
def test_multiple_emails_all_reported(self):
|
||||
report = (
|
||||
"Identity: [email protected] author\n"
|
||||
"Reviewer: [email protected]\n"
|
||||
)
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertTrue(res["flagged"])
|
||||
self.assertEqual(
|
||||
sorted(res["emails"]),
|
||||
["[email protected]", "[email protected]"],
|
||||
)
|
||||
|
||||
def test_justified_email_with_explanation_is_not_flagged(self):
|
||||
report = "\n".join([
|
||||
"Identity: jcwalker3 / prgs-author",
|
||||
"Contact email: [email protected]",
|
||||
"Email required because two accounts share the username and the",
|
||||
"address is necessary to disambiguate identity.",
|
||||
])
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertFalse(res["flagged"])
|
||||
self.assertTrue(res["justified"])
|
||||
self.assertIn("[email protected]", res["emails"])
|
||||
|
||||
def test_email_without_justification_language_is_flagged(self):
|
||||
report = "Contact email: [email protected] just in case.\n"
|
||||
res = review_proofs.assess_email_disclosure(report)
|
||||
self.assertTrue(res["flagged"])
|
||||
self.assertFalse(res["justified"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -3,15 +3,62 @@ from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
SKILL_DIR = REPO_ROOT / "skills" / "llm-project-workflow"
|
||||
SKILL = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_review_merge_workflow_file_exists():
|
||||
path = SKILL_DIR / "workflows" / "review-merge-pr.md"
|
||||
text = path.read_text(encoding="utf-8")
|
||||
def test_skill_md_exists():
|
||||
assert SKILL_DIR.joinpath("SKILL.md").is_file()
|
||||
|
||||
|
||||
def test_all_workflow_files_exist():
|
||||
for name in (
|
||||
"review-merge-pr.md",
|
||||
"reconcile-landed-pr.md",
|
||||
"create-issue.md",
|
||||
"work-issue.md",
|
||||
):
|
||||
assert (SKILL_DIR / "workflows" / name).is_file(), name
|
||||
|
||||
|
||||
def test_skill_references_all_workflow_files():
|
||||
for name in (
|
||||
"workflows/review-merge-pr.md",
|
||||
"workflows/reconcile-landed-pr.md",
|
||||
"workflows/create-issue.md",
|
||||
"workflows/work-issue.md",
|
||||
):
|
||||
assert name in SKILL
|
||||
|
||||
|
||||
def test_skill_contains_mode_isolation_language():
|
||||
assert "## Mode isolation" in SKILL
|
||||
assert "review-merge-pr" in SKILL
|
||||
assert "reconcile-landed-pr" in SKILL
|
||||
assert "create-issue" in SKILL
|
||||
assert "work-issue" in SKILL
|
||||
assert "Do not mix modes" in SKILL or "do not mix modes" in SKILL.lower()
|
||||
|
||||
|
||||
def test_skill_is_router_not_monolithic_review_body():
|
||||
assert "This skill is a **router**" in SKILL or "router" in SKILL.lower()
|
||||
assert "## F. Review workflow" not in SKILL
|
||||
assert "gitea_mark_final_review_decision" not in SKILL
|
||||
assert "gitea_submit_pr_review" not in SKILL
|
||||
|
||||
|
||||
def test_skill_still_declares_controller_handoff_contract():
|
||||
assert "## Controller Handoff" in SKILL
|
||||
assert "assess_controller_handoff" in SKILL
|
||||
assert "## Global LLM Worktree Rule" in SKILL
|
||||
|
||||
|
||||
def test_review_merge_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "review-merge-pr.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "review-merge-pr" in text
|
||||
assert "## 0. Load the canonical workflow first" in text
|
||||
assert "## 26A. Terminal review mutation hard-stop" in text
|
||||
assert "## 11A. Skipped PRs are read-only" in text
|
||||
assert "## 35. Duplicate request-changes prevention" in text
|
||||
assert "## 37. Controller handoff schema" in text
|
||||
assert "INVENTORY_PAGINATION_UNPROVEN" in text
|
||||
assert "ALREADY_LANDED_RECONCILE_REQUIRED" in text
|
||||
@@ -23,29 +70,49 @@ def test_review_merge_final_report_schema_exists():
|
||||
assert "Controller Handoff" in text
|
||||
assert "Candidate head SHA:" in text
|
||||
assert "Terminal review mutation:" in text
|
||||
assert "Workspace mutations" in text # documented as rejected
|
||||
assert "Workspace mutations" in text
|
||||
|
||||
|
||||
def test_skill_router_points_to_review_merge_workflow():
|
||||
skill = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
|
||||
assert "## Task mode router" in skill
|
||||
assert "workflows/review-merge-pr.md" in skill
|
||||
assert "schemas/review-merge-final-report.md" in skill
|
||||
assert "Identify task mode before any mutation" in skill
|
||||
def test_reconcile_landed_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "reconcile-landed-pr.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "Do not review or merge normal PRs" in text
|
||||
assert "Already-landed proof" in text
|
||||
|
||||
|
||||
def test_skill_still_declares_controller_handoff_contract():
|
||||
skill = (SKILL_DIR / "SKILL.md").read_text(encoding="utf-8")
|
||||
assert "## Controller Handoff" in skill
|
||||
assert "assess_controller_handoff" in skill
|
||||
assert "## Global LLM Worktree Rule" in skill
|
||||
def test_create_issue_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "create-issue.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "## 9. Duplicate search before mutation" in text
|
||||
|
||||
|
||||
def test_review_pr_template_references_extracted_workflow():
|
||||
def test_work_issue_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
assert "Do not merge your own PR" in text
|
||||
assert "## 21. PR creation rules" in text
|
||||
|
||||
|
||||
def test_final_report_schemas_exist():
|
||||
for name in (
|
||||
"review-merge-final-report.md",
|
||||
"reconcile-landed-final-report.md",
|
||||
"create-issue-final-report.md",
|
||||
"work-issue-final-report.md",
|
||||
):
|
||||
assert (SKILL_DIR / "schemas" / name).is_file(), name
|
||||
|
||||
|
||||
def test_review_pr_template_references_workflow():
|
||||
text = (SKILL_DIR / "templates" / "review-pr.md").read_text(encoding="utf-8")
|
||||
assert "workflows/review-merge-pr.md" in text
|
||||
|
||||
|
||||
def test_merge_pr_template_references_extracted_workflow():
|
||||
def test_merge_pr_template_references_workflow():
|
||||
text = (SKILL_DIR / "templates" / "merge-pr.md").read_text(encoding="utf-8")
|
||||
assert "workflows/review-merge-pr.md" in text
|
||||
assert "workflows/review-merge-pr.md" in text
|
||||
|
||||
|
||||
def test_start_issue_template_references_work_issue_workflow():
|
||||
text = (SKILL_DIR / "templates" / "start-issue.md").read_text(encoding="utf-8")
|
||||
assert "workflows/work-issue.md" in text
|
||||
@@ -0,0 +1,102 @@
|
||||
"""Tests for final-report mutation ledger verifier (#331)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_mutation_ledger_report # noqa: E402
|
||||
|
||||
|
||||
class TestMutationLedgerReport(unittest.TestCase):
|
||||
def test_none_claim_with_performed_edit_blocks(self):
|
||||
report = (
|
||||
"Review decision: request_changes.\n"
|
||||
"File edits by reviewer: none\n"
|
||||
)
|
||||
action_log = [
|
||||
{"action": "Edited", "path": "walkthrough.md", "tracked": False},
|
||||
]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["file_edits_claimed_none"])
|
||||
|
||||
def test_reported_edit_passes(self):
|
||||
report = (
|
||||
"File edits by reviewer: Edited walkthrough.md (untracked)\n"
|
||||
"Mutation ledger: Edited walkthrough.md untracked in review worktree\n"
|
||||
)
|
||||
action_log = [
|
||||
{
|
||||
"action": "Edited",
|
||||
"path": "walkthrough.md",
|
||||
"tracked": False,
|
||||
"in_repo": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(
|
||||
report,
|
||||
action_log=action_log,
|
||||
walkthrough_explicitly_requested=True,
|
||||
)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_unreported_mutation_blocks(self):
|
||||
report = "File edits by reviewer: none\n"
|
||||
action_log = [{"action": "Wrote", "path": "/tmp/review-notes.txt"}]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("/tmp/review-notes.txt", result["unreported_paths"])
|
||||
|
||||
def test_outside_repo_requires_label(self):
|
||||
report = (
|
||||
"File edits by reviewer: Wrote /tmp/review-notes.txt\n"
|
||||
)
|
||||
action_log = [
|
||||
{
|
||||
"action": "Wrote",
|
||||
"path": "/tmp/review-notes.txt",
|
||||
"outside_repo": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("outside repo", " ".join(result["reasons"]).lower())
|
||||
|
||||
def test_gated_rejection_excluded_from_performed(self):
|
||||
report = "File edits by reviewer: none\nRejected gated calls: submit_pr_review blocked\n"
|
||||
action_log = [
|
||||
{
|
||||
"action": "Edited",
|
||||
"path": "walkthrough.md",
|
||||
"performed": False,
|
||||
"gated_rejected": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(report, action_log=action_log)
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertEqual(result["performed_mutations"], [])
|
||||
|
||||
def test_post_status_artifact_requires_final_git_status(self):
|
||||
report = (
|
||||
"File edits by reviewer: Created scratch/notes.md (untracked)\n"
|
||||
)
|
||||
action_log = [
|
||||
{
|
||||
"action": "Created",
|
||||
"path": "scratch/notes.md",
|
||||
"tracked": False,
|
||||
"after_git_status": True,
|
||||
},
|
||||
]
|
||||
result = assess_mutation_ledger_report(
|
||||
report,
|
||||
action_log=action_log,
|
||||
final_git_status_reported=False,
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("final git status", " ".join(result["reasons"]).lower())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Tests for REQUEST_CHANGES override proof before approval (#326)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from review_proofs import assess_request_changes_approval_proof # noqa: E402
|
||||
|
||||
HEAD_A = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
HEAD_B = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||
|
||||
|
||||
def _feedback(**overrides):
|
||||
base = {
|
||||
"success": True,
|
||||
"current_head_sha": HEAD_A,
|
||||
"has_blocking_change_requests": False,
|
||||
"author_pushed_after_request_changes": False,
|
||||
"reviews": [],
|
||||
}
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
def _blocking_review(**overrides):
|
||||
entry = {
|
||||
"reviewer": "reviewer-a",
|
||||
"verdict": "REQUEST_CHANGES",
|
||||
"body": "Tests fail on pinned head; fix test_commit_files_gate.",
|
||||
"submitted_at": "2026-07-07T01:00:00-05:00",
|
||||
"reviewed_head_sha": HEAD_A,
|
||||
"dismissed": False,
|
||||
"stale": False,
|
||||
}
|
||||
entry.update(overrides)
|
||||
return entry
|
||||
|
||||
|
||||
class TestRequestChangesApprovalProof(unittest.TestCase):
|
||||
def test_missing_feedback_blocks_approval(self):
|
||||
result = assess_request_changes_approval_proof(None)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_no_prior_request_changes_allows_approval(self):
|
||||
feedback = _feedback(
|
||||
reviews=[{
|
||||
"reviewer": "sysadmin",
|
||||
"verdict": "APPROVED",
|
||||
"body": "LGTM",
|
||||
"submitted_at": "2026-07-07T02:00:00-05:00",
|
||||
"reviewed_head_sha": HEAD_A,
|
||||
"dismissed": False,
|
||||
}]
|
||||
)
|
||||
result = assess_request_changes_approval_proof(feedback)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertIsNone(result["blocking_review"])
|
||||
|
||||
def test_changed_head_since_blocker_allows_approval(self):
|
||||
feedback = _feedback(
|
||||
current_head_sha=HEAD_B,
|
||||
author_pushed_after_request_changes=True,
|
||||
has_blocking_change_requests=True,
|
||||
reviews=[_blocking_review(reviewed_head_sha=HEAD_A)],
|
||||
)
|
||||
result = assess_request_changes_approval_proof(feedback)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertTrue(result["head_changed_since_blocker"])
|
||||
|
||||
def test_unchanged_head_without_override_blocks_approval(self):
|
||||
feedback = _feedback(
|
||||
has_blocking_change_requests=True,
|
||||
reviews=[_blocking_review()],
|
||||
)
|
||||
result = assess_request_changes_approval_proof(feedback)
|
||||
self.assertFalse(result["approve_allowed"])
|
||||
self.assertIn("override_reason", " ".join(result["reasons"]))
|
||||
|
||||
def test_unchanged_head_with_valid_override_allows_approval(self):
|
||||
blocker = _blocking_review()
|
||||
feedback = _feedback(
|
||||
has_blocking_change_requests=True,
|
||||
reviews=[blocker],
|
||||
)
|
||||
report = (
|
||||
"Blocker text: Tests fail on pinned head; fix test_commit_files_gate.\n"
|
||||
"Override: wrong_validation_environment — CI used stale worktree."
|
||||
)
|
||||
result = assess_request_changes_approval_proof(
|
||||
feedback,
|
||||
override_reason="wrong_validation_environment",
|
||||
override_explanation="CI used stale worktree; local rerun passed.",
|
||||
report_text=report,
|
||||
)
|
||||
self.assertTrue(result["approve_allowed"])
|
||||
self.assertEqual(
|
||||
result["blocking_review"]["blocking_reviewer"], "reviewer-a"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -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