feat(AGX1-275): per-RPC task permission rewire and 404/403 wrap

[asherfink]

Related work

Parent epic: AGX1-264 — per-task FGAC. Follow-ups bundled in AGX1-291.

This change is part of a 5-PR stack across 3 repos. Merge order: scaleapi/scaleapi#144783 (sgp-authz 0.7.1) ✅ already merged → scaleapi/scaleapi#145000 + scaleapi/scaleapi#145044 + scaleapi/agentex#353 → #246 → this PR.

Repo	PR	Purpose
scaleapi/scaleapi	scaleapi/scaleapi#144783 ✅ merged	sgp-authz 0.7.1 — Action.CANCEL
scaleapi/scaleapi	scaleapi/scaleapi#145000	register FGAC_AGENTEX_AUTH_SPARK flag
scaleapi/scaleapi	scaleapi/scaleapi#145044	add cancel to SGP's AgentexOperation enum + role map
scaleapi/agentex	scaleapi/agentex#353	agentex-auth per-account routing + cancel op
scaleapi/scale-agentex	#246	task creator audit columns + FGAC dual-write + flag
scaleapi/scale-agentex	this PR	per-RPC operation rewire + 404/403 wrap

Last in the stack — this is the route-side change that actually exercises the permissions written by #246.

Summary

• Routes each task-resource RPC to the correct AuthorizedOperationType instead of using execute everywhere: MESSAGE_SEND / EVENT_SEND → update, TASK_CANCEL → cancel, TASK_CREATE stays create.
• Broadens the execute → update swap across messages.py, checkpoints.py, and states.py so the editor role can send messages and update checkpoints/state without needing owner. The task SpiceDB schema defines permission update = (editor + owner) & internal_tenant_gate, so leaving these on execute (owner-only) would lock editors out of routine mutations.
• Collapses every task-resource denial into 404 (via ItemDoesNotExist) across all surfaces — path id, query id, body id, and name routes — so callers can no longer distinguish "task present in another tenant" from "task absent" by comparing 403 vs 404.
• Extracts the collapse helper to src/utils/task_authorization.py, reused from both the FastAPI dep factories and the RPC authorize hook.
• Unknown AgentRPCMethod values now raise NotImplementedError rather than falling through authz-free — defense-in-depth so a new RPC must be explicitly wired before it can dispatch.

What changed

• src/utils/task_authorization.py (new): check_task_or_collapse_to_404(authorization, task_id, operation) — the shared wrap.
• src/utils/authorization_shortcuts.py: DAuthorizedId / DAuthorizedQuery / DAuthorizedBodyId route task checks through the wrap; their inner deps no longer take a task_repository (parameter was unused). DAuthorizedName now applies the wrap when resource_type == AgentexResourceType.task — previously the name surface leaked 403 vs 404 because tasks.name is globally unique, so a probe checked the whole system rather than a single tenant.
• src/api/routes/agents.py _authorize_rpc_request: each task-resource branch routes through the wrap. The MESSAGE_SEND block with task_name is restructured to a try/else shape so a denied-update on an existing task surfaces as 404 (it must NOT fall through to the create-fallback except — that would silently promote a denied-update into a create check, which is a privilege escalation footgun). The unknown-method case (case _:) now raises NotImplementedError.
• src/api/routes/messages.py / checkpoints.py / states.py: execute → update for routine mutations.
• TASK_CREATE and the wildcard task("*") checks intentionally untouched.

AGX1-275 list deliverable — already covered

The AGX1-275 list-filtering deliverable (only return tasks the caller can read) is already met by existing infrastructure that this PR does not change:

• DAuthorizedResourceIds(AgentexResourceType.task) at src/utils/authorization_shortcuts.py:130 resolves the per-principal accessible task id set.
• It is wired into the list route at src/api/routes/tasks.py:82, which intersects user filters with the accessible set before hitting the repository.

