feat(schedules): register agent_schedule in authz on create/delete

[harvhan]

Summary

Registers an agent_schedule resource in the authorization service on schedule
create (with the owning agent as its parent edge) and deregisters it on delete,
so schedules participate in fine-grained authorization.

Schedules have no Postgres row — Temporal is the store and the authz selector is
the schedule id ({agent_id}--{schedule_name}). The register/deregister calls
therefore live in ScheduleService, alongside the Temporal write, so the
compensation boundary can be scoped to the Temporal create only.

Behavior

• Create: register the schedule (with parent=agent) before the
  Temporal create. Unconditional and fail-closed — a register failure
  aborts the create. If the Temporal create fails after a successful register,
  a compensating deregister removes the tuple so no orphan is left. The
  post-create read-back (describe) is outside the compensation scope: a describe
  failure must not deregister a schedule that was actually created.
• Delete: deregister after the Temporal delete, best-effort (logged,
  never blocks the delete).
• No resolvable creator (agent-bypass / internal paths with neither a user
  nor service-account identity): registration is skipped with a warning and the
  schedule is still created. Interim behavior until on-behalf-of-user identity
  is threaded.

Per-account routing (which authz backend the call lands on) is owned by the
authorization service; this repo calls register/deregister unconditionally.
Follows the agent_api_key register/deregister pattern (#248), which this PR is
stacked on.

Requires

A companion change in the authorization service mapping the schedule resource
type to its SpiceDB agent_schedule definition; without it the
register/deregister call is rejected before reaching the backend.

Changes

• schedule_service.py: register-before-create + compensate-on-temporal-failure;
  best-effort deregister on delete; _register_schedule_in_auth /
  _deregister_schedule_from_auth helpers; inject the authorization service.
• authorization_types.py: add the schedule resource type + factory.
• tests/integration/services/test_schedule_service_dual_write.py: 7 cases
  (parent-edge contract, fail-closed create, compensating deregister on Temporal
  failure, read-back failure does NOT compensate, best-effort deregister on
  delete, no-creator skip).
• tests/unit/services/test_schedule_service.py: fixture updated for the new
  constructor arg.

Test plan

• pytest tests/integration/services/test_schedule_service_dual_write.py — 7/7
• pytest tests/unit/services/test_schedule_service.py — 27/27
• pytest tests/unit/use_cases/test_schedules_use_case.py — 18/18
• ruff clean

Stacked on

#248 (base branch). Rebase onto the default branch once #248 merges.

🤖 Generated with Claude Code

Greptile Summary

This PR registers agent_schedule resources in the authorization service on schedule create and deregisters them on delete, following the same pattern established for agent_api_key in #248. Because schedules have no Postgres row (Temporal is the store), the auth dual-write lives entirely in ScheduleService alongside the Temporal write.

• schedule_service.py: Injects AuthorizationService, adds _register_schedule_in_auth (fail-closed, called before the Temporal create) and _deregister_schedule_from_auth (best-effort, never raises); compensating deregister on Temporal failure and best-effort deregister on delete are correctly wired. The no-creator skip path is intentionally left as interim behavior.
• authorization_types.py: Adds schedule to AgentexResourceType and factory methods on AgentexResource / AgentexResourceOptionalSelector, mirroring the api_key pattern exactly.
• Tests: 7 integration scenarios cover the complete state machine (fail-closed create, compensating deregister, read-back non-compensation, best-effort delete, no-creator skip); unit test fixture updated for the new constructor arg.

Confidence Score: 4/5

Safe to merge; the core register/compensate/deregister logic is correct and well-tested, with one minor documentation gap.

The implementation faithfully follows the agent_api_key pattern, the fail-closed + compensating-deregister design is correctly scoped (read-back failures intentionally outside the compensation boundary), and all seven stated behavioral guarantees are covered by dedicated tests. The only open item is that the interim-behavior note in the docstring has no linked Linear ticket, which could cause that follow-up to slip through.

schedule_service.py is the only file worth a close look — specifically the _register_schedule_in_auth docstring's untracked interim note and the interaction between the user_id/service_account_id early-exit and the _bypass() logic inside AuthorizationService.

Important Files Changed

Filename	Overview
agentex/src/domain/services/schedule_service.py	Core change: adds authorization service injection, register-before-create (fail-closed), compensating deregister on Temporal failure, and best-effort deregister on delete; logic is sound with one style concern around interim-behavior documentation.
agentex/src/api/schemas/authorization_types.py	Adds schedule to the AgentexResourceType enum and corresponding factory class methods on AgentexResource and AgentexResourceOptionalSelector; follows the existing api_key pattern exactly.
agentex/tests/integration/services/test_schedule_service_dual_write.py	New integration test file; covers all 7 documented scenarios (parent-edge contract, fail-closed create, compensating deregister, read-back non-compensation, best-effort delete deregister, no-creator skip) with clean mocking strategy.
agentex/tests/unit/services/test_schedule_service.py	Existing unit test fixture updated to inject the new authorization_service mock with a resolvable principal; no test logic changed beyond the constructor update.

Sequence Diagram

sequenceDiagram
    participant C as Caller
    participant SS as ScheduleService
    participant AS as AuthorizationService
    participant TA as TemporalAdapter

    C->>SS: create_schedule(agent, request)
    SS->>SS: build_schedule_id()
    SS->>SS: _register_schedule_in_auth()

    alt no creator identity
        SS-->>SS: log warning, return False
    else creator resolvable
        SS->>AS: "register_resource(schedule, parent=agent)"
        alt register fails
            AS-->>SS: Exception
            SS-->>C: raise (fail-closed, Temporal NOT called)
        else register succeeds
            AS-->>SS: "OK (registered=True)"
        end
    end

    SS->>TA: create_schedule(...)
    alt Temporal create fails
        TA-->>SS: Exception
        alt "registered=True"
            SS->>AS: deregister_resource(schedule) [best-effort]
        end
        SS-->>C: raise original exception
    else Temporal create succeeds
        TA-->>SS: OK
        SS->>SS: get_schedule() [read-back, outside compensation scope]
        SS-->>C: ScheduleResponse
    end

    C->>SS: delete_schedule(agent_id, schedule_name)
    SS->>TA: delete_schedule(schedule_id)
    TA-->>SS: OK
    SS->>AS: deregister_resource(schedule) [best-effort]
    AS-->>SS: OK or error (swallowed)
    SS-->>C: None

Loading

Prompt To Fix All With AI

Fix the following 1 code review issue. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 1
agentex/src/domain/services/schedule_service.py:129-132
**Interim note missing a linked ticket**

The docstring describes this skip as "the interim behavior until on-behalf-of-user identity is threaded," but there is no Linear task linked to track that follow-up. Per the team's convention, interim/TODO notes should reference a ticket so the work is searchable and doesn't get lost.

_{Reviews (1): Last reviewed commit: "feat(schedules): register agent_schedule..." | Re-trigger Greptile}

Context used:

• Rule used - Include ticket numbers in TODO comments to make cl... (source)

Learned From
scaleapi/scaleapi#126926