authorgravatar for 73941998+UtkarshUsername@users.noreply.github.comUtkarsh Patil <73941998+UtkarshUsername@users.noreply.github.com> 2026-03-19 01:39:57+05:30
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-03-18 13:09:57-07:00
log0de5742b04887439d827bb51d10c2e3e849f67ea
tree85aa238cb51a13cc0001079018950af4efa18188
parentb500a36a94524f9829b0675e602789c4a56f6477
signature Signed by PGP key B5690EEEBB952194

fix: consume plans on implementation (#1133)

23 files changed, 965 insertions(+), 16 deletions(-)

apps/server/src/orchestration/Layers/ProjectionPipeline.ts+4
......@@ -653,6 +653,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
653653 threadId: event.payload.threadId,
654654 turnId: event.payload.proposedPlan.turnId,
655655 planMarkdown: event.payload.proposedPlan.planMarkdown,
656 implementedAt: event.payload.proposedPlan.implementedAt,
657 implementationThreadId: event.payload.proposedPlan.implementationThreadId,
656658 createdAt: event.payload.proposedPlan.createdAt,
657659 updatedAt: event.payload.proposedPlan.updatedAt,
658660 });
......@@ -775,6 +777,8 @@ const makeOrchestrationProjectionPipeline = Effect.gen(function* () {
775777 yield* projectionTurnRepository.replacePendingTurnStart({
776778 threadId: event.payload.threadId,
777779 messageId: event.payload.messageId,
780 sourceProposedPlanThreadId: event.payload.sourceProposedPlan?.threadId ?? null,
781 sourceProposedPlanId: event.payload.sourceProposedPlan?.planId ?? null,
778782 requestedAt: event.payload.createdAt,
779783 });
780784 return;
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+35-1
......@@ -26,6 +26,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
2626
2727 yield* sql`DELETE FROM projection_projects`;
2828 yield* sql`DELETE FROM projection_state`;
29 yield* sql`DELETE FROM projection_thread_proposed_plans`;
2930 yield* sql`DELETE FROM projection_turns`;
3031
3132 yield* sql`
......@@ -101,6 +102,29 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
101102 )
102103 `;
103104
105 yield* sql`
106 INSERT INTO projection_thread_proposed_plans (
107 plan_id,
108 thread_id,
109 turn_id,
110 plan_markdown,
111 implemented_at,
112 implementation_thread_id,
113 created_at,
114 updated_at
115 )
116 VALUES (
117 'plan-1',
118 'thread-1',
119 'turn-1',
120 '# Ship it',
121 '2026-02-24T00:00:05.500Z',
122 'thread-2',
123 '2026-02-24T00:00:05.000Z',
124 '2026-02-24T00:00:05.500Z'
125 )
126 `;
127
104128 yield* sql`
105129 INSERT INTO projection_thread_activities (
106130 activity_id,
......@@ -253,7 +277,17 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => {
253277 updatedAt: "2026-02-24T00:00:05.000Z",
254278 },
255279 ],
256 proposedPlans: [],
280 proposedPlans: [
281 {
282 id: "plan-1",
283 turnId: asTurnId("turn-1"),
284 planMarkdown: "# Ship it",
285 implementedAt: "2026-02-24T00:00:05.500Z",
286 implementationThreadId: ThreadId.makeUnsafe("thread-2"),
287 createdAt: "2026-02-24T00:00:05.000Z",
288 updatedAt: "2026-02-24T00:00:05.500Z",
289 },
290 ],
257291 activities: [
258292 {
259293 id: asEventId("activity-1"),
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+4
......@@ -200,6 +200,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
200200 thread_id AS "threadId",
201201 turn_id AS "turnId",
202202 plan_markdown AS "planMarkdown",
203 implemented_at AS "implementedAt",
204 implementation_thread_id AS "implementationThreadId",
203205 created_at AS "createdAt",
204206 updated_at AS "updatedAt"
205207 FROM projection_thread_proposed_plans
......@@ -435,6 +437,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
435437 id: row.planId,
436438 turnId: row.turnId,
437439 planMarkdown: row.planMarkdown,
440 implementedAt: row.implementedAt,
441 implementationThreadId: row.implementationThreadId,
438442 createdAt: row.createdAt,
439443 updatedAt: row.updatedAt,
440444 });
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+519-3
......@@ -2,7 +2,11 @@ import fs from "node:fs";
22import os from "node:os";
33import path from "node:path";
44
5import type { OrchestrationReadModel, ProviderRuntimeEvent } from "@t3tools/contracts";
5import type {
6 OrchestrationReadModel,
7 ProviderRuntimeEvent,
8 ProviderSession,
9} from "@t3tools/contracts";
610import {
711 ApprovalRequestId,
812 CommandId,
......@@ -57,6 +61,7 @@ type LegacyProviderRuntimeEvent = {
5761
5862function createProviderServiceHarness() {
5963 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
64 const runtimeSessions: ProviderSession[] = [];
6065
6166 const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never;
6267 const service: ProviderServiceShape = {
......@@ -66,12 +71,21 @@ function createProviderServiceHarness() {
6671 respondToRequest: () => unsupported(),
6772 respondToUserInput: () => unsupported(),
6873 stopSession: () => unsupported(),
69 listSessions: () => Effect.succeed([]),
74 listSessions: () => Effect.succeed([...runtimeSessions]),
7075 getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }),
7176 rollbackConversation: () => unsupported(),
7277 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
7378 };
7479
80 const setSession = (session: ProviderSession): void => {
81 const existingIndex = runtimeSessions.findIndex((entry) => entry.threadId === session.threadId);
82 if (existingIndex >= 0) {
83 runtimeSessions[existingIndex] = session;
84 return;
85 }
86 runtimeSessions.push(session);
87 };
88
7589 const emit = (event: LegacyProviderRuntimeEvent): void => {
7690 Effect.runSync(PubSub.publish(runtimeEventPubSub, event as unknown as ProviderRuntimeEvent));
7791 };
......@@ -79,6 +93,7 @@ function createProviderServiceHarness() {
7993 return {
8094 service,
8195 emit,
96 setSession,
8297 };
8398}
8499
......@@ -86,11 +101,12 @@ async function waitForThread(
86101 engine: OrchestrationEngineShape,
87102 predicate: (thread: ProviderRuntimeTestThread) => boolean,
88103 timeoutMs = 2000,
104 threadId: ThreadId = asThreadId("thread-1"),
89105) {
90106 const deadline = Date.now() + timeoutMs;
91107 const poll = async (): Promise<ProviderRuntimeTestThread> => {
92108 const readModel = await Effect.runPromise(engine.getReadModel());
93 const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
109 const thread = readModel.threads.find((entry) => entry.id === threadId);
94110 if (thread && predicate(thread)) {
95111 return thread;
96112 }
......@@ -150,6 +166,7 @@ describe("ProviderRuntimeIngestion", () => {
150166 );
151167 const layer = ProviderRuntimeIngestionLive.pipe(
152168 Layer.provideMerge(orchestrationLayer),
169 Layer.provideMerge(SqlitePersistenceMemory),
153170 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
154171 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())),
155172 Layer.provideMerge(NodeServices.layer),
......@@ -205,10 +222,19 @@ describe("ProviderRuntimeIngestion", () => {
205222 createdAt,
206223 }),
207224 );
225 provider.setSession({
226 provider: "codex",
227 status: "ready",
228 runtimeMode: "approval-required",
229 threadId: ThreadId.makeUnsafe("thread-1"),
230 createdAt,
231 updatedAt: createdAt,
232 });
208233
209234 return {
210235 engine,
211236 emit: provider.emit,
237 setProviderSession: provider.setSession,
212238 drain,
213239 };
214240 }
......@@ -628,6 +654,496 @@ describe("ProviderRuntimeIngestion", () => {
628654 );
629655 });
630656
657 it("marks the source proposed plan implemented only after the target turn starts", async () => {
658 const harness = await createHarness();
659 const sourceThreadId = asThreadId("thread-plan");
660 const targetThreadId = asThreadId("thread-implement");
661 const sourceTurnId = asTurnId("turn-plan-source");
662 const targetTurnId = asTurnId("turn-plan-implement");
663 const createdAt = new Date().toISOString();
664
665 await Effect.runPromise(
666 harness.engine.dispatch({
667 type: "thread.create",
668 commandId: CommandId.makeUnsafe("cmd-thread-create-plan-source"),
669 threadId: sourceThreadId,
670 projectId: asProjectId("project-1"),
671 title: "Plan Source",
672 model: "gpt-5-codex",
673 interactionMode: "plan",
674 runtimeMode: "approval-required",
675 branch: null,
676 worktreePath: null,
677 createdAt,
678 }),
679 );
680 await Effect.runPromise(
681 harness.engine.dispatch({
682 type: "thread.session.set",
683 commandId: CommandId.makeUnsafe("cmd-session-set-plan-source"),
684 threadId: sourceThreadId,
685 session: {
686 threadId: sourceThreadId,
687 status: "ready",
688 providerName: "codex",
689 runtimeMode: "approval-required",
690 activeTurnId: null,
691 updatedAt: createdAt,
692 lastError: null,
693 },
694 createdAt,
695 }),
696 );
697 await Effect.runPromise(
698 harness.engine.dispatch({
699 type: "thread.create",
700 commandId: CommandId.makeUnsafe("cmd-thread-create-plan-target"),
701 threadId: targetThreadId,
702 projectId: asProjectId("project-1"),
703 title: "Plan Target",
704 model: "gpt-5-codex",
705 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
706 runtimeMode: "approval-required",
707 branch: null,
708 worktreePath: null,
709 createdAt,
710 }),
711 );
712 await Effect.runPromise(
713 harness.engine.dispatch({
714 type: "thread.session.set",
715 commandId: CommandId.makeUnsafe("cmd-session-set-plan-target"),
716 threadId: targetThreadId,
717 session: {
718 threadId: targetThreadId,
719 status: "ready",
720 providerName: "codex",
721 runtimeMode: "approval-required",
722 activeTurnId: null,
723 updatedAt: createdAt,
724 lastError: null,
725 },
726 createdAt,
727 }),
728 );
729 harness.setProviderSession({
730 provider: "codex",
731 status: "ready",
732 runtimeMode: "approval-required",
733 threadId: targetThreadId,
734 createdAt,
735 updatedAt: createdAt,
736 activeTurnId: targetTurnId,
737 });
738
739 harness.emit({
740 type: "turn.proposed.completed",
741 eventId: asEventId("evt-plan-source-completed"),
742 provider: "codex",
743 createdAt,
744 threadId: sourceThreadId,
745 turnId: sourceTurnId,
746 payload: {
747 planMarkdown: "# Source plan",
748 },
749 });
750
751 const sourceThreadWithPlan = await waitForThread(
752 harness.engine,
753 (thread) =>
754 thread.proposedPlans.some(
755 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
756 proposedPlan.id === "plan:thread-plan:turn:turn-plan-source" &&
757 proposedPlan.implementedAt === null,
758 ),
759 2_000,
760 sourceThreadId,
761 );
762 const sourcePlan = sourceThreadWithPlan.proposedPlans.find(
763 (entry: ProviderRuntimeTestProposedPlan) =>
764 entry.id === "plan:thread-plan:turn:turn-plan-source",
765 );
766 expect(sourcePlan).toBeDefined();
767 if (!sourcePlan) {
768 throw new Error("Expected source plan to exist.");
769 }
770
771 await Effect.runPromise(
772 harness.engine.dispatch({
773 type: "thread.turn.start",
774 commandId: CommandId.makeUnsafe("cmd-turn-start-plan-target"),
775 threadId: targetThreadId,
776 message: {
777 messageId: asMessageId("msg-plan-target"),
778 role: "user",
779 text: "PLEASE IMPLEMENT THIS PLAN:\n# Source plan",
780 attachments: [],
781 },
782 sourceProposedPlan: {
783 threadId: sourceThreadId,
784 planId: sourcePlan.id,
785 },
786 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
787 runtimeMode: "approval-required",
788 createdAt: new Date().toISOString(),
789 }),
790 );
791
792 const sourceThreadBeforeStart = await waitForThread(
793 harness.engine,
794 (thread) =>
795 thread.proposedPlans.some(
796 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
797 proposedPlan.id === sourcePlan.id && proposedPlan.implementedAt === null,
798 ),
799 2_000,
800 sourceThreadId,
801 );
802 expect(
803 sourceThreadBeforeStart.proposedPlans.find((entry) => entry.id === sourcePlan.id),
804 ).toMatchObject({
805 implementedAt: null,
806 implementationThreadId: null,
807 });
808
809 harness.emit({
810 type: "turn.started",
811 eventId: asEventId("evt-plan-target-started"),
812 provider: "codex",
813 createdAt: new Date().toISOString(),
814 threadId: targetThreadId,
815 turnId: targetTurnId,
816 });
817
818 const sourceThreadAfterStart = await waitForThread(
819 harness.engine,
820 (thread) =>
821 thread.proposedPlans.some(
822 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
823 proposedPlan.id === sourcePlan.id &&
824 proposedPlan.implementedAt !== null &&
825 proposedPlan.implementationThreadId === targetThreadId,
826 ),
827 2_000,
828 sourceThreadId,
829 );
830 expect(
831 sourceThreadAfterStart.proposedPlans.find((entry) => entry.id === sourcePlan.id),
832 ).toMatchObject({
833 implementationThreadId: "thread-implement",
834 });
835 });
836
837 it("does not mark the source proposed plan implemented for a rejected turn.started event", async () => {
838 const harness = await createHarness();
839 const sourceThreadId = asThreadId("thread-plan");
840 const targetThreadId = asThreadId("thread-1");
841 const sourceTurnId = asTurnId("turn-plan-source");
842 const activeTurnId = asTurnId("turn-already-running");
843 const staleTurnId = asTurnId("turn-stale-start");
844 const createdAt = new Date().toISOString();
845
846 await Effect.runPromise(
847 harness.engine.dispatch({
848 type: "thread.create",
849 commandId: CommandId.makeUnsafe("cmd-thread-create-plan-source-guarded"),
850 threadId: sourceThreadId,
851 projectId: asProjectId("project-1"),
852 title: "Plan Source",
853 model: "gpt-5-codex",
854 interactionMode: "plan",
855 runtimeMode: "approval-required",
856 branch: null,
857 worktreePath: null,
858 createdAt,
859 }),
860 );
861 await Effect.runPromise(
862 harness.engine.dispatch({
863 type: "thread.session.set",
864 commandId: CommandId.makeUnsafe("cmd-session-set-plan-source-guarded"),
865 threadId: sourceThreadId,
866 session: {
867 threadId: sourceThreadId,
868 status: "ready",
869 providerName: "codex",
870 runtimeMode: "approval-required",
871 activeTurnId: null,
872 updatedAt: createdAt,
873 lastError: null,
874 },
875 createdAt,
876 }),
877 );
878 harness.setProviderSession({
879 provider: "codex",
880 status: "running",
881 runtimeMode: "approval-required",
882 threadId: targetThreadId,
883 createdAt,
884 updatedAt: createdAt,
885 activeTurnId,
886 });
887
888 harness.emit({
889 type: "turn.started",
890 eventId: asEventId("evt-turn-started-already-running"),
891 provider: "codex",
892 createdAt,
893 threadId: targetThreadId,
894 turnId: activeTurnId,
895 });
896
897 await waitForThread(
898 harness.engine,
899 (thread) =>
900 thread.session?.status === "running" && thread.session?.activeTurnId === activeTurnId,
901 2_000,
902 targetThreadId,
903 );
904
905 harness.emit({
906 type: "turn.proposed.completed",
907 eventId: asEventId("evt-plan-source-completed-guarded"),
908 provider: "codex",
909 createdAt,
910 threadId: sourceThreadId,
911 turnId: sourceTurnId,
912 payload: {
913 planMarkdown: "# Source plan",
914 },
915 });
916
917 const sourceThreadWithPlan = await waitForThread(
918 harness.engine,
919 (thread) =>
920 thread.proposedPlans.some(
921 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
922 proposedPlan.id === "plan:thread-plan:turn:turn-plan-source" &&
923 proposedPlan.implementedAt === null,
924 ),
925 2_000,
926 sourceThreadId,
927 );
928 const sourcePlan = sourceThreadWithPlan.proposedPlans.find(
929 (entry: ProviderRuntimeTestProposedPlan) =>
930 entry.id === "plan:thread-plan:turn:turn-plan-source",
931 );
932 expect(sourcePlan).toBeDefined();
933 if (!sourcePlan) {
934 throw new Error("Expected source plan to exist.");
935 }
936
937 await Effect.runPromise(
938 harness.engine.dispatch({
939 type: "thread.turn.start",
940 commandId: CommandId.makeUnsafe("cmd-turn-start-plan-target-guarded"),
941 threadId: targetThreadId,
942 message: {
943 messageId: asMessageId("msg-plan-target-guarded"),
944 role: "user",
945 text: "PLEASE IMPLEMENT THIS PLAN:\n# Source plan",
946 attachments: [],
947 },
948 sourceProposedPlan: {
949 threadId: sourceThreadId,
950 planId: sourcePlan.id,
951 },
952 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
953 runtimeMode: "approval-required",
954 createdAt: new Date().toISOString(),
955 }),
956 );
957
958 harness.emit({
959 type: "turn.started",
960 eventId: asEventId("evt-turn-started-stale-plan-implementation"),
961 provider: "codex",
962 createdAt: new Date().toISOString(),
963 threadId: targetThreadId,
964 turnId: staleTurnId,
965 });
966
967 await harness.drain();
968
969 const readModel = await Effect.runPromise(harness.engine.getReadModel());
970 const sourceThreadAfterRejectedStart = readModel.threads.find(
971 (entry) => entry.id === sourceThreadId,
972 );
973 expect(
974 sourceThreadAfterRejectedStart?.proposedPlans.find((entry) => entry.id === sourcePlan.id),
975 ).toMatchObject({
976 implementedAt: null,
977 implementationThreadId: null,
978 });
979
980 const targetThreadAfterRejectedStart = readModel.threads.find(
981 (entry) => entry.id === targetThreadId,
982 );
983 expect(targetThreadAfterRejectedStart?.session?.status).toBe("running");
984 expect(targetThreadAfterRejectedStart?.session?.activeTurnId).toBe(activeTurnId);
985 });
986
987 it("does not mark the source proposed plan implemented for an unrelated turn.started when no thread active turn is tracked", async () => {
988 const harness = await createHarness();
989 const sourceThreadId = asThreadId("thread-plan");
990 const targetThreadId = asThreadId("thread-implement");
991 const sourceTurnId = asTurnId("turn-plan-source");
992 const expectedTurnId = asTurnId("turn-plan-implement");
993 const replayedTurnId = asTurnId("turn-replayed");
994 const createdAt = new Date().toISOString();
995
996 await Effect.runPromise(
997 harness.engine.dispatch({
998 type: "thread.create",
999 commandId: CommandId.makeUnsafe("cmd-thread-create-plan-source-unrelated"),
1000 threadId: sourceThreadId,
1001 projectId: asProjectId("project-1"),
1002 title: "Plan Source",
1003 model: "gpt-5-codex",
1004 interactionMode: "plan",
1005 runtimeMode: "approval-required",
1006 branch: null,
1007 worktreePath: null,
1008 createdAt,
1009 }),
1010 );
1011 await Effect.runPromise(
1012 harness.engine.dispatch({
1013 type: "thread.session.set",
1014 commandId: CommandId.makeUnsafe("cmd-session-set-plan-source-unrelated"),
1015 threadId: sourceThreadId,
1016 session: {
1017 threadId: sourceThreadId,
1018 status: "ready",
1019 providerName: "codex",
1020 runtimeMode: "approval-required",
1021 activeTurnId: null,
1022 updatedAt: createdAt,
1023 lastError: null,
1024 },
1025 createdAt,
1026 }),
1027 );
1028 await Effect.runPromise(
1029 harness.engine.dispatch({
1030 type: "thread.create",
1031 commandId: CommandId.makeUnsafe("cmd-thread-create-plan-target-unrelated"),
1032 threadId: targetThreadId,
1033 projectId: asProjectId("project-1"),
1034 title: "Plan Target",
1035 model: "gpt-5-codex",
1036 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1037 runtimeMode: "approval-required",
1038 branch: null,
1039 worktreePath: null,
1040 createdAt,
1041 }),
1042 );
1043 await Effect.runPromise(
1044 harness.engine.dispatch({
1045 type: "thread.session.set",
1046 commandId: CommandId.makeUnsafe("cmd-session-set-plan-target-unrelated"),
1047 threadId: targetThreadId,
1048 session: {
1049 threadId: targetThreadId,
1050 status: "ready",
1051 providerName: "codex",
1052 runtimeMode: "approval-required",
1053 activeTurnId: null,
1054 updatedAt: createdAt,
1055 lastError: null,
1056 },
1057 createdAt,
1058 }),
1059 );
1060
1061 harness.emit({
1062 type: "turn.proposed.completed",
1063 eventId: asEventId("evt-plan-source-completed-unrelated"),
1064 provider: "codex",
1065 createdAt,
1066 threadId: sourceThreadId,
1067 turnId: sourceTurnId,
1068 payload: {
1069 planMarkdown: "# Source plan",
1070 },
1071 });
1072
1073 const sourceThreadWithPlan = await waitForThread(
1074 harness.engine,
1075 (thread) =>
1076 thread.proposedPlans.some(
1077 (proposedPlan: ProviderRuntimeTestProposedPlan) =>
1078 proposedPlan.id === "plan:thread-plan:turn:turn-plan-source" &&
1079 proposedPlan.implementedAt === null,
1080 ),
1081 2_000,
1082 sourceThreadId,
1083 );
1084 const sourcePlan = sourceThreadWithPlan.proposedPlans.find(
1085 (entry: ProviderRuntimeTestProposedPlan) =>
1086 entry.id === "plan:thread-plan:turn:turn-plan-source",
1087 );
1088 expect(sourcePlan).toBeDefined();
1089 if (!sourcePlan) {
1090 throw new Error("Expected source plan to exist.");
1091 }
1092
1093 await Effect.runPromise(
1094 harness.engine.dispatch({
1095 type: "thread.turn.start",
1096 commandId: CommandId.makeUnsafe("cmd-turn-start-plan-target-unrelated"),
1097 threadId: targetThreadId,
1098 message: {
1099 messageId: asMessageId("msg-plan-target-unrelated"),
1100 role: "user",
1101 text: "PLEASE IMPLEMENT THIS PLAN:\n# Source plan",
1102 attachments: [],
1103 },
1104 sourceProposedPlan: {
1105 threadId: sourceThreadId,
1106 planId: sourcePlan.id,
1107 },
1108 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1109 runtimeMode: "approval-required",
1110 createdAt: new Date().toISOString(),
1111 }),
1112 );
1113
1114 harness.setProviderSession({
1115 provider: "codex",
1116 status: "running",
1117 runtimeMode: "approval-required",
1118 threadId: targetThreadId,
1119 createdAt,
1120 updatedAt: createdAt,
1121 activeTurnId: expectedTurnId,
1122 });
1123
1124 harness.emit({
1125 type: "turn.started",
1126 eventId: asEventId("evt-turn-started-unrelated-plan-implementation"),
1127 provider: "codex",
1128 createdAt: new Date().toISOString(),
1129 threadId: targetThreadId,
1130 turnId: replayedTurnId,
1131 });
1132
1133 await harness.drain();
1134
1135 const readModel = await Effect.runPromise(harness.engine.getReadModel());
1136 const sourceThreadAfterUnrelatedStart = readModel.threads.find(
1137 (entry) => entry.id === sourceThreadId,
1138 );
1139 expect(
1140 sourceThreadAfterUnrelatedStart?.proposedPlans.find((entry) => entry.id === sourcePlan.id),
1141 ).toMatchObject({
1142 implementedAt: null,
1143 implementationThreadId: null,
1144 });
1145 });
1146
6311147 it("finalizes buffered proposed-plan deltas into a first-class proposed plan on turn completion", async () => {
6321148 const harness = await createHarness();
6331149 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+111-1
......@@ -4,6 +4,7 @@ import {
44 CommandId,
55 MessageId,
66 type OrchestrationEvent,
7 type OrchestrationProposedPlanId,
78 CheckpointRef,
89 isToolLifecycleItemType,
910 ThreadId,
......@@ -15,6 +16,8 @@ import { Cache, Cause, Duration, Effect, Layer, Option, Ref, Stream } from "effe
1516import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
1617
1718import { ProviderService } from "../../provider/Services/ProviderService.ts";
19import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts";
20import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";
1821import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts";
1922import { isGitRepository } from "../../git/isRepo.ts";
2023import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
......@@ -480,6 +483,7 @@ function runtimeEventToActivities(
480483const make = Effect.gen(function* () {
481484 const orchestrationEngine = yield* OrchestrationEngineService;
482485 const providerService = yield* ProviderService;
486 const projectionTurnRepository = yield* ProjectionTurnRepository;
483487
484488 const assistantDeliveryModeRef = yield* Ref.make<AssistantDeliveryMode>(
485489 DEFAULT_ASSISTANT_DELIVERY_MODE,
......@@ -671,6 +675,8 @@ const make = Effect.gen(function* () {
671675 threadProposedPlans: ReadonlyArray<{
672676 id: string;
673677 createdAt: string;
678 implementedAt: string | null;
679 implementationThreadId: ThreadId | null;
674680 }>;
675681 planId: string;
676682 turnId?: TurnId;
......@@ -693,6 +699,8 @@ const make = Effect.gen(function* () {
693699 id: input.planId,
694700 turnId: input.turnId ?? null,
695701 planMarkdown,
702 implementedAt: existingPlan?.implementedAt ?? null,
703 implementationThreadId: existingPlan?.implementationThreadId ?? null,
696704 createdAt: existingPlan?.createdAt ?? input.createdAt,
697705 updatedAt: input.updatedAt,
698706 },
......@@ -706,6 +714,8 @@ const make = Effect.gen(function* () {
706714 threadProposedPlans: ReadonlyArray<{
707715 id: string;
708716 createdAt: string;
717 implementedAt: string | null;
718 implementationThreadId: ThreadId | null;
709719 }>;
710720 planId: string;
711721 turnId?: TurnId;
......@@ -772,6 +782,79 @@ const make = Effect.gen(function* () {
772782 ).pipe(Effect.asVoid);
773783 });
774784
785 const getSourceProposedPlanReferenceForPendingTurnStart = Effect.fnUntraced(function* (
786 threadId: ThreadId,
787 ) {
788 const pendingTurnStart = yield* projectionTurnRepository.getPendingTurnStartByThreadId({
789 threadId,
790 });
791 if (Option.isNone(pendingTurnStart)) {
792 return null;
793 }
794
795 const sourceThreadId = pendingTurnStart.value.sourceProposedPlanThreadId;
796 const sourcePlanId = pendingTurnStart.value.sourceProposedPlanId;
797 if (sourceThreadId === null || sourcePlanId === null) {
798 return null;
799 }
800
801 return {
802 sourceThreadId,
803 sourcePlanId,
804 } as const;
805 });
806
807 const getExpectedProviderTurnIdForThread = Effect.fnUntraced(function* (threadId: ThreadId) {
808 const sessions = yield* providerService.listSessions();
809 const session = sessions.find((entry) => entry.threadId === threadId);
810 return session?.activeTurnId;
811 });
812
813 const getSourceProposedPlanReferenceForAcceptedTurnStart = Effect.fnUntraced(function* (
814 threadId: ThreadId,
815 eventTurnId: TurnId | undefined,
816 ) {
817 if (eventTurnId === undefined) {
818 return null;
819 }
820
821 const expectedTurnId = yield* getExpectedProviderTurnIdForThread(threadId);
822 if (!sameId(expectedTurnId, eventTurnId)) {
823 return null;
824 }
825
826 return yield* getSourceProposedPlanReferenceForPendingTurnStart(threadId);
827 });
828
829 const markSourceProposedPlanImplemented = Effect.fnUntraced(function* (
830 sourceThreadId: ThreadId,
831 sourcePlanId: OrchestrationProposedPlanId,
832 implementationThreadId: ThreadId,
833 implementedAt: string,
834 ) {
835 const readModel = yield* orchestrationEngine.getReadModel();
836 const sourceThread = readModel.threads.find((entry) => entry.id === sourceThreadId);
837 const sourcePlan = sourceThread?.proposedPlans.find((entry) => entry.id === sourcePlanId);
838 if (!sourceThread || !sourcePlan || sourcePlan.implementedAt !== null) {
839 return;
840 }
841
842 yield* orchestrationEngine.dispatch({
843 type: "thread.proposed-plan.upsert",
844 commandId: CommandId.makeUnsafe(
845 `provider:source-proposed-plan-implemented:${implementationThreadId}:${crypto.randomUUID()}`,
846 ),
847 threadId: sourceThread.id,
848 proposedPlan: {
849 ...sourcePlan,
850 implementedAt,
851 implementationThreadId,
852 updatedAt: implementedAt,
853 },
854 createdAt: implementedAt,
855 });
856 });
857
775858 const processRuntimeEvent = (event: ProviderRuntimeEvent) =>
776859 Effect.gen(function* () {
777860 const readModel = yield* orchestrationEngine.getReadModel();
......@@ -812,6 +895,10 @@ const make = Effect.gen(function* () {
812895 return true;
813896 }
814897 })();
898 const acceptedTurnStartedSourcePlan =
899 event.type === "turn.started" && shouldApplyThreadLifecycle
900 ? yield* getSourceProposedPlanReferenceForAcceptedTurnStart(thread.id, eventTurnId)
901 : null;
815902
816903 if (
817904 event.type === "session.started" ||
......@@ -854,6 +941,26 @@ const make = Effect.gen(function* () {
854941 : (thread.session?.lastError ?? null);
855942
856943 if (shouldApplyThreadLifecycle) {
944 if (event.type === "turn.started" && acceptedTurnStartedSourcePlan !== null) {
945 yield* markSourceProposedPlanImplemented(
946 acceptedTurnStartedSourcePlan.sourceThreadId,
947 acceptedTurnStartedSourcePlan.sourcePlanId,
948 thread.id,
949 now,
950 ).pipe(
951 Effect.catchCause((cause) =>
952 Effect.logWarning(
953 "provider runtime ingestion failed to mark source proposed plan",
954 {
955 eventId: event.eventId,
956 eventType: event.type,
957 cause: Cause.pretty(cause),
958 },
959 ),
960 ),
961 );
962 }
963
857964 yield* orchestrationEngine.dispatch({
858965 type: "thread.session.set",
859966 commandId: providerCommandId(event, "thread-session-set"),
......@@ -1144,4 +1251,7 @@ const make = Effect.gen(function* () {
11441251 } satisfies ProviderRuntimeIngestionShape;
11451252});
11461253
1147export const ProviderRuntimeIngestionLive = Layer.effect(ProviderRuntimeIngestionService, make);
1254export const ProviderRuntimeIngestionLive = Layer.effect(
1255 ProviderRuntimeIngestionService,
1256 make,
1257).pipe(Layer.provide(ProjectionTurnRepositoryLive));
apps/server/src/orchestration/decider.ts+28-7
......@@ -262,11 +262,35 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
262262 }
263263
264264 case "thread.turn.start": {
265 yield* requireThread({
265 const targetThread = yield* requireThread({
266266 readModel,
267267 command,
268268 threadId: command.threadId,
269269 });
270 const sourceProposedPlan = command.sourceProposedPlan;
271 const sourceThread = sourceProposedPlan
272 ? yield* requireThread({
273 readModel,
274 command,
275 threadId: sourceProposedPlan.threadId,
276 })
277 : null;
278 const sourcePlan =
279 sourceProposedPlan && sourceThread
280 ? sourceThread.proposedPlans.find((entry) => entry.id === sourceProposedPlan.planId)
281 : null;
282 if (sourceProposedPlan && !sourcePlan) {
283 return yield* new OrchestrationCommandInvariantError({
284 commandType: command.type,
285 detail: `Proposed plan '${sourceProposedPlan.planId}' does not exist on thread '${sourceProposedPlan.threadId}'.`,
286 });
287 }
288 if (sourceThread && sourceThread.projectId !== targetThread.projectId) {
289 return yield* new OrchestrationCommandInvariantError({
290 commandType: command.type,
291 detail: `Proposed plan '${sourceProposedPlan?.planId}' belongs to thread '${sourceThread.id}' in a different project.`,
292 });
293 }
270294 const userMessageEvent: Omit<OrchestrationEvent, "sequence"> = {
271295 ...withEventBase({
272296 aggregateKind: "thread",
......@@ -306,12 +330,9 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand"
306330 ? { providerOptions: command.providerOptions }
307331 : {}),
308332 assistantDeliveryMode: command.assistantDeliveryMode ?? DEFAULT_ASSISTANT_DELIVERY_MODE,
309 runtimeMode:
310 readModel.threads.find((entry) => entry.id === command.threadId)?.runtimeMode ??
311 command.runtimeMode,
312 interactionMode:
313 readModel.threads.find((entry) => entry.id === command.threadId)?.interactionMode ??
314 command.interactionMode,
333 runtimeMode: targetThread.runtimeMode,
334 interactionMode: targetThread.interactionMode,
335 ...(sourceProposedPlan !== undefined ? { sourceProposedPlan } : {}),
315336 createdAt: command.createdAt,
316337 },
317338 };
apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts+8
......@@ -22,6 +22,8 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
2222 thread_id,
2323 turn_id,
2424 plan_markdown,
25 implemented_at,
26 implementation_thread_id,
2527 created_at,
2628 updated_at
2729 )
......@@ -30,6 +32,8 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
3032 ${row.threadId},
3133 ${row.turnId},
3234 ${row.planMarkdown},
35 ${row.implementedAt},
36 ${row.implementationThreadId},
3337 ${row.createdAt},
3438 ${row.updatedAt}
3539 )
......@@ -38,6 +42,8 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
3842 thread_id = excluded.thread_id,
3943 turn_id = excluded.turn_id,
4044 plan_markdown = excluded.plan_markdown,
45 implemented_at = excluded.implemented_at,
46 implementation_thread_id = excluded.implementation_thread_id,
4147 created_at = excluded.created_at,
4248 updated_at = excluded.updated_at
4349 `,
......@@ -52,6 +58,8 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
5258 thread_id AS "threadId",
5359 turn_id AS "turnId",
5460 plan_markdown AS "planMarkdown",
61 implemented_at AS "implementedAt",
62 implementation_thread_id AS "implementationThreadId",
5563 created_at AS "createdAt",
5664 updated_at AS "updatedAt"
5765 FROM projection_thread_proposed_plans
apps/server/src/persistence/Layers/ProjectionTurns.ts+6
......@@ -106,6 +106,8 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
106106 thread_id,
107107 turn_id,
108108 pending_message_id,
109 source_proposed_plan_thread_id,
110 source_proposed_plan_id,
109111 assistant_message_id,
110112 state,
111113 requested_at,
......@@ -120,6 +122,8 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
120122 ${row.threadId},
121123 NULL,
122124 ${row.messageId},
125 ${row.sourceProposedPlanThreadId},
126 ${row.sourceProposedPlanId},
123127 NULL,
124128 'pending',
125129 ${row.requestedAt},
......@@ -141,6 +145,8 @@ const makeProjectionTurnRepository = Effect.gen(function* () {
141145 SELECT
142146 thread_id AS "threadId",
143147 pending_message_id AS "messageId",
148 source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
149 source_proposed_plan_id AS "sourceProposedPlanId",
144150 requested_at AS "requestedAt"
145151 FROM projection_turns
146152 WHERE thread_id = ${threadId}
apps/server/src/persistence/Migrations.ts+4
......@@ -25,6 +25,8 @@ import Migration0010 from "./Migrations/010_ProjectionThreadsRuntimeMode.ts";
2525import Migration0011 from "./Migrations/011_OrchestrationThreadCreatedRuntimeMode.ts";
2626import Migration0012 from "./Migrations/012_ProjectionThreadsInteractionMode.ts";
2727import Migration0013 from "./Migrations/013_ProjectionThreadProposedPlans.ts";
28import Migration0014 from "./Migrations/014_ProjectionThreadProposedPlanImplementation.ts";
29import Migration0015 from "./Migrations/015_ProjectionTurnsSourceProposedPlan.ts";
2830import { Effect } from "effect";
2931
3032/**
......@@ -51,6 +53,8 @@ const loader = Migrator.fromRecord({
5153 "11_OrchestrationThreadCreatedRuntimeMode": Migration0011,
5254 "12_ProjectionThreadsInteractionMode": Migration0012,
5355 "13_ProjectionThreadProposedPlans": Migration0013,
56 "14_ProjectionThreadProposedPlanImplementation": Migration0014,
57 "15_ProjectionTurnsSourceProposedPlan": Migration0015,
5458});
5559
5660/**
apps/server/src/persistence/Migrations/014_ProjectionThreadProposedPlanImplementation.ts created+16
......@@ -0,0 +1,16 @@
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 ALTER TABLE projection_thread_proposed_plans
9 ADD COLUMN implemented_at TEXT
10 `;
11
12 yield* sql`
13 ALTER TABLE projection_thread_proposed_plans
14 ADD COLUMN implementation_thread_id TEXT
15 `;
16});
apps/server/src/persistence/Migrations/015_ProjectionTurnsSourceProposedPlan.ts created+16
......@@ -0,0 +1,16 @@
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 ALTER TABLE projection_turns
9 ADD COLUMN source_proposed_plan_thread_id TEXT
10 `;
11
12 yield* sql`
13 ALTER TABLE projection_turns
14 ADD COLUMN source_proposed_plan_id TEXT
15 `;
16});
apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts+2
......@@ -15,6 +15,8 @@ export const ProjectionThreadProposedPlan = Schema.Struct({
1515 threadId: ThreadId,
1616 turnId: Schema.NullOr(TurnId),
1717 planMarkdown: TrimmedNonEmptyString,
18 implementedAt: Schema.NullOr(IsoDateTime),
19 implementationThreadId: Schema.NullOr(ThreadId),
1820 createdAt: IsoDateTime,
1921 updatedAt: IsoDateTime,
2022});
apps/server/src/persistence/Services/ProjectionTurns.ts+3
......@@ -11,6 +11,7 @@ import {
1111 IsoDateTime,
1212 MessageId,
1313 NonNegativeInt,
14 OrchestrationProposedPlanId,
1415 OrchestrationCheckpointFile,
1516 OrchestrationCheckpointStatus,
1617 ThreadId,
......@@ -65,6 +66,8 @@ export type ProjectionTurnById = typeof ProjectionTurnById.Type;
6566export const ProjectionPendingTurnStart = Schema.Struct({
6667 threadId: ThreadId,
6768 messageId: MessageId,
69 sourceProposedPlanThreadId: Schema.NullOr(ThreadId),
70 sourceProposedPlanId: Schema.NullOr(OrchestrationProposedPlanId),
6871 requestedAt: IsoDateTime,
6972});
7073export type ProjectionPendingTurnStart = typeof ProjectionPendingTurnStart.Type;
apps/web/src/components/ChatView.browser.tsx+2
......@@ -368,6 +368,8 @@ function createSnapshotWithLongProposedPlan(): OrchestrationReadModel {
368368 id: "plan-browser-test",
369369 turnId: null,
370370 planMarkdown,
371 implementedAt: null,
372 implementationThreadId: null,
371373 createdAt: isoAt(1_000),
372374 updatedAt: isoAt(1_001),
373375 },
apps/web/src/components/ChatView.tsx+11-1
......@@ -55,6 +55,7 @@ import {
5555 deriveActivePlanState,
5656 findLatestProposedPlan,
5757 deriveWorkLogEntries,
58 hasActionableProposedPlan,
5859 hasToolActivityForTurn,
5960 isLatestTurnSettled,
6061 formatElapsed,
......@@ -721,7 +722,7 @@ export default function ChatView({ threadId }: ChatViewProps) {
721722 pendingUserInputs.length === 0 &&
722723 interactionMode === "plan" &&
723724 latestTurnSettled &&
724 activeProposedPlan !== null;
725 hasActionableProposedPlan(activeProposedPlan);
725726 const activePendingApproval = pendingApprovals[0] ?? null;
726727 const isComposerApprovalState = activePendingApproval !== null;
727728 const hasComposerHeader =
......@@ -2878,6 +2879,14 @@ export default function ChatView({ threadId }: ChatViewProps) {
28782879 assistantDeliveryMode: settings.enableAssistantStreaming ? "streaming" : "buffered",
28792880 runtimeMode,
28802881 interactionMode: nextInteractionMode,
2882 ...(nextInteractionMode === "default" && activeProposedPlan
2883 ? {
2884 sourceProposedPlan: {
2885 threadId: activeThread.id,
2886 planId: activeProposedPlan.id,
2887 },
2888 }
2889 : {}),
28812890 createdAt: messageCreatedAt,
28822891 });
28832892 // Optimistically open the plan sidebar when implementing (not refining).
......@@ -2902,6 +2911,7 @@ export default function ChatView({ threadId }: ChatViewProps) {
29022911 },
29032912 [
29042913 activeThread,
2914 activeProposedPlan,
29052915 beginSendPhase,
29062916 forceStickToBottom,
29072917 isConnecting,
apps/web/src/components/Sidebar.logic.test.ts+31
......@@ -141,6 +141,8 @@ describe("resolveThreadStatusPill", () => {
141141 createdAt: "2026-03-09T10:00:00.000Z",
142142 updatedAt: "2026-03-09T10:05:00.000Z",
143143 planMarkdown: "# Plan",
144 implementedAt: null,
145 implementationThreadId: null,
144146 },
145147 ],
146148 session: {
......@@ -155,6 +157,35 @@ describe("resolveThreadStatusPill", () => {
155157 ).toMatchObject({ label: "Plan Ready", pulse: false });
156158 });
157159
160 it("does not show plan ready after the proposed plan was implemented elsewhere", () => {
161 expect(
162 resolveThreadStatusPill({
163 thread: {
164 ...baseThread,
165 latestTurn: makeLatestTurn(),
166 proposedPlans: [
167 {
168 id: "plan-1" as never,
169 turnId: "turn-1" as never,
170 createdAt: "2026-03-09T10:00:00.000Z",
171 updatedAt: "2026-03-09T10:05:00.000Z",
172 planMarkdown: "# Plan",
173 implementedAt: "2026-03-09T10:06:00.000Z",
174 implementationThreadId: "thread-implement" as never,
175 },
176 ],
177 session: {
178 ...baseThread.session,
179 status: "ready",
180 orchestrationStatus: "ready",
181 },
182 },
183 hasPendingApprovals: false,
184 hasPendingUserInput: false,
185 }),
186 ).toMatchObject({ label: "Completed", pulse: false });
187 });
188
158189 it("shows completed when there is an unseen completion and no active blocker", () => {
159190 expect(
160191 resolveThreadStatusPill({
apps/web/src/components/Sidebar.logic.ts+8-2
......@@ -1,6 +1,10 @@
11import type { Thread } from "../types";
22import { cn } from "../lib/utils";
3import { findLatestProposedPlan, isLatestTurnSettled } from "../session-logic";
3import {
4 findLatestProposedPlan,
5 hasActionableProposedPlan,
6 isLatestTurnSettled,
7} from "../session-logic";
48
59export const THREAD_SELECTION_SAFE_SELECTOR = "[data-thread-item], [data-thread-selection-safe]";
610export type SidebarNewThreadEnvMode = "local" | "worktree";
......@@ -124,7 +128,9 @@ export function resolveThreadStatusPill(input: {
124128 !hasPendingUserInput &&
125129 thread.interactionMode === "plan" &&
126130 isLatestTurnSettled(thread.latestTurn, thread.session) &&
127 findLatestProposedPlan(thread.proposedPlans, thread.latestTurn?.turnId ?? null) !== null;
131 hasActionableProposedPlan(
132 findLatestProposedPlan(thread.proposedPlans, thread.latestTurn?.turnId ?? null),
133 );
128134 if (hasPlanReadyPrompt) {
129135 return {
130136 label: "Plan Ready",
apps/web/src/session-logic.test.ts+54-1
......@@ -1,4 +1,10 @@
1import { EventId, MessageId, TurnId, type OrchestrationThreadActivity } from "@t3tools/contracts";
1import {
2 EventId,
3 MessageId,
4 ThreadId,
5 TurnId,
6 type OrchestrationThreadActivity,
7} from "@t3tools/contracts";
28import { describe, expect, it } from "vitest";
39
410import {
......@@ -10,6 +16,7 @@ import {
1016 deriveTimelineEntries,
1117 deriveWorkLogEntries,
1218 findLatestProposedPlan,
19 hasActionableProposedPlan,
1320 hasToolActivityForTurn,
1421 isLatestTurnSettled,
1522} from "./session-logic";
......@@ -269,6 +276,8 @@ describe("findLatestProposedPlan", () => {
269276 id: "plan:thread-1:turn:turn-1",
270277 turnId: TurnId.makeUnsafe("turn-1"),
271278 planMarkdown: "# Older",
279 implementedAt: null,
280 implementationThreadId: null,
272281 createdAt: "2026-02-23T00:00:01.000Z",
273282 updatedAt: "2026-02-23T00:00:01.000Z",
274283 },
......@@ -276,6 +285,8 @@ describe("findLatestProposedPlan", () => {
276285 id: "plan:thread-1:turn:turn-1",
277286 turnId: TurnId.makeUnsafe("turn-1"),
278287 planMarkdown: "# Latest",
288 implementedAt: null,
289 implementationThreadId: null,
279290 createdAt: "2026-02-23T00:00:01.000Z",
280291 updatedAt: "2026-02-23T00:00:02.000Z",
281292 },
......@@ -283,6 +294,8 @@ describe("findLatestProposedPlan", () => {
283294 id: "plan:thread-1:turn:turn-2",
284295 turnId: TurnId.makeUnsafe("turn-2"),
285296 planMarkdown: "# Different turn",
297 implementedAt: null,
298 implementationThreadId: null,
286299 createdAt: "2026-02-23T00:00:03.000Z",
287300 updatedAt: "2026-02-23T00:00:03.000Z",
288301 },
......@@ -293,6 +306,8 @@ describe("findLatestProposedPlan", () => {
293306 id: "plan:thread-1:turn:turn-1",
294307 turnId: "turn-1",
295308 planMarkdown: "# Latest",
309 implementedAt: null,
310 implementationThreadId: null,
296311 createdAt: "2026-02-23T00:00:01.000Z",
297312 updatedAt: "2026-02-23T00:00:02.000Z",
298313 });
......@@ -305,6 +320,8 @@ describe("findLatestProposedPlan", () => {
305320 id: "plan:thread-1:turn:turn-1",
306321 turnId: TurnId.makeUnsafe("turn-1"),
307322 planMarkdown: "# First",
323 implementedAt: null,
324 implementationThreadId: null,
308325 createdAt: "2026-02-23T00:00:01.000Z",
309326 updatedAt: "2026-02-23T00:00:01.000Z",
310327 },
......@@ -312,6 +329,8 @@ describe("findLatestProposedPlan", () => {
312329 id: "plan:thread-1:turn:turn-2",
313330 turnId: TurnId.makeUnsafe("turn-2"),
314331 planMarkdown: "# Latest",
332 implementedAt: null,
333 implementationThreadId: null,
315334 createdAt: "2026-02-23T00:00:02.000Z",
316335 updatedAt: "2026-02-23T00:00:03.000Z",
317336 },
......@@ -323,6 +342,36 @@ describe("findLatestProposedPlan", () => {
323342 });
324343});
325344
345describe("hasActionableProposedPlan", () => {
346 it("returns true for an unimplemented proposed plan", () => {
347 expect(
348 hasActionableProposedPlan({
349 id: "plan-1",
350 turnId: TurnId.makeUnsafe("turn-1"),
351 planMarkdown: "# Plan",
352 implementedAt: null,
353 implementationThreadId: null,
354 createdAt: "2026-02-23T00:00:00.000Z",
355 updatedAt: "2026-02-23T00:00:01.000Z",
356 }),
357 ).toBe(true);
358 });
359
360 it("returns false for a proposed plan already implemented elsewhere", () => {
361 expect(
362 hasActionableProposedPlan({
363 id: "plan-1",
364 turnId: TurnId.makeUnsafe("turn-1"),
365 planMarkdown: "# Plan",
366 implementedAt: "2026-02-23T00:00:02.000Z",
367 implementationThreadId: ThreadId.makeUnsafe("thread-implement"),
368 createdAt: "2026-02-23T00:00:00.000Z",
369 updatedAt: "2026-02-23T00:00:02.000Z",
370 }),
371 ).toBe(false);
372 });
373});
374
326375describe("deriveWorkLogEntries", () => {
327376 it("omits tool started entries and keeps completed entries", () => {
328377 const activities: OrchestrationThreadActivity[] = [
......@@ -531,6 +580,8 @@ describe("deriveTimelineEntries", () => {
531580 id: "plan:thread-1:turn:turn-1",
532581 turnId: TurnId.makeUnsafe("turn-1"),
533582 planMarkdown: "# Ship it",
583 implementedAt: null,
584 implementationThreadId: null,
534585 createdAt: "2026-02-23T00:00:02.000Z",
535586 updatedAt: "2026-02-23T00:00:02.000Z",
536587 },
......@@ -550,6 +601,8 @@ describe("deriveTimelineEntries", () => {
550601 kind: "proposed-plan",
551602 proposedPlan: {
552603 planMarkdown: "# Ship it",
604 implementedAt: null,
605 implementationThreadId: null,
553606 },
554607 });
555608 });
apps/web/src/session-logic.ts+13
......@@ -7,6 +7,7 @@ import {
77 type ProviderKind,
88 type ToolLifecycleItemType,
99 type UserInputQuestion,
10 type ThreadId,
1011 type TurnId,
1112} from "@t3tools/contracts";
1213
......@@ -72,6 +73,8 @@ export interface LatestProposedPlanState {
7273 updatedAt: string;
7374 turnId: TurnId | null;
7475 planMarkdown: string;
76 implementedAt: string | null;
77 implementationThreadId: ThreadId | null;
7578}
7679
7780export type TimelineEntry =
......@@ -380,6 +383,8 @@ export function findLatestProposedPlan(
380383 updatedAt: matchingTurnPlan.updatedAt,
381384 turnId: matchingTurnPlan.turnId,
382385 planMarkdown: matchingTurnPlan.planMarkdown,
386 implementedAt: matchingTurnPlan.implementedAt,
387 implementationThreadId: matchingTurnPlan.implementationThreadId,
383388 };
384389 }
385390 }
......@@ -400,9 +405,17 @@ export function findLatestProposedPlan(
400405 updatedAt: latestPlan.updatedAt,
401406 turnId: latestPlan.turnId,
402407 planMarkdown: latestPlan.planMarkdown,
408 implementedAt: latestPlan.implementedAt,
409 implementationThreadId: latestPlan.implementationThreadId,
403410 };
404411}
405412
413export function hasActionableProposedPlan(
414 proposedPlan: LatestProposedPlanState | Pick<ProposedPlan, "implementedAt"> | null,
415): boolean {
416 return proposedPlan !== null && proposedPlan.implementedAt === null;
417}
418
406419export function deriveWorkLogEntries(
407420 activities: ReadonlyArray<OrchestrationThreadActivity>,
408421 latestTurnId: TurnId | undefined,
apps/web/src/store.ts+2
......@@ -304,6 +304,8 @@ export function syncServerReadModel(state: AppState, readModel: OrchestrationRea
304304 id: proposedPlan.id,
305305 turnId: proposedPlan.turnId,
306306 planMarkdown: proposedPlan.planMarkdown,
307 implementedAt: proposedPlan.implementedAt,
308 implementationThreadId: proposedPlan.implementationThreadId,
307309 createdAt: proposedPlan.createdAt,
308310 updatedAt: proposedPlan.updatedAt,
309311 })),
apps/web/src/types.ts+2
......@@ -53,6 +53,8 @@ export interface ProposedPlan {
5353 id: OrchestrationProposedPlanId;
5454 turnId: TurnId | null;
5555 planMarkdown: string;
56 implementedAt: string | null;
57 implementationThreadId: ThreadId | null;
5658 createdAt: string;
5759 updatedAt: string;
5860}
packages/contracts/src/orchestration.test.ts+76
......@@ -6,6 +6,7 @@ import {
66 DEFAULT_PROVIDER_INTERACTION_MODE,
77 DEFAULT_RUNTIME_MODE,
88 OrchestrationGetTurnDiffInput,
9 OrchestrationProposedPlan,
910 OrchestrationSession,
1011 ProjectCreateCommand,
1112 ThreadTurnStartCommand,
......@@ -21,6 +22,7 @@ const decodeThreadTurnStartCommand = Schema.decodeUnknownEffect(ThreadTurnStartC
2122const decodeThreadTurnStartRequestedPayload = Schema.decodeUnknownEffect(
2223 ThreadTurnStartRequestedPayload,
2324);
25const decodeOrchestrationProposedPlan = Schema.decodeUnknownEffect(OrchestrationProposedPlan);
2426const decodeOrchestrationSession = Schema.decodeUnknownEffect(OrchestrationSession);
2527const decodeThreadCreatedPayload = Schema.decodeUnknownEffect(ThreadCreatedPayload);
2628
......@@ -186,6 +188,31 @@ it.effect("accepts provider-scoped model options in thread.turn.start", () =>
186188 }),
187189);
188190
191it.effect("accepts a source proposed plan reference in thread.turn.start", () =>
192 Effect.gen(function* () {
193 const parsed = yield* decodeThreadTurnStartCommand({
194 type: "thread.turn.start",
195 commandId: "cmd-turn-source-plan",
196 threadId: "thread-2",
197 message: {
198 messageId: "msg-source-plan",
199 role: "user",
200 text: "implement this",
201 attachments: [],
202 },
203 sourceProposedPlan: {
204 threadId: "thread-1",
205 planId: "plan-1",
206 },
207 createdAt: "2026-01-01T00:00:00.000Z",
208 });
209 assert.deepStrictEqual(parsed.sourceProposedPlan, {
210 threadId: "thread-1",
211 planId: "plan-1",
212 });
213 }),
214);
215
189216it.effect(
190217 "decodes thread.turn-start-requested defaults for provider, runtime mode, and interaction mode",
191218 () =>
......@@ -198,9 +225,28 @@ it.effect(
198225 assert.strictEqual(parsed.provider, undefined);
199226 assert.strictEqual(parsed.runtimeMode, DEFAULT_RUNTIME_MODE);
200227 assert.strictEqual(parsed.interactionMode, DEFAULT_PROVIDER_INTERACTION_MODE);
228 assert.strictEqual(parsed.sourceProposedPlan, undefined);
201229 }),
202230);
203231
232it.effect("decodes thread.turn-start-requested source proposed plan metadata when present", () =>
233 Effect.gen(function* () {
234 const parsed = yield* decodeThreadTurnStartRequestedPayload({
235 threadId: "thread-2",
236 messageId: "msg-2",
237 sourceProposedPlan: {
238 threadId: "thread-1",
239 planId: "plan-1",
240 },
241 createdAt: "2026-01-01T00:00:00.000Z",
242 });
243 assert.deepStrictEqual(parsed.sourceProposedPlan, {
244 threadId: "thread-1",
245 planId: "plan-1",
246 });
247 }),
248);
249
204250it.effect("decodes orchestration session runtime mode defaults", () =>
205251 Effect.gen(function* () {
206252 const parsed = yield* decodeOrchestrationSession({
......@@ -216,3 +262,33 @@ it.effect("decodes orchestration session runtime mode defaults", () =>
216262 assert.strictEqual(parsed.runtimeMode, DEFAULT_RUNTIME_MODE);
217263 }),
218264);
265
266it.effect("defaults proposed plan implementation metadata for historical rows", () =>
267 Effect.gen(function* () {
268 const parsed = yield* decodeOrchestrationProposedPlan({
269 id: "plan-1",
270 turnId: "turn-1",
271 planMarkdown: "# Plan",
272 createdAt: "2026-01-01T00:00:00.000Z",
273 updatedAt: "2026-01-01T00:00:00.000Z",
274 });
275 assert.strictEqual(parsed.implementedAt, null);
276 assert.strictEqual(parsed.implementationThreadId, null);
277 }),
278);
279
280it.effect("preserves proposed plan implementation metadata when present", () =>
281 Effect.gen(function* () {
282 const parsed = yield* decodeOrchestrationProposedPlan({
283 id: "plan-2",
284 turnId: "turn-2",
285 planMarkdown: "# Plan",
286 implementedAt: "2026-01-02T00:00:00.000Z",
287 implementationThreadId: "thread-2",
288 createdAt: "2026-01-01T00:00:00.000Z",
289 updatedAt: "2026-01-02T00:00:00.000Z",
290 });
291 assert.strictEqual(parsed.implementedAt, "2026-01-02T00:00:00.000Z");
292 assert.strictEqual(parsed.implementationThreadId, "thread-2");
293 }),
294);
packages/contracts/src/orchestration.ts+10
......@@ -163,11 +163,18 @@ export const OrchestrationProposedPlan = Schema.Struct({
163163 id: OrchestrationProposedPlanId,
164164 turnId: Schema.NullOr(TurnId),
165165 planMarkdown: TrimmedNonEmptyString,
166 implementedAt: Schema.NullOr(IsoDateTime).pipe(Schema.withDecodingDefault(() => null)),
167 implementationThreadId: Schema.NullOr(ThreadId).pipe(Schema.withDecodingDefault(() => null)),
166168 createdAt: IsoDateTime,
167169 updatedAt: IsoDateTime,
168170});
169171export type OrchestrationProposedPlan = typeof OrchestrationProposedPlan.Type;
170172
173const SourceProposedPlanReference = Schema.Struct({
174 threadId: ThreadId,
175 planId: OrchestrationProposedPlanId,
176});
177
171178export const OrchestrationSessionStatus = Schema.Literals([
172179 "idle",
173180 "starting",
......@@ -374,6 +381,7 @@ export const ThreadTurnStartCommand = Schema.Struct({
374381 interactionMode: ProviderInteractionMode.pipe(
375382 Schema.withDecodingDefault(() => DEFAULT_PROVIDER_INTERACTION_MODE),
376383 ),
384 sourceProposedPlan: Schema.optional(SourceProposedPlanReference),
377385 createdAt: IsoDateTime,
378386});
379387
......@@ -394,6 +402,7 @@ const ClientThreadTurnStartCommand = Schema.Struct({
394402 assistantDeliveryMode: Schema.optional(AssistantDeliveryMode),
395403 runtimeMode: RuntimeMode,
396404 interactionMode: ProviderInteractionMode,
405 sourceProposedPlan: Schema.optional(SourceProposedPlanReference),
397406 createdAt: IsoDateTime,
398407});
399408
......@@ -676,6 +685,7 @@ export const ThreadTurnStartRequestedPayload = Schema.Struct({
676685 interactionMode: ProviderInteractionMode.pipe(
677686 Schema.withDecodingDefault(() => DEFAULT_PROVIDER_INTERACTION_MODE),
678687 ),
688 sourceProposedPlan: Schema.optional(SourceProposedPlanReference),
679689 createdAt: IsoDateTime,
680690});
681691