diff --git a/apps/server/integration/transferBudgetV2.integration.test.ts b/apps/server/integration/transferBudgetV2.integration.test.ts index 0ec8b1c37b291..8cc1a2c2521a6 100644 --- a/apps/server/integration/transferBudgetV2.integration.test.ts +++ b/apps/server/integration/transferBudgetV2.integration.test.ts @@ -42,6 +42,10 @@ import * as OrchestrationEventStore from "../src/persistence/OrchestrationEventS import * as EventStore from "../src/orchestration-v2/EventStore.ts"; import * as EventSink from "../src/orchestration-v2/EventSink.ts"; import * as ProjectionStore from "../src/orchestration-v2/ProjectionStore.ts"; +import * as Orchestrator from "../src/orchestration-v2/Orchestrator.ts"; +import * as ProviderSessionManager from "../src/orchestration-v2/ProviderSessionManager.ts"; +import * as CommandReceiptStore from "../src/orchestration-v2/CommandReceiptStore.ts"; +import * as ServerSettings from "../src/serverSettings.ts"; import * as ThreadManagementService from "../src/orchestration-v2/ThreadManagementService.ts"; import * as ProjectStore from "../src/orchestration-v2/ProjectStore.ts"; import * as ProjectService from "../src/project/ProjectService.ts"; @@ -81,18 +85,26 @@ const layerManagement = Layer.unwrap( Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; const sink = yield* EventSink.EventSinkV2; - return Layer.mock(ThreadManagementService.ThreadManagementService)({ - ensureLegacyTranscript: () => Effect.void, - getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), - getThreadSnapshotWindow: (id, options) => - projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), - getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie), - getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie), - readShellSnapshot: (options) => - projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie), - streamStoredEventsFrom: (input) => - sink.stream({ ...input, bounded: true }).pipe(Stream.orDie), - }); + return ThreadManagementService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), + Layer.mock(CommandReceiptStore.CommandReceiptStoreV2)({}), + Layer.mock(ServerSettings.ServerSettingsService)({}), + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), + getThreadSnapshotWindow: (id, options) => + projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), + getThreadShell: (id) => projections.getThreadShell(id).pipe(Effect.orDie), + getShellSnapshot: (options) => projections.getShellSnapshot(options).pipe(Effect.orDie), + readShellSnapshot: (options) => + projections.readShellSnapshot(options).pipe(Effect.map(Effect.orDie), Effect.orDie), + streamStoredEventsFrom: (input) => + sink.stream({ ...input, bounded: true }).pipe(Stream.orDie), + }), + ), + ), + ); }), ); const layerEnrichment = Layer.unwrap( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 4464d09916e33..e364a4e7637aa 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2366,11 +2366,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events, command, )({ - type: "thread.visited", + type: "thread.visit-recorded", threadId: command.threadId, providerInstanceId: thread.providerInstanceId, occurredAt: yield* DateTime.now, - payload: movesForward ? { ...thread, lastVisitedAt: visitedAt.value } : thread, + payload: { lastVisitedAt: movesForward ? visitedAt.value : thread.lastVisitedAt }, }); }); @@ -3282,11 +3282,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events, command, )({ - type: "thread.pull-request-link-updated", + ...(command.type === "thread.pull-request-link.sync" + ? { + type: "thread.pull-request-link-synced" as const, + payload: { + ...normalizeThreadPullRequestKey(command), + snapshot: command.snapshot, + stack: command.stack, + }, + } + : { type: "thread.pull-request-link-updated" as const, payload: link }), threadId: command.threadId, providerInstanceId: updatedThread.providerInstanceId, occurredAt: now, - payload: link, }); return; } diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index f8acaf065cb93..3cbbe183cce3f 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -217,11 +217,14 @@ export const layer: Layer.Layer< "thread.model-selection-updated", "thread.provider-switched", ]; - // Read-state events carry the whole thread too, but client reducers apply - // them without bumping updatedAt. They supersede only each other: a newer + // Read-state events do not bump updatedAt. They supersede only each other: a newer // visit must not delete the rename that carries the activity timestamp, // while any newer activity-bearing event supersedes both groups. - const READ_STATE_THREAD_EVENT_TYPES = ["thread.visited", "thread.marked-unread"]; + const READ_STATE_THREAD_EVENT_TYPES = [ + "thread.visited", + "thread.marked-unread", + "thread.visit-recorded", + ]; const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES); const readStateThreadEventTypes = new Set(READ_STATE_THREAD_EVENT_TYPES); @@ -280,7 +283,7 @@ export const layer: Layer.Layer< AND event.event_type IN ('message.updated', 'node.updated') THEN json_extract(event.payload_json, '$.id') WHEN event.application_event_version = 2 - AND event.event_type = 'thread.pull-request-link-updated' + AND event.event_type IN ('thread.pull-request-link-updated', 'thread.pull-request-link-synced') THEN json_array(json_extract(event.payload_json, '$.host'), json_extract(event.payload_json, '$.repository'), json_extract(event.payload_json, '$.number')) @@ -322,13 +325,15 @@ export const layer: Layer.Layer< obsolete.push(row.sequence); } else retainedReadStateThreadIds.add(row.stream_id); } else if ( - row.event_type === "thread.pull-request-link-updated" && + (row.event_type === "thread.pull-request-link-updated" || + row.event_type === "thread.pull-request-link-synced") && retainedThreadIds.has(row.stream_id) ) { // A newer whole-thread payload already carries this link's state. obsolete.push(row.sequence); } else if ( row.event_type === "thread.pull-request-link-updated" || + row.event_type === "thread.pull-request-link-synced" || row.event_type === "message.updated" || row.event_type === "node.updated" ) { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 27f20ea7673ca..09f99b59c2d15 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -13,7 +13,10 @@ import { threadErrorSummary, usageLimitRunPresentedAsLatest, } from "@t3tools/shared/orchestrationV2ThreadError"; -import { threadPullRequestsOf } from "@t3tools/shared/threadPullRequests"; +import { + applyThreadPullRequestLinkSync, + threadPullRequestsOf, +} from "@t3tools/shared/threadPullRequests"; import type { OrchestrationV2AppThread, OrchestrationV2CheckpointScope, @@ -758,12 +761,23 @@ export function applyToProjection( ...base, thread: event.payload, }; + // A host refresh is not activity, matching the stored projection. + case "thread.pull-request-link-synced": + return { + ...projection, + thread: applyThreadPullRequestLinkSync(projection.thread, event.payload), + }; case "thread.visited": case "thread.marked-unread": return { ...projection, thread: event.payload, }; + case "thread.visit-recorded": + return { + ...projection, + thread: { ...projection.thread, lastVisitedAt: event.payload.lastVisitedAt }, + }; case "run.created": case "run.updated": return withLocalVisibleTurnItems({ @@ -1930,6 +1944,28 @@ export const layer: Layer.Layer = `; break; } + case "thread.visit-recorded": { + const visitedAt = event.payload.lastVisitedAt; + yield* sql` + UPDATE orchestration_v2_projection_threads + SET payload_json = json_set(payload_json, '$.lastVisitedAt', + ${visitedAt === null ? null : DateTime.formatIso(visitedAt)}) + WHERE thread_id = ${event.threadId} + `; + break; + } + case "thread.pull-request-link-synced": { + const thread = yield* getThread(event.threadId); + const updated = applyThreadPullRequestLinkSync(thread, event.payload); + if (updated === thread) break; + const payloadJson = yield* encodeThreadPayload(updated); + yield* sql` + UPDATE orchestration_v2_projection_threads + SET payload_json = ${payloadJson} + WHERE thread_id = ${event.threadId} + `; + break; + } case "run.created": case "run.updated": { const payloadJson = yield* encodeRunPayload(event.payload); @@ -2704,6 +2740,9 @@ export const layer: Layer.Layer = event.type !== "thread.unpinned" && event.type !== "thread.pin-reordered" && event.type !== "thread.visited" && + event.type !== "thread.visit-recorded" && + event.type !== "thread.pull-request-link-synced" && + event.type !== "thread.pull-request-synced" && event.type !== "thread.marked-unread" && event.type !== "thread.metadata-updated" && event.type !== "thread.runtime-mode-updated" && diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 685dbf851feb5..8dfc55b4afab6 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -433,6 +433,8 @@ export const make = Effect.gen(function* () { switch (event.type) { case "thread.pull-request-link-updated": return event.payload.snapshot === null ? requestSync(event.payload) : Effect.void; + case "thread.pull-request-link-synced": + return Effect.void; case "thread.pull-request-synced": return Effect.forEach( visibleThreadPullRequests(event.payload.pullRequests ?? []).filter( diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index d83dd9578c279..55c0d9611fa60 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -21,6 +21,8 @@ import { type OrchestrationV2ConversationMessage, type OrchestrationV2CreationSource, type OrchestrationV2Run, + type OrchestrationV2SubscribeThreadInput, + type OrchestrationV2ThreadStreamItem, type OrchestrationV2ThreadShellSnapshot, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadTranscript, @@ -49,11 +51,30 @@ import * as Stream from "effect/Stream"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; -import { projectedRowEncodedBytes } from "./threadHistoryPaging.ts"; +import { + projectedRowEncodedBytes, + THREAD_HISTORY_PAGE_POLICY, + THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, +} from "./threadHistoryPaging.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import { DispatchModeLimit, exceededDispatchModeLimit } from "./DispatchModeLimit.ts"; -import { projectTurnItemForDetail } from "./WireProjection.ts"; +import { + projectTurnItemForDetail, + projectDomainEventForWire, + projectThreadProjectionForWire, +} from "./WireProjection.ts"; +import { rpcInitialItems } from "../rpcInitialItems.ts"; +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; +import { coalesceThreadLiveStream } from "./ThreadLiveEventCoalescer.ts"; +import { + buildBoundedThreadStreamSnapshot, + decideThreadResume, + isThreadReplayRawPayloadSafe, + shouldUseBoundedThreadSnapshot, + threadReplayEncodedBytes, + THREAD_RESUME_MAX_REPLAY_EVENTS, +} from "./ThreadStream.ts"; import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts"; import * as ServerSettings from "../serverSettings.ts"; @@ -310,7 +331,50 @@ export type ThreadManagementError = typeof ThreadManagementError.Type; type ThreadManagementFailure = ThreadManagementError | Orchestrator.OrchestratorV2Error; +export class ThreadManagementThreadStreamError extends Schema.TaggedError()( + "ThreadManagementThreadStreamError", + { + threadId: ThreadId, + operation: Schema.Literals([ + "hydrate", + "stream", + "replay", + "snapshot", + "prepareReplay", + "measureReplay", + "locateRecreated", + ]), + cause: Schema.Defect(), + }, +) { + override get message(): string { + switch (this.operation) { + case "hydrate": + return `Failed to hydrate migrated thread ${this.threadId}`; + case "stream": + return `Failed while streaming orchestration V2 thread ${this.threadId}`; + case "replay": + return `Failed while replaying orchestration V2 thread ${this.threadId}`; + case "snapshot": + return `Failed to load orchestration V2 thread ${this.threadId}`; + case "prepareReplay": + return `Failed to prepare orchestration V2 thread ${this.threadId} replay`; + case "measureReplay": + return `Failed to measure orchestration V2 thread ${this.threadId} replay`; + case "locateRecreated": + return `Failed to locate recreated orchestration V2 thread ${this.threadId}`; + } + } +} + export interface ThreadManagementServiceShape { + readonly streamThreadEventsForClient: ( + input: OrchestrationV2SubscribeThreadInput, + ) => Effect.Effect< + Stream.Stream, + ThreadManagementThreadStreamError, + OrchestrationEventStore.OrchestrationEventStore + >; readonly searchThreadStream: ( input: OrchestrationV2SearchThreadInput, ) => Stream.Stream; @@ -642,6 +706,232 @@ const make = Effect.gen(function* () { Effect.andThen(orchestrator.getThreadSnapshotWindow(threadId, options)), ); + const streamThreadEventsForClient = Effect.fn( + "orchestrationV2.threadManagement.streamThreadEventsForClient", + )(function* (input: OrchestrationV2SubscribeThreadInput) { + const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; + + yield* ensureLegacyTranscript(input.threadId).pipe( + Effect.mapError( + (cause) => + new ThreadManagementThreadStreamError({ + threadId: input.threadId, + operation: "hydrate", + cause, + }), + ), + ); + + const eventStreamFrom = (afterSequence: number) => + orchestrator + .streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + }) + .pipe( + Stream.mapEffect((stored) => + Effect.gen(function* () { + const event = stored.event; + if ( + input.acceptThreadFieldEvents !== true && + (event.type === "thread.pull-request-link-synced" || + event.type === "thread.visit-recorded") + ) { + // Older clients need the legacy full-thread event. The live stream reads + // committed metadata only; historical compact events use a fresh snapshot below. + const { thread } = yield* ensureProjectionTranscript(input.threadId).pipe( + Effect.andThen(orchestrator.getThreadRecords(input.threadId, [])), + ); + return { + kind: "event" as const, + sequence: stored.sequence, + event: projectDomainEventForWire({ + ...event, + type: + event.type === "thread.visit-recorded" + ? "thread.visited" + : "thread.pull-request-synced", + payload: thread, + }), + }; + } + return { + kind: "event" as const, + sequence: stored.sequence, + event: projectDomainEventForWire(event), + }; + }), + ), + coalesceThreadLiveStream, + Stream.mapError( + (cause) => + new ThreadManagementThreadStreamError({ + threadId: input.threadId, + operation: "stream", + cause, + }), + ), + ); + + const loadReplayThrough = (afterSequence: number, throughSequence: number) => + applicationEvents + .readAgentEvents({ + threadId: input.threadId, + afterSequence, + throughSequence, + limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1, + // A row a newer build wrote (e.g. after a downgrade) must not fail + // every resubscribe. The live tail resumes from `highWater`, so a + // dropped row never hides later events. + skipUnknownEventTypes: true, + }) + .pipe( + Stream.map((stored) => ({ + kind: "event" as const, + sequence: stored.sequence, + event: projectDomainEventForWire(stored.event), + })), + Stream.runCollect, + Effect.map((items) => Array.from(items)), + Effect.mapError( + (cause) => + new ThreadManagementThreadStreamError({ + threadId: input.threadId, + operation: "replay", + cause, + }), + ), + ); + + const completionMarker = + input.requestCompletionMarker === true + ? Stream.make({ kind: "synchronized" as const }) + : Stream.empty; + + const snapshotThenLive = Effect.fn("orchestrationV2.threadManagement.threadSnapshotThenLive")( + function* () { + const useBoundedSnapshot = shouldUseBoundedThreadSnapshot(input); + const snapshot = yield* ( + useBoundedSnapshot + ? getThreadSnapshotWindow(input.threadId, { + rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, + userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, + }) + : getThreadSnapshot(input.threadId) + ).pipe( + Effect.mapError( + (cause) => + new ThreadManagementThreadStreamError({ + threadId: input.threadId, + operation: "snapshot", + cause, + }), + ), + ); + const { snapshotSequence } = snapshot; + const snapshotItem = useBoundedSnapshot + ? buildBoundedThreadStreamSnapshot({ + ...snapshot, + compactTurnItems: input.acceptCompactTurnItems === true, + }) + : { + kind: "snapshot" as const, + snapshotSequence, + projection: projectThreadProjectionForWire(snapshot.projection), + }; + return Stream.concat( + Stream.concat(rpcInitialItems([snapshotItem]), completionMarker), + eventStreamFrom(snapshotSequence), + ); + }, + ); + + // The event sink subscribes to live events before reading the persisted + // tail, so no event published during the replay window is lost; overlapping + // events are deduped by sequence on the client. + if (input.afterSequence !== undefined) { + const highWater = yield* applicationEvents.latestAgentSequence(input.threadId).pipe( + Effect.mapError( + (cause) => + new ThreadManagementThreadStreamError({ + threadId: input.threadId, + operation: "prepareReplay", + cause, + }), + ), + ); + if (input.afterSequence > highWater) { + return yield* snapshotThenLive(); + } + const stats = yield* applicationEvents + .getAgentReplayStats({ + threadId: input.threadId, + afterSequence: input.afterSequence, + throughSequence: highWater, + maxEvents: THREAD_RESUME_MAX_REPLAY_EVENTS, + }) + .pipe( + Effect.mapError( + (cause) => + new ThreadManagementThreadStreamError({ + threadId: input.threadId, + operation: "measureReplay", + cause, + }), + ), + ); + // Bound stored JSON before decoding, then check projected event + // size separately. Neither byte count is a bound on process memory. + if ( + stats.eventCount > THREAD_RESUME_MAX_REPLAY_EVENTS || + !isThreadReplayRawPayloadSafe(stats.rawPayloadBytes) + ) { + return yield* snapshotThenLive(); + } + if (stats.hasCreateEvent) { + const shell = yield* orchestrator.getThreadShell(input.threadId).pipe( + Effect.mapError( + (cause) => + new ThreadManagementThreadStreamError({ + threadId: input.threadId, + operation: "locateRecreated", + cause, + }), + ), + ); + // A retained creation can belong to a thread already deleted. + // Only replace its bounded replay when a snapshot can exist. + if (shell !== null) return yield* snapshotThenLive(); + } + const replay = yield* loadReplayThrough(input.afterSequence, highWater); + if ( + input.acceptThreadFieldEvents !== true && + replay.some( + (item) => + item.event.type === "thread.pull-request-link-synced" || + item.event.type === "thread.visit-recorded", + ) + ) { + return yield* snapshotThenLive(); + } + const plan = decideThreadResume({ + afterSequence: input.afterSequence, + highWater, + replayEventCount: replay.length, + replayEncodedBytes: threadReplayEncodedBytes(replay), + }); + if (plan.mode === "snapshot") { + return yield* snapshotThenLive(); + } + return Stream.concat( + Stream.concat(rpcInitialItems(replay), completionMarker), + eventStreamFrom(highWater), + ); + } + + return yield* snapshotThenLive(); + }); + const assertDispatchTargetEnabled = (command: OrchestrationV2ServerCommand) => { switch (command.type) { case "message.dispatch": @@ -1119,6 +1409,7 @@ const make = Effect.gen(function* () { }); return ThreadManagementService.of({ + streamThreadEventsForClient, ensureLegacyTranscript, searchThreadStream: (input) => Stream.unwrap( diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 84454ff993bb2..12af97ff159fb 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -602,6 +602,7 @@ export const make = Effect.gen(function* () { return cleanUpSettledThread(event.threadId); case "thread.pull-request-link-updated": case "thread.pull-request-synced": + case "thread.pull-request-link-synced": case "provider-session.detached": return worker.enqueue(event.threadId); case "provider-session.updated": diff --git a/apps/server/src/orchestration-v2/ThreadStream.ts b/apps/server/src/orchestration-v2/ThreadStream.ts index 4ef21edd3b3c7..c4885f7bcd6a8 100644 --- a/apps/server/src/orchestration-v2/ThreadStream.ts +++ b/apps/server/src/orchestration-v2/ThreadStream.ts @@ -11,6 +11,12 @@ import { } from "./threadHistoryPaging.ts"; import { projectThreadProjectionForWire } from "./WireProjection.ts"; +export function shouldUseBoundedThreadSnapshot(input: { + readonly acceptBoundedSnapshot?: boolean; +}): boolean { + return input.acceptBoundedSnapshot === true; +} + export const THREAD_RESUME_MAX_REPLAY_EVENTS = 128; export const THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES = 1_048_576; diff --git a/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts index 9519b661cbf3f..8375a13f88560 100644 --- a/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts +++ b/apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts @@ -43,6 +43,10 @@ import { subscribeOrchestrationV2Thread } from "../ws.ts"; import * as OrchestrationHttp from "./http.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProjectStore from "./ProjectStore.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; +import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as ServerSettings from "../serverSettings.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; const decodeBounded = Schema.decodeUnknownSync( @@ -66,11 +70,22 @@ const store = Layer.mergeAll( ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)), SqlitePersistence.layerMemory, ); +const threadManagementLayer = (overrides: Partial) => + Layer.fresh(ThreadManagementService.layer).pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)(overrides), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), + Layer.mock(CommandReceiptStore.CommandReceiptStoreV2)({}), + Layer.mock(ServerSettings.ServerSettingsService)({}), + ), + ), + ); + const management = Layer.unwrap( Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; - return Layer.mock(ThreadManagementService.ThreadManagementService)({ - ensureLegacyTranscript: () => Effect.void, + return threadManagementLayer({ getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), getThreadSnapshotWindow: (id, options) => projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), @@ -278,7 +293,226 @@ function itemIds(projection: OrchestrationV2ThreadProjection): string[] { return projection.turnItems.map((item) => String(item.id)); } +it.effect.each(["thread.visit-recorded", "thread.pull-request-link-synced"] as const)( + "resumes persisted %s with a current legacy snapshot or opted-in replay", + (type) => + Effect.gen(function* () { + const events = yield* OrchestrationEventStore.OrchestrationEventStore; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const persist = Effect.fnUntraced(function* (event: OrchestrationV2DomainEvent) { + const [stored] = yield* events.appendAgentEvents({ events: [event] }); + yield* projections.apply(event); + return stored!; + }); + const at = yield* DateTime.now; + const refreshedAt = DateTime.add(at, { seconds: 1 }); + const renamedAt = DateTime.add(at, { seconds: 2 }); + yield* persist(threadCreated(PARENT, at)); + const thread = yield* projections.getThread(PARENT); + const link = { + host: "github.com", + repository: "acme/app", + number: 1, + url: "https://github.com/acme/app/pull/1", + source: "manual" as const, + linkedAt: DateTime.formatIso(at), + snapshot: null, + stack: null, + }; + const before = yield* persist({ + id: EventId.make("resume:link"), + type: "thread.pull-request-synced", + threadId: PARENT, + occurredAt: at, + payload: { ...thread, pullRequests: [link] }, + }); + const snapshot = { + state: "merged" as const, + title: "Merged", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: DateTime.formatIso(refreshedAt), + syncedAt: DateTime.formatIso(refreshedAt), + }; + const compact = yield* persist({ + id: EventId.make("resume:compact"), + threadId: PARENT, + occurredAt: refreshedAt, + ...(type === "thread.visit-recorded" + ? { type, payload: { lastVisitedAt: refreshedAt } } + : { + type, + payload: { + host: link.host, + repository: link.repository, + number: link.number, + snapshot, + stack: null, + }, + }), + }); + const refreshed = yield* projections.getThread(PARENT); + const expected = { + ...thread, + title: "Renamed after refresh", + updatedAt: renamedAt, + lastVisitedAt: type === "thread.visit-recorded" ? refreshedAt : null, + pullRequests: [type === "thread.pull-request-link-synced" ? { ...link, snapshot } : link], + }; + const renamed = yield* persist({ + id: EventId.make("resume:rename"), + type: "thread.metadata-updated", + threadId: PARENT, + occurredAt: renamedAt, + payload: { ...refreshed, title: expected.title, updatedAt: renamedAt }, + }); + + for (const acceptThreadFieldEvents of [undefined, false, true]) { + const items = yield* subscribeOrchestrationV2Thread({ + threadId: PARENT, + afterSequence: before.sequence, + requestCompletionMarker: true, + ...(acceptThreadFieldEvents === undefined ? {} : { acceptThreadFieldEvents }), + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.runCollect, + ), + ), + Effect.map((items) => + items.map((item) => + decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(item)))), + ), + ), + ); + if (acceptThreadFieldEvents === true) { + assert.deepEqual(items, [ + { kind: "event", sequence: compact.sequence, event: compact.event }, + { kind: "event", sequence: renamed.sequence, event: renamed.event }, + { kind: "synchronized" }, + ]); + } else { + assert.deepEqual( + items.map((item) => item.kind), + ["snapshot", "synchronized"], + ); + const first = items[0]; + if (first?.kind !== "snapshot") return assert.fail("missing reconnect snapshot"); + assert.strictEqual(first.snapshotSequence, renamed.sequence); + assert.deepEqual(first.projection.thread, expected); + } + } + }).pipe( + Effect.provide( + Layer.mergeAll(management, OrchestrationEventStore.layer).pipe(Layer.provideMerge(store)), + ), + ), +); + it.layer(TestLayer)("compact bounded snapshot transport", (it) => { + it.effect("delivers compact visits and PR refreshes to current and legacy subscribers", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const at = yield* DateTime.now; + const thread = yield* projections.getThread(PARENT); + const link = { + host: "github.com", + repository: "acme/app", + number: 1, + url: "https://github.com/acme/app/pull/1", + source: "manual" as const, + linkedAt: DateTime.formatIso(at), + snapshot: null, + stack: null, + }; + yield* projections.apply({ + id: EventId.make("compat:link"), + type: "thread.pull-request-synced", + threadId: PARENT, + occurredAt: at, + payload: { ...thread, pullRequests: [link] }, + }); + const snapshot = { + state: "merged" as const, + title: "Merged", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: DateTime.formatIso(at), + syncedAt: DateTime.formatIso(at), + }; + const events: OrchestrationV2DomainEvent[] = [ + { + id: EventId.make("compat:visit"), + type: "thread.visit-recorded", + threadId: PARENT, + occurredAt: at, + payload: { lastVisitedAt: at }, + }, + { + id: EventId.make("compat:sync"), + type: "thread.pull-request-link-synced", + threadId: PARENT, + occurredAt: at, + payload: { + host: link.host, + repository: link.repository, + number: link.number, + snapshot, + stack: null, + }, + }, + ]; + for (const event of events) yield* projections.apply(event); + const expected = yield* projections.getThread(PARENT); + assert.deepEqual(expected.lastVisitedAt, at); + assert.deepEqual(expected.pullRequests, [{ ...link, snapshot }]); + assert.deepEqual(expected.updatedAt, thread.updatedAt); + + const liveManagement = threadManagementLayer({ + getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie), + getThreadSnapshotWindow: (id, options) => + projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie), + getThreadRecords: (id, fields, filter) => + projections.getThreadRecords(id, fields, filter).pipe(Effect.orDie), + streamStoredEventsFrom: () => + Stream.fromIterable( + events.map((event, index) => ({ + sequence: index + 1, + commandId: null, + event, + })), + ), + }); + for (const acceptThreadFieldEvents of [false, true]) { + const items = yield* subscribeOrchestrationV2Thread({ + threadId: PARENT, + acceptThreadFieldEvents, + }).pipe( + Effect.flatMap((stream) => Stream.take(stream, 3).pipe(Stream.runCollect)), + Effect.provide(liveManagement), + ); + const decoded = items + .slice(1) + .map((item) => decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(item))))); + assert.deepEqual( + decoded.map((item) => (item.kind === "event" ? item.event.type : item.kind)), + acceptThreadFieldEvents + ? ["thread.visit-recorded", "thread.pull-request-link-synced"] + : ["thread.visited", "thread.pull-request-synced"], + ); + if (!acceptThreadFieldEvents) { + for (const item of decoded) { + if (item.kind !== "event") return assert.fail("missing event"); + assert.deepEqual(item.event.payload, expected); + } + } + } + }), + ); + describe.each([ ["fork", FORK], ["long run with a retained interrupt request", LONG], diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 1b6a7de8a944f..cecddcb16998d 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -5387,6 +5387,7 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", }); } const before = yield* orchestrator.getThreadShell(threadId); + yield* TestClock.adjust("1 second"); yield* orchestrator.dispatch({ type: "thread.pull-request-link.sync", commandId: CommandId.make("compact-pr-sync"), @@ -5415,12 +5416,13 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", SELECT event_type, payload_json FROM orchestration_events WHERE command_id = ${"compact-pr-sync"} `; + const after = yield* orchestrator.getThreadShell(threadId); + assert.deepEqual(after?.updatedAt, before?.updatedAt); assert.equal(rows.length, 1); - assert.equal(rows[0]?.event_type, "thread.pull-request-link-updated"); + assert.equal(rows[0]?.event_type, "thread.pull-request-link-synced"); const payload = JSON.parse(rows[0]!.payload_json); assert.equal(payload.number, 2); assert.isUndefined(payload.pullRequests); - const after = yield* orchestrator.getThreadShell(threadId); assert.equal(after?.pullRequests?.length, 3); assert.deepEqual(after?.pullRequests?.[0], before?.pullRequests?.[0]); assert.deepEqual(after?.pullRequests?.[2], before?.pullRequests?.[2]); @@ -5458,7 +5460,7 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", yield* maintenance.compactEventStore; const linkEvents = yield* sql<{ command_id: string }>` SELECT command_id FROM orchestration_events - WHERE stream_id = ${threadId} AND event_type = 'thread.pull-request-link-updated' + WHERE stream_id = ${threadId} AND event_type = 'thread.pull-request-link-synced' ORDER BY sequence `; assert.deepEqual( @@ -5477,7 +5479,7 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", const threadEvents = () => sql<{ event_type: string }>` SELECT event_type FROM orchestration_events WHERE stream_id = ${threadId} - AND event_type IN ('thread.metadata-updated', 'thread.visited') + AND event_type IN ('thread.metadata-updated', 'thread.visit-recorded') ORDER BY sequence `; yield* orchestrator.dispatch({ @@ -5500,8 +5502,17 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", yield* maintenance.compactEventStore; assert.deepEqual( (yield* threadEvents()).map((row) => row.event_type), - ["thread.metadata-updated", "thread.visited"], + ["thread.metadata-updated", "thread.visit-recorded"], ); + const [visit] = yield* sql<{ payload_json: string }>` + SELECT payload_json FROM orchestration_events + WHERE command_id = 'compact-pr-visit-again'`; + assert.deepEqual(JSON.parse(visit!.payload_json), { + lastVisitedAt: "2026-10-13T15:00:00.000Z", + }); + const visited = yield* orchestrator.getThreadShell(threadId); + assert.isTrue((yield* maintenance.rebuild).valid); + assert.deepEqual(yield* orchestrator.getThreadShell(threadId), visited); yield* orchestrator.dispatch({ type: "thread.metadata.update", commandId: CommandId.make("compact-pr-rename-again"), @@ -5513,6 +5524,9 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", (yield* threadEvents()).map((row) => row.event_type), ["thread.metadata-updated"], ); + const renamed = yield* orchestrator.getThreadShell(threadId); + assert.isTrue((yield* maintenance.rebuild).valid); + assert.deepEqual(yield* orchestrator.getThreadShell(threadId), renamed); }), ); }); diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 285f2cfe1f9bc..a4491ee88ee01 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -178,6 +178,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( const shellReads: ThreadId[] = []; const catchUp = { shellSnapshotReads: 0 }; const threads = ThreadManagementService.ThreadManagementService.of({ + streamThreadEventsForClient: () => Effect.succeed(Stream.empty), getThreadShell: (threadId) => Effect.sync(() => shellReads.push(threadId)).pipe( Effect.andThen(options.readShell?.(threadId) ?? Ref.get(currentShell)), diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index fb21aa6e70dd4..ae27c061df263 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -93,6 +93,7 @@ export function shouldPublishAgentAwarenessEvent( case "thread.metadata-updated": case "thread.pull-request-link-updated": case "thread.pull-request-synced": + case "thread.pull-request-link-synced": case "thread.model-selection-updated": case "thread.provider-switched": case "run.created": @@ -111,6 +112,7 @@ export function shouldPublishAgentAwarenessEvent( case "thread.pin-reordered": case "thread.active-reordered": case "thread.visited": + case "thread.visit-recorded": case "thread.marked-unread": case "thread.runtime-mode-updated": case "thread.interaction-mode-updated": diff --git a/apps/server/src/testkit/RpcInitialMemory.fixture.mjs b/apps/server/src/testkit/RpcInitialMemory.fixture.mjs index a9648abf67155..be8c2d78797d4 100644 --- a/apps/server/src/testkit/RpcInitialMemory.fixture.mjs +++ b/apps/server/src/testkit/RpcInitialMemory.fixture.mjs @@ -13,12 +13,18 @@ const [Effect, Stream, Layer, Deferred, Fiber] = await Promise.all( ["Effect", "Stream", "Layer", "Deferred", "Fiber"].map(load), ); const app = (file) => import(NodeURL.pathToFileURL(root + "/apps/server/src/" + file + ".ts")); -const [Prefix, Ws, Threads, Events] = await Promise.all([ - app("rpcInitialItems"), - app("ws"), - app("orchestration-v2/ThreadManagementService"), - app("persistence/OrchestrationEventStore"), -]); +const [Prefix, Ws, Threads, Events, Orchestrator, Sessions, Receipts, Settings] = await Promise.all( + [ + app("rpcInitialItems"), + app("ws"), + app("orchestration-v2/ThreadManagementService"), + app("persistence/OrchestrationEventStore"), + app("orchestration-v2/Orchestrator"), + app("orchestration-v2/ProviderSessionManager"), + app("orchestration-v2/CommandReceiptStore"), + app("serverSettings"), + ], +); const refs = []; const fibers = []; const checkpoints = []; @@ -37,20 +43,28 @@ function start(ready) { Stream.concat(Stream.never), ); const dependencies = Layer.mergeAll( - Layer.mock(Threads.ThreadManagementService)({ - ensureLegacyTranscript: () => Effect.void, - getThreadSnapshot: () => - Effect.sync(() => ({ - snapshotSequence: 1, - projection: { - messages: [history()], - contextHandoffs: [], - turnItems: [], - visibleTurnItems: [], - }, - })), - streamStoredEventsFrom: () => live, - }), + Threads.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadSnapshot: () => + Effect.sync(() => ({ + snapshotSequence: 1, + projection: { + messages: [history()], + contextHandoffs: [], + turnItems: [], + visibleTurnItems: [], + }, + })), + streamStoredEventsFrom: () => live, + }), + Layer.mock(Sessions.ProviderSessionManagerV2)({}), + Layer.mock(Receipts.CommandReceiptStoreV2)({}), + Layer.mock(Settings.ServerSettingsService)({}), + ), + ), + ), Layer.mock(Events.OrchestrationEventStore)({ latestAgentSequence: () => Effect.succeed(1), getAgentReplayStats: () => diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 51b704dbda39f..9223fd0217a7b 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -37,6 +37,10 @@ import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as EventStore from "./orchestration-v2/EventStore.ts"; import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; +import { shouldUseBoundedThreadSnapshot } from "./orchestration-v2/ThreadStream.ts"; +import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; +import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; +import * as CommandReceiptStore from "./orchestration-v2/CommandReceiptStore.ts"; import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; @@ -47,13 +51,24 @@ import * as ProjectService from "./project/ProjectService.ts"; import { hasCompatibleOrchestrationProtocol, resolveAvailableEditorsForConfig, - shouldUseBoundedThreadSnapshot, subscribeOrchestrationV2Shell, subscribeOrchestrationV2Thread, subscribeServerConfig, withLateEditorConfig, } from "./ws.ts"; +const threadManagementLayer = (overrides: Partial) => + ThreadManagement.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)(overrides), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), + Layer.mock(CommandReceiptStore.CommandReceiptStoreV2)({}), + Layer.mock(ServerSettings.ServerSettingsService)({}), + ), + ), + ); + it("accepts only the current orchestration protocol before websocket RPC setup", () => { assert.isTrue( hasCompatibleOrchestrationProtocol( @@ -404,21 +419,23 @@ it.live.each(["thread", "shell"] as const)( }).pipe(Effect.flatMap(consume)); }); const items = yield* subscription.pipe( - Effect.provideService(ThreadManagement.ThreadManagementService, { - ensureLegacyTranscript: () => Effect.void, - streamStoredEventsFrom: ( - input: Parameters< - ThreadManagement.ThreadManagementService["Service"]["streamStoredEventsFrom"] - >[0], - ) => { - assert.equal(input?.afterSequence, highWater!.sequence); - return store - .streamApplicationEvents({ afterSequence: input?.afterSequence ?? 0 }) - .pipe( - Stream.filter((event): event is OrchestrationV2StoredEvent => "event" in event), - ); - }, - } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provide( + threadManagementLayer({ + streamStoredEventsFrom: ( + input: Parameters< + ThreadManagement.ThreadManagementService["Service"]["streamStoredEventsFrom"] + >[0], + ) => { + assert.equal(input?.afterSequence, highWater!.sequence); + return store + .streamApplicationEvents({ afterSequence: input?.afterSequence ?? 0 }) + .pipe( + Stream.filter((event): event is OrchestrationV2StoredEvent => "event" in event), + Stream.orDie, + ); + }, + }), + ), Effect.provideService(ProjectStore.ProjectStoreV2, { listShells: () => Effect.succeed([]), } as unknown as ProjectStore.ProjectStoreV2["Service"]), @@ -506,10 +523,11 @@ it.effect("thread resume across compacted history keeps message order and final Stream.runCollect, ), ), - Effect.provideService(ThreadManagement.ThreadManagementService, { - ensureLegacyTranscript: () => Effect.void, - streamStoredEventsFrom: () => Stream.never, - } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provide( + threadManagementLayer({ + streamStoredEventsFrom: () => Stream.never, + }), + ), ); assert.deepEqual( Array.from(items, (item) => diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0980882707db4..afceec6e5ff07 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -41,6 +41,7 @@ import { type FileManagerRevealKind, type OrchestrationClientOrigin, type OrchestrationV2Command, + type OrchestrationV2SubscribeThreadInput, type GitActionProgressEvent, type GitManagerServiceError, type MessageId, @@ -142,23 +143,11 @@ import { } from "./orchestration-v2/ShellStream.ts"; import { ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION } from "./orchestration-v2/ProjectionStore.ts"; import { bufferLiveStream } from "./orchestration-v2/LiveStreamBudget.ts"; -import { coalesceThreadLiveStream } from "./orchestration-v2/ThreadLiveEventCoalescer.ts"; -import { - buildBoundedThreadStreamSnapshot, - decideThreadResume, - isThreadReplayRawPayloadSafe, - threadReplayEncodedBytes, - THREAD_RESUME_MAX_REPLAY_EVENTS, -} from "./orchestration-v2/ThreadStream.ts"; import { buildBoundedThreadProjection, - THREAD_HISTORY_PAGE_POLICY, THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, } from "./orchestration-v2/threadHistoryPaging.ts"; -import { - projectDomainEventForWire, - projectThreadProjectionForWire, -} from "./orchestration-v2/WireProjection.ts"; +import { projectThreadProjectionForWire } from "./orchestration-v2/WireProjection.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; @@ -718,12 +707,6 @@ export function hasCompatibleOrchestrationProtocol(url: URL): boolean { ); } -export function shouldUseBoundedThreadSnapshot(input: { - readonly acceptBoundedSnapshot?: boolean; -}): boolean { - return input.acceptBoundedSnapshot === true; -} - function readClientConnectionOrigin( request: HttpServerRequest.HttpServerRequest, ): OrchestrationClientOrigin { @@ -844,198 +827,33 @@ const enrichProjectShells = Effect.fn("ws.orchestrationV2.enrichProjectShells")( ); export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subscribeThread")( - function* (input: { - readonly threadId: ThreadId; - readonly afterSequence?: number; - readonly requestCompletionMarker?: boolean; - readonly acceptBoundedSnapshot?: boolean; - readonly acceptCompactTurnItems?: boolean; - }) { + function* (input: OrchestrationV2SubscribeThreadInput) { const threadManagement = yield* ThreadManagementService.ThreadManagementService; - const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; - yield* Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId, }); - yield* threadManagement.ensureLegacyTranscript(input.threadId).pipe( + return yield* threadManagement.streamThreadEventsForClient(input).pipe( Effect.mapError( (cause) => new OrchestrationV2GetThreadProjectionError({ threadId: input.threadId, - message: `Failed to hydrate migrated thread ${input.threadId}`, + message: cause.message, cause, }), ), - ); - - const eventStreamFrom = (afterSequence: number) => - threadManagement - .streamStoredEventsFrom({ - threadId: input.threadId, - afterSequence, - }) - .pipe( - Stream.map((stored) => ({ - kind: "event" as const, - sequence: stored.sequence, - event: projectDomainEventForWire(stored.event), - })), - coalesceThreadLiveStream, + Effect.map((stream) => + stream.pipe( Stream.mapError( (cause) => new OrchestrationV2GetThreadProjectionError({ threadId: input.threadId, - message: `Failed while streaming orchestration V2 thread ${input.threadId}`, + message: cause.message, cause, }), ), - ); - - const loadReplayThrough = (afterSequence: number, throughSequence: number) => - applicationEvents - .readAgentEvents({ - threadId: input.threadId, - afterSequence, - throughSequence, - limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1, - // A row a newer build wrote (e.g. after a downgrade) must not fail - // every resubscribe. The live tail resumes from `highWater`, so a - // dropped row never hides later events. - skipUnknownEventTypes: true, - }) - .pipe( - Stream.map((stored) => ({ - kind: "event" as const, - sequence: stored.sequence, - event: projectDomainEventForWire(stored.event), - })), - Stream.runCollect, - Effect.map((items) => Array.from(items)), - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed while replaying orchestration V2 thread ${input.threadId}`, - cause, - }), - ), - ); - - const completionMarker = - input.requestCompletionMarker === true - ? Stream.make({ kind: "synchronized" as const }) - : Stream.empty; - - const snapshotThenLive = Effect.fn("ws.orchestrationV2.threadSnapshotThenLive")(function* () { - const useBoundedSnapshot = shouldUseBoundedThreadSnapshot(input); - const snapshot = yield* ( - useBoundedSnapshot - ? threadManagement.getThreadSnapshotWindow(input.threadId, { - rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, - userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns, - }) - : threadManagement.getThreadSnapshot(input.threadId) - ).pipe( - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed to load orchestration V2 thread ${input.threadId}`, - cause, - }), ), - ); - const { snapshotSequence } = snapshot; - const snapshotItem = useBoundedSnapshot - ? buildBoundedThreadStreamSnapshot({ - ...snapshot, - compactTurnItems: input.acceptCompactTurnItems === true, - }) - : { - kind: "snapshot" as const, - snapshotSequence, - projection: projectThreadProjectionForWire(snapshot.projection), - }; - return Stream.concat( - Stream.concat(rpcInitialItems([snapshotItem]), completionMarker), - eventStreamFrom(snapshotSequence), - ); - }); - - // The event sink subscribes to live events before reading the persisted - // tail, so no event published during the replay window is lost; overlapping - // events are deduped by sequence on the client. - if (input.afterSequence !== undefined) { - const highWater = yield* applicationEvents.latestAgentSequence(input.threadId).pipe( - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed to prepare orchestration V2 thread ${input.threadId} replay`, - cause, - }), - ), - ); - if (input.afterSequence > highWater) { - return yield* snapshotThenLive(); - } - const stats = yield* applicationEvents - .getAgentReplayStats({ - threadId: input.threadId, - afterSequence: input.afterSequence, - throughSequence: highWater, - maxEvents: THREAD_RESUME_MAX_REPLAY_EVENTS, - }) - .pipe( - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed to measure orchestration V2 thread ${input.threadId} replay`, - cause, - }), - ), - ); - // Bound stored JSON before decoding, then check projected event - // size separately. Neither byte count is a bound on process memory. - if ( - stats.eventCount > THREAD_RESUME_MAX_REPLAY_EVENTS || - !isThreadReplayRawPayloadSafe(stats.rawPayloadBytes) - ) { - return yield* snapshotThenLive(); - } - if (stats.hasCreateEvent) { - const shell = yield* threadManagement.getThreadShell(input.threadId).pipe( - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed to locate recreated orchestration V2 thread ${input.threadId}`, - cause, - }), - ), - ); - // A retained creation can belong to a thread already deleted. - // Only replace its bounded replay when a snapshot can exist. - if (shell !== null) return yield* snapshotThenLive(); - } - const replay = yield* loadReplayThrough(input.afterSequence, highWater); - const plan = decideThreadResume({ - afterSequence: input.afterSequence, - highWater, - replayEventCount: replay.length, - replayEncodedBytes: threadReplayEncodedBytes(replay), - }); - if (plan.mode === "snapshot") { - return yield* snapshotThenLive(); - } - return Stream.concat( - Stream.concat(rpcInitialItems(replay), completionMarker), - eventStreamFrom(highWater), - ); - } - - return yield* snapshotThenLive(); + ), + ); }, ); diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index eb7a6217c7723..98c1523f5a54c 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts @@ -5,6 +5,7 @@ import { type OrchestrationV2Run, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, + EventId, MessageId, ProjectId, ProviderInstanceId, @@ -106,6 +107,80 @@ const emptyProjection = { } as OrchestrationV2ThreadProjection; describe("applyOrchestrationV2ProjectionEvent", () => { + it("applies compact visits without changing activity or transcript state", () => { + const visitedAt = DateTime.add(now, { hours: 1 }); + const projection = { ...emptyProjection, turnItems: [commandItem("retained")] }; + const next = applyOrchestrationV2ProjectionEvent(projection, { + id: EventId.make("compact-visit"), + type: "thread.visit-recorded", + threadId, + occurredAt: visitedAt, + payload: { lastVisitedAt: visitedAt }, + }); + expect(next?.thread).toEqual({ ...projection.thread, lastVisitedAt: visitedAt }); + expect(next?.updatedAt).toEqual(now); + expect(next?.turnItems).toBe(projection.turnItems); + }); + + it("refreshes one PR snapshot without changing link ownership, watches, or activity", () => { + const link = (number: number): ThreadPullRequestLink => ({ + host: "github.com", + repository: "pingdotgg/t3code", + number, + url: `https://github.com/pingdotgg/t3code/pull/${number}`, + source: "manual", + linkedAt: DateTime.formatIso(now), + snapshot: null, + stack: null, + watch: { + startedAt: DateTime.formatIso(now), + headSha: "head", + failedChecks: ["build"], + passed: false, + passedChecks: [], + remarksThrough: DateTime.formatIso(now), + remarkIds: [], + conflicting: false, + wakes: 0, + }, + }); + const links = [link(1), link(2)]; + const projection = { + ...emptyProjection, + thread: { ...emptyProjection.thread, pullRequests: links }, + turnItems: [commandItem("retained")], + }; + const snapshot = { + state: "merged" as const, + title: "Merged upstream", + headBranch: "feature", + baseBranch: "main", + isDraft: false, + updatedAt: DateTime.formatIso(now), + syncedAt: DateTime.formatIso(now), + }; + const next = applyOrchestrationV2ProjectionEvent(projection, { + id: EventId.make("compact-pr-snapshot"), + type: "thread.pull-request-link-synced", + threadId, + occurredAt: DateTime.add(now, { hours: 1 }), + payload: { + host: links[1]!.host, + repository: links[1]!.repository, + number: 2, + snapshot, + stack: null, + }, + }); + expect(next?.thread).toEqual({ + ...projection.thread, + pullRequests: [links[0], { ...links[1], snapshot }], + }); + expect(next?.thread.pullRequests?.[0]).toBe(links[0]); + expect(next?.turnItems).toBe(projection.turnItems); + expect(next?.updatedAt).toEqual(now); + }); + it("applies a compact PR sync without replacing other links or thread metadata", () => { const link = (number: number): ThreadPullRequestLink => ({ host: "github.com", diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index c93044e93f272..a17be785d46c5 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -1,4 +1,7 @@ -import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests"; +import { + applyThreadPullRequestLinkSync, + applyThreadPullRequestLinkUpdate, +} from "@t3tools/shared/threadPullRequests"; import type { OrchestrationV2DomainEvent, OrchestrationV2Run, @@ -225,6 +228,16 @@ export function applyOrchestrationV2ProjectionEvent( case "thread.model-selection-updated": case "thread.provider-switched": return { ...base, thread: event.payload }; + case "thread.pull-request-link-synced": + return { + ...projection, + thread: applyThreadPullRequestLinkSync(projection.thread, event.payload), + }; + case "thread.visit-recorded": + return { + ...projection, + thread: { ...projection.thread, lastVisitedAt: event.payload.lastVisitedAt }, + }; // Visited tracking is read state, not activity: skip the updatedAt bump. case "thread.visited": case "thread.marked-unread": diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index fc8b8e5008133..dfeab0bc9f704 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -853,6 +853,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make return { threadId, + acceptThreadFieldEvents: true as const, ...(canResume ? { afterSequence: sequence } : {}), ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), ...(acceptBoundedSnapshot diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 94a9eca331d13..65be14b9724c1 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1580,6 +1580,18 @@ export const OrchestrationV2RawProviderEvent = Schema.Struct({ }); export type OrchestrationV2RawProviderEvent = typeof OrchestrationV2RawProviderEvent.Type; +/** A host refresh carries one link snapshot. */ +const OrchestrationV2ThreadPullRequestLinkSynced = Schema.Struct({ + ...ThreadPullRequestKey.fields, + snapshot: ThreadPullRequestSnapshot, + stack: Schema.NullOr(ThreadPullRequestStack), +}); + +/** The thread's read watermark after a visit, the only field a visit changes. */ +const OrchestrationV2ThreadVisitRecorded = Schema.Struct({ + lastVisitedAt: Schema.NullOr(Schema.DateTimeUtc), +}); + const OrchestrationV2EventBase = Schema.Struct({ id: EventId, threadId: ThreadId, @@ -1623,6 +1635,16 @@ export const OrchestrationV2DomainEvent = Schema.Union([ ]), payload: OrchestrationV2AppThread, }), + Schema.Struct({ + ...OrchestrationV2EventBase.fields, + type: Schema.Literal("thread.pull-request-link-synced"), + payload: OrchestrationV2ThreadPullRequestLinkSynced, + }), + Schema.Struct({ + ...OrchestrationV2EventBase.fields, + type: Schema.Literal("thread.visit-recorded"), + payload: OrchestrationV2ThreadVisitRecorded, + }), Schema.Struct({ ...OrchestrationV2EventBase.fields, type: Schema.Literal("thread.pull-request-link-updated"), @@ -2430,6 +2452,19 @@ export const OrchestrationV2DomainEventJson = Schema.Union([ ]), payload: OrchestrationV2AppThreadJson, }), + Schema.Struct({ + ...OrchestrationV2JsonEventBaseFields, + type: Schema.Literal("thread.pull-request-link-synced"), + payload: OrchestrationV2ThreadPullRequestLinkSynced, + }), + Schema.Struct({ + ...OrchestrationV2JsonEventBaseFields, + type: Schema.Literal("thread.visit-recorded"), + payload: OrchestrationV2ThreadVisitRecorded.mapFields((fields) => ({ + ...fields, + lastVisitedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + })), + }), Schema.Struct({ ...OrchestrationV2JsonEventBaseFields, type: Schema.Literal("thread.pull-request-link-updated"), @@ -3190,6 +3225,8 @@ export const OrchestrationV2SubscribeThreadInput = Schema.Struct({ requestCompletionMarker: Schema.optionalKey(Schema.Boolean), acceptBoundedSnapshot: Schema.optionalKey(Schema.Boolean), acceptCompactTurnItems: Schema.optionalKey(Schema.Boolean), + /** Accepts compact read-watermark and pull-request-link events. */ + acceptThreadFieldEvents: Schema.optionalKey(Schema.Boolean), }); export type OrchestrationV2SubscribeThreadInput = typeof OrchestrationV2SubscribeThreadInput.Type; diff --git a/packages/shared/src/threadPullRequests.ts b/packages/shared/src/threadPullRequests.ts index 4e593fa358fc9..4dcc0fd8430fc 100644 --- a/packages/shared/src/threadPullRequests.ts +++ b/packages/shared/src/threadPullRequests.ts @@ -366,6 +366,36 @@ export function threadPullRequestsOf(thread: { ]; } +export function applyThreadPullRequestLinkSync( + thread: OrchestrationV2AppThread, + sync: ThreadPullRequestKey & Pick, +): OrchestrationV2AppThread { + const links = threadPullRequestsOf(thread); + const existing = links.find((link) => threadPullRequestKeysEqual(link, sync)); + if (existing === undefined) return thread; + const pullRequests = links.map((link) => + link === existing ? { ...link, snapshot: sync.snapshot, stack: sync.stack } : link, + ); + const linked = thread.linkedPullRequest; + return { + ...thread, + pullRequests, + ...(linked === undefined + ? {} + : { + linkedPullRequest: + linked && + pullRequests.some( + (link) => + link.source !== "stack-dismissed" && + threadPullRequestKeysEqual(link, legacyThreadPullRequestKey(linked)), + ) + ? linked + : null, + }), + }; +} + export function applyThreadPullRequestLinkUpdate( thread: OrchestrationV2AppThread, updated: ThreadPullRequestLink,