From 7777b133885a4551a00a17d575939afa17bd4817 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:26:23 +0000 Subject: [PATCH] feat(conversations): Add readConversationEvent tool Co-Authored-By: Thomas Jiang <20501235+thomasjiangcy@users.noreply.github.com> --- .../evals/conversation/continuity.eval.ts | 51 ++++ .../src/chat/conversations/event-access.ts | 233 +++++++++++++++++ .../src/chat/conversations/tool-source.ts | 2 +- packages/junior/src/chat/tools/index.ts | 2 + .../src/chat/tools/read-conversation-event.ts | 181 +++++++++++++ .../chat/tools/search-conversation-events.ts | 226 +--------------- .../read-conversation-event.test.ts | 247 ++++++++++++++++++ 7 files changed, 726 insertions(+), 216 deletions(-) create mode 100644 packages/junior/src/chat/conversations/event-access.ts create mode 100644 packages/junior/src/chat/tools/read-conversation-event.ts create mode 100644 packages/junior/tests/integration/read-conversation-event.test.ts diff --git a/packages/junior-evals/evals/conversation/continuity.eval.ts b/packages/junior-evals/evals/conversation/continuity.eval.ts index 7c8f94a6c0..7c59be319c 100644 --- a/packages/junior-evals/evals/conversation/continuity.eval.ts +++ b/packages/junior-evals/evals/conversation/continuity.eval.ts @@ -3,6 +3,19 @@ import { slackMention, reply } from "@junior-evals/fixture/inputs"; import { rubric, RubricJudge } from "@junior-evals/fixture/judge"; import { test } from "@junior-evals/fixture/test"; +// Junior compacts agent history above 90% of the context window, at about 4 +// characters per token. This build log is about 56,000 tokens, so it does not +// fit a 50,000 token window. Checksums are pseudo-random, so a summary cannot +// recreate one. +const SMALL_CONTEXT_WINDOW_TOKENS = 50_000; +const checksum = (step: number) => + (Math.imul(step + 1, 2_654_435_761) >>> 0).toString(16).padStart(8, "0"); +const BUILD_LOG = Array.from( + { length: 3_500 }, + (_, index) => + `2026-03-01T04:${String(index % 60).padStart(2, "0")}:00Z step ${String(index + 1).padStart(4, "0")} ok checksum ${checksum(index + 1)} region iad1`, +).join("\n"); + describe("Thread Continuity", () => { test("when a follow-up asks about the prior turn, recall the earlier budget context", async ({ run, @@ -25,4 +38,42 @@ describe("Thread Continuity", () => { expect(conversation.replies).toHaveLength(1); }); + + test("when a compacted tool result holds an exact detail, read the full result back", async ({ + agent, + }) => { + const { run } = await agent({ + limits: { contextWindowTokens: SMALL_CONTEXT_WINDOW_TOKENS }, + }); + const conversation = await run( + slackMention( + "What was the exact checksum on step 0042 of that build log? I need to paste it into the release ticket.", + ), + { + history: [ + slackMention("Can you pull the full build log for build 7421?"), + reply("Build 7421 finished. All 3,500 steps passed in iad1.", { + toolHistory: [ + { + name: "bash", + arguments: { command: "cat logs/build-7421.log" }, + result: { + exit_code: 0, + stdout: BUILD_LOG, + stderr: "", + timed_out: false, + }, + }, + ], + }), + ], + }, + ); + + expect(conversation.compactions).toBeGreaterThan(0); + // The checksum is only in the compacted tool result. + expect(conversation.replies.map((item) => item.text).join("\n")).toContain( + checksum(42), + ); + }); }); diff --git a/packages/junior/src/chat/conversations/event-access.ts b/packages/junior/src/chat/conversations/event-access.ts new file mode 100644 index 0000000000..297ef67506 --- /dev/null +++ b/packages/junior/src/chat/conversations/event-access.ts @@ -0,0 +1,233 @@ +/** + * Read access to a Conversation's durable event log for model-facing tools. + * The current Conversation tree is always readable. Other Conversations are + * readable only when they are public and in the same Slack workspace. + */ +import { + canExposeConversationPayload, + type ConversationPrivacy, +} from "@/chat/conversation-privacy"; +import type { ConversationEvent } from "@/chat/conversations/history"; +import type { ConversationStore } from "@/chat/conversations/store"; +import { getConversationStore } from "@/chat/db"; +import { ToolInputError } from "@/chat/tools/execution/tool-input-error"; +import type { ToolRuntimeContext } from "@/chat/tools/types"; + +interface ConversationEventAccessScope { + currentConversationId: string; + currentRootConversationId?: string; + provider?: string; + providerTenantId?: string; +} + +/** + * Throw a repairable tool error unless the active context may read the event + * log of `conversationId`. + */ +export async function assertCanReadConversationEvents( + context: ToolRuntimeContext, + conversationId: string, +): Promise { + const conversationStore = getConversationStore(); + const target = await conversationStore.get({ conversationId }); + if (!target) { + throw new ToolInputError(`Conversation not found: ${conversationId}`); + } + if (target.transcriptPurgedAtMs !== undefined) { + throw new ToolInputError( + `Conversation transcript was purged: ${conversationId}`, + ); + } + + const accessScope = await resolveAccessScope( + conversationStore, + context, + context.conversationId, + ); + // A Conversation with a parent uses the root privacy boundary. + const targetRootConversationId = await resolveRootConversationId( + conversationStore, + conversationId, + target.parentConversationId, + ); + const targetRoot = + targetRootConversationId === conversationId + ? target + : await conversationStore.get({ + conversationId: targetRootConversationId, + }); + assertScopeCanReadTarget({ + accessScope, + targetConversationId: conversationId, + targetRootConversationId, + targetVisibility: targetRoot?.visibility ?? target.visibility, + targetDestination: targetRoot?.destination ?? target.destination, + }); +} + +async function resolveAccessScope( + conversationStore: ConversationStore, + context: ToolRuntimeContext, + currentConversationId: string, +): Promise { + const current = await conversationStore.get({ + conversationId: currentConversationId, + }); + const currentRootConversationId = current + ? await resolveRootConversationId( + conversationStore, + currentConversationId, + current.parentConversationId, + ) + : currentConversationId; + + if (context.location?.provider === "slack") { + return { + currentConversationId, + currentRootConversationId, + provider: "slack", + providerTenantId: context.location.teamId, + }; + } + + // TODO(dcramer): Remove the Source and Destination fallbacks after every + // deployed Slack Conversation has a stored Location and AgentRun no longer + // has Destination. + if (context.source.kind === "slack") { + return { + currentConversationId, + currentRootConversationId, + provider: "slack", + providerTenantId: context.source.teamId, + }; + } + + if ( + context.destination.platform === "slack" && + "teamId" in context.destination + ) { + return { + currentConversationId, + currentRootConversationId, + provider: "slack", + providerTenantId: context.destination.teamId, + }; + } + + return { + currentConversationId, + currentRootConversationId, + }; +} + +async function resolveRootConversationId( + conversationStore: ConversationStore, + conversationId: string, + parentConversationId: string | undefined, +): Promise { + if (!parentConversationId) { + return conversationId; + } + + let cursor: string | undefined = parentConversationId; + const seen = new Set([conversationId]); + while (cursor) { + if (seen.has(cursor)) { + // Fail closed on cycles; callers treat missing root as unauthorized. + return conversationId; + } + seen.add(cursor); + const parent = await conversationStore.get({ conversationId: cursor }); + if (!parent) { + return conversationId; + } + if (!parent.parentConversationId) { + return parent.conversationId; + } + cursor = parent.parentConversationId; + } + return conversationId; +} + +function assertScopeCanReadTarget(args: { + accessScope: ConversationEventAccessScope; + targetConversationId: string; + targetRootConversationId?: string; + targetVisibility?: ConversationPrivacy; + targetDestination?: { + platform: string; + teamId?: string; + channelId?: string; + conversationId?: string; + }; +}): void { + const { + accessScope, + targetConversationId, + targetRootConversationId, + targetVisibility, + targetDestination, + } = args; + + if ( + targetConversationId === accessScope.currentConversationId || + (targetRootConversationId !== undefined && + targetRootConversationId === accessScope.currentRootConversationId) + ) { + return; + } + + const publicPayloadAllowed = canExposeConversationPayload( + targetVisibility ? { visibility: targetVisibility } : {}, + ); + if (!publicPayloadAllowed) { + throw new ToolInputError( + `Conversation events are not accessible: ${targetConversationId}`, + ); + } + + if ( + accessScope.provider === "slack" && + targetDestination?.platform === "slack" && + accessScope.providerTenantId && + targetDestination.teamId === accessScope.providerTenantId + ) { + return; + } + + throw new ToolInputError( + `Conversation events are not accessible: ${targetConversationId}`, + ); +} + +/** Replace replacement history with its item count; the original events stay readable by seq. */ +export function stripReplacementHistory( + data: ConversationEvent["data"], +): ConversationEvent["data"] | Record { + if (data.type === "unknown") { + return { + ...data, + payload: stripReplacementHistoryField(data.payload), + }; + } + if (data.type === "compaction" || data.type === "handoff") { + return stripReplacementHistoryField(data) as Record; + } + return data; +} + +function stripReplacementHistoryField(value: unknown): unknown { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return value; + } + const { replacementHistory, ...rest } = value as Record; + if (replacementHistory === undefined) { + return value; + } + return { + ...rest, + replacement_history_count: Array.isArray(replacementHistory) + ? replacementHistory.length + : 0, + }; +} diff --git a/packages/junior/src/chat/conversations/tool-source.ts b/packages/junior/src/chat/conversations/tool-source.ts index 38a2c0f5f3..448dfa88e0 100644 --- a/packages/junior/src/chat/conversations/tool-source.ts +++ b/packages/junior/src/chat/conversations/tool-source.ts @@ -1,5 +1,5 @@ /** Deferred tool catalog source for retained Junior conversations. */ export const CONVERSATIONS_TOOL_SOURCE = { id: "conversations", - description: "Search messages and events in retained conversations.", + description: "Search and read messages and events in retained conversations.", } as const; diff --git a/packages/junior/src/chat/tools/index.ts b/packages/junior/src/chat/tools/index.ts index 15b02d75cb..b67f59879d 100644 --- a/packages/junior/src/chat/tools/index.ts +++ b/packages/junior/src/chat/tools/index.ts @@ -59,6 +59,7 @@ import { createUserCustomizationTools } from "@/chat/tools/user-customization"; import { createPublishImageTool } from "@/chat/tools/publish-image"; import { createUnpublishImageTool } from "@/chat/tools/unpublish-image"; import { createLoadAttachmentTool } from "@/chat/tools/load-attachment"; +import { createReadConversationEventTool } from "@/chat/tools/read-conversation-event"; import { createSearchConversationEventsTool } from "@/chat/tools/search-conversation-events"; import { createSearchConversationBriefsTool } from "@/chat/tools/search-conversation-briefs"; import { createHandoffTool } from "@/chat/tools/handoff/tool"; @@ -145,6 +146,7 @@ export function createTools( ...createUserCustomizationTools(context), }; tools.searchConversationEvents = createSearchConversationEventsTool(context); + tools.readConversationEvent = createReadConversationEventTool(context); if (context.conversationPrivacy === "public") { tools.searchConversationBriefs = createSearchConversationBriefsTool( slackContext diff --git a/packages/junior/src/chat/tools/read-conversation-event.ts b/packages/junior/src/chat/tools/read-conversation-event.ts new file mode 100644 index 0000000000..c46c694d95 --- /dev/null +++ b/packages/junior/src/chat/tools/read-conversation-event.ts @@ -0,0 +1,181 @@ +import { z } from "zod"; +import { + assertCanReadConversationEvents, + stripReplacementHistory, +} from "@/chat/conversations/event-access"; +import type { ConversationEvent } from "@/chat/conversations/history"; +import { CONVERSATIONS_TOOL_SOURCE } from "@/chat/conversations/tool-source"; +import { getConversationEventStore } from "@/chat/db"; +import { juniorToolOutputSchema } from "@/chat/tool-support/structured-result"; +import { zodTool } from "@/chat/tool-support/zod-tool"; +import { ToolInputError } from "@/chat/tools/execution/tool-input-error"; +import type { ToolRuntimeContext } from "@/chat/tools/types"; + +const DEFAULT_LIMIT = 20_000; +const MAX_LIMIT = 50_000; + +const readConversationEventOutputSchema = juniorToolOutputSchema.extend({ + conversation_id: z.string().min(1), + seq: z.number().int().nonnegative(), + history_version: z.number().int().nonnegative(), + created_at: z.iso.datetime(), + type: z.string().min(1).describe("Event type."), + tool_name: z + .string() + .min(1) + .optional() + .describe("Tool that produced a tool_result event."), + is_error: z + .boolean() + .optional() + .describe("Whether a tool_result event was an error."), + text: z.string().describe("Event text from offset, at most limit chars."), + offset: z.number().int().nonnegative(), + total_chars: z.number().int().nonnegative(), + has_more: z.boolean().describe("Whether text continues after this slice."), +}); + +/** Create a deferred tool that reads one full event of a conversation log. */ +export function createReadConversationEventTool(context: ToolRuntimeContext) { + return zodTool({ + description: + "Read the full text of one conversation event by seq, in pages. Works for events from earlier history versions, such as cleared tool results.", + exposure: "deferred", + source: CONVERSATIONS_TOOL_SOURCE, + annotations: { + destructiveHint: false, + idempotentHint: true, + openWorldHint: false, + readOnlyHint: true, + }, + inputSchema: z + .object({ + seq: z.number().int().nonnegative().describe("Event sequence number."), + conversation_id: z + .string() + .trim() + .min(1) + .nullable() + .describe( + "Conversation ID. Omit or null for the current conversation.", + ) + .optional(), + offset: z + .number() + .int() + .nonnegative() + .nullable() + .describe("First char to return. Default 0.") + .optional(), + limit: z + .number() + .int() + .min(1) + .max(MAX_LIMIT) + .nullable() + .describe( + `Maximum chars. Default ${DEFAULT_LIMIT}; max ${MAX_LIMIT}.`, + ) + .optional(), + }) + .strict(), + outputSchema: readConversationEventOutputSchema, + execute: async (input) => { + const conversationId = + input.conversation_id?.trim() || context.conversationId; + const offset = input.offset ?? 0; + const limit = input.limit ?? DEFAULT_LIMIT; + + await assertCanReadConversationEvents(context, conversationId); + // The log query has no history-version filter, so this finds events + // that a compaction or handoff replaced. + const page = await getConversationEventStore().query(conversationId, { + afterSeq: input.seq - 1, + beforeSeq: input.seq + 1, + limit: 1, + }); + const event = page.events[0]; + if (!event) { + throw new ToolInputError( + `Conversation event not found: seq ${input.seq} in ${conversationId}`, + ); + } + + const text = eventText(event.data); + if (offset > 0 && offset >= text.length) { + throw new ToolInputError( + `offset ${offset} is past the end of the event text (${text.length} chars)`, + ); + } + const slice = text.slice(offset, offset + limit); + return { + conversation_id: conversationId, + seq: event.seq, + history_version: event.historyVersion, + created_at: new Date(event.createdAtMs).toISOString(), + type: + event.data.type === "unknown" + ? event.data.originalType + : event.data.type, + ...toolResultFields(event.data), + text: slice, + offset, + total_chars: text.length, + has_more: offset + slice.length < text.length, + }; + }, + }); +} + +function toolResultFields(data: ConversationEvent["data"]) { + if (data.type !== "tool_result") return undefined; + return { + ...(typeof data.toolName === "string" && data.toolName + ? { tool_name: data.toolName } + : undefined), + ...(typeof data.isError === "boolean" + ? { is_error: data.isError } + : undefined), + }; +} + +/** + * Render agent history items as the text the model saw, and messages as their + * text. Other events render as JSON without replacement history, which stays + * readable through its original events. + */ +function eventText(data: ConversationEvent["data"]): string { + switch (data.type) { + case "user_message": + case "assistant_message": + case "tool_result": + return contentText(data.content); + case "message": + case "message_updated": + return data.text; + default: + return JSON.stringify(stripReplacementHistory(data), null, 2); + } +} + +function contentText(content: unknown): string { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + return content.map(contentPartText).join("\n"); +} + +function contentPartText(part: unknown): string { + if (!part || typeof part !== "object") return ""; + const value = part as Record; + switch (value.type) { + case "text": + return typeof value.text === "string" ? value.text : ""; + case "image": + return `[image: ${typeof value.mimeType === "string" ? value.mimeType : "unknown type"}]`; + case "toolCall": + return `[tool call: ${String(value.name)} ${JSON.stringify(value.arguments ?? {})}]`; + default: + // Thinking blocks and unknown parts are not returned. + return ""; + } +} diff --git a/packages/junior/src/chat/tools/search-conversation-events.ts b/packages/junior/src/chat/tools/search-conversation-events.ts index c7236b075d..effcdb87a1 100644 --- a/packages/junior/src/chat/tools/search-conversation-events.ts +++ b/packages/junior/src/chat/tools/search-conversation-events.ts @@ -1,17 +1,16 @@ import { z } from "zod"; -import { - canExposeConversationPayload, - type ConversationPrivacy, -} from "@/chat/conversation-privacy"; import { KNOWN_CONVERSATION_EVENT_TYPES, type ConversationEvent, type ConversationEventPage, type KnownConversationEventType, } from "@/chat/conversations/history"; -import type { ConversationStore } from "@/chat/conversations/store"; +import { + assertCanReadConversationEvents, + stripReplacementHistory, +} from "@/chat/conversations/event-access"; import { CONVERSATIONS_TOOL_SOURCE } from "@/chat/conversations/tool-source"; -import { getConversationEventStore, getConversationStore } from "@/chat/db"; +import { getConversationEventStore } from "@/chat/db"; import { juniorToolOutputSchema } from "@/chat/tool-support/structured-result"; import { zodTool } from "@/chat/tool-support/zod-tool"; import { ToolInputError } from "@/chat/tools/execution/tool-input-error"; @@ -44,7 +43,11 @@ const projectedEventSchema = z .min(1) .optional() .describe("Retry-stable event identity, when assigned."), - data: z.record(z.string(), z.unknown()).describe("Bounded event data."), + data: z + .record(z.string(), z.unknown()) + .describe( + "Bounded event data. Read an omitted payload with readConversationEvent.", + ), }) .strict(); @@ -61,13 +64,6 @@ const searchConversationEventsOutputSchema = juniorToolOutputSchema.extend({ .describe("Events omitted by the response byte limit."), }); -interface ConversationEventAccessScope { - currentConversationId: string; - currentRootConversationId?: string; - provider?: string; - providerTenantId?: string; -} - /** Create a deferred tool that searches one conversation's durable event log. */ export function createSearchConversationEventsTool( context: ToolRuntimeContext, @@ -145,41 +141,7 @@ export function createSearchConversationEventsTool( ); } - const conversationStore = getConversationStore(); - const target = await conversationStore.get({ conversationId }); - if (!target) { - throw new ToolInputError(`Conversation not found: ${conversationId}`); - } - if (target.transcriptPurgedAtMs !== undefined) { - throw new ToolInputError( - `Conversation transcript was purged: ${conversationId}`, - ); - } - - const accessScope = await resolveAccessScope( - conversationStore, - context, - currentConversationId, - ); - // A Conversation with a parent uses the root privacy boundary. - const targetRootConversationId = await resolveRootConversationId( - conversationStore, - conversationId, - target.parentConversationId, - ); - const targetRoot = - targetRootConversationId === conversationId - ? target - : await conversationStore.get({ - conversationId: targetRootConversationId, - }); - assertCanSearchConversationEvents({ - accessScope, - targetConversationId: conversationId, - targetRootConversationId, - targetVisibility: targetRoot?.visibility ?? target.visibility, - targetDestination: targetRoot?.destination ?? target.destination, - }); + await assertCanReadConversationEvents(context, conversationId); const page = await getConversationEventStore().query(conversationId, { limit, @@ -206,141 +168,6 @@ export function createSearchConversationEventsTool( }); } -async function resolveAccessScope( - conversationStore: ConversationStore, - context: ToolRuntimeContext, - currentConversationId: string, -): Promise { - const current = await conversationStore.get({ - conversationId: currentConversationId, - }); - const currentRootConversationId = current - ? await resolveRootConversationId( - conversationStore, - currentConversationId, - current.parentConversationId, - ) - : currentConversationId; - - if (context.location?.provider === "slack") { - return { - currentConversationId, - currentRootConversationId, - provider: "slack", - providerTenantId: context.location.teamId, - }; - } - - // TODO(dcramer): Remove the Source and Destination fallbacks after every - // deployed Slack Conversation has a stored Location and AgentRun no longer - // has Destination. - if (context.source.kind === "slack") { - return { - currentConversationId, - currentRootConversationId, - provider: "slack", - providerTenantId: context.source.teamId, - }; - } - - if ( - context.destination.platform === "slack" && - "teamId" in context.destination - ) { - return { - currentConversationId, - currentRootConversationId, - provider: "slack", - providerTenantId: context.destination.teamId, - }; - } - - return { - currentConversationId, - currentRootConversationId, - }; -} - -async function resolveRootConversationId( - conversationStore: ConversationStore, - conversationId: string, - parentConversationId: string | undefined, -): Promise { - if (!parentConversationId) { - return conversationId; - } - - let cursor: string | undefined = parentConversationId; - const seen = new Set([conversationId]); - while (cursor) { - if (seen.has(cursor)) { - // Fail closed on cycles; callers treat missing root as unauthorized. - return conversationId; - } - seen.add(cursor); - const parent = await conversationStore.get({ conversationId: cursor }); - if (!parent) { - return conversationId; - } - if (!parent.parentConversationId) { - return parent.conversationId; - } - cursor = parent.parentConversationId; - } - return conversationId; -} - -function assertCanSearchConversationEvents(args: { - accessScope: ConversationEventAccessScope; - targetConversationId: string; - targetRootConversationId?: string; - targetVisibility?: ConversationPrivacy; - targetDestination?: { - platform: string; - teamId?: string; - channelId?: string; - conversationId?: string; - }; -}): void { - const { - accessScope, - targetConversationId, - targetRootConversationId, - targetVisibility, - targetDestination, - } = args; - - if ( - targetConversationId === accessScope.currentConversationId || - (targetRootConversationId !== undefined && - targetRootConversationId === accessScope.currentRootConversationId) - ) { - return; - } - - const publicPayloadAllowed = canExposeConversationPayload( - targetVisibility ? { visibility: targetVisibility } : {}, - ); - if (!publicPayloadAllowed) { - throw new ToolInputError( - `Conversation events are not accessible: ${targetConversationId}`, - ); - } - - if ( - accessScope.provider === "slack" && - targetDestination?.platform === "slack" && - accessScope.providerTenantId && - targetDestination.teamId === accessScope.providerTenantId - ) { - return; - } - - throw new ToolInputError( - `Conversation events are not accessible: ${targetConversationId}`, - ); -} - function projectEventsForTool( page: ConversationEventPage, options: { newestFirst: boolean }, @@ -431,35 +258,4 @@ function eventIdentity(data: Record) { ); } -function stripReplacementHistory( - data: ConversationEvent["data"], -): ConversationEvent["data"] | Record { - if (data.type === "unknown") { - return { - ...data, - payload: stripReplacementHistoryField(data.payload), - }; - } - if (data.type === "compaction" || data.type === "handoff") { - return stripReplacementHistoryField(data) as Record; - } - return data; -} - -function stripReplacementHistoryField(value: unknown): unknown { - if (!value || typeof value !== "object" || Array.isArray(value)) { - return value; - } - const { replacementHistory, ...rest } = value as Record; - if (replacementHistory === undefined) { - return value; - } - return { - ...rest, - replacement_history_count: Array.isArray(replacementHistory) - ? replacementHistory.length - : 0, - }; -} - export type { KnownConversationEventType }; diff --git a/packages/junior/tests/integration/read-conversation-event.test.ts b/packages/junior/tests/integration/read-conversation-event.test.ts new file mode 100644 index 0000000000..07d46a2c47 --- /dev/null +++ b/packages/junior/tests/integration/read-conversation-event.test.ts @@ -0,0 +1,247 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createResumeState } from "@/chat/agent/resume"; +import { runCodeModeScript } from "@/chat/codemode/bridge"; +import { + closeDb, + getConversationEventStore, + getConversationStore, +} from "@/chat/db"; +import { createToolActionReview } from "@/chat/guardian/gate"; +import type { ToolActionReviewer } from "@/chat/guardian/proposal"; +import { SkillSandbox } from "@/chat/sandbox/skill-sandbox"; +import { planToolExposure } from "@/chat/tool-exposure"; +import { createToolPipeline } from "@/chat/tool-support/pi-tool-adapter"; +import { createTools } from "@/chat/tools"; +import type { ToolRuntimeContext } from "@/chat/tools/types"; + +const CURRENT_CONVERSATION_ID = "slack:C123:1700000000.100000"; +const CODEMODE_CALL_ID = "call-codemode"; +const SOURCE = { kind: "scheduled_automation" } as const; + +function context(): ToolRuntimeContext { + return { + conversationId: CURRENT_CONVERSATION_ID, + destination: { platform: "slack", teamId: "T123", channelId: "C123" }, + location: { + id: "location:T123:C123", + provider: "slack", + teamId: "T123", + channelId: "C123", + threadTs: "1700000000.100000", + }, + source: SOURCE, + egress: { + async fetch() { + return new Response("ok"); + }, + }, + workspace: {} as ToolRuntimeContext["workspace"], + }; +} + +/** Run a Code Mode script against the tools a real turn exposes to scripts. */ +async function runScript(source: string) { + const review = vi.fn(async () => ({ + decision: "allow", + reason: "Read-only lookup.", + riskLevel: "low", + rule: "none", + userAuthorization: "high", + })); + const actionReview = createToolActionReview({ + context: { + actor: { platform: "slack", teamId: "T123", userId: "U123" }, + conversationId: CURRENT_CONVERSATION_ID, + destination: context().destination, + source: SOURCE, + userIntent: () => "What was the checksum in the earlier deploy log?", + }, + onDecision: async () => undefined, + onFatal: () => undefined, + reviewer: { review }, + }); + const resume = createResumeState({ + checkpoint: { resumed: false, sliceId: 1 }, + conversationId: CURRENT_CONVERSATION_ID, + destination: context().destination, + durability: {}, + recordActiveMcpProviders: async () => undefined, + runSource: SOURCE, + startedAtMs: Date.now(), + surface: "internal", + turnId: "turn-read-event", + }); + const { scriptTools } = planToolExposure(createTools([], {}, context())); + const { result } = await runCodeModeScript({ + countToolCall: resume.countToolCall, + pipeline: createToolPipeline({ + actionReview, + catalogTools: {}, + conversationPrivacy: "public", + sandbox: new SkillSandbox([], []), + spanContext: {}, + }), + source, + toolCallId: CODEMODE_CALL_ID, + tools: scriptTools, + }); + return result; +} + +async function recordSlackConversation(args: { + channelId: string; + conversationId: string; + visibility: "private" | "public"; +}) { + await getConversationStore().recordActivity({ + conversationId: args.conversationId, + destination: { + platform: "slack", + teamId: "T123", + channelId: args.channelId, + }, + nowMs: 1, + source: "slack", + visibility: args.visibility, + }); +} + +describe("readConversationEvent", () => { + afterEach(async () => { + await closeDb(); + }); + + it("reads a large tool result from a replaced history version in pages", async () => { + await recordSlackConversation({ + channelId: "C123", + conversationId: CURRENT_CONVERSATION_ID, + visibility: "public", + }); + const log = Array.from( + { length: 1_500 }, + (_, index) => `step ${index} checksum ${(index * 7_919).toString(16)}`, + ).join("\n"); + const events = getConversationEventStore(); + const [toolResult] = await events.append(CURRENT_CONVERSATION_ID, [ + { + createdAtMs: 1, + data: { + type: "tool_result", + toolCallId: "call-bash", + toolName: "bash", + isError: false, + content: [{ type: "text", text: log }], + timestamp: 1, + }, + }, + ]); + await events.replaceHistory(CURRENT_CONVERSATION_ID, { + createdAtMs: 2, + data: { + type: "compaction", + modelProfile: "standard", + modelId: "test/model", + summary: "Ran a deploy.", + replacementHistory: [], + }, + }); + expect( + await events.loadCurrentHistoryVersion(CURRENT_CONVERSATION_ID), + ).toBe(toolResult!.historyVersion + 1); + + const result = await runScript(` + const pages = []; + let offset = 0; + while (true) { + const page = await tools.readConversationEvent({ + seq: ${toolResult!.seq}, + offset, + limit: 15000, + }); + pages.push(page); + if (!page.has_more) break; + offset += page.text.length; + } + return { + first: { ...pages[0], text: undefined }, + pageCount: pages.length, + text: pages.map((page) => page.text).join(""), + }; + `); + + expect(result.ok).toBe(true); + const value = result.value as { + first: Record; + pageCount: number; + text: string; + }; + expect(value.text).toBe(log); + expect(value.pageCount).toBe(Math.ceil(log.length / 15_000)); + expect(value.first).toMatchObject({ + conversation_id: CURRENT_CONVERSATION_ID, + seq: toolResult!.seq, + history_version: toolResult!.historyVersion, + type: "tool_result", + tool_name: "bash", + is_error: false, + offset: 0, + total_chars: log.length, + has_more: true, + }); + + const missing = await runScript(` + try { + await tools.readConversationEvent({ seq: 999 }); + } catch (error) { + return error.message; + } + `); + expect(missing.value).toContain( + `Conversation event not found: seq 999 in ${CURRENT_CONVERSATION_ID}`, + ); + }); + + it("refuses events of a private conversation outside the current tree", async () => { + await recordSlackConversation({ + channelId: "C123", + conversationId: CURRENT_CONVERSATION_ID, + visibility: "public", + }); + const privateConversationId = "slack:D999:1700000000.300000"; + await recordSlackConversation({ + channelId: "D999", + conversationId: privateConversationId, + visibility: "private", + }); + const [message] = await getConversationEventStore().append( + privateConversationId, + [ + { + createdAtMs: 1, + data: { + type: "message", + messageId: "m-1", + role: "user", + text: "private text", + }, + }, + ], + ); + + const result = await runScript(` + try { + return await tools.readConversationEvent({ + conversation_id: ${JSON.stringify(privateConversationId)}, + seq: ${message!.seq}, + }); + } catch (error) { + return error.message; + } + `); + + expect(result.value).toContain( + `Conversation events are not accessible: ${privateConversationId}`, + ); + expect(JSON.stringify(result)).not.toContain("private text"); + }); +});