feat(agent): added implementation authorization gate for branch/PR tools

- Added `is_implementation_authorizer` check requiring OWNER or allowlisted maintainer to authorize implementation work.
- Blocked `gh_push_branch` and `gh_open_pr` for unclassified/enhancement/proposal issues without explicit directive authorization.
- Auto-allowed bug and documentation issues without requiring a directive.
- Propagated `authorizes_impl` flag through events, server, tasks, and worker bindings.

python/robomp/src/github_events.py

@@ -31,6 +31,7 @@ class RouteDecision:
     directive_body: str | None = None
     directive_author: str | None = None
     directive_pragmas: tuple[tuple[str, str], ...] = ()
+    directive_authorizes_impl: bool = False
 
     @property
     def should_queue(self) -> bool:
@@ -125,6 +126,20 @@ def is_maintainer(
     return False
 
 
+def is_implementation_authorizer(
+    login: str | None,
+    association: str | None,
+    *,
+    maintainers: frozenset[str],
+) -> bool:
+    """Return whether this author may authorize implementation work."""
+    if isinstance(login, str) and login and login.lower() in maintainers:
+        return True
+    if isinstance(association, str) and association.upper() == "OWNER":
+        return True
+    return False
+
+
 def route(
     event_type: str,
     payload: Mapping[str, Any],
@@ -184,6 +199,7 @@ def route(
                 "directive_body": cleaned,
                 "directive_author": rb_login,
                 "directive_pragmas": pragmas,
+                "directive_authorizes_impl": False,
             }
         if not is_maintainer(login, assoc, maintainers=maintainers):
             return {}
@@ -191,11 +207,13 @@ def route(
         if stripped is None:
             return {}
         cleaned, pragmas = parse_pragmas(stripped)
+        authorizes_impl = is_implementation_authorizer(login, assoc, maintainers=maintainers)
         return {
             "directive": True,
             "directive_body": cleaned,
             "directive_author": login,
             "directive_pragmas": pragmas,
+            "directive_authorizes_impl": authorizes_impl,
         }
 
     if event_type == "issues":
@@ -352,6 +370,7 @@ __all__ = [
     "TRUSTED_ASSOCIATIONS",
     "extract_mention",
     "is_maintainer",
+    "is_implementation_authorizer",
     "rate_limit_cap",
     "route",
     "verify_signature",

python/robomp/src/host_tools.py

@@ -113,6 +113,9 @@ class ToolBindings:
     # True only for incoming-PR review tasks. Review tools require it; mutating
     # branch/PR publication tools reject when it is set.
     review_mode: bool = False
+    # Current task is driven by an allowlist/OWNER maintainer directive that
+    # authorizes implementation. Gates first-PR creation on non-bug/doc issues.
+    impl_authorized: bool = False
     slot_uid: int | None = None
     # Set by the worker before launching omp. Carries the abort-task signal
     # back out to the worker; `None` for unit tests that exercise tools
@@ -623,6 +626,7 @@ def _build_push_branch(bindings: ToolBindings) -> HostTool[Any, Any]:
             msg = "refusing to push: PR review worktrees are read-only."
             _audit(bindings, "gh_push_branch", args, error=msg)
             _raise_command(msg)
+        _enforce_impl_authorization(bindings, "gh_push_branch", args, action="push branch")
         branch = str(args.get("branch") or bindings.workspace.branch)
         skip = bool(args.get("skip_checks", False))
         # Same gate as gh_open_pr — formatter + check before bytes leave the
@@ -664,6 +668,7 @@ def _build_open_pr(bindings: ToolBindings) -> HostTool[Any, Any]:
             msg = "refusing to open PR: PR review tasks are read-only."
             _audit(bindings, "gh_open_pr", args, error=msg)
             _raise_command(msg)
+        _enforce_impl_authorization(bindings, "gh_open_pr", args, action="open PR")
         title = args.get("title")
         body = args.get("body")
         if not isinstance(title, str) or not title.strip():
@@ -982,6 +987,7 @@ def _build_fetch_thread(bindings: ToolBindings) -> HostTool[Any, Any]:
 
 
 _PRIMARY_TYPES = ("bug", "enhancement", "question", "proposal", "documentation", "invalid", "duplicate")
+_AUTO_PR_CLASSIFICATIONS = frozenset({"bug", "documentation"})
 _PRIORITIES = ("prio:p0", "prio:p1", "prio:p2", "prio:p3")
 _FUNCTIONAL = ("agent", "tool", "tui", "cli", "prompting", "sdk", "auth", "setup", "ux", "providers")
 _PLATFORMS = ("platform:linux", "platform:macos", "platform:windows", "platform:wsl")
@@ -990,6 +996,35 @@ _PR_TYPES = ("feat", "fix", "docs", "refactor", "perf", "test", "chore", "ci", "
 _CLOSING_ISSUE_RE = re.compile(r"\b(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?)\s+#(\d+)", re.IGNORECASE)
 
 
+def _enforce_impl_authorization(
+    bindings: ToolBindings,
+    tool_name: str,
+    args: Mapping[str, Any],
+    *,
+    action: str,
+) -> None:
+    """Refuse first publish on issue classes that require maintainer authorization."""
+    if bindings.impl_authorized:
+        return
+    row = bindings.db.get_issue(bindings.issue_key)
+    if row is not None:
+        if row.pr_number is not None:
+            return
+        classification = row.classification
+        if classification in _AUTO_PR_CLASSIFICATIONS:
+            return
+    else:
+        classification = None
+    classification_phrase = f"classified `{classification}`" if classification else "not classified"
+    msg = (
+        f"refusing to {action}: issue #{bindings.issue.number} is {classification_phrase}; "
+        "a repo OWNER or allowlisted maintainer must @-mention you with an explicit go-ahead "
+        "before any branch/PR. Post your analysis with `gh_post_comment` and stop."
+    )
+    _audit(bindings, tool_name, args, error=msg)
+    _raise_command(msg)
+
+
 def _require_review_mode(bindings: ToolBindings, name: str, args: Mapping[str, Any]) -> None:
     if bindings.review_mode:
         return

python/robomp/src/manual_triage.py

@@ -53,9 +53,7 @@ def parse_issue_ref(ref: str) -> tuple[str, int]:
     cleaned = ref.strip()
     match = _ISSUE_REF.match(cleaned) or _ISSUE_URL.match(cleaned)
     if match is None:
-        raise InvalidIssueRef(
-            f"expected owner/repo#NN or https://github.com/owner/repo/issues/NN, got {ref!r}"
-        )
+        raise InvalidIssueRef(f"expected owner/repo#NN or https://github.com/owner/repo/issues/NN, got {ref!r}")
     return f"{match.group('owner')}/{match.group('repo')}", int(match.group("number"))
 
 

python/robomp/src/prompts/directive.md

@@ -24,7 +24,7 @@ Read the thread first — reviewer bots (e.g. `chatgpt-codex-connector`) often r
 
 Then branch on request type:
 
-- **Code change** → commit on `{{workspace.branch}}`. NEVER open a second PR; push to this branch. `gh_push_branch` / `gh_open_pr` run `bun run fix` + `bun check` before contacting the remote — you do NOT. After pushing, reply with ONE `gh_post_comment` summarizing the fix, one line per concrete change. Directive bundles multiple issues (e.g. several inline review comments)? Address each and group them in the reply.
+- **Code change** → commit on `{{workspace.branch}}`. NEVER open a second PR; push to this branch. `gh_push_branch` / `gh_open_pr` run `bun run fix` + `bun check` before contacting the remote — you do NOT. If these tools refuse on an enhancement/proposal because the directive author lacks implementation authority, reply with ONE `gh_post_comment` explaining that a repo OWNER or allowlisted maintainer must explicitly authorize implementation, then stop. After pushing, reply with ONE `gh_post_comment` summarizing the fix, one line per concrete change. Directive bundles multiple issues (e.g. several inline review comments)? Address each and group them in the reply.
 - **Question / clarification** → one `gh_post_comment`. No code change.
 - **Explicit stop / drop this** → one ack comment, then halt.
 - **Ambiguous** → exactly one clarifying question, then stop. NEVER guess.

python/robomp/src/prompts/followup_comment.md

@@ -17,7 +17,7 @@ Thread context: {{origin.description}}. PR state: `{{state.pr_status}}`.
 Decide what to do:
 
 - **New repro info?** Re-run via `repro_record`, then `gh_post_comment` with the outcome.
-- **PR change requested?** Amend `{{workspace.branch}}` and push; NEVER open a second PR. Reply with a short `gh_post_comment` naming what changed.
+- **PR change requested?** Amend `{{workspace.branch}}` and push only for an already-open PR / authorized implementation; NEVER open a second PR, and NEVER open the first PR for an unauthorized enhancement/proposal. Reply with a short `gh_post_comment` naming what changed.
 - **Confirmation or unrelated question?** Reply with one `gh_post_comment`. Leave code untouched.
 - **Bot author or no actionable content?** No-op.
 

python/robomp/src/server.py

@@ -379,6 +379,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
                 "body": decision.directive_body,
                 "author": decision.directive_author,
                 "pragmas": [list(item) for item in decision.directive_pragmas],
+                "authorizes_impl": decision.directive_authorizes_impl,
             }
 
         if not decision.should_queue:

python/robomp/src/tasks.py

@@ -54,7 +54,12 @@ def _directive_from_payload(payload: Mapping[str, Any]) -> DirectiveInfo | None:
                 k, v = entry
                 if isinstance(k, str) and isinstance(v, str):
                     pragmas.append((k, v))
-    return DirectiveInfo(body=body, author=author, pragmas=tuple(pragmas))
+    return DirectiveInfo(
+        body=body,
+        author=author,
+        pragmas=tuple(pragmas),
+        authorizes_impl=bool(raw.get("authorizes_impl")),
+    )
 
 
 async def _fetch_thread(

python/robomp/src/worker.py

@@ -86,6 +86,7 @@ class DirectiveInfo:
     author: str
     thread: tuple[ThreadMessage, ...] = ()
     pragmas: tuple[tuple[str, str], ...] = ()
+    authorizes_impl: bool = False
 
 
 def _resolve_pragma_overrides(
@@ -685,6 +686,7 @@ async def run_task(
         inbound_thread_number=pr_number,
         inbound_is_pr=pr_number is not None,
         review_mode=review_mode,
+        impl_authorized=bool(directive is not None and directive.authorizes_impl),
         slot_uid=inputs.slot_uid,
         abort=AbortController(),
     )

python/robomp/tests/test_github_events.py

@@ -5,6 +5,7 @@ import hmac
 
 from robomp.github_events import (
     extract_mention,
+    is_implementation_authorizer,
     is_maintainer,
     rate_limit_cap,
     route,
@@ -545,6 +546,17 @@ def test_is_maintainer_rejects_contributor_and_none() -> None:
     assert is_maintainer(None, "OWNER", maintainers=frozenset())  # association still wins
 
 
+def test_is_implementation_authorizer_accepts_allowlist_and_owner() -> None:
+    assert is_implementation_authorizer("can1357", None, maintainers=frozenset({"can1357"}))
+    assert is_implementation_authorizer("Can1357", "NONE", maintainers=frozenset({"can1357"}))
+    assert is_implementation_authorizer("stranger", "OWNER", maintainers=frozenset())
+
+
+def test_is_implementation_authorizer_rejects_non_owner_associations() -> None:
+    for assoc in ("MEMBER", "COLLABORATOR", "NONE", "CONTRIBUTOR", None):
+        assert not is_implementation_authorizer("stranger", assoc, maintainers=frozenset()), assoc
+
+
 def test_route_directive_set_on_issue_comment_when_owner_mentions_bot() -> None:
     decision = route(
         "issue_comment",
@@ -565,6 +577,7 @@ def test_route_directive_set_on_issue_comment_when_owner_mentions_bot() -> None:
     assert decision.directive is True
     assert decision.directive_body == "please refactor X"
     assert decision.directive_author == "can1357"
+    assert decision.directive_authorizes_impl is True
 
 
 def test_route_directive_set_when_login_in_maintainers_list() -> None:
@@ -587,6 +600,28 @@ def test_route_directive_set_when_login_in_maintainers_list() -> None:
     assert decision.directive is True
     assert decision.directive_body == "do it"
     assert decision.directive_author == "can1357"
+    assert decision.directive_authorizes_impl is True
+
+
+def test_route_directive_from_collaborator_does_not_authorize_impl() -> None:
+    decision = route(
+        "issue_comment",
+        {
+            "action": "created",
+            "comment": {
+                "user": {"login": "oldschoola"},
+                "author_association": "COLLABORATOR",
+                "body": "@robomp-bot go ahead with the plan",
+            },
+            "issue": {"number": 9},
+            "repository": {"full_name": "octo/widget"},
+        },
+        allowlist=ALLOWLIST,
+        bot_login=BOT,
+    )
+    assert decision.directive is True
+    assert decision.directive_body == "go ahead with the plan"
+    assert decision.directive_authorizes_impl is False
 
 
 def test_route_directive_unset_for_random_user_even_with_mention() -> None:
@@ -719,6 +754,7 @@ def test_route_reviewer_bot_review_comment_is_directive() -> None:
     assert decision.directive is True
     assert decision.directive_body == "This branch leaks memory."
     assert decision.directive_author == "chatgpt-codex-connector"
+    assert decision.directive_authorizes_impl is False
 
 
 def test_route_random_bot_still_skipped_when_not_in_reviewer_list() -> None:

python/robomp/tests/test_host_tools.py

@@ -350,6 +350,7 @@ def test_gh_post_comment_propagates_github_error(db: Database, tmp_path: Path) -
 def test_gh_open_pr_requires_template_sections(db: Database, tmp_path: Path) -> None:
     transport = httpx.MockTransport(lambda r: httpx.Response(500))
     bindings, loop, t = _bindings(db, tmp_path, transport)
+    db.set_issue_classification(bindings.issue_key, "bug")
     try:
         tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
         with pytest.raises(RpcCommandError) as exc:
@@ -942,6 +943,90 @@ def test_review_mode_rejects_push_and_open_pr_before_repo_commands(db: Database,
     assert calls == []
 
 
+def test_impl_gate_rejects_unauthorized_proposal_before_repo_commands(
+    db: Database, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    calls: list[list[str] | tuple[str, ...]] = []
+
+    def record_repo_command(_bindings: ToolBindings, cmd: list[str] | tuple[str, ...], *, timeout: float | None = None):
+        del timeout
+        calls.append(cmd)
+        raise AssertionError("repo command must not run before implementation authorization")
+
+    bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
+    db.set_issue_classification(bindings.issue_key, "proposal")
+    monkeypatch.setattr(host_tools, "_run_repo_command", record_repo_command)
+    try:
+        push = next(x for x in build(bindings) if x.name == "gh_push_branch")
+        open_pr = next(x for x in build(bindings) if x.name == "gh_open_pr")
+        with pytest.raises(RpcCommandError) as push_exc:
+            push.execute({}, _ctx())
+        with pytest.raises(RpcCommandError) as pr_exc:
+            open_pr.execute({"title": "fix: x", "body": "invalid"}, _ctx())
+    finally:
+        _stop_loop(loop, t)
+
+    for msg in (str(push_exc.value), str(pr_exc.value)):
+        assert "classified `proposal`" in msg
+        assert "OWNER or allowlisted maintainer" in msg
+        assert "gh_post_comment" in msg
+    assert calls == []
+    rows = db._conn.execute(
+        "SELECT tool, error FROM tool_calls WHERE tool IN ('gh_push_branch', 'gh_open_pr') ORDER BY id"
+    ).fetchall()
+    assert [row["tool"] for row in rows] == ["gh_push_branch", "gh_open_pr"]
+    assert all("classified `proposal`" in row["error"] for row in rows)
+
+
+def test_impl_gate_allows_authorized_proposal_to_reach_pr_validation(db: Database, tmp_path: Path) -> None:
+    from dataclasses import replace
+
+    bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
+    db.set_issue_classification(bindings.issue_key, "proposal")
+    bindings = replace(bindings, impl_authorized=True)
+    try:
+        tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
+        with pytest.raises(RpcCommandError) as exc:
+            tool.execute({"title": "fix: x", "body": ""}, _ctx())
+    finally:
+        _stop_loop(loop, t)
+
+    msg = str(exc.value)
+    assert "requires a non-empty 'body'" in msg
+    assert "OWNER or allowlisted maintainer" not in msg
+
+
+def test_impl_gate_allows_bug_without_directive_to_reach_pr_validation(db: Database, tmp_path: Path) -> None:
+    bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
+    db.set_issue_classification(bindings.issue_key, "bug")
+    try:
+        tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
+        with pytest.raises(RpcCommandError) as exc:
+            tool.execute({"title": "fix: x", "body": ""}, _ctx())
+    finally:
+        _stop_loop(loop, t)
+
+    msg = str(exc.value)
+    assert "requires a non-empty 'body'" in msg
+    assert "OWNER or allowlisted maintainer" not in msg
+
+
+def test_impl_gate_allows_existing_proposal_pr_to_reach_pr_validation(db: Database, tmp_path: Path) -> None:
+    bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda _r: httpx.Response(500)))
+    db.set_issue_classification(bindings.issue_key, "proposal")
+    db.set_issue_pr(bindings.issue_key, 7)
+    try:
+        tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
+        with pytest.raises(RpcCommandError) as exc:
+            tool.execute({"title": "fix: x", "body": ""}, _ctx())
+    finally:
+        _stop_loop(loop, t)
+
+    msg = str(exc.value)
+    assert "requires a non-empty 'body'" in msg
+    assert "OWNER or allowlisted maintainer" not in msg
+
+
 def test_classify_issue_on_pr_thread_is_noop(db: Database, tmp_path: Path) -> None:
     """On PR threads the tool must not hit GitHub and must not raise."""
     calls: list[str] = []
@@ -1279,6 +1364,7 @@ def test_gh_push_branch_rejects_wrong_identity(db: Database, tmp_path: Path) ->
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         with pytest.raises(RpcCommandError) as exc:
             tool.execute({}, _ctx())
@@ -1397,6 +1483,7 @@ def test_gh_open_pr_rejects_wrong_identity_before_push_or_pr(db: Database, tmp_p
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
         body = "## Repro\nrepro\n\n## Cause\ncause\n\n## Fix\nfix\n\n## Verification\nran tests\n\nFixes #42\n"
         with pytest.raises(RpcCommandError) as exc:
@@ -1517,6 +1604,7 @@ def test_gh_push_branch_rejects_invalid_identity_scan_range(db: Database, tmp_pa
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         with pytest.raises(RpcCommandError) as exc:
             tool.execute({}, _ctx())
@@ -1542,6 +1630,7 @@ def test_gh_push_branch_rejects_invalid_identity_scan_range(db: Database, tmp_pa
 def test_gh_open_pr_requires_closes_keyword(db: Database, tmp_path: Path) -> None:
     """gh_open_pr refuses if the body has the four sections but no Fixes/Closes/Resolves keyword."""
     bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(lambda r: httpx.Response(500)))
+    db.set_issue_classification(bindings.issue_key, "bug")
     try:
         tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
         body = "## Repro\nrepro\n\n## Cause\ncause\n\n## Fix\nfix\n\n## Verification\nran tests\n"
@@ -1574,6 +1663,7 @@ def test_gh_open_pr_refuses_failed_bun_check_before_push_or_pr(
         )
 
     bindings, loop, t = _bindings(db, tmp_path, httpx.MockTransport(handler))
+    db.set_issue_classification(bindings.issue_key, "bug")
     fakebin = tmp_path / "fakebin"
     fakebin.mkdir()
     fake_bun = fakebin / "bun"
@@ -1713,6 +1803,7 @@ def test_gh_push_branch_rejects_dirty_worktree(db: Database, tmp_path: Path) ->
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         with pytest.raises(RpcCommandError) as exc:
             tool.execute({}, _ctx())
@@ -1860,6 +1951,7 @@ def test_gh_push_branch_runs_fix_and_check_before_pushing(
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         result = tool.execute({}, _ctx())
     finally:
@@ -1993,6 +2085,7 @@ def test_gh_push_branch_force_with_lease_recovers_after_amend(db: Database, tmp_
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         tool.execute({}, _ctx())
 
@@ -2171,6 +2264,7 @@ def test_gh_push_branch_aborts_on_failed_bun_check(
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         with pytest.raises(RpcCommandError) as exc:
             tool.execute({}, _ctx())
@@ -2326,6 +2420,7 @@ def test_gh_push_branch_skip_checks_bypasses_failing_bun_check(
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         result = tool.execute({"skip_checks": True}, _ctx())
     finally:
@@ -2463,6 +2558,7 @@ def test_gh_push_branch_skip_checks_still_refuses_dirty_worktree(
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         with pytest.raises(RpcCommandError) as exc:
             tool.execute({"skip_checks": True}, _ctx())
@@ -2629,6 +2725,7 @@ def test_gh_open_pr_runs_fix_then_check_and_commits_fixup(
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
         body = "## Repro\nrepro\n\n## Cause\ncause\n\n## Fix\nfix\n\n## Verification\nran tests\n\nFixes #42\n"
         result = tool.execute({"title": "fix: x", "body": body}, _ctx())
@@ -2802,6 +2899,7 @@ def test_gh_open_pr_refuses_dirty_worktree_before_fix(
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         push_tool = next(x for x in build(bindings) if x.name == "gh_push_branch")
         with pytest.raises(RpcCommandError) as exc:
             push_tool.execute({}, _ctx())
@@ -2985,6 +3083,7 @@ def test_gh_open_pr_skips_fix_when_no_script(db: Database, tmp_path: Path, monke
             branch=ws.branch,
             session_dir=str(ws.session_dir),
         )
+        db.set_issue_classification(bindings.issue_key, "bug")
         tool = next(x for x in build(bindings) if x.name == "gh_open_pr")
         body = "## Repro\nrepro\n\n## Cause\ncause\n\n## Fix\nfix\n\n## Verification\nran tests\n\nFixes #42\n"
         result = tool.execute({"title": "fix: x", "body": body}, _ctx())

python/robomp/tests/test_server.py

@@ -1500,7 +1500,7 @@ def test_webhook_directive_on_unknown_issue_is_queued_with_metadata(env) -> None
     assert row is not None
     assert row.state == "queued"
     directive = row.payload.get("_robomp_directive")
-    assert directive == {"body": "please refactor X", "author": "can1357", "pragmas": []}
+    assert directive == {"body": "please refactor X", "author": "can1357", "pragmas": [], "authorizes_impl": True}
 
 
 def test_webhook_maintainer_bypasses_rate_limit(

python/robomp/tests/test_tasks_directive.py

@@ -19,12 +19,14 @@ def test_directive_from_payload_parses_pragmas() -> None:
     assert directive.body == "do the thing"
     assert directive.author == "can1357"
     assert directive.pragmas == (("model", "gpt"), ("thinking", "low"))
+    assert directive.authorizes_impl is False
 
 
 def test_directive_from_payload_missing_pragmas_is_empty_tuple() -> None:
     directive = _directive_from_payload({"_robomp_directive": {"body": "x", "author": "can1357"}})
     assert directive is not None
     assert directive.pragmas == ()
+    assert directive.authorizes_impl is False
 
 
 def test_directive_from_payload_drops_malformed_pragma_entries() -> None:
@@ -46,6 +48,20 @@ def test_directive_from_payload_drops_malformed_pragma_entries() -> None:
     assert directive.pragmas == (("model", "gpt"),)
 
 
+def test_directive_from_payload_parses_implementation_authorization() -> None:
+    directive = _directive_from_payload(
+        {
+            "_robomp_directive": {
+                "body": "do the thing",
+                "author": "can1357",
+                "authorizes_impl": True,
+            }
+        }
+    )
+    assert directive is not None
+    assert directive.authorizes_impl is True
+
+
 def test_directive_from_payload_returns_none_for_missing_directive() -> None:
     assert _directive_from_payload({}) is None
     assert _directive_from_payload({"_robomp_directive": "not-a-mapping"}) is None

python/robomp/tests/test_worker.py

@@ -157,6 +157,40 @@ def _patch_worker(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
     )
 
 
+@pytest.mark.asyncio
+async def test_run_task_sets_impl_authorized_from_directive(
+    tmp_path: Path, settings: Settings, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    inputs, _bindings = _make_inputs(tmp_path, settings, session_has_jsonl=False)
+    captured: dict[str, bool] = {}
+
+    monkeypatch.setattr(worker, "_build_prompt", lambda *args, **kwargs: "prompt")
+
+    def fake_run_rpc_blocking(
+        _inputs: worker.TaskInputs,
+        *,
+        task_kind: str,
+        prompt: str,
+        loop: asyncio.AbstractEventLoop,
+        bindings: worker.ToolBindings,
+        directive: worker.DirectiveInfo | None = None,
+    ) -> str:
+        del task_kind, prompt, loop, directive
+        captured["impl_authorized"] = bindings.impl_authorized
+        return "ok"
+
+    monkeypatch.setattr(worker, "_run_rpc_blocking", fake_run_rpc_blocking)
+
+    result = await worker.run_task(
+        task_kind="triage_issue",
+        inputs=inputs,
+        directive=worker.DirectiveInfo(body="go ahead", author="can1357", authorizes_impl=True),
+    )
+
+    assert result == "ok"
+    assert captured == {"impl_authorized": True}
+
+
 @pytest.mark.asyncio
 async def test_run_rpc_passes_continue_when_session_jsonl_present(tmp_path: Path, settings: Settings) -> None:
     inputs, bindings = _make_inputs(tmp_path, settings, session_has_jsonl=True)