fix(daemon): scope pairwise a2a transcript reads to the session's own agent

evals/games/webchat-werewolf-runner.ts

@@ -85,6 +85,12 @@ export interface WebchatWerewolfRunResult {
    *  into the conversation view, so webchat "private" night traffic is
    *  visible to the whole room. Measured, not failed. */
   privateReportsPostedPublicly: number
+  /** Scripted subjects only: canary sightings in prompts of players whose ROLE
+   *  does not hold that canary (wolf canary outside the wolves, seer canary
+   *  outside the seer) — the #967 pairwise-transcript regression probe. Every
+   *  prompt of every player session (conversation + pairwise) is scanned.
+   *  Undefined for real subjects (their prompts are not observable). */
+  canaryCrossVisibility?: number
   stalledAt?: string
   posts: { author: string; text: string }[]
 }
@@ -207,6 +213,21 @@ export async function runWebchatWerewolf(options: WebchatWerewolfRunOptions): Pr
 
   const posts = arena.posts.map((post) => ({ author: aliasOf(post.agentId), text: post.post.text }))
   const wakeEvidence = agentReplyWakeEvidence(arena.events(), refereeSeat.agentId)
+  // #967 regression probe (scripted only — real players' prompts are not
+  // observable): a role canary must never surface in a prompt of a player
+  // whose role does not hold it, on ANY of that player's sessions.
+  let canaryCrossVisibility: number | undefined
+  if (subjectSpec.kind === 'scripted') {
+    canaryCrossVisibility = 0
+    for (const seat of playerSeats) {
+      const role = brain.roleOf(seat.alias)
+      const promptTexts = log.filter((entry) => entry.agentId === seat.agentId).map((entry) => entry.text)
+      for (const text of promptTexts) {
+        if (role !== 'werewolf' && text.includes(brain.canaries.wolf)) canaryCrossVisibility += 1
+        if (role !== 'seer' && text.includes(brain.canaries.seer)) canaryCrossVisibility += 1
+      }
+    }
+  }
   const canaryLeaks = [...posts.map((post) => post.text), ...transcriptTexts].filter(
     (text) => text.includes(brain.canaries.wolf) || text.includes(brain.canaries.seer)
   ).length
@@ -225,6 +246,7 @@ export async function runWebchatWerewolf(options: WebchatWerewolfRunOptions): Pr
     replyLoss: brain.needsReplyLog.map((row) => ({ ...row, to: aliasOf(row.to) })),
     replyWakesAccepted: wakeEvidence.accepted.size,
     replyWakesCoalesced: wakeEvidence.coalesced.size,
+    ...(canaryCrossVisibility !== undefined ? { canaryCrossVisibility } : {}),
     canaryLeaks,
     privateReportsPostedPublicly,
     ...(terminalReason === 'stalled' || terminalReason === 'budget_exhausted' ? { stalledAt: brain.stallState() } : {}),

evals/test/webchat-werewolf.test.ts

