diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index 0ec5b90af2ab..0b3cdb78589a 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -106,6 +106,7 @@ const FLEET_SHELL_BUDGET = { // About 14% above the observed 75 KB. The window holds the working run only: 2 nodes, 2 subagents, // 1 plan, 1 provider turn and the last finished turn's checkpoint. When finished runs ride along // it holds 82 nodes, 82 subagents, 41 plans, 41 provider turns, 40 checkpoints and 740 KB. +// The working run's ready handoff ships without its 256 KB history; with it the open is 339 KB. const STALE_TODO_BUDGET = { decodedBytes: 85_000, rows: 75, diff --git a/apps/server/integration/TransferBudgetV2Fixture.integration.ts b/apps/server/integration/TransferBudgetV2Fixture.integration.ts index 85ac051673c1..9a857120c512 100644 --- a/apps/server/integration/TransferBudgetV2Fixture.integration.ts +++ b/apps/server/integration/TransferBudgetV2Fixture.integration.ts @@ -2,6 +2,7 @@ import { CheckpointId, CheckpointRef, CheckpointScopeId, + ContextHandoffId, EventId, MessageId, NodeId, @@ -446,6 +447,7 @@ export const STALE_TODO_THREAD_ID = ThreadId.make("transfer-budget-stale-todo-li export const STALE_TODO_SETTLED_RUNS = 40; export const STALE_TODO_SUBAGENTS_PER_RUN = 2; export const STALE_TODO_LIVE_ROWS = 600; +const STALE_TODO_HANDOFF_HISTORY_BYTES = 256 * 1024; const staleTodoCheckpointId = (index: number) => CheckpointId.make(`checkpoint:thread:${STALE_TODO_THREAD_ID}:ordinal:${index + 1}`); /** The checkpoint clients read the last finished turn's diff from. */ @@ -742,7 +744,50 @@ export function staleTodoListThreadBatches( exitCode: 0, }, }); - if (live) return events; + if (live) { + // A ready handoff rides along with every window. Its history and delivery are for the + // provider turn start and must not reach a client. + const earlierItemId = TurnItemId.make(`turn-item:stale-todo:${index}:handoff-source`); + events.push({ + ...common, + id: EventId.make(`stale-todo-handoff-${index}`), + type: "context-handoff.updated", + payload: { + id: ContextHandoffId.make(`handoff:run:${runKey}`), + transferId: null, + threadId, + targetRunId: runId, + fromProviderThreadIds: [providerThreadId], + toProviderThreadId: providerThreadId, + coveredRunOrdinals: { from: 1, to: index }, + strategy: "full_thread_summary", + status: "ready", + summaryMessageId: null, + summaryText: text(index, 6, 1_024), + history: { + messages: [ + { + role: "user", + text: text(index, 7, STALE_TODO_HANDOFF_HISTORY_BYTES), + threadId, + runId: null, + itemId: earlierItemId, + providerThreadId, + status: "completed", + kind: "user_message", + }, + ], + coverage: "full", + omittedItems: 0, + }, + delivery: { nativeThreadId: "native", status: "injected", itemIds: [earlierItemId] }, + createdByProviderInstanceId: providerInstanceId, + createdAt: now, + updatedAt: now, + }, + }); + return events; + } message("assistant", text(index, 5, 1_200)); const checkpointId = staleTodoCheckpointId(index); diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index 8c5fe2447ced..313c583eecf7 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -448,6 +448,8 @@ it.live("reports V2 thread HTTP and WebSocket transfer budgets", () => staleTodo.subagents.filter((subagent) => subagent.runId === STALE_TODO_LIVE_RUN_ID), STALE_TODO_SUBAGENTS_PER_RUN, ); + // The byte cap below also passes when the window loses the handoff. + assert.lengthOf(staleTodo.contextHandoffs, 1); const staleTodoColdOpen = { http: staleTodoHttp, rows: staleTodo.visibleTurnItems.length, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a311d1ed6a7b..3827f69c0948 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -428,6 +428,10 @@ export interface OrchestratorV2Shape { options: ProjectionTimelinePageOptions, ) => Effect.Effect; readonly getMessageCount: (threadId: ThreadId) => Effect.Effect; + readonly getTranscriptByteFloor: ( + threadId: ThreadId, + limit: number, + ) => Effect.Effect; readonly getTurnItem: (input: { readonly threadId: ThreadId; readonly itemId: TurnItemId; @@ -15418,6 +15422,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .getMessageCount(threadId) .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))); }), + getTranscriptByteFloor: Effect.fn("OrchestratorV2.getTranscriptByteFloor")( + function (threadId, limit) { + return projectionStore + .getTranscriptByteFloor(threadId, limit) + .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))); + }, + ), getTurnItem: Effect.fn("OrchestratorV2.getTurnItem")(function (input) { return projectionStore .getTurnItem(input) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 580cdc24cac4..53cd77a88f85 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -9,6 +9,7 @@ import { ContextTransferId, CheckpointRef, CheckpointScopeId, + ContextHandoffId, MessageId, type ModelSelection, type OrchestrationV2ProviderThread, @@ -55,6 +56,7 @@ import { THREAD_HISTORY_PAGE_POLICY, THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT, } from "./threadHistoryPaging.ts"; +import { buildBoundedThreadStreamSnapshot } from "./ThreadStream.ts"; import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; import { layerReaderDatabase, traceSqlRouting } from "../persistence/testkit/SqliteRouting.ts"; @@ -2594,6 +2596,142 @@ it.effect("SQL snapshot windows keep cards for requests still pending", () => }).pipe(Effect.provide(layerTest)), ); +it.effect("SQL snapshot windows leave handoff bodies in SQLite without changing the wire", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:handoff-window"); + yield* store.apply({ + id: EventId.make("event:handoff-window:thread"), + type: "thread.created", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId: ProjectId.make("project:handoff-window"), + title: "Handoff window", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + }); + const providerThreadId = ProviderThreadId.make("provider-thread:handoff-window"); + const earlierItemId = TurnItemId.make("item:handoff-window:earlier"); + const handoff = (id: string, status: "ready" | "superseded") => ({ + id: ContextHandoffId.make(`handoff:handoff-window:${id}`), + transferId: null, + threadId, + targetRunId: RunId.make("run:handoff-window"), + fromProviderThreadIds: [providerThreadId], + toProviderThreadId: providerThreadId, + coveredRunOrdinals: { from: 1, to: 1 }, + strategy: "full_thread_summary" as const, + status, + summaryMessageId: null, + summaryText: `Summary ${id}`, + history: { + messages: [ + { + role: "user" as const, + text: "Earlier request ".repeat(1_000), + threadId, + runId: null, + itemId: earlierItemId, + providerThreadId, + status: "completed", + kind: "user_message", + }, + ], + coverage: "full", + omittedItems: 0, + }, + delivery: { nativeThreadId: "native", status: "injected" as const, itemIds: [earlierItemId] }, + createdByProviderInstanceId: providerInstanceId, + createdAt: now, + updatedAt: now, + }); + // The window keeps a ready handoff and one its rows show; the third is outside it. + const handoffs = [ + handoff("ready", "ready"), + handoff("shown", "superseded"), + handoff("old", "superseded"), + ]; + for (const payload of handoffs) { + yield* store.apply({ + id: EventId.make(`event:${payload.id}`), + type: "context-handoff.updated", + threadId, + occurredAt: now, + payload, + }); + } + yield* store.apply({ + id: EventId.make("event:handoff-window:item"), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make("item:handoff-window:handoff"), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "handoff", + contextHandoffId: handoffs[1]!.id, + fromProviderThreadIds: [providerThreadId], + toProviderThreadId: providerThreadId, + fromProviderInstanceIds: [providerInstanceId], + toProviderInstanceId: providerInstanceId, + strategy: "full_thread_summary", + }, + }); + + const window = yield* store.getThreadSnapshotWindow(threadId, { rowLimit: 10 }); + const windowHandoffs = window.projection.contextHandoffs; + assert.deepEqual( + windowHandoffs.map((row) => row.id), + [handoffs[0]!.id, handoffs[1]!.id], + ); + for (const row of windowHandoffs) { + assert.isUndefined(row.history); + assert.isUndefined(row.delivery); + } + const full = yield* store.getThreadSnapshot(threadId); + assert.deepEqual(full.projection.contextHandoffs, handoffs); + assert.deepEqual( + buildBoundedThreadStreamSnapshot(window), + buildBoundedThreadStreamSnapshot({ + ...window, + projection: { ...window.projection, contextHandoffs: handoffs.slice(0, 2) }, + }), + ); + }).pipe(Effect.provide(layerTest)), +); + it.effect("lists only active subagent IDs without loading shell data", () => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2a0fdc7518d1..292aa61ecdd6 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -472,6 +472,11 @@ export interface ProjectionStoreV2Shape { options: ProjectionTimelinePageOptions, ) => Effect.Effect; readonly getMessageCount: (threadId: ThreadId) => Effect.Effect; + /** Bytes of tool output a transcript export is sure to hold, or 0 when it can fit in `limit`. */ + readonly getTranscriptByteFloor: ( + threadId: ThreadId, + limit: number, + ) => Effect.Effect; readonly getSubagentCount: (threadId: ThreadId) => Effect.Effect; /** Child threads of the thread's app-owned tasks, in task order, without decoding the tasks. */ readonly listAppOwnedChildThreadIds: ( @@ -716,6 +721,7 @@ export interface ProjectionStoreV2Shape { }, ProjectionStoreV2Error >; + /** Context handoffs come without `history` and `delivery`; no window reader uses them. */ readonly getThreadSnapshotWindow: ( threadId: ThreadId, options: { @@ -4003,7 +4009,9 @@ export const layer: Layer.Layer = WHERE thread_id = ${threadId} ORDER BY rowid ASC ` : sql` - SELECT payload_json FROM orchestration_v2_projection_context_handoffs AS handoff + -- Megabytes on a long thread, and projectContextHandoffForWire drops both. + SELECT json_remove(payload_json, '$.history', '$.delivery') AS payload_json + FROM orchestration_v2_projection_context_handoffs AS handoff WHERE handoff.thread_id = ${threadId} AND ( handoff.status IN ('pending', 'ready') @@ -5915,6 +5923,41 @@ export const layer: Layer.Layer = Effect.map((rows) => rows[0]?.count ?? 0), Effect.mapError(controlReadError(threadId)), ); + // Tool output reaches the export unchanged, so the bytes of its text, also inside a structured + // result, are a floor for the export. Other stored bytes are not: decoding drops fields. A + // rolled back run is the only way the export hides a tool row, and rows a fork inherits only + // add to it. The output sum parses payloads, so it runs only when the tool rows exceed `limit`. + const getTranscriptByteFloor: ProjectionStoreV2Shape["getTranscriptByteFloor"] = ( + threadId, + limit, + ) => + sql<{ bytes: number }>` + WITH tool_rows AS ( + SELECT payload_json FROM orchestration_v2_projection_turn_items + WHERE thread_id = ${threadId} + AND type IN ('command_execution', 'dynamic_tool') + AND (run_id IS NULL OR run_id NOT IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} AND status = 'rolled_back' + )) + ) + SELECT CASE + WHEN (SELECT COALESCE(SUM(octet_length(payload_json)), 0) FROM tool_rows) <= ${limit} + THEN 0 + ELSE ( + -- json_valid is false past SQLite's nesting limit, where json_tree would fail. + SELECT COALESCE(SUM(CASE WHEN json_valid(payload_json) THEN ( + SELECT COALESCE(SUM(octet_length(leaf.value)), 0) + FROM json_tree(tool_rows.payload_json, '$.output') AS leaf + WHERE leaf.type = 'text' + ) ELSE 0 END), 0) + FROM tool_rows + ) + END AS bytes + `.pipe( + Effect.map((rows) => rows[0]?.bytes ?? 0), + Effect.mapError(controlReadError(threadId)), + ); const getSubagentCount: ProjectionStoreV2Shape["getSubagentCount"] = (threadId) => sql<{ count: number; @@ -7995,6 +8038,7 @@ export const layer: Layer.Layer = hasUnpairedRunInterruptRequest: readOnReader(hasUnpairedRunInterruptRequest), isNodeStored: readOnReader(isNodeStored), getMessageCount: readOnReader(getMessageCount), + getTranscriptByteFloor: readOnReader(getTranscriptByteFloor), getSubagentCount: readOnReader(getSubagentCount), listAppOwnedChildThreadIds: readOnReader(listAppOwnedChildThreadIds), hasPendingDelegatedCompletion: readOnReader(hasPendingDelegatedCompletion), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index a8bc76977a8a..34ff5a763110 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -239,6 +239,7 @@ it.effect( getThreadHistoryPage: () => Effect.die("unused"), getTimelinePage: () => Effect.die("Unused timeline read"), getMessageCount: () => Effect.die("unused message count"), + getTranscriptByteFloor: () => Effect.die("unused transcript byte floor"), hasPendingDelegatedCompletion: () => Effect.die("unused hasPendingDelegatedCompletion"), getSubagentCount: () => Effect.die("unused subagent count"), listAppOwnedChildThreadIds: () => Effect.die("unused listAppOwnedChildThreadIds"), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 4613d9d1f4f7..ab846252c9e5 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -890,6 +890,7 @@ it.effect( Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), Layer.provide( Layer.mock(Orchestrator.OrchestratorV2)({ + getTranscriptByteFloor: () => Effect.succeed(0), getThreadSnapshot: (threadId) => Queue.offer(started, threadId).pipe( Effect.andThen(Deferred.await(gates.get(threadId)!)), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 8a0d2d5df600..e1a1abdc1819 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -687,7 +687,15 @@ const make = Effect.gen(function* () { const transcriptExportPermits = yield* Semaphore.make(2); const getThreadTranscript = Effect.fn("orchestrationV2.threadManagement.getThreadTranscript")( function* (threadId: ThreadId) { - const { projection } = yield* getThreadSnapshot(threadId); + yield* ensureProjectionTranscript(threadId); + const byteFloor = yield* orchestrator.getTranscriptByteFloor( + threadId, + PROVIDER_SEND_TURN_MAX_FILE_BYTES, + ); + if (byteFloor > PROVIDER_SEND_TURN_MAX_FILE_BYTES) { + return yield* new ThreadTranscriptTooLargeError({ threadId }); + } + const { projection } = yield* orchestrator.getThreadSnapshot(threadId); const environment = yield* ServerEnvironment.ServerEnvironmentIdentity; const transcript = { threadId: projection.thread.id, diff --git a/apps/server/src/orchestration-v2/http.test.ts b/apps/server/src/orchestration-v2/http.test.ts index 570d34468d37..a32b4c63507d 100644 --- a/apps/server/src/orchestration-v2/http.test.ts +++ b/apps/server/src/orchestration-v2/http.test.ts @@ -9,6 +9,7 @@ import { ORCHESTRATION_PROTOCOL_HEADER, ORCHESTRATION_PROTOCOL_VERSION_TEXT, OrchestrationV2AppThread, + OrchestrationV2Run, OrchestrationV2TurnItem, OrchestrationV2ThreadTranscript, PROVIDER_SEND_TURN_MAX_FILE_BYTES, @@ -21,6 +22,8 @@ import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import * as HttpRouter from "effect/http/HttpRouter"; import * as HttpPlatform from "effect/http/HttpPlatform"; import * as Etag from "effect/http/Etag"; @@ -96,6 +99,24 @@ const items = Schema.decodeUnknownSync(Schema.Array(OrchestrationV2TurnItem))( })), ); +const rolledBackRun = Schema.decodeSync(OrchestrationV2Run)({ + id: "run:rolled-back", + threadId: "source-thread", + ordinal: 1, + providerInstanceId: "codex", + modelSelection: { instanceId: "codex", model: "gpt-5.4" }, + providerThreadId: null, + userMessageId: "message:rolled-back", + rootNodeId: null, + activeAttemptId: null, + status: "rolled_back", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, +}); + // Serves the orchestration HTTP routes over real stores; `allowed` gates the read scope. const serveOrchestrationHttp = (allowed: () => boolean) => Effect.gen(function* () { @@ -173,6 +194,18 @@ it.effect("exports full history with size, scope, and missing-thread checks", () ), ); + const readTranscriptError = Context.get( + services, + ThreadManagementService.ThreadManagementService, + ) + .getThreadTranscript(thread.id) + .pipe( + Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { + getEnvironmentId: Effect.succeed(environmentId), + }), + Effect.flip, + ); + const emptyResponse = yield* read(thread.id); expect(emptyResponse.status).toBe(200); expect( @@ -227,23 +260,80 @@ it.effect("exports full history with size, scope, and missing-thread checks", () }); const response = yield* read(thread.id); expect(response.status).toBe(expectedStatus); - if (expectedStatus === 400) { - const error = yield* Context.get(services, ThreadManagementService.ThreadManagementService) - .getThreadTranscript(thread.id) - .pipe( - Effect.provideService(ServerEnvironment.ServerEnvironmentIdentity, { - getEnvironmentId: Effect.succeed(environmentId), - }), - Effect.flip, - ); - expect(error).toBeInstanceOf(ThreadManagementService.ThreadTranscriptTooLargeError); - expect(yield* Effect.promise(() => response.json())).toMatchObject({ - _tag: "EnvironmentRequestInvalidError", - reason: "thread_transcript_too_large", - }); - } + if (expectedStatus === 200) continue; + expect(yield* readTranscriptError).toBeInstanceOf( + ThreadManagementService.ThreadTranscriptTooLargeError, + ); + expect(yield* Effect.promise(() => response.json())).toMatchObject({ + _tag: "EnvironmentRequestInvalidError", + reason: "thread_transcript_too_large", + }); } + const sql = Context.get(services, SqlClient.SqlClient); + const commandItemId = items.find((item) => item.type === "command_execution")!.id; + const setCommandPayload = (payload: Statement.Fragment) => sql` + UPDATE orchestration_v2_projection_turn_items SET payload_json = ${payload} + WHERE turn_item_id = ${commandItemId} + `; + const fill = (bytes: number) => sql`printf('%.*c', ${bytes}, 'x')`; + // Stored bytes the export drops do not count, and neither does a row SQLite cannot parse: + // only what the export keeps can reject it. + yield* setCommandPayload(sql`json_set( + payload_json, '$.output', 'x', + '$.droppedLegacyField', ${fill(PROVIDER_SEND_TURN_MAX_FILE_BYTES + 1)} + )`); + expect((yield* read(thread.id)).status).toBe(200); + const deepField = sql`'"droppedDeepField":' || printf('%.*c', 1001, '[') || printf('%.*c', 1001, ']') || ','`; + yield* setCommandPayload( + sql`replace(payload_json, '"droppedLegacyField":', ${deepField} || '"droppedLegacyField":')`, + ); + expect((yield* read(thread.id)).status).toBe(200); + // Tool output over the limit rejects the export, here as the content array a dynamic tool + // stores. Its only read is the size. + yield* sql` + UPDATE orchestration_v2_projection_turn_items SET type = 'dynamic_tool' + WHERE turn_item_id = ${commandItemId} + `; + yield* setCommandPayload(sql`json_set( + replace(payload_json, ${deepField}, ''), '$.type', 'dynamic_tool', '$.toolName', NULL, + '$.output', + json_array(json_object('type', 'text', 'text', ${fill(PROVIDER_SEND_TURN_MAX_FILE_BYTES + 1)})) + )`); + const reads: Array = []; + expect( + yield* readTranscriptError.pipe( + Effect.provideService(Statement.CurrentTransformer, (statement) => { + reads.push(statement.compile()[0]); + return Effect.succeed(statement); + }), + ), + ).toBeInstanceOf(ThreadManagementService.ThreadTranscriptTooLargeError); + expect(reads).toEqual([expect.stringContaining("octet_length(payload_json)")]); + expect((yield* read(thread.id)).status).toBe(400); + // The export leaves a rolled back run out, so its output cannot reject it. + yield* eventSink.write({ + events: [ + { + id: EventId.make("create-rolled-back-run"), + type: "run.created", + threadId: thread.id, + occurredAt: now, + payload: rolledBackRun, + }, + ], + }); + yield* sql` + UPDATE orchestration_v2_projection_turn_items SET run_id = ${rolledBackRun.id} + WHERE turn_item_id = ${commandItemId} + `; + yield* setCommandPayload(sql`json_set(payload_json, '$.runId', ${rolledBackRun.id})`); + const withoutRolledBackRun = yield* read(thread.id); + expect(withoutRolledBackRun.status).toBe(200); + expect( + (yield* decodeTranscript(yield* Effect.promise(() => withoutRolledBackRun.json()))).items, + ).toHaveLength(199); + expect((yield* read(ThreadId.make("missing"))).status).toBe(404); allowed = false; expect((yield* read(thread.id)).status).toBe(403); diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index e641933b313f..294189600fea 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -41,7 +41,8 @@ The v2 checks pin these invariants: requested. Explicit cross-environment transcript exports preserve full tool bodies so the destination agent can read the context. The export rejects timeline JSON above the file attachment limit - before sending it; the client still applies its attachment limit to the complete file. + before sending it, and before decoding any row when the tool output text alone is over that + limit; the client still applies its attachment limit to the complete file. - Live command output reaches a client only through `orchestration.subscribeCommandOutput`, which an expanded command row opens and closes with itself. Raw output deltas are never orchestration events: adapters send them to the @@ -71,8 +72,9 @@ send and caps the decoded bytes and rows of the bounded snapshot and one older p agent turn, and of the cold-connect shell for a sidebar of settled threads and idle subagents. It also caps the bytes and the node, subagent, plan, provider turn and checkpoint counts of the bounded snapshot for a thread of 40 finished runs behind one working run, so finished runs cannot -ride along with a window. When clients gain an opt-in, add it to the test's client queries and give -a fixture the data it saves. +ride along with a window. The working run has a ready context handoff, so its history cannot ride +along either. When clients gain an opt-in, add it to the test's client queries and give a fixture +the data it saves. ## Event store and startup