authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-05 15:34:51-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-05 23:00:41-08:00
log640a2cca9f9effc434d1296d7b625ee09bcba64e
tree632727f823634dbb3708fecf821b34b242bf7ee8
parent12edc3455cf675d2e8000f508b38b79bc7743a88

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

28 files changed, 1113 insertions(+), 323 deletions(-)

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts+1
...@@ -61,6 +61,7 @@ function makeSnapshot(input: {...@@ -61,6 +61,7 @@ function makeSnapshot(input: {
61 deletedAt: null,61 deletedAt: null,
62 messages: [],62 messages: [],
63 activities: [],63 activities: [],
64 proposedPlans: [],
64 checkpoints: [65 checkpoints: [
65 {66 {
66 turnId: TurnId.makeUnsafe("turn-1"),67 turnId: TurnId.makeUnsafe("turn-1"),
apps/server/src/orchestration/Layers/ProjectionPipeline.ts+84
...@@ -18,6 +18,10 @@ import {...@@ -18,6 +18,10 @@ import {
18 type ProjectionThreadMessage,18 type ProjectionThreadMessage,
19 ProjectionThreadMessageRepository,19 ProjectionThreadMessageRepository,
20} from "../../persistence/Services/ProjectionThreadMessages.ts";20} from "../../persistence/Services/ProjectionThreadMessages.ts";
21import {
22 type ProjectionThreadProposedPlan,
23 ProjectionThreadProposedPlanRepository,
24} from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
21import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts";25import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts";
22import {26import {
23 type ProjectionTurn,27 type ProjectionTurn,
...@@ -29,6 +33,7 @@ import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/Projec...@@ -29,6 +33,7 @@ import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/Projec
29import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts";33import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts";
30import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";34import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";
31import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts";35import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts";
36import { ProjectionThreadProposedPlanRepositoryLive } from "../../persistence/Layers/ProjectionThreadProposedPlans.ts";
32import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts";37import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts";
33import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";38import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";
34import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts";39import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts";
...@@ -48,6 +53,7 @@ export const ORCHESTRATION_PROJECTOR_NAMES = {...@@ -48,6 +53,7 @@ export const ORCHESTRATION_PROJECTOR_NAMES = {
48 projects: "projection.projects",53 projects: "projection.projects",
49 threads: "projection.threads",54 threads: "projection.threads",
50 threadMessages: "projection.thread-messages",55 threadMessages: "projection.thread-messages",
56 threadProposedPlans: "projection.thread-proposed-plans",
51 threadActivities: "projection.thread-activities",57 threadActivities: "projection.thread-activities",
52 threadSessions: "projection.thread-sessions",58 threadSessions: "projection.thread-sessions",
53 threadTurns: "projection.thread-turns",59 threadTurns: "projection.thread-turns",
...@@ -189,6 +195,26 @@ function retainProjectionActivitiesAfterRevert(...@@ -189,6 +195,26 @@ function retainProjectionActivitiesAfterRevert(
189 );195 );
190}196}
191197
198function retainProjectionProposedPlansAfterRevert(
199 proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>,
200 turns: ReadonlyArray<ProjectionTurn>,
201 turnCount: number,
202): ReadonlyArray<ProjectionThreadProposedPlan> {
203 const retainedTurnIds = new Set<string>(
204 turns
205 .filter(
206 (turn) =>
207 turn.turnId !== null &&
208 turn.checkpointTurnCount !== null &&
209 turn.checkpointTurnCount <= turnCount,
210 )
211 .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])),
212 );
213 return proposedPlans.filter(
214 (proposedPlan) => proposedPlan.turnId === null || retainedTurnIds.has(proposedPlan.turnId),
215 );
216}
217
192function collectThreadAttachmentRelativePaths(218function collectThreadAttachmentRelativePaths(
193 threadId: string,219 threadId: string,
194 messages: ReadonlyArray<ProjectionThreadMessage>,220 messages: ReadonlyArray<ProjectionThreadMessage>,
...@@ -320,6 +346,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -320,6 +346,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
320 const projectionProjectRepository = yield* ProjectionProjectRepository;346 const projectionProjectRepository = yield* ProjectionProjectRepository;
321 const projectionThreadRepository = yield* ProjectionThreadRepository;347 const projectionThreadRepository = yield* ProjectionThreadRepository;
322 const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository;348 const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository;
349 const projectionThreadProposedPlanRepository = yield* ProjectionThreadProposedPlanRepository;
323 const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;350 const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;
324 const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository;351 const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository;
325 const projectionTurnRepository = yield* ProjectionTurnRepository;352 const projectionTurnRepository = yield* ProjectionTurnRepository;
...@@ -474,6 +501,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -474,6 +501,7 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
474 }501 }
475502
476 case "thread.message-sent":503 case "thread.message-sent":
504 case "thread.proposed-plan-upserted":
477 case "thread.activity-appended": {505 case "thread.activity-appended": {
478 const existingRow = yield* projectionThreadRepository.getById({506 const existingRow = yield* projectionThreadRepository.getById({
479 threadId: event.payload.threadId,507 threadId: event.payload.threadId,
...@@ -615,6 +643,57 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -615,6 +643,57 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
615 }643 }
616 });644 });
617645
646 const applyThreadProposedPlansProjection: ProjectorDefinition["apply"] = (
647 event,
648 _attachmentSideEffects,
649 ) =>
650 Effect.gen(function* () {
651 switch (event.type) {
652 case "thread.proposed-plan-upserted":
653 yield* projectionThreadProposedPlanRepository.upsert({
654 planId: event.payload.proposedPlan.id,
655 threadId: event.payload.threadId,
656 turnId: event.payload.proposedPlan.turnId,
657 planMarkdown: event.payload.proposedPlan.planMarkdown,
658 createdAt: event.payload.proposedPlan.createdAt,
659 updatedAt: event.payload.proposedPlan.updatedAt,
660 });
661 return;
662
663 case "thread.reverted": {
664 const existingRows = yield* projectionThreadProposedPlanRepository.listByThreadId({
665 threadId: event.payload.threadId,
666 });
667 if (existingRows.length === 0) {
668 return;
669 }
670
671 const existingTurns = yield* projectionTurnRepository.listByThreadId({
672 threadId: event.payload.threadId,
673 });
674 const keptRows = retainProjectionProposedPlansAfterRevert(
675 existingRows,
676 existingTurns,
677 event.payload.turnCount,
678 );
679 if (keptRows.length === existingRows.length) {
680 return;
681 }
682
683 yield* projectionThreadProposedPlanRepository.deleteByThreadId({
684 threadId: event.payload.threadId,
685 });
686 yield* Effect.forEach(keptRows, projectionThreadProposedPlanRepository.upsert, {
687 concurrency: 1,
688 }).pipe(Effect.asVoid);
689 return;
690 }
691
692 default:
693 return;
694 }
695 });
696
618 const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = (697 const applyThreadActivitiesProjection: ProjectorDefinition["apply"] = (
619 event,698 event,
620 _attachmentSideEffects,699 _attachmentSideEffects,
...@@ -1023,6 +1102,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {...@@ -1023,6 +1102,10 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
1023 name: ORCHESTRATION_PROJECTOR_NAMES.threadMessages,1102 name: ORCHESTRATION_PROJECTOR_NAMES.threadMessages,
1024 apply: applyThreadMessagesProjection,1103 apply: applyThreadMessagesProjection,
1025 },1104 },
1105 {
1106 name: ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans,
1107 apply: applyThreadProposedPlansProjection,
1108 },
1026 {1109 {
1027 name: ORCHESTRATION_PROJECTOR_NAMES.threadActivities,1110 name: ORCHESTRATION_PROJECTOR_NAMES.threadActivities,
1028 apply: applyThreadActivitiesProjection,1111 apply: applyThreadActivitiesProjection,
...@@ -1142,6 +1225,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect(...@@ -1142,6 +1225,7 @@ export const OrchestrationProjectionPipelineLive = Layer.effect(
1142 Layer.provideMerge(ProjectionProjectRepositoryLive),1225 Layer.provideMerge(ProjectionProjectRepositoryLive),
1143 Layer.provideMerge(ProjectionThreadRepositoryLive),1226 Layer.provideMerge(ProjectionThreadRepositoryLive),
1144 Layer.provideMerge(ProjectionThreadMessageRepositoryLive),1227 Layer.provideMerge(ProjectionThreadMessageRepositoryLive),
1228 Layer.provideMerge(ProjectionThreadProposedPlanRepositoryLive),
1145 Layer.provideMerge(ProjectionThreadActivityRepositoryLive),1229 Layer.provideMerge(ProjectionThreadActivityRepositoryLive),
1146 Layer.provideMerge(ProjectionThreadSessionRepositoryLive),1230 Layer.provideMerge(ProjectionThreadSessionRepositoryLive),
1147 Layer.provideMerge(ProjectionTurnRepositoryLive),1231 Layer.provideMerge(ProjectionTurnRepositoryLive),
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+1
...@@ -260,6 +260,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {...@@ -260,6 +260,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
260 updatedAt: "2026-02-24T00:00:05.000Z",260 updatedAt: "2026-02-24T00:00:05.000Z",
261 },261 },
262 ],262 ],
263 proposedPlans: [],
263 activities: [264 activities: [
264 {265 {
265 id: asEventId("activity-1"),266 id: asEventId("activity-1"),
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+45
...@@ -10,6 +10,7 @@ import {...@@ -10,6 +10,7 @@ import {
10 type OrchestrationCheckpointSummary,10 type OrchestrationCheckpointSummary,
11 type OrchestrationLatestTurn,11 type OrchestrationLatestTurn,
12 type OrchestrationMessage,12 type OrchestrationMessage,
13 type OrchestrationProposedPlan,
13 type OrchestrationProject,14 type OrchestrationProject,
14 type OrchestrationSession,15 type OrchestrationSession,
15 type OrchestrationThread,16 type OrchestrationThread,
...@@ -30,6 +31,7 @@ import { ProjectionProject } from "../../persistence/Services/ProjectionProjects...@@ -30,6 +31,7 @@ import { ProjectionProject } from "../../persistence/Services/ProjectionProjects
30import { ProjectionState } from "../../persistence/Services/ProjectionState.ts";31import { ProjectionState } from "../../persistence/Services/ProjectionState.ts";
31import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";32import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";
32import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionThreadMessages.ts";33import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionThreadMessages.ts";
34import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
33import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";35import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";
34import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";36import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";
35import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";37import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
...@@ -50,6 +52,7 @@ const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields(...@@ -50,6 +52,7 @@ const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields(
50 attachments: Schema.NullOr(Schema.fromJsonString(Schema.Array(ChatAttachment))),52 attachments: Schema.NullOr(Schema.fromJsonString(Schema.Array(ChatAttachment))),
51 }),53 }),
52);54);
55const ProjectionThreadProposedPlanDbRowSchema = ProjectionThreadProposedPlan;
53const ProjectionThreadDbRowSchema = ProjectionThread;56const ProjectionThreadDbRowSchema = ProjectionThread;
54const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(57const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
55 Struct.assign({58 Struct.assign({
...@@ -78,6 +81,7 @@ const REQUIRED_SNAPSHOT_PROJECTORS = [...@@ -78,6 +81,7 @@ const REQUIRED_SNAPSHOT_PROJECTORS = [
78 ORCHESTRATION_PROJECTOR_NAMES.projects,81 ORCHESTRATION_PROJECTOR_NAMES.projects,
79 ORCHESTRATION_PROJECTOR_NAMES.threads,82 ORCHESTRATION_PROJECTOR_NAMES.threads,
80 ORCHESTRATION_PROJECTOR_NAMES.threadMessages,83 ORCHESTRATION_PROJECTOR_NAMES.threadMessages,
84 ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans,
81 ORCHESTRATION_PROJECTOR_NAMES.threadActivities,85 ORCHESTRATION_PROJECTOR_NAMES.threadActivities,
82 ORCHESTRATION_PROJECTOR_NAMES.threadSessions,86 ORCHESTRATION_PROJECTOR_NAMES.threadSessions,
83 ORCHESTRATION_PROJECTOR_NAMES.checkpoints,87 ORCHESTRATION_PROJECTOR_NAMES.checkpoints,
...@@ -186,6 +190,23 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -186,6 +190,23 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
186 `,190 `,
187 });191 });
188192
193 const listThreadProposedPlanRows = SqlSchema.findAll({
194 Request: Schema.Void,
195 Result: ProjectionThreadProposedPlanDbRowSchema,
196 execute: () =>
197 sql`
198 SELECT
199 plan_id AS "planId",
200 thread_id AS "threadId",
201 turn_id AS "turnId",
202 plan_markdown AS "planMarkdown",
203 created_at AS "createdAt",
204 updated_at AS "updatedAt"
205 FROM projection_thread_proposed_plans
206 ORDER BY thread_id ASC, created_at ASC, plan_id ASC
207 `,
208 });
209
189 const listThreadActivityRows = SqlSchema.findAll({210 const listThreadActivityRows = SqlSchema.findAll({
190 Request: Schema.Void,211 Request: Schema.Void,
191 Result: ProjectionThreadActivityDbRowSchema,212 Result: ProjectionThreadActivityDbRowSchema,
...@@ -291,6 +312,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -291,6 +312,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
291 projectRows,312 projectRows,
292 threadRows,313 threadRows,
293 messageRows,314 messageRows,
315 proposedPlanRows,
294 activityRows,316 activityRows,
295 sessionRows,317 sessionRows,
296 checkpointRows,318 checkpointRows,
...@@ -321,6 +343,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -321,6 +343,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
321 ),343 ),
322 ),344 ),
323 ),345 ),
346 listThreadProposedPlanRows(undefined).pipe(
347 Effect.mapError(
348 toPersistenceSqlOrDecodeError(
349 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:query",
350 "ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:decodeRows",
351 ),
352 ),
353 ),
324 listThreadActivityRows(undefined).pipe(354 listThreadActivityRows(undefined).pipe(
325 Effect.mapError(355 Effect.mapError(
326 toPersistenceSqlOrDecodeError(356 toPersistenceSqlOrDecodeError(
...@@ -364,6 +394,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -364,6 +394,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
364 ]);394 ]);
365395
366 const messagesByThread = new Map<string, Array<OrchestrationMessage>>();396 const messagesByThread = new Map<string, Array<OrchestrationMessage>>();
397 const proposedPlansByThread = new Map<string, Array<OrchestrationProposedPlan>>();
367 const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();398 const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();
368 const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();399 const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();
369 const sessionsByThread = new Map<string, OrchestrationSession>();400 const sessionsByThread = new Map<string, OrchestrationSession>();
...@@ -397,6 +428,19 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -397,6 +428,19 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
397 messagesByThread.set(row.threadId, threadMessages);428 messagesByThread.set(row.threadId, threadMessages);
398 }429 }
399430
431 for (const row of proposedPlanRows) {
432 updatedAt = maxIso(updatedAt, row.updatedAt);
433 const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? [];
434 threadProposedPlans.push({
435 id: row.planId,
436 turnId: row.turnId,
437 planMarkdown: row.planMarkdown,
438 createdAt: row.createdAt,
439 updatedAt: row.updatedAt,
440 });
441 proposedPlansByThread.set(row.threadId, threadProposedPlans);
442 }
443
400 for (const row of activityRows) {444 for (const row of activityRows) {
401 updatedAt = maxIso(updatedAt, row.createdAt);445 updatedAt = maxIso(updatedAt, row.createdAt);
402 const threadActivities = activitiesByThread.get(row.threadId) ?? [];446 const threadActivities = activitiesByThread.get(row.threadId) ?? [];
...@@ -494,6 +538,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -494,6 +538,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
494 updatedAt: row.updatedAt,538 updatedAt: row.updatedAt,
495 deletedAt: row.deletedAt,539 deletedAt: row.deletedAt,
496 messages: messagesByThread.get(row.threadId) ?? [],540 messages: messagesByThread.get(row.threadId) ?? [],
541 proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
497 activities: activitiesByThread.get(row.threadId) ?? [],542 activities: activitiesByThread.get(row.threadId) ?? [],
498 checkpoints: checkpointsByThread.get(row.threadId) ?? [],543 checkpoints: checkpointsByThread.get(row.threadId) ?? [],
499 session: sessionsByThread.get(row.threadId) ?? null,544 session: sessionsByThread.get(row.threadId) ?? null,
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+43-35
...@@ -103,6 +103,7 @@ async function waitForThread(...@@ -103,6 +103,7 @@ async function waitForThread(
103type ProviderRuntimeTestReadModel = OrchestrationReadModel;103type ProviderRuntimeTestReadModel = OrchestrationReadModel;
104type ProviderRuntimeTestThread = ProviderRuntimeTestReadModel["threads"][number];104type ProviderRuntimeTestThread = ProviderRuntimeTestReadModel["threads"][number];
105type ProviderRuntimeTestMessage = ProviderRuntimeTestThread["messages"][number];105type ProviderRuntimeTestMessage = ProviderRuntimeTestThread["messages"][number];
106type ProviderRuntimeTestProposedPlan = ProviderRuntimeTestThread["proposedPlans"][number];
106type ProviderRuntimeTestActivity = ProviderRuntimeTestThread["activities"][number];107type ProviderRuntimeTestActivity = ProviderRuntimeTestThread["activities"][number];
107type ProviderRuntimeTestCheckpoint = ProviderRuntimeTestThread["checkpoints"][number];108type ProviderRuntimeTestCheckpoint = ProviderRuntimeTestThread["checkpoints"][number];
108109
...@@ -633,41 +634,35 @@ describe("ProviderRuntimeIngestion", () => {...@@ -633,41 +634,35 @@ describe("ProviderRuntimeIngestion", () => {
633 expect(message?.streaming).toBe(false);634 expect(message?.streaming).toBe(false);
634 });635 });
635636
636 it("projects completed plan items into finalized proposed-plan assistant messages", async () => {637 it("projects completed plan items into first-class proposed plans", async () => {
637 const harness = await createHarness();638 const harness = await createHarness();
638 const now = new Date().toISOString();639 const now = new Date().toISOString();
639640
640 harness.emit({641 harness.emit({
641 type: "item.completed",642 type: "turn.proposed.completed",
642 eventId: asEventId("evt-plan-item-completed"),643 eventId: asEventId("evt-plan-item-completed"),
643 provider: "codex",644 provider: "codex",
644 createdAt: now,645 createdAt: now,
645 threadId: asThreadId("thread-1"),646 threadId: asThreadId("thread-1"),
646 turnId: asTurnId("turn-plan-final"),647 turnId: asTurnId("turn-plan-final"),
647 itemId: asItemId("item-plan-final"),
648 payload: {648 payload: {
649 itemType: "plan",649 planMarkdown: "## Ship plan\n\n- wire projection\n- render follow-up",
650 status: "completed",
651 detail: "## Ship plan\n\n- wire projection\n- render follow-up",
652 },650 },
653 });651 });
654652
655 const thread = await waitForThread(harness.engine, (entry) => {653 const thread = await waitForThread(harness.engine, (entry) =>
656 const message = entry.messages.find(654 entry.proposedPlans.some(
657 (candidate: ProviderRuntimeTestMessage) => candidate.id === "assistant:item-plan-final",655 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
658 );656 proposedPlan.id === "plan:thread-1:turn:turn-plan-final",
659 return message?.streaming === false && message.text.includes("<proposed_plan>");657 ),
660 });
661 const message = thread.messages.find(
662 (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-plan-final",
663 );658 );
664 expect(message?.text).toBe(659 const proposedPlan = thread.proposedPlans.find(
665 "<proposed_plan>\n## Ship plan\n\n- wire projection\n- render follow-up\n</proposed_plan>",660 (entry: ProviderRuntimeTestProposedPlan) => entry.id === "plan:thread-1:turn:turn-plan-final",
666 );661 );
667 expect(message?.streaming).toBe(false);662 expect(proposedPlan?.planMarkdown).toBe("## Ship plan\n\n- wire projection\n- render follow-up");
668 });663 });
669664
670 it("finalizes buffered plan deltas into a proposed-plan assistant message on turn completion", async () => {665 it("finalizes buffered proposed-plan deltas into a first-class proposed plan on turn completion", async () => {
671 const harness = await createHarness();666 const harness = await createHarness();
672 const now = new Date().toISOString();667 const now = new Date().toISOString();
673668
...@@ -687,28 +682,24 @@ describe("ProviderRuntimeIngestion", () => {...@@ -687,28 +682,24 @@ describe("ProviderRuntimeIngestion", () => {
687 );682 );
688683
689 harness.emit({684 harness.emit({
690 type: "content.delta",685 type: "turn.proposed.delta",
691 eventId: asEventId("evt-plan-delta-1"),686 eventId: asEventId("evt-plan-delta-1"),
692 provider: "codex",687 provider: "codex",
693 createdAt: now,688 createdAt: now,
694 threadId: asThreadId("thread-1"),689 threadId: asThreadId("thread-1"),
695 turnId: asTurnId("turn-plan-buffer"),690 turnId: asTurnId("turn-plan-buffer"),
696 itemId: asItemId("item-plan-buffer"),
697 payload: {691 payload: {
698 streamKind: "plan_text",
699 delta: "## Buffered plan\n\n- first",692 delta: "## Buffered plan\n\n- first",
700 },693 },
701 });694 });
702 harness.emit({695 harness.emit({
703 type: "content.delta",696 type: "turn.proposed.delta",
704 eventId: asEventId("evt-plan-delta-2"),697 eventId: asEventId("evt-plan-delta-2"),
705 provider: "codex",698 provider: "codex",
706 createdAt: now,699 createdAt: now,
707 threadId: asThreadId("thread-1"),700 threadId: asThreadId("thread-1"),
708 turnId: asTurnId("turn-plan-buffer"),701 turnId: asTurnId("turn-plan-buffer"),
709 itemId: asItemId("item-plan-buffer"),
710 payload: {702 payload: {
711 streamKind: "plan_text",
712 delta: "\n- second",703 delta: "\n- second",
713 },704 },
714 });705 });
...@@ -724,19 +715,16 @@ describe("ProviderRuntimeIngestion", () => {...@@ -724,19 +715,16 @@ describe("ProviderRuntimeIngestion", () => {
724 },715 },
725 });716 });
726717
727 const thread = await waitForThread(harness.engine, (entry) => {718 const thread = await waitForThread(harness.engine, (entry) =>
728 const message = entry.messages.find(719 entry.proposedPlans.some(
729 (candidate: ProviderRuntimeTestMessage) => candidate.id === "assistant:item-plan-buffer",720 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
730 );721 proposedPlan.id === "plan:thread-1:turn:turn-plan-buffer",
731 return message?.streaming === false && message.text.includes("<proposed_plan>");722 ),
732 });
733 const message = thread.messages.find(
734 (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-plan-buffer",
735 );723 );
736 expect(message?.text).toBe(724 const proposedPlan = thread.proposedPlans.find(
737 "<proposed_plan>\n## Buffered plan\n\n- first\n- second\n</proposed_plan>",725 (entry: ProviderRuntimeTestProposedPlan) => entry.id === "plan:thread-1:turn:turn-plan-buffer",
738 );726 );
739 expect(message?.streaming).toBe(false);727 expect(proposedPlan?.planMarkdown).toBe("## Buffered plan\n\n- first\n- second");
740 });728 });
741729
742 it("buffers assistant deltas by default until completion", async () => {730 it("buffers assistant deltas by default until completion", async () => {
...@@ -1370,12 +1358,27 @@ describe("ProviderRuntimeIngestion", () => {...@@ -1370,12 +1358,27 @@ describe("ProviderRuntimeIngestion", () => {
1370 summary: "<proposed_plan>\n# Plan title\n</proposed_plan>",1358 summary: "<proposed_plan>\n# Plan title\n</proposed_plan>",
1371 },1359 },
1372 });1360 });
1361 harness.emit({
1362 type: "turn.proposed.completed",
1363 eventId: asEventId("evt-task-proposed-plan-completed"),
1364 provider: "codex",
1365 createdAt: now,
1366 threadId: asThreadId("thread-1"),
1367 turnId: asTurnId("turn-task-1"),
1368 payload: {
1369 planMarkdown: "# Plan title",
1370 },
1371 });
13731372
1374 const thread = await waitForThread(1373 const thread = await waitForThread(
1375 harness.engine,1374 harness.engine,
1376 (entry) =>1375 (entry) =>
1377 entry.activities.some(1376 entry.activities.some(
1378 (activity: ProviderRuntimeTestActivity) => activity.kind === "task.completed",1377 (activity: ProviderRuntimeTestActivity) => activity.kind === "task.completed",
1378 ) &&
1379 entry.proposedPlans.some(
1380 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
1381 proposedPlan.id === "plan:thread-1:turn:turn-task-1",
1379 ),1382 ),
1380 );1383 );
13811384
...@@ -1406,6 +1409,11 @@ describe("ProviderRuntimeIngestion", () => {...@@ -1406,6 +1409,11 @@ describe("ProviderRuntimeIngestion", () => {
1406 );1409 );
1407 expect(completed?.kind).toBe("task.completed");1410 expect(completed?.kind).toBe("task.completed");
1408 expect(completedPayload?.detail).toBe("<proposed_plan>\n# Plan title\n</proposed_plan>");1411 expect(completedPayload?.detail).toBe("<proposed_plan>\n# Plan title\n</proposed_plan>");
1412 expect(
1413 thread.proposedPlans.find(
1414 (entry: ProviderRuntimeTestProposedPlan) => entry.id === "plan:thread-1:turn:turn-task-1",
1415 )?.planMarkdown,
1416 ).toBe("# Plan title");
1409 });1417 });
14101418
1411 it("projects structured user input request and resolution as thread activities", async () => {1419 it("projects structured user input request and resolution as thread activities", async () => {
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+172-47
...@@ -26,10 +26,10 @@ const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId...@@ -26,10 +26,10 @@ const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId
26const DEFAULT_ASSISTANT_DELIVERY_MODE: AssistantDeliveryMode = "buffered";26const DEFAULT_ASSISTANT_DELIVERY_MODE: AssistantDeliveryMode = "buffered";
27const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;27const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000;
28const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);28const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120);
29const MESSAGE_STREAM_KIND_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
30const MESSAGE_STREAM_KIND_BY_MESSAGE_ID_TTL = Duration.minutes(120);
31const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;29const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY = 20_000;
32const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);30const BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL = Duration.minutes(120);
31const BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY = 10_000;
32const BUFFERED_PROPOSED_PLAN_BY_ID_TTL = Duration.minutes(120);
33const MAX_BUFFERED_ASSISTANT_CHARS = 24_000;33const MAX_BUFFERED_ASSISTANT_CHARS = 24_000;
34const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0";34const STRICT_PROVIDER_LIFECYCLE_GUARD = process.env.T3CODE_STRICT_PROVIDER_LIFECYCLE_GUARD !== "0";
3535
...@@ -67,12 +67,27 @@ function truncateDetail(value: string, limit = 180): string {...@@ -67,12 +67,27 @@ function truncateDetail(value: string, limit = 180): string {
67 return value.length > limit ? `${value.slice(0, limit - 3)}...` : value;67 return value.length > limit ? `${value.slice(0, limit - 3)}...` : value;
68}68}
6969
70function wrapProposedPlanMessage(planMarkdown: string | undefined): string | undefined {70function normalizeProposedPlanMarkdown(planMarkdown: string | undefined): string | undefined {
71 const trimmed = planMarkdown?.trim();71 const trimmed = planMarkdown?.trim();
72 if (!trimmed) {72 if (!trimmed) {
73 return undefined;73 return undefined;
74 }74 }
75 return `<proposed_plan>\n${trimmed}\n</proposed_plan>`;75 return trimmed;
76}
77
78function proposedPlanIdForTurn(threadId: ThreadId, turnId: TurnId): string {
79 return `plan:${threadId}:turn:${turnId}`;
80}
81
82function proposedPlanIdFromEvent(event: ProviderRuntimeEvent, threadId: ThreadId): string {
83 const turnId = toTurnId(event.turnId);
84 if (turnId) {
85 return proposedPlanIdForTurn(threadId, turnId);
86 }
87 if (event.itemId) {
88 return `plan:${threadId}:item:${event.itemId}`;
89 }
90 return `plan:${threadId}:event:${event.eventId}`;
76}91}
7792
78function asString(value: unknown): string | undefined {93function asString(value: unknown): string | undefined {
...@@ -480,18 +495,18 @@ const make = Effect.gen(function* () {...@@ -480,18 +495,18 @@ const make = Effect.gen(function* () {
480 lookup: () => Effect.succeed(new Set<MessageId>()),495 lookup: () => Effect.succeed(new Set<MessageId>()),
481 });496 });
482497
483 const messageStreamKindByMessageId = yield* Cache.make<MessageId, "assistant_text" | "plan_text">({
484 capacity: MESSAGE_STREAM_KIND_BY_MESSAGE_ID_CACHE_CAPACITY,
485 timeToLive: MESSAGE_STREAM_KIND_BY_MESSAGE_ID_TTL,
486 lookup: () => Effect.succeed("assistant_text"),
487 });
488
489 const bufferedAssistantTextByMessageId = yield* Cache.make<MessageId, string>({498 const bufferedAssistantTextByMessageId = yield* Cache.make<MessageId, string>({
490 capacity: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY,499 capacity: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_CACHE_CAPACITY,
491 timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL,500 timeToLive: BUFFERED_MESSAGE_TEXT_BY_MESSAGE_ID_TTL,
492 lookup: () => Effect.succeed(""),501 lookup: () => Effect.succeed(""),
493 });502 });
494503
504 const bufferedProposedPlanById = yield* Cache.make<string, { text: string; createdAt: string }>({
505 capacity: BUFFERED_PROPOSED_PLAN_BY_ID_CACHE_CAPACITY,
506 timeToLive: BUFFERED_PROPOSED_PLAN_BY_ID_TTL,
507 lookup: () => Effect.succeed({ text: "", createdAt: "" }),
508 });
509
495 const rememberAssistantMessageId = (510 const rememberAssistantMessageId = (
496 threadId: ThreadId,511 threadId: ThreadId,
497 turnId: TurnId,512 turnId: TurnId,
...@@ -545,16 +560,6 @@ const make = Effect.gen(function* () {...@@ -545,16 +560,6 @@ const make = Effect.gen(function* () {
545 const clearAssistantMessageIdsForTurn = (threadId: ThreadId, turnId: TurnId) =>560 const clearAssistantMessageIdsForTurn = (threadId: ThreadId, turnId: TurnId) =>
546 Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(threadId, turnId));561 Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(threadId, turnId));
547562
548 const rememberMessageStreamKind = (
549 messageId: MessageId,
550 streamKind: "assistant_text" | "plan_text",
551 ) => Cache.set(messageStreamKindByMessageId, messageId, streamKind);
552
553 const getMessageStreamKind = (messageId: MessageId) =>
554 Cache.getOption(messageStreamKindByMessageId, messageId).pipe(
555 Effect.map((streamKind) => Option.getOrElse(streamKind, () => "assistant_text" as const)),
556 );
557
558 const appendBufferedAssistantText = (messageId: MessageId, delta: string) =>563 const appendBufferedAssistantText = (messageId: MessageId, delta: string) =>
559 Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe(564 Cache.getOption(bufferedAssistantTextByMessageId, messageId).pipe(
560 Effect.flatMap((existingText) =>565 Effect.flatMap((existingText) =>
...@@ -587,11 +592,30 @@ const make = Effect.gen(function* () {...@@ -587,11 +592,30 @@ const make = Effect.gen(function* () {
587 const clearBufferedAssistantText = (messageId: MessageId) =>592 const clearBufferedAssistantText = (messageId: MessageId) =>
588 Cache.invalidate(bufferedAssistantTextByMessageId, messageId);593 Cache.invalidate(bufferedAssistantTextByMessageId, messageId);
589594
590 const clearAssistantMessageState = (messageId: MessageId) =>595 const appendBufferedProposedPlan = (planId: string, delta: string, createdAt: string) =>
591 Effect.all([596 Cache.getOption(bufferedProposedPlanById, planId).pipe(
592 clearBufferedAssistantText(messageId),597 Effect.flatMap((existingEntry) => {
593 Cache.invalidate(messageStreamKindByMessageId, messageId),598 const existing = Option.getOrUndefined(existingEntry);
594 ]).pipe(Effect.asVoid);599 return Cache.set(bufferedProposedPlanById, planId, {
600 text: `${existing?.text ?? ""}${delta}`,
601 createdAt: existing?.createdAt && existing.createdAt.length > 0 ? existing.createdAt : createdAt,
602 });
603 }),
604 );
605
606 const takeBufferedProposedPlan = (planId: string) =>
607 Cache.getOption(bufferedProposedPlanById, planId).pipe(
608 Effect.flatMap((existingEntry) =>
609 Cache.invalidate(bufferedProposedPlanById, planId).pipe(
610 Effect.as(Option.getOrUndefined(existingEntry)),
611 ),
612 ),
613 );
614
615 const clearBufferedProposedPlan = (planId: string) =>
616 Cache.invalidate(bufferedProposedPlanById, planId);
617
618 const clearAssistantMessageState = (messageId: MessageId) => clearBufferedAssistantText(messageId);
595619
596 const finalizeAssistantMessage = (input: {620 const finalizeAssistantMessage = (input: {
597 event: ProviderRuntimeEvent;621 event: ProviderRuntimeEvent;
...@@ -604,17 +628,10 @@ const make = Effect.gen(function* () {...@@ -604,17 +628,10 @@ const make = Effect.gen(function* () {
604 fallbackText?: string;628 fallbackText?: string;
605 }) =>629 }) =>
606 Effect.gen(function* () {630 Effect.gen(function* () {
607 const messageStreamKind = yield* getMessageStreamKind(input.messageId);
608 const bufferedText = yield* takeBufferedAssistantText(input.messageId);631 const bufferedText = yield* takeBufferedAssistantText(input.messageId);
609 const bufferedFinalText =
610 bufferedText.length > 0
611 ? messageStreamKind === "plan_text"
612 ? (wrapProposedPlanMessage(bufferedText) ?? "")
613 : bufferedText
614 : "";
615 const text =632 const text =
616 bufferedFinalText.length > 0633 bufferedText.length > 0
617 ? bufferedFinalText634 ? bufferedText
618 : (input.fallbackText?.trim().length ?? 0) > 0635 : (input.fallbackText?.trim().length ?? 0) > 0
619 ? input.fallbackText!636 ? input.fallbackText!
620 : "";637 : "";
...@@ -642,10 +659,84 @@ const make = Effect.gen(function* () {...@@ -642,10 +659,84 @@ const make = Effect.gen(function* () {
642 yield* clearAssistantMessageState(input.messageId);659 yield* clearAssistantMessageState(input.messageId);
643 });660 });
644661
662 const upsertProposedPlan = (input: {
663 event: ProviderRuntimeEvent;
664 threadId: ThreadId;
665 threadProposedPlans: ReadonlyArray<{
666 id: string;
667 createdAt: string;
668 }>;
669 planId: string;
670 turnId?: TurnId;
671 planMarkdown: string | undefined;
672 createdAt: string;
673 updatedAt: string;
674 }) =>
675 Effect.gen(function* () {
676 const planMarkdown = normalizeProposedPlanMarkdown(input.planMarkdown);
677 if (!planMarkdown) {
678 return;
679 }
680
681 const existingPlan = input.threadProposedPlans.find((entry) => entry.id === input.planId);
682 yield* orchestrationEngine.dispatch({
683 type: "thread.proposed-plan.upsert",
684 commandId: providerCommandId(input.event, "proposed-plan-upsert"),
685 threadId: input.threadId,
686 proposedPlan: {
687 id: input.planId,
688 turnId: input.turnId ?? null,
689 planMarkdown,
690 createdAt: existingPlan?.createdAt ?? input.createdAt,
691 updatedAt: input.updatedAt,
692 },
693 createdAt: input.updatedAt,
694 });
695 });
696
697 const finalizeBufferedProposedPlan = (input: {
698 event: ProviderRuntimeEvent;
699 threadId: ThreadId;
700 threadProposedPlans: ReadonlyArray<{
701 id: string;
702 createdAt: string;
703 }>;
704 planId: string;
705 turnId?: TurnId;
706 fallbackMarkdown?: string;
707 updatedAt: string;
708 }) =>
709 Effect.gen(function* () {
710 const bufferedPlan = yield* takeBufferedProposedPlan(input.planId);
711 const bufferedMarkdown = normalizeProposedPlanMarkdown(bufferedPlan?.text);
712 const fallbackMarkdown = normalizeProposedPlanMarkdown(input.fallbackMarkdown);
713 const planMarkdown = bufferedMarkdown ?? fallbackMarkdown;
714 if (!planMarkdown) {
715 return;
716 }
717
718 yield* upsertProposedPlan({
719 event: input.event,
720 threadId: input.threadId,
721 threadProposedPlans: input.threadProposedPlans,
722 planId: input.planId,
723 ...(input.turnId ? { turnId: input.turnId } : {}),
724 planMarkdown,
725 createdAt:
726 bufferedPlan?.createdAt && bufferedPlan.createdAt.length > 0
727 ? bufferedPlan.createdAt
728 : input.updatedAt,
729 updatedAt: input.updatedAt,
730 });
731 yield* clearBufferedProposedPlan(input.planId);
732 });
733
645 const clearTurnStateForSession = (threadId: ThreadId) =>734 const clearTurnStateForSession = (threadId: ThreadId) =>
646 Effect.gen(function* () {735 Effect.gen(function* () {
647 const prefix = `${threadId}:`;736 const prefix = `${threadId}:`;
737 const proposedPlanPrefix = `plan:${threadId}:`;
648 const turnKeys = Array.from(yield* Cache.keys(turnMessageIdsByTurnKey));738 const turnKeys = Array.from(yield* Cache.keys(turnMessageIdsByTurnKey));
739 const proposedPlanKeys = Array.from(yield* Cache.keys(bufferedProposedPlanById));
649 yield* Effect.forEach(740 yield* Effect.forEach(
650 turnKeys,741 turnKeys,
651 (key) =>742 (key) =>
...@@ -665,6 +756,14 @@ const make = Effect.gen(function* () {...@@ -665,6 +756,14 @@ const make = Effect.gen(function* () {
665 }),756 }),
666 { concurrency: 1 },757 { concurrency: 1 },
667 ).pipe(Effect.asVoid);758 ).pipe(Effect.asVoid);
759 yield* Effect.forEach(
760 proposedPlanKeys,
761 (key) =>
762 key.startsWith(proposedPlanPrefix)
763 ? Cache.invalidate(bufferedProposedPlanById, key)
764 : Effect.void,
765 { concurrency: 1 },
766 ).pipe(Effect.asVoid);
668 });767 });
669768
670 const processRuntimeEvent = (event: ProviderRuntimeEvent) =>769 const processRuntimeEvent = (event: ProviderRuntimeEvent) =>
...@@ -768,24 +867,20 @@ const make = Effect.gen(function* () {...@@ -768,24 +867,20 @@ const make = Effect.gen(function* () {
768 }867 }
769868
770 const assistantDelta =869 const assistantDelta =
771 event.type === "content.delta" &&870 event.type === "content.delta" && event.payload.streamKind === "assistant_text"
772 (event.payload.streamKind === "assistant_text" || event.payload.streamKind === "plan_text")
773 ? event.payload.delta871 ? event.payload.delta
774 : undefined;872 : undefined;
873 const proposedPlanDelta =
874 event.type === "turn.proposed.delta" ? event.payload.delta : undefined;
775875
776 if (assistantDelta && assistantDelta.length > 0) {876 if (assistantDelta && assistantDelta.length > 0) {
777 const assistantMessageId = MessageId.makeUnsafe(877 const assistantMessageId = MessageId.makeUnsafe(
778 `assistant:${event.itemId ?? event.turnId ?? event.eventId}`,878 `assistant:${event.itemId ?? event.turnId ?? event.eventId}`,
779 );879 );
780 const turnId = toTurnId(event.turnId);880 const turnId = toTurnId(event.turnId);
781 const messageStreamKind =
782 event.type === "content.delta" && event.payload.streamKind === "plan_text"
783 ? "plan_text"
784 : "assistant_text";
785 if (turnId) {881 if (turnId) {
786 yield* rememberAssistantMessageId(thread.id, turnId, assistantMessageId);882 yield* rememberAssistantMessageId(thread.id, turnId, assistantMessageId);
787 }883 }
788 yield* rememberMessageStreamKind(assistantMessageId, messageStreamKind);
789884
790 const assistantDeliveryMode = yield* Ref.get(assistantDeliveryModeRef);885 const assistantDeliveryMode = yield* Ref.get(assistantDeliveryModeRef);
791 if (assistantDeliveryMode === "buffered") {886 if (assistantDeliveryMode === "buffered") {
...@@ -814,15 +909,24 @@ const make = Effect.gen(function* () {...@@ -814,15 +909,24 @@ const make = Effect.gen(function* () {
814 }909 }
815 }910 }
816911
912 if (proposedPlanDelta && proposedPlanDelta.length > 0) {
913 const planId = proposedPlanIdFromEvent(event, thread.id);
914 yield* appendBufferedProposedPlan(planId, proposedPlanDelta, now);
915 }
916
817 const assistantCompletion =917 const assistantCompletion =
818 event.type === "item.completed" &&918 event.type === "item.completed" && event.payload.itemType === "assistant_message"
819 (event.payload.itemType === "assistant_message" || event.payload.itemType === "plan")
820 ? {919 ? {
821 messageId: MessageId.makeUnsafe(`assistant:${event.itemId ?? event.turnId ?? event.eventId}`),920 messageId: MessageId.makeUnsafe(`assistant:${event.itemId ?? event.turnId ?? event.eventId}`),
822 fallbackText:921 fallbackText: event.payload.detail,
823 event.payload.itemType === "plan"922 }
824 ? wrapProposedPlanMessage(event.payload.detail)923 : undefined;
825 : event.payload.detail,924 const proposedPlanCompletion =
925 event.type === "turn.proposed.completed"
926 ? {
927 planId: proposedPlanIdFromEvent(event, thread.id),
928 turnId: toTurnId(event.turnId),
929 planMarkdown: event.payload.planMarkdown,
826 }930 }
827 : undefined;931 : undefined;
828932
...@@ -851,6 +955,18 @@ const make = Effect.gen(function* () {...@@ -851,6 +955,18 @@ const make = Effect.gen(function* () {
851 }955 }
852 }956 }
853957
958 if (proposedPlanCompletion) {
959 yield* finalizeBufferedProposedPlan({
960 event,
961 threadId: thread.id,
962 threadProposedPlans: thread.proposedPlans,
963 planId: proposedPlanCompletion.planId,
964 ...(proposedPlanCompletion.turnId ? { turnId: proposedPlanCompletion.turnId } : {}),
965 fallbackMarkdown: proposedPlanCompletion.planMarkdown,
966 updatedAt: now,
967 });
968 }
969
854 if (event.type === "turn.completed") {970 if (event.type === "turn.completed") {
855 const turnId = toTurnId(event.turnId);971 const turnId = toTurnId(event.turnId);
856 if (turnId) {972 if (turnId) {
...@@ -870,6 +986,15 @@ const make = Effect.gen(function* () {...@@ -870,6 +986,15 @@ const make = Effect.gen(function* () {
870 { concurrency: 1 },986 { concurrency: 1 },
871 ).pipe(Effect.asVoid);987 ).pipe(Effect.asVoid);
872 yield* clearAssistantMessageIdsForTurn(thread.id, turnId);988 yield* clearAssistantMessageIdsForTurn(thread.id, turnId);
989
990 yield* finalizeBufferedProposedPlan({
991 event,
992 threadId: thread.id,
993 threadProposedPlans: thread.proposedPlans,
994 planId: proposedPlanIdForTurn(thread.id, turnId),
995 turnId,
996 updatedAt: now,
997 });
873 }998 }
874 }999 }
8751000
apps/server/src/orchestration/Schemas.ts+2
...@@ -8,6 +8,7 @@ import {...@@ -8,6 +8,7 @@ import {
8 ThreadInteractionModeSetPayload as ContractsThreadInteractionModeSetPayloadSchema,8 ThreadInteractionModeSetPayload as ContractsThreadInteractionModeSetPayloadSchema,
9 ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema,9 ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema,
10 ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema,10 ThreadMessageSentPayload as ContractsThreadMessageSentPayloadSchema,
11 ThreadProposedPlanUpsertedPayload as ContractsThreadProposedPlanUpsertedPayloadSchema,
11 ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema,12 ThreadSessionSetPayload as ContractsThreadSessionSetPayloadSchema,
12 ThreadTurnDiffCompletedPayload as ContractsThreadTurnDiffCompletedPayloadSchema,13 ThreadTurnDiffCompletedPayload as ContractsThreadTurnDiffCompletedPayloadSchema,
13 ThreadRevertedPayload as ContractsThreadRevertedPayloadSchema,14 ThreadRevertedPayload as ContractsThreadRevertedPayloadSchema,
...@@ -31,6 +32,7 @@ export const ThreadInteractionModeSetPayload = ContractsThreadInteractionModeSet...@@ -31,6 +32,7 @@ export const ThreadInteractionModeSetPayload = ContractsThreadInteractionModeSet
31export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema;32export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema;
3233
33export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema;34export const MessageSentPayloadSchema = ContractsThreadMessageSentPayloadSchema;
35export const ThreadProposedPlanUpsertedPayload = ContractsThreadProposedPlanUpsertedPayloadSchema;
34export const ThreadSessionSetPayload = ContractsThreadSessionSetPayloadSchema;36export const ThreadSessionSetPayload = ContractsThreadSessionSetPayloadSchema;
35export const ThreadTurnDiffCompletedPayload = ContractsThreadTurnDiffCompletedPayloadSchema;37export const ThreadTurnDiffCompletedPayload = ContractsThreadTurnDiffCompletedPayloadSchema;
36export const ThreadRevertedPayload = ContractsThreadRevertedPayloadSchema;38export const ThreadRevertedPayload = ContractsThreadRevertedPayloadSchema;
apps/server/src/orchestration/commandInvariants.test.ts+2
...@@ -61,6 +61,7 @@ const readModel: OrchestrationReadModel = {...@@ -61,6 +61,7 @@ const readModel: OrchestrationReadModel = {
61 messages: [],61 messages: [],
62 session: null,62 session: null,
63 activities: [],63 activities: [],
64 proposedPlans: [],
64 checkpoints: [],65 checkpoints: [],
65 deletedAt: null,66 deletedAt: null,
66 },67 },
...@@ -79,6 +80,7 @@ const readModel: OrchestrationReadModel = {...@@ -79,6 +80,7 @@ const readModel: OrchestrationReadModel = {
79 messages: [],80 messages: [],
80 session: null,81 session: null,
81 activities: [],82 activities: [],
83 proposedPlans: [],
82 checkpoints: [],84 checkpoints: [],
83 deletedAt: null,85 deletedAt: null,
84 },86 },
apps/server/src/orchestration/decider.ts+21
...@@ -508,6 +508,27 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"...@@ -508,6 +508,27 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
508 };508 };
509 }509 }
510510
511 case "thread.proposed-plan.upsert": {
512 yield* requireThread({
513 readModel,
514 command,
515 threadId: command.threadId,
516 });
517 return {
518 ...withEventBase({
519 aggregateKind: "thread",
520 aggregateId: command.threadId,
521 occurredAt: command.createdAt,
522 commandId: command.commandId,
523 }),
524 type: "thread.proposed-plan-upserted",
525 payload: {
526 threadId: command.threadId,
527 proposedPlan: command.proposedPlan,
528 },
529 };
530 }
531
511 case "thread.turn.diff.complete": {532 case "thread.turn.diff.complete": {
512 yield* requireThread({533 yield* requireThread({
513 readModel,534 readModel,
apps/server/src/orchestration/projector.ts+47
...@@ -18,6 +18,7 @@ import {...@@ -18,6 +18,7 @@ import {
18 ThreadDeletedPayload,18 ThreadDeletedPayload,
19 ThreadInteractionModeSetPayload,19 ThreadInteractionModeSetPayload,
20 ThreadMetaUpdatedPayload,20 ThreadMetaUpdatedPayload,
21 ThreadProposedPlanUpsertedPayload,
21 ThreadRuntimeModeSetPayload,22 ThreadRuntimeModeSetPayload,
22 ThreadRevertedPayload,23 ThreadRevertedPayload,
23 ThreadSessionSetPayload,24 ThreadSessionSetPayload,
...@@ -126,6 +127,15 @@ function retainThreadActivitiesAfterRevert(...@@ -126,6 +127,15 @@ function retainThreadActivitiesAfterRevert(
126 );127 );
127}128}
128129
130function retainThreadProposedPlansAfterRevert(
131 proposedPlans: ReadonlyArray<OrchestrationThread["proposedPlans"][number]>,
132 retainedTurnIds: ReadonlySet<string>,
133): ReadonlyArray<OrchestrationThread["proposedPlans"][number]> {
134 return proposedPlans.filter(
135 (proposedPlan) => proposedPlan.turnId === null || retainedTurnIds.has(proposedPlan.turnId),
136 );
137}
138
129function compareThreadActivities(139function compareThreadActivities(
130 left: OrchestrationThread["activities"][number],140 left: OrchestrationThread["activities"][number],
131 right: OrchestrationThread["activities"][number],141 right: OrchestrationThread["activities"][number],
...@@ -435,6 +445,38 @@ export function projectEvent(...@@ -435,6 +445,38 @@ export function projectEvent(
435 };445 };
436 });446 });
437447
448 case "thread.proposed-plan-upserted":
449 return Effect.gen(function* () {
450 const payload = yield* decodeForEvent(
451 ThreadProposedPlanUpsertedPayload,
452 event.payload,
453 event.type,
454 "payload",
455 );
456 const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);
457 if (!thread) {
458 return nextBase;
459 }
460
461 const proposedPlans = [
462 ...thread.proposedPlans.filter((entry) => entry.id !== payload.proposedPlan.id),
463 payload.proposedPlan,
464 ]
465 .toSorted(
466 (left, right) =>
467 left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id),
468 )
469 .slice(-200);
470
471 return {
472 ...nextBase,
473 threads: updateThread(nextBase.threads, payload.threadId, {
474 proposedPlans,
475 updatedAt: event.occurredAt,
476 }),
477 };
478 });
479
438 case "thread.turn-diff-completed":480 case "thread.turn-diff-completed":
439 return Effect.gen(function* () {481 return Effect.gen(function* () {
440 const payload = yield* decodeForEvent(482 const payload = yield* decodeForEvent(
...@@ -511,6 +553,10 @@ export function projectEvent(...@@ -511,6 +553,10 @@ export function projectEvent(
511 retainedTurnIds,553 retainedTurnIds,
512 payload.turnCount,554 payload.turnCount,
513 ).slice(-MAX_THREAD_MESSAGES);555 ).slice(-MAX_THREAD_MESSAGES);
556 const proposedPlans = retainThreadProposedPlansAfterRevert(
557 thread.proposedPlans,
558 retainedTurnIds,
559 ).slice(-200);
514 const activities = retainThreadActivitiesAfterRevert(thread.activities, retainedTurnIds);560 const activities = retainThreadActivitiesAfterRevert(thread.activities, retainedTurnIds);
515561
516 const latestCheckpoint = checkpoints.at(-1) ?? null;562 const latestCheckpoint = checkpoints.at(-1) ?? null;
...@@ -531,6 +577,7 @@ export function projectEvent(...@@ -531,6 +577,7 @@ export function projectEvent(
531 threads: updateThread(nextBase.threads, payload.threadId, {577 threads: updateThread(nextBase.threads, payload.threadId, {
532 checkpoints,578 checkpoints,
533 messages,579 messages,
580 proposedPlans,
534 activities,581 activities,
535 latestTurn,582 latestTurn,
536 updatedAt: event.occurredAt,583 updatedAt: event.occurredAt,
apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts created+104
...@@ -0,0 +1,104 @@
1import { Effect, Layer } from "effect";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
3import * as SqlSchema from "effect/unstable/sql/SqlSchema";
4
5import { toPersistenceSqlError } from "../Errors.ts";
6import {
7 DeleteProjectionThreadProposedPlansInput,
8 ListProjectionThreadProposedPlansInput,
9 ProjectionThreadProposedPlan,
10 ProjectionThreadProposedPlanRepository,
11 type ProjectionThreadProposedPlanRepositoryShape,
12} from "../Services/ProjectionThreadProposedPlans.ts";
13
14const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
15 const sql = yield* SqlClient.SqlClient;
16
17 const upsertProjectionThreadProposedPlanRow = SqlSchema.void({
18 Request: ProjectionThreadProposedPlan,
19 execute: (row) => sql`
20 INSERT INTO projection_thread_proposed_plans (
21 plan_id,
22 thread_id,
23 turn_id,
24 plan_markdown,
25 created_at,
26 updated_at
27 )
28 VALUES (
29 ${row.planId},
30 ${row.threadId},
31 ${row.turnId},
32 ${row.planMarkdown},
33 ${row.createdAt},
34 ${row.updatedAt}
35 )
36 ON CONFLICT (plan_id)
37 DO UPDATE SET
38 thread_id = excluded.thread_id,
39 turn_id = excluded.turn_id,
40 plan_markdown = excluded.plan_markdown,
41 created_at = excluded.created_at,
42 updated_at = excluded.updated_at
43 `,
44 });
45
46 const listProjectionThreadProposedPlanRows = SqlSchema.findAll({
47 Request: ListProjectionThreadProposedPlansInput,
48 Result: ProjectionThreadProposedPlan,
49 execute: ({ threadId }) => sql`
50 SELECT
51 plan_id AS "planId",
52 thread_id AS "threadId",
53 turn_id AS "turnId",
54 plan_markdown AS "planMarkdown",
55 created_at AS "createdAt",
56 updated_at AS "updatedAt"
57 FROM projection_thread_proposed_plans
58 WHERE thread_id = ${threadId}
59 ORDER BY created_at ASC, plan_id ASC
60 `,
61 });
62
63 const deleteProjectionThreadProposedPlanRows = SqlSchema.void({
64 Request: DeleteProjectionThreadProposedPlansInput,
65 execute: ({ threadId }) => sql`
66 DELETE FROM projection_thread_proposed_plans
67 WHERE thread_id = ${threadId}
68 `,
69 });
70
71 const upsert: ProjectionThreadProposedPlanRepositoryShape["upsert"] = (row) =>
72 upsertProjectionThreadProposedPlanRow(row).pipe(
73 Effect.mapError(
74 toPersistenceSqlError("ProjectionThreadProposedPlanRepository.upsert:query"),
75 ),
76 );
77
78 const listByThreadId: ProjectionThreadProposedPlanRepositoryShape["listByThreadId"] = (input) =>
79 listProjectionThreadProposedPlanRows(input).pipe(
80 Effect.mapError(
81 toPersistenceSqlError("ProjectionThreadProposedPlanRepository.listByThreadId:query"),
82 ),
83 );
84
85 const deleteByThreadId: ProjectionThreadProposedPlanRepositoryShape["deleteByThreadId"] = (
86 input,
87 ) =>
88 deleteProjectionThreadProposedPlanRows(input).pipe(
89 Effect.mapError(
90 toPersistenceSqlError("ProjectionThreadProposedPlanRepository.deleteByThreadId:query"),
91 ),
92 );
93
94 return {
95 upsert,
96 listByThreadId,
97 deleteByThreadId,
98 } satisfies ProjectionThreadProposedPlanRepositoryShape;
99});
100
101export const ProjectionThreadProposedPlanRepositoryLive = Layer.effect(
102 ProjectionThreadProposedPlanRepository,
103 makeProjectionThreadProposedPlanRepository,
104);
apps/server/src/persistence/Migrations.ts+2
...@@ -24,6 +24,7 @@ import Migration0009 from "./Migrations/009_ProviderSessionRuntimeMode.ts";...@@ -24,6 +24,7 @@ import Migration0009 from "./Migrations/009_ProviderSessionRuntimeMode.ts";
24import Migration0010 from "./Migrations/010_ProjectionThreadsRuntimeMode.ts";24import Migration0010 from "./Migrations/010_ProjectionThreadsRuntimeMode.ts";
25import Migration0011 from "./Migrations/011_OrchestrationThreadCreatedRuntimeMode.ts";25import Migration0011 from "./Migrations/011_OrchestrationThreadCreatedRuntimeMode.ts";
26import Migration0012 from "./Migrations/012_ProjectionThreadsInteractionMode.ts";26import Migration0012 from "./Migrations/012_ProjectionThreadsInteractionMode.ts";
27import Migration0013 from "./Migrations/013_ProjectionThreadProposedPlans.ts";
27import { Effect } from "effect";28import { Effect } from "effect";
2829
29/**30/**
...@@ -49,6 +50,7 @@ const loader = Migrator.fromRecord({...@@ -49,6 +50,7 @@ const loader = Migrator.fromRecord({
49 "10_ProjectionThreadsRuntimeMode": Migration0010,50 "10_ProjectionThreadsRuntimeMode": Migration0010,
50 "11_OrchestrationThreadCreatedRuntimeMode": Migration0011,51 "11_OrchestrationThreadCreatedRuntimeMode": Migration0011,
51 "12_ProjectionThreadsInteractionMode": Migration0012,52 "12_ProjectionThreadsInteractionMode": Migration0012,
53 "13_ProjectionThreadProposedPlans": Migration0013,
52});54});
5355
54/**56/**
apps/server/src/persistence/Migrations/013_ProjectionThreadProposedPlans.ts created+22
...@@ -0,0 +1,22 @@
1import * as Effect from "effect/Effect";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 yield* sql`
8 CREATE TABLE IF NOT EXISTS projection_thread_proposed_plans (
9 plan_id TEXT PRIMARY KEY,
10 thread_id TEXT NOT NULL,
11 turn_id TEXT,
12 plan_markdown TEXT NOT NULL,
13 created_at TEXT NOT NULL,
14 updated_at TEXT NOT NULL
15 )
16 `;
17
18 yield* sql`
19 CREATE INDEX IF NOT EXISTS idx_projection_thread_proposed_plans_thread_created
20 ON projection_thread_proposed_plans(thread_id, created_at)
21 `;
22});
apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts created+52
...@@ -0,0 +1,52 @@
1import {
2 IsoDateTime,
3 OrchestrationProposedPlanId,
4 ThreadId,
5 TrimmedNonEmptyString,
6 TurnId,
7} from "@t3tools/contracts";
8import { Schema, ServiceMap } from "effect";
9import type { Effect } from "effect";
10
11import type { ProjectionRepositoryError } from "../Errors.ts";
12
13export const ProjectionThreadProposedPlan = Schema.Struct({
14 planId: OrchestrationProposedPlanId,
15 threadId: ThreadId,
16 turnId: Schema.NullOr(TurnId),
17 planMarkdown: TrimmedNonEmptyString,
18 createdAt: IsoDateTime,
19 updatedAt: IsoDateTime,
20});
21export type ProjectionThreadProposedPlan = typeof ProjectionThreadProposedPlan.Type;
22
23export const ListProjectionThreadProposedPlansInput = Schema.Struct({
24 threadId: ThreadId,
25});
26export type ListProjectionThreadProposedPlansInput =
27 typeof ListProjectionThreadProposedPlansInput.Type;
28
29export const DeleteProjectionThreadProposedPlansInput = Schema.Struct({
30 threadId: ThreadId,
31});
32export type DeleteProjectionThreadProposedPlansInput =
33 typeof DeleteProjectionThreadProposedPlansInput.Type;
34
35export interface ProjectionThreadProposedPlanRepositoryShape {
36 readonly upsert: (
37 proposedPlan: ProjectionThreadProposedPlan,
38 ) => Effect.Effect<void, ProjectionRepositoryError>;
39 readonly listByThreadId: (
40 input: ListProjectionThreadProposedPlansInput,
41 ) => Effect.Effect<ReadonlyArray<ProjectionThreadProposedPlan>, ProjectionRepositoryError>;
42 readonly deleteByThreadId: (
43 input: DeleteProjectionThreadProposedPlansInput,
44 ) => Effect.Effect<void, ProjectionRepositoryError>;
45}
46
47export class ProjectionThreadProposedPlanRepository extends ServiceMap.Service<
48 ProjectionThreadProposedPlanRepository,
49 ProjectionThreadProposedPlanRepositoryShape
50>()(
51 "t3/persistence/Services/ProjectionThreadProposedPlans/ProjectionThreadProposedPlanRepository",
52) {}
apps/server/src/provider/Layers/CodexAdapter.test.ts+46-7
...@@ -270,7 +270,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {...@@ -270,7 +270,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
270 }),270 }),
271 );271 );
272272
273 it.effect("maps completed plan items to canonical item.completed plan events", () =>273 it.effect("maps completed plan items to canonical proposed-plan completion events", () =>
274 Effect.gen(function* () {274 Effect.gen(function* () {
275 const adapter = yield* CodexAdapter;275 const adapter = yield* CodexAdapter;
276 const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);276 const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);
...@@ -300,13 +300,46 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {...@@ -300,13 +300,46 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
300 if (firstEvent._tag !== "Some") {300 if (firstEvent._tag !== "Some") {
301 return;301 return;
302 }302 }
303 assert.equal(firstEvent.value.type, "item.completed");303 assert.equal(firstEvent.value.type, "turn.proposed.completed");
304 if (firstEvent.value.type !== "item.completed") {304 if (firstEvent.value.type !== "turn.proposed.completed") {
305 return;
306 }
307 assert.equal(firstEvent.value.turnId, "turn-1");
308 assert.equal(firstEvent.value.payload.planMarkdown, "## Final plan\n\n- one\n- two");
309 }),
310 );
311
312 it.effect("maps plan deltas to canonical proposed-plan delta events", () =>
313 Effect.gen(function* () {
314 const adapter = yield* CodexAdapter;
315 const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);
316
317 lifecycleManager.emit("event", {
318 id: asEventId("evt-plan-delta"),
319 kind: "notification",
320 provider: "codex",
321 createdAt: new Date().toISOString(),
322 method: "item/plan/delta",
323 threadId: asThreadId("thread-1"),
324 turnId: asTurnId("turn-1"),
325 itemId: asItemId("plan_1"),
326 payload: {
327 delta: "## Final plan",
328 },
329 } satisfies ProviderEvent);
330
331 const firstEvent = yield* Fiber.join(firstEventFiber);
332
333 assert.equal(firstEvent._tag, "Some");
334 if (firstEvent._tag !== "Some") {
335 return;
336 }
337 assert.equal(firstEvent.value.type, "turn.proposed.delta");
338 if (firstEvent.value.type !== "turn.proposed.delta") {
305 return;339 return;
306 }340 }
307 assert.equal(firstEvent.value.itemId, "plan_1");341 assert.equal(firstEvent.value.turnId, "turn-1");
308 assert.equal(firstEvent.value.payload.itemType, "plan");342 assert.equal(firstEvent.value.payload.delta, "## Final plan");
309 assert.equal(firstEvent.value.payload.detail, "## Final plan\n\n- one\n- two");
310 }),343 }),
311 );344 );
312345
...@@ -487,7 +520,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {...@@ -487,7 +520,7 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
487 it.effect("maps Codex task and reasoning event chunks into canonical runtime events", () =>520 it.effect("maps Codex task and reasoning event chunks into canonical runtime events", () =>
488 Effect.gen(function* () {521 Effect.gen(function* () {
489 const adapter = yield* CodexAdapter;522 const adapter = yield* CodexAdapter;
490 const eventsFiber = yield* Stream.runCollect(Stream.take(adapter.streamEvents, 4)).pipe(523 const eventsFiber = yield* Stream.runCollect(Stream.take(adapter.streamEvents, 5)).pipe(
491 Effect.forkChild,524 Effect.forkChild,
492 );525 );
493526
...@@ -592,6 +625,12 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {...@@ -592,6 +625,12 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
592 assert.equal(events[3].payload.taskId, "turn-structured-1");625 assert.equal(events[3].payload.taskId, "turn-structured-1");
593 assert.equal(events[3].payload.summary, "<proposed_plan>\n# Ship it\n</proposed_plan>");626 assert.equal(events[3].payload.summary, "<proposed_plan>\n# Ship it\n</proposed_plan>");
594 }627 }
628
629 assert.equal(events[4]?.type, "turn.proposed.completed");
630 if (events[4]?.type === "turn.proposed.completed") {
631 assert.equal(events[4].turnId, "turn-structured-1");
632 assert.equal(events[4].payload.planMarkdown, "# Ship it");
633 }
595 }),634 }),
596 );635 );
597});636});
apps/server/src/provider/Layers/CodexAdapter.ts+75-6
...@@ -374,8 +374,6 @@ function contentStreamKindFromMethod(...@@ -374,8 +374,6 @@ function contentStreamKindFromMethod(
374 return "reasoning_text";374 return "reasoning_text";
375 case "item/reasoning/summaryTextDelta":375 case "item/reasoning/summaryTextDelta":
376 return "reasoning_summary_text";376 return "reasoning_summary_text";
377 case "item/plan/delta":
378 return "plan_text";
379 case "item/commandExecution/outputDelta":377 case "item/commandExecution/outputDelta":
380 return "command_output";378 return "command_output";
381 case "item/fileChange/outputDelta":379 case "item/fileChange/outputDelta":
...@@ -385,6 +383,14 @@ function contentStreamKindFromMethod(...@@ -385,6 +383,14 @@ function contentStreamKindFromMethod(
385 }383 }
386}384}
387385
386const PROPOSED_PLAN_BLOCK_REGEX = /<proposed_plan>\s*([\s\S]*?)\s*<\/proposed_plan>/i;
387
388function extractProposedPlanMarkdown(text: string | undefined): string | undefined {
389 const match = text ? PROPOSED_PLAN_BLOCK_REGEX.exec(text) : null;
390 const planMarkdown = match?.[1]?.trim();
391 return planMarkdown && planMarkdown.length > 0 ? planMarkdown : undefined;
392}
393
388function asRuntimeItemId(itemId: ProviderItemId): RuntimeItemId {394function asRuntimeItemId(itemId: ProviderItemId): RuntimeItemId {
389 return RuntimeItemId.makeUnsafe(itemId);395 return RuntimeItemId.makeUnsafe(itemId);
390}396}
...@@ -487,7 +493,7 @@ function mapItemLifecycle(...@@ -487,7 +493,7 @@ function mapItemLifecycle(
487 return undefined;493 return undefined;
488 }494 }
489495
490 const detail = itemDetail(source, payload ?? source);496 const detail = itemDetail(source, payload ?? {});
491 const status =497 const status =
492 lifecycle === "item.started"498 lifecycle === "item.started"
493 ? "inProgress"499 ? "inProgress"
...@@ -800,6 +806,28 @@ function mapToRuntimeEvents(...@@ -800,6 +806,28 @@ function mapToRuntimeEvents(
800 }806 }
801807
802 if (event.method === "item/completed") {808 if (event.method === "item/completed") {
809 const payload = asObject(event.payload);
810 const item = asObject(payload?.item);
811 const source = item ?? payload;
812 if (!source) {
813 return [];
814 }
815 const itemType = source ? toCanonicalItemType(source.type ?? source.kind) : "unknown";
816 if (itemType === "plan") {
817 const detail = itemDetail(source, payload ?? {});
818 if (!detail) {
819 return [];
820 }
821 return [
822 {
823 ...runtimeEventBase(event, canonicalThreadId),
824 type: "turn.proposed.completed",
825 payload: {
826 planMarkdown: detail,
827 },
828 },
829 ];
830 }
803 const completed = mapItemLifecycle(event, canonicalThreadId, "item.completed");831 const completed = mapItemLifecycle(event, canonicalThreadId, "item.completed");
804 return completed ? [completed] : [];832 return completed ? [completed] : [];
805 }833 }
...@@ -812,9 +840,28 @@ function mapToRuntimeEvents(...@@ -812,9 +840,28 @@ function mapToRuntimeEvents(
812 return updated ? [updated] : [];840 return updated ? [updated] : [];
813 }841 }
814842
843 if (event.method === "item/plan/delta") {
844 const delta =
845 event.textDelta ??
846 asString(payload?.delta) ??
847 asString(payload?.text) ??
848 asString(asObject(payload?.content)?.text);
849 if (!delta || delta.length === 0) {
850 return [];
851 }
852 return [
853 {
854 ...runtimeEventBase(event, canonicalThreadId),
855 type: "turn.proposed.delta",
856 payload: {
857 delta,
858 },
859 },
860 ];
861 }
862
815 if (863 if (
816 event.method === "item/agentMessage/delta" ||864 event.method === "item/agentMessage/delta" ||
817 event.method === "item/plan/delta" ||
818 event.method === "item/commandExecution/outputDelta" ||865 event.method === "item/commandExecution/outputDelta" ||
819 event.method === "item/fileChange/outputDelta" ||866 event.method === "item/fileChange/outputDelta" ||
820 event.method === "item/reasoning/summaryTextDelta" ||867 event.method === "item/reasoning/summaryTextDelta" ||
...@@ -919,10 +966,22 @@ function mapToRuntimeEvents(...@@ -919,10 +966,22 @@ function mapToRuntimeEvents(
919 if (event.method === "codex/event/task_complete") {966 if (event.method === "codex/event/task_complete") {
920 const msg = codexEventMessage(payload);967 const msg = codexEventMessage(payload);
921 const taskId = asString(payload?.id) ?? asString(msg?.turn_id);968 const taskId = asString(payload?.id) ?? asString(msg?.turn_id);
969 const proposedPlanMarkdown = extractProposedPlanMarkdown(asString(msg?.last_agent_message));
922 if (!taskId) {970 if (!taskId) {
923 return [];971 if (!proposedPlanMarkdown) {
972 return [];
973 }
974 return [
975 {
976 ...codexEventBase(event, canonicalThreadId),
977 type: "turn.proposed.completed",
978 payload: {
979 planMarkdown: proposedPlanMarkdown,
980 },
981 },
982 ];
924 }983 }
925 return [984 const events: ProviderRuntimeEvent[] = [
926 {985 {
927 ...codexEventBase(event, canonicalThreadId),986 ...codexEventBase(event, canonicalThreadId),
928 type: "task.completed",987 type: "task.completed",
...@@ -935,6 +994,16 @@ function mapToRuntimeEvents(...@@ -935,6 +994,16 @@ function mapToRuntimeEvents(
935 },994 },
936 },995 },
937 ];996 ];
997 if (proposedPlanMarkdown) {
998 events.push({
999 ...codexEventBase(event, canonicalThreadId),
1000 type: "turn.proposed.completed",
1001 payload: {
1002 planMarkdown: proposedPlanMarkdown,
1003 },
1004 });
1005 }
1006 return events;
938 }1007 }
9391008
940 if (event.method === "codex/event/agent_reasoning") {1009 if (event.method === "codex/event/agent_reasoning") {
apps/web/src/components/ChatView.tsx+105-76
...@@ -72,6 +72,7 @@ import {...@@ -72,6 +72,7 @@ import {
72 derivePhase,72 derivePhase,
73 deriveTimelineEntries,73 deriveTimelineEntries,
74 deriveActivePlanState,74 deriveActivePlanState,
75 findLatestProposedPlan,
75 type PendingApproval,76 type PendingApproval,
76 type PendingUserInput,77 type PendingUserInput,
77 PROVIDER_OPTIONS,78 PROVIDER_OPTIONS,
...@@ -93,8 +94,6 @@ import {...@@ -93,8 +94,6 @@ import {
93 buildPlanImplementationThreadTitle,94 buildPlanImplementationThreadTitle,
94 buildPlanImplementationPrompt,95 buildPlanImplementationPrompt,
95 buildProposedPlanMarkdownFilename,96 buildProposedPlanMarkdownFilename,
96 findLatestProposedPlanMessage,
97 parseProposedPlanMessage,
98 proposedPlanTitle,97 proposedPlanTitle,
99} from "../proposedPlan";98} from "../proposedPlan";
100import { truncateTitle } from "../truncateTitle";99import { truncateTitle } from "../truncateTitle";
...@@ -334,6 +333,7 @@ function buildLocalDraftThread(...@@ -334,6 +333,7 @@ function buildLocalDraftThread(
334 worktreePath: draftThread.worktreePath,333 worktreePath: draftThread.worktreePath,
335 turnDiffSummaries: [],334 turnDiffSummaries: [],
336 activities: [],335 activities: [],
336 proposedPlans: [],
337 };337 };
338}338}
339339
...@@ -904,11 +904,11 @@ export default function ChatView({ threadId }: ChatViewProps) {...@@ -904,11 +904,11 @@ export default function ChatView({ threadId }: ChatViewProps) {
904 if (!latestTurnSettled) {904 if (!latestTurnSettled) {
905 return null;905 return null;
906 }906 }
907 return findLatestProposedPlanMessage(907 return findLatestProposedPlan(
908 activeThread?.messages ?? [],908 activeThread?.proposedPlans ?? [],
909 activeLatestTurn?.assistantMessageId ?? null,909 activeLatestTurn?.turnId ?? null,
910 );910 );
911 }, [activeLatestTurn?.assistantMessageId, activeThread?.messages, latestTurnSettled]);911 }, [activeLatestTurn?.turnId, activeThread?.proposedPlans, latestTurnSettled]);
912 const activePlan = useMemo(912 const activePlan = useMemo(
913 () => deriveActivePlanState(threadActivities, activeLatestTurn?.turnId ?? undefined),913 () => deriveActivePlanState(threadActivities, activeLatestTurn?.turnId ?? undefined),
914 [activeLatestTurn?.turnId, threadActivities],914 [activeLatestTurn?.turnId, threadActivities],
...@@ -1055,8 +1055,8 @@ export default function ChatView({ threadId }: ChatViewProps) {...@@ -1055,8 +1055,8 @@ export default function ChatView({ threadId }: ChatViewProps) {
1055 return [...serverMessagesWithPreviewHandoff, ...pendingMessages];1055 return [...serverMessagesWithPreviewHandoff, ...pendingMessages];
1056 }, [serverMessages, attachmentPreviewHandoffByMessageId, optimisticUserMessages]);1056 }, [serverMessages, attachmentPreviewHandoffByMessageId, optimisticUserMessages]);
1057 const timelineEntries = useMemo(1057 const timelineEntries = useMemo(
1058 () => deriveTimelineEntries(timelineMessages, workLogEntries),1058 () => deriveTimelineEntries(timelineMessages, activeThread?.proposedPlans ?? [], workLogEntries),
1059 [timelineMessages, workLogEntries],1059 [activeThread?.proposedPlans, timelineMessages, workLogEntries],
1060 );1060 );
1061 const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } =1061 const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } =
1062 useTurnDiffSummaries(activeThread);1062 useTurnDiffSummaries(activeThread);
...@@ -2279,8 +2279,8 @@ export default function ChatView({ threadId }: ChatViewProps) {...@@ -2279,8 +2279,8 @@ export default function ChatView({ threadId }: ChatViewProps) {
2279 if (showPlanFollowUpPrompt && activeProposedPlan) {2279 if (showPlanFollowUpPrompt && activeProposedPlan) {
2280 const followUpText =2280 const followUpText =
2281 trimmed.length > 02281 trimmed.length > 0
2282 ? trimmed2282 ? trimmed
2283 : buildPlanImplementationPrompt(activeProposedPlan.plan.planMarkdown);2283 : buildPlanImplementationPrompt(activeProposedPlan.planMarkdown);
2284 const nextInteractionMode = trimmed.length > 0 ? "plan" : "default";2284 const nextInteractionMode = trimmed.length > 0 ? "plan" : "default";
2285 promptRef.current = "";2285 promptRef.current = "";
2286 clearComposerDraftContent(activeThread.id);2286 clearComposerDraftContent(activeThread.id);
...@@ -2815,7 +2815,7 @@ export default function ChatView({ threadId }: ChatViewProps) {...@@ -2815,7 +2815,7 @@ export default function ChatView({ threadId }: ChatViewProps) {
28152815
2816 const createdAt = new Date().toISOString();2816 const createdAt = new Date().toISOString();
2817 const nextThreadId = newThreadId();2817 const nextThreadId = newThreadId();
2818 const planMarkdown = activeProposedPlan.plan.planMarkdown;2818 const planMarkdown = activeProposedPlan.planMarkdown;
2819 const implementationPrompt = buildPlanImplementationPrompt(planMarkdown);2819 const implementationPrompt = buildPlanImplementationPrompt(planMarkdown);
2820 const nextThreadTitle = truncateTitle(buildPlanImplementationThreadTitle(planMarkdown));2820 const nextThreadTitle = truncateTitle(buildPlanImplementationThreadTitle(planMarkdown));
2821 const nextThreadModel: ModelSlug =2821 const nextThreadModel: ModelSlug =
...@@ -3334,7 +3334,7 @@ export default function ChatView({ threadId }: ChatViewProps) {...@@ -3334,7 +3334,7 @@ export default function ChatView({ threadId }: ChatViewProps) {
3334 >3334 >
3335 <MessagesTimeline3335 <MessagesTimeline
3336 key={activeThread.id}3336 key={activeThread.id}
3337 hasMessages={timelineMessages.length > 0}3337 hasMessages={timelineEntries.length > 0}
3338 isWorking={isWorking}3338 isWorking={isWorking}
3339 activeTurnInProgress={!latestTurnSettled}3339 activeTurnInProgress={!latestTurnSettled}
3340 activeTurnStartedAt={activeLatestTurn?.startedAt ?? null}3340 activeTurnStartedAt={activeLatestTurn?.startedAt ?? null}
...@@ -3387,8 +3387,8 @@ export default function ChatView({ threadId }: ChatViewProps) {...@@ -3387,8 +3387,8 @@ export default function ChatView({ threadId }: ChatViewProps) {
3387 ) : showPlanFollowUpPrompt && activeProposedPlan ? (3387 ) : showPlanFollowUpPrompt && activeProposedPlan ? (
3388 <div className="rounded-t-[19px] border-b border-border/65 bg-muted/20">3388 <div className="rounded-t-[19px] border-b border-border/65 bg-muted/20">
3389 <ComposerPlanFollowUpBanner3389 <ComposerPlanFollowUpBanner
3390 key={activeProposedPlan.message.id}3390 key={activeProposedPlan.id}
3391 planTitle={proposedPlanTitle(activeProposedPlan.plan.planMarkdown) ?? null}3391 planTitle={proposedPlanTitle(activeProposedPlan.planMarkdown) ?? null}
3392 />3392 />
3393 </div>3393 </div>
3394 ) : null}3394 ) : null}
...@@ -4205,7 +4205,6 @@ const ComposerPendingUserInputPanel = memo(function ComposerPendingUserInputPane...@@ -4205,7 +4205,6 @@ const ComposerPendingUserInputPanel = memo(function ComposerPendingUserInputPane
4205 <ComposerPendingUserInputCard4205 <ComposerPendingUserInputCard
4206 key={activePrompt.requestId}4206 key={activePrompt.requestId}
4207 prompt={activePrompt}4207 prompt={activePrompt}
4208 pendingPromptCount={pendingUserInputs.length}
4209 isResponding={respondingRequestIds.includes(activePrompt.requestId)}4208 isResponding={respondingRequestIds.includes(activePrompt.requestId)}
4210 answers={answers}4209 answers={answers}
4211 questionIndex={questionIndex}4210 questionIndex={questionIndex}
...@@ -4216,14 +4215,12 @@ const ComposerPendingUserInputPanel = memo(function ComposerPendingUserInputPane...@@ -4216,14 +4215,12 @@ const ComposerPendingUserInputPanel = memo(function ComposerPendingUserInputPane
42164215
4217const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard({4216const ComposerPendingUserInputCard = memo(function ComposerPendingUserInputCard({
4218 prompt,4217 prompt,
4219 pendingPromptCount,
4220 isResponding,4218 isResponding,
4221 answers,4219 answers,
4222 questionIndex,4220 questionIndex,
4223 onSelectOption,4221 onSelectOption,
4224}: {4222}: {
4225 prompt: PendingUserInput;4223 prompt: PendingUserInput;
4226 pendingPromptCount: number;
4227 isResponding: boolean;4224 isResponding: boolean;
4228 answers: Record<string, PendingUserInputDraftAnswer>;4225 answers: Record<string, PendingUserInputDraftAnswer>;
4229 questionIndex: number;4226 questionIndex: number;
...@@ -4448,33 +4445,29 @@ const ChangedFilesTree = memo(function ChangedFilesTree(props: {...@@ -4448,33 +4445,29 @@ const ChangedFilesTree = memo(function ChangedFilesTree(props: {
4448 return (4445 return (
4449 <div className="space-y-0.5">4446 <div className="space-y-0.5">
4450 {treeNodes.map((node) => renderTreeNode(node, 0))}4447 {treeNodes.map((node) => renderTreeNode(node, 0))}
4451const ProposedPlanMessage = memo(function ProposedPlanMessage({4448 </div>
4452 text,4449 );
4450});
4451
4452const ProposedPlanCard = memo(function ProposedPlanCard({
4453 planMarkdown,
4453 cwd,4454 cwd,
4454 workspaceRoot,4455 workspaceRoot,
4455 isStreaming,
4456}: {4456}: {
4457 text: string;4457 planMarkdown: string;
4458 cwd: string | undefined;4458 cwd: string | undefined;
4459 workspaceRoot: string | undefined;4459 workspaceRoot: string | undefined;
4460 isStreaming: boolean;
4461}) {4460}) {
4462 const proposedPlan = parseProposedPlanMessage(text);
4463 const [expanded, setExpanded] = useState(false);4461 const [expanded, setExpanded] = useState(false);
4464 const [isSaveDialogOpen, setIsSaveDialogOpen] = useState(false);4462 const [isSaveDialogOpen, setIsSaveDialogOpen] = useState(false);
4465 const [savePath, setSavePath] = useState("");4463 const [savePath, setSavePath] = useState("");
4466 const [isSavingToWorkspace, setIsSavingToWorkspace] = useState(false);4464 const [isSavingToWorkspace, setIsSavingToWorkspace] = useState(false);
4467 const savePathInputId = useId();4465 const savePathInputId = useId();
44684466 const title = proposedPlanTitle(planMarkdown) ?? "Proposed plan";
4469 if (!proposedPlan) {4467 const lineCount = planMarkdown.split("\n").length;
4470 return <ChatMarkdown text={text} cwd={cwd} isStreaming={isStreaming} />;4468 const canCollapse = planMarkdown.length > 900 || lineCount > 20;
4471 }4469 const downloadFilename = buildProposedPlanMarkdownFilename(planMarkdown);
44724470 const saveContents = normalizePlanMarkdownForExport(planMarkdown);
4473 const title = proposedPlanTitle(proposedPlan.planMarkdown) ?? "Proposed plan";
4474 const lineCount = proposedPlan.planMarkdown.split("\n").length;
4475 const canCollapse = proposedPlan.planMarkdown.length > 900 || lineCount > 20;
4476 const downloadFilename = buildProposedPlanMarkdownFilename(proposedPlan.planMarkdown);
4477 const saveContents = normalizePlanMarkdownForExport(proposedPlan.planMarkdown);
44784471
4479 const handleDownload = () => {4472 const handleDownload = () => {
4480 downloadTextFile(downloadFilename, saveContents);4473 downloadTextFile(downloadFilename, saveContents);
...@@ -4540,46 +4533,40 @@ const ProposedPlanMessage = memo(function ProposedPlanMessage({...@@ -4540,46 +4533,40 @@ const ProposedPlanMessage = memo(function ProposedPlanMessage({
4540 };4533 };
45414534
4542 return (4535 return (
4543 <div className="space-y-3">4536 <div className="rounded-[24px] border border-border/80 bg-card/70 p-4 sm:p-5">
4544 {proposedPlan.beforeText ? (4537 <div className="flex flex-wrap items-center justify-between gap-3">
4545 <ChatMarkdown text={proposedPlan.beforeText} cwd={cwd} isStreaming={isStreaming} />4538 <div className="flex min-w-0 items-center gap-2">
4546 ) : null}4539 <Badge variant="secondary">Plan</Badge>
45474540 <p className="truncate text-sm font-medium text-foreground">{title}</p>
4548 <div className="rounded-[24px] border border-border/80 bg-card/70 p-4 sm:p-5">
4549 <div className="flex flex-wrap items-center justify-between gap-3">
4550 <div className="flex min-w-0 items-center gap-2">
4551 <Badge variant="secondary">Plan</Badge>
4552 <p className="truncate text-sm font-medium text-foreground">{title}</p>
4553 </div>
4554 <Menu>
4555 <MenuTrigger
4556 render={<Button aria-label="Plan actions" size="icon-xs" variant="outline" />}
4557 >
4558 <EllipsisIcon aria-hidden="true" className="size-4" />
4559 </MenuTrigger>
4560 <MenuPopup align="end">
4561 <MenuItem onClick={handleDownload}>Download as markdown</MenuItem>
4562 <MenuItem onClick={openSaveDialog} disabled={!workspaceRoot || isSavingToWorkspace}>
4563 Save to workspace
4564 </MenuItem>
4565 </MenuPopup>
4566 </Menu>
4567 </div>4541 </div>
4568 <div className="mt-4">4542 <Menu>
4569 <div className={cn("relative", canCollapse && !expanded && "max-h-104 overflow-hidden")}>4543 <MenuTrigger
4570 <ChatMarkdown text={proposedPlan.planMarkdown} cwd={cwd} isStreaming={false} />4544 render={<Button aria-label="Plan actions" size="icon-xs" variant="outline" />}
4571 {canCollapse && !expanded ? (4545 >
4572 <div className="pointer-events-none absolute inset-x-0 bottom-0 h-24 bg-linear-to-t from-card/95 via-card/80 to-transparent" />4546 <EllipsisIcon aria-hidden="true" className="size-4" />
4573 ) : null}4547 </MenuTrigger>
4574 </div>4548 <MenuPopup align="end">
4575 {canCollapse ? (4549 <MenuItem onClick={handleDownload}>Download as markdown</MenuItem>
4576 <div className="mt-4 flex justify-center">4550 <MenuItem onClick={openSaveDialog} disabled={!workspaceRoot || isSavingToWorkspace}>
4577 <Button size="sm" variant="outline" onClick={() => setExpanded((value) => !value)}>4551 Save to workspace
4578 {expanded ? "Collapse plan" : "Expand plan"}4552 </MenuItem>
4579 </Button>4553 </MenuPopup>
4580 </div>4554 </Menu>
4555 </div>
4556 <div className="mt-4">
4557 <div className={cn("relative", canCollapse && !expanded && "max-h-104 overflow-hidden")}>
4558 <ChatMarkdown text={planMarkdown} cwd={cwd} isStreaming={false} />
4559 {canCollapse && !expanded ? (
4560 <div className="pointer-events-none absolute inset-x-0 bottom-0 h-24 bg-linear-to-t from-card/95 via-card/80 to-transparent" />
4581 ) : null}4561 ) : null}
4582 </div>4562 </div>
4563 {canCollapse ? (
4564 <div className="mt-4 flex justify-center">
4565 <Button size="sm" variant="outline" onClick={() => setExpanded((value) => !value)}>
4566 {expanded ? "Collapse plan" : "Expand plan"}
4567 </Button>
4568 </div>
4569 ) : null}
4583 </div>4570 </div>
45844571
4585 <Dialog4572 <Dialog
...@@ -4629,10 +4616,6 @@ const ProposedPlanMessage = memo(function ProposedPlanMessage({...@@ -4629,10 +4616,6 @@ const ProposedPlanMessage = memo(function ProposedPlanMessage({
4629 </DialogFooter>4616 </DialogFooter>
4630 </DialogPopup>4617 </DialogPopup>
4631 </Dialog>4618 </Dialog>
4632
4633 {proposedPlan.afterText ? (
4634 <ChatMarkdown text={proposedPlan.afterText} cwd={cwd} isStreaming={isStreaming} />
4635 ) : null}
4636 </div>4619 </div>
4637 );4620 );
4638});4621});
...@@ -4662,6 +4645,7 @@ interface MessagesTimelineProps {...@@ -4662,6 +4645,7 @@ interface MessagesTimelineProps {
46624645
4663type TimelineEntry = ReturnType<typeof deriveTimelineEntries>[number];4646type TimelineEntry = ReturnType<typeof deriveTimelineEntries>[number];
4664type TimelineMessage = Extract<TimelineEntry, { kind: "message" }>["message"];4647type TimelineMessage = Extract<TimelineEntry, { kind: "message" }>["message"];
4648type TimelineProposedPlan = Extract<TimelineEntry, { kind: "proposed-plan" }>["proposedPlan"];
4665type TimelineWorkEntry = Extract<TimelineEntry, { kind: "work" }>["entry"];4649type TimelineWorkEntry = Extract<TimelineEntry, { kind: "work" }>["entry"];
4666type TimelineRow =4650type TimelineRow =
4667 | {4651 | {
...@@ -4677,8 +4661,33 @@ type TimelineRow =...@@ -4677,8 +4661,33 @@ type TimelineRow =
4677 message: TimelineMessage;4661 message: TimelineMessage;
4678 showCompletionDivider: boolean;4662 showCompletionDivider: boolean;
4679 }4663 }
4664 | {
4665 kind: "proposed-plan";
4666 id: string;
4667 createdAt: string;
4668 proposedPlan: TimelineProposedPlan;
4669 }
4680 | { kind: "working"; id: string; createdAt: string | null };4670 | { kind: "working"; id: string; createdAt: string | null };
46814671
4672function estimateTimelineMessageHeight(message: TimelineMessage): number {
4673 const textLength = message.text.length;
4674 if (message.role === "assistant") {
4675 const estimatedLines = Math.max(1, Math.ceil(textLength / 72));
4676 return 78 + Math.min(estimatedLines * 22, 820);
4677 }
4678
4679 const estimatedLines = Math.max(1, Math.ceil(textLength / 56));
4680 const attachmentCount = message.attachments?.length ?? 0;
4681 const attachmentRows = Math.ceil(attachmentCount / 2);
4682 const attachmentHeight = attachmentRows * 124;
4683 return 96 + Math.min(estimatedLines * 22, 620) + attachmentHeight;
4684}
4685
4686function estimateTimelineProposedPlanHeight(proposedPlan: TimelineProposedPlan): number {
4687 const estimatedLines = Math.max(1, Math.ceil(proposedPlan.planMarkdown.length / 72));
4688 return 120 + Math.min(estimatedLines * 22, 880);
4689}
4690
4682const MessagesTimeline = memo(function MessagesTimeline({4691const MessagesTimeline = memo(function MessagesTimeline({
4683 hasMessages,4692 hasMessages,
4684 isWorking,4693 isWorking,
...@@ -4757,6 +4766,16 @@ const MessagesTimeline = memo(function MessagesTimeline({...@@ -4757,6 +4766,16 @@ const MessagesTimeline = memo(function MessagesTimeline({
4757 continue;4766 continue;
4758 }4767 }
47594768
4769 if (timelineEntry.kind === "proposed-plan") {
4770 nextRows.push({
4771 kind: "proposed-plan",
4772 id: timelineEntry.id,
4773 createdAt: timelineEntry.createdAt,
4774 proposedPlan: timelineEntry.proposedPlan,
4775 });
4776 continue;
4777 }
4778
4760 nextRows.push({4779 nextRows.push({
4761 kind: "message",4780 kind: "message",
4762 id: timelineEntry.id,4781 id: timelineEntry.id,
...@@ -4831,6 +4850,7 @@ const MessagesTimeline = memo(function MessagesTimeline({...@@ -4831,6 +4850,7 @@ const MessagesTimeline = memo(function MessagesTimeline({
4831 const row = rows[index];4850 const row = rows[index];
4832 if (!row) return 96;4851 if (!row) return 96;
4833 if (row.kind === "work") return 112;4852 if (row.kind === "work") return 112;
4853 if (row.kind === "proposed-plan") return estimateTimelineProposedPlanHeight(row.proposedPlan);
4834 if (row.kind === "working") return 40;4854 if (row.kind === "working") return 40;
4835 return estimateTimelineMessageHeight(row.message, { timelineWidthPx });4855 return estimateTimelineMessageHeight(row.message, { timelineWidthPx });
4836 },4856 },
...@@ -5044,10 +5064,9 @@ const MessagesTimeline = memo(function MessagesTimeline({...@@ -5044,10 +5064,9 @@ const MessagesTimeline = memo(function MessagesTimeline({
5044 </div>5064 </div>
5045 )}5065 )}
5046 <div className="min-w-0 px-1 py-0.5">5066 <div className="min-w-0 px-1 py-0.5">
5047 <ProposedPlanMessage5067 <ChatMarkdown
5048 text={messageText}5068 text={messageText}
5049 cwd={markdownCwd}5069 cwd={markdownCwd}
5050 workspaceRoot={workspaceRoot}
5051 isStreaming={Boolean(row.message.streaming)}5070 isStreaming={Boolean(row.message.streaming)}
5052 />5071 />
5053 {(() => {5072 {(() => {
...@@ -5119,6 +5138,16 @@ const MessagesTimeline = memo(function MessagesTimeline({...@@ -5119,6 +5138,16 @@ const MessagesTimeline = memo(function MessagesTimeline({
5119 );5138 );
5120 })()}5139 })()}
51215140
5141 {row.kind === "proposed-plan" && (
5142 <div className="min-w-0 px-1 py-0.5">
5143 <ProposedPlanCard
5144 planMarkdown={row.proposedPlan.planMarkdown}
5145 cwd={markdownCwd}
5146 workspaceRoot={workspaceRoot}
5147 />
5148 </div>
5149 )}
5150
5122 {row.kind === "working" && (5151 {row.kind === "working" && (
5123 <div className="flex items-center gap-2 py-0.5 pl-1.5">5152 <div className="flex items-center gap-2 py-0.5 pl-1.5">
5124 <span className="h-1.5 w-1.5 shrink-0 rounded-full bg-muted-foreground/30" />5153 <span className="h-1.5 w-1.5 shrink-0 rounded-full bg-muted-foreground/30" />
apps/web/src/proposedPlan.test.ts-73
...@@ -1,85 +1,12 @@...@@ -1,85 +1,12 @@
1import { MessageId } from "@t3tools/contracts";
2import { describe, expect, it } from "vitest";1import { describe, expect, it } from "vitest";
32
4import {3import {
5 buildPlanImplementationThreadTitle,4 buildPlanImplementationThreadTitle,
6 buildPlanImplementationPrompt,5 buildPlanImplementationPrompt,
7 buildProposedPlanMarkdownFilename,6 buildProposedPlanMarkdownFilename,
8 findLatestProposedPlanMessage,
9 parseProposedPlanMessage,
10 proposedPlanTitle,7 proposedPlanTitle,
11} from "./proposedPlan";8} from "./proposedPlan";
129
13describe("parseProposedPlanMessage", () => {
14 it("extracts the proposed plan block and surrounding assistant text", () => {
15 expect(
16 parseProposedPlanMessage(
17 "First I checked the repo.\n\n<proposed_plan>\n# Ship plan\n\n- step 1\n</proposed_plan>\n\nImplement this plan?",
18 ),
19 ).toEqual({
20 beforeText: "First I checked the repo.",
21 planMarkdown: "# Ship plan\n\n- step 1",
22 afterText: "Implement this plan?",
23 });
24 });
25
26 it("returns null when the assistant message has no plan block", () => {
27 expect(parseProposedPlanMessage("Plain assistant text")).toBeNull();
28 });
29});
30
31describe("findLatestProposedPlanMessage", () => {
32 it("prefers the latest turn assistant message id when it contains a proposed plan", () => {
33 const match = findLatestProposedPlanMessage(
34 [
35 {
36 id: MessageId.makeUnsafe("assistant:older-plan"),
37 role: "assistant",
38 text: "<proposed_plan>\n# Older plan\n</proposed_plan>",
39 createdAt: "2026-03-05T00:00:00.000Z",
40 streaming: false,
41 },
42 {
43 id: MessageId.makeUnsafe("assistant:latest-plan"),
44 role: "assistant",
45 text: "<proposed_plan>\n# Latest plan\n</proposed_plan>",
46 createdAt: "2026-03-05T00:01:00.000Z",
47 streaming: false,
48 },
49 ],
50 "assistant:latest-plan",
51 );
52
53 expect(match?.message.id).toBe("assistant:latest-plan");
54 expect(match?.plan.planMarkdown).toBe("# Latest plan");
55 });
56
57 it("falls back to the latest assistant proposed plan when no latest-turn message id is available", () => {
58 const match = findLatestProposedPlanMessage(
59 [
60 {
61 id: MessageId.makeUnsafe("assistant:plain"),
62 role: "assistant",
63 text: "No plan here",
64 createdAt: "2026-03-05T00:00:00.000Z",
65 streaming: false,
66 },
67 {
68 id: MessageId.makeUnsafe("assistant:plan"),
69 role: "assistant",
70 text: "<proposed_plan>\n# Fallback plan\n</proposed_plan>",
71 createdAt: "2026-03-05T00:01:00.000Z",
72 streaming: false,
73 },
74 ],
75 null,
76 );
77
78 expect(match?.message.id).toBe("assistant:plan");
79 expect(match?.plan.planMarkdown).toBe("# Fallback plan");
80 });
81});
82
83describe("proposedPlanTitle", () => {10describe("proposedPlanTitle", () => {
84 it("reads the first markdown heading as the plan title", () => {11 it("reads the first markdown heading as the plan title", () => {
85 expect(proposedPlanTitle("# Integrate RPC\n\nBody")).toBe("Integrate RPC");12 expect(proposedPlanTitle("# Integrate RPC\n\nBody")).toBe("Integrate RPC");
apps/web/src/proposedPlan.ts-76
...@@ -1,42 +1,3 @@...@@ -1,42 +1,3 @@
1import type { MessageId } from "@t3tools/contracts";
2
3import type { ChatMessage } from "./types";
4
5export interface ProposedPlanBlock {
6 beforeText: string;
7 planMarkdown: string;
8 afterText: string;
9}
10
11export interface ProposedPlanMessageMatch {
12 message: ChatMessage;
13 plan: ProposedPlanBlock;
14}
15
16const PROPOSED_PLAN_REGEX = /<proposed_plan>\s*([\s\S]*?)\s*<\/proposed_plan>/;
17
18export function parseProposedPlanMessage(text: string): ProposedPlanBlock | null {
19 const match = PROPOSED_PLAN_REGEX.exec(text);
20 if (!match) {
21 return null;
22 }
23
24 const fullMatch = match[0];
25 const content = match[1];
26 const start = match.index;
27 const end = start + fullMatch.length;
28 const planMarkdown = content?.trim();
29 if (!planMarkdown) {
30 return null;
31 }
32
33 return {
34 beforeText: text.slice(0, start).trim(),
35 planMarkdown,
36 afterText: text.slice(end).trim(),
37 };
38}
39
40export function proposedPlanTitle(planMarkdown: string): string | null {1export function proposedPlanTitle(planMarkdown: string): string | null {
41 const heading = planMarkdown.match(/^\s{0,3}#{1,6}\s+(.+)$/m)?.[1]?.trim();2 const heading = planMarkdown.match(/^\s{0,3}#{1,6}\s+(.+)$/m)?.[1]?.trim();
42 return heading && heading.length > 0 ? heading : null;3 return heading && heading.length > 0 ? heading : null;
...@@ -51,43 +12,6 @@ function sanitizePlanFileSegment(input: string): string {...@@ -51,43 +12,6 @@ function sanitizePlanFileSegment(input: string): string {
51 return sanitized.length > 0 ? sanitized : "plan";12 return sanitized.length > 0 ? sanitized : "plan";
52}13}
5314
54export function findLatestProposedPlanMessage(
55 messages: readonly ChatMessage[],
56 assistantMessageId: MessageId | string | null | undefined,
57): ProposedPlanMessageMatch | null {
58 if (assistantMessageId) {
59 const matchingMessage = messages.find(
60 (message) => message.id === assistantMessageId && message.role === "assistant",
61 );
62 if (matchingMessage) {
63 const parsedPlan = parseProposedPlanMessage(matchingMessage.text);
64 if (parsedPlan) {
65 return {
66 message: matchingMessage,
67 plan: parsedPlan,
68 };
69 }
70 }
71 }
72
73 for (let index = messages.length - 1; index >= 0; index -= 1) {
74 const message = messages[index];
75 if (!message || message.role !== "assistant") {
76 continue;
77 }
78 const parsedPlan = parseProposedPlanMessage(message.text);
79 if (!parsedPlan) {
80 continue;
81 }
82 return {
83 message,
84 plan: parsedPlan,
85 };
86 }
87
88 return null;
89}
90
91export function buildPlanImplementationPrompt(planMarkdown: string): string {15export function buildPlanImplementationPrompt(planMarkdown: string): string {
92 return `PLEASE IMPLEMENT THIS PLAN:\n${planMarkdown.trim()}`;16 return `PLEASE IMPLEMENT THIS PLAN:\n${planMarkdown.trim()}`;
93}17}
apps/web/src/session-logic.test.ts+107-1
...@@ -1,4 +1,4 @@...@@ -1,4 +1,4 @@
1import { EventId, TurnId, type OrchestrationThreadActivity } from "@t3tools/contracts";1import { EventId, MessageId, TurnId, type OrchestrationThreadActivity } from "@t3tools/contracts";
2import { describe, expect, it } from "vitest";2import { describe, expect, it } from "vitest";
33
4import {4import {
...@@ -6,7 +6,9 @@ import {...@@ -6,7 +6,9 @@ import {
6 PROVIDER_OPTIONS,6 PROVIDER_OPTIONS,
7 derivePendingApprovals,7 derivePendingApprovals,
8 derivePendingUserInputs,8 derivePendingUserInputs,
9 deriveTimelineEntries,
9 deriveWorkLogEntries,10 deriveWorkLogEntries,
11 findLatestProposedPlan,
10 hasToolActivityForTurn,12 hasToolActivityForTurn,
11 isLatestTurnSettled,13 isLatestTurnSettled,
12} from "./session-logic";14} from "./session-logic";
...@@ -257,6 +259,69 @@ describe("deriveActivePlanState", () => {...@@ -257,6 +259,69 @@ describe("deriveActivePlanState", () => {
257 });259 });
258});260});
259261
262describe("findLatestProposedPlan", () => {
263 it("prefers the latest proposed plan for the active turn", () => {
264 expect(
265 findLatestProposedPlan(
266 [
267 {
268 id: "plan:thread-1:turn:turn-1",
269 turnId: TurnId.makeUnsafe("turn-1"),
270 planMarkdown: "# Older",
271 createdAt: "2026-02-23T00:00:01.000Z",
272 updatedAt: "2026-02-23T00:00:01.000Z",
273 },
274 {
275 id: "plan:thread-1:turn:turn-1",
276 turnId: TurnId.makeUnsafe("turn-1"),
277 planMarkdown: "# Latest",
278 createdAt: "2026-02-23T00:00:01.000Z",
279 updatedAt: "2026-02-23T00:00:02.000Z",
280 },
281 {
282 id: "plan:thread-1:turn:turn-2",
283 turnId: TurnId.makeUnsafe("turn-2"),
284 planMarkdown: "# Different turn",
285 createdAt: "2026-02-23T00:00:03.000Z",
286 updatedAt: "2026-02-23T00:00:03.000Z",
287 },
288 ],
289 TurnId.makeUnsafe("turn-1"),
290 ),
291 ).toEqual({
292 id: "plan:thread-1:turn:turn-1",
293 turnId: "turn-1",
294 planMarkdown: "# Latest",
295 createdAt: "2026-02-23T00:00:01.000Z",
296 updatedAt: "2026-02-23T00:00:02.000Z",
297 });
298 });
299
300 it("falls back to the most recently updated proposed plan", () => {
301 const latestPlan = findLatestProposedPlan(
302 [
303 {
304 id: "plan:thread-1:turn:turn-1",
305 turnId: TurnId.makeUnsafe("turn-1"),
306 planMarkdown: "# First",
307 createdAt: "2026-02-23T00:00:01.000Z",
308 updatedAt: "2026-02-23T00:00:01.000Z",
309 },
310 {
311 id: "plan:thread-1:turn:turn-2",
312 turnId: TurnId.makeUnsafe("turn-2"),
313 planMarkdown: "# Latest",
314 createdAt: "2026-02-23T00:00:02.000Z",
315 updatedAt: "2026-02-23T00:00:03.000Z",
316 },
317 ],
318 null,
319 );
320
321 expect(latestPlan?.planMarkdown).toBe("# Latest");
322 });
323});
324
260describe("deriveWorkLogEntries", () => {325describe("deriveWorkLogEntries", () => {
261 it("omits tool started entries and keeps completed entries", () => {326 it("omits tool started entries and keeps completed entries", () => {
262 const activities: OrchestrationThreadActivity[] = [327 const activities: OrchestrationThreadActivity[] = [
...@@ -338,6 +403,47 @@ describe("deriveWorkLogEntries", () => {...@@ -338,6 +403,47 @@ describe("deriveWorkLogEntries", () => {
338 });403 });
339});404});
340405
406describe("deriveTimelineEntries", () => {
407 it("includes proposed plans alongside messages and work entries in chronological order", () => {
408 const entries = deriveTimelineEntries(
409 [
410 {
411 id: MessageId.makeUnsafe("message-1"),
412 role: "assistant",
413 text: "hello",
414 createdAt: "2026-02-23T00:00:01.000Z",
415 streaming: false,
416 },
417 ],
418 [
419 {
420 id: "plan:thread-1:turn:turn-1",
421 turnId: TurnId.makeUnsafe("turn-1"),
422 planMarkdown: "# Ship it",
423 createdAt: "2026-02-23T00:00:02.000Z",
424 updatedAt: "2026-02-23T00:00:02.000Z",
425 },
426 ],
427 [
428 {
429 id: "work-1",
430 createdAt: "2026-02-23T00:00:03.000Z",
431 label: "Ran tests",
432 tone: "tool",
433 },
434 ],
435 );
436
437 expect(entries.map((entry) => entry.kind)).toEqual(["message", "proposed-plan", "work"]);
438 expect(entries[1]).toMatchObject({
439 kind: "proposed-plan",
440 proposedPlan: {
441 planMarkdown: "# Ship it",
442 },
443 });
444 });
445});
446
341describe("hasToolActivityForTurn", () => {447describe("hasToolActivityForTurn", () => {
342 it("returns false when turn id is missing", () => {448 it("returns false when turn id is missing", () => {
343 const activities: OrchestrationThreadActivity[] = [449 const activities: OrchestrationThreadActivity[] = [
apps/web/src/session-logic.ts+68-2
...@@ -2,12 +2,13 @@ import {...@@ -2,12 +2,13 @@ import {
2 ApprovalRequestId,2 ApprovalRequestId,
3 type OrchestrationLatestTurn,3 type OrchestrationLatestTurn,
4 type OrchestrationThreadActivity,4 type OrchestrationThreadActivity,
5 type OrchestrationProposedPlanId,
5 type ProviderKind,6 type ProviderKind,
6 type UserInputQuestion,7 type UserInputQuestion,
7 type TurnId,8 type TurnId,
8} from "@t3tools/contracts";9} from "@t3tools/contracts";
910
10import type { ChatMessage, SessionPhase, ThreadSession, TurnDiffSummary } from "./types";11import type { ChatMessage, ProposedPlan, SessionPhase, ThreadSession, TurnDiffSummary } from "./types";
1112
12export const PROVIDER_OPTIONS: Array<{13export const PROVIDER_OPTIONS: Array<{
13 value: ProviderKind;14 value: ProviderKind;
...@@ -50,6 +51,14 @@ export interface ActivePlanState {...@@ -50,6 +51,14 @@ export interface ActivePlanState {
50 }>;51 }>;
51}52}
5253
54export interface LatestProposedPlanState {
55 id: OrchestrationProposedPlanId;
56 createdAt: string;
57 updatedAt: string;
58 turnId: TurnId | null;
59 planMarkdown: string;
60}
61
53export type TimelineEntry =62export type TimelineEntry =
54 | {63 | {
55 id: string;64 id: string;
...@@ -57,6 +66,12 @@ export type TimelineEntry =...@@ -57,6 +66,12 @@ export type TimelineEntry =
57 createdAt: string;66 createdAt: string;
58 message: ChatMessage;67 message: ChatMessage;
59 }68 }
69 | {
70 id: string;
71 kind: "proposed-plan";
72 createdAt: string;
73 proposedPlan: ProposedPlan;
74 }
60 | {75 | {
61 id: string;76 id: string;
62 kind: "work";77 kind: "work";
...@@ -327,6 +342,48 @@ export function deriveActivePlanState(...@@ -327,6 +342,48 @@ export function deriveActivePlanState(
327 };342 };
328}343}
329344
345export function findLatestProposedPlan(
346 proposedPlans: ReadonlyArray<ProposedPlan>,
347 latestTurnId: TurnId | string | null | undefined,
348): LatestProposedPlanState | null {
349 if (latestTurnId) {
350 const matchingTurnPlan = [...proposedPlans]
351 .filter((proposedPlan) => proposedPlan.turnId === latestTurnId)
352 .toSorted(
353 (left, right) =>
354 left.updatedAt.localeCompare(right.updatedAt) || left.id.localeCompare(right.id),
355 )
356 .at(-1);
357 if (matchingTurnPlan) {
358 return {
359 id: matchingTurnPlan.id,
360 createdAt: matchingTurnPlan.createdAt,
361 updatedAt: matchingTurnPlan.updatedAt,
362 turnId: matchingTurnPlan.turnId,
363 planMarkdown: matchingTurnPlan.planMarkdown,
364 };
365 }
366 }
367
368 const latestPlan = [...proposedPlans]
369 .toSorted(
370 (left, right) =>
371 left.updatedAt.localeCompare(right.updatedAt) || left.id.localeCompare(right.id),
372 )
373 .at(-1);
374 if (!latestPlan) {
375 return null;
376 }
377
378 return {
379 id: latestPlan.id,
380 createdAt: latestPlan.createdAt,
381 updatedAt: latestPlan.updatedAt,
382 turnId: latestPlan.turnId,
383 planMarkdown: latestPlan.planMarkdown,
384 };
385}
386
330export function deriveWorkLogEntries(387export function deriveWorkLogEntries(
331 activities: ReadonlyArray<OrchestrationThreadActivity>,388 activities: ReadonlyArray<OrchestrationThreadActivity>,
332 latestTurnId: TurnId | undefined,389 latestTurnId: TurnId | undefined,
...@@ -381,6 +438,7 @@ export function hasToolActivityForTurn(...@@ -381,6 +438,7 @@ export function hasToolActivityForTurn(
381438
382export function deriveTimelineEntries(439export function deriveTimelineEntries(
383 messages: ChatMessage[],440 messages: ChatMessage[],
441 proposedPlans: ProposedPlan[],
384 workEntries: WorkLogEntry[],442 workEntries: WorkLogEntry[],
385): TimelineEntry[] {443): TimelineEntry[] {
386 const messageRows: TimelineEntry[] = messages.map((message) => ({444 const messageRows: TimelineEntry[] = messages.map((message) => ({
...@@ -389,13 +447,21 @@ export function deriveTimelineEntries(...@@ -389,13 +447,21 @@ export function deriveTimelineEntries(
389 createdAt: message.createdAt,447 createdAt: message.createdAt,
390 message,448 message,
391 }));449 }));
450 const proposedPlanRows: TimelineEntry[] = proposedPlans.map((proposedPlan) => ({
451 id: proposedPlan.id,
452 kind: "proposed-plan",
453 createdAt: proposedPlan.createdAt,
454 proposedPlan,
455 }));
392 const workRows: TimelineEntry[] = workEntries.map((entry) => ({456 const workRows: TimelineEntry[] = workEntries.map((entry) => ({
393 id: entry.id,457 id: entry.id,
394 kind: "work",458 kind: "work",
395 createdAt: entry.createdAt,459 createdAt: entry.createdAt,
396 entry,460 entry,
397 }));461 }));
398 return [...messageRows, ...workRows].toSorted((a, b) => a.createdAt.localeCompare(b.createdAt));462 return [...messageRows, ...proposedPlanRows, ...workRows].toSorted((a, b) =>
463 a.createdAt.localeCompare(b.createdAt),
464 );
399}465}
400466
401export function inferCheckpointTurnCountByTurnId(467export function inferCheckpointTurnCountByTurnId(
apps/web/src/store.test.ts+2
...@@ -17,6 +17,7 @@ function makeThread(overrides: Partial<Thread> = {}): Thread {...@@ -17,6 +17,7 @@ function makeThread(overrides: Partial<Thread> = {}): Thread {
17 messages: [],17 messages: [],
18 turnDiffSummaries: [],18 turnDiffSummaries: [],
19 activities: [],19 activities: [],
20 proposedPlans: [],
20 error: null,21 error: null,
21 createdAt: "2026-02-13T00:00:00.000Z",22 createdAt: "2026-02-13T00:00:00.000Z",
22 latestTurn: null,23 latestTurn: null,
...@@ -59,6 +60,7 @@ function makeReadModelThread(overrides: Partial<OrchestrationReadModel["threads"...@@ -59,6 +60,7 @@ function makeReadModelThread(overrides: Partial<OrchestrationReadModel["threads"
59 deletedAt: null,60 deletedAt: null,
60 messages: [],61 messages: [],
61 activities: [],62 activities: [],
63 proposedPlans: [],
62 checkpoints: [],64 checkpoints: [],
63 session: null,65 session: null,
64 ...overrides,66 ...overrides,
apps/web/src/store.ts+7
...@@ -289,6 +289,13 @@ export function syncServerReadModel(...@@ -289,6 +289,13 @@ export function syncServerReadModel(
289 };289 };
290 return normalizedMessage;290 return normalizedMessage;
291 }),291 }),
292 proposedPlans: thread.proposedPlans.map((proposedPlan) => ({
293 id: proposedPlan.id,
294 turnId: proposedPlan.turnId,
295 planMarkdown: proposedPlan.planMarkdown,
296 createdAt: proposedPlan.createdAt,
297 updatedAt: proposedPlan.updatedAt,
298 })),
292 error: thread.session?.lastError ?? null,299 error: thread.session?.lastError ?? null,
293 createdAt: thread.createdAt,300 createdAt: thread.createdAt,
294 latestTurn: thread.latestTurn,301 latestTurn: thread.latestTurn,
apps/web/src/types.ts+10
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import type {1import type {
2 OrchestrationLatestTurn,2 OrchestrationLatestTurn,
3 OrchestrationProposedPlanId,
3 OrchestrationSessionStatus,4 OrchestrationSessionStatus,
4 OrchestrationThreadActivity,5 OrchestrationThreadActivity,
5 ProjectScript as ContractProjectScript,6 ProjectScript as ContractProjectScript,
...@@ -48,6 +49,14 @@ export interface ChatMessage {...@@ -48,6 +49,14 @@ export interface ChatMessage {
48 streaming: boolean;49 streaming: boolean;
49}50}
5051
52export interface ProposedPlan {
53 id: OrchestrationProposedPlanId;
54 turnId: TurnId | null;
55 planMarkdown: string;
56 createdAt: string;
57 updatedAt: string;
58}
59
51export interface TurnDiffFileChange {60export interface TurnDiffFileChange {
52 path: string;61 path: string;
53 kind?: string | undefined;62 kind?: string | undefined;
...@@ -84,6 +93,7 @@ export interface Thread {...@@ -84,6 +93,7 @@ export interface Thread {
84 interactionMode: InteractionMode;93 interactionMode: InteractionMode;
85 session: ThreadSession | null;94 session: ThreadSession | null;
86 messages: ChatMessage[];95 messages: ChatMessage[];
96 proposedPlans: ProposedPlan[];
87 error: string | null;97 error: string | null;
88 createdAt: string;98 createdAt: string;
89 latestTurn: OrchestrationLatestTurn | null;99 latestTurn: OrchestrationLatestTurn | null;
apps/web/src/worktreeCleanup.test.ts+1
...@@ -17,6 +17,7 @@ function makeThread(overrides: Partial<Thread> = {}): Thread {...@@ -17,6 +17,7 @@ function makeThread(overrides: Partial<Thread> = {}): Thread {
17 messages: [],17 messages: [],
18 turnDiffSummaries: [],18 turnDiffSummaries: [],
19 activities: [],19 activities: [],
20 proposedPlans: [],
20 error: null,21 error: null,
21 createdAt: "2026-02-13T00:00:00.000Z",22 createdAt: "2026-02-13T00:00:00.000Z",
22 latestTurn: null,23 latestTurn: null,
packages/contracts/src/orchestration.ts+40
...@@ -136,6 +136,18 @@ export const OrchestrationMessage = Schema.Struct({...@@ -136,6 +136,18 @@ export const OrchestrationMessage = Schema.Struct({
136});136});
137export type OrchestrationMessage = typeof OrchestrationMessage.Type;137export type OrchestrationMessage = typeof OrchestrationMessage.Type;
138138
139export const OrchestrationProposedPlanId = TrimmedNonEmptyString;
140export type OrchestrationProposedPlanId = typeof OrchestrationProposedPlanId.Type;
141
142export const OrchestrationProposedPlan = Schema.Struct({
143 id: OrchestrationProposedPlanId,
144 turnId: Schema.NullOr(TurnId),
145 planMarkdown: TrimmedNonEmptyString,
146 createdAt: IsoDateTime,
147 updatedAt: IsoDateTime,
148});
149export type OrchestrationProposedPlan = typeof OrchestrationProposedPlan.Type;
150
139export const OrchestrationSessionStatus = Schema.Literals([151export const OrchestrationSessionStatus = Schema.Literals([
140 "idle",152 "idle",
141 "starting",153 "starting",
...@@ -234,6 +246,9 @@ export const OrchestrationThread = Schema.Struct({...@@ -234,6 +246,9 @@ export const OrchestrationThread = Schema.Struct({
234 updatedAt: IsoDateTime,246 updatedAt: IsoDateTime,
235 deletedAt: Schema.NullOr(IsoDateTime),247 deletedAt: Schema.NullOr(IsoDateTime),
236 messages: Schema.Array(OrchestrationMessage),248 messages: Schema.Array(OrchestrationMessage),
249 proposedPlans: Schema.Array(OrchestrationProposedPlan).pipe(
250 Schema.withDecodingDefault(() => []),
251 ),
237 activities: Schema.Array(OrchestrationThreadActivity),252 activities: Schema.Array(OrchestrationThreadActivity),
238 checkpoints: Schema.Array(OrchestrationCheckpointSummary),253 checkpoints: Schema.Array(OrchestrationCheckpointSummary),
239 session: Schema.NullOr(OrchestrationSession),254 session: Schema.NullOr(OrchestrationSession),
...@@ -467,6 +482,14 @@ export const ThreadMessageAssistantCompleteCommand = Schema.Struct({...@@ -467,6 +482,14 @@ export const ThreadMessageAssistantCompleteCommand = Schema.Struct({
467 createdAt: IsoDateTime,482 createdAt: IsoDateTime,
468});483});
469484
485export const ThreadProposedPlanUpsertCommand = Schema.Struct({
486 type: Schema.Literal("thread.proposed-plan.upsert"),
487 commandId: CommandId,
488 threadId: ThreadId,
489 proposedPlan: OrchestrationProposedPlan,
490 createdAt: IsoDateTime,
491});
492
470export const ThreadTurnDiffCompleteCommand = Schema.Struct({493export const ThreadTurnDiffCompleteCommand = Schema.Struct({
471 type: Schema.Literal("thread.turn.diff.complete"),494 type: Schema.Literal("thread.turn.diff.complete"),
472 commandId: CommandId,495 commandId: CommandId,
...@@ -501,6 +524,7 @@ export const InternalOrchestrationCommand = Schema.Union([...@@ -501,6 +524,7 @@ export const InternalOrchestrationCommand = Schema.Union([
501 ThreadSessionSetCommand,524 ThreadSessionSetCommand,
502 ThreadMessageAssistantDeltaCommand,525 ThreadMessageAssistantDeltaCommand,
503 ThreadMessageAssistantCompleteCommand,526 ThreadMessageAssistantCompleteCommand,
527 ThreadProposedPlanUpsertCommand,
504 ThreadTurnDiffCompleteCommand,528 ThreadTurnDiffCompleteCommand,
505 ThreadActivityAppendCommand,529 ThreadActivityAppendCommand,
506 ThreadRevertCompleteCommand,530 ThreadRevertCompleteCommand,
...@@ -531,6 +555,7 @@ export const OrchestrationEventType = Schema.Literals([...@@ -531,6 +555,7 @@ export const OrchestrationEventType = Schema.Literals([
531 "thread.reverted",555 "thread.reverted",
532 "thread.session-stop-requested",556 "thread.session-stop-requested",
533 "thread.session-set",557 "thread.session-set",
558 "thread.proposed-plan-upserted",
534 "thread.turn-diff-completed",559 "thread.turn-diff-completed",
535 "thread.activity-appended",560 "thread.activity-appended",
536]);561]);
...@@ -674,6 +699,11 @@ export const ThreadSessionSetPayload = Schema.Struct({...@@ -674,6 +699,11 @@ export const ThreadSessionSetPayload = Schema.Struct({
674 session: OrchestrationSession,699 session: OrchestrationSession,
675});700});
676701
702export const ThreadProposedPlanUpsertedPayload = Schema.Struct({
703 threadId: ThreadId,
704 proposedPlan: OrchestrationProposedPlan,
705});
706
677export const ThreadTurnDiffCompletedPayload = Schema.Struct({707export const ThreadTurnDiffCompletedPayload = Schema.Struct({
678 threadId: ThreadId,708 threadId: ThreadId,
679 turnId: TurnId,709 turnId: TurnId,
...@@ -811,6 +841,11 @@ export const OrchestrationEvent = Schema.Union([...@@ -811,6 +841,11 @@ export const OrchestrationEvent = Schema.Union([
811 type: Schema.Literal("thread.session-set"),841 type: Schema.Literal("thread.session-set"),
812 payload: ThreadSessionSetPayload,842 payload: ThreadSessionSetPayload,
813 }),843 }),
844 Schema.Struct({
845 ...EventBaseFields,
846 type: Schema.Literal("thread.proposed-plan-upserted"),
847 payload: ThreadProposedPlanUpsertedPayload,
848 }),
814 Schema.Struct({849 Schema.Struct({
815 ...EventBaseFields,850 ...EventBaseFields,
816 type: Schema.Literal("thread.turn-diff-completed"),851 type: Schema.Literal("thread.turn-diff-completed"),
...@@ -910,6 +945,11 @@ export const OrchestrationPersistedEvent = Schema.Union([...@@ -910,6 +945,11 @@ export const OrchestrationPersistedEvent = Schema.Union([
910 eventType: Schema.Literal("thread.session-set"),945 eventType: Schema.Literal("thread.session-set"),
911 payload: ThreadSessionSetPayload,946 payload: ThreadSessionSetPayload,
912 }),947 }),
948 Schema.Struct({
949 ...PersistedEventBaseFields,
950 eventType: Schema.Literal("thread.proposed-plan-upserted"),
951 payload: ThreadProposedPlanUpsertedPayload,
952 }),
913 Schema.Struct({953 Schema.Struct({
914 ...PersistedEventBaseFields,954 ...PersistedEventBaseFields,
915 eventType: Schema.Literal("thread.turn-diff-completed"),955 eventType: Schema.Literal("thread.turn-diff-completed"),
packages/contracts/src/providerRuntime.test.ts+22
...@@ -32,6 +32,26 @@ describe("ProviderRuntimeEvent", () => {...@@ -32,6 +32,26 @@ describe("ProviderRuntimeEvent", () => {
32 expect(parsed.payload.plan[1]?.status).toBe("inProgress");32 expect(parsed.payload.plan[1]?.status).toBe("inProgress");
33 });33 });
3434
35 it("decodes proposed-plan completion events", () => {
36 const parsed = decodeRuntimeEvent({
37 type: "turn.proposed.completed",
38 eventId: "event-proposed-plan-1",
39 provider: "codex",
40 createdAt: "2026-02-28T00:00:00.000Z",
41 threadId: "thread-1",
42 turnId: "turn-1",
43 payload: {
44 planMarkdown: "# Ship it",
45 },
46 });
47
48 expect(parsed.type).toBe("turn.proposed.completed");
49 if (parsed.type !== "turn.proposed.completed") {
50 throw new Error("expected turn.proposed.completed");
51 }
52 expect(parsed.payload.planMarkdown).toBe("# Ship it");
53 });
54
35 it("decodes user-input.requested with structured questions", () => {55 it("decodes user-input.requested with structured questions", () => {
36 const parsed = decodeRuntimeEvent({56 const parsed = decodeRuntimeEvent({
37 type: "user-input.requested",57 type: "user-input.requested",
...@@ -39,6 +59,7 @@ describe("ProviderRuntimeEvent", () => {...@@ -39,6 +59,7 @@ describe("ProviderRuntimeEvent", () => {
39 provider: "claudeCode",59 provider: "claudeCode",
40 sessionId: "runtime-session-2",60 sessionId: "runtime-session-2",
41 createdAt: "2026-02-28T00:00:01.000Z",61 createdAt: "2026-02-28T00:00:01.000Z",
62 threadId: "thread-2",
42 requestId: "request-1",63 requestId: "request-1",
43 payload: {64 payload: {
44 questions: [65 questions: [
...@@ -76,6 +97,7 @@ describe("ProviderRuntimeEvent", () => {...@@ -76,6 +97,7 @@ describe("ProviderRuntimeEvent", () => {
76 provider: "claudeCode",97 provider: "claudeCode",
77 sessionId: "runtime-session-2",98 sessionId: "runtime-session-2",
78 createdAt: "2026-02-28T00:00:02.000Z",99 createdAt: "2026-02-28T00:00:02.000Z",
100 threadId: "thread-2",
79 requestId: "request-1",101 requestId: "request-1",
80 payload: {102 payload: {
81 answers: {103 answers: {
packages/contracts/src/providerRuntime.ts+32
...@@ -154,6 +154,8 @@ export const ProviderRuntimeEventType = Schema.Literals([...@@ -154,6 +154,8 @@ export const ProviderRuntimeEventType = Schema.Literals([
154 "turn.completed",154 "turn.completed",
155 "turn.aborted",155 "turn.aborted",
156 "turn.plan.updated",156 "turn.plan.updated",
157 "turn.proposed.delta",
158 "turn.proposed.completed",
157 "turn.diff.updated",159 "turn.diff.updated",
158 "item.started",160 "item.started",
159 "item.updated",161 "item.updated",
...@@ -202,6 +204,8 @@ export const TurnStartedType = Schema.Literal("turn.started");...@@ -202,6 +204,8 @@ export const TurnStartedType = Schema.Literal("turn.started");
202export const TurnCompletedType = Schema.Literal("turn.completed");204export const TurnCompletedType = Schema.Literal("turn.completed");
203export const TurnAbortedType = Schema.Literal("turn.aborted");205export const TurnAbortedType = Schema.Literal("turn.aborted");
204export const TurnPlanUpdatedType = Schema.Literal("turn.plan.updated");206export const TurnPlanUpdatedType = Schema.Literal("turn.plan.updated");
207export const TurnProposedDeltaType = Schema.Literal("turn.proposed.delta");
208export const TurnProposedCompletedType = Schema.Literal("turn.proposed.completed");
205export const TurnDiffUpdatedType = Schema.Literal("turn.diff.updated");209export const TurnDiffUpdatedType = Schema.Literal("turn.diff.updated");
206export const ItemStartedType = Schema.Literal("item.started");210export const ItemStartedType = Schema.Literal("item.started");
207export const ItemUpdatedType = Schema.Literal("item.updated");211export const ItemUpdatedType = Schema.Literal("item.updated");
...@@ -349,6 +353,16 @@ export const TurnPlanUpdatedPayload = Schema.Struct({...@@ -349,6 +353,16 @@ export const TurnPlanUpdatedPayload = Schema.Struct({
349});353});
350export type TurnPlanUpdatedPayload = typeof TurnPlanUpdatedPayload.Type;354export type TurnPlanUpdatedPayload = typeof TurnPlanUpdatedPayload.Type;
351355
356export const TurnProposedDeltaPayload = Schema.Struct({
357 delta: Schema.String,
358});
359export type TurnProposedDeltaPayload = typeof TurnProposedDeltaPayload.Type;
360
361export const TurnProposedCompletedPayload = Schema.Struct({
362 planMarkdown: TrimmedNonEmptyStringSchema,
363});
364export type TurnProposedCompletedPayload = typeof TurnProposedCompletedPayload.Type;
365
352export const TurnDiffUpdatedPayload = Schema.Struct({366export const TurnDiffUpdatedPayload = Schema.Struct({
353 unifiedDiff: Schema.String,367 unifiedDiff: Schema.String,
354});368});
...@@ -677,6 +691,22 @@ export const ProviderRuntimeTurnPlanUpdatedEvent = Schema.Struct({...@@ -677,6 +691,22 @@ export const ProviderRuntimeTurnPlanUpdatedEvent = Schema.Struct({
677});691});
678export type ProviderRuntimeTurnPlanUpdatedEvent = typeof ProviderRuntimeTurnPlanUpdatedEvent.Type;692export type ProviderRuntimeTurnPlanUpdatedEvent = typeof ProviderRuntimeTurnPlanUpdatedEvent.Type;
679693
694export const ProviderRuntimeTurnProposedDeltaEvent = Schema.Struct({
695 ...ProviderRuntimeEventBase.fields,
696 type: TurnProposedDeltaType,
697 payload: TurnProposedDeltaPayload,
698});
699export type ProviderRuntimeTurnProposedDeltaEvent =
700 typeof ProviderRuntimeTurnProposedDeltaEvent.Type;
701
702export const ProviderRuntimeTurnProposedCompletedEvent = Schema.Struct({
703 ...ProviderRuntimeEventBase.fields,
704 type: TurnProposedCompletedType,
705 payload: TurnProposedCompletedPayload,
706});
707export type ProviderRuntimeTurnProposedCompletedEvent =
708 typeof ProviderRuntimeTurnProposedCompletedEvent.Type;
709
680export const ProviderRuntimeTurnDiffUpdatedEvent = Schema.Struct({710export const ProviderRuntimeTurnDiffUpdatedEvent = Schema.Struct({
681 ...ProviderRuntimeEventBase.fields,711 ...ProviderRuntimeEventBase.fields,
682 type: TurnDiffUpdatedType,712 type: TurnDiffUpdatedType,
...@@ -893,6 +923,8 @@ export const ProviderRuntimeEventV2 = Schema.Union([...@@ -893,6 +923,8 @@ export const ProviderRuntimeEventV2 = Schema.Union([
893 ProviderRuntimeTurnCompletedEvent,923 ProviderRuntimeTurnCompletedEvent,
894 ProviderRuntimeTurnAbortedEvent,924 ProviderRuntimeTurnAbortedEvent,
895 ProviderRuntimeTurnPlanUpdatedEvent,925 ProviderRuntimeTurnPlanUpdatedEvent,
926 ProviderRuntimeTurnProposedDeltaEvent,
927 ProviderRuntimeTurnProposedCompletedEvent,
896 ProviderRuntimeTurnDiffUpdatedEvent,928 ProviderRuntimeTurnDiffUpdatedEvent,
897 ProviderRuntimeItemStartedEvent,929 ProviderRuntimeItemStartedEvent,
898 ProviderRuntimeItemUpdatedEvent,930 ProviderRuntimeItemUpdatedEvent,