feat: register work-issue task routing for role-aware MCP (#139)

- Map work_issue/work-issue to author role and gitea.pr.create in resolver
- Route work-issue sessions through role_session_router before mutations
- Expose work_issue in runtime context task capabilities
- Add work-issue workflow source verifier and canonical routing docs
- Tests for resolver, router, and final-report verifier

Closes #139

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-07 09:30:38 -04:00
co-authored by Claude Opus 4.8
parent 7b71113b4d
commit c260ef15fb
9 changed files with 258 additions and 0 deletions
+8
View File
@@ -91,6 +91,8 @@ def test_work_issue_workflow_contract():
assert "canonical: true" in text
assert "Do not merge your own PR" in text
assert "## 21. PR creation rules" in text
assert "gitea_route_task_session" in text
assert "work-issue" in text
def test_final_report_schemas_exist():
@@ -136,6 +138,12 @@ def test_create_issue_final_report_verifier_exported():
assert callable(assess_create_issue_final_report)
def test_work_issue_final_report_verifier_exported():
from review_proofs import assess_work_issue_final_report
assert callable(assess_work_issue_final_report)
def test_merge_simulation_verifier_exported():
from review_proofs import assess_merge_simulation_report
+26
View File
@@ -121,6 +121,32 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertTrue(res["allowed_in_current_session"])
self.assertFalse(res["different_mcp_namespace_required"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_work_issue_author_profile_allowed(self, _auth, _api):
with patch.dict(os.environ, self._env("author-profile")):
for task in ("work_issue", "work-issue"):
res = mcp_server.gitea_resolve_task_capability(
task=task, remote="prgs"
)
self.assertEqual(res["requested_task"], task)
self.assertEqual(
res["required_operation_permission"], "gitea.pr.create"
)
self.assertEqual(res["required_role_kind"], "author")
self.assertTrue(res["allowed_in_current_session"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_work_issue_reviewer_profile_blocked(self, _auth, _api):
with patch.dict(os.environ, self._env("reviewer-profile")):
res = mcp_server.gitea_resolve_task_capability(
task="work_issue", remote="prgs"
)
self.assertFalse(res["allowed_in_current_session"])
self.assertEqual(res["required_role_kind"], "author")
self.assertTrue(res["different_mcp_namespace_required"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_issue_comment_task(self, _auth, _api):
+49
View File
@@ -38,6 +38,9 @@ from review_proofs import ( # noqa: E402
assess_create_issue_final_report,
assess_create_issue_mode_isolation,
assess_create_issue_workflow_source,
assess_work_issue_final_report,
assess_work_issue_mode_isolation,
assess_work_issue_workflow_source,
assess_edited_pr_inventory_coverage,
assess_empty_queue_report,
assess_fresh_issue_selection,
@@ -2318,6 +2321,52 @@ class TestCreateIssueFinalReport(unittest.TestCase):
self.assertTrue(result["downgraded"])
class TestWorkIssueFinalReport(unittest.TestCase):
"""#139: work-issue final-report verifier coverage."""
def _good_report(self):
return "\n".join([
"Task mode: work-issue",
"Workflow source: skills/llm-project-workflow/workflows/work-issue.md",
"## Controller Handoff",
"- Task: work-issue",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: jcwalker3 / prgs-author",
"- Active profile: prgs-author",
"- Runtime context: prgs-author on prgs",
"- Selected issue: #139",
"- Branch name: feat/issue-139-role-aware-work-issue-routing",
"- PR number: not created in this session",
"- File edits by author: role_session_router.py",
"- Blockers: none",
"- Current status: implementing routing",
"- Safe next action: open PR",
"- Next: open PR",
"- Safety statement: no review/merge",
])
def test_complete_work_issue_report_earns_a(self):
result = assess_work_issue_final_report(self._good_report())
self.assertEqual(result["grade"], "A")
self.assertFalse(result["downgraded"])
def test_missing_workflow_source_downgrades(self):
report = self._good_report().replace(
"workflows/work-issue.md", "SKILL.md only"
)
result = assess_work_issue_workflow_source(report)
self.assertTrue(result["downgraded"])
def test_review_field_in_handoff_downgrades(self):
report = self._good_report().replace(
"- Safety statement:",
"- Review decision: APPROVE\n- Safety statement:",
)
result = assess_work_issue_mode_isolation(report)
self.assertTrue(result["downgraded"])
class TestValidationEnvironmentProof(unittest.TestCase):
"""Issue #311: exact validation command and execution environment."""
+25
View File
@@ -157,6 +157,31 @@ class TestRoleSessionRouter(unittest.TestCase):
)
self.assertEqual(result.get("number"), 999)
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_work_issue_task_under_author_profile_allowed(self, _auth, _api):
with patch.dict(os.environ, self._env("prgs-author")):
for task_type in ("work_issue", "work-issue"):
route = mcp_server.gitea_route_task_session(
task_type=task_type, remote="prgs"
)
self.assertEqual(
route["route_result"], role_session_router.ROUTE_ALLOWED
)
self.assertTrue(route["downstream_allowed"])
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_work_issue_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
with patch.dict(os.environ, self._env("prgs-reviewer")):
route = mcp_server.gitea_route_task_session(
task_type="work-issue", remote="prgs"
)
self.assertEqual(
route["route_result"], role_session_router.ROUTE_TO_AUTHOR
)
self.assertFalse(route["downstream_allowed"])
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):