feat(tracker): FGAC authz on agent_task_tracker routes (AGX1-307) (#265)

## Summary

Adds FGAC authorization to all three `agent_task_tracker` routes by
delegating to the parent task. Trackers have no SpiceDB type of their
own so no schema changes, `register_resource` calls, or dual-write are
needed.

- **GET /tracker/{tracker_id}** checks `task.read` (view) on the parent
task, resolved via `tracker.task_id`. A denied or missing tracker
collapses to 404 rather than 403, so callers cannot probe cross-tenant
existence by comparing status codes. Side effect: a missing tracker now
returns 404 (previously 400) because the parent-task fetch runs before
the handler body.
- **GET /tracker** uses `DAuthorizedResourceIds(task)` to get the
caller's viewable task set and drops trackers under unauthorized tasks
silently. An explicit `?task_id=` the caller cannot view returns an
empty list, never a 404. `agent_id` remains a plain filter.
- **PUT /tracker/{tracker_id}** now checks `task.execute` on the parent
task (matching the message and checkpoint write routes), with the same
404 collapse on denial. This closes a cross-tenant write gap where the
read routes were locked down but the mutating route was left unenforced.

Also cleans up copy-paste artifacts from the original file (variable
naming, log strings, stale docstring args) and removes an unreachable
`task_id` scalar path in `use_case.list` whose only caller already
passes `task_ids`.

## Test plan

- Existing integration tests continue to pass (20 integration, 1 unit)
- New integration tests in `test_agent_task_tracker_authz_api.py`:
  - List with authz disabled returns all trackers (bypass path)
  - List with zero authorized tasks and no filter returns empty list
- PUT authorized returns 200 and records exactly one `task.execute`
check on the parent task
  - PUT unauthorized returns 404

## Rollout note

PUT enforcement is behind the same flag as the read routes. Before
enabling tracker FGAC, confirm the agent runtime that commits
cursor/status updates either runs with an authz bypass (agent API key)
or holds `task.execute` on the parent task.

<!-- greptile_comment -->

<h3>Greptile Summary</h3>

This PR adds fine-grained access control (FGAC) to the three
`agent_task_tracker` routes by delegating authorization to the parent
task. No SpiceDB schema changes are needed since trackers inherit task
permissions.

- **GET /tracker/{tracker_id}** and **PUT /tracker/{tracker_id}** use
`DAuthorizedId` with `TaskChildResourceType.agent_task_tracker` to
resolve `tracker → task_id` before the handler runs, checking
`task.read` and `task.execute` respectively; a missing or unauthorized
tracker collapses to 404.
- **GET /tracker** uses `DAuthorizedResourceIds(task)` to scope the
result set to the caller's authorized tasks, with explicit `?task_id=`
filters silently clamped to the authorized set. The use case signature
changes from a scalar `task_id` to a list `task_ids` to support the `IN
(...)` SQL pattern (with `[]` producing zero rows via SQLAlchemy's `IN
()` handling).
- Side-effect documented in the PR: a missing tracker on GET/PUT now
returns 404 (previously 400) because the parent-task resolution runs
unconditionally in the dependency, even on the bypass path.

<details><summary><h3>Confidence Score: 4/5</h3></summary>

The change is safe to merge. Authorization logic is correct for all
three routes and consistent with the existing state/message patterns.
The documented behavioral change (missing tracker → 404 instead of 400)
is intentional and the existing test suite has been updated to reflect
it.

The authz logic across GET, PUT, and list routes is well-tested and
correct. The only observations are a linear list-membership scan on
authorized_task_ids in the filter route and a redundant DB fetch of the
tracker in the GET handler — both minor and non-blocking.

agentex/src/api/routes/agent_task_tracker.py — the list-route
authorized_task_ids membership check and the GET handler's double
tracker fetch are both worth a second look before scaling.
</details>

<h3>Important Files Changed</h3>

| Filename | Overview |
|----------|----------|
| agentex/src/api/routes/agent_task_tracker.py | Adds DAuthorizedId
(read/execute) to GET/PUT handlers and DAuthorizedResourceIds to the
list handler; effective_task_ids logic is correct but uses a linear list
membership check for the explicit ?task_id= case. |
| agentex/src/utils/authorization_shortcuts.py | Adds
DAgentTaskTrackerRepository to _get_parent_task_id registry and threads
it through DAuthorizedId/DAuthorizedQuery; pattern is consistent with
existing state/message entries. |
| agentex/src/domain/use_cases/agent_task_tracker_use_case.py |
Signature change from scalar task_id to list task_ids; empty-list case
correctly produces a falsy IN () filter via SQLAlchemy; None correctly
bypasses the filter. |
|
agentex/tests/integration/api/agent_task_tracker/test_agent_task_tracker_authz_api.py
| New authz integration tests covering bypass, zero-authorized-tasks,
authorized/unauthorized GET and PUT, and explicit task_id filter cases;
mock call inspection relies on positional-arg indexing. |
|
agentex/tests/integration/api/agent_task_tracker/test_agent_task_tracker_api.py
| Updates non-existent-tracker test from 400 to 404 to reflect the
documented behaviour change caused by DAuthorizedId resolving before the
handler body. |
| agentex/tests/unit/api/test_agent_api_keys_authz.py | Adds
tracker_repository MagicMock to existing DAuthorizedId unit test
call-sites to match the updated dependency signature. |
| agentex/src/api/schemas/authorization_types.py | Adds
agent_task_tracker to TaskChildResourceType enum; minimal and correct
change. |

</details>

<details><summary><h3>Sequence Diagram</h3></summary>

```mermaid
sequenceDiagram
    participant C as Caller
    participant R as Router
    participant DI as DAuthorizedId / DAuthorizedResourceIds
    participant TR as TrackerRepository
    participant AS as AuthorizationService
    participant UC as TrackerUseCase

    Note over R,DI: GET /tracker/{tracker_id} & PUT /tracker/{tracker_id}
    C->>R: "GET/PUT /tracker/{tracker_id}"
    R->>DI: resolve DAuthorizedId(agent_task_tracker, read/execute)
    DI->>TR: "get(id=tracker_id) → task_id"
    alt tracker missing
        TR-->>DI: ItemDoesNotExist
        DI-->>C: 404
    end
    DI->>AS: check(task.read / task.execute, task_id)
    alt unauthorized
        AS-->>DI: AuthorizationError → ItemDoesNotExist
        DI-->>C: 404
    end
    DI-->>R: tracker_id (authorized)
    R->>UC: get/update tracker
    UC->>TR: get/update(tracker_id)
    TR-->>UC: AgentTaskTrackerEntity
    UC-->>R: entity
    R-->>C: 200

    Note over R,DI: GET /tracker (list)
    C->>R: "GET /tracker[?task_id=X]"
    R->>DI: resolve DAuthorizedResourceIds(task, read)
    DI->>AS: list_resources(task, read)
    AS-->>DI: authorized_task_ids (list) or None (bypass)
    DI-->>R: authorized_task_ids
    R->>R: compute effective_task_ids
    R->>UC: "list(task_ids=effective_task_ids)"
    UC->>TR: "list(filters={task_id: IN(effective_task_ids)})"
    TR-->>UC: filtered trackers
    UC-->>R: entities
    R-->>C: 200 list
```
</details>

<a
href="https://app.greptile.com/api/ide/cursor?prompt=Fix%20the%20following%202%20code%20review%20issues.%20Work%20through%20them%20one%20at%20a%20time%2C%20proposing%20concise%20fixes.%0A%0A---%0A%0A%23%23%23%20Issue%201%20of%202%0Aagentex%2Fsrc%2Fapi%2Froutes%2Fagent_task_tracker.py%3A73-76%0AWhen%20authz%20is%20enabled%20and%20an%20explicit%20%60%3Ftask_id%3D%60%20is%20provided%2C%20%60task_id%20in%20authorized_task_ids%60%20performs%20a%20linear%20O%28n%29%20scan%20over%20a%20%60list%5Bstr%5D%60.%20If%20a%20caller%20is%20authorized%20for%20thousands%20of%20tasks%2C%20this%20membership%20test%20dominates%20on%20every%20filtered%20list%20request.%20Converting%20to%20a%20%60set%60%20at%20the%20point%20of%20use%20eliminates%20the%20overhead.%0A%0A%60%60%60suggestion%0A%20%20%20%20elif%20task_id%20is%20not%20None%3A%0A%20%20%20%20%20%20%20%20%23%20Explicit%20task_id%20is%20only%20honored%20if%20the%20caller%20is%20authorized%20for%20it%3B%0A%20%20%20%20%20%20%20%20%23%20otherwise%20the%20result%20set%20is%20empty%20%28IN%20%28%29%29.%0A%20%20%20%20%20%20%20%20authorized_task_id_set%20%3D%20set%28authorized_task_ids%29%0A%20%20%20%20%20%20%20%20effective_task_ids%20%3D%20%5Btask_id%5D%20if%20task_id%20in%20authorized_task_id_set%20else%20%5B%5D%0A%60%60%60%0A%0A%23%23%23%20Issue%202%20of%202%0Aagentex%2Fsrc%2Fapi%2Froutes%2Fagent_task_tracker.py%3A41-50%0A**Double%20DB%20fetch%20of%20tracker%20on%20GET**%0A%0A%60DAuthorizedId%60%20already%20calls%20%60tracker_repository.get%28id%3Dtracker_id%29%60%20inside%20%60_get_parent_task_id%60%20to%20resolve%20%60task_id%60.%20The%20handler%20then%20issues%20a%20second%20identical%20%60get%60%20via%20%60get_agent_task_tracker%60.%20For%20the%20GET%20path%20this%20means%20two%20sequential%20round-trips%20to%20retrieve%20the%20same%20row.%20The%20existing%20%60state%60%20and%20%60message%60%20routes%20share%20this%20pattern%2C%20so%20this%20is%20consistent%20with%20the%20codebase%2C%20but%20it%20may%20be%20worth%20caching%20the%20resolved%20entity%20in%20the%20dependency%20for%20tracker%20routes%20given%20how%20frequently%20the%20cursor-commit%20path%20is%20called.%0A%0A&pr=265&platform=github"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCursorDark.svg?v=3"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCursor.svg?v=3"><img
alt="Fix All in Cursor"
src="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCursor.svg?v=3"
height="20"></picture></a> <a
href="https://app.greptile.com/ide/claude-code?prompt=Fix%20the%20following%202%20code%20review%20issues.%20Work%20through%20them%20one%20at%20a%20time%2C%20proposing%20concise%20fixes.%0A%0A---%0A%0A%23%23%23%20Issue%201%20of%202%0Aagentex%2Fsrc%2Fapi%2Froutes%2Fagent_task_tracker.py%3A73-76%0AWhen%20authz%20is%20enabled%20and%20an%20explicit%20%60%3Ftask_id%3D%60%20is%20provided%2C%20%60task_id%20in%20authorized_task_ids%60%20performs%20a%20linear%20O%28n%29%20scan%20over%20a%20%60list%5Bstr%5D%60.%20If%20a%20caller%20is%20authorized%20for%20thousands%20of%20tasks%2C%20this%20membership%20test%20dominates%20on%20every%20filtered%20list%20request.%20Converting%20to%20a%20%60set%60%20at%20the%20point%20of%20use%20eliminates%20the%20overhead.%0A%0A%60%60%60suggestion%0A%20%20%20%20elif%20task_id%20is%20not%20None%3A%0A%20%20%20%20%20%20%20%20%23%20Explicit%20task_id%20is%20only%20honored%20if%20the%20caller%20is%20authorized%20for%20it%3B%0A%20%20%20%20%20%20%20%20%23%20otherwise%20the%20result%20set%20is%20empty%20%28IN%20%28%29%29.%0A%20%20%20%20%20%20%20%20authorized_task_id_set%20%3D%20set%28authorized_task_ids%29%0A%20%20%20%20%20%20%20%20effective_task_ids%20%3D%20%5Btask_id%5D%20if%20task_id%20in%20authorized_task_id_set%20else%20%5B%5D%0A%60%60%60%0A%0A%23%23%23%20Issue%202%20of%202%0Aagentex%2Fsrc%2Fapi%2Froutes%2Fagent_task_tracker.py%3A41-50%0A**Double%20DB%20fetch%20of%20tracker%20on%20GET**%0A%0A%60DAuthorizedId%60%20already%20calls%20%60tracker_repository.get%28id%3Dtracker_id%29%60%20inside%20%60_get_parent_task_id%60%20to%20resolve%20%60task_id%60.%20The%20handler%20then%20issues%20a%20second%20identical%20%60get%60%20via%20%60get_agent_task_tracker%60.%20For%20the%20GET%20path%20this%20means%20two%20sequential%20round-trips%20to%20retrieve%20the%20same%20row.%20The%20existing%20%60state%60%20and%20%60message%60%20routes%20share%20this%20pattern%2C%20so%20this%20is%20consistent%20with%20the%20codebase%2C%20but%20it%20may%20be%20worth%20caching%20the%20resolved%20entity%20in%20the%20dependency%20for%20tracker%20routes%20given%20how%20frequently%20the%20cursor-commit%20path%20is%20called.%0A%0A&repo=scaleapi%2Fscale-agentex&pr=265&platform=github"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInClaudeDark.svg?v=3"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInClaude.svg?v=3"><img
alt="Fix All in Claude Code"
src="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInClaude.svg?v=3"
height="20"></picture></a> <a
href="https://chatgpt.com/codex/deeplink?prompt=IMPORTANT%3A%20Work%20in%20the%20repository%20%22scaleapi%2Fscale-agentex%22%20on%20the%20existing%20branch%20%22asher.fink%2Fagx1-307-tracker-route-fgac%22.%20Checkout%20that%20branch%20%E2%80%94%20do%20NOT%20create%20a%20new%20branch%20or%20open%20a%20new%20PR.%20Push%20your%20changes%20to%20%22asher.fink%2Fagx1-307-tracker-route-fgac%22.%0A%0AFix%20the%20following%202%20code%20review%20issues.%20Work%20through%20them%20one%20at%20a%20time%2C%20proposing%20concise%20fixes.%0A%0A---%0A%0A%23%23%23%20Issue%201%20of%202%0Aagentex%2Fsrc%2Fapi%2Froutes%2Fagent_task_tracker.py%3A73-76%0AWhen%20authz%20is%20enabled%20and%20an%20explicit%20%60%3Ftask_id%3D%60%20is%20provided%2C%20%60task_id%20in%20authorized_task_ids%60%20performs%20a%20linear%20O%28n%29%20scan%20over%20a%20%60list%5Bstr%5D%60.%20If%20a%20caller%20is%20authorized%20for%20thousands%20of%20tasks%2C%20this%20membership%20test%20dominates%20on%20every%20filtered%20list%20request.%20Converting%20to%20a%20%60set%60%20at%20the%20point%20of%20use%20eliminates%20the%20overhead.%0A%0A%60%60%60suggestion%0A%20%20%20%20elif%20task_id%20is%20not%20None%3A%0A%20%20%20%20%20%20%20%20%23%20Explicit%20task_id%20is%20only%20honored%20if%20the%20caller%20is%20authorized%20for%20it%3B%0A%20%20%20%20%20%20%20%20%23%20otherwise%20the%20result%20set%20is%20empty%20%28IN%20%28%29%29.%0A%20%20%20%20%20%20%20%20authorized_task_id_set%20%3D%20set%28authorized_task_ids%29%0A%20%20%20%20%20%20%20%20effective_task_ids%20%3D%20%5Btask_id%5D%20if%20task_id%20in%20authorized_task_id_set%20else%20%5B%5D%0A%60%60%60%0A%0A%23%23%23%20Issue%202%20of%202%0Aagentex%2Fsrc%2Fapi%2Froutes%2Fagent_task_tracker.py%3A41-50%0A**Double%20DB%20fetch%20of%20tracker%20on%20GET**%0A%0A%60DAuthorizedId%60%20already%20calls%20%60tracker_repository.get%28id%3Dtracker_id%29%60%20inside%20%60_get_parent_task_id%60%20to%20resolve%20%60task_id%60.%20The%20handler%20then%20issues%20a%20second%20identical%20%60get%60%20via%20%60get_agent_task_tracker%60.%20For%20the%20GET%20path%20this%20means%20two%20sequential%20round-trips%20to%20retrieve%20the%20same%20row.%20The%20existing%20%60state%60%20and%20%60message%60%20routes%20share%20this%20pattern%2C%20so%20this%20is%20consistent%20with%20the%20codebase%2C%20but%20it%20may%20be%20worth%20caching%20the%20resolved%20entity%20in%20the%20dependency%20for%20tracker%20routes%20given%20how%20frequently%20the%20cursor-commit%20path%20is%20called.%0A%0A"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCodexDark.svg?v=3"><source
media="(prefers-color-scheme: light)"
srcset="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCodex.svg?v=3"><img
alt="Fix All in Codex"
src="https://greptile-static-assets.s3.amazonaws.com/badges/FixAllInCodex.svg?v=3"
height="20"></picture></a>

<details><summary>Prompt To Fix All With AI</summary>

`````markdown
Fix the following 2 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 2
agentex/src/api/routes/agent_task_tracker.py:73-76
When authz is enabled and an explicit `?task_id=` is provided, `task_id in authorized_task_ids` performs a linear O(n) scan over a `list[str]`. If a caller is authorized for thousands of tasks, this membership test dominates on every filtered list request. Converting to a `set` at the point of use eliminates the overhead.

```suggestion
    elif task_id is not None:
# Explicit task_id is only honored if the caller is authorized for it;
        # otherwise the result set is empty (IN ()).
        authorized_task_id_set = set(authorized_task_ids)
effective_task_ids = [task_id] if task_id in authorized_task_id_set else
[]
```

### Issue 2 of 2
agentex/src/api/routes/agent_task_tracker.py:41-50
**Double DB fetch of tracker on GET**

`DAuthorizedId` already calls `tracker_repository.get(id=tracker_id)` inside `_get_parent_task_id` to resolve `task_id`. The handler then issues a second identical `get` via `get_agent_task_tracker`. For the GET path this means two sequential round-trips to retrieve the same row. The existing `state` and `message` routes share this pattern, so this is consistent with the codebase, but it may be worth caching the resolved entity in the dependency for tracker routes given how frequently the cursor-commit path is called.

`````

</details>

<sub>Reviews (1): Last reviewed commit: ["feat(tracker): FGAC authz on
agent\_task\_..."](5ffc036)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=35398752)</sub>

<!-- /greptile_comment -->

Author: asherfink

agentex/src/api/routes/agent_task_tracker.py

@@ -4,10 +4,19 @@
     AgentTaskTracker,
     UpdateAgentTaskTrackerRequest,
 )
+from src.api.schemas.authorization_types import (
+    AgentexResourceType,
+    AuthorizedOperationType,
+    TaskChildResourceType,
+)
 from src.domain.exceptions import ClientError
 from src.domain.use_cases.agent_task_tracker_use_case import (
     DAgentTaskTrackerUseCase,
 )
+from src.utils.authorization_shortcuts import (
+    DAuthorizedId,
+    DAuthorizedResourceIds,
+)
 from src.utils.logging import make_logger
 
 logger = make_logger(__name__)
@@ -22,21 +31,22 @@
     description="Get agent task tracker by tracker ID",
 )
 async def get_agent_task_tracker(
-    tracker_id: str,
+    tracker_id: DAuthorizedId(
+        TaskChildResourceType.agent_task_tracker,
+        AuthorizedOperationType.read,
+        param_name="tracker_id",
+    ),
     agent_task_tracker_use_case: DAgentTaskTrackerUseCase,
 ) -> AgentTaskTracker:
-    """
-    Get agent task tracker for a specific agent and task.
-    """
     try:
-        state = await agent_task_tracker_use_case.get_agent_task_tracker(
+        tracker = await agent_task_tracker_use_case.get_agent_task_tracker(
             tracker_id=tracker_id
         )
-        return AgentTaskTracker.model_validate(state)
+        return AgentTaskTracker.model_validate(tracker)
     except ClientError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
     except Exception as e:
-        logger.error(f"Error getting processing state: {e}", exc_info=True)
+        logger.error(f"Error getting agent task tracker: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail="Internal server error") from e
 
 
@@ -48,19 +58,28 @@ async def get_agent_task_tracker(
 )
 async def filter_agent_task_tracker(
     agent_task_tracker_use_case: DAgentTaskTrackerUseCase,
+    authorized_task_ids: DAuthorizedResourceIds(AgentexResourceType.task),
     agent_id: str | None = Query(None, description="Agent ID"),
     task_id: str | None = Query(None, description="Task ID"),
     limit: int = Query(50, description="Limit", ge=1),
     page_number: int = Query(1, description="Page number", ge=1),
     order_by: str | None = Query(None, description="Field to order by"),
     order_direction: str = Query("desc", description="Order direction (asc or desc)"),
 ) -> list[AgentTaskTracker]:
-    """
-    Filter agent task tracker by query parameters.
-    """
+    if authorized_task_ids is None:
+        # Authz bypassed: honor the explicit task_id filter if given, else no
+        # task restriction.
+        effective_task_ids = [task_id] if task_id else None
+    elif task_id is not None:
+        # Explicit task_id is only honored if the caller is authorized for it;
+        # otherwise the result set is empty (IN ()).
+        effective_task_ids = [task_id] if task_id in authorized_task_ids else []
+    else:
+        effective_task_ids = authorized_task_ids
+
     agent_task_tracker_entities = await agent_task_tracker_use_case.list(
         agent_id=agent_id,
-        task_id=task_id,
+        task_ids=effective_task_ids,
         limit=limit,
         page_number=page_number,
         order_by=order_by,
@@ -79,23 +98,24 @@ async def filter_agent_task_tracker(
     description="Update agent task tracker by tracker ID",
 )
 async def update_agent_task_tracker(
-    tracker_id: str,
+    tracker_id: DAuthorizedId(
+        TaskChildResourceType.agent_task_tracker,
+        AuthorizedOperationType.execute,
+        param_name="tracker_id",
+    ),
     request: UpdateAgentTaskTrackerRequest,
     agent_task_tracker_use_case: DAgentTaskTrackerUseCase,
 ) -> AgentTaskTracker:
-    """
-    Update agent task tracker
-    """
     try:
-        state = await agent_task_tracker_use_case.update_agent_task_tracker(
+        tracker = await agent_task_tracker_use_case.update_agent_task_tracker(
             tracker_id=tracker_id,
             last_processed_event_id=request.last_processed_event_id,
             status=request.status,
             status_reason=request.status_reason,
         )
-        return AgentTaskTracker.model_validate(state)
+        return AgentTaskTracker.model_validate(tracker)
     except ClientError as e:
         raise HTTPException(status_code=400, detail=str(e)) from e
     except Exception as e:
-        logger.error(f"Error committing cursor: {e}", exc_info=True)
+        logger.error(f"Error updating agent task tracker: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail="Internal server error") from e

agentex/src/api/schemas/authorization_types.py

@@ -25,6 +25,7 @@ class TaskChildResourceType(StrEnum):
 
     state = "state"
     message = "message"
+    agent_task_tracker = "agent_task_tracker"
 
 
 class AgentexResource(BaseModel):

agentex/src/domain/use_cases/agent_task_tracker_use_case.py

@@ -27,18 +27,16 @@ async def list(
         limit: int,
         page_number: int,
         agent_id: str | None = None,
-        task_id: str | None = None,
+        task_ids: list[str] | None = None,
         order_by: str | None = None,
         order_direction: str = "desc",
     ) -> list[AgentTaskTrackerEntity]:
-        """
-        List agent task trackers.
-        """
+        # task_ids=[] produces IN () -> zero rows; None means no task filter.
         filters = {}
         if agent_id:
             filters["agent_id"] = agent_id
-        if task_id:
-            filters["task_id"] = task_id
+        if task_ids is not None:
+            filters["task_id"] = task_ids
 
         return await self._tracker_repository.list(
             filters=filters if filters else None,
@@ -55,25 +53,7 @@ async def update_agent_task_tracker(
         status: str | None = None,
         status_reason: str | None = None,
     ) -> AgentTaskTrackerEntity:
-        """
-        Commit cursor position for an agent-task combination.
-
-        Args:
-            agent_id: The agent ID
-            task_id: The task ID
-            last_processed_event_id: The last processed event ID (None to leave unchanged)
-            status: Processing status
-            status_reason: Optional status reason
-
-        Returns:
-            Updated AgentTaskTrackerEntity object
-
-        Raises:
-            ClientError: If invalid parameters or cursor moves backwards
-        """
-        # Validate inputs
         try:
-            # Commit the cursor
             updated_state = await self._tracker_repository.update_agent_task_tracker(
                 id=tracker_id,
                 status=status,

agentex/src/utils/authorization_shortcuts.py

@@ -9,6 +9,9 @@
     TaskChildResourceType,
 )
 from src.domain.repositories.agent_repository import DAgentRepository
+from src.domain.repositories.agent_task_tracker_repository import (
+    DAgentTaskTrackerRepository,
+)
 from src.domain.repositories.task_message_repository import DTaskMessageRepository
 from src.domain.repositories.task_repository import DTaskRepository
 from src.domain.repositories.task_state_repository import DTaskStateRepository
@@ -22,11 +25,13 @@ async def _get_parent_task_id(
     resource_id: str,
     state_repository: DTaskStateRepository,
     message_repository: DTaskMessageRepository,
+    tracker_repository: DAgentTaskTrackerRepository,
 ) -> str:
     """Get the parent task ID for a task-child resource."""
     registry = {
         TaskChildResourceType.state: state_repository,
         TaskChildResourceType.message: message_repository,
+        TaskChildResourceType.agent_task_tracker: tracker_repository,
     }
 
     repository = registry[resource_type]
@@ -46,6 +51,7 @@ async def _ensure_authorized_id(
         authorization: DAuthorizationService,
         state_repository: DTaskStateRepository,
         message_repository: DTaskMessageRepository,
+        tracker_repository: DAgentTaskTrackerRepository,
         resource_id: str = Path(..., alias=param_name),
     ) -> str:
         # For child resources, check the parent task. Collapse a denied check
@@ -57,6 +63,7 @@ async def _ensure_authorized_id(
                 resource_id,
                 state_repository,
                 message_repository,
+                tracker_repository,
             )
             await check_task_or_collapse_to_404(authorization, task_id, operation)
         elif resource_type == AgentexResourceType.task:
@@ -92,6 +99,7 @@ async def _ensure_authorized_query(
         authorization: DAuthorizationService,
         state_repository: DTaskStateRepository,
         message_repository: DTaskMessageRepository,
+        tracker_repository: DAgentTaskTrackerRepository,
         resource_id: str = Query(..., alias=param_name, description=description),
     ) -> str:
         # For child resources, check the parent task. Collapse a denied check
@@ -103,6 +111,7 @@ async def _ensure_authorized_query(
                 resource_id,
                 state_repository,
                 message_repository,
+                tracker_repository,
             )
             await check_task_or_collapse_to_404(authorization, task_id, operation)
         elif resource_type == AgentexResourceType.task:

agentex/tests/integration/api/agent_task_tracker/test_agent_task_tracker_api.py

@@ -315,13 +315,19 @@ async def test_commit_cursor_tracker_nonexistent_returns_400(
         # Then - Should return 400
         assert response.status_code == 400
 
-    async def test_get_tracker_non_existent_returns_400(self, isolated_client):
-        """Test getting a non-existent tracker returns proper error"""
+    async def test_get_tracker_non_existent_returns_404(self, isolated_client):
+        """Test getting a non-existent tracker returns 404.
+
+        The ``DAuthorizedId`` dependency resolves ``tracker -> task_id`` before
+        the handler runs (on every request, even when authz is bypassed). A
+        missing tracker raises ``ItemDoesNotExist`` during that resolution,
+        which surfaces as a 404 rather than the handler's ClientError 400.
+        """
         # When - Get a non-existent tracker
         response = await isolated_client.get("/tracker/non-existent-tracker-id")
 
-        # Then - Should return 400 (based on the actual API behavior)
-        assert response.status_code == 400
+        # Then - Should return 404 (parent-task resolution raises ItemDoesNotExist)
+        assert response.status_code == 404
 
     @mock.patch(
         "src.api.schemas.agent_task_tracker.AgentTaskTracker.model_validate",