From d1a87a7d1f43e8d18bc278a6b0199988a208ba7d Mon Sep 17 00:00:00 2001 From: Poytr1 Date: Fri, 14 Aug 2026 17:35:22 +0800 Subject: [PATCH] fix(daemon): scope pairwise a2a transcript reads to the session's own agent (#967) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every postless toAgent child of one caller shares the synthetic a2a: channel + the caller's thread, so all pairwise sessions of one caller land in ONE physical transcript thread — and the §8.5 catch-up and turn-context refresh, built for genuinely shared conversations, replayed SIBLINGS' private deliveries (role assignments, canaries included) and reports into each child's prompts. Measured in the webchat Werewolf real run (#941). The sessions are pairwise; the reads now are too. isSyntheticA2aChannel (cp-collab-routes, beside the coordinate minting) marks the synthetic channel; the store gains transcriptSince(Revision)ForAgent — the same sender/recipient/transcript_recipient delivery scope the console session views already use, factored into one shared predicate — and the three context seams (§8.5 catch-up in session-manager, the thread-context refresh, localInvalidatingEvents) route through the scoped reads exactly when the session sits on a synthetic a2a channel. Every a2a writer already stamps recipient, so a pair's own rows are unaffected. Ordinary shared conversations (channels, webchat rosters, co-hosted participants) are untouched. Pinned by packages/daemon/test/a2a-transcript-privacy.test.ts (red without the fix: the sibling's canary appears in the child's prompt) and, at game level, by the new canaryCrossVisibility=0 audit in the scripted webchat Werewolf gate (every player prompt scanned across conversation AND pairwise sessions). Closes #967. Co-Authored-By: Claude Fable 5 --- evals/games/webchat-werewolf-runner.ts | 22 ++++++ evals/test/webchat-werewolf.test.ts | 6 ++ packages/daemon/src/cp/cp-collab-routes.ts | 17 ++++ packages/daemon/src/daemon.ts | 18 ++++- .../daemon/src/session/session-manager.ts | 14 +++- packages/daemon/src/session/thread-context.ts | 17 +++- packages/daemon/src/store/local-store.ts | 74 +++++++++++++++--- .../test/a2a-transcript-privacy.test.ts | Bin 0 -> 3868 bytes 8 files changed, 148 insertions(+), 20 deletions(-) create mode 100644 packages/daemon/test/a2a-transcript-privacy.test.ts diff --git a/evals/games/webchat-werewolf-runner.ts b/evals/games/webchat-werewolf-runner.ts index 895433742..7b2d44aca 100644 --- a/evals/games/webchat-werewolf-runner.ts +++ b/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() } : {}), diff --git a/evals/test/webchat-werewolf.test.ts b/evals/test/webchat-werewolf.test.ts index 78299c10f..7dd0a419f 100644 --- a/evals/test/webchat-werewolf.test.ts +++ b/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. diff --git a/packages/daemon/src/cp/cp-collab-routes.ts b/packages/daemon/src/cp/cp-collab-routes.ts index 98a559fa3..610b6bc79 100644 --- a/packages/daemon/src/cp/cp-collab-routes.ts +++ b/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>() diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index 79a17d937..9ab8ab449 100644 --- a/packages/daemon/src/daemon.ts +++ b/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) } diff --git a/packages/daemon/src/session/session-manager.ts b/packages/daemon/src/session/session-manager.ts index e60285aae..f091cbe61 100644 --- a/packages/daemon/src/session/session-manager.ts +++ b/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:` 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. diff --git a/packages/daemon/src/session/thread-context.ts b/packages/daemon/src/session/thread-context.ts index 75ed5769d..6d26bbd57 100644 --- a/packages/daemon/src/session/thread-context.ts +++ b/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 + /** Read only rows this agent sent or received. Set by the daemon for sessions + * on a synthetic pairwise `a2a:` 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) diff --git a/packages/daemon/src/store/local-store.ts b/packages/daemon/src/store/local-store.ts index 37e598681..b2f55d4d0 100644 --- a/packages/daemon/src/store/local-store.ts +++ b/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:` 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:` 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 diff --git a/packages/daemon/test/a2a-transcript-privacy.test.ts b/packages/daemon/test/a2a-transcript-privacy.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..7d64a5ac33cf918f880aa38aeb2329e11f7f1ba2 GIT binary patch literal 3868 zcmcInYi<)s5Z>QDMM)sd4DusEyX<0KvV&0~g(M0HgaG08xE(i)r-$jD7%Q^GA@@q%kzKx_2gXJEU}~vN7oi9Ul!S6W9OF7rsgkzLZS>;W-cKheujw_Zf z2`VzNkY!{XT6H$%g3>e8RVvC-L9-9=1bQT3pB^6$0%j#$h?GSpz1xO9ef&MR=3omr zz;Y^2vuX@=EFe&ju}|;Hv4yT^CKfam=aOd17&%UcF8u6kr=eypHWx6?g+G!_k{=40 z+9oOdKUcJ*36N8CuBd9 zTjN?5dI3kgNdN@uZwLO_{6?;Sj)djM2Pog znP{YKt8UA1wtsN=@J){o9UZhpv~|*7pNyiQuy=T}@2uws#suRt(RznBcu9&BOvBXl z>Ej<)Xo*|b6BA0CyBIu%vEmp4wge_ax?vGXaOKGjky>4%#J#LRjo;huhS+C~=%vrp z>u#+m@}%i6f-wM1)@fj2U%M;mp-A8P#=m}1pDl+z0W0bLSET#Pp(~MP;)}@P_uK%) zxtx=ykN5&)q`4sL=8kLJO{&d?hyLrNNmDfZBOilq2nN+Q#9Sy#HUAhZN*jCQcjDSG z2oiBl+G$NGK(fg&wfHuh)OVX9o&ioTJOGz-(PG$y*!y&GQNTo8%J*0L%vIB6T(b>| z|J;`+}jTXKz%gV71CW`y#2S^FJ3rBv-VUi^6m6u7|>b{J*U44X!yvB3Y=2Sn@W$e4MHdyyW`%Kzk z2>Ptv$Lu?l`I()@yQm>1`JnKdQ2R}CQ2Q&Wj(&eU)c%*Du5`l|3Y@X{^zqN_PWA}4 zdTI}gzJPMqxIuN@OU&cl++ z{u1GA+3lg9@NIzh|E%pVLEXNx_E8 jDtbW$_HumYWZ8NuT^f+Os0ZE>*X7||! literal 0 HcmV?d00001