No additional code is needed here — the list surface is already authorized end-to-end.

Pre-merge wire dependency

The execute → update swap changes the operation literal that hits agentex-auth's check endpoint. The SGP gateway forwards the literal verbatim, so the SGP permissions backend must resolve both update and cancel against the existing OWNER grant before this deploys:

• update is already in SGP's AgentexOperation enum on master — no work needed.
• cancel is added by scaleapi/scaleapi#145044. This PR is held until that PR ships everywhere.

For Spark-routed accounts the dependency chain is scaleapi/agentex#353 + sgp-authz 0.7.1 (already merged) → ResourceRole.OWNER → Action.CANCEL, all of which are in place upstream of this PR.

Tests

• tests/unit/api/test_tasks_authz.py — 17 tests pass:
  • 8 TestPerRpcOperationRouting tests (incl. MESSAGE_SEND create-fallback preserved through the restructure).
  • 2 TestCheckTaskOrCollapseTo404 tests (allow + denied-collapses-to-404).
  • 3 TestDAuthorizedBodyIdTaskWrap tests.
  • 3 TestDAuthorizedNameTaskWrap tests (denied-task → 404, allow returns name, agent path unaffected).
  • Wire-contract test pins cancel → "cancel" (mirrors agentex-auth's).
• Ruff + ruff-format clean (pre-commit hooks).

Out of scope / follow-ups (tracked in AGX1-291)

• /agents/name/{agent_name} has the same leak shape — agent FGAC is outside AGX1-264.
• Restoring the 403/404 split for same-tenant calls once tasks carry tenant/account_id scope at the data layer (AGX1-290).

Greptile Summary

This PR is the final route-side piece of the AGX1-264 per-task FGAC stack. It rewires _authorize_rpc_request to use operation-specific literals (update for MESSAGE_SEND/EVENT_SEND, cancel for TASK_CANCEL), collapses every task-resource authorization denial into a 404 (via a shared check_task_or_collapse_to_404 helper), and guards the unknown-method case with NotImplementedError so new RPCs cannot slip through authorization-free.

• src/utils/task_authorization.py (new): check_task_or_collapse_to_404 wraps any AuthorizationError on a task resource into ItemDoesNotExist, eliminating the cross-tenant existence leak that tasks.name (globally unique) would otherwise expose via 403 vs 404 divergence.
• src/utils/authorization_shortcuts.py: All four dep factories (DAuthorizedId, DAuthorizedQuery, DAuthorizedBodyId, DAuthorizedName) now route task-type resources through the collapse helper; the MESSAGE_SEND-with-task_name branch uses a try/else shape so a denied-update can never fall through to the create-fallback check (privilege escalation footgun addressed).
• src/api/routes/{messages,checkpoints}.py: execute → update swap for routine mutations so the editor role can send messages and update checkpoints/state without needing owner.

Confidence Score: 4/5

Safe to merge once the upstream wire dependency (scaleapi/scaleapi#145044 shipping cancel into SGP's AgentexOperation enum) is confirmed live; the execute→update swap for MESSAGE_SEND/EVENT_SEND is already safe per SGP master.

The routing logic, collapse helper, and try/else privilege-escalation guard are all correct and well-tested. The one remaining gate — cancel reaching every SGP region before this deploys — is explicitly documented in the PR description as a hold condition; no in-repo verification of that live state exists. The change is otherwise clean and the score reflects that single external dependency rather than any defect in the code itself.

No files contain defects. The pre-merge wire dependency on scaleapi/scaleapi#145044 is the only open gate, documented in the PR description and in the existing review comment on agents.py.

Important Files Changed

Filename	Overview
agentex/src/utils/task_authorization.py	New helper that collapses AuthorizationError → ItemDoesNotExist (404) for task resources; logic is simple, well-documented with a TODO for the 403/404 restore, and fully unit-tested.
agentex/src/api/routes/agents.py	Per-RPC operation rewire is correct; try/else shape for MESSAGE_SEND task-name path prevents the denied-update → create-fallback privilege escalation; unknown-method guard raises NotImplementedError defensively.
agentex/src/utils/authorization_shortcuts.py	All dep factories now route task resources through the collapse helper; DAuthorizedName's lookup-before-authz pattern is safe because both absent-name and denied-name paths produce 404.
agentex/src/api/routes/messages.py	execute → update swap on all four message mutation endpoints; straightforward mechanical change, correctly enables editor-role access per the SpiceDB schema.
agentex/src/api/routes/checkpoints.py	execute → update swap on put_checkpoint and put_writes; import block reordered alphabetically as a side effect, no functional change.
agentex/src/api/schemas/authorization_types.py	Adds cancel to AuthorizedOperationType StrEnum; wire-format value "cancel" is pinned by a dedicated unit test.
agentex/tests/unit/api/test_tasks_authz.py	17 tests covering per-RPC routing, 404-collapse semantics, body-id and name-path dep factories, wire-format contract pins, and the unwired-method fail-closed case; tests are well-structured and isolate the authz gateway via mocks.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    RPC["Incoming AgentRPC Request"] --> MATCH{"match request.method"}

    MATCH -->|TASK_CREATE| CREATE_CHECK["authorization.check\ntask(*) · create"]
    MATCH -->|MESSAGE_SEND| MSG_ID{"task_id present?"}
    MATCH -->|EVENT_SEND| EVT_ID{"task_id present?"}
    MATCH -->|TASK_CANCEL| CXL_ID{"task_id present?"}
    MATCH -->|unknown| NIE["raise NotImplementedError\n→ 500"]

    MSG_ID -->|yes| MSG_WRAP_ID["check_task_or_collapse_to_404\ntask_id · update"]
    MSG_ID -->|no, task_name| MSG_LOOKUP["task_service.get_task(name)"]
    MSG_ID -->|no task_id/name| MSG_CREATE["authorization.check\ntask(*) · create"]
    MSG_LOOKUP -->|ItemDoesNotExist| MSG_CREATE
    MSG_LOOKUP -->|found| MSG_WRAP_NAME["check_task_or_collapse_to_404\nexisting_task.id · update"]

    EVT_ID -->|yes| EVT_WRAP_ID["check_task_or_collapse_to_404\ntask_id · update"]
    EVT_ID -->|no, task_name| EVT_LOOKUP["task_service.get_task(name)"]
    EVT_ID -->|no task_id/name| EVT_ERR["raise ValueError → 400"]
    EVT_LOOKUP --> EVT_WRAP_NAME["check_task_or_collapse_to_404\nexisting_task.id · update"]

    CXL_ID -->|yes| CXL_WRAP_ID["check_task_or_collapse_to_404\ntask_id · cancel"]
    CXL_ID -->|no, task_name| CXL_LOOKUP["task_service.get_task(name)"]
    CXL_ID -->|no task_id/name| CXL_ERR["raise ValueError → 400"]
    CXL_LOOKUP --> CXL_WRAP_NAME["check_task_or_collapse_to_404\nexisting_task.id · cancel"]

    subgraph WRAP["check_task_or_collapse_to_404"]
        direction TB
        AZ["authorization.check(task, op)"] -->|AuthorizationError| FOLD["raise ItemDoesNotExist → 404"]
        AZ -->|ok| PASS["return (allow)"]
    end

    MSG_WRAP_ID & MSG_WRAP_NAME & EVT_WRAP_ID & EVT_WRAP_NAME & CXL_WRAP_ID & CXL_WRAP_NAME --> WRAP

Loading

_{Reviews (4): Last reviewed commit: "feat(AGX1-275): per-RPC task permission ..." | Re-trigger Greptile}

[harvhan]

nit: make this typed

[asherfink]

done, typed it as AuthorizationService