gitea_mcp_server.py

@@ -5291,6 +5291,7 @@ _RUNTIME_CAPABILITY_TASKS = (
     "create_issue",
     "comment_issue",
     "create_pr",
+    "work_issue",
     "review_pr",
     "merge_pr",
     "close_issue",
@@ -5343,6 +5344,7 @@ def _build_runtime_task_capabilities(
         "create_issue": "can_create_issues",
         "comment_issue": "can_comment_on_issues",
         "create_pr": "can_author_prs",
+        "work_issue": "can_work_issues",
         "review_pr": "can_review_prs",
         "merge_pr": "can_merge_prs",
         "close_issue": "can_close_issues",

review_proofs.py

@@ -2328,6 +2328,30 @@ CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
     ("Terminal review mutation", ("terminal review mutation",)),
 )
 
+WORK_ISSUE_WORKFLOW_MARKERS = (
+    "workflows/work-issue.md",
+    "work-issue.md",
+)
+
+WORK_ISSUE_TASK_MARKERS = (
+    "work-issue",
+    "work issue",
+)
+
+WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
+    ("Selected PR", ("selected pr",)),
+    ("Review decision", ("review decision",)),
+    ("Merge result", ("merge result",)),
+    ("Merge preflight", ("merge preflight",)),
+    ("Already-landed gate", ("already-landed gate",)),
+    ("Pinned reviewed head", ("pinned reviewed head",)),
+    ("Terminal review mutation", ("terminal review mutation",)),
+    ("Duplicate search terms", ("duplicate search terms",)),
+    ("Issues created", ("issues created",)),
+    ("Issues skipped as duplicates", ("issues skipped as duplicates",)),
+    ("Requested issue task", ("requested issue task",)),
+)
+
 
 def _handoff_section_lines(report_text):
     """Return the lines of the Controller Handoff section, or None."""
@@ -3520,6 +3544,104 @@ def assess_create_issue_workflow_source(report_text: str) -> dict:
     }
 
 
+def assess_work_issue_workflow_source(report_text: str) -> dict:
+    """#139: work-issue reports must cite the canonical workflow file."""
+    lower = (report_text or "").lower()
+    reasons = []
+    if not any(marker in lower for marker in WORK_ISSUE_WORKFLOW_MARKERS):
+        reasons.append(
+            "work-issue report missing workflow source "
+            "(workflows/work-issue.md)"
+        )
+    if not any(marker in lower for marker in WORK_ISSUE_TASK_MARKERS):
+        reasons.append(
+            "work-issue report missing task mode declaration (work-issue)"
+        )
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_work_issue_mode_isolation(report_text: str) -> dict:
+    """#139: reject review-merge/create-issue handoff fields in work-issue runs."""
+    section = _handoff_section_lines(report_text)
+    reasons = []
+    if section is None:
+        return {
+            "complete": True,
+            "downgraded": False,
+            "reasons": [],
+        }
+
+    labels = []
+    for line in section:
+        stripped = line.strip().lstrip("-*").strip()
+        if ":" in stripped:
+            labels.append(stripped.split(":", 1)[0].strip().lower())
+
+    for name, aliases in WORK_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS:
+        if any(
+            label.startswith(alias)
+            for label in labels
+            for alias in aliases
+        ):
+            reasons.append(
+                f"work-issue handoff must not include cross-mode field "
+                f"'{name}'"
+            )
+
+    legacy_workspace = any(
+        label.startswith("workspace mutations") for label in labels
+    )
+    precise_categories = any(
+        label.startswith("file edits by author")
+        for label in labels
+    )
+    if legacy_workspace and not precise_categories:
+        reasons.append(
+            "work-issue handoff must use precise mutation categories "
+            "instead of legacy 'Workspace mutations'"
+        )
+
+    return {
+        "complete": not reasons,
+        "downgraded": bool(reasons),
+        "reasons": reasons,
+    }
+
+
+def assess_work_issue_final_report(report_text: str) -> dict:
+    """#139: composite verifier for work-issue final reports."""
+    checks = {
+        "workflow_source": assess_work_issue_workflow_source(report_text),
+        "mode_isolation": assess_work_issue_mode_isolation(report_text),
+    }
+
+    reasons = []
+    downgraded = False
+    for name, result in checks.items():
+        verdict = result.get("verdict")
+        if verdict in ("missing", "incomplete"):
+            downgraded = True
+            reasons.extend(result.get("reasons") or [])
+        elif result.get("downgraded") or not result.get("complete", True):
+            downgraded = True
+            reasons.extend(
+                f"{name}: {r}" for r in (result.get("reasons") or [])
+            )
+
+    grade = "A" if not downgraded else "downgraded"
+    return {
+        "grade": grade,
+        "downgraded": downgraded,
+        "checks": checks,
+        "reasons": reasons,
+        "complete": not downgraded,
+    }
+
+
 def assess_create_issue_mode_isolation(report_text: str) -> dict:
     """#337: reject work-issue/review-merge handoff fields in create-issue runs."""
     section = _handoff_section_lines(report_text)

skills/llm-project-workflow/workflows/work-issue.md

@@ -83,6 +83,20 @@ Examples:
 
 If capability cannot be proven, stop and produce a recovery handoff only.
 
+### 2A. Pre-task role routing (#139)
+
+Before issue selection or any mutation, resolve the composite author task:
+
+* `gitea_route_task_session(task_type="work-issue", remote=…)` — must return
+  `route_result: allowed_current_session` and `downstream_allowed: true`
+* `gitea_resolve_task_capability(task="work_issue", remote=…)` — must show
+  `required_role_kind: author` and `allowed_in_current_session: true`
+
+Hyphen (`work-issue`) and underscore (`work_issue`) aliases are equivalent.
+If routing returns `ambiguous_task_stop`, `wrong_role_stop`, or
+`route_to_author_session`, stop and produce a recovery handoff only — do not
+fall back to reviewer tools or guess a different profile.
+
 ## 3. Stop immediately on blocked infrastructure
 
 If any of the following appears, stop immediately:

task_capability_map.py

@@ -104,6 +104,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    "work_issue": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
+    "work-issue": {
+        "permission": "gitea.pr.create",
+        "role": "author",
+    },
     "reconcile_landed_pr": {
         "permission": "gitea.read",
         "role": "author",

tests/test_llm_workflow_split.py

@@ -100,6 +100,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_pr_queue_cleanup_workflow_contract():
@@ -181,6 +183,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
 

tests/test_resolve_task_capability.py

@@ -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):

tests/test_review_proofs.py

@@ -40,6 +40,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,
@@ -2320,6 +2323,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."""
 

tests/test_role_session_router.py

@@ -192,6 +192,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):