feat(push): idempotency keys for push, and the monitor drops a re-published push (TASK-3585) (#2071)

POST .../items/{ref}/push accepts op_key (body) or Idempotency-Key,
claimed after every refusal. Push writes no row, so its answer cannot
commit with it; per the lead's ruling the key is completed AFTER the
publish (store.CompleteIdempotencyKey, its own transaction):
- a clean publish stores the response; a same-key retry replays it and
  publishes nothing;
- an unconfirmed publish stores 409 push_outcome_unknown, so a retry
  says "check your agent session" instead of delivering again;
- a closed bus published nothing and releases the key.

A crash between the publish and the completion lets the lease expire and
a retry take the claim over and publish again. Every publish therefore
carries a push_id derived from (workspace, user, key), the same across a
takeover, on watchevents.Notification and the stream payload, and
`pad watch --stream --for-session` drops a push_id it printed in the last
~256 events / 10 minutes, across reconnects. A monitor that predates
push_id prints every publish (at-least-once, as before).

CLI: `pad push --op-key` (fresh per invocation by default) and one keyed
retry after a 5xx, only against a server advertising idempotency_keys.

Claude-Session: https://claude.ai/code/session_01Q9WssCjFuHjSfLABoW4tmS

cmd/pad/cmd_push.go

@@ -3,6 +3,7 @@ package main
 import (
 	"errors"
 	"fmt"
+	"github.com/google/uuid"
 	"os"
 	"strings"
 
@@ -30,7 +31,7 @@ const maxPushMessageLenForHelp = 4096
 // notification-etiquette section for the receiving-agent behavior
 // contract.
 func pushCmd() *cobra.Command {
-	var message string
+	var message, pushOpKey string
 
 	cmd := &cobra.Command{
 		Use:     "push <ref>",
@@ -75,7 +76,13 @@ func pushCmd() *cobra.Command {
 			client, _ := getClient()
 			ws := getWorkspace()
 
-			result, err := client.PushItem(ws, args[0], trimmed)
+			// TASK-3585: one idempotency key per invocation, so the
+			// client's single retry, or a re-run with --op-key, does not
+			// deliver the instruction twice.
+			if pushOpKey == "" {
+				pushOpKey = uuid.NewString()
+			}
+			result, err := client.PushItemKeyed(ws, args[0], trimmed, pushOpKey)
 			if err != nil {
 				// AMBIGUOUS FAILURES GET SAID OUT LOUD (codex round 9 on
 				// BUG-2699). The handler publishes BEFORE it writes its
@@ -83,9 +90,10 @@ func pushCmd() *cobra.Command {
 				// nowhere: a response lost in transit, a truncated body, a
 				// gateway envelope, or the server's own push_unconfirmed all
 				// reach here identically to a clean refusal. A user who reads
-				// "error" and re-runs the command delivers the push twice —
-				// there is no idempotency key, and the receiving agent acts
-				// on each copy.
+				// "error" and re-runs the command delivers the push twice:
+				// the re-run gets a NEW key (TASK-3585 keys one invocation,
+				// and the client already made its one keyed retry), and the
+				// receiving agent acts on each copy.
 				//
 				// The web dialog has drawn this distinction since TASK-2588;
 				// the CLI had not, so the same outcome produced a safe
@@ -106,6 +114,7 @@ func pushCmd() *cobra.Command {
 		},
 	}
 
+	cmd.Flags().StringVar(&pushOpKey, "op-key", "", "idempotency key: re-running the SAME push with the same key answers the first push instead of delivering it again (default: a new key per invocation)")
 	cmd.Flags().StringVarP(&message, "message", "m", "",
 		fmt.Sprintf("instruction text to push (required, max %d characters after whitespace collapse)", maxPushMessageLenForHelp))
 	return cmd

cmd/pad/cmd_watch.go

@@ -11,6 +11,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"strings"
+	"sync"
 	"syscall"
 	"text/tabwriter"
 	"time"
@@ -67,8 +68,56 @@ type watchStreamPayload struct {
 	Kind      string `json:"kind"`
 	Actor     string `json:"actor"`
 	Summary   string `json:"summary"`
+	// PushID is set on a push (TASK-3585). A re-publish of the same push
+	// (a retry after a server crash) carries the same id; seenPushes drops it.
+	PushID string `json:"push_id,omitempty"`
 }
 
+// recentSet remembers ids for a bounded time and count, so the monitor can
+// drop a push it has already printed (TASK-3585). A monitor that predates
+// push_id prints every publish: at-least-once, as before.
+type recentSet struct {
+	mu    sync.Mutex
+	max   int
+	ttl   time.Duration
+	now   func() time.Time
+	order []string
+	at    map[string]time.Time
+}
+
+func newRecentSet(max int, ttl time.Duration) *recentSet {
+	return &recentSet{max: max, ttl: ttl, now: time.Now, at: map[string]time.Time{}}
+}
+
+// seen reports whether id was recorded within the window, and records it
+// when it was not.
+func (s *recentSet) seen(id string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	now := s.now()
+	if t, ok := s.at[id]; ok && now.Sub(t) <= s.ttl {
+		return true
+	}
+	// Evict from the oldest end: anything past the ttl, and enough to make
+	// room for this one under max. An expired copy of id goes here too.
+	for len(s.order) > 0 {
+		oldest := s.order[0]
+		if len(s.order) < s.max && now.Sub(s.at[oldest]) <= s.ttl && oldest != id {
+			break
+		}
+		delete(s.at, oldest)
+		s.order = s.order[1:]
+	}
+	s.at[id] = now
+	s.order = append(s.order, id)
+	return false
+}
+
+// seenPushes is the monitor's memory of printed pushes: about the last 256,
+// for ten minutes. One monitor runs per process, so one set serves every
+// reconnect.
+var seenPushes = newRecentSet(256, 10*time.Minute)
+
 // formatMonitorLine renders one notification as the exact stdout-line
 // contract `pad watch --stream --for-session` promises (DOC-2479):
 // "PAD demo/TASK-214 → done (Dave): fix verified" — one line, facts
@@ -622,7 +671,8 @@ func streamWatchEventsTo(w io.Writer, resp *http.Response, lastEventID string) s
 				}
 				if data != "" {
 					var payload watchStreamPayload
-					if err := json.Unmarshal([]byte(data), &payload); err == nil {
+					if err := json.Unmarshal([]byte(data), &payload); err == nil &&
+						(payload.PushID == "" || !seenPushes.seen(payload.PushID)) {
 						fmt.Fprintln(w, formatMonitorLine(payload))
 						resyncAnnounced = false
 					}

cmd/pad/task3585_monitor_dedupe_test.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"bytes"
+	"strings"
+	"testing"
+	"time"
+)
+
+// TASK-3585: the monitor prints a push once, however many times it is
+// published under the same push_id (a re-publish after a claim takeover).
+func TestTASK3585_MonitorDropsARepublishedPush(t *testing.T) {
+	prev := seenPushes
+	seenPushes = newRecentSet(256, 10*time.Minute)
+	t.Cleanup(func() { seenPushes = prev })
+	armSessionForTest(t)
+
+	ev := func(id, pushID string) string {
+		return "id: " + id + "\nevent: notification\ndata: {\"id\":" + id +
+			",\"workspace\":\"w\",\"item_ref\":\"TASK-1\",\"kind\":\"push\",\"summary\":\"triage this\",\"push_id\":\"" + pushID + "\"}\n\n"
+	}
+	var buf bytes.Buffer
+	streamWatchEventsTo(&buf, fakeSSEResponse(ev("1", "push_a")+ev("2", "push_a")+ev("3", "push_b")), "")
+	if n := strings.Count(buf.String(), "triage this"); n != 2 {
+		t.Fatalf("want push_a once and push_b once (2 lines), got %d:\n%s", n, buf.String())
+	}
+	// The set outlives one connection: a reconnect replaying push_a is dropped too.
+	buf.Reset()
+	streamWatchEventsTo(&buf, fakeSSEResponse(ev("4", "push_a")), "3")
+	if buf.Len() != 0 {
+		t.Fatalf("a re-publish on a new connection was printed: %q", buf.String())
+	}
+}
+
+func TestTASK3585_RecentSetBounds(t *testing.T) {
+	now := time.Unix(0, 0)
+	s := newRecentSet(2, time.Minute)
+	s.now = func() time.Time { return now }
+	if s.seen("a") || s.seen("b") || !s.seen("a") {
+		t.Fatal("a and b are new; a again is a repeat")
+	}
+	s.seen("c") // over max: evicts the oldest (a)
+	if s.seen("a") {
+		t.Fatal("a should have been evicted by count")
+	}
+	now = now.Add(2 * time.Minute)
+	if s.seen("c") {
+		t.Fatal("c should have expired")
+	}
+}

docs/api.md

@@ -72,7 +72,8 @@ REST API at `/api/v1/`. Key endpoints:
 - `GET/POST /workspaces/{ws}/items/{slug}/links` — item relationships (blocks/blocked-by, parent/child)
 - `GET /search?q=query&workspace=slug` — full-text search. A `collection` filter resolves like a collection in the item routes (BUG-2659): exact slug first, then the singular/alias fallback, and an archived collection still claims its name. It resolves once per workspace in scope, so `collection=task` means `task` where one exists and `tasks` where it does not. `GET /server/capabilities` advertises it as `search_collection_resolution`; the CLI sends the name as typed only to a server that does, and expands the shorthand itself otherwise. `include_archived=true` (TASK-2864) adds soft-deleted items on every path (full text, ref and number lookups, total, facets), each carrying `deleted_at`. The permission filter is unchanged, and an individual item grant names live items only, so a guest sees an archived item only through a collection grant. The web search box's `is:archived` token sends it; not on the CLI or MCP
 - `GET /api/v1/events?workspace=slug` — SSE real-time events (workspace-scoped)
-- `GET /api/v1/events/stream` — SSE watch/push notifications (USER-scoped, spans every workspace the caller belongs to; backs `pad watch --stream`). With `?access=true` (TASK-3272) the stream also carries `workspace_access_changed` addressed to the caller — `{workspace_id, change}` with change `gained` / `lost` / `deleted` / `restored` / `purged` — published after every door that changes whether an EXISTING account reaches a workspace (signup-time doors publish nothing: the new account holds no stream yet). Opt-in because an installed `pad watch` prints any kind it does not know as an item line into an agent session. It is a refetch hint: correctness lives on the read-filtered `GET /me/workspace-tabs`, and a duplicate or missed event under concurrent grant writes is benign. Such a stream counts against `PAD_SSE_MAX_PER_USER` like any other
+- `GET /api/v1/events/stream` — SSE watch/push notifications (USER-scoped, spans every workspace the caller belongs to; backs `pad watch --stream`). With `?access=true` (TASK-3272) the stream also carries `workspace_access_changed` addressed to the caller — `{workspace_id, change}` with change `gained` / `lost` / `deleted` / `restored` / `purged` — published after every door that changes whether an EXISTING account reaches a workspace (signup-time doors publish nothing: the new account holds no stream yet). Opt-in because an installed `pad watch` prints any kind it does not know as an item line into an agent session. It is a refetch hint: correctness lives on the read-filtered `GET /me/workspace-tabs`, and a duplicate or missed event under concurrent grant writes is benign. Such a stream counts against `PAD_SSE_MAX_PER_USER` like any other. A push event carries `push_id` (TASK-3585, omitted on every other kind): the same id on every publish of one keyed push, so a consumer drops one it has seen (`pad watch --stream --for-session` keeps ~256 ids for 10 minutes); a monitor that predates it is at-least-once
+- `POST /workspaces/{ws}/items/{ref}/push` — idempotency key (TASK-3585): `op_key` in the body or `Idempotency-Key`, claimed after every refusal. Push writes no row, so the key is completed AFTER the publish, outside any transaction: a clean publish stores the response and a same-key retry replays it without publishing; an unconfirmed publish (`push_unconfirmed`, 502) stores 409 `push_outcome_unknown` for a retry, which publishes nothing; a closed bus (503) releases the key. A crash between publish and completion lets the 60s lease expire and a retry publish again, under the same `push_id`, which is what a monitor dedupes on. The CLI keys every push and retries once after a 5xx, only against a server advertising `idempotency_keys`
 
 **Every long-lived connection ends when the CREDENTIAL that opened it stops being valid** — both SSE endpoints and the `/api/v1/collab/{itemID}` WebSocket, checked on each connection's existing revalidation tick (BUG-3007). The invariant is deliberately wider than "the session was destroyed": a revoked PAT ends a stream too, because a revoked PAT still streaming is the same defect for a CLI or MCP caller that a destroyed session is for a browser. Before this, all three kept running indefinitely — measured at 150-180s past a logout, and 64-120s past a revocation, in both cases with `/auth/me` on that credential answering `401`. The pre-existing per-tick checks are about the PRINCIPAL (is this user still a member, is the item still visible) and none of them changes when a credential dies, which is why the two are separate questions rather than one. A request carrying NO credential AND no resolved principal — the fresh-install window, the legacy no-auth path — is unaffected: there is nothing to invalidate. A request that carries a resolved USER but no record of how the credential was established is CLOSED, not exempted: the middleware stamps that on every accept point, so its absence means an accept point recorded who without recording what, and a connection nobody can re-check is the same defect under a new name. (Both MCP accept points are in that state by design today — the PAT one records `api_token` and is re-checked; the OAuth one has no liveness door for an opaque token yet, so it fails closed. Neither is reachable: no MCP route is long-lived.) A store ERROR is not a revocation and keeps the connection until the next tick, so a database blip does not disconnect the fleet.
 

internal/cli/client.go

@@ -641,9 +641,27 @@ type PushResult struct {
 // --stream --for-session` consumes. Transient, fire-and-forget — see
 // server.handlePushToItem's doc comment for the no-durability rationale.
 func (c *Client) PushItem(wsSlug, itemSlug, message string) (*PushResult, error) {
+	return c.PushItemKeyed(wsSlug, itemSlug, message, "")
+}
+
+// PushItemKeyed is PushItem with an idempotency key (TASK-3585). With a key
+// it retries once with that key after a transport error or a 5xx, and only
+// against a server that advertises idempotency_keys: there the retry
+// answers the first push (or 409 push_outcome_unknown when that push could
+// not be confirmed) instead of delivering it again.
+func (c *Client) PushItemKeyed(wsSlug, itemSlug, message, opKey string) (*PushResult, error) {
 	body := map[string]string{"message": message}
+	if opKey != "" {
+		body["op_key"] = opKey
+	}
+	path := "/workspaces/" + wsSlug + "/items/" + itemSlug + "/push"
 	var result PushResult
-	return &result, c.post("/workspaces/"+wsSlug+"/items/"+itemSlug+"/push", body, &result)
+	err := c.post(path, body, &result)
+	if err != nil && opKey != "" && retryableWithKey(err) && c.serverHonorsKeys() {
+		result = PushResult{}
+		err = c.post(path, body, &result)
+	}
+	return &result, err
 }
 
 // ListWatches returns every watch the current user holds, across all

internal/cli/task3585_push_retry_test.go

@@ -0,0 +1,69 @@
+package cli
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+)
+
+// TASK-3585: a keyed push retries once with the same key after a 5xx, and
+// only against a server that advertises idempotency_keys.
+func pushKeyServer(t *testing.T, honorsKeys bool, statuses ...int) (*Client, func() []string) {
+	t.Helper()
+	var mu sync.Mutex
+	var keys []string
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/capabilities") {
+			_ = json.NewEncoder(w).Encode(map[string]any{"idempotency_keys": honorsKeys})
+			return
+		}
+		raw, _ := io.ReadAll(r.Body)
+		var b map[string]any
+		_ = json.Unmarshal(raw, &b)
+		k, _ := b["op_key"].(string)
+		mu.Lock()
+		n := len(keys)
+		keys = append(keys, k)
+		mu.Unlock()
+		status := http.StatusOK
+		if n < len(statuses) {
+			status = statuses[n]
+		}
+		w.WriteHeader(status)
+		if status >= 400 {
+			_, _ = w.Write([]byte(`{"error":{"code":"push_unconfirmed","message":"m"}}`))
+			return
+		}
+		_, _ = w.Write([]byte(`{"ref":"TASK-1","workspace":"w","pushed":true,"message":"m"}`))
+	}))
+	t.Cleanup(ts.Close)
+	t.Setenv("HOME", t.TempDir())
+	return NewClientFromURL(ts.URL), func() []string { mu.Lock(); defer mu.Unlock(); return append([]string(nil), keys...) }
+}
+
+func TestTASK3585_PushRetriesOnceWithTheSameKey(t *testing.T) {
+	c, keys := pushKeyServer(t, true, http.StatusBadGateway)
+	if _, err := c.PushItemKeyed("w", "TASK-1", "m", "k-1"); err != nil {
+		t.Fatalf("a 502 then a 200 must succeed: %v", err)
+	}
+	if got := keys(); len(got) != 2 || got[0] != "k-1" || got[1] != "k-1" {
+		t.Fatalf("want two attempts with one key; got %v", got)
+	}
+}
+
+func TestTASK3585_PushNoRetryWithoutTheCapabilityOrAKey(t *testing.T) {
+	c, keys := pushKeyServer(t, false, http.StatusBadGateway)
+	_, _ = c.PushItemKeyed("w", "TASK-1", "m", "k-2")
+	if got := keys(); len(got) != 1 {
+		t.Fatalf("retried a push against a server that ignores keys: %d attempts", len(got))
+	}
+	c2, keys2 := pushKeyServer(t, true, http.StatusBadGateway)
+	_, _ = c2.PushItem("w", "TASK-1", "m")
+	if got := keys2(); len(got) != 1 {
+		t.Fatalf("retried a keyless push: %d attempts", len(got))
+	}
+}

internal/server/handlers_push.go

@@ -30,6 +30,9 @@ type pushRequest struct {
 	// silently dropping a field this response documents at length. It
 	// still never SENDS target_session_id.
 	TargetSessionID string `json:"target_session_id,omitempty"`
+	// OpKey is the push's idempotency key in the body (TASK-3585). Not part
+	// of the request hash.
+	OpKey string `json:"op_key,omitempty"`
 }
 
 // pushResponse is the body of a successful push (dispatcher review round
@@ -333,6 +336,31 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) {
 			"item_ref", item.Ref, "error", presenceErr)
 		deliveredSessionsUnknown = true
 	}
