diff --git a/internal/server/handlers_push.go b/internal/server/handlers_push.go index 32ea9257..f6414fbb 100644 --- a/internal/server/handlers_push.go +++ b/internal/server/handlers_push.go @@ -13,6 +13,15 @@ import ( // pushRequest is the body of POST .../items/{itemSlug}/push. type pushRequest struct { Message string `json:"message"` + // TargetSessionID optionally narrows delivery to one of the caller's + // OWN live sessions from the S1 presence registry (PLAN-2558 S5, + // TASK-2588; GET /api/v1/sessions is where a caller learns the id). + // Omitted (the pre-S5 shape) means broadcast to every one of the + // caller's connected sessions, unchanged. API + TS client + web + // picker only in this slice, per CONVE-1741 — no CLI flag, no MCP + // surface; internal/cli's PushResult mirror simply never sends or + // reads this field. + TargetSessionID string `json:"target_session_id,omitempty"` } // pushResponse is the body of a successful push (dispatcher review round @@ -24,10 +33,51 @@ type pushRequest struct { type pushResponse struct { Ref string `json:"ref"` Workspace string `json:"workspace"` - Pushed bool `json:"pushed"` - Message string `json:"message"` + // Pushed means "accepted and processed", not "delivered" — it is + // true even when a TARGETED push's publish was skipped because the + // id matched no live session (dispatcher ruling, TASK-2588 round 2: + // broadcast-with-no-listeners has always returned exactly this shape + // — true, with nothing to receive it — and a targeted miss is not + // given a different contract just because DeliveredSessions can now + // say more about it). DeliveredSessions is the delivery signal; do + // not read Pushed as one. + Pushed bool `json:"pushed"` + Message string `json:"message"` + // DeliveredSessions counts how many of the caller's own live sessions + // (S1 presence registry, `target_session_id`-filtered if one was + // given) matched — PLAN-2558 S5, TASK-2588. This is a PREDICTION read + // from the registry, not a delivery receipt: it carries the exact + // same staleness window as GET /api/v1/sessions (session_presence.go's + // LiveSession doc comment — up to ~30s behind an ungracefully-dropped + // connection) and there is still no ack from the receiving side. A + // vanished or cross-user target_session_id is 0, the same as "nothing + // connected" — deliberately not a distinct error, so the CLI's pre-S5 + // behavior is unchanged by construction. + // + // SNAPSHOTTED BEFORE THE PUBLISH, not after (dispatcher review round + // 1, codex): counting post-publish raced the very thing it reports on + // — a targeted session could receive the notification and then + // disconnect before the count read, reporting 0 on a push that had + // already landed exactly once. handlePushToItem reads presence FIRST + // and, for a targeted push, skips the publish entirely when the + // target isn't in that snapshot — see its doc comment — so a 0 here + // is never a race, it's a guarantee: nothing was sent. + DeliveredSessions int `json:"delivered_sessions"` } +// maxPushTargetSessionIDLen bounds target_session_id (dispatcher review +// round 1, codex). It's trimmed but otherwise unvalidated — see its doc +// comment in pushRequest — and decodeJSON allows request bodies up to +// 2 MiB, so without a cap an authenticated caller could park arbitrarily +// large garbage strings in the bus's 1024-entry replay buffer on every +// push. 256 runes is comfortably above any id the S1 presence registry +// actually issues (a uuid.NewString() is 36) — a registry-issued id can +// never be rejected by this bound, so nothing a real client sends is +// ever affected; this exists purely to keep an unmatchable payload out +// of shared memory, not to constrain the id format (still opaque, still +// no format enforced beyond length). +const maxPushTargetSessionIDLen = 256 + // maxPushMessageLen bounds a push's instruction text, measured in runes // AFTER whitespace collapse (dispatcher review round 1). Two // constraints in tension set this: a push message is a free-form @@ -49,15 +99,20 @@ const maxPushMessageLen = 4096 // notification (IDEA-2544 Phase 1) — the "push this to my agent" verb: // an explicit, user-authored instruction bound to an item, delivered to // every one of the pushing user's OWN connected monitor sessions via -// GET /api/v1/events/stream. Unlike watch/assignment notifications, -// this has no durable backing (Dave's product call: fire-and-forget is -// acceptable for v1 — no inbox, no "no session connected" warning; the -// bus's replay buffer is the only resilience a push gets). +// GET /api/v1/events/stream, or to exactly one of them when the request +// names a target_session_id (PLAN-2558 S5, TASK-2588). Unlike watch/ +// assignment notifications, this has no durable backing (Dave's product +// call: fire-and-forget is acceptable for v1 — no inbox, no "no session +// connected" warning; the bus's replay buffer is the only resilience a +// push gets). // // Self-addressed only: pushing into someone else's session is a consent // question, not a code question (IDEA-2544 plan), so TargetUserID is // always set to the CALLER's own ID, never a request-supplied target — -// there is no cross-user push in Phase 1. +// there is no cross-user push. TargetSessionID (S5) does not relax this: +// it can only narrow delivery WITHIN the sessions ListForUser(userID) +// already scopes to, never address a session outside it — see +// deliveredSessionCount. // // POST /api/v1/workspaces/{slug}/items/{itemSlug}/push func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) { @@ -129,23 +184,88 @@ func (s *Server) handlePushToItem(w http.ResponseWriter, r *http.Request) { actor, _ := actorFromRequest(r) actorName := actorNameFromRequest(r) + // Trimmed, not otherwise validated beyond the length cap below: a + // session id is opaque to this handler (see deliveredSessionCount and + // Notification.TargetSessionID) — an id that names no live session of + // userID's just matches nothing, there is no format to enforce. + targetSessionID := strings.TrimSpace(input.TargetSessionID) + if length := len([]rune(targetSessionID)); length > maxPushTargetSessionIDLen { + writeError(w, http.StatusBadRequest, "bad_request", + fmt.Sprintf("target_session_id must be %d characters or fewer (got %d)", maxPushTargetSessionIDLen, length)) + return + } - s.watchEvents.Publish(watchevents.Notification{ - WorkspaceID: workspaceID, - ItemID: item.ID, - CollectionID: item.CollectionID, - ItemRef: item.Ref, - Kind: watchevents.KindPush, - Actor: actor, - ActorName: actorName, - Summary: message, - TargetUserID: userID, - }) + // Presence is read BEFORE the publish, not after (dispatcher review + // round 1, codex — see DeliveredSessions' doc comment for the race a + // post-publish count had). For a TARGETED push whose id isn't in this + // snapshot, the publish is skipped entirely: session ids are + // per-connection and never reused (session_presence.go's + // MemorySessionPresence.Add mints a fresh uuid per Add call), so a + // target absent right now can never later be matched by the SAME + // connection reconnecting under that id, nor by the bus's replay + // buffer (which only serves a resumed connection presenting its own + // prior Last-Event-ID, not an arbitrary target id). The notification + // would therefore be a guaranteed no-op — skipping it is what makes + // delivered_sessions=0 an honest guarantee rather than a snapshot + // that a slower reader could still race. Broadcast (targetSessionID + // == "") keeps the original fire-and-forget posture and always + // publishes, same as pre-S5 — its count is a pre-publish snapshot of + // who's connected, not a promise that count still holds by the time + // delivery happens, which is the same staleness every presence + // answer on this surface already carries. + deliveredSessions := deliveredSessionCount(s.sessionPresence, userID, targetSessionID) + if targetSessionID == "" || deliveredSessions > 0 { + s.watchEvents.Publish(watchevents.Notification{ + WorkspaceID: workspaceID, + ItemID: item.ID, + CollectionID: item.CollectionID, + ItemRef: item.Ref, + Kind: watchevents.KindPush, + Actor: actor, + ActorName: actorName, + Summary: message, + TargetUserID: userID, + TargetSessionID: targetSessionID, + }) + } writeJSON(w, http.StatusOK, pushResponse{ - Ref: item.Ref, - Workspace: ws.Slug, - Pushed: true, - Message: message, + Ref: item.Ref, + Workspace: ws.Slug, + Pushed: true, + Message: message, + DeliveredSessions: deliveredSessions, }) } + +// deliveredSessionCount answers "how many of userID's own live sessions +// will this push's delivery predicate match?" (PLAN-2558 S5, TASK-2588). +// It reads the SAME self-scoped list session_presence.go's +// SessionPresence.ListForUser already restricts every other consumer to +// (handlers_sessions.go's GET /api/v1/sessions is the other one) — that +// scoping is what makes a targetSessionID belonging to a DIFFERENT user +// structurally indistinguishable from a vanished one: it is simply never +// in userID's own list, so it falls out to 0 without any cross-user +// lookup or special-casing here. +// +// A nil presence (no registry wired) answers 0 rather than guessing — +// consistent with handleListSessions' own refusal to report an empty +// list as "nobody connected" when it genuinely cannot tell; here there +// is no error channel to say "can't tell" through (pushResponse.Pushed +// is still true — the notification really was published), so 0 is the +// closest honest answer available. +func deliveredSessionCount(presence SessionPresence, userID, targetSessionID string) int { + if presence == nil { + return 0 + } + sessions := presence.ListForUser(userID) + if targetSessionID == "" { + return len(sessions) + } + for _, sess := range sessions { + if sess.ID == targetSessionID { + return 1 + } + } + return 0 +} diff --git a/internal/server/handlers_push_test.go b/internal/server/handlers_push_test.go index 88c95d33..119cf166 100644 --- a/internal/server/handlers_push_test.go +++ b/internal/server/handlers_push_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/PerpetualSoftware/pad/internal/models" "github.com/PerpetualSoftware/pad/internal/watchevents" ) @@ -191,3 +192,295 @@ func TestPushToItem_InvisibleItemDenied(t *testing.T) { t.Fatalf("expected 404, got %d (body: %s)", rr.Code, rr.Body.String()) } } + +// TestPushToItem_TargetedSessionReceivesOnly is PLAN-2558 S5's (TASK-2588) +// core positive case: with TWO of the caller's own sessions connected, a +// push naming one of their ids reaches exactly that one, not the other, +// and delivered_sessions reports 1 — not the registry size of 2. +func TestPushToItem_TargetedSessionReceivesOnly(t *testing.T) { + t.Parallel() + srv := testServerWithPresence(t) + slug, item, tok, _ := setupWatchTestUser(t, srv) + ts := httptest.NewServer(srv) + defer ts.Close() + + ctxA, cancelA := context.WithCancel(context.Background()) + defer cancelA() + chA := connectWatchStream(ctxA, t, ts.URL, tok.Token) + waitForWatchEvent(t, chA, 3*time.Second) // connected — A's Add() has happened + + ctxB, cancelB := context.WithCancel(context.Background()) + defer cancelB() + chB := connectWatchStream(ctxB, t, ts.URL, tok.Token) + waitForWatchEvent(t, chB, 3*time.Second) // connected — B's Add() has happened, strictly after A's + + status, sessions := getSessions(t, ts.URL, tok.Token) + if status != http.StatusOK || len(sessions.Sessions) != 2 { + t.Fatalf("expected 2 live sessions, got status=%d sessions=%+v", status, sessions) + } + // ListForUser orders oldest-connection-first (session_presence.go), and + // A connected strictly before B, so sessions[0] is A's own id. + targetID := sessions.Sessions[0].ID + + rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/push", tok.Token, + map[string]interface{}{"message": "for A only", "target_session_id": targetID}) + if rr.Code != http.StatusOK { + t.Fatalf("push: %d %s", rr.Code, rr.Body.String()) + } + var resp pushResponse + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("parse response: %v", err) + } + if resp.DeliveredSessions != 1 { + t.Fatalf("expected delivered_sessions=1 for a session-targeted push, got %d", resp.DeliveredSessions) + } + + ev := waitForWatchEvent(t, chA, 3*time.Second) + var payload watchEventPayload + if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil { + t.Fatalf("parse payload: %v", err) + } + if payload.Summary != "for A only" { + t.Fatalf("expected the targeted session to receive the push, got summary %q", payload.Summary) + } + + assertNoWatchEventForRef(t, chB, item.Ref, 300*time.Millisecond) +} + +// TestPushToItem_TargetedVanishedSessionMisses is S5's honest-miss case: +// a target_session_id that names no live session (mistyped, or expired) +// gets a 200 with delivered_sessions=0, never mis-delivered to a session +// that IS connected but wasn't the one addressed. +func TestPushToItem_TargetedVanishedSessionMisses(t *testing.T) { + t.Parallel() + srv := testServerWithPresence(t) + slug, item, tok, _ := setupWatchTestUser(t, srv) + ts := httptest.NewServer(srv) + defer ts.Close() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ch := connectWatchStream(ctx, t, ts.URL, tok.Token) + waitForWatchEvent(t, ch, 3*time.Second) // connected + + // Bus growth, not just "this OTHER session didn't get it" (dispatcher + // review round 1, codex): a targeted miss must skip the publish + // entirely, not merely fail to match anyone downstream — see + // pushResponse.DeliveredSessions' doc comment on why. This is the + // seam that fails if the pre-publish snapshot-and-skip ever gets + // reordered back to publish-then-count. + before := len(srv.watchEvents.EventsSince(0)) + + rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/push", tok.Token, + map[string]interface{}{"message": "nobody home", "target_session_id": "sess-does-not-exist"}) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200 for a vanished target (honest miss, not an error), got %d (body: %s)", rr.Code, rr.Body.String()) + } + var resp pushResponse + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("parse response: %v", err) + } + if resp.DeliveredSessions != 0 { + t.Fatalf("expected delivered_sessions=0 for a vanished target, got %d", resp.DeliveredSessions) + } + if !resp.Pushed { + t.Fatal("expected pushed=true — a targeted miss is still a successfully PROCESSED push (matching broadcast's existing no-listeners semantics), even though nothing was published to the bus") + } + if after := len(srv.watchEvents.EventsSince(0)); after != before { + t.Fatalf("expected a targeted miss to skip the publish entirely (guaranteed no-op), but the bus grew from %d to %d entries", before, after) + } + + assertNoWatchEventForRef(t, ch, item.Ref, 300*time.Millisecond) +} + +// TestPushToItem_TargetedSessionOfAnotherUserTreatedAsVanished pins the +// self-addressed boundary (dispatcher constraint, TASK-2588): a +// target_session_id that names a REAL, currently-connected session +// belonging to a DIFFERENT user must behave exactly like a vanished id — +// 200, delivered_sessions=0 — never leak to that other user's session and +// never let the pusher probe whether a given id exists on the server. +func TestPushToItem_TargetedSessionOfAnotherUserTreatedAsVanished(t *testing.T) { + t.Parallel() + srv := testServerWithPresence(t) + slug, item, tokA, _ := setupWatchTestUser(t, srv) + userB, err := srv.store.CreateUser(models.UserCreate{ + Email: "push-target-test-b@example.com", + Name: "Push Target Tester B", + Password: "pw-test-12345", + }) + if err != nil { + t.Fatalf("CreateUser: %v", err) + } + tokB, err := srv.store.CreateAPIToken(userB.ID, models.APITokenCreate{Name: "push-target-test-b"}, 0, 0) + if err != nil { + t.Fatalf("CreateAPIToken: %v", err) + } + ts := httptest.NewServer(srv) + defer ts.Close() + + ctxA, cancelA := context.WithCancel(context.Background()) + defer cancelA() + chA := connectWatchStream(ctxA, t, ts.URL, tokA.Token) + waitForWatchEvent(t, chA, 3*time.Second) + + ctxB, cancelB := context.WithCancel(context.Background()) + defer cancelB() + chB := connectWatchStream(ctxB, t, ts.URL, tokB.Token) + waitForWatchEvent(t, chB, 3*time.Second) + + _, bSessions := getSessions(t, ts.URL, tokB.Token) + if len(bSessions.Sessions) != 1 { + t.Fatalf("expected user B to see exactly 1 live session, got %+v", bSessions) + } + bSessionID := bSessions.Sessions[0].ID + + // See TestPushToItem_TargetedVanishedSessionMisses: a cross-user id + // must skip the publish exactly like a genuinely vanished one — the + // bus must not grow, not merely fail to reach B downstream. + before := len(srv.watchEvents.EventsSince(0)) + + rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/push", tokA.Token, + map[string]interface{}{"message": "should reach nobody", "target_session_id": bSessionID}) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200, got %d (body: %s)", rr.Code, rr.Body.String()) + } + var resp pushResponse + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("parse response: %v", err) + } + if resp.DeliveredSessions != 0 { + t.Fatalf("expected delivered_sessions=0 for another user's session id — a cross-user id must be indistinguishable from a vanished one, got %d", resp.DeliveredSessions) + } + if after := len(srv.watchEvents.EventsSince(0)); after != before { + t.Fatalf("expected a cross-user target to skip the publish entirely, but the bus grew from %d to %d entries", before, after) + } + + assertNoWatchEventForRef(t, chA, item.Ref, 300*time.Millisecond) + assertNoWatchEventForRef(t, chB, item.Ref, 300*time.Millisecond) +} + +// TestPushToItem_TargetSessionIDOverLengthRejected covers the 400-over-cap +// guard (dispatcher review round 1, codex): target_session_id is opaque +// and unvalidated by format, but not unbounded — decodeJSON allows request +// bodies up to 2 MiB, so without a length cap an authenticated caller +// could park arbitrarily large garbage strings in the bus's replay +// buffer on every push. The bus must not grow, matching the "reject +// before publish" posture the length check shares with the message cap. +func TestPushToItem_TargetSessionIDOverLengthRejected(t *testing.T) { + t.Parallel() + srv := testServerWithPresence(t) + slug, item, tok, _ := setupWatchTestUser(t, srv) + + before := len(srv.watchEvents.EventsSince(0)) + overLong := strings.Repeat("a", maxPushTargetSessionIDLen+1) + rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/push", tok.Token, + map[string]interface{}{"message": "triage this", "target_session_id": overLong}) + if rr.Code != http.StatusBadRequest { + t.Fatalf("expected 400 for an over-cap target_session_id, got %d (body: %s)", rr.Code, rr.Body.String()) + } + if after := len(srv.watchEvents.EventsSince(0)); after != before { + t.Fatalf("expected an over-cap target_session_id to be rejected before publish, but the bus grew from %d to %d entries", before, after) + } +} + +// TestPushToItem_TargetSessionIDAtCapAccepted is the at-boundary +// counterpart: a target_session_id exactly at the cap must NOT be +// rejected (it is still a miss — nothing that long is a real registry +// id — but the length check itself must not be off-by-one). +func TestPushToItem_TargetSessionIDAtCapAccepted(t *testing.T) { + t.Parallel() + srv := testServerWithPresence(t) + slug, item, tok, _ := setupWatchTestUser(t, srv) + + atCap := strings.Repeat("a", maxPushTargetSessionIDLen) + rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/push", tok.Token, + map[string]interface{}{"message": "triage this", "target_session_id": atCap}) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200 for a target_session_id exactly at the cap, got %d (body: %s)", rr.Code, rr.Body.String()) + } + var resp pushResponse + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("parse response: %v", err) + } + if resp.DeliveredSessions != 0 { + t.Fatalf("expected delivered_sessions=0 — an at-cap id still names no live session, got %d", resp.DeliveredSessions) + } +} + +// TestPushToItem_BroadcastDeliveredSessionsCountsLiveSessions covers the +// broadcast (omitted target_session_id) mode with delivered_sessions +// wired up: it reaches every one of the caller's own connected sessions, +// and the count is the actual number that matched — 2, from +// ListForUser(userID) — not any GLOBAL registry size. +func TestPushToItem_BroadcastDeliveredSessionsCountsLiveSessions(t *testing.T) { + t.Parallel() + srv := testServerWithPresence(t) + slug, item, tok, _ := setupWatchTestUser(t, srv) + ts := httptest.NewServer(srv) + defer ts.Close() + + ctxA, cancelA := context.WithCancel(context.Background()) + defer cancelA() + chA := connectWatchStream(ctxA, t, ts.URL, tok.Token) + waitForWatchEvent(t, chA, 3*time.Second) + + ctxB, cancelB := context.WithCancel(context.Background()) + defer cancelB() + chB := connectWatchStream(ctxB, t, ts.URL, tok.Token) + waitForWatchEvent(t, chB, 3*time.Second) + + rr := bearerJSON(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/push", tok.Token, + map[string]interface{}{"message": "for everyone"}) + if rr.Code != http.StatusOK { + t.Fatalf("push: %d %s", rr.Code, rr.Body.String()) + } + var resp pushResponse + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("parse response: %v", err) + } + if resp.DeliveredSessions != 2 { + t.Fatalf("expected delivered_sessions=2 for a broadcast with 2 live sessions, got %d", resp.DeliveredSessions) + } + + for _, ch := range []<-chan watchSSEEvent{chA, chB} { + ev := waitForWatchEvent(t, ch, 3*time.Second) + var payload watchEventPayload + if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil { + t.Fatalf("parse payload: %v", err) + } + if payload.Summary != "for everyone" { + t.Fatalf("expected both sessions to receive the broadcast push, got summary %q", payload.Summary) + } + } +} + +// TestPushToItem_OmittedTargetSessionIDMatchesPreS5RequestShape is the +// wire-level compatibility leg (dispatcher plan): a request body in +// EXACTLY the pre-S5 shape — no target_session_id key at all, not even +// an empty string — must still be accepted and behave as pure broadcast, +// against a server with no presence registry at all (mirroring every +// pre-S5 push test's setup via testServerWithWatchEvents). A vanished +// registry answers delivered_sessions=0 honestly rather than erroring — +// there is no new error channel for a caller that never asked to target +// anything. +func TestPushToItem_OmittedTargetSessionIDMatchesPreS5RequestShape(t *testing.T) { + t.Parallel() + srv := testServerWithWatchEvents(t) // NOT testServerWithPresence — no registry + slug, item, tok, _ := setupWatchTestUser(t, srv) + + rr := bearerCall(t, srv, "POST", "/api/v1/workspaces/"+slug+"/items/"+item.Slug+"/push", tok.Token, + []byte(`{"message":"triage this"}`)) + if rr.Code != http.StatusOK { + t.Fatalf("push: %d %s", rr.Code, rr.Body.String()) + } + var resp pushResponse + if err := json.Unmarshal(rr.Body.Bytes(), &resp); err != nil { + t.Fatalf("parse response: %v", err) + } + if resp.Ref != item.Ref || resp.Workspace != slug || !resp.Pushed || resp.Message != "triage this" { + t.Fatalf("pre-S5 fields must be unchanged for a pre-S5 request body, got %+v", resp) + } + if resp.DeliveredSessions != 0 { + t.Fatalf("expected delivered_sessions=0 with no presence registry wired, got %d", resp.DeliveredSessions) + } +} diff --git a/internal/server/handlers_watch_events.go b/internal/server/handlers_watch_events.go index 3d36277a..998238d5 100644 --- a/internal/server/handlers_watch_events.go +++ b/internal/server/handlers_watch_events.go @@ -141,8 +141,16 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) // The identity is whatever the client claimed in its request // headers (S2, TASK-2560) — sanitized, never verified, and empty // for any client that doesn't say. See SessionIdentity. + // Hoisted out of the if-block (PLAN-2558 S5, TASK-2588): a targeted + // push's predicate (Notification.TargetSessionID) is evaluated in the + // select loop below, which needs this connection's own registry id in + // scope. Left as the zero value when there is no registry — a + // targeted push can then never match THIS connection, but an + // untargeted (broadcast) one is unaffected, since watchNotificationVisible + // only compares TargetSessionID when the notification actually set one. + var sessionID string if s.sessionPresence != nil { - sessionID := s.sessionPresence.Add(user.ID, parseSessionIdentity(r)) + sessionID = s.sessionPresence.Add(user.ID, parseSessionIdentity(r)) defer s.sessionPresence.Remove(user.ID, sessionID) } @@ -182,7 +190,7 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) flusher.Flush() } else { for _, n := range missed { - if !watchNotificationVisible(watches, visCache.forWorkspace(n.WorkspaceID), user.ID, n) { + if !watchNotificationVisible(watches, visCache.forWorkspace(n.WorkspaceID), user.ID, sessionID, n) { continue } if err := writeSSEEvent(w, "notification", n.ID, watchEventPayloadFor(s, n)); err != nil { @@ -216,7 +224,7 @@ func (s *Server) handleWatchEventsStream(w http.ResponseWriter, r *http.Request) if !ok { return } - if !watchNotificationVisible(watches, visCache.forWorkspace(n.WorkspaceID), user.ID, n) { + if !watchNotificationVisible(watches, visCache.forWorkspace(n.WorkspaceID), user.ID, sessionID, n) { continue } if err := writeSSEEvent(w, "notification", n.ID, watchEventPayloadFor(s, n)); err != nil { @@ -413,7 +421,13 @@ func (s *Server) loadWatchPredicates(r *http.Request, userID string) (map[string // handler so the DR-2 filtering rules are unit-testable without a live // SSE connection — mirrors sseEventVisibleFor's role in // handlers_events.go. -func watchNotificationVisible(watches map[string]string, vis watchAccessVisibility, userID string, n watchevents.Notification) bool { +// +// sessionID is THIS connection's own S1 presence-registry id (empty if +// the server has no presence registry wired) — the same id +// Notification.TargetSessionID's doc comment describes evaluating +// against (PLAN-2558 S5, TASK-2588). Unused by every kind except +// KindPush. +func watchNotificationVisible(watches map[string]string, vis watchAccessVisibility, userID string, sessionID string, n watchevents.Notification) bool { // TASK-2533 codex round 2 finding 2: the caller's CURRENT access to // the notification's collection/item is checked FIRST and applies // UNIFORMLY to every kind below — watch-matched and addressed-to-you @@ -469,12 +483,23 @@ func watchNotificationVisible(watches map[string]string, vis watchAccessVisibili // original code fell through to the watch-map check on a // non-matching TargetUserID, so any unconditional watcher on the item // received every push addressed to every other user, instruction - // text included). Phase 4's session targeting is expected to inherit - // this same "addressed traffic is exclusive of watch-matched + // text included). PLAN-2558 S5's session targeting (TASK-2588) + // inherits this same "addressed traffic is exclusive of watch-matched // delivery" semantics, so it's pinned explicitly here rather than // left to fall out of the shared TargetUserID field's shape. if n.Kind == watchevents.KindPush { - return n.TargetUserID != "" && n.TargetUserID == userID + if n.TargetUserID == "" || n.TargetUserID != userID { + return false + } + // TargetSessionID narrows delivery to one of userID's own live + // sessions (empty = every one of them, i.e. broadcast is this + // same check with an always-true second leg — one delivery path, + // not two). Deliberately compared AFTER the TargetUserID check + // above, never before: a session id is meaningless without first + // confirming this connection belongs to the addressed user at + // all, and evaluating it first would invite conflating "wrong + // user" with "wrong session" in some future edit. + return n.TargetSessionID == "" || n.TargetSessionID == sessionID } predicate, watched := watches[n.ItemID] diff --git a/internal/server/handlers_watch_events_test.go b/internal/server/handlers_watch_events_test.go index 6fcee727..f03fa205 100644 --- a/internal/server/handlers_watch_events_test.go +++ b/internal/server/handlers_watch_events_test.go @@ -578,7 +578,7 @@ func TestWatchNotificationVisible_UnconditionalWatch(t *testing.T) { t.Parallel() watches := map[string]string{"item-1": ""} n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindComment} - if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected an unconditional watch to match any notification on the item") } } @@ -587,7 +587,7 @@ func TestWatchNotificationVisible_UnwatchedItemDenied(t *testing.T) { t.Parallel() watches := map[string]string{"item-1": ""} n := watchevents.Notification{ItemID: "item-2", Kind: watchevents.KindComment} - if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected an unwatched item's notification to be denied") } } @@ -596,7 +596,7 @@ func TestWatchNotificationVisible_PredicateGatesNonStatusKinds(t *testing.T) { t.Parallel() watches := map[string]string{"item-1": "status=done"} n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindComment} - if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected a predicated watch to suppress a comment notification") } } @@ -606,10 +606,10 @@ func TestWatchNotificationVisible_PredicateMatchesOnlyTargetValue(t *testing.T) watches := map[string]string{"item-1": "status=done"} wrong := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindStatusChange, StatusFieldKey: "status", ToStatus: "in-progress"} right := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindStatusChange, StatusFieldKey: "status", ToStatus: "done"} - if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", wrong) { + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", wrong) { t.Fatal("expected the non-matching status transition to be denied") } - if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", right) { + if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", right) { t.Fatal("expected the matching status transition to be visible") } } @@ -624,10 +624,10 @@ func TestWatchNotificationVisible_PredicateMatchesOnlyTargetValue(t *testing.T) func TestWatchNotificationVisible_AssignmentToYouNeedsAWatch(t *testing.T) { t.Parallel() n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindAssignment, AssignedUserID: "user-1"} - if watchNotificationVisible(map[string]string{}, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if watchNotificationVisible(map[string]string{}, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected an assignment-to-you notification to be denied with no watch on the item") } - if !watchNotificationVisible(map[string]string{"item-1": ""}, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if !watchNotificationVisible(map[string]string{"item-1": ""}, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected the same assignment to be visible to a caller holding an unconditional watch") } } @@ -636,7 +636,7 @@ func TestWatchNotificationVisible_AssignmentToSomeoneElseDenied(t *testing.T) { t.Parallel() watches := map[string]string{} n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindAssignment, AssignedUserID: "user-2"} - if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected an assignment to someone else to be denied") } } @@ -655,7 +655,7 @@ func TestWatchNotificationVisible_AssignmentToSomeoneElseVisibleToWatcher(t *tes t.Parallel() watches := map[string]string{"item-1": ""} // unconditional watch n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindAssignment, AssignedUserID: "user-2"} - if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected an unconditional watcher to see an assignment to someone else — an assignment is an item-level fact, not private dispatch") } } @@ -668,11 +668,63 @@ func TestWatchNotificationVisible_PushToYou(t *testing.T) { t.Parallel() watches := map[string]string{} // no watches at all n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindPush, TargetUserID: "user-1"} - if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected a push-to-you notification to be visible with no watch") } } +// TestWatchNotificationVisible_PushSessionTargetedMatchDelivers is +// PLAN-2558 S5's (TASK-2588) core positive case: a push naming a +// specific session id is visible on the connection whose OWN registry +// id matches it, even though TargetUserID alone would already have +// allowed it — this pins that the narrower predicate doesn't accidentally +// deny a matching session. +func TestWatchNotificationVisible_PushSessionTargetedMatchDelivers(t *testing.T) { + t.Parallel() + watches := map[string]string{} + n := watchevents.Notification{ + ItemID: "item-1", Kind: watchevents.KindPush, + TargetUserID: "user-1", TargetSessionID: "sess-a", + } + if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "sess-a", n) { + t.Fatal("expected a session-targeted push to be visible on the matching session") + } +} + +// TestWatchNotificationVisible_PushSessionTargetedMismatchDenied is S5's +// core negative case: TargetUserID matching is NOT sufficient once +// TargetSessionID is set — a push aimed at one of the user's sessions +// must not leak to another of that SAME user's own connected sessions. +func TestWatchNotificationVisible_PushSessionTargetedMismatchDenied(t *testing.T) { + t.Parallel() + watches := map[string]string{} + n := watchevents.Notification{ + ItemID: "item-1", Kind: watchevents.KindPush, + TargetUserID: "user-1", TargetSessionID: "sess-a", + } + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "sess-b", n) { + t.Fatal("expected a session-targeted push to be denied on a different session of the same user") + } +} + +// TestWatchNotificationVisible_PushEmptySessionTargetMatchesAnySession +// pins that an untargeted (broadcast) push is unaffected by S5: an +// empty TargetSessionID must still reach every one of the user's own +// sessions regardless of that connection's own registry id — including +// the zero-value id a connection gets when there is no presence +// registry wired at all (see the sessionID hoist in +// handleWatchEventsStream). +func TestWatchNotificationVisible_PushEmptySessionTargetMatchesAnySession(t *testing.T) { + t.Parallel() + watches := map[string]string{} + n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindPush, TargetUserID: "user-1"} + for _, sessionID := range []string{"sess-a", "sess-b", ""} { + if !watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", sessionID, n) { + t.Fatalf("expected a broadcast push (empty TargetSessionID) to be visible regardless of this connection's session id %q", sessionID) + } + } +} + // TestWatchNotificationVisible_PushToSomeoneElseDenied mirrors // TestWatchNotificationVisible_AssignmentToSomeoneElseDenied for // KindPush: Phase 1 only ever publishes self-addressed pushes, but the @@ -682,7 +734,7 @@ func TestWatchNotificationVisible_PushToSomeoneElseDenied(t *testing.T) { t.Parallel() watches := map[string]string{} n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindPush, TargetUserID: "user-2"} - if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected a push addressed to someone else to be denied") } } @@ -703,7 +755,7 @@ func TestWatchNotificationVisible_PushToSomeoneElseDeniedEvenWithUnconditionalWa t.Parallel() watches := map[string]string{"item-1": ""} // unconditional watch on the item n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindPush, TargetUserID: "user-2"} - if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected a push addressed to someone else to be denied even when the caller holds an unconditional watch on the item") } } @@ -715,7 +767,7 @@ func TestWatchNotificationVisible_PushToSomeoneElseDeniedEvenWithPredicateWatch( t.Parallel() watches := map[string]string{"item-1": "status=done"} n := watchevents.Notification{ItemID: "item-1", Kind: watchevents.KindPush, TargetUserID: "user-2"} - if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", n) { + if watchNotificationVisible(watches, watchAccessVisibility{fullAccess: true}, "user-1", "", n) { t.Fatal("expected a push addressed to someone else to be denied even when the caller holds a predicated watch on the item") } } @@ -732,12 +784,12 @@ func TestWatchNotificationVisible_PushStillGatedByAccess(t *testing.T) { ItemID: "item-1", CollectionID: "coll-1", Kind: watchevents.KindPush, TargetUserID: "user-1", } - if watchNotificationVisible(watches, deny, "user-1", n) { + if watchNotificationVisible(watches, deny, "user-1", "", n) { t.Fatal("expected a push-to-you notification to be denied when the caller has no current access to the item's collection") } allow := watchAccessVisibility{visibleCollIDs: map[string]bool{"coll-1": true}} - if !watchNotificationVisible(watches, allow, "user-1", n) { + if !watchNotificationVisible(watches, allow, "user-1", "", n) { t.Fatal("expected a push-to-you notification to be visible once the caller has current access to the item's collection") } } @@ -767,12 +819,12 @@ func TestWatchNotificationVisible_AssignmentStillGatedByAccess(t *testing.T) { ItemID: "item-1", CollectionID: "coll-1", Kind: watchevents.KindAssignment, AssignedUserID: "user-1", } - if watchNotificationVisible(watches, deny, "user-1", n) { + if watchNotificationVisible(watches, deny, "user-1", "", n) { t.Fatal("expected a watch-matched assignment notification to be denied when the caller has no current access to the item's collection") } allow := watchAccessVisibility{visibleCollIDs: map[string]bool{"coll-1": true}} - if !watchNotificationVisible(watches, allow, "user-1", n) { + if !watchNotificationVisible(watches, allow, "user-1", "", n) { t.Fatal("expected the same notification to be visible once the collection is in the caller's current access") } } diff --git a/internal/watchevents/watchevents.go b/internal/watchevents/watchevents.go index f58d7366..35e0a4b5 100644 --- a/internal/watchevents/watchevents.go +++ b/internal/watchevents/watchevents.go @@ -43,13 +43,18 @@ const ( // eventually exists. KindAsk = "ask" // KindPush is IDEA-2544 Phase 1's human→harness addressed-dispatch - // event: POST .../items/{itemSlug}/push publishes exactly one of - // these, self-addressed (TargetUserID == the pushing user), any time - // a user wants to put an item + instruction in front of their own - // harness right now rather than waiting on assignment/watch - // semantics. KindAsk is push's reserved sibling in the other - // direction (harness→human) — see TargetUserID's doc comment for the - // shared envelope shape the two are meant to converge on. + // event: POST .../items/{itemSlug}/push accepts one of these, + // self-addressed (TargetUserID == the pushing user), any time a user + // wants to put an item + instruction in front of their own harness + // right now rather than waiting on assignment/watch semantics. NOT + // "publishes exactly one" unconditionally as of PLAN-2558 S5 + // (TASK-2588): handlePushToItem decides whether to publish at all — + // a session-targeted request whose id matches no live session skips + // the publish entirely (a guaranteed no-op; see TargetSessionID and + // pushResponse.DeliveredSessions' doc comments in handlers_push.go). + // KindAsk is push's reserved sibling in the other direction + // (harness→human) — see TargetUserID's doc comment for the shared + // envelope shape the two are meant to converge on. KindPush = "push" ) @@ -104,6 +109,21 @@ type Notification struct { // notification could have been addressed to, and echoing it back // would leak a user ID for no consumer that needs it. TargetUserID string + // TargetSessionID narrows TargetUserID to one of that user's live + // event-stream connections (PLAN-2558 S5, TASK-2588). Populated only + // on Kind == KindPush, and only when the pusher named a specific + // session id from the S1 presence registry (GET /api/v1/sessions); + // empty means "every one of TargetUserID's connected sessions" — + // broadcast is targeted-with-an-empty-predicate, not a separate + // code path. watchNotificationVisible checks this against the + // SAME session id session_presence.go handed the connection at + // Add() time, exactly parallel to how TargetUserID gates against + // the connected caller's user id. An id that names no live session + // (vanished, mistyped, or — deliberately indistinguishable from + // either — belonging to a DIFFERENT user) simply matches nothing: + // there is no separate "not found" signal here, by design, because + // this field must never become an existence oracle across users. + TargetSessionID string // StatusFieldKey / ToStatus are populated on Kind == KindStatusChange, // mirroring models.ItemMutationSignal, so the `--until field=value` // watch predicate can be evaluated against a Notification directly diff --git a/web/src/lib/api/client.ts b/web/src/lib/api/client.ts index 1ca3b9bc..eb0cbe98 100644 --- a/web/src/lib/api/client.ts +++ b/web/src/lib/api/client.ts @@ -1423,27 +1423,45 @@ export const api = { /** * Push an instruction about this item to the caller's OWN connected - * agent sessions (IDEA-2544 Phase 1 / PLAN-2558 S3). + * agent sessions (IDEA-2544 Phase 1 / PLAN-2558 S3), or to exactly + * one of them when `targetSessionId` is given (PLAN-2558 S5, + * TASK-2588 — an id from `api.sessions.list()`). * * Fire-and-forget: there is no durable inbox and no ack, so a - * resolved promise means "published to the bus", never "an agent read - * it". Pair the call site with `api.sessions.list()` so the user - * knows whether anything is listening BEFORE they click — that pairing - * is the entire reason this endpoint has a web surface. + * resolved promise means "accepted and processed" (`pushed: true` in + * ItemPushResult), never "an agent read it" — and, as of PLAN-2558 + * S5, not even "published to the bus" unconditionally: a targeted + * request whose `targetSessionId` matches no live session resolves + * with `delivered_sessions: 0` and the server skips the publish + * entirely (see ItemPushResult's doc comment). Pair the call site + * with `api.sessions.list()` so the user knows whether anything is + * listening BEFORE they click — that pairing is the entire reason + * this endpoint has a web surface. * - * Self-addressed only; there is no target parameter, by design (see - * handlePushToItem). `message` is collapsed and bounded server-side — - * use `$lib/push/message` to apply the same rules in the composer so + * Self-addressed only; there is no cross-user target, by design (see + * handlePushToItem) — `targetSessionId` can only narrow delivery + * within the caller's OWN sessions, never address anyone else's. + * `message` is collapsed and bounded server-side — use + * `$lib/push/message` to apply the same rules in the composer so * over-length text is caught before it becomes a 400. * + * Omitting `targetSessionId` sends the exact pre-S5 body shape + * (`{ message }`, no `target_session_id` key at all) — existing + * broadcast call sites (QuickActionsMenu's S4 dispatch) are + * unaffected by this parameter's addition. + * * NEVER auto-retry this. Like the copy endpoint it carries no * idempotency key, so a retry after an ambiguous failure can deliver - * the same instruction twice. + * the same instruction twice. A targeted miss (`delivered_sessions + * === 0` in the response) is the one exception a caller may act on + * safely — see ItemPushResult's doc comment. */ - push: (ws: string, itemSlug: string, message: string) => + push: (ws: string, itemSlug: string, message: string, targetSessionId?: string) => request(`/workspaces/${ws}/items/${itemSlug}/push`, { method: 'POST', - body: JSON.stringify({ message }) + body: JSON.stringify( + targetSessionId ? { message, target_session_id: targetSessionId } : { message } + ) }) }, diff --git a/web/src/lib/components/items/PushToAgentDialog.svelte b/web/src/lib/components/items/PushToAgentDialog.svelte index 575e0274..7e3a249b 100644 --- a/web/src/lib/components/items/PushToAgentDialog.svelte +++ b/web/src/lib/components/items/PushToAgentDialog.svelte @@ -1,7 +1,19 @@ + {#if sessionCount > 0} +
+ + +
+ {/if} +
@@ -594,6 +709,12 @@ server tells you not to run would cost honesty in the case that actually ships. /* Border / background / padding come from app.css's global `input, textarea, select` rule — only the box behaviour is local. */ + /* Border / background / padding come from app.css's global + `input, textarea, select` rule, same as .composer above. */ + .target-picker { + width: 100%; + box-sizing: border-box; + } .composer { width: 100%; box-sizing: border-box; diff --git a/web/src/lib/components/items/PushToAgentDialog.svelte.test.ts b/web/src/lib/components/items/PushToAgentDialog.svelte.test.ts index 3b23f37a..37c5fb08 100644 --- a/web/src/lib/components/items/PushToAgentDialog.svelte.test.ts +++ b/web/src/lib/components/items/PushToAgentDialog.svelte.test.ts @@ -95,6 +95,12 @@ function textarea(): HTMLTextAreaElement { return el as HTMLTextAreaElement; } +/** The session-target picker (PLAN-2558 S5), or null when the presence + * state has no live sessions to pick between (it isn't rendered at all). */ +function targetPicker(): HTMLSelectElement | null { + return document.querySelector('select'); +} + /** Mount, then let the initial presence read settle. */ async function mountSettled(props: Record = {}) { const result = render(PushToAgentDialog, { props: baseProps(props) }); @@ -112,7 +118,8 @@ beforeEach(() => { ref: 'TASK-5', workspace: 'docapp', pushed: true, - message: 'ok' + message: 'ok', + delivered_sessions: 1 }); sessionsListMock.mockReset().mockResolvedValue(sessions(1)); toastMock.mockReset(); @@ -504,3 +511,229 @@ describe('PushToAgentDialog — message handling', () => { expect(pushMock).toHaveBeenCalledTimes(1); }); }); + +describe('PushToAgentDialog — session targeting (PLAN-2558 S5, TASK-2588)', () => { + it('renders a broadcast default plus one option per connected session', async () => { + sessionsListMock.mockResolvedValue(sessions(2)); + await mountSettled(); + + const picker = targetPicker(); + if (!picker) throw new Error('expected a target picker with 2 live sessions'); + const options = Array.from(picker.options).map((o) => ({ value: o.value, text: o.textContent })); + expect(options[0]).toEqual({ value: '', text: expect.stringContaining('All connected sessions (2)') }); + expect(options[1]).toEqual({ value: 'session-id-0', text: 'docapp-0 (pid 1000)' }); + expect(options[2]).toEqual({ value: 'session-id-1', text: 'docapp-1 (pid 1001)' }); + }); + + it('does not render a picker with zero sessions — nothing to pick between', async () => { + sessionsListMock.mockResolvedValue(sessions(0)); + await mountSettled(); + expect(targetPicker()).toBeNull(); + }); + + it('defaults to broadcast: an untouched send keeps the exact pre-S5 3-argument call', async () => { + sessionsListMock.mockResolvedValue(sessions(2)); + await mountSettled(); + + await fireEvent.click(button('Push')); + await tick(); + + // No 4th argument at all — not even an explicit `undefined` — so this + // stays byte-identical to every pre-S5 broadcast call site + // (QuickActionsMenu's S4 dispatch included). + expect(pushMock).toHaveBeenCalledWith( + 'docapp', + 'fix-the-thing', + 'Take a look at TASK-5 — Fix the thing' + ); + }); + + it('selecting a session passes its id as the 4th push argument', async () => { + sessionsListMock.mockResolvedValue(sessions(2)); + await mountSettled(); + + const picker = targetPicker(); + if (!picker) throw new Error('expected a target picker'); + await fireEvent.change(picker, { target: { value: 'session-id-1' } }); + await fireEvent.click(button('Push')); + await tick(); + + expect(pushMock).toHaveBeenCalledWith( + 'docapp', + 'fix-the-thing', + 'Take a look at TASK-5 — Fix the thing', + 'session-id-1' + ); + }); + + it('a targeted miss (delivered_sessions=0) toasts, keeps the dialog open, drops back to broadcast, and re-reads presence — because zero delivery means nothing to duplicate by resending', async () => { + sessionsListMock.mockResolvedValue(sessions(2)); + pushMock.mockResolvedValueOnce({ + ref: 'TASK-5', + workspace: 'docapp', + pushed: true, + message: 'ok', + delivered_sessions: 0 + }); + const onclose = vi.fn(); + await mountSettled({ onclose }); + + const picker = targetPicker(); + if (!picker) throw new Error('expected a target picker'); + await fireEvent.change(picker, { target: { value: 'session-id-0' } }); + const presenceReadsBeforeSend = sessionsListMock.mock.calls.length; + + await fireEvent.click(button('Push')); + await tick(); + await tick(); + + expect(toastMock).toHaveBeenCalledWith('that session is gone — refresh the list', 'error'); + // Still open — unlike every OTHER successful-response outcome, a + // definitive zero-delivery result is safe to leave re-armed rather + // than closing, since nothing was actually sent to duplicate. + expect(onclose).not.toHaveBeenCalled(); + // The stale selection drops back to broadcast, and presence is + // re-read so the picker reflects who's actually still connected. + expect(targetPicker()?.value).toBe(''); + expect(sessionsListMock.mock.calls.length).toBeGreaterThan(presenceReadsBeforeSend); + }); + + it('a successful targeted push (delivered_sessions > 0) closes normally, same as broadcast', async () => { + sessionsListMock.mockResolvedValue(sessions(2)); + pushMock.mockResolvedValueOnce({ + ref: 'TASK-5', + workspace: 'docapp', + pushed: true, + message: 'ok', + delivered_sessions: 1 + }); + const onclose = vi.fn(); + await mountSettled({ onclose }); + + const picker = targetPicker(); + if (!picker) throw new Error('expected a target picker'); + await fireEvent.change(picker, { target: { value: 'session-id-0' } }); + await fireEvent.click(button('Push')); + await tick(); + await tick(); + + expect(onclose).toHaveBeenCalled(); + expect(toastMock).toHaveBeenCalledTimes(1); + const [msg] = toastMock.mock.calls[0] ?? []; + expect(String(msg)).toMatch(/isn’t confirmed|is not confirmed/); + }); + + it('a refresh that removes the selected session drops the picker back to broadcast — the next send carries no target_session_id', async () => { + vi.useFakeTimers(); + sessionsListMock.mockResolvedValue(sessions(2)); + render(PushToAgentDialog, { props: baseProps() }); + await vi.advanceTimersByTimeAsync(0); + flushSync(); + + const picker = targetPicker(); + if (!picker) throw new Error('expected a target picker'); + await fireEvent.change(picker, { target: { value: 'session-id-0' } }); + expect(targetPicker()?.value).toBe('session-id-0'); + + // The next poll's list no longer has session-id-0 — the OTHER + // session is still there, so this is a picker-visible refresh, not + // a drop to zero (which is already covered by presence-honesty's + // "session drops mid-compose" test). + sessionsListMock.mockResolvedValue({ + count: 1, + sessions: [ + { + id: 'session-id-1', + label: 'docapp-1', + pid: 1001, + connected_at: new Date(Date.now() - 60_000).toISOString() + } + ] + }); + await vi.advanceTimersByTimeAsync(10_000); + flushSync(); + + // A