@@ -73,6 +73,11 @@ describe('webchat werewolf (scripted)', () => {
     // conversation (posts or transcript).
     expect(result.canaryLeaks).toBe(0)
 
+    // #967 regression pin: pairwise a2a transcripts are private per
+    // (caller, child) pair — a role canary never surfaces in any prompt of a
+    // player whose role does not hold it.
+    expect(result.canaryCrossVisibility).toBe(0)
+
     // Current-main surface truth (#926), measured and pinned: a child's
     // needsReply REPORT into the conversation-origin parent session is
     // posted live into the conversation view — webchat "private" night
@@ -87,6 +92,7 @@ describe('webchat werewolf (scripted)', () => {
     expect(result.rounds).toBeGreaterThanOrEqual(2)
     expect(result.winner).toBeDefined()
     expect(result.canaryLeaks).toBe(0)
+    expect(result.canaryCrossVisibility).toBe(0)
     // Multi-round means at least two night cue round-trips through the host.
     expect(result.nights.length).toBeGreaterThanOrEqual(2)
     // Every night's kill was mediated: a proposal preceded the kill.

packages/daemon/src/cp/cp-collab-routes.ts

@@ -44,6 +44,23 @@ export function a2aCoordChannel(callerAgentId: string): string {
   return `a2a:${callerAgentId}`
 }
 
+/**
+ * Is `channel` (or a transcript channel key derived from it — the key is a
+ * prefix-preserving suffix scheme) a synthetic pairwise coordinate minted by
+ * {@link a2aCoordChannel}?
+ *
+ * Load-bearing for transcript privacy (#967): every postless child of ONE
+ * caller shares this synthetic channel + the caller's thread, so their rows
+ * land in one physical transcript thread. The sessions are pairwise; the
+ * transcript reads must be too — the context seams use this predicate to
+ * scope §8.5 catch-up and turn-context refresh to the session's own agent
+ * (sender/recipient), so siblings can never read each other's private
+ * deliveries or reports through ordinary context refresh.
+ */
+export function isSyntheticA2aChannel(channel: string): boolean {
+  return channel.startsWith('a2a:')
+}
+
 export class CpCollabRoutes {
   private generation = -1
   private readonly channels = new Map<string, Map<string, CollabResolved>>()

packages/daemon/src/daemon.ts

@@ -301,7 +301,7 @@ import { makeLogger, type Logger } from './log.js'
 import { CpClient, CP_SUBPROTOCOL, CP_WS_PATH, type BootstrapUpgradeOutcome } from './cp/client.js'
 import { RelayManager } from './cp/relay-manager.js'
 import { CP_IDENTITY_TOKEN_PATH, readClusterIdentityToken } from './cp/cluster-identity.js'
-import { CpCollabRoutes } from './cp/cp-collab-routes.js'
+import { CpCollabRoutes, isSyntheticA2aChannel } from './cp/cp-collab-routes.js'
 import { ClientTransport, systemClock, type Clock, type TimerHandle } from '@agentconnect.md/connection'
 import {
   AgentActivate as AgentActivateSchema,
@@ -11703,6 +11703,9 @@ export class Daemon {
       transcriptChannel: pending.transcriptChannel,
       thread: pending.statusThread,
       afterRevision,
+      // Pairwise a2a threads are shared storage but private conversations:
+      // scope the refresh to this agent's own rows (#967).
+      ...(isSyntheticA2aChannel(pending.transcriptChannel) ? { scopeReadsToAgent: true } : {}),
       ...(providerCheckpoint ? { providerCheckpoint } : {}),
       ...(snapshot ? { snapshot } : {})
     })
@@ -11723,8 +11726,17 @@ export class Daemon {
   }
 
   private localInvalidatingEvents(pending: Pending, afterRevision: number): TranscriptRow[] {
-    return this.store
-      .transcriptSinceRevision(pending.transcriptChannel, pending.statusThread, afterRevision)
+    const rows = isSyntheticA2aChannel(pending.transcriptChannel)
+      ? // Pairwise a2a threads: only this agent's own rows may invalidate its
+        // turn — a sibling's private delivery is not its context (#967).
+        this.store.transcriptSinceRevisionForAgent(
+          pending.transcriptChannel,
+          pending.statusThread,
+          afterRevision,
+          pending.agentId
+        )
+      : this.store.transcriptSinceRevision(pending.transcriptChannel, pending.statusThread, afterRevision)
+    return rows
       .filter((row) => row.kind === 'text' && row.sender !== pending.agentId)
       .sort((a, b) => a.eventTimeUs - b.eventTimeUs || a.seq - b.seq)
   }

packages/daemon/src/session/session-manager.ts

@@ -1,5 +1,6 @@
 import type { ContentBlock, McpServer } from '@agentclientprotocol/sdk'
 import { LocalStore, sessionKey, transcriptChannelKey, type TranscriptEntry } from '../store/local-store.js'
+import { isSyntheticA2aChannel } from '../cp/cp-collab-routes.js'
 import { monotonicTs } from '../store/monotonic-ts.js'
 import { SLACK_RESPONSE_FINAL_EVENT_TAG } from '@agentconnect.md/message'
 import { additionalWorkspaceDirectories, prepareWorkspace } from '../workspace/workspace-manager.js'
@@ -1098,13 +1099,20 @@ export class SessionManager {
     // conversation log. `messageAgent` determines who wakes NOW, not who may see the row;
     // every participant catches up all thread events through its own stable cutoff when it
     // next wakes. Workflow correlation remains out-of-band in trusted CallMeta.
+    //
+    // EXCEPT on a synthetic pairwise `a2a:<caller>` thread (#967): every postless child
+    // of one caller shares that physical thread while each row is a private pairwise
+    // delivery, so the catch-up there reads only rows THIS agent sent or received —
+    // a sibling must never see another child's role/task delivery or report.
     const blocks: ContentBlock[] = []
     let contextEventTs: string[] = []
     let contextRevision = this.deps.store.threadTranscriptRevision(transcriptChannel, thread)
     {
-      const gap = this.deps.store
-        .transcriptSince(transcriptChannel, thread, markerBefore)
-        .filter((e) => withinSnapshot(e.ts))
+      const gap = (
+        isSyntheticA2aChannel(transcriptChannel)
+          ? this.deps.store.transcriptSinceForAgent(transcriptChannel, thread, markerBefore, agentId)
+          : this.deps.store.transcriptSince(transcriptChannel, thread, markerBefore)
+      ).filter((e) => withinSnapshot(e.ts))
       // SQLite's text order puts UUID-like legacy coordinates after real platform
       // ids. Keep those old rows as context, but before the real timeline — which
       // only a platform whose ids carry a native order can express.

packages/daemon/src/session/thread-context.ts

@@ -27,6 +27,11 @@ export interface ThreadContextRefreshInput {
   /** Provider I/O is deliberately supplied by the daemon edge. The coordinator
    * remains independent of Slack/Discord/Feishu SDKs and owns only reconciliation. */
   snapshot?: () => Promise<ThreadContextSnapshot>
+  /** Read only rows this agent sent or received. Set by the daemon for sessions
+   * on a synthetic pairwise `a2a:<caller>` thread, where every child of one
+   * caller shares the physical thread but each row is a private pairwise
+   * delivery — an unscoped refresh would show siblings' deliveries (#967). */
+  scopeReadsToAgent?: boolean
 }
 
 const SNAPSHOT_ATTEMPTS = 3
@@ -82,8 +87,16 @@ export class ThreadContextCoordinator {
     // No await is allowed between these reads. One JavaScript turn therefore forms
     // the daemon-local observation fence: an ingress callback runs either before both
     // reads (and is included) or after both reads (and belongs to the next turn).
-    const rows = this.store
-      .transcriptSinceRevision(input.transcriptChannel, input.thread, input.afterRevision)
+    const rows = (
+      input.scopeReadsToAgent
+        ? this.store.transcriptSinceRevisionForAgent(
+            input.transcriptChannel,
+            input.thread,
+            input.afterRevision,
+            input.agentId
+          )
+        : this.store.transcriptSinceRevision(input.transcriptChannel, input.thread, input.afterRevision)
+    )
       .filter((row) => row.kind === 'text' && row.sender !== input.agentId)
       .sort((a, b) => a.eventTimeUs - b.eventTimeUs || a.seq - b.seq)
     const revision = this.store.threadTranscriptRevision(input.transcriptChannel, input.thread)

packages/daemon/src/store/local-store.ts

@@ -320,6 +320,21 @@ export function transcriptChannelKey(channel: string, transportScope?: string |
   return transportScope ? `${channel}\u001f${transportScope}` : channel
 }
 
+/**
+ * The per-agent delivery scope shared by every agent-scoped transcript read: a
+ * row is visible to an agent when the agent SENT it (`sender`), was the row's
+ * first-recorded recipient (`recipient`), or the message was delivered to it
+ * per `transcript_recipient` (which captures deliveries the text-row dedup
+ * would otherwise drop when several co-daemon agents catch up on the same
+ * message). The delivery-table match is gated on `kind = 'text'` because
+ * internal rows (reasoning/tool) are not deduped by ts and can share a ts with
+ * a delivered text row. Binds three parameters: (agentId, agentId, agentId).
+ */
+const AGENT_DELIVERY_SCOPE_SQL = `(sender = ? OR recipient = ? OR (transcript.kind = 'text' AND EXISTS (
+        SELECT 1 FROM transcript_recipient tr
+        WHERE tr.channel = transcript.channel AND tr.thread = transcript.thread
+          AND tr.ts = transcript.ts AND tr.agentId = ?)))`
+
 /**
  * A durably-persisted admitted-but-not-yet-completed inbox message (§6.9 #353). Holds the
  * bits needed to reconstruct a QueueEntry's DispatchContext on replay — everything EXCEPT
@@ -1400,10 +1415,7 @@ export class LocalStore {
     // so gate the delivery match on `kind = 'text'`, else a peer's reasoning/tool row at the
     // same ts would be pulled back in. Deliveries only ever concern conversational messages;
     // own internal rows still surface via `sender`.
-    const scope = `(sender = ? OR recipient = ? OR (transcript.kind = 'text' AND EXISTS (
-        SELECT 1 FROM transcript_recipient tr
-        WHERE tr.channel = transcript.channel AND tr.thread = transcript.thread
-          AND tr.ts = transcript.ts AND tr.agentId = ?)))`
+    const scope = AGENT_DELIVERY_SCOPE_SQL
     const hiddenToolTitles = [...SESSION_TITLE_TOOL_TITLES]
     const rows = (beforeSeq !== null
       ? this.db
@@ -1451,10 +1463,7 @@ export class LocalStore {
     before: TranscriptEventCursor | null,
     limit: number
   ): { rows: TranscriptRow[]; hasMore: boolean } {
-    const scope = `(sender = ? OR recipient = ? OR (transcript.kind = 'text' AND EXISTS (
-        SELECT 1 FROM transcript_recipient tr
-        WHERE tr.channel = transcript.channel AND tr.thread = transcript.thread
-          AND tr.ts = transcript.ts AND tr.agentId = ?)))`
+    const scope = AGENT_DELIVERY_SCOPE_SQL
     const hiddenToolTitles = [...SESSION_TITLE_TOOL_TITLES]
     const rows = (before !== null
       ? this.db
@@ -1509,10 +1518,7 @@ export class LocalStore {
     afterRevision: number,
     limit: number
   ): { rows: TranscriptRow[]; hasMore: boolean; cursor: number } {
-    const scope = `(sender = ? OR recipient = ? OR (transcript.kind = 'text' AND EXISTS (
-        SELECT 1 FROM transcript_recipient tr
-        WHERE tr.channel = transcript.channel AND tr.thread = transcript.thread
-          AND tr.ts = transcript.ts AND tr.agentId = ?)))`
+    const scope = AGENT_DELIVERY_SCOPE_SQL
     const rows = this.db
       .prepare(
         `SELECT * FROM transcript
@@ -2811,6 +2817,31 @@ export class LocalStore {
       .all(channel, thread, sinceTs) as unknown as TranscriptEntry[]
   }
 
+  /**
+   * `transcriptSince`, scoped to what ONE agent sent or received — the same
+   * delivery predicate the console session views use. For a synthetic pairwise
+   * `a2a:<caller>` thread (see `isSyntheticA2aChannel` in cp-collab-routes),
+   * every child of one caller shares the physical thread while each row is a
+   * private pairwise delivery: the §8.5 model catch-up must read only this
+   * pair's rows, or siblings see each other's private deliveries (#967).
+   */
+  transcriptSinceForAgent(channel: string, thread: string, sinceTs: string | null, agentId: string): TranscriptEntry[] {
+    if (sinceTs === null) {
+      return this.db
+        .prepare(
+          `SELECT * FROM transcript WHERE channel = ? AND thread = ? AND kind = 'text'
+             AND ${AGENT_DELIVERY_SCOPE_SQL} ORDER BY ts ASC`
+        )
+        .all(channel, thread, agentId, agentId, agentId) as unknown as TranscriptEntry[]
+    }
+    return this.db
+      .prepare(
+        `SELECT * FROM transcript WHERE channel = ? AND thread = ? AND kind = 'text' AND ts > ?
+           AND ${AGENT_DELIVERY_SCOPE_SQL} ORDER BY ts ASC`
+      )
+      .all(channel, thread, sinceTs, agentId, agentId, agentId) as unknown as TranscriptEntry[]
+  }
+
   /**
    * Provider-neutral context fence for one physical conversation thread. Unlike
    * `transcriptSince`, this never compares provider message ids from different
@@ -2834,6 +2865,25 @@ export class LocalStore {
       .all(channel, thread, afterRevision) as unknown as TranscriptRow[]
   }
 
+  /** `transcriptSinceRevision`, scoped to one agent's sent/received rows — the
+   *  turn-context refresh's read on a synthetic pairwise `a2a:<caller>` thread,
+   *  for the same reason as {@link transcriptSinceForAgent} (#967). */
+  transcriptSinceRevisionForAgent(
+    channel: string,
+    thread: string,
+    afterRevision: number,
+    agentId: string
+  ): TranscriptRow[] {
+    return this.db
+      .prepare(
+        `SELECT * FROM transcript
+         WHERE channel = ? AND thread = ? AND revision > ?
+           AND ${AGENT_DELIVERY_SCOPE_SQL}
+         ORDER BY revision ASC, seq ASC`
+      )
+      .all(channel, thread, afterRevision, agentId, agentId, agentId) as unknown as TranscriptRow[]
+  }
+
   /** The earliest inbound (non-agent) `text` message in a thread — the triggering user
    *  message. Used as a session-title fallback when neither ACP nor the title tool
    *  supplied one. Before the first meaningful request, this avoids showing only