+
+	// TASK-3585: an idempotency key, claimed after every refusal above and
+	// before the publish. Push writes no row, so its answer cannot commit
+	// with it: the key is completed AFTER the publish, outside any
+	// transaction. A crash in between lets the lease expire and a retry
+	// publish again, which is why every publish carries a push_id derived
+	// from the key: a monitor that has printed it drops the repeat.
+	key, kerr := idempotencyKey(r, input.OpKey)
+	if kerr != nil {
+		writeError(w, http.StatusBadRequest, "bad_request", kerr.Error())
+		return
+	}
+	hold, ok := s.claimKey(w, r, workspaceID, idempotencyRoutePush, item.ID,
+		pushRequest{Message: message, TargetSessionID: targetSessionID}, key,
+		func(status int, stored string) {
+			w.Header().Set("Content-Type", "application/json")
+			w.WriteHeader(status)
+			_, _ = w.Write([]byte(stored))
+		})
+	if !ok {
+		return
+	}
+	defer s.release(hold)
+	pushID := pushIDFor(workspaceID, userID, key)
+
 	if targetSessionID == "" || deliveredSessions > 0 {
 		// The ONE direct Bus.Publish call in this package, and the only
 		// one that acts on the result (BUG-2699 — see
@@ -346,6 +374,7 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) {
 		// instruction is gone, and the caller is the only one who can do
 		// anything about that.
 		if err := s.watchEvents.Publish(watchevents.Notification{
+			PushID:          pushID,
 			WorkspaceID:     workspaceID,
 			ItemID:          item.ID,
 			CollectionID:    item.CollectionID,
@@ -357,6 +386,13 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) {
 			TargetUserID:    userID,
 			TargetSessionID: targetSessionID,
 		}); err != nil {
+			// An unconfirmed publish may have delivered: the key records
+			// that, so a same-key retry answers 409 push_outcome_unknown
+			// instead of delivering a second time. A closed bus published
+			// nothing, and the deferred release frees the key.
+			if hold != nil && !errors.Is(err, watchevents.ErrBusClosed) {
+				s.completeKey(hold, http.StatusConflict, pushOutcomeUnknownEnvelope(item.Ref))
+			}
 			writePushPublishError(w, err, item.Ref)
 			return
 		}
@@ -371,6 +407,9 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) {
 	if !deliveredSessionsUnknown {
 		resp.DeliveredSessions = &deliveredSessions
 	}
+	if hold != nil {
+		s.completeKey(hold, http.StatusOK, resp)
+	}
 	writeJSON(w, http.StatusOK, resp)
 }
 

