feat(schedules): agent run schedules (v1) (#335)

## Summary

Adds per-agent **run schedules**: recurring schedules that fire a task
and deliver a configured initial input on a cron/interval cadence.
**Replaces the prior `schedules` implementation** (a bare-workflow
scheduler) on the same API path.

Each schedule is a Postgres row (the source of truth) plus a Temporal
Schedule that acts purely as the recurring clock (it carries only the
row id). On each fire, a thin, deterministic workflow runs a single
activity that creates a task and delivers the initial input via the same
path as a manual run — `message/send` for sync agents, `event/send` for
agentic agents — attributed to the schedule's stored creator principal.

## Feature flag

The API is gated behind `ENABLE_AGENT_RUN_SCHEDULES` (matches the
existing `ENABLE_HEALTH_CHECK_WORKFLOW` pattern), **disabled by default
in every environment** — when off, the routes are not registered at all.
Enable per-environment when ready to test (e.g. locally
`ENABLE_AGENT_RUN_SCHEDULES=true ./dev.sh`). The OpenAPI spec/SDK
document the endpoints regardless of the runtime default.

## Removed / breaking changes

This PR **deletes the previous `schedules` feature** (routes, schemas,
service, use case, and its tests). The old endpoint scheduled a raw
Temporal workflow and stored nothing in Postgres; the new one schedules
an agent *run* and is Postgres-backed. Because the API path
`/agents/{agent_id}/schedules` is reused with new semantics, this is
**breaking for existing consumers of the old endpoint**:

- `POST /agents/{agent_id}/schedules` — request/response schema changed
(schedules an agent run, not a bare workflow)
- `POST …/{name}/unpause` → **renamed** to `…/{name}/resume`
- Path param `{schedule_name}` → `{name}` (cosmetic)
- Adds the new `agent_run_schedules` table (the old scheduler was
Temporal-only)

(`…/{name}/trigger` is preserved — see below.)

## Endpoints

`/agents/{agent_id}/schedules`:
- `POST` — create
- `GET` — list (served from Postgres; no per-row Temporal call)
- `GET /{name}` — get (includes live Temporal state: next/last fire,
action count)
- `PATCH /{name}` — partial update (cadence, window, input, params,
paused; cron/interval stay mutually exclusive)
- `POST /{name}/pause` · `POST /{name}/resume`
- `POST /{name}/trigger` — immediate out-of-band run
- `DELETE /{name}`

## Implementation notes

- `ScheduledAgentRunWorkflow` (thin/deterministic) +
`launch_scheduled_agent_run` activity (all side effects live in the
activity).
- Deterministic per-fire task name makes `task/create` idempotent on
activity retry; a delivered marker guards against duplicate input
delivery.
- Fire-time authorization re-check under the stored creator principal —
a revoked creator stops firing cleanly.
- Scheduled tasks get a `task_metadata.display_name` (`Scheduled
Message: <name> · <fire time>`), stamped with the nominal fire time
(stable across retries) so they render with a label instead of "Unnamed
task".
- `delete`/`pause`/`resume`/`update` tolerate a missing Temporal
schedule so a partial failure can't strand an un-cleanable row.
- New `agent_run_schedules` table migration (new-table create;
schema-only, non-blocking).

## Testing

- 30 unit tests (service, activity, use case, env flag) pass, covering
create/list/get/update/pause/resume/trigger/delete, idempotency,
validation, and flag parsing.
- Verified end-to-end locally (flag on): both delivery paths (sync
`message/send` and agentic `event/send`), plus
pause/resume/update/trigger/delete reflected consistently in Postgres
and Temporal.
- Verified on a dev cluster (branch image, flag on): create → Temporal
schedule → worker fires on schedule → `message/send` delivered, with the
row persisted and the creator principal captured from real auth.

## Deployment dependency (authz provider)

Dev verification surfaced this: on a cluster using the SGP authz
provider (`AUTH_PROVIDER=sgp`), **the provider must learn the new
`schedule` resource type before this is usable there.** Today its
`/v1/authz/check` returns **422** for a `schedule` resource, so:

- **Create works** (it gates on `agent.update`, and `register` of the
`schedule` resource is tolerated).
- **Every op gated on the schedule resource — `GET /{name}`, `pause`,
`resume`, `trigger`, `PATCH`, `DELETE` — returns 422** until the
provider handles
`check`/`grant`/`revoke`/`register`/`deregister`/`search` for `schedule`
(mirroring `agent`/`task`/`api_key`).

This is provider-side work (the `schedule` type is already part of the
documented auth-provider contract); it should land alongside this
feature's rollout. Environments with authz disabled or a permissive
provider are unaffected.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- greptile_comment -->

<h3>Greptile Summary</h3>

This PR replaces the previous bare-workflow scheduler with a fully
Postgres-backed agent run scheduling system. Each schedule is a Postgres
row (source of truth) paired with a Temporal Schedule acting as the
recurring clock; on each fire, a thin deterministic workflow runs a
single activity that creates a fresh task and delivers the configured
initial input under the stored creator principal.

- Adds `agent_run_schedules` table, ORM, repository, service, use case,
and full CRUD API
(`POST`/`GET`/`PATCH`/`DELETE`/`pause`/`resume`/`trigger`) gated behind
`ENABLE_AGENT_RUN_SCHEDULES`.
- Implements idempotent fire-and-deliver via deterministic task names
and a `scheduled_input_delivered` marker; all previously flagged issues
(partial-delete strand, list-path sequential Temporal RPCs, fire-time
stamp drift, manual-trigger paused bypass, auth-filter/limit ordering,
mutually-exclusive cadence validation) have been addressed in the
current code.
- The Temporal worker unconditionally registers the new workflow and
activities so that in-flight scheduled executions continue to be handled
even when the API flag is disabled.

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

Safe to merge; the core fire-and-deliver path is well-guarded, all
previously flagged issues are addressed, and the feature is off by
default.

All six issues called out in prior review threads have been resolved in
the current code. The new activity is idempotent (deterministic task
name + delivered marker), the rollback logic on failed creates is
correct, and the migration is non-blocking. Remaining observations are
non-blocking design trade-offs with documented intent.

agentex/src/temporal/scheduled_agent_run_factory.py — per-fire engine
allocation in `build_acp_use_case_for_principal` is worth revisiting if
fire rates grow. agentex/src/adapters/temporal/adapter_temporal.py —
string-based not-found detection is fragile to upstream message changes.
</details>

<details><summary><h3>Important Files Changed</h3></summary>

| Filename | Overview |
|----------|----------|
| agentex/src/domain/services/agent_run_schedule_service.py | Core
service managing the dual Postgres+Temporal state. All previously
flagged issues addressed. `trigger_schedule` uses `start_workflow` (not
`trigger_now`), so manual fires won't update Temporal Schedule live
stats. |
| agentex/src/temporal/activities/scheduled_agent_run_activities.py |
Well-structured with idempotent task creation, delivered marker guard,
fire-time authz re-check, and correct manual-trigger paused bypass. |
| agentex/src/adapters/temporal/adapter_temporal.py | Adds
`update_schedule`; fixes `start_workflow` to use `args=` keyword.
Not-found detection via string matching is fragile. |
| agentex/src/api/routes/agent_run_schedules.py | Clean route layer;
`_check_schedule_or_collapse_to_404` imported with private prefix is a
minor style concern. |
| agentex/src/temporal/scheduled_agent_run_factory.py | Per-fire factory
creates a new DB engine on each activity invocation; worth revisiting
under high fire rates. |
| agentex/src/api/schemas/agent_run_schedules.py | Schema validators
correctly enforce single cadence at create and prevent both-non-null
cadences at update. |
|
agentex/database/migrations/alembic/versions/2026_06_22_1200_add_agent_run_schedules_3b1c9d2e4f6a.py
| Non-blocking schema-only migration; unique index enforces permanently
reserved names consistently with application logic. |
| agentex/src/utils/schedule_metrics.py | Bounded categorical tags only
— no high-cardinality entity IDs. |
| agentex/src/temporal/workflows/scheduled_agent_run_workflow.py | Thin
deterministic wrapper; correctly passes trigger_type to the activity. |

</details>

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

<a href="#gh-light-mode-only">

```mermaid
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant API as API Route
    participant SVC as AgentRunScheduleService
    participant PG as Postgres
    participant TP as Temporal Schedule
    participant WF as ScheduledAgentRunWorkflow
    participant ACT as launch_scheduled_agent_run
    participant ACP as AgentsACPUseCase

    Note over API,TP: Create schedule
    API->>SVC: create_schedule(agent, request, creator_principal)
    SVC->>PG: create(AgentRunScheduleEntity)
    SVC->>TP: "create_schedule(temporal_id, args=[row.id])"
    TP-->>SVC: ScheduleHandle

    Note over TP,ACP: Each cron/interval fire
    TP->>WF: start ScheduledAgentRunWorkflow(schedule_id)
    WF->>ACT: launch_scheduled_agent_run(schedule_id, fire_id, trigger_type)
    ACT->>PG: get schedule row
    ACT->>ACT: fire-time authz re-check (creator_principal)
    ACT->>ACP: task/create (deterministic name)
    ACT->>ACP: message/send or event/send
    ACT->>PG: mark scheduled_input_delivered

    Note over API,TP: Manual trigger
    API->>SVC: trigger_schedule(agent_id, name)
    SVC->>TP: "start_workflow(ScheduledAgentRunWorkflow, args=[row.id, 'manual'])"
    TP-->>API: schedule response (async, no task info)
```

</a>
<a href="#gh-dark-mode-only">

```mermaid
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant API as API Route
    participant SVC as AgentRunScheduleService
    participant PG as Postgres
    participant TP as Temporal Schedule
    participant WF as ScheduledAgentRunWorkflow
    participant ACT as launch_scheduled_agent_run
    participant ACP as AgentsACPUseCase

    Note over API,TP: Create schedule
    API->>SVC: create_schedule(agent, request, creator_principal)
    SVC->>PG: create(AgentRunScheduleEntity)
    SVC->>TP: "create_schedule(temporal_id, args=[row.id])"
    TP-->>SVC: ScheduleHandle

    Note over TP,ACP: Each cron/interval fire
    TP->>WF: start ScheduledAgentRunWorkflow(schedule_id)
    WF->>ACT: launch_scheduled_agent_run(schedule_id, fire_id, trigger_type)
    ACT->>PG: get schedule row
    ACT->>ACT: fire-time authz re-check (creator_principal)
    ACT->>ACP: task/create (deterministic name)
    ACT->>ACP: message/send or event/send
    ACT->>PG: mark scheduled_input_delivered

    Note over API,TP: Manual trigger
    API->>SVC: trigger_schedule(agent_id, name)
    SVC->>TP: "start_workflow(ScheduledAgentRunWorkflow, args=[row.id, 'manual'])"
    TP-->>API: schedule response (async, no task info)
```

</a>
</details>

<sub>Reviews (10): Last reviewed commit: ["fix(temporal): pass workflow
args via
ar..."](e6ea64b)
| [Re-trigger
Greptile](https://app.greptile.com/api/retrigger?id=39353769)</sub>

<!-- /greptile_comment -->

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 3 people

agentex/database/migrations/alembic/versions/2026_06_22_1200_add_agent_run_schedules_3b1c9d2e4f6a.py

@@ -0,0 +1,83 @@
+"""add agent_run_schedules
+
+Revision ID: 3b1c9d2e4f6a
+Revises: c7a1b2d3e4f5
+Create Date: 2026-06-22 12:00:00.000000
+
+Creates the agent_run_schedules table backing the scheduled-agent-runs feature.
+Schema-only and idempotent: the table and its indexes are created
+with IF NOT EXISTS-style guards (Alembic create_table on a fresh table), and the
+indexes target the just-created table so they are non-blocking by construction.
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "3b1c9d2e4f6a"
+down_revision: str | None = "c7a1b2d3e4f5"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "agent_run_schedules",
+        sa.Column("id", sa.String(), nullable=False),
+        sa.Column("agent_id", sa.String(length=64), nullable=False),
+        sa.Column("name", sa.String(length=256), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("cron_expression", sa.String(), nullable=True),
+        sa.Column("interval_seconds", sa.Integer(), nullable=True),
+        sa.Column("timezone", sa.String(), server_default="UTC", nullable=False),
+        sa.Column("start_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("end_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("paused", sa.Boolean(), server_default="false", nullable=False),
+        sa.Column("creator_principal", sa.JSON(), nullable=False),
+        sa.Column("task_params", sa.JSON(), nullable=True),
+        sa.Column("task_metadata", sa.JSON(), nullable=True),
+        sa.Column("initial_input", sa.JSON(), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=True,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=True,
+        ),
+        # Soft-delete marker: NULL = active, set = tombstoned for audit. Deleted
+        # rows keep occupying their (agent_id, name) so names are not reusable.
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        # Monotonic record version reserved for future optimistic concurrency /
+        # change history. Added now (brand-new table) to avoid a later backfill;
+        # not enforced yet.
+        sa.Column("version", sa.Integer(), nullable=False, server_default="1"),
+        sa.ForeignKeyConstraint(["agent_id"], ["agents.id"]),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # Indexes target the table created in this same migration, so they hold no
+    # write-blocking lock against live traffic (the table has no rows yet).
+    op.create_index(
+        "uq_agent_run_schedules_agent_name",
+        "agent_run_schedules",
+        ["agent_id", "name"],
+        unique=True,
+    )
+    op.create_index(
+        "idx_agent_run_schedules_agent",
+        "agent_run_schedules",
+        ["agent_id"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("idx_agent_run_schedules_agent", table_name="agent_run_schedules")
+    op.drop_index("uq_agent_run_schedules_agent_name", table_name="agent_run_schedules")
+    op.drop_table("agent_run_schedules")

agentex/docker-compose.yml

@@ -166,6 +166,8 @@ services:
       - MONGODB_DATABASE_NAME=agentex
       - WATCHFILES_FORCE_POLLING=true
       - ENABLE_HEALTH_CHECK_WORKFLOW=true
+      # Disabled by default; enable when testing, e.g. `ENABLE_AGENT_RUN_SCHEDULES=true ./dev.sh`.
+      - ENABLE_AGENT_RUN_SCHEDULES=${ENABLE_AGENT_RUN_SCHEDULES:-false}
       - AGENTEX_SERVER_TASK_QUEUE=agentex-server
       - ALLOWED_ORIGINS=http://localhost:3000
       - OTEL_EXPORTER_OTLP_ENDPOINT=http://agentex-otel-collector:4317