diff --git a/apps/server/package.json b/apps/server/package.json index 709ef2ba8ad5a1894fcd62e06917914cc93f5dfd..e05937f2b3856271675a0cd75e85a8a236ed3582 100644 --- a/apps/server/package.json +++ b/apps/server/package.json @@ -18,6 +18,7 @@ }, "dependencies": { "@effect/platform-node": "4.0.0-beta.10", + "@pierre/diffs": "^1.1.0-beta.16", "effect": "4.0.0-beta.10", "node-pty": "^1.1.0", "open": "^10.1.0", diff --git a/apps/server/src/checkpointing/Diffs.test.ts b/apps/server/src/checkpointing/Diffs.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..e81e509364218e60b015772fcb67f67862563547 --- /dev/null +++ b/apps/server/src/checkpointing/Diffs.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from "vitest"; + +import { parseTurnDiffFilesFromUnifiedDiff } from "./Diffs.ts"; + +describe("parseTurnDiffFilesFromUnifiedDiff", () => { + it("returns empty list for empty diff", () => { + expect(parseTurnDiffFilesFromUnifiedDiff("")).toEqual([]); + }); + + it("parses per-file additions and deletions", () => { + const diff = [ + "diff --git a/a.txt b/a.txt", + "index 1111111..2222222 100644", + "--- a/a.txt", + "+++ b/a.txt", + "@@ -1,2 +1,3 @@", + " one", + "-two", + "+two updated", + "+three", + "diff --git a/src/b.ts b/src/b.ts", + "index 3333333..4444444 100644", + "--- a/src/b.ts", + "+++ b/src/b.ts", + "@@ -3,2 +3,0 @@", + "-old", + "-stale", + "", + ].join("\n"); + + expect(parseTurnDiffFilesFromUnifiedDiff(diff)).toEqual([ + { path: "a.txt", additions: 2, deletions: 1 }, + { path: "src/b.ts", additions: 0, deletions: 2 }, + ]); + }); + + it("parses rename-only diffs with zero line changes", () => { + const diff = [ + "diff --git a/src/old.ts b/src/new.ts", + "similarity index 100%", + "rename from src/old.ts", + "rename to src/new.ts", + "", + ].join("\n"); + + expect(parseTurnDiffFilesFromUnifiedDiff(diff)).toEqual([ + { path: "src/new.ts", additions: 0, deletions: 0 }, + ]); + }); + + it("normalizes CRLF input before parsing", () => { + const diff = [ + "diff --git a/a.txt b/a.txt", + "index 1111111..2222222 100644", + "--- a/a.txt", + "+++ b/a.txt", + "@@ -1 +1,2 @@", + "-one", + "+one updated", + "+two", + "", + ].join("\r\n"); + + expect(parseTurnDiffFilesFromUnifiedDiff(diff)).toEqual([ + { path: "a.txt", additions: 2, deletions: 1 }, + ]); + }); +}); diff --git a/apps/server/src/checkpointing/Diffs.ts b/apps/server/src/checkpointing/Diffs.ts new file mode 100644 index 0000000000000000000000000000000000000000..1db56411051bca100457a0ec52baaf84db820561 --- /dev/null +++ b/apps/server/src/checkpointing/Diffs.ts @@ -0,0 +1,25 @@ +import { parsePatchFiles } from "@pierre/diffs"; + +export interface TurnDiffFileSummary { + readonly path: string; + readonly additions: number; + readonly deletions: number; +} + +export function parseTurnDiffFilesFromUnifiedDiff(diff: string): ReadonlyArray { + const normalized = diff.replace(/\r\n/g, "\n").trim(); + if (normalized.length === 0) { + return []; + } + + const parsedPatches = parsePatchFiles(normalized); + const files = parsedPatches.flatMap((patch) => + patch.files.map((file) => ({ + path: file.name, + additions: file.hunks.reduce((total, hunk) => total + hunk.additionLines, 0), + deletions: file.hunks.reduce((total, hunk) => total + hunk.deletionLines, 0), + })), + ); + + return files.toSorted((left, right) => left.path.localeCompare(right.path)); +} diff --git a/apps/server/src/orchestration/Layer.test.ts b/apps/server/src/orchestration/Layer.test.ts index 27f29b8e76d4ef4400b829d92d92ed5cd9654f69..e82811c12056686e7fbfdb341aa6bb9266f24033 100644 --- a/apps/server/src/orchestration/Layer.test.ts +++ b/apps/server/src/orchestration/Layer.test.ts @@ -1,5 +1,5 @@ import type { OrchestrationEvent } from "@t3tools/contracts"; -import { Effect, Layer, ManagedRuntime } from "effect"; +import { Effect, Layer, ManagedRuntime, Queue, Stream } from "effect"; import { describe, expect, it } from "vitest"; import { @@ -65,33 +65,6 @@ describe("OrchestrationEngine", () => { await system.dispose(); }); - it("fans out read-model updates to subscribers", async () => { - const system = await createOrchestrationSystem(); - const engine = system.engine; - const updates: number[] = []; - const unsubscribe = await system.run( - engine.subscribeToReadModel((snapshot) => { - updates.push(snapshot.sequence); - }), - ); - await system.run( - engine.dispatch({ - type: "thread.create", - commandId: "cmd-thread", - threadId: "thread-2", - projectId: "project-2", - title: "fanout", - model: "gpt-5-codex", - branch: null, - worktreePath: null, - createdAt: new Date().toISOString(), - }), - ); - unsubscribe(); - expect(updates.length).toBeGreaterThan(0); - await system.dispose(); - }); - it("replays append-only events from sequence", async () => { const system = await createOrchestrationSystem(); const engine = system.engine; @@ -123,6 +96,46 @@ describe("OrchestrationEngine", () => { await system.dispose(); }); + it("streams persisted domain events in order", async () => { + const system = await createOrchestrationSystem(); + const engine = system.engine; + const eventTypes: string[] = []; + await system.run( + Effect.gen(function* () { + const eventQueue = yield* Queue.unbounded(); + yield* Effect.forkScoped( + Stream.take(engine.streamDomainEvents, 2).pipe( + Stream.runForEach((event) => Queue.offer(eventQueue, event).pipe(Effect.asVoid)), + ), + ); + yield* Effect.sleep("10 millis"); + const createdAt = new Date().toISOString(); + yield* engine.dispatch({ + type: "thread.create", + commandId: "cmd-stream-domain-a", + threadId: "thread-stream-domain", + projectId: "project-stream-domain", + title: "domain-stream", + model: "gpt-5-codex", + branch: null, + worktreePath: null, + createdAt, + }); + yield* engine.dispatch({ + type: "thread.meta.update", + commandId: "cmd-stream-domain-b", + threadId: "thread-stream-domain", + title: "domain-stream-updated", + createdAt, + }); + eventTypes.push((yield* Queue.take(eventQueue)).type); + eventTypes.push((yield* Queue.take(eventQueue)).type); + }).pipe(Effect.scoped), + ); + expect(eventTypes).toEqual(["thread.created", "thread.meta-updated"]); + await system.dispose(); + }); + it("stores completed turn summaries even when no files changed", async () => { const system = await createOrchestrationSystem(); const engine = system.engine; diff --git a/apps/server/src/orchestration/Layer.ts b/apps/server/src/orchestration/Layer.ts index 6847c693d3302e4bd7a94b38d0e4101be8cbd42f..c03029b93ae7ae6be31446c1c7ce8eb323e6fd63 100644 --- a/apps/server/src/orchestration/Layer.ts +++ b/apps/server/src/orchestration/Layer.ts @@ -4,12 +4,11 @@ import type { OrchestrationReadModel, } from "@t3tools/contracts"; import { OrchestrationCommandSchema } from "@t3tools/contracts"; -import { Deferred, Effect, Layer, PubSub, Queue, Schema } from "effect"; +import { Deferred, Effect, Layer, PubSub, Queue, Schema, Stream } from "effect"; import { createLogger } from "../logger.ts"; import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents.ts"; import { - toListenerCallbackError, toOrchestrationCommandDecodeError, toOrchestrationJsonParseError, type OrchestrationDispatchError, @@ -182,42 +181,14 @@ const makeOrchestrationEngine = Effect.gen(function* () { let readModel = createEmptyReadModel(new Date().toISOString()); const commandQueue = yield* Queue.unbounded(); - const readModelPubSub = yield* PubSub.unbounded(); const eventPubSub = yield* PubSub.unbounded(); - const readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>(); - const domainEventListeners = new Set<(event: OrchestrationEvent) => void>(); - - const notifyReadModelListeners = (snapshot: OrchestrationReadModel) => - Effect.forEach(readModelListeners, (listener) => - Effect.try({ - try: () => listener(snapshot), - catch: toListenerCallbackError("read-model"), - }), - ).pipe(Effect.asVoid); - - const notifyDomainEventListeners = (event: OrchestrationEvent) => - Effect.forEach(domainEventListeners, (listener) => - Effect.try({ - try: () => listener(event), - catch: toListenerCallbackError("domain-event"), - }), - ).pipe(Effect.asVoid); - const processEnvelope = (envelope: CommandEnvelope): Effect.Effect => Effect.gen(function* () { const eventBase = mapCommandToEvent(envelope.command); const savedEvent = yield* eventStore.append(eventBase); readModel = yield* reduceEvent(readModel, savedEvent); - - const snapshot = readModel; - yield* Effect.all([ - PubSub.publish(eventPubSub, savedEvent), - PubSub.publish(readModelPubSub, snapshot), - ]); - - yield* notifyDomainEventListeners(savedEvent); - yield* notifyReadModelListeners(snapshot); + yield* PubSub.publish(eventPubSub, savedEvent); yield* Deferred.succeed(envelope.result, { sequence: savedEvent.sequence }); }).pipe(Effect.catch((error) => Deferred.fail(envelope.result, error).pipe(Effect.asVoid))); @@ -269,29 +240,15 @@ const makeOrchestrationEngine = Effect.gen(function* () { return yield* dispatch(decoded); }); - const subscribeToReadModel: OrchestrationEngineShape["subscribeToReadModel"] = (callback) => - Effect.sync(() => { - readModelListeners.add(callback); - return () => { - readModelListeners.delete(callback); - }; - }); - - const subscribeToDomainEvents: OrchestrationEngineShape["subscribeToDomainEvents"] = (callback) => - Effect.sync(() => { - domainEventListeners.add(callback); - return () => { - domainEventListeners.delete(callback); - }; - }); + const streamDomainEvents: OrchestrationEngineShape["streamDomainEvents"] = + Stream.fromPubSub(eventPubSub); return { getSnapshot, replayEvents, dispatchUnknown, dispatch, - subscribeToReadModel, - subscribeToDomainEvents, + streamDomainEvents, } satisfies OrchestrationEngineShape; }); diff --git a/apps/server/src/orchestration/Service.ts b/apps/server/src/orchestration/Service.ts index 692c6ca4998261bcd7c2b7a8a1400eabd3b82718..bffd33e50823b42f291123199e772bbae36d7906 100644 --- a/apps/server/src/orchestration/Service.ts +++ b/apps/server/src/orchestration/Service.ts @@ -16,7 +16,7 @@ import type { OrchestrationReadModel, } from "@t3tools/contracts"; import { ServiceMap } from "effect"; -import type { Effect } from "effect"; +import type { Effect, Stream } from "effect"; import type { OrchestrationDispatchError, OrchestrationEngineError } from "./Errors.ts"; import type { OrchestrationEventRepositoryError } from "../persistence/Errors.ts"; @@ -62,24 +62,9 @@ export interface OrchestrationEngineShape { ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>; /** - * Subscribe to read model updates. - * - * @param callback - Handler invoked on every read model update. - * @returns Effect containing an unsubscribe function. + * Stream persisted domain events in dispatch order. */ - readonly subscribeToReadModel: ( - callback: (snapshot: OrchestrationReadModel) => void, - ) => Effect.Effect<() => void, never, never>; - - /** - * Subscribe to domain event fan-out. - * - * @param callback - Handler invoked for every persisted domain event. - * @returns Effect containing an unsubscribe function. - */ - readonly subscribeToDomainEvents: ( - callback: (event: OrchestrationEvent) => void, - ) => Effect.Effect<() => void, never, never>; + readonly streamDomainEvents: Stream.Stream; } /** diff --git a/apps/server/src/orchestration/reducer.test.ts b/apps/server/src/orchestration/reducer.test.ts index bf3cc633d72ea0b3893da09b04ac0c053111242b..97cb5a9d14afef839a0b941ee06308c479cbf70e 100644 --- a/apps/server/src/orchestration/reducer.test.ts +++ b/apps/server/src/orchestration/reducer.test.ts @@ -95,4 +95,164 @@ describe("orchestration reducer", () => { expect(next.updatedAt).toBe("2026-01-01T00:00:00.000Z"); expect(next.threads).toEqual([]); }); + + it("tracks latest turn timing from session lifecycle events", async () => { + const createdAt = "2026-02-23T08:00:00.000Z"; + const startedAt = "2026-02-23T08:00:05.000Z"; + const completedAt = "2026-02-23T08:00:11.250Z"; + const model = createEmptyReadModel(createdAt); + + const createdEvent: OrchestrationEvent = { + sequence: 1, + eventId: "thread-created", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: createdAt, + commandId: "cmd-create", + payload: { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5.3-codex", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }; + + const runningSessionEvent: OrchestrationEvent = { + sequence: 2, + eventId: "session-running", + type: "thread.session-set", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: startedAt, + commandId: "cmd-running", + payload: { + threadId: "thread-1", + session: { + sessionId: "session-1", + status: "running", + provider: "codex", + threadId: "thread-1", + activeTurnId: "turn-1", + createdAt, + updatedAt: startedAt, + lastError: null, + }, + }, + }; + + const readySessionEvent: OrchestrationEvent = { + sequence: 3, + eventId: "session-ready", + type: "thread.session-set", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: completedAt, + commandId: "cmd-ready", + payload: { + threadId: "thread-1", + session: { + sessionId: "session-1", + status: "ready", + provider: "codex", + threadId: "thread-1", + activeTurnId: null, + createdAt, + updatedAt: completedAt, + lastError: null, + }, + }, + }; + + const afterCreate = await Effect.runPromise(reduceEvent(model, createdEvent)); + const afterRunning = await Effect.runPromise(reduceEvent(afterCreate, runningSessionEvent)); + const afterReady = await Effect.runPromise(reduceEvent(afterRunning, readySessionEvent)); + const thread = afterReady.threads[0]; + + expect(thread?.latestTurnId).toBe("turn-1"); + expect(thread?.latestTurnStartedAt).toBe(startedAt); + expect(thread?.latestTurnCompletedAt).toBe(completedAt); + expect(thread?.latestTurnDurationMs).toBe(6250); + }); + + it("marks assistant messages completed using non-streaming updates", async () => { + const createdAt = "2026-02-23T09:00:00.000Z"; + const deltaAt = "2026-02-23T09:00:01.000Z"; + const completeAt = "2026-02-23T09:00:03.500Z"; + const model = createEmptyReadModel(createdAt); + + const createdEvent: OrchestrationEvent = { + sequence: 1, + eventId: "thread-created", + type: "thread.created", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: createdAt, + commandId: "cmd-create", + payload: { + id: "thread-1", + projectId: "project-1", + title: "demo", + model: "gpt-5.3-codex", + branch: null, + worktreePath: null, + createdAt, + updatedAt: createdAt, + }, + }; + + const deltaEvent: OrchestrationEvent = { + sequence: 2, + eventId: "assistant-delta", + type: "message.sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: deltaAt, + commandId: "cmd-delta", + payload: { + id: "assistant:msg-1", + role: "assistant", + text: "hello", + threadId: "thread-1", + createdAt: deltaAt, + streaming: true, + }, + }; + + const completionEvent: OrchestrationEvent = { + sequence: 3, + eventId: "assistant-complete", + type: "message.sent", + aggregateType: "thread", + aggregateId: "thread-1", + occurredAt: completeAt, + commandId: "cmd-complete", + payload: { + id: "assistant:msg-1", + role: "assistant", + text: "", + threadId: "thread-1", + createdAt: completeAt, + streaming: false, + }, + }; + + const afterCreate = await Effect.runPromise(reduceEvent(model, createdEvent)); + const afterDelta = await Effect.runPromise(reduceEvent(afterCreate, deltaEvent)); + const afterComplete = await Effect.runPromise(reduceEvent(afterDelta, completionEvent)); + const message = afterComplete.threads[0]?.messages[0]; + + expect(message).toEqual({ + id: "assistant:msg-1", + role: "assistant", + text: "hello", + createdAt: deltaAt, + completedAt: completeAt, + streaming: false, + }); + }); }); diff --git a/apps/server/src/orchestration/reducer.ts b/apps/server/src/orchestration/reducer.ts index 40320779c1f6af4979ff69071f6b97342d156c0d..bf643572eebf597946d7bbe06bf41ca44bd27b4c 100644 --- a/apps/server/src/orchestration/reducer.ts +++ b/apps/server/src/orchestration/reducer.ts @@ -204,6 +204,7 @@ export function reduceEvent( role: payload.role, text: payload.text, createdAt: payload.createdAt, + ...(payload.streaming ? {} : { completedAt: event.occurredAt }), streaming: payload.streaming, }, event.type, @@ -222,6 +223,9 @@ export function reduceEvent( : entry.text, streaming: message.streaming, createdAt: entry.createdAt, + ...(message.streaming + ? { completedAt: undefined } + : { completedAt: message.completedAt ?? event.occurredAt }), } : entry, ) @@ -260,12 +264,51 @@ export function reduceEvent( "session", ); + const sessionUpdatedAt = session.updatedAt; + const turnTimingPatch: ThreadPatch = + session.status === "running" && session.activeTurnId + ? (() => { + const isSameTurn = thread.latestTurnId === session.activeTurnId; + return { + latestTurnId: session.activeTurnId, + latestTurnStartedAt: + isSameTurn && thread.latestTurnStartedAt + ? thread.latestTurnStartedAt + : sessionUpdatedAt, + ...(isSameTurn + ? {} + : { + latestTurnCompletedAt: null, + latestTurnDurationMs: null, + }), + }; + })() + : session.activeTurnId === null && thread.latestTurnId && thread.latestTurnStartedAt + ? (() => { + const latestTurnCompletedAt = thread.latestTurnCompletedAt ?? sessionUpdatedAt; + const startedAtMs = Date.parse(thread.latestTurnStartedAt); + const completedAtMs = Date.parse(latestTurnCompletedAt); + const latestTurnDurationMs = + !Number.isNaN(startedAtMs) && + !Number.isNaN(completedAtMs) && + completedAtMs >= startedAtMs + ? completedAtMs - startedAtMs + : thread.latestTurnDurationMs; + + return { + latestTurnCompletedAt, + latestTurnDurationMs, + }; + })() + : {}; + return { ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { session, updatedAt: event.occurredAt, error: session.lastError, + ...turnTimingPatch, }), }; }); @@ -303,12 +346,23 @@ export function reduceEvent( ...thread.turnDiffSummaries.filter((summary) => summary.turnId !== payload.turnId), nextSummary, ].toSorted((left, right) => left.completedAt.localeCompare(right.completedAt)); + const startedAtMs = thread.latestTurnStartedAt + ? Date.parse(thread.latestTurnStartedAt) + : Number.NaN; + const completedAtMs = Date.parse(nextSummary.completedAt); + const latestTurnDurationMs = + !Number.isNaN(startedAtMs) && + !Number.isNaN(completedAtMs) && + completedAtMs >= startedAtMs + ? completedAtMs - startedAtMs + : thread.latestTurnDurationMs; return { ...nextBase, threads: updateThread(nextBase.threads, payload.threadId, { latestTurnId: payload.turnId, latestTurnCompletedAt: nextSummary.completedAt, + latestTurnDurationMs, turnDiffSummaries, updatedAt: event.occurredAt, }), diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 632fd0a922126f1a1bbdf7980c5f78a608bc80c0..c8c6bb334e5029e05382a89dd1986abb4aacfcbe 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -170,6 +170,45 @@ const lifecycleManager = new FakeCodexManager(); const lifecycleLayer = it.layer(makeCodexAdapterLive({ manager: lifecycleManager })); lifecycleLayer("CodexAdapterLive lifecycle", (it) => { + it.effect("maps completed agent message items to canonical message.completed events", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + const event: ProviderEvent = { + id: "evt-msg-complete", + kind: "notification", + provider: "codex", + sessionId: "sess-1", + createdAt: new Date().toISOString(), + method: "item/completed", + threadId: "thread-1", + turnId: "turn-1", + itemId: "msg_1", + payload: { + item: { + type: "agentMessage", + id: "msg_1", + }, + }, + }; + + lifecycleManager.emit("event", event); + const firstEvent = yield* Fiber.join(firstEventFiber); + + assert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + assert.equal(firstEvent.value.type, "message.completed"); + if (firstEvent.value.type !== "message.completed") { + return; + } + assert.equal(firstEvent.value.itemId, "msg_1"); + assert.equal(firstEvent.value.turnId, "turn-1"); + }), + ); + it.effect("emits canonical events on stream and supports consumer interruption", () => Effect.gen(function* () { const adapter = yield* CodexAdapter; diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index d8055a54460a0704c339f9ebc58ed0c19f0ad086..c3e88d112becc5ff041856efb4eb74559fd82748 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -161,6 +161,39 @@ function itemDetail( return undefined; } +function mapMessageCompletedEvent(event: ProviderEvent): ProviderRuntimeEvent | undefined { + if (event.method !== "item/completed") { + return undefined; + } + + const payload = asObject(event.payload); + const item = asObject(payload?.item); + if (!item) { + return undefined; + } + + const normalizedType = normalizeItemType(item.type ?? item.kind); + if (!normalizedType.includes("agent message")) { + return undefined; + } + + const itemId = event.itemId ?? asString(item.id); + if (!itemId) { + return undefined; + } + + return { + type: "message.completed", + eventId: event.id, + provider: event.provider, + sessionId: event.sessionId, + createdAt: event.createdAt, + itemId, + ...(event.threadId ? { threadId: event.threadId } : {}), + ...(event.turnId ? { turnId: event.turnId } : {}), + }; +} + function mapToolEvent(event: ProviderEvent): ProviderRuntimeEvent | undefined { if (event.method !== "item/started" && event.method !== "item/completed") { return undefined; @@ -267,6 +300,11 @@ function mapToRuntimeEvents(event: ProviderEvent): ReadonlyArray, orchestrationEngine: OrchestrationEngineShape, ) { - unsubscribeDomainEvents = await runtime.runPromise( - orchestrationEngine.subscribeToDomainEvents((event) => { - orchestrationDomainEventLogger?.write({ - observedAt: new Date().toISOString(), - event, - }); - const push: WsPush = { - type: "push", - channel: ORCHESTRATION_WS_CHANNELS.domainEvent, - data: event, - }; - const message = JSON.stringify(push); - let recipients = 0; - for (const client of clients) { - if (client.readyState === client.OPEN) { - client.send(message); - recipients += 1; - } - } - logOutgoingPush(push, recipients); - }), + const subscriptionScope = await runtime.runPromise(Scope.make("sequential")); + await runtime.runPromise( + Effect.gen(function* () { + yield* Effect.forkScoped( + Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => + Effect.sync(() => { + orchestrationDomainEventLogger?.write({ + observedAt: new Date().toISOString(), + event, + }); + const push: WsPush = { + type: "push", + channel: ORCHESTRATION_WS_CHANNELS.domainEvent, + data: event, + }; + const message = JSON.stringify(push); + let recipients = 0; + for (const client of clients) { + if (client.readyState === client.OPEN) { + client.send(message); + recipients += 1; + } + } + logOutgoingPush(push, recipients); + }), + ), + ); + }).pipe(Scope.provide(subscriptionScope)), ); + + unsubscribeDomainEvents = () => { + runtime.runFork(Scope.close(subscriptionScope, Exit.void)); + }; } async function attachProviderSubscriptions( @@ -350,7 +359,9 @@ export function createServer(options: ServerOptions) { const processEvent = (event: ProviderRuntimeEvent) => Effect.gen(function* () { const snapshot = yield* orchestrationEngine.getSnapshot(); - const thread = snapshot.threads.find((entry) => entry.session?.sessionId === event.sessionId); + const thread = snapshot.threads.find( + (entry) => entry.session?.sessionId === event.sessionId, + ); if (!thread) return; const now = event.createdAt; @@ -393,9 +404,29 @@ export function createServer(options: ServerOptions) { }); } + if (event.type === "message.completed") { + const assistantMessageId = `assistant:${event.itemId}`; + if (event.turnId) { + rememberAssistantMessageId(event.sessionId, event.turnId, assistantMessageId); + } + yield* orchestrationEngine.dispatch({ + type: "message.send", + commandId: crypto.randomUUID(), + threadId: thread.id, + messageId: assistantMessageId, + role: "assistant", + text: "", + streaming: false, + createdAt: now, + }); + } + if (event.type === "turn.completed") { if (event.turnId) { - const assistantMessageIds = getAssistantMessageIdsForTurn(event.sessionId, event.turnId); + const assistantMessageIds = getAssistantMessageIdsForTurn( + event.sessionId, + event.turnId, + ); yield* Effect.forEach(assistantMessageIds, (assistantMessageId) => orchestrationEngine.dispatch({ type: "message.send", @@ -417,6 +448,30 @@ export function createServer(options: ServerOptions) { } if (event.type === "checkpoint.captured" && event.turnId) { + const files: ReadonlyArray = yield* providerService + .getCheckpointDiff({ + sessionId: event.sessionId, + fromTurnCount: Math.max(0, event.turnCount - 1), + toTurnCount: event.turnCount, + }) + .pipe( + Effect.map((result) => parseTurnDiffFilesFromUnifiedDiff(result.diff)), + Effect.catch((error) => + Effect.sync(() => { + logger.warn("failed to derive checkpoint file summary", { + sessionId: event.sessionId, + turnId: event.turnId, + turnCount: event.turnCount, + error: + error instanceof Error + ? error.message + : typeof error === "string" + ? error + : "unknown error", + }); + }).pipe(Effect.as([] as ReadonlyArray)), + ), + ); const assistantMessageId = getLatestAssistantMessageIdForTurn(event.sessionId, event.turnId) ?? `assistant:${event.turnId}`; @@ -427,7 +482,7 @@ export function createServer(options: ServerOptions) { turnId: event.turnId, completedAt: now, ...(event.status ? { status: event.status } : {}), - files: [], + files, assistantMessageId, checkpointTurnCount: event.turnCount, createdAt: now, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 8c68aa3a82108f808f1e0ca634194b2ace2b174f..c308809c55710c4e9ba6e6328228e7f481259c8e 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -26,12 +26,11 @@ import { useRef, useState, } from "react"; -import { useMutation, useQueries, useQuery, useQueryClient } from "@tanstack/react-query"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { useDebouncedValue } from "@tanstack/react-pacer"; import { useNavigate, useSearch } from "@tanstack/react-router"; import { type VirtualItem, useVirtualizer } from "@tanstack/react-virtual"; import { gitBranchesQueryOptions, gitCreateWorktreeMutationOptions } from "~/lib/gitReactQuery"; -import { checkpointDiffQueryOptions } from "~/lib/providerReactQuery"; import { projectSearchEntriesQueryOptions } from "~/lib/projectReactQuery"; import { serverConfigQueryOptions, serverQueryKeys } from "~/lib/serverReactQuery"; @@ -55,7 +54,6 @@ import { import { derivePendingApprovals, derivePhase, - deriveTurnDiffFilesFromUnifiedDiff, deriveTimelineEntries, type PendingApproval, deriveWorkLogEntries, @@ -70,7 +68,6 @@ import { DEFAULT_THREAD_TERMINAL_ID, MAX_THREAD_TERMINAL_COUNT, type ChatImageAttachment, - type TurnDiffFileChange, type TurnDiffSummary, } from "../types"; import { basenameOfPath, getVscodeIconUrlForEntry } from "../vscode-icons"; @@ -186,33 +183,6 @@ interface ExpandedImagePreview { name: string; } -type TurnCheckpointFilesState = - | { - status: "pending" | "loading"; - files: TurnDiffFileChange[]; - errorMessage?: undefined; - } - | { - status: "error"; - files: TurnDiffFileChange[]; - errorMessage: string; - } - | { - status: "ready"; - files: TurnDiffFileChange[]; - errorMessage?: undefined; - }; - -function checkpointErrorMessage(error: unknown): string { - if (error instanceof Error && error.message.trim().length > 0) { - return error.message; - } - if (typeof error === "string" && error.trim().length > 0) { - return error; - } - return "Checkpoint diff is unavailable."; -} - type ComposerCommandItem = | { id: string; @@ -432,17 +402,21 @@ export default function ChatView({ threadId }: ChatViewProps) { ); const phase = derivePhase(activeThread?.session ?? null); const isWorking = phase === "running" || isSending || isConnecting || isRevertingCheckpoint; - const activeTurnId = activeThread?.session?.activeTurnId ?? undefined; const nowIso = new Date(nowTick).toISOString(); const threadActivities = activeThread?.activities ?? []; const workLogEntries = useMemo( () => deriveWorkLogEntries(threadActivities, undefined), [threadActivities], ); - const latestTurnWorkEntries = useMemo( - () => deriveWorkLogEntries(threadActivities, activeTurnId ?? activeThread?.latestTurnId), - [activeThread?.latestTurnId, activeTurnId, threadActivities], - ); + const latestTurnHasToolActivity = useMemo(() => { + const latestTurnId = activeThread?.latestTurnId; + if (!latestTurnId) { + return false; + } + return threadActivities.some( + (activity) => activity.turnId === latestTurnId && activity.tone === "tool", + ); + }, [activeThread?.latestTurnId, threadActivities]); const pendingApprovals = useMemo( () => derivePendingApprovals(threadActivities), [threadActivities], @@ -453,14 +427,6 @@ export default function ChatView({ threadId }: ChatViewProps) { ); const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = useTurnDiffSummaries(activeThread); - const assistantCompletionByItemId = useMemo(() => { - const completionByMessageId = new Map(); - for (const summary of turnDiffSummaries) { - if (!summary.assistantMessageId) continue; - completionByMessageId.set(summary.assistantMessageId, summary.completedAt); - } - return completionByMessageId; - }, [turnDiffSummaries]); const turnDiffSummaryByAssistantMessageId = useMemo(() => { const byMessageId = new Map(); for (const summary of turnDiffSummaries) { @@ -469,69 +435,6 @@ export default function ChatView({ threadId }: ChatViewProps) { } return byMessageId; }, [turnDiffSummaries]); - const checkpointDiffInputsByTurn = useMemo( - () => - turnDiffSummaries - .filter((summary) => summary.assistantMessageId) - .map((summary) => { - const checkpointTurnCount = - summary.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[summary.turnId]; - return { - turnId: summary.turnId, - checkpointTurnCount: - typeof checkpointTurnCount === "number" ? checkpointTurnCount : null, - }; - }), - [inferredCheckpointTurnCountByTurnId, turnDiffSummaries], - ); - const checkpointDiffQueriesByTurn = useQueries({ - queries: checkpointDiffInputsByTurn.map((input) => - checkpointDiffQueryOptions(api, { - sessionId: activeSessionId, - threadRuntimeId: activeThreadRuntimeId, - fromTurnCount: - typeof input.checkpointTurnCount === "number" - ? Math.max(0, input.checkpointTurnCount - 1) - : null, - toTurnCount: input.checkpointTurnCount, - cacheScope: `turn:${input.turnId}`, - }), - ), - }); - const turnCheckpointFilesByTurnId = useMemo(() => { - const byTurnId = new Map(); - for (let index = 0; index < checkpointDiffInputsByTurn.length; index += 1) { - const input = checkpointDiffInputsByTurn[index]; - const query = checkpointDiffQueriesByTurn[index]; - if (!input || !query) continue; - - if (!activeSessionId || typeof input.checkpointTurnCount !== "number") { - byTurnId.set(input.turnId, { status: "pending", files: [] }); - continue; - } - - const diff = query.data?.diff; - if (typeof diff === "string") { - byTurnId.set(input.turnId, { - status: "ready", - files: deriveTurnDiffFilesFromUnifiedDiff(diff), - }); - continue; - } - - if (query.isError) { - byTurnId.set(input.turnId, { - status: "error", - files: [], - errorMessage: checkpointErrorMessage(query.error), - }); - continue; - } - - byTurnId.set(input.turnId, { status: "loading", files: [] }); - } - return byTurnId; - }, [activeSessionId, checkpointDiffInputsByTurn, checkpointDiffQueriesByTurn]); const revertTurnCountByUserMessageId = useMemo(() => { const byUserMessageId = new Map(); for (let index = 0; index < timelineEntries.length; index += 1) { @@ -568,9 +471,7 @@ export default function ChatView({ threadId }: ChatViewProps) { const completionSummary = useMemo(() => { if (!activeThread?.latestTurnStartedAt) return null; if (!activeThread.latestTurnCompletedAt) return null; - if (!latestTurnWorkEntries.some((entry) => entry.tone === "tool")) { - return null; - } + if (!latestTurnHasToolActivity) return null; if ( typeof activeThread.latestTurnDurationMs === "number" && @@ -589,7 +490,7 @@ export default function ChatView({ threadId }: ChatViewProps) { activeThread?.latestTurnStartedAt, activeThread?.latestTurnCompletedAt, activeThread?.latestTurnDurationMs, - latestTurnWorkEntries, + latestTurnHasToolActivity, ]); const completionDividerBeforeEntryId = useMemo(() => { if (!activeThread?.latestTurnStartedAt) return null; @@ -2039,9 +1940,7 @@ export default function ChatView({ threadId }: ChatViewProps) { timelineEntries={timelineEntries} completionDividerBeforeEntryId={completionDividerBeforeEntryId} completionSummary={completionSummary} - assistantCompletionByItemId={assistantCompletionByItemId} turnDiffSummaryByAssistantMessageId={turnDiffSummaryByAssistantMessageId} - turnCheckpointFilesByTurnId={turnCheckpointFilesByTurnId} nowIso={nowIso} expandedWorkGroups={expandedWorkGroups} onToggleWorkGroup={onToggleWorkGroup} @@ -2505,9 +2404,7 @@ interface MessagesTimelineProps { timelineEntries: ReturnType; completionDividerBeforeEntryId: string | null; completionSummary: string | null; - assistantCompletionByItemId: Map; turnDiffSummaryByAssistantMessageId: Map; - turnCheckpointFilesByTurnId: Map; nowIso: string; expandedWorkGroups: Record; onToggleWorkGroup: (groupId: string) => void; @@ -2542,9 +2439,7 @@ const MessagesTimeline = memo(function MessagesTimeline({ timelineEntries, completionDividerBeforeEntryId, completionSummary, - assistantCompletionByItemId, turnDiffSummaryByAssistantMessageId, - turnCheckpointFilesByTurnId, nowIso, expandedWorkGroups, onToggleWorkGroup, @@ -2793,10 +2688,8 @@ const MessagesTimeline = memo(function MessagesTimeline({ row.message.id, ); if (!turnSummary) return null; - const checkpointFilesState = turnCheckpointFilesByTurnId.get( - turnSummary.turnId, - ); - const checkpointFiles = checkpointFilesState?.files ?? []; + const checkpointFiles = turnSummary.files; + if (checkpointFiles.length === 0) return null; const summaryStat = checkpointFiles.reduce( (acc, file) => { if ( @@ -2812,10 +2705,7 @@ const MessagesTimeline = memo(function MessagesTimeline({ }, { additions: 0, deletions: 0 }, ); - const changedFileCountLabel = - checkpointFilesState?.status === "ready" - ? String(checkpointFiles.length) - : "..."; + const changedFileCountLabel = String(checkpointFiles.length); return (
@@ -2843,66 +2733,40 @@ const MessagesTimeline = memo(function MessagesTimeline({ View diff
- {!checkpointFilesState || - checkpointFilesState.status === "pending" ? ( -

- Waiting for checkpoint metadata... -

- ) : checkpointFilesState?.status === "loading" ? ( -

- Loading checkpoint diff... -

- ) : checkpointFilesState?.status === "error" ? ( -

- {checkpointFilesState.errorMessage} -

- ) : checkpointFiles.length > 0 ? ( -
- {checkpointFiles.map((file) => ( - - ))} -
- ) : ( -

- No changed files. -

- )} +
+ {checkpointFiles.map((file) => ( + + ))} +
); })()} @@ -2911,10 +2775,7 @@ const MessagesTimeline = memo(function MessagesTimeline({ row.message.createdAt, row.message.streaming ? formatElapsed(row.message.createdAt, nowIso) - : formatElapsed( - row.message.createdAt, - assistantCompletionByItemId.get(row.message.id), - ), + : formatElapsed(row.message.createdAt, row.message.completedAt), )}

diff --git a/apps/web/src/components/DiffPanel.tsx b/apps/web/src/components/DiffPanel.tsx index 4b867209108223ed0e1ac0dcf4d9d504facdab99..3a25f678132fe6b74841025dd1b7c8a59952bafb 100644 --- a/apps/web/src/components/DiffPanel.tsx +++ b/apps/web/src/components/DiffPanel.tsx @@ -104,14 +104,28 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { const activeSessionId = activeThread?.session?.sessionId ?? null; const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = useTurnDiffSummaries(activeThread); + const orderedTurnDiffSummaries = useMemo( + () => + [...turnDiffSummaries].toSorted((left, right) => { + const leftTurnCount = + left.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[left.turnId] ?? 0; + const rightTurnCount = + right.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[right.turnId] ?? 0; + if (leftTurnCount !== rightTurnCount) { + return rightTurnCount - leftTurnCount; + } + return right.completedAt.localeCompare(left.completedAt); + }), + [inferredCheckpointTurnCountByTurnId, turnDiffSummaries], + ); const selectedTurnId = diffSearch.diffTurnId ?? null; const selectedFilePath = selectedTurnId !== null ? (diffSearch.diffFilePath ?? null) : null; const selectedTurn = selectedTurnId === null ? undefined - : (turnDiffSummaries.find((summary) => summary.turnId === selectedTurnId) ?? - turnDiffSummaries[0]); + : (orderedTurnDiffSummaries.find((summary) => summary.turnId === selectedTurnId) ?? + orderedTurnDiffSummaries[0]); const selectedCheckpointTurnCount = selectedTurn && (selectedTurn.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[selectedTurn.turnId]); @@ -126,7 +140,7 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { [selectedCheckpointTurnCount], ); const conversationCheckpointTurnCount = useMemo(() => { - const turnCounts = turnDiffSummaries + const turnCounts = orderedTurnDiffSummaries .map( (summary) => summary.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[summary.turnId], @@ -137,7 +151,7 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { } const latest = Math.max(...turnCounts); return latest > 0 ? latest : undefined; - }, [inferredCheckpointTurnCountByTurnId, turnDiffSummaries]); + }, [inferredCheckpointTurnCountByTurnId, orderedTurnDiffSummaries]); const conversationCheckpointRange = useMemo( () => !selectedTurn && typeof conversationCheckpointTurnCount === "number" @@ -152,11 +166,11 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { ? selectedCheckpointRange : conversationCheckpointRange; const conversationCacheScope = useMemo(() => { - if (selectedTurn || turnDiffSummaries.length === 0) { + if (selectedTurn || orderedTurnDiffSummaries.length === 0) { return null; } - return `conversation:${turnDiffSummaries.map((summary) => summary.turnId).join(",")}`; - }, [selectedTurn, turnDiffSummaries]); + return `conversation:${orderedTurnDiffSummaries.map((summary) => summary.turnId).join(",")}`; + }, [orderedTurnDiffSummaries, selectedTurn]); const activeCheckpointDiffQuery = useQuery( checkpointDiffQueryOptions(api, { sessionId: activeSessionId, @@ -181,6 +195,8 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { : null; const selectedPatch = selectedTurn ? selectedTurnCheckpointDiff : conversationCheckpointDiff; + const hasResolvedPatch = typeof selectedPatch === "string"; + const hasNoNetChanges = hasResolvedPatch && selectedPatch.trim().length === 0; const renderablePatch = useMemo(() => getRenderablePatch(selectedPatch), [selectedPatch]); const renderableFiles = useMemo(() => { if (!renderablePatch || renderablePatch.kind !== "files") { @@ -275,7 +291,7 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) { useEffect(() => { updateTurnStripScrollState(); - }, [turnDiffSummaries, selectedTurnId, updateTurnStripScrollState]); + }, [orderedTurnDiffSummaries, selectedTurnId, updateTurnStripScrollState]); useEffect(() => { const element = turnStripRef.current; @@ -358,7 +374,7 @@ export default function DiffPanel({ mode = "inline" }: DiffPanelProps) {
All turns
- {turnDiffSummaries.map((summary, index) => ( + {orderedTurnDiffSummaries.map((summary) => (