Add run dedupe key and idempotent create flow

docs/hackbot/api.md

@@ -7,20 +7,21 @@ component that knows the agent catalog, and the only writer of run state.
 
 ### Public — `X-API-Key`
 
-| Method | Path                              | Does                                                   |
-| ------ | --------------------------------- | ------------------------------------------------------ |
-| GET    | `/agents`                         | The catalog, each with its input JSON schema           |
-| POST   | `/agents/{agent}/runs`            | Validate inputs, create a run, start an execution      |
-| GET    | `/runs`                           | List runs; filter by `agent`, `status`, `requested_by` |
-| GET    | `/runs/{run_id}`                  | One run: status, inputs, summary, artifacts            |
-| GET    | `/runs/{run_id}/artifacts/{path}` | A short-lived signed GCS download URL                  |
-| GET    | `/runs/{run_id}/actions`          | Recorded actions and their apply state                 |
-| POST   | `/runs/{run_id}/actions/apply`    | Apply all pending actions (idempotent)                 |
-| GET    | `/health`                         | Health check                                           |
+| Method | Path                              | Does                                                                 |
+| ------ | --------------------------------- | -------------------------------------------------------------------- |
+| GET    | `/agents`                         | The catalog, each with its input JSON schema                         |
+| POST   | `/agents/{agent}/runs`            | Validate inputs, create a run, start an execution                    |
+| GET    | `/runs`                           | List runs; filter by `agent`, `status`, `requested_by`, `dedupe_key` |
+| GET    | `/runs/{run_id}`                  | One run: status, inputs, summary, artifacts                          |
+| GET    | `/runs/{run_id}/artifacts/{path}` | A short-lived signed GCS download URL                                |
+| GET    | `/runs/{run_id}/actions`          | Recorded actions and their apply state                               |
+| POST   | `/runs/{run_id}/actions/apply`    | Apply all pending actions (idempotent)                               |
+| GET    | `/health`                         | Health check                                                         |
 
 `POST /agents/{agent}/runs` accepts an `X-On-Behalf-Of` header carrying the requesting
 user's email, stored as `requested_by` — the caller is a trusted service (the UI), so this
