fix(daemon): an older Slack reply's cutoff is the ask's first close and the first tagged answer's place

Review round 3 on the needs-me read.

- The close cutoff for an untagged older Slack reply is now the ask's first close
  (its earliest transition out of an open state), whatever its state now. Before,
  it followed the current state, so reopening an ask, or closing it again, turned
  a thread reply already excluded as correspondence into an answer.
- The first tagged answer's cutoff is now its place in the read's own order
  (ts_created, then rowid) rather than its timestamp. An older reply recorded
  just before it in the same millisecond is kept; one recorded after it isn't.

Cost: a genuine untagged answer that arrived only after an earlier close and a
reopen is no longer listed in `answers`. The ask's humanAnswers and its
transitions still carry it, and tagged answers after the reopen are listed.

docs/reference/daemon-http-clients.md

@@ -151,6 +151,40 @@ An outbox entry's `deliveryState` reports the send attempt; only `delivered` con
 an open FYI, or a closed ask is not that receipt. The rig-wide reads and their limits are described below; the existing
 `outbox/list?senderSession=<session>` read remains available.
 
+### What needs a person, and how their answers link back (`team-operations.v1`)
+
+`GET /api/team-operations/capabilities` says which team-operations jobs this daemon supports, separately from its
+build:
+
+```json
+{ "contract": "team-operations.v1",
+  "capabilities": { "needs-me.read": "supported", "needs-me.answer": "not-supported", "needs-me.follow": "not-supported" },
+  "build": { "semver": "…", "commit": "…" } }
+```
+
+A capability reads `supported`, `not-supported` or `unknown`, and a missing key means `unknown`. A daemon without the
+route (HTTP 404) predates the contract: keep reading what it already returns. In a source run, `build` is
+`{ semver: null, commit: null }`.
+
+With `needs-me.read` supported:
+
+- **What needs the person:** `GET /api/attention?view=<person-id>` (the same read as the Feed). Each queue item adds
+  `revision` and `about`. `revision` is the request's opaque decision revision: the same value as that row's
+  `decisionRevision`, and what an answer can name as `expectedRevision` so it applies only to the version the person
+  read. `about` lists the requests it concerns by id: a row parked on the person is itself the request, and an ask is
+  about the open rows blocked on it (at most 100; empty when none are recorded).
+- **A snapshot watermark:** the response's `watermark` is the newest event `seq` read before the snapshot. Follow
+  `GET /api/events` from it with `Last-Event-ID: <watermark>`: every event after the snapshot is replayed, some possibly
+  already reflected in it. Recording one answer of several emits no event yet, so re-read the snapshot (or the row)
+  rather than relying on events alone for partial answers.
+- **The answers returned to a request:** `GET /api/queue/<qitemId>` adds `answers`, oldest first, each
+  `{ qitemId, by, at, text, episode, recordedIdentity }`. They are found by the answer row's link to the request, not
+  by its text, so don't parse the `In reply to:` line in an answer's body. `text` is the person's answer without that
+  footer. `episode` names the park an answer belongs to, or is null. `recordedIdentity` is the identity provenance
+  recorded when the answer row was created. An answer row's own read has `replyTo`, the request it answers. A partial
+  set of choices stays on the request's `humanAnswers` until the whole set is returned. A runner's permission request
+  or a host pairing request is answered in place, with no answer row.
+
 ## Route details
 
 - **A rig's queue history:** `GET /api/queue/list?rig=<rig>&updatedSince=<ISO time>&sort=updated` returns the

packages/cli/src/mcp-server.ts

@@ -305,15 +305,17 @@ function projectForExternal<T>(row: T, view: ExternalView): T {
   return fields as T;
 }
 