internal/server/handlers_watch_events.go

@@ -56,6 +56,9 @@ type watchEventPayload struct {
 	// may no longer resolve (purged), where Workspace falls back to the ID.
 	WorkspaceID string `json:"workspace_id,omitempty"`
 	Change      string `json:"change,omitempty"`
+	// PushID is set only on a push (TASK-3585) and omitted otherwise: a
+	// monitor that has already printed this push_id drops the event.
+	PushID string `json:"push_id,omitempty"`
 }
 
 // handleWatchEventsStream streams the caller's watch/nudge notifications.
@@ -862,6 +865,7 @@ func watchEventPayloadFor(s *Server, n watchevents.Notification) watchEventPaylo
 		Kind:      n.Kind,
 		Actor:     actor,
 		Summary:   n.Summary,
+		PushID:    n.PushID,
 	}
 	if n.Kind == watchevents.KindWorkspaceAccessChanged {
 		p.WorkspaceID = n.WorkspaceID

internal/server/idempotency_keys.go

@@ -6,8 +6,11 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"log/slog"
 	"net/http"
 
+	"github.com/google/uuid"
+
 	"github.com/PerpetualSoftware/pad/internal/models"
 	"github.com/PerpetualSoftware/pad/internal/store"
 )
@@ -51,10 +54,8 @@ func (s *Server) claimKey(w http.ResponseWriter, r *http.Request, workspaceID, r
 	if key == "" {
 		return nil, true
 	}
-	body, _ := json.Marshal(canon)
-	sum := sha256.Sum256([]byte(route + "\x00" + scope + "\x00" + string(body)))
 	userID := currentUserID(r)
-	c, err := s.store.ClaimIdempotencyKey(workspaceID, userID, key, route, hex.EncodeToString(sum[:]), idempotencyLease)
+	c, err := s.store.ClaimIdempotencyKey(workspaceID, userID, key, route, idemRequestHash(route, scope, canon), idempotencyLease)
 	if err != nil {
 		writeInternalError(w, err)
 		return nil, false
@@ -149,3 +150,54 @@ func (s *Server) replayAppend(w http.ResponseWriter, r *http.Request, visIDs []s
 // request hash. Its answer is the whole first ItemCopyResult: the warnings
 // it carries (dropped fields, attachments cloned) cannot be recomputed later.
 const idempotencyRouteCopy = "item.copy"
+
+// idempotencyRoutePush names the push route in a key's request hash.
+const idempotencyRoutePush = "item.push"
+
+// pushOutcomeUnknownCode answers a same-key retry of a push whose publish
+// could not be confirmed (TASK-3585): it may have been delivered, so it is
+// not published again.
+const pushOutcomeUnknownCode = "push_outcome_unknown"
+
+// pushOutcomeUnknownEnvelope is the stored answer for that retry, in the
+// API's error envelope.
+func pushOutcomeUnknownEnvelope(ref string) any {
+	return map[string]any{"error": map[string]any{
+		"code": pushOutcomeUnknownCode,
+		"message": "An earlier push with this key could not be confirmed; it may or may not have been delivered, so it is not sent again. " +
+			"Check your agent session; to send it anyway, use a NEW key.",
+		"details": map[string]any{"ref": ref},
+	}}
+}
+
+// completeKey records an answer for a write that has no transaction of its
+// own (push). A failure is logged, not returned: the write already
+// happened, and the push_id still lets a monitor drop a re-publish.
+func (s *Server) completeKey(h *idemHold, status int, answer any) {
+	b, err := json.Marshal(answer)
+	if err == nil {
+		err = s.store.CompleteIdempotencyKey(h.workspaceID, h.userID, h.key, h.token, status, string(b))
+	}
+	if err != nil {
+		slog.Warn("idempotency key not completed after a write with no transaction", "key", h.key, "error", err)
+	}
+}
+
+// pushIDFor is the push_id a publish carries (TASK-3585). With a key it is
+// derived from (workspace, user, key), so a re-publish after a claim
+// takeover carries the same id; without one it is fresh.
+func pushIDFor(workspaceID, userID, key string) string {
+	if key == "" {
+		return uuid.NewString()
+	}
+	sum := sha256.Sum256([]byte(workspaceID + "\x00" + userID + "\x00" + key))
+	return "push_" + hex.EncodeToString(sum[:16])
+}
+
+// idemRequestHash is the request half of a key's identity: the route, its
+// scope (the item) and the canonical request body.
+func idemRequestHash(route, scope string, canon any) string {
+	body, _ := json.Marshal(canon)
+	sum := sha256.Sum256([]byte(route + "\x00" + scope + "\x00" + string(body)))
+	return hex.EncodeToString(sum[:])
+}

internal/server/task3585_push_key_test.go

@@ -0,0 +1,141 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/PerpetualSoftware/pad/internal/watchevents"
+)
+
+// TASK-3585: idempotency keys on push. Push writes no row, so the key is
+// completed after the publish; every publish carries a push_id derived from
+// the key so a monitor can drop a re-publish after a claim takeover.
+
+// capturingBus records what was published and fails with err when set.
+type capturingBus struct {
+	mu   sync.Mutex
+	err  error
+	sent []watchevents.Notification
+}
+
+func (b *capturingBus) Publish(n watchevents.Notification) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.sent = append(b.sent, n)
+	return b.err
+}
+func (b *capturingBus) published() []watchevents.Notification {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return append([]watchevents.Notification(nil), b.sent...)
+}
+func (b *capturingBus) setErr(err error) { b.mu.Lock(); b.err = err; b.mu.Unlock() }
+func (b *capturingBus) Subscribe() (chan watchevents.Notification, <-chan struct{}, error) {
+	return nil, nil, nil
+}
+func (b *capturingBus) SubscribeAndReplaySince(context.Context, int64) (chan watchevents.Notification, []watchevents.Notification, <-chan struct{}, error) {
+	return nil, nil, nil, nil
+}
+func (b *capturingBus) Unsubscribe(chan watchevents.Notification) {}
+func (b *capturingBus) EventsSince(int64) []watchevents.Notification {
+	return nil
+}
+func (b *capturingBus) Close() {}
+
+func pushOnce(t *testing.T, srv *Server, slug, itemSlug, token, key string) *http.Response {
+	t.Helper()
+	rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+itemSlug+"/push", token,
+		map[string]interface{}{"message": "triage this", "op_key": key})
+	return rr.Result()
+}
+
+func TestTASK3585_PushReplaysWithoutRedelivering(t *testing.T) {
+	srv := testServer(t)
+	bus := &capturingBus{}
+	srv.SetWatchEventsBus(bus)
+	slug, item, tok, _ := setupWatchTestUser(t, srv)
+
+	first := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-1")
+	again := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-1")
+	if first.StatusCode != http.StatusOK || again.StatusCode != http.StatusOK || again.Header.Get("Idempotent-Replayed") != "true" {
+		t.Fatalf("replay: %d/%d", first.StatusCode, again.StatusCode)
+	}
+	if n := len(bus.published()); n != 1 {
+		t.Fatalf("a same-key retry published again: %d publishes", n)
+	}
+	if bus.published()[0].PushID == "" {
+		t.Fatal("a push must carry a push_id")
+	}
+}
+
+func TestTASK3585_AmbiguousPushIsNotRedelivered(t *testing.T) {
+	srv := testServer(t)
+	bus := &capturingBus{err: errors.New("redis: i/o timeout")}
+	srv.SetWatchEventsBus(bus)
+	slug, item, tok, _ := setupWatchTestUser(t, srv)
+
+	first := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-2")
+	if first.StatusCode != http.StatusBadGateway {
+		t.Fatalf("premise: an unconfirmed publish answers 502; got %d", first.StatusCode)
+	}
+	bus.setErr(nil)
+	again := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-2")
+	var env struct {
+		Error struct{ Code string } `json:"error"`
+	}
+	_ = json.NewDecoder(again.Body).Decode(&env)
+	if again.StatusCode != http.StatusConflict || env.Error.Code != pushOutcomeUnknownCode {
+		t.Fatalf("retry after an ambiguous publish: %d %q", again.StatusCode, env.Error.Code)
+	}
+	if n := len(bus.published()); n != 1 {
+		t.Fatalf("the ambiguous push was delivered again: %d publishes", n)
+	}
+}
+
+func TestTASK3585_ClosedBusReleasesThePushKey(t *testing.T) {
+	srv := testServer(t)
+	bus := &capturingBus{err: watchevents.ErrBusClosed}
+	srv.SetWatchEventsBus(bus)
+	slug, item, tok, _ := setupWatchTestUser(t, srv)
+
+	if r := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-3"); r.StatusCode != http.StatusServiceUnavailable {
+		t.Fatalf("premise: a closed bus answers 503; got %d", r.StatusCode)
+	}
+	bus.setErr(nil)
+	again := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-3")
+	if again.StatusCode != http.StatusOK || again.Header.Get("Idempotent-Replayed") != "" {
+		t.Fatalf("a push that published nothing must be sendable again with its key: %d", again.StatusCode)
+	}
+}
+
+// A server that dies between publishing and completing the key leaves a
+// claim whose lease expires. The retry takes it over and publishes again,
+// with the SAME push_id, which is what lets a monitor drop the repeat.
+func TestTASK3585_PushTakeoverKeepsThePushID(t *testing.T) {
+	srv := testServer(t)
+	bus := &capturingBus{}
+	srv.SetWatchEventsBus(bus)
+	slug, item, tok, user := setupWatchTestUser(t, srv)
+	ws, err := srv.store.GetWorkspaceBySlug(slug)
+	if err != nil || ws == nil {
+		t.Fatalf("workspace: %v", err)
+	}
+	hash := idemRequestHash(idempotencyRoutePush, item.ID, pushRequest{Message: "triage this"})
+	// The dead first attempt: claimed, never completed, lease already over.
+	if _, err := srv.store.ClaimIdempotencyKey(ws.ID, user.ID, "p-4", idempotencyRoutePush, hash, -time.Second); err != nil {
+		t.Fatal(err)
+	}
+	r := pushOnce(t, srv, slug, item.Slug, tok.Token, "p-4")
+	if r.StatusCode != http.StatusOK {
+		t.Fatalf("takeover: %d", r.StatusCode)
+	}
+	sent := bus.published()
+	if len(sent) != 1 || sent[0].PushID != pushIDFor(ws.ID, user.ID, "p-4") {
+		t.Fatalf("the takeover must publish the key's stable push_id: %+v", sent)
+	}
+}