-is attribution, not authentication.
+is attribution, not authentication. It also takes a `dedupe_key` query parameter, which
+decides whether the request starts a run at all (see [Deduplication](#deduplication)).
 
 Artifact downloads are restricted to artifacts already listed on the run, which both scopes
 the download to that run's prefix and prevents probing unrelated objects.
@@ -68,15 +69,49 @@ To turn it on: **Interactivity & Shortcuts → Request URL** =
 
 ```
 validate payload against the agent's input schema  ── 422 on mismatch
+insert Run(status=pending) and commit              ── the key is now claimed
+   ...unless the unique index rejects it           ── 200 + the run that won
 mint a V4 signed POST policy scoped to runs/<run_id>/
-insert Run(status=pending)
 trigger a Cloud Run Job execution with env overrides on the `agent` container
 store execution_name
 ```
 
 Env overrides are the run id, the results bucket/prefix/policy, and the inputs mapped from
-the schema. If the trigger fails the run is marked `failed` with the reason and the caller
-gets a 502 — a run row always exists, so a failed dispatch is visible rather than lost.
+the schema. If the policy or the trigger fails the run is marked `failed` with the reason and
+the caller gets a 502 — a run row always exists, so a failed start is visible rather than
+lost.
+
+The insert comes **first**, so a duplicate is turned away before anything is prepared for
+it, and it **commits** before the dispatch, so no transaction is held open across a call to
+Cloud Run (the pool is 5 plus 5 overflow) and the row survives a crash mid-dispatch. Neither
+is what makes deduplication correct, the unique index is. The cost is that a crash in between
+leaves a pending run with no execution, which the stale-run sweep finalizes.
+
+## Deduplication
+
+External triggers fan out: 20 build tasks fail on one push, Phabricator retries a delivery,
+a Slack button gets clicked twice. A caller **gives the work a key** with `?dedupe_key=`,
+scoped to the agent, and the rule is the whole design:
+
+> a key belongs to one run, for good.
+
+A new run answers `201`. A request whose key another run already holds answers `200` with
+that run, which is not an error and needs no handling; `GET /runs?dedupe_key=…` finds it
+again later. The key is coalescing rather than a request fingerprint: requests sharing one
+may carry different inputs, and the first to arrive is the one that runs.
+
+Recurrence lives in the key, since only the caller knows whether the work may happen again:
+
+| Intent                             | Key                                      |
+| ---------------------------------- | ---------------------------------------- |
+| Investigate this push exactly once | `push:<project>:<revision>`              |
+| ...but let tomorrow try again      | `push:<project>:<revision>:<YYYY-MM-DD>` |
+| Handle this delivery exactly once  | `phab-txn:<phid>`, `ni:<flag-id>`        |
+| Always run                         | (omit the parameter)                     |
+
+A run keeps its key whatever becomes of it, a failed dispatch included, so a repeated
+trigger gets the failure rather than a silent retry. A transient error therefore spends that
+key: key the work differently, or run unkeyed.
 
 ## Run states
 
@@ -144,6 +179,14 @@ Two tables, defined in [app/database/models.py](../../services/hackbot-api/app/d
 **`runs`** is the system of record for a run — its inputs, execution name, summary,
 artifacts and terminal state. Listing orders by `created_at desc, run_id desc` rather than
 timestamp alone, so offset paging stays stable when two runs share a timestamp.
+`dedupe_key` carries one index, `uq_runs_dedupe_key` over `(dedupe_key, agent)`, **unique**:
+
+- **Unique**, because a key names one run for good, so the index is the decision about a
+  duplicate trigger rather than a check on one. Runs without a key are unconstrained, since
+  Postgres treats each NULL as distinct.
+- **`dedupe_key` first**, so the same index answers `GET /runs?dedupe_key=…`, which carries
+  no agent. The reverse order would constrain exactly the same thing but only serve lookups
+  that name the agent.
 
 **`run_actions`** holds one row per entry in a run's `summary.json` actions, unique on
 `(run_id, idx)`, carrying that action's apply state. That uniqueness is what makes replays

libs/hackbot-client/hackbot_client/client.py

@@ -26,17 +26,26 @@ async def trigger_run(
         inputs: Mapping[str, Any],
         *,
         on_behalf_of: str | None = None,
+        dedupe_key: str | None = None,
     ) -> RunRef:
-        """Create an agent run and return the API's typed run reference."""
+        """Create an agent run and return the API's typed run reference.
+
+        `dedupe_key` keys the work the run does, and a key belongs to one run
+        for good: repeated triggers carrying it are no-ops, answered with the
+        same run reference.
+        """
         headers = {"X-API-Key": self._api_key}
         if on_behalf_of is not None:
             headers["X-On-Behalf-Of"] = on_behalf_of
 
+        params = {} if dedupe_key is None else {"dedupe_key": dedupe_key}
+
         async with httpx.AsyncClient(timeout=self._timeout_seconds) as client:
             response = await client.post(
                 f"{self._base_url}/agents/{agent_name}/runs",
                 json=dict(inputs),
                 headers=headers,
+                params=params,
             )
 
         response.raise_for_status()

libs/hackbot-client/tests/test_client.py

@@ -33,8 +33,8 @@ async def __aenter__(self):
         async def __aexit__(self, *exc):
             return False
 
-        async def post(self, url, json=None, headers=None):
-            captured.update(url=url, json=json, headers=headers)
+        async def post(self, url, json=None, headers=None, params=None):
+            captured.update(url=url, json=json, headers=headers, params=params)
             response.request = httpx.Request("POST", url)
             return response
 
@@ -66,6 +66,8 @@ async def test_trigger_run_posts_inputs_and_returns_typed_reference(monkeypatch)
             "X-API-Key": "secret",
             "X-On-Behalf-Of": "user@example.com",
         },
+        # No key means no client-assigned id on the create call.
+        "params": {},
     }
 
 

services/hackbot-api/alembic/versions/a7d4e9c21b83_runs_dedupe_key.py

@@ -0,0 +1,40 @@
+"""Name the work a run does, so duplicate triggers can collapse onto it.
+
+Revision ID: a7d4e9c21b83
+Revises: f3c8a1d5b2e7
+Create Date: 2026-09-08 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a7d4e9c21b83"
+down_revision: Union[str, Sequence[str], None] = "f3c8a1d5b2e7"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.add_column("runs", sa.Column("dedupe_key", sa.String(), nullable=True))
+    # Unique, because a key belongs to one run for good: the constraint is what
+    # decides a duplicate trigger, not just a check on one. `dedupe_key` leads
+    # so the index also serves the `?dedupe_key=` listing filter, which carries
+    # no agent. Existing rows all have a NULL key, and NULLs do not collide, so
+    # this is safe to add to a populated table.
+    op.create_index(
+        "uq_runs_dedupe_key",
+        "runs",
+        ["dedupe_key", "agent"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_index("uq_runs_dedupe_key", table_name="runs")
+    op.drop_column("runs", "dedupe_key")

services/hackbot-api/app/database/models.py

@@ -1,7 +1,15 @@
 from datetime import datetime
 from uuid import UUID
 
-from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy import (
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+)
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
 from sqlalchemy.ext.asyncio import AsyncAttrs
@@ -15,12 +23,14 @@ class Base(AsyncAttrs, DeclarativeBase):
 
 class Run(Base):
     __tablename__ = "runs"
+    __table_args__ = (Index("uq_runs_dedupe_key", "dedupe_key", "agent", unique=True),)
 
     run_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
     agent: Mapped[str] = mapped_column(String, nullable=False, index=True)
     status: Mapped[str] = mapped_column(String, nullable=False, index=True)
     inputs: Mapped[dict] = mapped_column(JSONB, nullable=False)
     requested_by: Mapped[str | None] = mapped_column(String, nullable=True, index=True)
+    dedupe_key: Mapped[str | None] = mapped_column(String, nullable=True)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         server_default=func.now(),