From 640a2cca9f9effc434d1296d7b625ee09bcba64e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 5 Mar 2026 15:34:51 -0800 Subject: [PATCH] Persist proposed plans as first-class thread projection data - Add `thread.proposed-plan-upserted` orchestration flow from ingestion through decider/projector - Store proposed plans in a dedicated projection table and include them in snapshot queries - Update provider runtime/web handling and tests to use proposed plans instead of assistant message wrappers --- .../Layers/CheckpointDiffQuery.test.ts | 1 + .../Layers/ProjectionPipeline.ts | 84 +++++++ .../Layers/ProjectionSnapshotQuery.test.ts | 1 + .../Layers/ProjectionSnapshotQuery.ts | 45 ++++ .../Layers/ProviderRuntimeIngestion.test.ts | 78 ++++--- .../Layers/ProviderRuntimeIngestion.ts | 219 ++++++++++++++---- apps/server/src/orchestration/Schemas.ts | 2 + .../orchestration/commandInvariants.test.ts | 2 + apps/server/src/orchestration/decider.ts | 21 ++ apps/server/src/orchestration/projector.ts | 47 ++++ .../Layers/ProjectionThreadProposedPlans.ts | 104 +++++++++ apps/server/src/persistence/Migrations.ts | 2 + .../013_ProjectionThreadProposedPlans.ts | 22 ++ .../Services/ProjectionThreadProposedPlans.ts | 52 +++++ .../src/provider/Layers/CodexAdapter.test.ts | 53 ++++- .../src/provider/Layers/CodexAdapter.ts | 81 ++++++- apps/web/src/components/ChatView.tsx | 181 +++++++++------ apps/web/src/proposedPlan.test.ts | 73 ------ apps/web/src/proposedPlan.ts | 76 ------ apps/web/src/session-logic.test.ts | 108 ++++++++- apps/web/src/session-logic.ts | 70 +++++- apps/web/src/store.test.ts | 2 + apps/web/src/store.ts | 7 + apps/web/src/types.ts | 10 + apps/web/src/worktreeCleanup.test.ts | 1 + packages/contracts/src/orchestration.ts | 40 ++++ .../contracts/src/providerRuntime.test.ts | 22 ++ packages/contracts/src/providerRuntime.ts | 32 +++ 28 files changed, 1113 insertions(+), 323 deletions(-) create mode 100644 apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts create mode 100644 apps/server/src/persistence/Migrations/013_ProjectionThreadProposedPlans.ts create mode 100644 apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts diff --git a/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts index dce489cbb9f15125c23e214a51773dbd1ef68136..2f79ea9d5a5364d1b4adea535c4f793fef968c8c 100644 --- a/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts @@ -61,6 +61,7 @@ function makeSnapshot(input: { deletedAt: null, messages: [], activities: [], + proposedPlans: [], checkpoints: [ { turnId: TurnId.makeUnsafe("turn-1"), diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index cd8111fbcd13dd83b529a8c6f3b21a86f582b7ad..24b81d514ab198a926be61fca67dae942b4169e8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -18,6 +18,10 @@ import { type ProjectionThreadMessage, ProjectionThreadMessageRepository, } from "../../persistence/Services/ProjectionThreadMessages.ts"; +import { + type ProjectionThreadProposedPlan, + ProjectionThreadProposedPlanRepository, +} from "../../persistence/Services/ProjectionThreadProposedPlans.ts"; import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts"; import { type ProjectionTurn, @@ -29,6 +33,7 @@ import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/Projec import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts"; import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts"; import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts"; +import { ProjectionThreadProposedPlanRepositoryLive } from "../../persistence/Layers/ProjectionThreadProposedPlans.ts"; import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts"; import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts"; @@ -48,6 +53,7 @@ export const ORCHESTRATION_PROJECTOR_NAMES = { projects: "projection.projects", threads: "projection.threads", threadMessages: "projection.thread-messages", + threadProposedPlans: "projection.thread-proposed-plans", threadActivities: "projection.thread-activities", threadSessions: "projection.thread-sessions", threadTurns: "projection.thread-turns", @@ -189,6 +195,26 @@ function retainProjectionActivitiesAfterRevert( ); } +function retainProjectionProposedPlansAfterRevert( + proposedPlans: ReadonlyArray, + turns: ReadonlyArray, + turnCount: number, +): ReadonlyArray { + const retainedTurnIds = new Set( + turns + .filter( + (turn) => + turn.turnId !== null && + turn.checkpointTurnCount !== null && + turn.checkpointTurnCount <= turnCount, + ) + .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])), + ); + return proposedPlans.filter( + (proposedPlan) => proposedPlan.turnId === null || retainedTurnIds.has(proposedPlan.turnId), + ); +} + function collectThreadAttachmentRelativePaths( threadId: string, messages: ReadonlyArray, @@ -320,6 +346,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { const projectionProjectRepository = yield* ProjectionProjectRepository; const projectionThreadRepository = yield* ProjectionThreadRepository; const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository; + const projectionThreadProposedPlanRepository = yield* ProjectionThreadProposedPlanRepository; const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository; const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository; const projectionTurnRepository = yield* ProjectionTurnRepository; @@ -474,6 +501,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } case "thread.message-sent": + case "thread.proposed-plan-upserted": case "thread.activity-appended": { const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, @@ -615,6 +643,57 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { } }); + const applyThreadProposedPlansProjection: ProjectorDefinition["apply"] = ( + event, + _attachmentSideEffects, + ) => + Effect.gen(function* () { + switch (event.type) { + case "thread.proposed-plan-upserted": + yield* projectionThreadProposedPlanRepository.upsert({ + planId: event.payload.proposedPlan.id, + threadId: event.payload.threadId, + turnId: event.payload.proposedPlan.turnId, + planMarkdown: event.payload.proposedPlan.planMarkdown, + createdAt: event.payload.proposedPlan.createdAt, + updatedAt: event.payload.proposedPlan.updatedAt, + }); + return; + + case "thread.reverted": { + const existingRows = yield* projectionThreadProposedPlanRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + if (existingRows.length === 0) { + return; + } + + const existingTurns = yield* projectionTurnRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + const keptRows = retainProjectionProposedPlansAfterRevert( + existingRows, + existingTurns, + event.payload.turnCount, + ); + if (keptRows.length === existingRows.length) { + return; + } + + yield* projectionThreadProposedPlanRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + yield* Effect.forEach(keptRows, projectionThreadProposedPlanRepository.upsert, { + concurrency: 1, + }).pipe(Effect.asVoid); + return; + } + + default: + return; + } + }); + const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = ( event, _attachmentSideEffects, @@ -1023,6 +1102,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () { name: ORCHESTRATION_PROJECTOR_NAMES.threadMessages, apply: applyThreadMessagesProjection, }, + { + name: ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans, + apply: applyThreadProposedPlansProjection, + }, { name: ORCHESTRATION_PROJECTOR_NAMES.threadActivities, apply: applyThreadActivitiesProjection, @@ -1142,6 +1225,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect( Layer.provideMerge(ProjectionProjectRepositoryLive), Layer.provideMerge(ProjectionThreadRepositoryLive), Layer.provideMerge(ProjectionThreadMessageRepositoryLive), + Layer.provideMerge(ProjectionThreadProposedPlanRepositoryLive), Layer.provideMerge(ProjectionThreadActivityRepositoryLive), Layer.provideMerge(ProjectionThreadSessionRepositoryLive), Layer.provideMerge(ProjectionTurnRepositoryLive), diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 23074a741986ef5d00b3639ba9cc5100169f1ffc..e7e9cd4e1271745f33951ae7c244a77b6f0d0a62 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -260,6 +260,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { updatedAt: "2026-02-24T00:00:05.000Z", }, ], + proposedPlans: [], activities: [ { id: asEventId("activity-1"), diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 8af14b5ed0a973895f83dbe0e5b77a657ea6cd10..5fd38a5401750576273abb84c5915c76318b53e8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -10,6 +10,7 @@ import { type OrchestrationCheckpointSummary, type OrchestrationLatestTurn, type OrchestrationMessage, + type OrchestrationProposedPlan, type OrchestrationProject, type OrchestrationSession, type OrchestrationThread, @@ -30,6 +31,7 @@ import { ProjectionProject } from "../../persistence/Services/ProjectionProjects import { ProjectionState } from "../../persistence/Services/ProjectionState.ts"; import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts"; import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionThreadMessages.ts"; +import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts"; import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts"; import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts"; import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts"; @@ -50,6 +52,7 @@ const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields( attachments: Schema.NullOr(Schema.fromJsonString(Schema.Array(ChatAttachment))), }), ); +const ProjectionThreadProposedPlanDbRowSchema = ProjectionThreadProposedPlan; const ProjectionThreadDbRowSchema = ProjectionThread; const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( Struct.assign({ @@ -78,6 +81,7 @@ const REQUIRED_SNAPSHOT_PROJECTORS = [ ORCHESTRATION_PROJECTOR_NAMES.projects, ORCHESTRATION_PROJECTOR_NAMES.threads, ORCHESTRATION_PROJECTOR_NAMES.threadMessages, + ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans, ORCHESTRATION_PROJECTOR_NAMES.threadActivities, ORCHESTRATION_PROJECTOR_NAMES.threadSessions, ORCHESTRATION_PROJECTOR_NAMES.checkpoints, @@ -186,6 +190,23 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const listThreadProposedPlanRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProjectionThreadProposedPlanDbRowSchema, + execute: () => + sql` + SELECT + plan_id AS "planId", + thread_id AS "threadId", + turn_id AS "turnId", + plan_markdown AS "planMarkdown", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM projection_thread_proposed_plans + ORDER BY thread_id ASC, created_at ASC, plan_id ASC + `, + }); + const listThreadActivityRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionThreadActivityDbRowSchema, @@ -291,6 +312,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { projectRows, threadRows, messageRows, + proposedPlanRows, activityRows, sessionRows, checkpointRows, @@ -321,6 +343,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), + listThreadProposedPlanRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:query", + "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:decodeRows", + ), + ), + ), listThreadActivityRows(undefined).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -364,6 +394,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ]); const messagesByThread = new Map>(); + const proposedPlansByThread = new Map>(); const activitiesByThread = new Map>(); const checkpointsByThread = new Map>(); const sessionsByThread = new Map(); @@ -397,6 +428,19 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { messagesByThread.set(row.threadId, threadMessages); } + for (const row of proposedPlanRows) { + updatedAt = maxIso(updatedAt, row.updatedAt); + const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? []; + threadProposedPlans.push({ + id: row.planId, + turnId: row.turnId, + planMarkdown: row.planMarkdown, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }); + proposedPlansByThread.set(row.threadId, threadProposedPlans); + } + for (const row of activityRows) { updatedAt = maxIso(updatedAt, row.createdAt); const threadActivities = activitiesByThread.get(row.threadId) ?? []; @@ -494,6 +538,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { updatedAt: row.updatedAt, deletedAt: row.deletedAt, messages: messagesByThread.get(row.threadId) ?? [], + proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], activities: activitiesByThread.get(row.threadId) ?? [], checkpoints: checkpointsByThread.get(row.threadId) ?? [], session: sessionsByThread.get(row.threadId) ?? null, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 8fd64abd0203dc15c2d194845c9cd039b02e2be2..ec42c4ec7a53a797e8a59a6d93b08615e63f4551 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -103,6 +103,7 @@ async function waitForThread( type ProviderRuntimeTestReadModel = OrchestrationReadModel; type ProviderRuntimeTestThread = ProviderRuntimeTestReadModel["threads"][number]; type ProviderRuntimeTestMessage = ProviderRuntimeTestThread["messages"][number]; +type ProviderRuntimeTestProposedPlan = ProviderRuntimeTestThread["proposedPlans"][number]; type ProviderRuntimeTestActivity = ProviderRuntimeTestThread["activities"][number]; type ProviderRuntimeTestCheckpoint = ProviderRuntimeTestThread["checkpoints"][number]; @@ -633,41 +634,35 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); - it("projects completed plan items into finalized proposed-plan assistant messages", async () => { + it("projects completed plan items into first-class proposed plans", async () => { const harness = await createHarness(); const now = new Date().toISOString(); harness.emit({ - type: "item.completed", + type: "turn.proposed.completed", eventId: asEventId("evt-plan-item-completed"), provider: "codex", createdAt: now, threadId: asThreadId("thread-1"), turnId: asTurnId("turn-plan-final"), - itemId: asItemId("item-plan-final"), payload: { - itemType: "plan", - status: "completed", - detail: "## Ship plan\n\n- wire projection\n- render follow-up", + planMarkdown: "## Ship plan\n\n- wire projection\n- render follow-up", }, }); - const thread = await waitForThread(harness.engine, (entry) => { - const message = entry.messages.find( - (candidate: ProviderRuntimeTestMessage) => candidate.id === "assistant:item-plan-final", - ); - return message?.streaming === false && message.text.includes(""); - }); - const message = thread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-plan-final", + const thread = await waitForThread(harness.engine, (entry) => + entry.proposedPlans.some( + (proposedPlan: ProviderRuntimeTestProposedPlan) => + proposedPlan.id === "plan:thread-1:turn:turn-plan-final", + ), ); - expect(message?.text).toBe( - "\n## Ship plan\n\n- wire projection\n- render follow-up\n", + const proposedPlan = thread.proposedPlans.find( + (entry: ProviderRuntimeTestProposedPlan) => entry.id === "plan:thread-1:turn:turn-plan-final", ); - expect(message?.streaming).toBe(false); + expect(proposedPlan?.planMarkdown).toBe("## Ship plan\n\n- wire projection\n- render follow-up"); }); - it("finalizes buffered plan deltas into a proposed-plan assistant message on turn completion", async () => { + it("finalizes buffered proposed-plan deltas into a first-class proposed plan on turn completion", async () => { const harness = await createHarness(); const now = new Date().toISOString(); @@ -687,28 +682,24 @@ describe("ProviderRuntimeIngestion", () => { ); harness.emit({ - type: "content.delta", + type: "turn.proposed.delta", eventId: asEventId("evt-plan-delta-1"), provider: "codex", createdAt: now, threadId: asThreadId("thread-1"), turnId: asTurnId("turn-plan-buffer"), - itemId: asItemId("item-plan-buffer"), payload: { - streamKind: "plan_text", delta: "## Buffered plan\n\n- first", }, }); harness.emit({ - type: "content.delta", + type: "turn.proposed.delta", eventId: asEventId("evt-plan-delta-2"), provider: "codex", createdAt: now, threadId: asThreadId("thread-1"), turnId: asTurnId("turn-plan-buffer"), - itemId: asItemId("item-plan-buffer"), payload: { - streamKind: "plan_text", delta: "\n- second", }, }); @@ -724,19 +715,16 @@ describe("ProviderRuntimeIngestion", () => { }, }); - const thread = await waitForThread(harness.engine, (entry) => { - const message = entry.messages.find( - (candidate: ProviderRuntimeTestMessage) => candidate.id === "assistant:item-plan-buffer", - ); - return message?.streaming === false && message.text.includes(""); - }); - const message = thread.messages.find( - (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-plan-buffer", + const thread = await waitForThread(harness.engine, (entry) => + entry.proposedPlans.some( + (proposedPlan: ProviderRuntimeTestProposedPlan) => + proposedPlan.id === "plan:thread-1:turn:turn-plan-buffer", + ), ); - expect(message?.text).toBe( - "\n## Buffered plan\n\n- first\n- second\n", + const proposedPlan = thread.proposedPlans.find( + (entry: ProviderRuntimeTestProposedPlan) => entry.id === "plan:thread-1:turn:turn-plan-buffer", ); - expect(message?.streaming).toBe(false); + expect(proposedPlan?.planMarkdown).toBe("## Buffered plan\n\n- first\n- second"); }); it("buffers assistant deltas by default until completion", async () => { @@ -1370,12 +1358,27 @@ describe("ProviderRuntimeIngestion", () => { summary: "\n# Plan title\n", }, }); + harness.emit({ + type: "turn.proposed.completed", + eventId: asEventId("evt-task-proposed-plan-completed"), + provider: "codex", + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-task-1"), + payload: { + planMarkdown: "# Plan title", + }, + }); const thread = await waitForThread( harness.engine, (entry) => entry.activities.some( (activity: ProviderRuntimeTestActivity) => activity.kind === "task.completed", + ) && + entry.proposedPlans.some( + (proposedPlan: ProviderRuntimeTestProposedPlan) => + proposedPlan.id === "plan:thread-1:turn:turn-task-1", ), ); @@ -1406,6 +1409,11 @@ describe("ProviderRuntimeIngestion", () => { ); expect(completed?.kind).toBe("task.completed"); expect(completedPayload?.detail).toBe("\n# Plan title\n"); + expect( + thread.proposedPlans.find( + (entry: ProviderRuntimeTestProposedPlan) => entry.id === "plan:thread-1:turn:turn-task-1", + )?.planMarkdown, + ).toBe("# Plan title"); }); it("projects structured user input request and resolution as thread activities", async () => { diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 6ab4653b293872105512073603c3744d7f11f52a..04f6d8523437310d547f4f7629c1a82e96da6386 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -26,10 +26,10 @@ const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId const DEFAULT_ASSISTANT_DELIVERY_MODE: AssistantDeliveryMode = "buffered"; const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000; const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120); -const MESSAGE_STREAM_KIND_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000; -const MESSAGE_STREAM_KIND_BY_MESSAGE_ID_TTL = Duration.minutes(120); const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000; const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120); +const BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY = 10_000; +const BUFFERED_PROPOSED_PLAN_BY_ID_TTL = Duration.minutes(120); const MAX_BUFFERED_ASSISTANT_CHARS = 24_000; const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0"; @@ -67,12 +67,27 @@ function truncateDetail(value: string, limit = 180): string { return value.length > limit ? `${value.slice(0, limit - 3)}...` : value; } -function wrapProposedPlanMessage(planMarkdown: string | undefined): string | undefined { +function normalizeProposedPlanMarkdown(planMarkdown: string | undefined): string | undefined { const trimmed = planMarkdown?.trim(); if (!trimmed) { return undefined; } - return `\n${trimmed}\n`; + return trimmed; +} + +function proposedPlanIdForTurn(threadId: ThreadId, turnId: TurnId): string { + return `plan:${threadId}:turn:${turnId}`; +} + +function proposedPlanIdFromEvent(event: ProviderRuntimeEvent, threadId: ThreadId): string { + const turnId = toTurnId(event.turnId); + if (turnId) { + return proposedPlanIdForTurn(threadId, turnId); + } + if (event.itemId) { + return `plan:${threadId}:item:${event.itemId}`; + } + return `plan:${threadId}:event:${event.eventId}`; } function asString(value: unknown): string | undefined { @@ -480,18 +495,18 @@ const make = Effect.gen(function* () { lookup: () => Effect.succeed(new Set()), }); - const messageStreamKindByMessageId = yield* Cache.make({ - capacity: MESSAGE_STREAM_KIND_BY_MESSAGE_ID_CACHE_CAPACITY, - timeToLive: MESSAGE_STREAM_KIND_BY_MESSAGE_ID_TTL, - lookup: () => Effect.succeed("assistant_text"), - }); - const bufferedAssistantTextByMessageId = yield* Cache.make({ capacity: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY, timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL, lookup: () => Effect.succeed(""), }); + const bufferedProposedPlanById = yield* Cache.make({ + capacity: BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY, + timeToLive: BUFFERED_PROPOSED_PLAN_BY_ID_TTL, + lookup: () => Effect.succeed({ text: "", createdAt: "" }), + }); + const rememberAssistantMessageId = ( threadId: ThreadId, turnId: TurnId, @@ -545,16 +560,6 @@ const make = Effect.gen(function* () { const clearAssistantMessageIdsForTurn = (threadId: ThreadId, turnId: TurnId) => Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(threadId, turnId)); - const rememberMessageStreamKind = ( - messageId: MessageId, - streamKind: "assistant_text" | "plan_text", - ) => Cache.set(messageStreamKindByMessageId, messageId, streamKind); - - const getMessageStreamKind = (messageId: MessageId) => - Cache.getOption(messageStreamKindByMessageId, messageId).pipe( - Effect.map((streamKind) => Option.getOrElse(streamKind, () => "assistant_text" as const)), - ); - const appendBufferedAssistantText = (messageId: MessageId, delta: string) => Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe( Effect.flatMap((existingText) => @@ -587,11 +592,30 @@ const make = Effect.gen(function* () { const clearBufferedAssistantText = (messageId: MessageId) => Cache.invalidate(bufferedAssistantTextByMessageId, messageId); - const clearAssistantMessageState = (messageId: MessageId) => - Effect.all([ - clearBufferedAssistantText(messageId), - Cache.invalidate(messageStreamKindByMessageId, messageId), - ]).pipe(Effect.asVoid); + const appendBufferedProposedPlan = (planId: string, delta: string, createdAt: string) => + Cache.getOption(bufferedProposedPlanById, planId).pipe( + Effect.flatMap((existingEntry) => { + const existing = Option.getOrUndefined(existingEntry); + return Cache.set(bufferedProposedPlanById, planId, { + text: `${existing?.text ?? ""}${delta}`, + createdAt: existing?.createdAt && existing.createdAt.length > 0 ? existing.createdAt : createdAt, + }); + }), + ); + + const takeBufferedProposedPlan = (planId: string) => + Cache.getOption(bufferedProposedPlanById, planId).pipe( + Effect.flatMap((existingEntry) => + Cache.invalidate(bufferedProposedPlanById, planId).pipe( + Effect.as(Option.getOrUndefined(existingEntry)), + ), + ), + ); + + const clearBufferedProposedPlan = (planId: string) => + Cache.invalidate(bufferedProposedPlanById, planId); + + const clearAssistantMessageState = (messageId: MessageId) => clearBufferedAssistantText(messageId); const finalizeAssistantMessage = (input: { event: ProviderRuntimeEvent; @@ -604,17 +628,10 @@ const make = Effect.gen(function* () { fallbackText?: string; }) => Effect.gen(function* () { - const messageStreamKind = yield* getMessageStreamKind(input.messageId); const bufferedText = yield* takeBufferedAssistantText(input.messageId); - const bufferedFinalText = - bufferedText.length > 0 - ? messageStreamKind === "plan_text" - ? (wrapProposedPlanMessage(bufferedText) ?? "") - : bufferedText - : ""; const text = - bufferedFinalText.length > 0 - ? bufferedFinalText + bufferedText.length > 0 + ? bufferedText : (input.fallbackText?.trim().length ?? 0) > 0 ? input.fallbackText! : ""; @@ -642,10 +659,84 @@ const make = Effect.gen(function* () { yield* clearAssistantMessageState(input.messageId); }); + const upsertProposedPlan = (input: { + event: ProviderRuntimeEvent; + threadId: ThreadId; + threadProposedPlans: ReadonlyArray<{ + id: string; + createdAt: string; + }>; + planId: string; + turnId?: TurnId; + planMarkdown: string | undefined; + createdAt: string; + updatedAt: string; + }) => + Effect.gen(function* () { + const planMarkdown = normalizeProposedPlanMarkdown(input.planMarkdown); + if (!planMarkdown) { + return; + } + + const existingPlan = input.threadProposedPlans.find((entry) => entry.id === input.planId); + yield* orchestrationEngine.dispatch({ + type: "thread.proposed-plan.upsert", + commandId: providerCommandId(input.event, "proposed-plan-upsert"), + threadId: input.threadId, + proposedPlan: { + id: input.planId, + turnId: input.turnId ?? null, + planMarkdown, + createdAt: existingPlan?.createdAt ?? input.createdAt, + updatedAt: input.updatedAt, + }, + createdAt: input.updatedAt, + }); + }); + + const finalizeBufferedProposedPlan = (input: { + event: ProviderRuntimeEvent; + threadId: ThreadId; + threadProposedPlans: ReadonlyArray<{ + id: string; + createdAt: string; + }>; + planId: string; + turnId?: TurnId; + fallbackMarkdown?: string; + updatedAt: string; + }) => + Effect.gen(function* () { + const bufferedPlan = yield* takeBufferedProposedPlan(input.planId); + const bufferedMarkdown = normalizeProposedPlanMarkdown(bufferedPlan?.text); + const fallbackMarkdown = normalizeProposedPlanMarkdown(input.fallbackMarkdown); + const planMarkdown = bufferedMarkdown ?? fallbackMarkdown; + if (!planMarkdown) { + return; + } + + yield* upsertProposedPlan({ + event: input.event, + threadId: input.threadId, + threadProposedPlans: input.threadProposedPlans, + planId: input.planId, + ...(input.turnId ? { turnId: input.turnId } : {}), + planMarkdown, + createdAt: + bufferedPlan?.createdAt && bufferedPlan.createdAt.length > 0 + ? bufferedPlan.createdAt + : input.updatedAt, + updatedAt: input.updatedAt, + }); + yield* clearBufferedProposedPlan(input.planId); + }); + const clearTurnStateForSession = (threadId: ThreadId) => Effect.gen(function* () { const prefix = `${threadId}:`; + const proposedPlanPrefix = `plan:${threadId}:`; const turnKeys = Array.from(yield* Cache.keys(turnMessageIdsByTurnKey)); + const proposedPlanKeys = Array.from(yield* Cache.keys(bufferedProposedPlanById)); yield* Effect.forEach( turnKeys, (key) => @@ -665,6 +756,14 @@ const make = Effect.gen(function* () { }), { concurrency: 1 }, ).pipe(Effect.asVoid); + yield* Effect.forEach( + proposedPlanKeys, + (key) => + key.startsWith(proposedPlanPrefix) + ? Cache.invalidate(bufferedProposedPlanById, key) + : Effect.void, + { concurrency: 1 }, + ).pipe(Effect.asVoid); }); const processRuntimeEvent = (event: ProviderRuntimeEvent) => @@ -768,24 +867,20 @@ const make = Effect.gen(function* () { } const assistantDelta = - event.type === "content.delta" && - (event.payload.streamKind === "assistant_text" || event.payload.streamKind === "plan_text") + event.type === "content.delta" && event.payload.streamKind === "assistant_text" ? event.payload.delta : undefined; + const proposedPlanDelta = + event.type === "turn.proposed.delta" ? event.payload.delta : undefined; if (assistantDelta && assistantDelta.length > 0) { const assistantMessageId = MessageId.makeUnsafe( `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, ); const turnId = toTurnId(event.turnId); - const messageStreamKind = - event.type === "content.delta" && event.payload.streamKind === "plan_text" - ? "plan_text" - : "assistant_text"; if (turnId) { yield* rememberAssistantMessageId(thread.id, turnId, assistantMessageId); } - yield* rememberMessageStreamKind(assistantMessageId, messageStreamKind); const assistantDeliveryMode = yield* Ref.get(assistantDeliveryModeRef); if (assistantDeliveryMode === "buffered") { @@ -814,15 +909,24 @@ const make = Effect.gen(function* () { } } + if (proposedPlanDelta && proposedPlanDelta.length > 0) { + const planId = proposedPlanIdFromEvent(event, thread.id); + yield* appendBufferedProposedPlan(planId, proposedPlanDelta, now); + } + const assistantCompletion = - event.type === "item.completed" && - (event.payload.itemType === "assistant_message" || event.payload.itemType === "plan") + event.type === "item.completed" && event.payload.itemType === "assistant_message" ? { messageId: MessageId.makeUnsafe(`assistant:${event.itemId ?? event.turnId ?? event.eventId}`), - fallbackText: - event.payload.itemType === "plan" - ? wrapProposedPlanMessage(event.payload.detail) - : event.payload.detail, + fallbackText: event.payload.detail, + } + : undefined; + const proposedPlanCompletion = + event.type === "turn.proposed.completed" + ? { + planId: proposedPlanIdFromEvent(event, thread.id), + turnId: toTurnId(event.turnId), + planMarkdown: event.payload.planMarkdown, } : undefined; @@ -851,6 +955,18 @@ const make = Effect.gen(function* () { } } + if (proposedPlanCompletion) { + yield* finalizeBufferedProposedPlan({ + event, + threadId: thread.id, + threadProposedPlans: thread.proposedPlans, + planId: proposedPlanCompletion.planId, + ...(proposedPlanCompletion.turnId ? { turnId: proposedPlanCompletion.turnId } : {}), + fallbackMarkdown: proposedPlanCompletion.planMarkdown, + updatedAt: now, + }); + } + if (event.type === "turn.completed") { const turnId = toTurnId(event.turnId); if (turnId) { @@ -870,6 +986,15 @@ const make = Effect.gen(function* () { { concurrency: 1 }, ).pipe(Effect.asVoid); yield* clearAssistantMessageIdsForTurn(thread.id, turnId); + + yield* finalizeBufferedProposedPlan({ + event, + threadId: thread.id, + threadProposedPlans: thread.proposedPlans, + planId: proposedPlanIdForTurn(thread.id, turnId), + turnId, + updatedAt: now, + }); } } diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 9a6922c56d0f36e09e05df05e4d6ffbacfba8ca0..c96385cad1c4155626ad551d366eebba0b5c508e 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -8,6 +8,7 @@ import { ThreadInteractionModeSetPayload as ContractsThreadInteractionModeSetPayloadSchema, ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema, ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema, + ThreadProposedPlanUpsertedPayload as ContractsThreadProposedPlanUpsertedPayloadSchema, ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema, ThreadTurnDiffCompletedPayload as ContractsThreadTurnDiffCompletedPayloadSchema, ThreadRevertedPayload as ContractsThreadRevertedPayloadSchema, @@ -31,6 +32,7 @@ export const ThreadInteractionModeSetPayload = ContractsThreadInteractionModeSet export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema; export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema; +export const ThreadProposedPlanUpsertedPayload = ContractsThreadProposedPlanUpsertedPayloadSchema; export const ThreadSessionSetPayload = ContractsThreadSessionSetPayloadSchema; export const ThreadTurnDiffCompletedPayload = ContractsThreadTurnDiffCompletedPayloadSchema; export const ThreadRevertedPayload = ContractsThreadRevertedPayloadSchema; diff --git a/apps/server/src/orchestration/commandInvariants.test.ts b/apps/server/src/orchestration/commandInvariants.test.ts index ae2ec25f9793be345e162ace7b07a17ad0b78d04..f95e4db754bc20f575c2529a42a9e2138ee3e189 100644 --- a/apps/server/src/orchestration/commandInvariants.test.ts +++ b/apps/server/src/orchestration/commandInvariants.test.ts @@ -61,6 +61,7 @@ const readModel: OrchestrationReadModel = { messages: [], session: null, activities: [], + proposedPlans: [], checkpoints: [], deletedAt: null, }, @@ -79,6 +80,7 @@ const readModel: OrchestrationReadModel = { messages: [], session: null, activities: [], + proposedPlans: [], checkpoints: [], deletedAt: null, }, diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 48afbcea40b0654bde83a31def418c1a757026d5..fcc5dba08598b36e14c1702657cc8463b6c9cb11 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -508,6 +508,27 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.proposed-plan.upsert": { + yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + return { + ...withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + }), + type: "thread.proposed-plan-upserted", + payload: { + threadId: command.threadId, + proposedPlan: command.proposedPlan, + }, + }; + } + case "thread.turn.diff.complete": { yield* requireThread({ readModel, diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index af4af5e2cd8bdcced51ca8bec45e50758d8c8eff..c0badfe95880d5af248c2b0d7a3d844373641472 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -18,6 +18,7 @@ import { ThreadDeletedPayload, ThreadInteractionModeSetPayload, ThreadMetaUpdatedPayload, + ThreadProposedPlanUpsertedPayload, ThreadRuntimeModeSetPayload, ThreadRevertedPayload, ThreadSessionSetPayload, @@ -126,6 +127,15 @@ function retainThreadActivitiesAfterRevert( ); } +function retainThreadProposedPlansAfterRevert( + proposedPlans: ReadonlyArray, + retainedTurnIds: ReadonlySet, +): ReadonlyArray { + return proposedPlans.filter( + (proposedPlan) => proposedPlan.turnId === null || retainedTurnIds.has(proposedPlan.turnId), + ); +} + function compareThreadActivities( left: OrchestrationThread["activities"][number], right: OrchestrationThread["activities"][number], @@ -435,6 +445,38 @@ export function projectEvent( }; }); + case "thread.proposed-plan-upserted": + return Effect.gen(function* () { + const payload = yield* decodeForEvent( + ThreadProposedPlanUpsertedPayload, + event.payload, + event.type, + "payload", + ); + const thread = nextBase.threads.find((entry) => entry.id === payload.threadId); + if (!thread) { + return nextBase; + } + + const proposedPlans = [ + ...thread.proposedPlans.filter((entry) => entry.id !== payload.proposedPlan.id), + payload.proposedPlan, + ] + .toSorted( + (left, right) => + left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id), + ) + .slice(-200); + + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + proposedPlans, + updatedAt: event.occurredAt, + }), + }; + }); + case "thread.turn-diff-completed": return Effect.gen(function* () { const payload = yield* decodeForEvent( @@ -511,6 +553,10 @@ export function projectEvent( retainedTurnIds, payload.turnCount, ).slice(-MAX_THREAD_MESSAGES); + const proposedPlans = retainThreadProposedPlansAfterRevert( + thread.proposedPlans, + retainedTurnIds, + ).slice(-200); const activities = retainThreadActivitiesAfterRevert(thread.activities, retainedTurnIds); const latestCheckpoint = checkpoints.at(-1) ?? null; @@ -531,6 +577,7 @@ export function projectEvent( threads: updateThread(nextBase.threads, payload.threadId, { checkpoints, messages, + proposedPlans, activities, latestTurn, updatedAt: event.occurredAt, diff --git a/apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts b/apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts new file mode 100644 index 0000000000000000000000000000000000000000..24446e04dc38e8ec0aa0765a9d4dc793f95b0573 --- /dev/null +++ b/apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts @@ -0,0 +1,104 @@ +import { Effect, Layer } from "effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; + +import { toPersistenceSqlError } from "../Errors.ts"; +import { + DeleteProjectionThreadProposedPlansInput, + ListProjectionThreadProposedPlansInput, + ProjectionThreadProposedPlan, + ProjectionThreadProposedPlanRepository, + type ProjectionThreadProposedPlanRepositoryShape, +} from "../Services/ProjectionThreadProposedPlans.ts"; + +const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const upsertProjectionThreadProposedPlanRow = SqlSchema.void({ + Request: ProjectionThreadProposedPlan, + execute: (row) => sql` + INSERT INTO projection_thread_proposed_plans ( + plan_id, + thread_id, + turn_id, + plan_markdown, + created_at, + updated_at + ) + VALUES ( + ${row.planId}, + ${row.threadId}, + ${row.turnId}, + ${row.planMarkdown}, + ${row.createdAt}, + ${row.updatedAt} + ) + ON CONFLICT (plan_id) + DO UPDATE SET + thread_id = excluded.thread_id, + turn_id = excluded.turn_id, + plan_markdown = excluded.plan_markdown, + created_at = excluded.created_at, + updated_at = excluded.updated_at + `, + }); + + const listProjectionThreadProposedPlanRows = SqlSchema.findAll({ + Request: ListProjectionThreadProposedPlansInput, + Result: ProjectionThreadProposedPlan, + execute: ({ threadId }) => sql` + SELECT + plan_id AS "planId", + thread_id AS "threadId", + turn_id AS "turnId", + plan_markdown AS "planMarkdown", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM projection_thread_proposed_plans + WHERE thread_id = ${threadId} + ORDER BY created_at ASC, plan_id ASC + `, + }); + + const deleteProjectionThreadProposedPlanRows = SqlSchema.void({ + Request: DeleteProjectionThreadProposedPlansInput, + execute: ({ threadId }) => sql` + DELETE FROM projection_thread_proposed_plans + WHERE thread_id = ${threadId} + `, + }); + + const upsert: ProjectionThreadProposedPlanRepositoryShape["upsert"] = (row) => + upsertProjectionThreadProposedPlanRow(row).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionThreadProposedPlanRepository.upsert:query"), + ), + ); + + const listByThreadId: ProjectionThreadProposedPlanRepositoryShape["listByThreadId"] = (input) => + listProjectionThreadProposedPlanRows(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionThreadProposedPlanRepository.listByThreadId:query"), + ), + ); + + const deleteByThreadId: ProjectionThreadProposedPlanRepositoryShape["deleteByThreadId"] = ( + input, + ) => + deleteProjectionThreadProposedPlanRows(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProjectionThreadProposedPlanRepository.deleteByThreadId:query"), + ), + ); + + return { + upsert, + listByThreadId, + deleteByThreadId, + } satisfies ProjectionThreadProposedPlanRepositoryShape; +}); + +export const ProjectionThreadProposedPlanRepositoryLive = Layer.effect( + ProjectionThreadProposedPlanRepository, + makeProjectionThreadProposedPlanRepository, +); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 83836a7973299a5e55997455cf9723ca082d76e2..7deb890dd854eb7223597d94b1251eeed6f54fd5 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -24,6 +24,7 @@ import Migration0009 from "./Migrations/009_ProviderSessionRuntimeMode.ts"; import Migration0010 from "./Migrations/010_ProjectionThreadsRuntimeMode.ts"; import Migration0011 from "./Migrations/011_OrchestrationThreadCreatedRuntimeMode.ts"; import Migration0012 from "./Migrations/012_ProjectionThreadsInteractionMode.ts"; +import Migration0013 from "./Migrations/013_ProjectionThreadProposedPlans.ts"; import { Effect } from "effect"; /** @@ -49,6 +50,7 @@ const loader = Migrator.fromRecord({ "10_ProjectionThreadsRuntimeMode": Migration0010, "11_OrchestrationThreadCreatedRuntimeMode": Migration0011, "12_ProjectionThreadsInteractionMode": Migration0012, + "13_ProjectionThreadProposedPlans": Migration0013, }); /** diff --git a/apps/server/src/persistence/Migrations/013_ProjectionThreadProposedPlans.ts b/apps/server/src/persistence/Migrations/013_ProjectionThreadProposedPlans.ts new file mode 100644 index 0000000000000000000000000000000000000000..2af642c838a321f22fb311cd4a415da774b4a204 --- /dev/null +++ b/apps/server/src/persistence/Migrations/013_ProjectionThreadProposedPlans.ts @@ -0,0 +1,22 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS projection_thread_proposed_plans ( + plan_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + turn_id TEXT, + plan_markdown TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_thread_proposed_plans_thread_created + ON projection_thread_proposed_plans(thread_id, created_at) + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts b/apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts new file mode 100644 index 0000000000000000000000000000000000000000..ee662d52be029b167348c44d4d8e8770b4680aec --- /dev/null +++ b/apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts @@ -0,0 +1,52 @@ +import { + IsoDateTime, + OrchestrationProposedPlanId, + ThreadId, + TrimmedNonEmptyString, + TurnId, +} from "@t3tools/contracts"; +import { Schema, ServiceMap } from "effect"; +import type { Effect } from "effect"; + +import type { ProjectionRepositoryError } from "../Errors.ts"; + +export const ProjectionThreadProposedPlan = Schema.Struct({ + planId: OrchestrationProposedPlanId, + threadId: ThreadId, + turnId: Schema.NullOr(TurnId), + planMarkdown: TrimmedNonEmptyString, + createdAt: IsoDateTime, + updatedAt: IsoDateTime, +}); +export type ProjectionThreadProposedPlan = typeof ProjectionThreadProposedPlan.Type; + +export const ListProjectionThreadProposedPlansInput = Schema.Struct({ + threadId: ThreadId, +}); +export type ListProjectionThreadProposedPlansInput = + typeof ListProjectionThreadProposedPlansInput.Type; + +export const DeleteProjectionThreadProposedPlansInput = Schema.Struct({ + threadId: ThreadId, +}); +export type DeleteProjectionThreadProposedPlansInput = + typeof DeleteProjectionThreadProposedPlansInput.Type; + +export interface ProjectionThreadProposedPlanRepositoryShape { + readonly upsert: ( + proposedPlan: ProjectionThreadProposedPlan, + ) => Effect.Effect; + readonly listByThreadId: ( + input: ListProjectionThreadProposedPlansInput, + ) => Effect.Effect, ProjectionRepositoryError>; + readonly deleteByThreadId: ( + input: DeleteProjectionThreadProposedPlansInput, + ) => Effect.Effect; +} + +export class ProjectionThreadProposedPlanRepository extends ServiceMap.Service< + ProjectionThreadProposedPlanRepository, + ProjectionThreadProposedPlanRepositoryShape +>()( + "t3/persistence/Services/ProjectionThreadProposedPlans/ProjectionThreadProposedPlanRepository", +) {} diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 8d76583413be7d123861470193fbea74bcfccff9..1d54d83e3a759a324057d4756e7d971f3e574237 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -270,7 +270,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { }), ); - it.effect("maps completed plan items to canonical item.completed plan events", () => + it.effect("maps completed plan items to canonical proposed-plan completion events", () => Effect.gen(function* () { const adapter = yield* CodexAdapter; const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); @@ -300,13 +300,46 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { if (firstEvent._tag !== "Some") { return; } - assert.equal(firstEvent.value.type, "item.completed"); - if (firstEvent.value.type !== "item.completed") { + assert.equal(firstEvent.value.type, "turn.proposed.completed"); + if (firstEvent.value.type !== "turn.proposed.completed") { return; } - assert.equal(firstEvent.value.itemId, "plan_1"); - assert.equal(firstEvent.value.payload.itemType, "plan"); - assert.equal(firstEvent.value.payload.detail, "## Final plan\n\n- one\n- two"); + assert.equal(firstEvent.value.turnId, "turn-1"); + assert.equal(firstEvent.value.payload.planMarkdown, "## Final plan\n\n- one\n- two"); + }), + ); + + it.effect("maps plan deltas to canonical proposed-plan delta events", () => + Effect.gen(function* () { + const adapter = yield* CodexAdapter; + const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild); + + lifecycleManager.emit("event", { + id: asEventId("evt-plan-delta"), + kind: "notification", + provider: "codex", + createdAt: new Date().toISOString(), + method: "item/plan/delta", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-1"), + itemId: asItemId("plan_1"), + payload: { + delta: "## Final plan", + }, + } satisfies ProviderEvent); + + const firstEvent = yield* Fiber.join(firstEventFiber); + + assert.equal(firstEvent._tag, "Some"); + if (firstEvent._tag !== "Some") { + return; + } + assert.equal(firstEvent.value.type, "turn.proposed.delta"); + if (firstEvent.value.type !== "turn.proposed.delta") { + return; + } + assert.equal(firstEvent.value.turnId, "turn-1"); + assert.equal(firstEvent.value.payload.delta, "## Final plan"); }), ); @@ -487,7 +520,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { it.effect("maps Codex task and reasoning event chunks into canonical runtime events", () => Effect.gen(function* () { const adapter = yield* CodexAdapter; - const eventsFiber = yield* Stream.runCollect(Stream.take(adapter.streamEvents, 4)).pipe( + const eventsFiber = yield* Stream.runCollect(Stream.take(adapter.streamEvents, 5)).pipe( Effect.forkChild, ); @@ -592,6 +625,12 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => { assert.equal(events[3].payload.taskId, "turn-structured-1"); assert.equal(events[3].payload.summary, "\n# Ship it\n"); } + + assert.equal(events[4]?.type, "turn.proposed.completed"); + if (events[4]?.type === "turn.proposed.completed") { + assert.equal(events[4].turnId, "turn-structured-1"); + assert.equal(events[4].payload.planMarkdown, "# Ship it"); + } }), ); }); diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index d4558a8fd235d39930d2893e4d980a947a3bfbe0..8f547abf260b9fc9bd4af883790638aec10a5507 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -374,8 +374,6 @@ function contentStreamKindFromMethod( return "reasoning_text"; case "item/reasoning/summaryTextDelta": return "reasoning_summary_text"; - case "item/plan/delta": - return "plan_text"; case "item/commandExecution/outputDelta": return "command_output"; case "item/fileChange/outputDelta": @@ -385,6 +383,14 @@ function contentStreamKindFromMethod( } } +const PROPOSED_PLAN_BLOCK_REGEX = /\s*([\s\S]*?)\s*<\/proposed_plan>/i; + +function extractProposedPlanMarkdown(text: string | undefined): string | undefined { + const match = text ? PROPOSED_PLAN_BLOCK_REGEX.exec(text) : null; + const planMarkdown = match?.[1]?.trim(); + return planMarkdown && planMarkdown.length > 0 ? planMarkdown : undefined; +} + function asRuntimeItemId(itemId: ProviderItemId): RuntimeItemId { return RuntimeItemId.makeUnsafe(itemId); } @@ -487,7 +493,7 @@ function mapItemLifecycle( return undefined; } - const detail = itemDetail(source, payload ?? source); + const detail = itemDetail(source, payload ?? {}); const status = lifecycle === "item.started" ? "inProgress" @@ -800,6 +806,28 @@ function mapToRuntimeEvents( } if (event.method === "item/completed") { + const payload = asObject(event.payload); + const item = asObject(payload?.item); + const source = item ?? payload; + if (!source) { + return []; + } + const itemType = source ? toCanonicalItemType(source.type ?? source.kind) : "unknown"; + if (itemType === "plan") { + const detail = itemDetail(source, payload ?? {}); + if (!detail) { + return []; + } + return [ + { + ...runtimeEventBase(event, canonicalThreadId), + type: "turn.proposed.completed", + payload: { + planMarkdown: detail, + }, + }, + ]; + } const completed = mapItemLifecycle(event, canonicalThreadId, "item.completed"); return completed ? [completed] : []; } @@ -812,9 +840,28 @@ function mapToRuntimeEvents( return updated ? [updated] : []; } + if (event.method === "item/plan/delta") { + const delta = + event.textDelta ?? + asString(payload?.delta) ?? + asString(payload?.text) ?? + asString(asObject(payload?.content)?.text); + if (!delta || delta.length === 0) { + return []; + } + return [ + { + ...runtimeEventBase(event, canonicalThreadId), + type: "turn.proposed.delta", + payload: { + delta, + }, + }, + ]; + } + if ( event.method === "item/agentMessage/delta" || - event.method === "item/plan/delta" || event.method === "item/commandExecution/outputDelta" || event.method === "item/fileChange/outputDelta" || event.method === "item/reasoning/summaryTextDelta" || @@ -919,10 +966,22 @@ function mapToRuntimeEvents( if (event.method === "codex/event/task_complete") { const msg = codexEventMessage(payload); const taskId = asString(payload?.id) ?? asString(msg?.turn_id); + const proposedPlanMarkdown = extractProposedPlanMarkdown(asString(msg?.last_agent_message)); if (!taskId) { - return []; + if (!proposedPlanMarkdown) { + return []; + } + return [ + { + ...codexEventBase(event, canonicalThreadId), + type: "turn.proposed.completed", + payload: { + planMarkdown: proposedPlanMarkdown, + }, + }, + ]; } - return [ + const events: ProviderRuntimeEvent[] = [ { ...codexEventBase(event, canonicalThreadId), type: "task.completed", @@ -935,6 +994,16 @@ function mapToRuntimeEvents( }, }, ]; + if (proposedPlanMarkdown) { + events.push({ + ...codexEventBase(event, canonicalThreadId), + type: "turn.proposed.completed", + payload: { + planMarkdown: proposedPlanMarkdown, + }, + }); + } + return events; } if (event.method === "codex/event/agent_reasoning") { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index d6400508c5b19ae576754c9abeae16d90ab83ed9..55d34424aad4a85eec434f35746bdc301bbf5dcb 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -72,6 +72,7 @@ import { derivePhase, deriveTimelineEntries, deriveActivePlanState, + findLatestProposedPlan, type PendingApproval, type PendingUserInput, PROVIDER_OPTIONS, @@ -93,8 +94,6 @@ import { buildPlanImplementationThreadTitle, buildPlanImplementationPrompt, buildProposedPlanMarkdownFilename, - findLatestProposedPlanMessage, - parseProposedPlanMessage, proposedPlanTitle, } from "../proposedPlan"; import { truncateTitle } from "../truncateTitle"; @@ -334,6 +333,7 @@ function buildLocalDraftThread( worktreePath: draftThread.worktreePath, turnDiffSummaries: [], activities: [], + proposedPlans: [], }; } @@ -904,11 +904,11 @@ export default function ChatView({ threadId }: ChatViewProps) { if (!latestTurnSettled) { return null; } - return findLatestProposedPlanMessage( - activeThread?.messages ?? [], - activeLatestTurn?.assistantMessageId ?? null, + return findLatestProposedPlan( + activeThread?.proposedPlans ?? [], + activeLatestTurn?.turnId ?? null, ); - }, [activeLatestTurn?.assistantMessageId, activeThread?.messages, latestTurnSettled]); + }, [activeLatestTurn?.turnId, activeThread?.proposedPlans, latestTurnSettled]); const activePlan = useMemo( () => deriveActivePlanState(threadActivities, activeLatestTurn?.turnId ?? undefined), [activeLatestTurn?.turnId, threadActivities], @@ -1055,8 +1055,8 @@ export default function ChatView({ threadId }: ChatViewProps) { return [...serverMessagesWithPreviewHandoff, ...pendingMessages]; }, [serverMessages, attachmentPreviewHandoffByMessageId, optimisticUserMessages]); const timelineEntries = useMemo( - () => deriveTimelineEntries(timelineMessages, workLogEntries), - [timelineMessages, workLogEntries], + () => deriveTimelineEntries(timelineMessages, activeThread?.proposedPlans ?? [], workLogEntries), + [activeThread?.proposedPlans, timelineMessages, workLogEntries], ); const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = useTurnDiffSummaries(activeThread); @@ -2279,8 +2279,8 @@ export default function ChatView({ threadId }: ChatViewProps) { if (showPlanFollowUpPrompt && activeProposedPlan) { const followUpText = trimmed.length > 0 - ? trimmed - : buildPlanImplementationPrompt(activeProposedPlan.plan.planMarkdown); + ? trimmed + : buildPlanImplementationPrompt(activeProposedPlan.planMarkdown); const nextInteractionMode = trimmed.length > 0 ? "plan" : "default"; promptRef.current = ""; clearComposerDraftContent(activeThread.id); @@ -2815,7 +2815,7 @@ export default function ChatView({ threadId }: ChatViewProps) { const createdAt = new Date().toISOString(); const nextThreadId = newThreadId(); - const planMarkdown = activeProposedPlan.plan.planMarkdown; + const planMarkdown = activeProposedPlan.planMarkdown; const implementationPrompt = buildPlanImplementationPrompt(planMarkdown); const nextThreadTitle = truncateTitle(buildPlanImplementationThreadTitle(planMarkdown)); const nextThreadModel: ModelSlug = @@ -3334,7 +3334,7 @@ export default function ChatView({ threadId }: ChatViewProps) { > 0} + hasMessages={timelineEntries.length > 0} isWorking={isWorking} activeTurnInProgress={!latestTurnSettled} activeTurnStartedAt={activeLatestTurn?.startedAt ?? null} @@ -3387,8 +3387,8 @@ export default function ChatView({ threadId }: ChatViewProps) { ) : showPlanFollowUpPrompt && activeProposedPlan ? (
) : null} @@ -4205,7 +4205,6 @@ const ComposerPendingUserInputPanel = memo(function ComposerPendingUserInputPane ; questionIndex: number; @@ -4448,33 +4445,29 @@ const ChangedFilesTree = memo(function ChangedFilesTree(props: { return (
{treeNodes.map((node) => renderTreeNode(node, 0))} -const ProposedPlanMessage = memo(function ProposedPlanMessage({ - text, +
+ ); +}); + +const ProposedPlanCard = memo(function ProposedPlanCard({ + planMarkdown, cwd, workspaceRoot, - isStreaming, }: { - text: string; + planMarkdown: string; cwd: string | undefined; workspaceRoot: string | undefined; - isStreaming: boolean; }) { - const proposedPlan = parseProposedPlanMessage(text); const [expanded, setExpanded] = useState(false); const [isSaveDialogOpen, setIsSaveDialogOpen] = useState(false); const [savePath, setSavePath] = useState(""); const [isSavingToWorkspace, setIsSavingToWorkspace] = useState(false); const savePathInputId = useId(); - - if (!proposedPlan) { - return ; - } - - const title = proposedPlanTitle(proposedPlan.planMarkdown) ?? "Proposed plan"; - const lineCount = proposedPlan.planMarkdown.split("\n").length; - const canCollapse = proposedPlan.planMarkdown.length > 900 || lineCount > 20; - const downloadFilename = buildProposedPlanMarkdownFilename(proposedPlan.planMarkdown); - const saveContents = normalizePlanMarkdownForExport(proposedPlan.planMarkdown); + const title = proposedPlanTitle(planMarkdown) ?? "Proposed plan"; + const lineCount = planMarkdown.split("\n").length; + const canCollapse = planMarkdown.length > 900 || lineCount > 20; + const downloadFilename = buildProposedPlanMarkdownFilename(planMarkdown); + const saveContents = normalizePlanMarkdownForExport(planMarkdown); const handleDownload = () => { downloadTextFile(downloadFilename, saveContents); @@ -4540,46 +4533,40 @@ const ProposedPlanMessage = memo(function ProposedPlanMessage({ }; return ( -
- {proposedPlan.beforeText ? ( - - ) : null} - -
-
-
- Plan -

{title}

-
- - } - > - - - Download as markdown - - Save to workspace - - - +
+
+
+ Plan +

{title}

-
-
- - {canCollapse && !expanded ? ( -
- ) : null} -
- {canCollapse ? ( -
- -
+ + } + > + + + Download as markdown + + Save to workspace + + + +
+
+
+ + {canCollapse && !expanded ? ( +
) : null}
+ {canCollapse ? ( +
+ +
+ ) : null}
- - {proposedPlan.afterText ? ( - - ) : null}
); }); @@ -4662,6 +4645,7 @@ interface MessagesTimelineProps { type TimelineEntry = ReturnType[number]; type TimelineMessage = Extract["message"]; +type TimelineProposedPlan = Extract["proposedPlan"]; type TimelineWorkEntry = Extract["entry"]; type TimelineRow = | { @@ -4677,8 +4661,33 @@ type TimelineRow = message: TimelineMessage; showCompletionDivider: boolean; } + | { + kind: "proposed-plan"; + id: string; + createdAt: string; + proposedPlan: TimelineProposedPlan; + } | { kind: "working"; id: string; createdAt: string | null }; +function estimateTimelineMessageHeight(message: TimelineMessage): number { + const textLength = message.text.length; + if (message.role === "assistant") { + const estimatedLines = Math.max(1, Math.ceil(textLength / 72)); + return 78 + Math.min(estimatedLines * 22, 820); + } + + const estimatedLines = Math.max(1, Math.ceil(textLength / 56)); + const attachmentCount = message.attachments?.length ?? 0; + const attachmentRows = Math.ceil(attachmentCount / 2); + const attachmentHeight = attachmentRows * 124; + return 96 + Math.min(estimatedLines * 22, 620) + attachmentHeight; +} + +function estimateTimelineProposedPlanHeight(proposedPlan: TimelineProposedPlan): number { + const estimatedLines = Math.max(1, Math.ceil(proposedPlan.planMarkdown.length / 72)); + return 120 + Math.min(estimatedLines * 22, 880); +} + const MessagesTimeline = memo(function MessagesTimeline({ hasMessages, isWorking, @@ -4757,6 +4766,16 @@ const MessagesTimeline = memo(function MessagesTimeline({ continue; } + if (timelineEntry.kind === "proposed-plan") { + nextRows.push({ + kind: "proposed-plan", + id: timelineEntry.id, + createdAt: timelineEntry.createdAt, + proposedPlan: timelineEntry.proposedPlan, + }); + continue; + } + nextRows.push({ kind: "message", id: timelineEntry.id, @@ -4831,6 +4850,7 @@ const MessagesTimeline = memo(function MessagesTimeline({ const row = rows[index]; if (!row) return 96; if (row.kind === "work") return 112; + if (row.kind === "proposed-plan") return estimateTimelineProposedPlanHeight(row.proposedPlan); if (row.kind === "working") return 40; return estimateTimelineMessageHeight(row.message, { timelineWidthPx }); }, @@ -5044,10 +5064,9 @@ const MessagesTimeline = memo(function MessagesTimeline({
)}
- {(() => { @@ -5119,6 +5138,16 @@ const MessagesTimeline = memo(function MessagesTimeline({ ); })()} + {row.kind === "proposed-plan" && ( +
+ +
+ )} + {row.kind === "working" && (
diff --git a/apps/web/src/proposedPlan.test.ts b/apps/web/src/proposedPlan.test.ts index 7bdf17b2f06d6949ba911c16c09ee37e1463ccdf..02e4e78aed5763b1c7fa3b7144c08b85ef44651e 100644 --- a/apps/web/src/proposedPlan.test.ts +++ b/apps/web/src/proposedPlan.test.ts @@ -1,85 +1,12 @@ -import { MessageId } from "@t3tools/contracts"; import { describe, expect, it } from "vitest"; import { buildPlanImplementationThreadTitle, buildPlanImplementationPrompt, buildProposedPlanMarkdownFilename, - findLatestProposedPlanMessage, - parseProposedPlanMessage, proposedPlanTitle, } from "./proposedPlan"; -describe("parseProposedPlanMessage", () => { - it("extracts the proposed plan block and surrounding assistant text", () => { - expect( - parseProposedPlanMessage( - "First I checked the repo.\n\n\n# Ship plan\n\n- step 1\n\n\nImplement this plan?", - ), - ).toEqual({ - beforeText: "First I checked the repo.", - planMarkdown: "# Ship plan\n\n- step 1", - afterText: "Implement this plan?", - }); - }); - - it("returns null when the assistant message has no plan block", () => { - expect(parseProposedPlanMessage("Plain assistant text")).toBeNull(); - }); -}); - -describe("findLatestProposedPlanMessage", () => { - it("prefers the latest turn assistant message id when it contains a proposed plan", () => { - const match = findLatestProposedPlanMessage( - [ - { - id: MessageId.makeUnsafe("assistant:older-plan"), - role: "assistant", - text: "\n# Older plan\n", - createdAt: "2026-03-05T00:00:00.000Z", - streaming: false, - }, - { - id: MessageId.makeUnsafe("assistant:latest-plan"), - role: "assistant", - text: "\n# Latest plan\n", - createdAt: "2026-03-05T00:01:00.000Z", - streaming: false, - }, - ], - "assistant:latest-plan", - ); - - expect(match?.message.id).toBe("assistant:latest-plan"); - expect(match?.plan.planMarkdown).toBe("# Latest plan"); - }); - - it("falls back to the latest assistant proposed plan when no latest-turn message id is available", () => { - const match = findLatestProposedPlanMessage( - [ - { - id: MessageId.makeUnsafe("assistant:plain"), - role: "assistant", - text: "No plan here", - createdAt: "2026-03-05T00:00:00.000Z", - streaming: false, - }, - { - id: MessageId.makeUnsafe("assistant:plan"), - role: "assistant", - text: "\n# Fallback plan\n", - createdAt: "2026-03-05T00:01:00.000Z", - streaming: false, - }, - ], - null, - ); - - expect(match?.message.id).toBe("assistant:plan"); - expect(match?.plan.planMarkdown).toBe("# Fallback plan"); - }); -}); - describe("proposedPlanTitle", () => { it("reads the first markdown heading as the plan title", () => { expect(proposedPlanTitle("# Integrate RPC\n\nBody")).toBe("Integrate RPC"); diff --git a/apps/web/src/proposedPlan.ts b/apps/web/src/proposedPlan.ts index b34ab7523431499419f005e4c8cec40f19c2baf8..fdfe859726b95051f41a278093ecfa305345ebd1 100644 --- a/apps/web/src/proposedPlan.ts +++ b/apps/web/src/proposedPlan.ts @@ -1,42 +1,3 @@ -import type { MessageId } from "@t3tools/contracts"; - -import type { ChatMessage } from "./types"; - -export interface ProposedPlanBlock { - beforeText: string; - planMarkdown: string; - afterText: string; -} - -export interface ProposedPlanMessageMatch { - message: ChatMessage; - plan: ProposedPlanBlock; -} - -const PROPOSED_PLAN_REGEX = /\s*([\s\S]*?)\s*<\/proposed_plan>/; - -export function parseProposedPlanMessage(text: string): ProposedPlanBlock | null { - const match = PROPOSED_PLAN_REGEX.exec(text); - if (!match) { - return null; - } - - const fullMatch = match[0]; - const content = match[1]; - const start = match.index; - const end = start + fullMatch.length; - const planMarkdown = content?.trim(); - if (!planMarkdown) { - return null; - } - - return { - beforeText: text.slice(0, start).trim(), - planMarkdown, - afterText: text.slice(end).trim(), - }; -} - export function proposedPlanTitle(planMarkdown: string): string | null { const heading = planMarkdown.match(/^\s{0,3}#{1,6}\s+(.+)$/m)?.[1]?.trim(); return heading && heading.length > 0 ? heading : null; @@ -51,43 +12,6 @@ function sanitizePlanFileSegment(input: string): string { return sanitized.length > 0 ? sanitized : "plan"; } -export function findLatestProposedPlanMessage( - messages: readonly ChatMessage[], - assistantMessageId: MessageId | string | null | undefined, -): ProposedPlanMessageMatch | null { - if (assistantMessageId) { - const matchingMessage = messages.find( - (message) => message.id === assistantMessageId && message.role === "assistant", - ); - if (matchingMessage) { - const parsedPlan = parseProposedPlanMessage(matchingMessage.text); - if (parsedPlan) { - return { - message: matchingMessage, - plan: parsedPlan, - }; - } - } - } - - for (let index = messages.length - 1; index >= 0; index -= 1) { - const message = messages[index]; - if (!message || message.role !== "assistant") { - continue; - } - const parsedPlan = parseProposedPlanMessage(message.text); - if (!parsedPlan) { - continue; - } - return { - message, - plan: parsedPlan, - }; - } - - return null; -} - export function buildPlanImplementationPrompt(planMarkdown: string): string { return `PLEASE IMPLEMENT THIS PLAN:\n${planMarkdown.trim()}`; } diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index bb8625942c64d8342873b050572130193f6ebc4a..eebf1170493c1076b124673bc755b1ad800402fe 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -1,4 +1,4 @@ -import { EventId, TurnId, type OrchestrationThreadActivity } from "@t3tools/contracts"; +import { EventId, MessageId, TurnId, type OrchestrationThreadActivity } from "@t3tools/contracts"; import { describe, expect, it } from "vitest"; import { @@ -6,7 +6,9 @@ import { PROVIDER_OPTIONS, derivePendingApprovals, derivePendingUserInputs, + deriveTimelineEntries, deriveWorkLogEntries, + findLatestProposedPlan, hasToolActivityForTurn, isLatestTurnSettled, } from "./session-logic"; @@ -257,6 +259,69 @@ describe("deriveActivePlanState", () => { }); }); +describe("findLatestProposedPlan", () => { + it("prefers the latest proposed plan for the active turn", () => { + expect( + findLatestProposedPlan( + [ + { + id: "plan:thread-1:turn:turn-1", + turnId: TurnId.makeUnsafe("turn-1"), + planMarkdown: "# Older", + createdAt: "2026-02-23T00:00:01.000Z", + updatedAt: "2026-02-23T00:00:01.000Z", + }, + { + id: "plan:thread-1:turn:turn-1", + turnId: TurnId.makeUnsafe("turn-1"), + planMarkdown: "# Latest", + createdAt: "2026-02-23T00:00:01.000Z", + updatedAt: "2026-02-23T00:00:02.000Z", + }, + { + id: "plan:thread-1:turn:turn-2", + turnId: TurnId.makeUnsafe("turn-2"), + planMarkdown: "# Different turn", + createdAt: "2026-02-23T00:00:03.000Z", + updatedAt: "2026-02-23T00:00:03.000Z", + }, + ], + TurnId.makeUnsafe("turn-1"), + ), + ).toEqual({ + id: "plan:thread-1:turn:turn-1", + turnId: "turn-1", + planMarkdown: "# Latest", + createdAt: "2026-02-23T00:00:01.000Z", + updatedAt: "2026-02-23T00:00:02.000Z", + }); + }); + + it("falls back to the most recently updated proposed plan", () => { + const latestPlan = findLatestProposedPlan( + [ + { + id: "plan:thread-1:turn:turn-1", + turnId: TurnId.makeUnsafe("turn-1"), + planMarkdown: "# First", + createdAt: "2026-02-23T00:00:01.000Z", + updatedAt: "2026-02-23T00:00:01.000Z", + }, + { + id: "plan:thread-1:turn:turn-2", + turnId: TurnId.makeUnsafe("turn-2"), + planMarkdown: "# Latest", + createdAt: "2026-02-23T00:00:02.000Z", + updatedAt: "2026-02-23T00:00:03.000Z", + }, + ], + null, + ); + + expect(latestPlan?.planMarkdown).toBe("# Latest"); + }); +}); + describe("deriveWorkLogEntries", () => { it("omits tool started entries and keeps completed entries", () => { const activities: OrchestrationThreadActivity[] = [ @@ -338,6 +403,47 @@ describe("deriveWorkLogEntries", () => { }); }); +describe("deriveTimelineEntries", () => { + it("includes proposed plans alongside messages and work entries in chronological order", () => { + const entries = deriveTimelineEntries( + [ + { + id: MessageId.makeUnsafe("message-1"), + role: "assistant", + text: "hello", + createdAt: "2026-02-23T00:00:01.000Z", + streaming: false, + }, + ], + [ + { + id: "plan:thread-1:turn:turn-1", + turnId: TurnId.makeUnsafe("turn-1"), + planMarkdown: "# Ship it", + createdAt: "2026-02-23T00:00:02.000Z", + updatedAt: "2026-02-23T00:00:02.000Z", + }, + ], + [ + { + id: "work-1", + createdAt: "2026-02-23T00:00:03.000Z", + label: "Ran tests", + tone: "tool", + }, + ], + ); + + expect(entries.map((entry) => entry.kind)).toEqual(["message", "proposed-plan", "work"]); + expect(entries[1]).toMatchObject({ + kind: "proposed-plan", + proposedPlan: { + planMarkdown: "# Ship it", + }, + }); + }); +}); + describe("hasToolActivityForTurn", () => { it("returns false when turn id is missing", () => { const activities: OrchestrationThreadActivity[] = [ diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 61a3b11bcf4a255e38a491374582da69cf6c9b57..2ded1b984b15bdbc8133084ecf69dd50461e84f8 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -2,12 +2,13 @@ import { ApprovalRequestId, type OrchestrationLatestTurn, type OrchestrationThreadActivity, + type OrchestrationProposedPlanId, type ProviderKind, type UserInputQuestion, type TurnId, } from "@t3tools/contracts"; -import type { ChatMessage, SessionPhase, ThreadSession, TurnDiffSummary } from "./types"; +import type { ChatMessage, ProposedPlan, SessionPhase, ThreadSession, TurnDiffSummary } from "./types"; export const PROVIDER_OPTIONS: Array<{ value: ProviderKind; @@ -50,6 +51,14 @@ export interface ActivePlanState { }>; } +export interface LatestProposedPlanState { + id: OrchestrationProposedPlanId; + createdAt: string; + updatedAt: string; + turnId: TurnId | null; + planMarkdown: string; +} + export type TimelineEntry = | { id: string; @@ -57,6 +66,12 @@ export type TimelineEntry = createdAt: string; message: ChatMessage; } + | { + id: string; + kind: "proposed-plan"; + createdAt: string; + proposedPlan: ProposedPlan; + } | { id: string; kind: "work"; @@ -327,6 +342,48 @@ export function deriveActivePlanState( }; } +export function findLatestProposedPlan( + proposedPlans: ReadonlyArray, + latestTurnId: TurnId | string | null | undefined, +): LatestProposedPlanState | null { + if (latestTurnId) { + const matchingTurnPlan = [...proposedPlans] + .filter((proposedPlan) => proposedPlan.turnId === latestTurnId) + .toSorted( + (left, right) => + left.updatedAt.localeCompare(right.updatedAt) || left.id.localeCompare(right.id), + ) + .at(-1); + if (matchingTurnPlan) { + return { + id: matchingTurnPlan.id, + createdAt: matchingTurnPlan.createdAt, + updatedAt: matchingTurnPlan.updatedAt, + turnId: matchingTurnPlan.turnId, + planMarkdown: matchingTurnPlan.planMarkdown, + }; + } + } + + const latestPlan = [...proposedPlans] + .toSorted( + (left, right) => + left.updatedAt.localeCompare(right.updatedAt) || left.id.localeCompare(right.id), + ) + .at(-1); + if (!latestPlan) { + return null; + } + + return { + id: latestPlan.id, + createdAt: latestPlan.createdAt, + updatedAt: latestPlan.updatedAt, + turnId: latestPlan.turnId, + planMarkdown: latestPlan.planMarkdown, + }; +} + export function deriveWorkLogEntries( activities: ReadonlyArray, latestTurnId: TurnId | undefined, @@ -381,6 +438,7 @@ export function hasToolActivityForTurn( export function deriveTimelineEntries( messages: ChatMessage[], + proposedPlans: ProposedPlan[], workEntries: WorkLogEntry[], ): TimelineEntry[] { const messageRows: TimelineEntry[] = messages.map((message) => ({ @@ -389,13 +447,21 @@ export function deriveTimelineEntries( createdAt: message.createdAt, message, })); + const proposedPlanRows: TimelineEntry[] = proposedPlans.map((proposedPlan) => ({ + id: proposedPlan.id, + kind: "proposed-plan", + createdAt: proposedPlan.createdAt, + proposedPlan, + })); const workRows: TimelineEntry[] = workEntries.map((entry) => ({ id: entry.id, kind: "work", createdAt: entry.createdAt, entry, })); - return [...messageRows, ...workRows].toSorted((a, b) => a.createdAt.localeCompare(b.createdAt)); + return [...messageRows, ...proposedPlanRows, ...workRows].toSorted((a, b) => + a.createdAt.localeCompare(b.createdAt), + ); } export function inferCheckpointTurnCountByTurnId( diff --git a/apps/web/src/store.test.ts b/apps/web/src/store.test.ts index 03e2874821d76d8c56f7635b0d8294ff9355280a..1b279eea5720ee911366b74f1156a08d7d517870 100644 --- a/apps/web/src/store.test.ts +++ b/apps/web/src/store.test.ts @@ -17,6 +17,7 @@ function makeThread(overrides: Partial = {}): Thread { messages: [], turnDiffSummaries: [], activities: [], + proposedPlans: [], error: null, createdAt: "2026-02-13T00:00:00.000Z", latestTurn: null, @@ -59,6 +60,7 @@ function makeReadModelThread(overrides: Partial ({ + id: proposedPlan.id, + turnId: proposedPlan.turnId, + planMarkdown: proposedPlan.planMarkdown, + createdAt: proposedPlan.createdAt, + updatedAt: proposedPlan.updatedAt, + })), error: thread.session?.lastError ?? null, createdAt: thread.createdAt, latestTurn: thread.latestTurn, diff --git a/apps/web/src/types.ts b/apps/web/src/types.ts index 0e9ded5a1aba19ca4f907739ac09a9a132660179..82e7dd34d77b7231337b7bab81a0e1c1b31e9499 100644 --- a/apps/web/src/types.ts +++ b/apps/web/src/types.ts @@ -1,5 +1,6 @@ import type { OrchestrationLatestTurn, + OrchestrationProposedPlanId, OrchestrationSessionStatus, OrchestrationThreadActivity, ProjectScript as ContractProjectScript, @@ -48,6 +49,14 @@ export interface ChatMessage { streaming: boolean; } +export interface ProposedPlan { + id: OrchestrationProposedPlanId; + turnId: TurnId | null; + planMarkdown: string; + createdAt: string; + updatedAt: string; +} + export interface TurnDiffFileChange { path: string; kind?: string | undefined; @@ -84,6 +93,7 @@ export interface Thread { interactionMode: InteractionMode; session: ThreadSession | null; messages: ChatMessage[]; + proposedPlans: ProposedPlan[]; error: string | null; createdAt: string; latestTurn: OrchestrationLatestTurn | null; diff --git a/apps/web/src/worktreeCleanup.test.ts b/apps/web/src/worktreeCleanup.test.ts index bdbcd94d3e8cd6f596bea43e5025a6ad5a05e3a9..516df6046a503ada0ca3663963b84e33b62d5208 100644 --- a/apps/web/src/worktreeCleanup.test.ts +++ b/apps/web/src/worktreeCleanup.test.ts @@ -17,6 +17,7 @@ function makeThread(overrides: Partial = {}): Thread { messages: [], turnDiffSummaries: [], activities: [], + proposedPlans: [], error: null, createdAt: "2026-02-13T00:00:00.000Z", latestTurn: null, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 8600f5b91dd0e56ac19856c3375a2bb9ce3e9263..57ac2e6f10844081777dd3db644ee0d68db9b920 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -136,6 +136,18 @@ export const OrchestrationMessage = Schema.Struct({ }); export type OrchestrationMessage = typeof OrchestrationMessage.Type; +export const OrchestrationProposedPlanId = TrimmedNonEmptyString; +export type OrchestrationProposedPlanId = typeof OrchestrationProposedPlanId.Type; + +export const OrchestrationProposedPlan = Schema.Struct({ + id: OrchestrationProposedPlanId, + turnId: Schema.NullOr(TurnId), + planMarkdown: TrimmedNonEmptyString, + createdAt: IsoDateTime, + updatedAt: IsoDateTime, +}); +export type OrchestrationProposedPlan = typeof OrchestrationProposedPlan.Type; + export const OrchestrationSessionStatus = Schema.Literals([ "idle", "starting", @@ -234,6 +246,9 @@ export const OrchestrationThread = Schema.Struct({ updatedAt: IsoDateTime, deletedAt: Schema.NullOr(IsoDateTime), messages: Schema.Array(OrchestrationMessage), + proposedPlans: Schema.Array(OrchestrationProposedPlan).pipe( + Schema.withDecodingDefault(() => []), + ), activities: Schema.Array(OrchestrationThreadActivity), checkpoints: Schema.Array(OrchestrationCheckpointSummary), session: Schema.NullOr(OrchestrationSession), @@ -467,6 +482,14 @@ export const ThreadMessageAssistantCompleteCommand = Schema.Struct({ createdAt: IsoDateTime, }); +export const ThreadProposedPlanUpsertCommand = Schema.Struct({ + type: Schema.Literal("thread.proposed-plan.upsert"), + commandId: CommandId, + threadId: ThreadId, + proposedPlan: OrchestrationProposedPlan, + createdAt: IsoDateTime, +}); + export const ThreadTurnDiffCompleteCommand = Schema.Struct({ type: Schema.Literal("thread.turn.diff.complete"), commandId: CommandId, @@ -501,6 +524,7 @@ export const InternalOrchestrationCommand = Schema.Union([ ThreadSessionSetCommand, ThreadMessageAssistantDeltaCommand, ThreadMessageAssistantCompleteCommand, + ThreadProposedPlanUpsertCommand, ThreadTurnDiffCompleteCommand, ThreadActivityAppendCommand, ThreadRevertCompleteCommand, @@ -531,6 +555,7 @@ export const OrchestrationEventType = Schema.Literals([ "thread.reverted", "thread.session-stop-requested", "thread.session-set", + "thread.proposed-plan-upserted", "thread.turn-diff-completed", "thread.activity-appended", ]); @@ -674,6 +699,11 @@ export const ThreadSessionSetPayload = Schema.Struct({ session: OrchestrationSession, }); +export const ThreadProposedPlanUpsertedPayload = Schema.Struct({ + threadId: ThreadId, + proposedPlan: OrchestrationProposedPlan, +}); + export const ThreadTurnDiffCompletedPayload = Schema.Struct({ threadId: ThreadId, turnId: TurnId, @@ -811,6 +841,11 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.session-set"), payload: ThreadSessionSetPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.proposed-plan-upserted"), + payload: ThreadProposedPlanUpsertedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.turn-diff-completed"), @@ -910,6 +945,11 @@ export const OrchestrationPersistedEvent = Schema.Union([ eventType: Schema.Literal("thread.session-set"), payload: ThreadSessionSetPayload, }), + Schema.Struct({ + ...PersistedEventBaseFields, + eventType: Schema.Literal("thread.proposed-plan-upserted"), + payload: ThreadProposedPlanUpsertedPayload, + }), Schema.Struct({ ...PersistedEventBaseFields, eventType: Schema.Literal("thread.turn-diff-completed"), diff --git a/packages/contracts/src/providerRuntime.test.ts b/packages/contracts/src/providerRuntime.test.ts index b5c1821974fbe1b78b2beb9cd2c76809e9c8f372..7f578c276eacf7392e4d0718c3e055c93e775d1f 100644 --- a/packages/contracts/src/providerRuntime.test.ts +++ b/packages/contracts/src/providerRuntime.test.ts @@ -32,6 +32,26 @@ describe("ProviderRuntimeEvent", () => { expect(parsed.payload.plan[1]?.status).toBe("inProgress"); }); + it("decodes proposed-plan completion events", () => { + const parsed = decodeRuntimeEvent({ + type: "turn.proposed.completed", + eventId: "event-proposed-plan-1", + provider: "codex", + createdAt: "2026-02-28T00:00:00.000Z", + threadId: "thread-1", + turnId: "turn-1", + payload: { + planMarkdown: "# Ship it", + }, + }); + + expect(parsed.type).toBe("turn.proposed.completed"); + if (parsed.type !== "turn.proposed.completed") { + throw new Error("expected turn.proposed.completed"); + } + expect(parsed.payload.planMarkdown).toBe("# Ship it"); + }); + it("decodes user-input.requested with structured questions", () => { const parsed = decodeRuntimeEvent({ type: "user-input.requested", @@ -39,6 +59,7 @@ describe("ProviderRuntimeEvent", () => { provider: "claudeCode", sessionId: "runtime-session-2", createdAt: "2026-02-28T00:00:01.000Z", + threadId: "thread-2", requestId: "request-1", payload: { questions: [ @@ -76,6 +97,7 @@ describe("ProviderRuntimeEvent", () => { provider: "claudeCode", sessionId: "runtime-session-2", createdAt: "2026-02-28T00:00:02.000Z", + threadId: "thread-2", requestId: "request-1", payload: { answers: { diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index c7ce78c5046f6e833b8e013315522d57b32f2af1..4ef6e62cf053b5ccbff58d03bc3deee52a3a6696 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -154,6 +154,8 @@ export const ProviderRuntimeEventType = Schema.Literals([ "turn.completed", "turn.aborted", "turn.plan.updated", + "turn.proposed.delta", + "turn.proposed.completed", "turn.diff.updated", "item.started", "item.updated", @@ -202,6 +204,8 @@ export const TurnStartedType = Schema.Literal("turn.started"); export const TurnCompletedType = Schema.Literal("turn.completed"); export const TurnAbortedType = Schema.Literal("turn.aborted"); export const TurnPlanUpdatedType = Schema.Literal("turn.plan.updated"); +export const TurnProposedDeltaType = Schema.Literal("turn.proposed.delta"); +export const TurnProposedCompletedType = Schema.Literal("turn.proposed.completed"); export const TurnDiffUpdatedType = Schema.Literal("turn.diff.updated"); export const ItemStartedType = Schema.Literal("item.started"); export const ItemUpdatedType = Schema.Literal("item.updated"); @@ -349,6 +353,16 @@ export const TurnPlanUpdatedPayload = Schema.Struct({ }); export type TurnPlanUpdatedPayload = typeof TurnPlanUpdatedPayload.Type; +export const TurnProposedDeltaPayload = Schema.Struct({ + delta: Schema.String, +}); +export type TurnProposedDeltaPayload = typeof TurnProposedDeltaPayload.Type; + +export const TurnProposedCompletedPayload = Schema.Struct({ + planMarkdown: TrimmedNonEmptyStringSchema, +}); +export type TurnProposedCompletedPayload = typeof TurnProposedCompletedPayload.Type; + export const TurnDiffUpdatedPayload = Schema.Struct({ unifiedDiff: Schema.String, }); @@ -677,6 +691,22 @@ export const ProviderRuntimeTurnPlanUpdatedEvent = Schema.Struct({ }); export type ProviderRuntimeTurnPlanUpdatedEvent = typeof ProviderRuntimeTurnPlanUpdatedEvent.Type; +export const ProviderRuntimeTurnProposedDeltaEvent = Schema.Struct({ + ...ProviderRuntimeEventBase.fields, + type: TurnProposedDeltaType, + payload: TurnProposedDeltaPayload, +}); +export type ProviderRuntimeTurnProposedDeltaEvent = + typeof ProviderRuntimeTurnProposedDeltaEvent.Type; + +export const ProviderRuntimeTurnProposedCompletedEvent = Schema.Struct({ + ...ProviderRuntimeEventBase.fields, + type: TurnProposedCompletedType, + payload: TurnProposedCompletedPayload, +}); +export type ProviderRuntimeTurnProposedCompletedEvent = + typeof ProviderRuntimeTurnProposedCompletedEvent.Type; + export const ProviderRuntimeTurnDiffUpdatedEvent = Schema.Struct({ ...ProviderRuntimeEventBase.fields, type: TurnDiffUpdatedType, @@ -893,6 +923,8 @@ export const ProviderRuntimeEventV2 = Schema.Union([ ProviderRuntimeTurnCompletedEvent, ProviderRuntimeTurnAbortedEvent, ProviderRuntimeTurnPlanUpdatedEvent, + ProviderRuntimeTurnProposedDeltaEvent, + ProviderRuntimeTurnProposedCompletedEvent, ProviderRuntimeTurnDiffUpdatedEvent, ProviderRuntimeItemStartedEvent, ProviderRuntimeItemUpdatedEvent, -- 2.54.0