-/** A handed-on item as a scoped caller receives it: without the daemon's own successor fields (successors,
- *  successorsTruncated, successorNote), which name rows and remote ids regardless of this caller's scope. Its answer is
- *  show's scope-checked successor (successorOf). */
-function withoutDaemonSuccessors<T>(row: T): T {
+/** An item as a scoped caller receives it: without the daemon's own fields that carry other rows regardless of this
+ *  caller's scope. Those are the successor fields (successors, successorsTruncated, successorNote), which name rows and
+ *  remote ids; a handed-on item's answer is show's scope-checked successor (successorOf). And the answers people returned
+ *  to the item (answers), which are other rows, from people, to its asker. */
+function withoutDaemonLinkedRows<T>(row: T): T {
   if (!row || typeof row !== "object" || Array.isArray(row)) return row;
   const fields = { ...(row as Record<string, unknown>) };
   delete fields.successors;
   delete fields.successorsTruncated;
   delete fields.successorNote;
+  delete fields.answers;
   return fields as T;
 }
 
@@ -1445,7 +1447,7 @@ export function createMcpServer(client: DaemonClient, options: { seat?: string;
         const view = externalView(scope);
         let successor = handedOn ? await successorOf(client, qitemId, row.closureTarget, view ? scope!.item : undefined) : undefined;
         if (scope && !view && successor && "destinationSession" in successor && !scope.successor(successor.destinationSession)) successor = { unknown: true, code: "out_of_scope" };
-        const shownItem = scope ? withoutDaemonSuccessors(item.data) : item.data;
+        const shownItem = scope ? withoutDaemonLinkedRows(item.data) : item.data;
         const data = view
           ? { item: projectForExternal(shownItem, view), transitions: Array.isArray(transitions.data) ? transitions.data.map((t) => projectForExternal(t, view)) : transitions.data }
           : { item: shownItem, transitions: transitions.data };

packages/cli/test/mcp-http-grant-service-daemon.test.ts

@@ -989,6 +989,21 @@ describe("rig mcp serve --http --grant-service on the daemon's routes", () => {
     expect(back.text).not.toContain("qitem-split-hidden");
   });
 
+  it("a scoped show of its own item never carries the answers a person returned to it; the owner's read does", async () => {
+    const served = await listen();
+    const asExternal = await connect(served.url);
+    const own = (await asExternal("rig_queue_create", { destination: "lead@team-a", body: "my task" })).structured.qitemId as string;
+    // The item asked of a person, and their answer row back to the caller: another row, from a person, linked to it.
+    db.prepare("UPDATE queue_items SET destination_session = 'human-x@external' WHERE qitem_id = ?").run(own);
+    await repo.createHumanAnswer({ sourceSession: "human-x@external", destinationSession: CALLER, body: "The person's private reasoning",
+      inboundReplyTo: own, nudge: false } as never);
+    expect(await (await fetch(`${daemonUrl}/api/queue/${own}`)).json()).toMatchObject({ answers: [{ by: "human-x@external" }] });
+    const shown = await asExternal("rig_queue_show", { qitemId: own });
+    expect(shown.isError).toBe(false);
+    expect(shown.structured.item).not.toHaveProperty("answers");
+    expect(shown.text).not.toContain("private reasoning");
+  });
+
   it("an id-reuse conflict never names another item's parties, and still names the external caller's own", async () => {
     const served = await listen();
     const asExternal = await connect(served.url);

packages/daemon/src/attention-surface.ts

@@ -25,6 +25,11 @@ export interface AttentionItem {
   /** The settings page where the person decides this request directly, without an agent (a pairing request's Approve and
    *  Deny): a TUI settings category, and the request it opens on there. */
   settingsLink?: { category: string; focus: string; label: string };
+  /** A queue request's decision revision: the same opaque token its single-row read returns and a settling act can name. */
+  revision?: string;
+  /** The requests this is about, by ID: a parked row is itself the request; an ask is about the open rows blocked on it
+   *  (at most 100). Empty when no request is recorded as waiting on it. */
+  about?: string[];
 }
 export interface AttentionDetail {
   item: AttentionItem;
@@ -38,6 +43,9 @@ export interface AttentionRead {
   sources: Array<{ source: string; state: "available" | "unavailable" | "partial"; detail: string }>;
   detail: AttentionDetail | null;
   detailError: string | null;
+  /** The event log's newest seq, read before this snapshot: following /api/events from it (Last-Event-ID) replays every
+   *  event since, possibly some already reflected here, never one missed. Absent without an event log. */
+  watermark?: number;
   /** Whose queue this Feed shows (`view`), and who acts on it. A view, never an identity: acts on what it shows carry the
    *  primary user's authority, the person at the terminal, whoever's queue it is. `requested` names a view that matched no
    *  registered person. Absent when no person is registered. */

packages/daemon/src/db/all-migrations.ts

@@ -9,6 +9,7 @@ import { queueInboxCheckinsSchema } from "./migrations/158_queue_inbox_checkins.
 import { queueItemsHandedOffFromIndexSchema } from "./migrations/159_queue_items_handed_off_from_index.js";
 import { networkBindingsSchema } from "./migrations/160_network_bindings.js";
 import { networkReceiptsSchema } from "./migrations/161_network_receipts.js";
+import { queueItemsReplyToIndexSchema } from "./migrations/164_queue_items_reply_to_index.js";
 import { activityGrantsSchema } from "./migrations/143_activity_grants.js";
 import { environmentBindingsSchema } from "./migrations/144_environment_bindings.js";
 import { nodePermissionsToSchema } from "./migrations/145_node_permissions_to.js";
@@ -325,4 +326,5 @@ export const ALL_MIGRATIONS: Migration[] = [
   queueItemsHandedOffFromIndexSchema,
   networkBindingsSchema,
   networkReceiptsSchema,
+  queueItemsReplyToIndexSchema,
 ];

packages/daemon/src/db/migrations/164_queue_items_reply_to_index.ts

@@ -0,0 +1,10 @@
+import type { Migration } from "../migrate.js";
+
+/** An ask's answers and its decision revision find rows by their `reply_to`. Without this, each lookup scans the whole
+ *  queue, once per request on the attention read. Partial: only replies carry the column. */
+export const queueItemsReplyToIndexSchema: Migration = {
+  name: "164_queue_items_reply_to_index.sql",
+  sql: `
+    CREATE INDEX IF NOT EXISTS idx_queue_items_reply_to ON queue_items(reply_to) WHERE reply_to IS NOT NULL;
+  `,
+};

packages/daemon/src/domain/queue-repository.ts

@@ -2190,6 +2190,51 @@ export class QueueRepository {
       .map((row) => ({ qitemId: row.qitem_id, body: row.body ?? "" }));
   }
 
+  /** Every answer returned to this ask, oldest first, under getHumanAnswer's rule for whoever answered: an ask to a person
+   *  (never an update), a row back to the asker linked by reply_to, from someone who may answer it (never a shared
+   *  channel). An answer row carries the human-answer tag; the unnamed primary user's own is one. An untagged Slack reply
+   *  (from before that tag) counts only when it came before the ask's first tagged answer (in this read's order) and no
+   *  later than the ask's first close, even if it was reopened since, and only each person's first: after an answer or a
+   *  close, a reply on the thread is correspondence. Each gives who, when,
+   *  their words without the footer, the park it answered (null for the ask itself or an older Slack reply), and the
+   *  identity recorded when the answer row was created. An ask answered in place (a runner's permission request, a host
+   *  pairing request) has no answer row: its answer is on the ask. */
+  answersForAsk(askId: string): Array<{ qitemId: string; by: string; at: string; text: string; episode: string | null;
+    recordedIdentity: { provenance: string | null; principal: string | null } | null }> {
+    const ask = this.getById(askId);
+    if (!this.hasReplyToColumn || !ask || ask.humanIntent === "update" || !this.isPersonAddress(ask.destinationSession)) return [];
+    const candidates = (this.db.prepare(`SELECT qitem_id, source_session, ts_created,
+        EXISTS (SELECT 1 FROM json_each(queue_items.tags) WHERE value = 'human-answer') AS tagged
+      FROM queue_items
+      WHERE reply_to = ? AND destination_session = ?
+        AND (EXISTS (SELECT 1 FROM json_each(queue_items.tags) WHERE value = 'human-answer')
+          OR (qitem_id LIKE 'qitem-slack-inbound-%' AND COALESCE(summary, '') NOT LIKE '[Recovered after gap] %'))
+      ORDER BY ts_created, rowid`).all(askId, ask.sourceSession) as Array<{ qitem_id: string; source_session: string; ts_created: string; tagged: number }>)
+      .filter((row) => !this.isSharedChannel(row.source_session)
+        && (this.answersAsk(ask, row.source_session) || (row.tagged === 1 && row.source_session === UNATTRIBUTED_ACTOR)));
+    const firstTagged = candidates.findIndex((row) => row.tagged === 1);
+    const history = this.transitionLog.listForQitem(askId);
+    // The ask's first close, whatever its state now: a reopen doesn't make later thread replies its answers.
+    const firstClose = history.find((transition) => !isBlockerLive(transition.state))?.ts;
+    const legacyFirst = new Set<string>();
+    const rows = candidates.filter((row, index) => row.tagged === 1 || ((firstTagged < 0 || index < firstTagged)
+      && (!firstClose || row.ts_created <= firstClose) && !legacyFirst.has(row.source_session) && !!legacyFirst.add(row.source_session)));
+    if (rows.length === 0) return [];
+    // An answer to a park's question has the ID made from that park's episode, so the episode is read back from the ID.
+    const episodes = new Map(history
+      .filter((transition) => transition.ownerNotificationKind === "human-required")
+      .map((transition) => `${askId}:${transition.transitionId}`)
+      .map((key) => [humanAnswerId(askId, key), key] as const));
+    return rows.flatMap(({ qitem_id }) => {
+      const row = this.getById(qitem_id);
+      if (!row) return [];
+      const created = this.transitionLog.listForQitem(qitem_id)[0];
+      return [{ qitemId: row.qitemId, by: row.sourceSession, at: row.tsCreated, text: humanAnswerText(row.body),
+        episode: episodes.get(row.qitemId) ?? null,
+        recordedIdentity: created ? { provenance: created.identityProvenance, principal: created.identityPrincipal } : null }];
+    });
+  }
+
   /** This person's answer to the current question of a decision parked on them, if one was returned. */
   currentEpisodeAnswer(qitemId: string, actorSession: string): QueueItem | null {
     const episode = this.currentHumanRequiredEpisode(qitemId);

packages/daemon/src/domain/team-operations.ts

@@ -0,0 +1,27 @@
+import { BUILD_INFO, type BuildInfo } from "../build-info.js";
+
+/** The team-operations contract an app reads before relying on a job's fields: one contract version with per-job
+ *  capabilities, distinct from the build label. A capability is supported, not-supported or unknown; an absent key reads
+ *  unknown. */
+export const TEAM_OPERATIONS_CONTRACT = "team-operations.v1";
+
+export type CapabilityValue = "supported" | "not-supported" | "unknown";
+
+/** needs-me: see what needs me, answer as me, follow it through. This build reads it (what needs the person, each
+ *  request's revision and what it's about, the answers returned to it, and a snapshot watermark). Answering under this
+ *  contract (a stable answer identity) and following a request through to its result aren't in it yet; the existing
+ *  answer paths keep working outside it. */
+export const TEAM_OPERATIONS_CAPABILITIES: Record<string, CapabilityValue> = {
+  "needs-me.read": "supported",
+  "needs-me.answer": "not-supported",
+  "needs-me.follow": "not-supported",
+};
+
+export function teamOperationsCapabilities(info: BuildInfo = BUILD_INFO) {
+  return {
+    contract: TEAM_OPERATIONS_CONTRACT,
+    capabilities: TEAM_OPERATIONS_CAPABILITIES,
+    // A source run has null build fields rather than an invented version.
+    build: { semver: info.semver, commit: info.commit },
+  };
+}

packages/daemon/src/routes/attention.ts

@@ -3,6 +3,7 @@ import * as fs from "node:fs";
 import * as path from "node:path";
 import type { AttentionRead, AttentionItem } from "../attention-surface.js";
 import type { QueueRepository } from "../domain/queue-repository.js";
+import type { EventBus } from "../domain/event-bus.js";
 import { isHumanSeatSessionRef } from "../domain/session-name.js";
 import { unansweredQuestions } from "../domain/human-questions.js";
 import type { HealthProjectionService } from "../domain/health-detectors.js";
@@ -20,6 +21,8 @@ export function attentionRoutes(): Hono {
   app.get("/", c => {
     const wanted = c.req.query("item");
     const result: AttentionRead = { scope: "instance", readAt: new Date().toISOString(), items: [], sources: [], detail: null, detailError: null };
+    const eventBus = c.get("eventBus" as never) as EventBus | undefined;
+    if (eventBus) result.watermark = eventBus.maxSeq();
     // Whether an item is in the queue this Feed shows; an item opened by id is shown in full whatever the view.
     let inView = (_recipient: string | null | undefined): boolean => true;
     const add = (item: AttentionItem, lines: () => string[], files: () => Array<{ label: string; path: string }> = () => []) => {
@@ -86,7 +89,8 @@ export function attentionRoutes(): Hono {
           ...(questions ? { humanQuestions: questions, humanAnswers: q.humanAnswers ?? {} } : {}),
           // A pairing request is also decided directly on this host's settings page, opened on this request (its row).
           ...(q.evidenceRef?.startsWith("pair-request:") ? { settingsLink: { category: "instance", focus: q.qitemId, label: "Open pairing settings (Approve or Deny directly)" } } : {}),
-          summary: q.summary || q.body.trim().split(/\r?\n/).find(Boolean) || "Request summary unavailable", urgency: q.priority, unblocks, at: q.tsUpdated, scope: p ? `project ${p.id}` : "instance · project unknown", project: p ? { id: p.id, root: p.root } : null, source: `/api/queue/${encodeURIComponent(q.qitemId)}` };
+          summary: q.summary || q.body.trim().split(/\r?\n/).find(Boolean) || "Request summary unavailable", urgency: q.priority, unblocks, at: q.tsUpdated, scope: p ? `project ${p.id}` : "instance · project unknown", project: p ? { id: p.id, root: p.root } : null, source: `/api/queue/${encodeURIComponent(q.qitemId)}`,
+          revision: queue.decisionRevision(q), about: toPerson ? deps.slice(0, 100).map(d => d.qitem_id) : [q.qitemId] };
         const unbound = noConnector(toPerson ? q.destinationSession : q.blockedOn);
         const notSent = unbound || quiet.has(q.qitemId);
         const route = unanswered

packages/daemon/src/routes/queue.ts

@@ -1546,6 +1546,8 @@ export function queueRoutes(): Hono {
       ...item,
       // #2446: what a settling Mission Control act can be guarded with (expectedRevision).
       decisionRevision: repo.decisionRevision(item),
+      // The answer rows returned to this row, read by their link rather than their text; empty when none came back.
+      answers: repo.answersForAsk(qitemId),
       // A closed row's outcome (its note and reason) lives only in the transition log; omitted while the row is open.
       ...(closing ? { closingTransition: {
         transitionId: closing.transitionId,

packages/daemon/src/server.ts

@@ -38,6 +38,7 @@ import type { ClaimService } from "./domain/claim-service.js";
 import type { SelfAttachService } from "./domain/self-attach-service.js";
 import { rigsRoutes } from "./routes/rigs.js";
 import { OPERATION_CONTRACT, OPERATION_KINDS } from "./domain/operation-correlation.js";
+import { teamOperationsCapabilities } from "./domain/team-operations.js";
 import { sessionsRoutes, nodesRoutes, sessionAdminRoutes } from "./routes/sessions.js";
 import { adaptersRoutes } from "./routes/adapters.js";
 import { eventsRoute } from "./routes/events.js";
@@ -795,6 +796,7 @@ export function createApp(deps: AppDeps): Hono {
 
   app.route("/api/rigs", rigsRoutes);
   app.get("/api/operations/capabilities", (c) => c.json({ contract: OPERATION_CONTRACT, kinds: OPERATION_KINDS }));
+  app.get("/api/team-operations/capabilities", (c) => c.json(teamOperationsCapabilities()));
   app.route("/api/rigs/:rigId/sessions", sessionsRoutes);
   // Slice 24 — per-rig CMUX workspace launcher.
   app.route("/api/rigs/:rigId/cmux", rigCmuxRoutes);

packages/daemon/test/helpers/test-app.ts

@@ -6,6 +6,7 @@ import { queueInboxCheckinsSchema } from "../../src/db/migrations/158_queue_inbo
 import { queueItemsHandedOffFromIndexSchema } from "../../src/db/migrations/159_queue_items_handed_off_from_index.js";
 import { networkBindingsSchema } from "../../src/db/migrations/160_network_bindings.js";
 import { networkReceiptsSchema } from "../../src/db/migrations/161_network_receipts.js";
+import { queueItemsReplyToIndexSchema } from "../../src/db/migrations/164_queue_items_reply_to_index.js";
 import { activityGrantsSchema } from "../../src/db/migrations/143_activity_grants.js";
 import { environmentBindingsSchema } from "../../src/db/migrations/144_environment_bindings.js";
 import { nodePermissionsToSchema } from "../../src/db/migrations/145_node_permissions_to.js";
@@ -168,7 +169,7 @@ import fs from "node:fs";
 
 /** Seam B R6: the canonical full-fixture migration list, exported so file-backed
  *  DB-reopen tests migrate IDENTICALLY to createFullTestDb. */
-export const migrationsForFullTestDb = [coreSchema, bindingsSessionsSchema, eventsSchema, snapshotsSchema, checkpointsSchema, resumeMetadataSchema, nodeSpecFieldsSchema, packagesSchema, installJournalSchema, journalSeqSchema, bootstrapSchema, discoverySchema, discoveryFkFix, agentspecRebootSchema, startupContextSchema, chatMessagesSchema, podNamespaceSchema, contextUsageSchema, externalCliAttachmentSchema, rigServicesSchema, seatHandoverObservabilitySchema, nodeCodexConfigProfileSchema, nodePermissionPolicySchema, rigPermissionPolicySchema, nodePolicyProvenanceSchema, rigPolicyProvenanceSchema, streamItemsSchema, queueItemsSchema, queueTransitionsSchema, rigPolicySchema, rigArchiveSchema, resumeProvenanceSchema, resumeVerificationSchema, seatIdentityVerdictsSchema, selfHostIdentitySchema, occupantTenuresSchema, daemonLifecycleSchema, watchdogJobsSchema, occupantGenerationStampsSchema, projectionManifestSchema, watchdogTargetGenerationSchema, appliedLaunchObservationsSchema, appliedLaunchObservationInvalidationsSchema, threadSeatMapSchema, queueTransitionWakesSchema, nodeSessionSourceSchema, scopedOperatingPostureSchema, humanNotificationIntentSchema, reviewReadIndexesSchema, inventoryEventIndexesSchema, rigClaudeManagedBlockFileSchema, nodePermissionSelectionsSchema, humanReplyToSchema, humanQuestionsSchema, nodeEffortSchema, rigNonInterruptiveSchema, rigInstallRootSchema, threadPartMapSchema, queueTransitionsQitemIdOrderSchema, queueItemsRecoveryTagsSchema, resumeRotationSchema, sessionStopIntentSchema, nodeEffortChoiceSchema, modePostureCoexistSchema, seatEnvSchema, appliedLaunchDaemonOptOutSchema, sessionCommunityHarnessSchema, threadSeatMapChannelKeySchema, queueItemWakeInputBoxSchema, providerAccountReadingsSchema, humanQuietWindowsSchema, seatIdentityPanePidSchema, seatIdentityProcessEvidenceSchema, rigStartupLayersSchema, humanOperatingModesSchema, queueDelegatedDecisionSchema, nodeAdvisorSchema, operationCorrelationSchema, terminalPolicySchema, seatRuntimeSwitchesSchema, claudeTokenReadingsSchema, nodeLaunchRecipeSchema, nodeClaudeTokenRefSchema, nodeClaudePermissionsSchema, nodeCodexPermissionsSchema, escalationTargetRefsSchema, sessionAgentRunnerSchema, nodeAccountHomeSchema, nodeClaudeMcpSchema, activityGrantsSchema, environmentBindingsSchema, nodePermissionsToSchema, nodeStartupGuidanceFileSchema, managedBlockWritesSchema, streamIntakeCursorSchema, nodeLaunchWrapperSchema, piAccountSwitchesSchema, eventsRetentionMarksSchema, queueErasureSchema, queueErasureBatchesSchema, queueInboxCheckinsSchema, queueItemsHandedOffFromIndexSchema, networkBindingsSchema, networkReceiptsSchema];
+export const migrationsForFullTestDb = [coreSchema, bindingsSessionsSchema, eventsSchema, snapshotsSchema, checkpointsSchema, resumeMetadataSchema, nodeSpecFieldsSchema, packagesSchema, installJournalSchema, journalSeqSchema, bootstrapSchema, discoverySchema, discoveryFkFix, agentspecRebootSchema, startupContextSchema, chatMessagesSchema, podNamespaceSchema, contextUsageSchema, externalCliAttachmentSchema, rigServicesSchema, seatHandoverObservabilitySchema, nodeCodexConfigProfileSchema, nodePermissionPolicySchema, rigPermissionPolicySchema, nodePolicyProvenanceSchema, rigPolicyProvenanceSchema, streamItemsSchema, queueItemsSchema, queueTransitionsSchema, rigPolicySchema, rigArchiveSchema, resumeProvenanceSchema, resumeVerificationSchema, seatIdentityVerdictsSchema, selfHostIdentitySchema, occupantTenuresSchema, daemonLifecycleSchema, watchdogJobsSchema, occupantGenerationStampsSchema, projectionManifestSchema, watchdogTargetGenerationSchema, appliedLaunchObservationsSchema, appliedLaunchObservationInvalidationsSchema, threadSeatMapSchema, queueTransitionWakesSchema, nodeSessionSourceSchema, scopedOperatingPostureSchema, humanNotificationIntentSchema, reviewReadIndexesSchema, inventoryEventIndexesSchema, rigClaudeManagedBlockFileSchema, nodePermissionSelectionsSchema, humanReplyToSchema, humanQuestionsSchema, nodeEffortSchema, rigNonInterruptiveSchema, rigInstallRootSchema, threadPartMapSchema, queueTransitionsQitemIdOrderSchema, queueItemsRecoveryTagsSchema, resumeRotationSchema, sessionStopIntentSchema, nodeEffortChoiceSchema, modePostureCoexistSchema, seatEnvSchema, appliedLaunchDaemonOptOutSchema, sessionCommunityHarnessSchema, threadSeatMapChannelKeySchema, queueItemWakeInputBoxSchema, providerAccountReadingsSchema, humanQuietWindowsSchema, seatIdentityPanePidSchema, seatIdentityProcessEvidenceSchema, rigStartupLayersSchema, humanOperatingModesSchema, queueDelegatedDecisionSchema, nodeAdvisorSchema, operationCorrelationSchema, terminalPolicySchema, seatRuntimeSwitchesSchema, claudeTokenReadingsSchema, nodeLaunchRecipeSchema, nodeClaudeTokenRefSchema, nodeClaudePermissionsSchema, nodeCodexPermissionsSchema, escalationTargetRefsSchema, sessionAgentRunnerSchema, nodeAccountHomeSchema, nodeClaudeMcpSchema, activityGrantsSchema, environmentBindingsSchema, nodePermissionsToSchema, nodeStartupGuidanceFileSchema, managedBlockWritesSchema, streamIntakeCursorSchema, nodeLaunchWrapperSchema, piAccountSwitchesSchema, eventsRetentionMarksSchema, queueErasureSchema, queueErasureBatchesSchema, queueInboxCheckinsSchema, queueItemsHandedOffFromIndexSchema, networkBindingsSchema, networkReceiptsSchema, queueItemsReplyToIndexSchema];
 
 /**
  * P24 — the DECLARED exclusions for {@link migrationsForFullTestDb}. That list is deliberately a

packages/daemon/test/needs-me-read.test.ts

@@ -0,0 +1,207 @@
+// team-operations.v1, needs-me.read: what needs the person, each request's revision and what it's about, and the answers
+// returned to it, read by their links rather than by searching answer text. Real SQLite (all migrations), the real queue,
+// the Mission Control answer path, the queue's single-row read and the attention read. Nothing native.
+import { afterEach, describe, expect, it } from "vitest";
+import Database from "better-sqlite3";
+import { Hono } from "hono";
+import { migrate } from "../src/db/migrate.js";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { QueueRepository } from "../src/domain/queue-repository.js";
+import { MissionControlWriteContract } from "../src/domain/mission-control/mission-control-write-contract.js";
+import { MissionControlActionLog } from "../src/domain/mission-control/mission-control-action-log.js";
+import { missionControlRoutes } from "../src/routes/mission-control.js";
+import { queueRoutes } from "../src/routes/queue.js";
+import { attentionRoutes } from "../src/routes/attention.js";
+import type { AttentionRead } from "../src/attention-surface.js";
+import { teamOperationsCapabilities } from "../src/domain/team-operations.js";
+
+const PERSON = "human-test@external", ASKER = "requester@test-rig";
+const registry = { ok: true as const, entities: [{ entityId: "human-test", class: "human" as const, displayName: "human-test",
+  address: PERSON, connectorBindings: [], prefs: { deliveryClass: "A" as const } }] };
+const QUESTIONS = [
+  { id: "q1", question: "Ship it?", options: [{ id: "yes", label: "Yes" }, { id: "no", label: "No" }] },
+  { id: "q2", question: "Tell the team?", options: [{ id: "yes", label: "Yes" }, { id: "no", label: "No" }] },
+];
+
+const closes: Array<() => void> = [];
+afterEach(() => { for (const close of closes.splice(0)) close(); });
+
+function fixture() {
+  const db = new Database(":memory:"); migrate(db, ALL_MIGRATIONS); const bus = new EventBus(db);
+  const repo = new QueueRepository(db, bus, { validateRig: () => true, loadHumanRegistry: () => registry,
+    transport: { send: async () => ({ ok: true, verified: true, sent: true }) } });
+  const contract = new MissionControlWriteContract({ db, eventBus: bus, queueRepo: repo, actionLog: new MissionControlActionLog(db) });
+  const app = new Hono();
+  app.use("*", async (c, next) => {
+    c.set("queueRepo" as never, repo as never); c.set("eventBus" as never, bus as never);
+    c.set("missionControlWriteContract" as never, contract as never);
+    c.set("settingsStore" as never, { resolveOne: () => ({ value: "/nonexistent-needs-me-fixture" }) } as never);
+    await next();
+  });
+  app.route("/api/mission-control", missionControlRoutes());
+  app.route("/api/queue", queueRoutes());
+  app.route("/api/attention", attentionRoutes());
+  closes.push(() => db.close());
+  const ask = async () => (await repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "Synthetic two-question ask",
+    summary: "Ship?", humanIntent: "decision", humanQuestions: QUESTIONS, nudge: false })).qitemId;
+  const read = async (id: string) => (await app.request(`/api/queue/${id}`)).json() as Promise<Record<string, any>>;
+  const attention = async () => (await app.request("/api/attention")).json() as Promise<AttentionRead>;
+  const act = async (body: Record<string, unknown>) => {
+    const response = await app.request("/api/mission-control/action", { method: "POST",
+      headers: { "Content-Type": "application/json", "X-OpenRig-Session": PERSON }, body: JSON.stringify({ actorSession: PERSON, ...body }) });
+    return { status: response.status, body: await response.json() as Record<string, any> };
+  };
+  return { db, bus, repo, ask, read, attention, act };
+}
+
+describe("needs-me.read: answers on the ask, by link", () => {
+  it("an ask lists no answer until its set is returned, then the one answer row, with who, when and the words", async () => {
+    const f = fixture(); const id = await f.ask();
+    expect((await f.read(id)).answers).toEqual([]);
+    expect(await f.act({ verb: "answer", qitemId: id, questionId: "q1", optionId: "yes" })).toMatchObject({ status: 200 });
+    // A partial answer is recorded on the ask (humanAnswers); nothing has been returned to the asker yet.
+    expect(await f.read(id)).toMatchObject({ humanAnswers: { q1: "yes" }, answers: [] });
+    const done = await f.act({ verb: "answer", qitemId: id, questionId: "q2", optionId: "no" });
+    expect(done).toMatchObject({ status: 200 });
+    const { answers } = await f.read(id);
+    expect(answers).toEqual([{ qitemId: done.body.createdQitemId, by: PERSON, at: expect.any(String),
+      text: "Ship it?: Yes\nTell the team?: No", episode: null, recordedIdentity: expect.anything() }]);
+    // The other direction is the answer row's own link.
+    expect(await f.read(answers[0].qitemId)).toMatchObject({ replyTo: id, answers: [] });
+  });
+
+  it("an answer in the person's words reads back as those words, without the footer", async () => {
+    const f = fixture(); const id = await f.ask();
+    expect(await f.act({ verb: "answer", qitemId: id, decision: "Hold until Friday" })).toMatchObject({ status: 200 });
+    expect((await f.read(id)).answers).toMatchObject([{ by: PERSON, text: "Hold until Friday" }]);
+  });
+
+  it("names the park an answer belongs to when the ask is parked on its person", async () => {
+    const f = fixture(); const id = await f.ask();
+    f.repo.update({ qitemId: id, actorSession: ASKER, state: "blocked", blockedOn: PERSON, summary: "Still need your call",
+      evidenceRef: "/fixture/decision", transitionNote: "parked on its person" });
+    const episode = f.repo.currentHumanRequiredEpisode(id)!.key;
+    // As the Slack path records a parked ask's answer: under that park's question.
+    const parked = await f.repo.createHumanAnswer({ sourceSession: PERSON, destinationSession: ASKER, body: "Go ahead\n\n---\nIn reply to: x",
+      tags: ["human-answer"], inboundReplyTo: id, episodeKey: episode, nudge: false });
+    expect((await f.read(id)).answers).toMatchObject([{ qitemId: parked.qitemId, by: PERSON, text: "Go ahead", episode }]);
+  });
+
+  it("counts only real answers: not correspondence, another reply after the answer, a reply to an update, or a row to someone else", async () => {
+    const f = fixture();
+    const legacy = (qitemId: string, replyTo: string, source: string, destination: string, at = new Date().toISOString()) => {
+      f.db.prepare("INSERT INTO queue_items(qitem_id,ts_created,ts_updated,source_session,destination_session,state,priority,body,reply_to,tags) VALUES(?,?,?,?,?,?,?,?,?,?)")
+        .run(qitemId, at, at, source, destination, "pending", "routine", `Words in ${qitemId}`, replyTo, "[]");
+    };
+    // An older Slack reply from before the answer tag: the asked person's first one counts, and a second one doesn't.
+    const old = await f.ask();
+    legacy("qitem-slack-inbound-old-1", old, PERSON, ASKER);
+    legacy("qitem-slack-inbound-old-2", old, PERSON, ASKER);
+    // A reply from someone who can't answer the ask, and one sent to someone other than the asker.
+    legacy("qitem-slack-inbound-agent", old, "owner@test-rig", ASKER);
+    legacy("qitem-slack-inbound-elsewhere", old, PERSON, "owner@test-rig");
+    expect((await f.read(old)).answers).toEqual([expect.objectContaining({ qitemId: "qitem-slack-inbound-old-1", episode: null })]);
+    // Once a tagged answer exists, a later untagged reply on the ask is correspondence, not a second answer.
+    const id = await f.ask();
+    const done = await f.act({ verb: "answer", qitemId: id, decision: "Ship it" });
+    legacy("qitem-slack-inbound-later", id, PERSON, ASKER);
+    expect((await f.read(id)).answers).toEqual([expect.objectContaining({ qitemId: done.body.createdQitemId, text: "Ship it" })]);
+    // Replies posted to an update's thread are never its answers.
+    const update = (await f.repo.create({ sourceSession: ASKER, destinationSession: PERSON, body: "FYI", humanIntent: "update", nudge: false })).qitemId;
+    legacy("qitem-slack-inbound-on-update", update, PERSON, ASKER);
+    expect((await f.read(update)).answers).toEqual([]);
+    // Closed with no answer row (answered in place), then a reply on its thread: correspondence, not its answer.
+    const inPlace = await f.ask();
+    f.repo.update({ qitemId: inPlace, actorSession: PERSON, state: "done", closureReason: "no-follow-on", transitionNote: "decided in place" });
+    legacy("qitem-slack-inbound-after-close", inPlace, PERSON, ASKER, new Date(Date.now() + 60_000).toISOString());
+    expect((await f.read(inPlace)).answers).toEqual([]);
+    // Reopening the ask, or closing it again later, doesn't make that reply its answer.
+    f.repo.update({ qitemId: inPlace, actorSession: ASKER, state: "pending", reopen: true, transitionNote: "reopened to ask again" });
+    expect((await f.read(inPlace)).answers).toEqual([]);
+    f.repo.update({ qitemId: inPlace, actorSession: PERSON, state: "done", closureReason: "no-follow-on", transitionNote: "decided again" });
+    expect((await f.read(inPlace)).answers).toEqual([]);
+  });
+
+  it("at the same time as the first tagged answer, an older reply counts only when it was recorded first", async () => {
+    const f = fixture();
+    const at = new Date(Date.now() - 60_000).toISOString();
+    const insert = (qitemId: string, askId: string, tags: string) => f.db.prepare("INSERT INTO queue_items(qitem_id,ts_created,ts_updated,source_session,destination_session,state,priority,body,reply_to,tags) VALUES(?,?,?,?,?,?,?,?,?,?)")
+      .run(qitemId, at, at, PERSON, ASKER, "pending", "routine", `Words in ${qitemId}`, askId, tags);
+    const legacyFirst = await f.ask();
+    insert("qitem-slack-inbound-tie-first", legacyFirst, "[]");
+    insert("qitem-tagged-tie-second", legacyFirst, '["human-answer"]');
+    expect((await f.read(legacyFirst)).answers.map((a: { qitemId: string }) => a.qitemId)).toEqual(["qitem-slack-inbound-tie-first", "qitem-tagged-tie-second"]);
+    const taggedFirst = await f.ask();
+    insert("qitem-tagged-tie-first", taggedFirst, '["human-answer"]');
+    insert("qitem-slack-inbound-tie-second", taggedFirst, "[]");
+    expect((await f.read(taggedFirst)).answers.map((a: { qitemId: string }) => a.qitemId)).toEqual(["qitem-tagged-tie-first"]);
+  });
+
+  it("keeps a real older Slack answer that came before a re-park's tagged answer, and lists both", async () => {
+    const f = fixture(); const id = await f.ask();
+    const at = new Date(Date.now() - 60_000).toISOString();
+    f.db.prepare("INSERT INTO queue_items(qitem_id,ts_created,ts_updated,source_session,destination_session,state,priority,body,reply_to,tags) VALUES(?,?,?,?,?,?,?,?,?,?)")
+      .run("qitem-slack-inbound-before-tag", at, at, PERSON, ASKER, "pending", "routine", "Yes, ship it", id, "[]");
+    // That answer closed the ask; later it was deliberately reopened onto its person, and answered again under the tag.
+    f.repo.update({ qitemId: id, actorSession: ASKER, state: "done", closureReason: "no-follow-on", transitionNote: "answered in Slack" });
+    f.repo.update({ qitemId: id, actorSession: ASKER, state: "blocked", blockedOn: PERSON, reopen: true, summary: "One more call",
+      evidenceRef: "/fixture/decision", transitionNote: "reopened for a follow-up decision" });
+    const episode = f.repo.currentHumanRequiredEpisode(id)!.key;
+    const parked = await f.repo.createHumanAnswer({ sourceSession: PERSON, destinationSession: ASKER, body: "Also tell the team",
+      tags: ["human-answer"], inboundReplyTo: id, episodeKey: episode, nudge: false });
+    expect((await f.read(id)).answers).toMatchObject([
+      { qitemId: "qitem-slack-inbound-before-tag", text: "Yes, ship it", episode: null },
+      { qitemId: parked.qitemId, text: "Also tell the team", episode },
+    ]);
+  });
+
+  it("finds an ask's replies through the reply_to index, not a scan of the queue", async () => {
+    const f = fixture(); const id = await f.ask();
+    const replyToQueries = new Set<string>();
+    const prepare = f.db.prepare.bind(f.db);
+    f.db.prepare = ((sql: string) => {
+      if (/reply_to = \?/.test(sql)) replyToQueries.add(sql);
+      return prepare(sql);
+    }) as typeof f.db.prepare;
+    await f.read(id);
+    await f.attention();
+    f.db.prepare = prepare;
+    // The answer lookup and the decision revision's reply lookups, as the reads actually ran them.
+    expect(replyToQueries.size).toBeGreaterThanOrEqual(2);
+    for (const sql of replyToQueries) {
+      const params = (sql.match(/\?/g) ?? []).map(() => "x");
+      const plan = (f.db.prepare(`EXPLAIN QUERY PLAN ${sql}`).all(...params) as Array<{ detail: string }>).map((row) => row.detail).join("\n");
+      expect(plan, sql).toMatch(/SEARCH queue_items USING INDEX idx_queue_items_reply_to/);
+    }
+  });
+});
+
+describe("needs-me.read: the attention snapshot", () => {
+  it("gives each request its revision and what it's about, and a watermark read before the snapshot", async () => {
+    const f = fixture(); const id = await f.ask();
+    const waiting = await f.repo.create({ sourceSession: ASKER, destinationSession: "owner@test-rig", body: "Waits on the ask", nudge: false });
+    f.repo.claim({ qitemId: waiting.qitemId, destinationSession: "owner@test-rig" });
+    f.repo.update({ qitemId: waiting.qitemId, actorSession: "owner@test-rig", state: "blocked", blockedOn: id });
+    const park = await f.repo.create({ sourceSession: ASKER, destinationSession: "owner@test-rig", body: "Parked on the person", nudge: false });
+    f.repo.claim({ qitemId: park.qitemId, destinationSession: "owner@test-rig" });
+    f.repo.update({ qitemId: park.qitemId, actorSession: "owner@test-rig", state: "blocked", blockedOn: PERSON });
+    const before = f.bus.maxSeq();
+    const snapshot = await f.attention();
+    expect(snapshot.watermark).toBe(before);
+    const item = (qitemId: string) => snapshot.items.find((i) => i.id === `queue:${qitemId}`);
+    expect(item(id)).toMatchObject({ revision: (await f.read(id)).decisionRevision, about: [waiting.qitemId] });
+    expect(item(park.qitemId)).toMatchObject({ revision: (await f.read(park.qitemId)).decisionRevision, about: [park.qitemId] });
+  });
+});
+
+describe("team-operations.v1 capabilities", () => {
+  it("names the contract, needs-me's capabilities and the build label separately", () => {
+    expect(teamOperationsCapabilities({ semver: "9.9.9", commit: "abc", dirty: false, builtAt: null })).toEqual({
+      contract: "team-operations.v1",
+      capabilities: { "needs-me.read": "supported", "needs-me.answer": "not-supported", "needs-me.follow": "not-supported" },
+      build: { semver: "9.9.9", commit: "abc" },
+    });
+    expect(teamOperationsCapabilities().build).toEqual({ semver: null, commit: null });
+  });
+});