internal/store/idempotency.go

@@ -119,6 +119,20 @@ func (s *Store) CompleteIdempotencyKeyTx(tx *sql.Tx, workspaceID, userID, key, t
 	return nil
 }
 
+// CompleteIdempotencyKey is CompleteIdempotencyKeyTx for a write that has
+// no transaction of its own (push, TASK-3585), in a transaction of its own.
+func (s *Store) CompleteIdempotencyKey(workspaceID, userID, key, token string, status int, response string) error {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	if err := s.CompleteIdempotencyKeyTx(tx, workspaceID, userID, key, token, status, response); err != nil {
+		return err
+	}
+	return tx.Commit()
+}
+
 // ReleaseIdempotencyKey gives up a claim the write refused, so a corrected
 // retry with the same key can proceed (lead ruling: refusals are not
 // stored). A pending claim is deleted; an unknown one keeps its record and

internal/watchevents/watchevents.go

@@ -145,7 +145,12 @@ const (
 type Notification struct {
 	// ID is a monotonic sequence number assigned by the Bus on Publish,
 	// used for Last-Event-ID resume. Zero until Publish assigns it.
-	ID          int64
+	ID int64
+	// PushID identifies one push across re-publishes (TASK-3585): for a
+	// keyed push it is derived from the key, so a re-publish after a claim
+	// takeover carries the same id and a monitor drops the repeat. Empty on
+	// every other kind.
+	PushID      string
 	WorkspaceID string
 	ItemID      string
 	// CollectionID is the watched/assigned item's collection, needed so