authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-06 02:56:42-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-06 02:56:42-07:00
loge4e9fa9a0b1a77e393ab6cffd75ec22dbb258bdc
tree10bd68fad603b5cc257dc165080fa58ed89c5605
parente63ddb48e2fd23854a0b4a480b32cbf33e601981
signature Signed by PGP key B5690EEEBB952194

perf(server): finish runtime messages without full thread reads (#10120)

Co-authored-by: Ross Cawston <rcawston@users.noreply.github.com>

11 files changed, 690 insertions(+), 269 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+208-59
......@@ -286,9 +286,20 @@ describe("ProviderRuntimeIngestion", () => {
286286 Layer.provide(RepositoryIdentityResolver.layer),
287287 Layer.provide(SqlitePersistenceMemory),
288288 );
289 const ingestionProjectionSnapshotLayer = Layer.effect(
290 ProjectionSnapshotQuery,
291 Effect.gen(function* () {
292 const query = yield* ProjectionSnapshotQuery;
293 return ProjectionSnapshotQuery.of({
294 ...query,
295 getThreadDetailById: () =>
296 Effect.die("provider runtime ingestion must not hydrate thread detail"),
297 });
298 }),
299 ).pipe(Layer.provide(projectionSnapshotLayer));
289300 const layer = ProviderRuntimeIngestionLive.pipe(
290301 Layer.provideMerge(orchestrationLayer),
291 Layer.provideMerge(projectionSnapshotLayer),
302 Layer.provideMerge(ingestionProjectionSnapshotLayer),
292303 // Single shared liveness instance across ingestion (writer), the
293304 // engine, and the snapshot query (reader).
294305 Layer.provideMerge(ThreadBackgroundLiveness.layer),
......@@ -483,6 +494,89 @@ describe("ProviderRuntimeIngestion", () => {
483494 ]);
484495 });
485496
497 it.each(["turn.completed", "turn.aborted"] as const)(
498 "finalizes old buffered text on late %s without stopping the newer turn",
499 async (terminalType) => {
500 const harness = await createHarness({
501 serverSettings: { enableLegacyTokenStreaming: false },
502 });
503 const threadId = asThreadId("thread-1");
504 const oldTurnId = asTurnId("old-buffered-turn");
505 const newTurnId = asTurnId("new-active-turn");
506 const base = {
507 provider: ProviderDriverKind.make("opencode"),
508 threadId,
509 createdAt: "2026-01-01T00:00:01.000Z",
510 };
511 await harness.emitAndDrain([
512 {
513 ...base,
514 type: "turn.started",
515 eventId: asEventId("old-buffered-started"),
516 turnId: oldTurnId,
517 },
518 {
519 ...base,
520 type: "content.delta",
521 eventId: asEventId("old-buffered-delta"),
522 turnId: oldTurnId,
523 itemId: asItemId("old-buffered-message"),
524 payload: { streamKind: "assistant_text", delta: "Keep the old answer." },
525 },
526 ]);
527 await harness.dispatch({
528 type: "thread.turn.start",
529 commandId: CommandId.make("start-new-while-old-finishes"),
530 threadId,
531 message: {
532 messageId: asMessageId("new-turn-prompt"),
533 role: "user",
534 text: "Continue",
535 attachments: [],
536 },
537 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
538 runtimeMode: "approval-required",
539 createdAt: base.createdAt,
540 });
541 harness.setProviderSession({
542 provider: base.provider,
543 status: "running",
544 runtimeMode: "approval-required",
545 threadId,
546 createdAt: base.createdAt,
547 updatedAt: base.createdAt,
548 activeTurnId: newTurnId,
549 });
550 await harness.emitAndDrain([
551 {
552 ...base,
553 type: "turn.started",
554 eventId: asEventId("new-active-started"),
555 turnId: newTurnId,
556 },
557 {
558 ...base,
559 type: terminalType,
560 eventId: asEventId("old-buffered-terminal"),
561 turnId: oldTurnId,
562 payload:
563 terminalType === "turn.completed"
564 ? { state: "completed" }
565 : { reason: "Interrupted by user." },
566 },
567 ]);
568 const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId);
569 expect(thread?.session).toMatchObject({ activeTurnId: newTurnId, status: "running" });
570 expect(thread?.messages).toContainEqual(
571 expect.objectContaining({
572 turnId: oldTurnId,
573 text: "Keep the old answer.",
574 streaming: false,
575 }),
576 );
577 },
578 );
579
486580 it.each([
487581 { source: "the previous turn", turnId: asTurnId("opencode-stopped-turn") },
488582 { source: "an unspecified turn", turnId: undefined },
......@@ -1690,6 +1784,31 @@ describe("ProviderRuntimeIngestion", () => {
16901784 ).toMatchObject({
16911785 implementationThreadId: "thread-implement",
16921786 });
1787 const implementedPlan = sourceThreadAfterStart.proposedPlans.find(
1788 (entry) => entry.id === sourcePlan.id,
1789 );
1790 await harness.emitAndDrain([
1791 {
1792 type: "turn.proposed.completed",
1793 eventId: asEventId("evt-plan-source-late-completion"),
1794 provider: ProviderDriverKind.make("codex"),
1795 createdAt: "2026-01-01T00:01:00.000Z",
1796 threadId: sourceThreadId,
1797 turnId: sourceTurnId,
1798 payload: { planMarkdown: "# Source plan with late details" },
1799 },
1800 ]);
1801 const sourceAfterLateCompletion = (await harness.readModel()).threads.find(
1802 (entry) => entry.id === sourceThreadId,
1803 );
1804 expect(
1805 sourceAfterLateCompletion?.proposedPlans.find((entry) => entry.id === sourcePlan.id),
1806 ).toMatchObject({
1807 planMarkdown: "# Source plan with late details",
1808 createdAt: sourcePlan.createdAt,
1809 implementedAt: implementedPlan?.implementedAt,
1810 implementationThreadId: targetThreadId,
1811 });
16931812 });
16941813
16951814 it("does not mark the source proposed plan implemented for a rejected turn.started event", async () => {
......@@ -2120,7 +2239,7 @@ describe("ProviderRuntimeIngestion", () => {
21202239 type: "turn.proposed.delta",
21212240 eventId: asEventId("evt-plan-delta-1"),
21222241 provider: ProviderDriverKind.make("codex"),
2123 createdAt: now,
2242 createdAt: "",
21242243 threadId: asThreadId("thread-1"),
21252244 turnId: asTurnId("turn-plan-buffer"),
21262245 payload: {
......@@ -2131,7 +2250,7 @@ describe("ProviderRuntimeIngestion", () => {
21312250 type: "turn.proposed.delta",
21322251 eventId: asEventId("evt-plan-delta-2"),
21332252 provider: ProviderDriverKind.make("codex"),
2134 createdAt: now,
2253 createdAt: "",
21352254 threadId: asThreadId("thread-1"),
21362255 turnId: asTurnId("turn-plan-buffer"),
21372256 payload: {
......@@ -2161,6 +2280,42 @@ describe("ProviderRuntimeIngestion", () => {
21612280 entry.id === "plan:thread-1:turn:turn-plan-buffer",
21622281 );
21632282 expect(proposedPlan?.planMarkdown).toBe("## Buffered plan\n\n- first\n- second");
2283 expect(proposedPlan?.createdAt).toBe(now);
2284 });
2285
2286 it("releases a blank completed plan before a late replacement", async () => {
2287 const harness = await createHarness();
2288 const threadId = asThreadId("thread-1");
2289 const turnId = asTurnId("blank-plan-turn");
2290 const base = { provider: ProviderDriverKind.make("codex"), threadId, turnId };
2291 const replacementTime = "2026-01-01T00:00:02.000Z";
2292 await harness.emitAndDrain([
2293 {
2294 ...base,
2295 type: "turn.proposed.delta",
2296 eventId: asEventId("blank-plan-delta"),
2297 createdAt: "2026-01-01T00:00:00.000Z",
2298 payload: { delta: " \n " },
2299 },
2300 {
2301 ...base,
2302 type: "turn.completed",
2303 eventId: asEventId("blank-plan-completed"),
2304 createdAt: "2026-01-01T00:00:01.000Z",
2305 payload: { state: "completed" },
2306 },
2307 {
2308 ...base,
2309 type: "turn.proposed.completed",
2310 eventId: asEventId("late-plan-completed"),
2311 createdAt: replacementTime,
2312 payload: { planMarkdown: "# Replacement plan" },
2313 },
2314 ]);
2315 const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId);
2316 expect(thread?.proposedPlans).toEqual([
2317 expect.objectContaining({ planMarkdown: "# Replacement plan", createdAt: replacementTime }),
2318 ]);
21642319 });
21652320
21662321 it("buffers assistant deltas with one lifecycle query per event until completion", async () => {
......@@ -4037,71 +4192,65 @@ describe("ProviderRuntimeIngestion", () => {
40374192 expect(completedPayload?.title).toBe("wait for codex review to finish");
40384193 });
40394194
4040 it("titles task completion from persisted activities after the description cache is swept", async () => {
4195 it("recovers a task title past untitled progress after the cache is swept", async () => {
40414196 const harness = await createHarness();
40424197 const now = "2026-01-01T00:00:00.000Z";
4198 const threadId = asThreadId("thread-1");
4199 const turnId = asTurnId("turn-swept-task");
4200 const provider = ProviderDriverKind.make("claudeAgent");
40434201
4044 harness.emit({
4045 type: "task.progress",
4046 eventId: asEventId("evt-swept-task-progress"),
4047 provider: ProviderDriverKind.make("claudeAgent"),
4048 createdAt: now,
4049 threadId: asThreadId("thread-1"),
4050 turnId: asTurnId("turn-swept-task"),
4051 payload: {
4052 taskId: "swept-task-1",
4053 description: "Watch round-3 CI and bots",
4202 await harness.emitAndDrain([
4203 {
4204 type: "task.started",
4205 eventId: asEventId("evt-swept-task-started"),
4206 provider,
4207 createdAt: now,
4208 threadId,
4209 turnId,
4210 payload: { taskId: "swept-task-1", description: "Watch round-3 CI and bots" },
4211 },
4212 ]);
4213 // Older saved progress rows can have no title even when the start has one.
4214 await harness.dispatch({
4215 type: "thread.activity.append",
4216 commandId: CommandId.make("cmd-swept-task-progress"),
4217 threadId,
4218 activity: {
4219 id: asEventId("evt-swept-task-progress"),
4220 kind: "task.progress",
4221 tone: "info",
40544222 summary: "Polling CI checks.",
4223 payload: { taskId: "swept-task-1" },
4224 turnId,
4225 createdAt: "2026-01-01T00:00:01.000Z",
40554226 },
4227 createdAt: "2026-01-01T00:00:01.000Z",
40564228 });
4057
4058 await waitForThread(harness.readModel, (entry) =>
4059 entry.activities.some(
4060 (activity: ProviderRuntimeTestActivity) =>
4061 activity.id === "task-progress:thread-1:swept-task-1",
4062 ),
4063 );
4064
4065 // session.exited sweeps the in-memory description cache; the completion
4066 // that follows must recover the name from persisted activities.
4067 harness.emit({
4068 type: "session.exited",
4069 eventId: asEventId("evt-swept-task-session-exited"),
4070 provider: ProviderDriverKind.make("claudeAgent"),
4071 createdAt: now,
4072 threadId: asThreadId("thread-1"),
4073 payload: {},
4074 });
4075
4076 harness.emit({
4077 type: "task.completed",
4078 eventId: asEventId("evt-swept-task-completed"),
4079 provider: ProviderDriverKind.make("claudeAgent"),
4080 createdAt: now,
4081 threadId: asThreadId("thread-1"),
4082 turnId: asTurnId("turn-swept-task"),
4083 payload: {
4084 taskId: "swept-task-1",
4085 status: "completed",
4086 summary: "CI is green.",
4229 await harness.emitAndDrain([
4230 {
4231 type: "session.exited",
4232 eventId: asEventId("evt-swept-task-session-exited"),
4233 provider,
4234 createdAt: "2026-01-01T00:00:02.000Z",
4235 threadId,
4236 payload: {},
40874237 },
4088 });
4089
4090 const thread = await waitForThread(harness.readModel, (entry) =>
4091 entry.activities.some(
4092 (activity: ProviderRuntimeTestActivity) => activity.id === "evt-swept-task-completed",
4093 ),
4094 );
4238 {
4239 type: "task.completed",
4240 eventId: asEventId("evt-swept-task-completed"),
4241 provider,
4242 createdAt: "2026-01-01T00:00:03.000Z",
4243 threadId,
4244 turnId,
4245 payload: { taskId: "swept-task-1", status: "completed", summary: "CI is green." },
4246 },
4247 ]);
40954248
4096 const completed = thread.activities.find(
4097 (activity: ProviderRuntimeTestActivity) => activity.id === "evt-swept-task-completed",
4249 const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId);
4250 const completed = thread?.activities.find(
4251 (activity) => activity.id === "evt-swept-task-completed",
40984252 );
4099 const completedPayload =
4100 completed?.payload && typeof completed.payload === "object"
4101 ? (completed.payload as Record<string, unknown>)
4102 : undefined;
4103
4104 expect(completedPayload?.title).toBe("Watch round-3 CI and bots");
4253 expect(completed?.payload).toMatchObject({ title: "Watch round-3 CI and bots" });
41054254 });
41064255
41074256 it("projects structured user input request and resolution as thread activities", async () => {
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+116-197
......@@ -4,8 +4,7 @@ import {
44 CommandId,
55 MessageId,
66 type OrchestrationEvent,
7 type OrchestrationMessage,
8 type OrchestrationProposedPlanId,
7 OrchestrationProposedPlanId,
98 CheckpointRef,
109 classifyTaskAgentKind,
1110 EventId,
......@@ -14,8 +13,6 @@ import {
1413 type ThreadTokenUsageSnapshot,
1514 TurnId,
1615 type OrchestrationCheckpointSummary,
17 type OrchestrationProposedPlan,
18 type OrchestrationThread,
1916 type OrchestrationThreadActivity,
2017 type ProviderRuntimeEvent,
2118 RuntimeRequestId,
......@@ -39,6 +36,10 @@ import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/Projectio
3936import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";
4037import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";
4138import * as CheckpointStore from "../../checkpointing/CheckpointStore.ts";
39import { ProjectionThreadMessageRepository } from "../../persistence/Services/ProjectionThreadMessages.ts";
40import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts";
41import { ProjectionThreadProposedPlanRepository } from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
42import { ProjectionThreadProposedPlanRepositoryLive } from "../../persistence/Layers/ProjectionThreadProposedPlans.ts";
4243import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";
4344import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts";
4445import { ThreadPlanProgressService } from "../ThreadPlanProgress.ts";
......@@ -54,13 +55,12 @@ import { canReplaceThreadTitle } from "../threadTitles.ts";
5455
5556const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`;
5657const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`;
57const TASK_TITLE_ACTIVITY_KINDS = ["task.started", "task.progress"] as const;
5858
5959// Fallback when the in-memory description cache no longer has the task name
6060// (server restart, session-exit sweep, TTL/capacity eviction): earlier
6161// task.started/task.progress activities for the task are persisted with it.
6262function findTaskTitleInActivities(
63 activities: ReadonlyArray<OrchestrationThreadActivity> | undefined,
63 activities: ReadonlyArray<{ readonly kind: string; readonly payload: unknown }> | undefined,
6464 taskId: string,
6565): string | undefined {
6666 if (!activities) {
......@@ -138,57 +138,6 @@ function sameId(left: string | null | undefined, right: string | null | undefine
138138 return left === right;
139139}
140140
141function hasAssistantMessageForTurn(
142 messages: ReadonlyArray<OrchestrationMessage>,
143 turnId: TurnId,
144 options?: { readonly streamingOnly?: boolean },
145): boolean {
146 for (let index = 0; index < messages.length; index += 1) {
147 const message = messages[index];
148 if (!message) {
149 continue;
150 }
151 if (message.role !== "assistant" || message.turnId !== turnId) {
152 continue;
153 }
154 if (options?.streamingOnly === true && !message.streaming) {
155 continue;
156 }
157 return true;
158 }
159 return false;
160}
161
162function findMessageById(
163 messages: ReadonlyArray<OrchestrationMessage>,
164 messageId: MessageId,
165): OrchestrationMessage | undefined {
166 for (let index = 0; index < messages.length; index += 1) {
167 const message = messages[index];
168 if (message?.id === messageId) {
169 return message;
170 }
171 }
172 return undefined;
173}
174
175function findProposedPlanById(
176 proposedPlans: ReadonlyArray<
177 Pick<OrchestrationProposedPlan, "id" | "createdAt" | "implementedAt" | "implementationThreadId">
178 >,
179 planId: string,
180):
181 | Pick<OrchestrationProposedPlan, "id" | "createdAt" | "implementedAt" | "implementationThreadId">
182 | undefined {
183 for (let index = 0; index < proposedPlans.length; index += 1) {
184 const proposedPlan = proposedPlans[index];
185 if (proposedPlan?.id === planId) {
186 return proposedPlan;
187 }
188 }
189 return undefined;
190}
191
192141function hasCheckpointForTurn(
193142 checkpoints: ReadonlyArray<OrchestrationCheckpointSummary>,
194143 turnId: TurnId,
......@@ -264,16 +213,15 @@ function buildContextWindowActivityPayload(
264213}
265214
266215function compactedTokenCountsFromActivities(
267 activities: ReadonlyArray<OrchestrationThreadActivity> | undefined,
216 activities: ReadonlyArray<
217 Pick<OrchestrationThreadActivity, "kind" | "payload" | "sequence" | "createdAt">
218 >,
268219): { readonly beforeTokens: number; readonly afterTokens: number } | undefined {
269 const lastCompactionIndex = activities?.findLastIndex(
220 const lastCompactionIndex = activities.findLastIndex(
270221 (activity) => activity.kind === "context-compaction",
271222 );
272 const lastCompaction =
273 lastCompactionIndex !== undefined && lastCompactionIndex >= 0
274 ? activities?.[lastCompactionIndex]
275 : undefined;
276 const activitiesSinceLastCompaction = activities?.slice((lastCompactionIndex ?? -1) + 1) ?? [];
223 const lastCompaction = activities[lastCompactionIndex];
224 const activitiesSinceLastCompaction = activities.slice(lastCompactionIndex + 1);
277225 const usedTokens = activitiesSinceLastCompaction.flatMap((activity) => {
278226 if (activity.kind !== "context-window.updated") return [];
279227 if (lastCompaction !== undefined) {
......@@ -956,6 +904,8 @@ const make = Effect.gen(function* () {
956904 const orchestrationEngine = yield* OrchestrationEngineService;
957905 const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
958906 const providerService = yield* ProviderService;
907 const projectionThreadMessages = yield* ProjectionThreadMessageRepository;
908 const projectionThreadProposedPlans = yield* ProjectionThreadProposedPlanRepository;
959909 const projectionTurnRepository = yield* ProjectionTurnRepository;
960910 const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;
961911 const serverSettingsService = yield* ServerSettingsService;
......@@ -1013,21 +963,22 @@ const make = Effect.gen(function* () {
1013963 ),
1014964 );
1015965
1016 const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (
966 const resolveThreadRuntimeContext = Effect.fn("resolveThreadRuntimeContext")(function* (
1017967 threadId: ThreadId,
1018 activityKinds: ReadonlyArray<string> = [],
1019968 ) {
1020969 return yield* projectionSnapshotQuery
1021 .getThreadDetailById(threadId, { activityKinds })
970 .getThreadRuntimeContext(threadId)
1022971 .pipe(Effect.map(Option.getOrUndefined));
1023972 });
1024973
1025 const resolveThreadRuntimeContext = Effect.fn("resolveThreadRuntimeContext")(function* (
974 const getThreadMessageById = Effect.fn("getThreadMessageById")(function* (
1026975 threadId: ThreadId,
976 messageId: MessageId,
1027977 ) {
1028 return yield* projectionSnapshotQuery
1029 .getThreadRuntimeContext(threadId)
1030 .pipe(Effect.map(Option.getOrUndefined));
978 const message = yield* projectionThreadMessages.getByMessageId({ messageId });
979 return Option.filter(message, (entry) => entry.threadId === threadId).pipe(
980 Option.getOrUndefined,
981 );
1031982 });
1032983
1033984 const rememberAssistantMessageId = (threadId: ThreadId, turnId: TurnId, messageId: MessageId) =>
......@@ -1195,15 +1146,6 @@ const make = Effect.gen(function* () {
11951146 }),
11961147 );
11971148
1198 const takeBufferedProposedPlan = (planId: string) =>
1199 Cache.getOption(bufferedProposedPlanById, planId).pipe(
1200 Effect.flatMap((existingEntry) =>
1201 Cache.invalidate(bufferedProposedPlanById, planId).pipe(
1202 Effect.as(Option.getOrUndefined(existingEntry)),
1203 ),
1204 ),
1205 );
1206
12071149 const clearBufferedProposedPlan = (planId: string) =>
12081150 Cache.invalidate(bufferedProposedPlanById, planId);
12091151
......@@ -1357,83 +1299,45 @@ const make = Effect.gen(function* () {
13571299 }
13581300 });
13591301
1360 const upsertProposedPlan = (input: {
1361 event: ProviderRuntimeEvent;
1362 threadId: ThreadId;
1363 threadProposedPlans: ReadonlyArray<{
1364 id: string;
1365 createdAt: string;
1366 implementedAt: string | null;
1367 implementationThreadId: ThreadId | null;
1368 }>;
1369 planId: string;
1370 turnId?: TurnId;
1371 planMarkdown: string | undefined;
1372 createdAt: string;
1373 updatedAt: string;
1374 }) =>
1375 Effect.gen(function* () {
1376 const planMarkdown = normalizeProposedPlanMarkdown(input.planMarkdown);
1377 if (!planMarkdown) {
1378 return;
1379 }
1380
1381 const existingPlan = findProposedPlanById(input.threadProposedPlans, input.planId);
1382 yield* orchestrationEngine.dispatch({
1383 type: "thread.proposed-plan.upsert",
1384 commandId: yield* providerCommandId(input.event, "proposed-plan-upsert"),
1385 threadId: input.threadId,
1386 proposedPlan: {
1387 id: input.planId,
1388 turnId: input.turnId ?? null,
1389 planMarkdown,
1390 implementedAt: existingPlan?.implementedAt ?? null,
1391 implementationThreadId: existingPlan?.implementationThreadId ?? null,
1392 createdAt: existingPlan?.createdAt ?? input.createdAt,
1393 updatedAt: input.updatedAt,
1394 },
1395 createdAt: input.updatedAt,
1396 });
1397 });
1398
1399 const finalizeBufferedProposedPlan = (input: {
1302 const finalizeBufferedProposedPlan = Effect.fn("finalizeBufferedProposedPlan")(function* (input: {
14001303 event: ProviderRuntimeEvent;
14011304 threadId: ThreadId;
1402 threadProposedPlans: ReadonlyArray<{
1403 id: string;
1404 createdAt: string;
1405 implementedAt: string | null;
1406 implementationThreadId: ThreadId | null;
1407 }>;
14081305 planId: string;
14091306 turnId?: TurnId;
14101307 fallbackMarkdown?: string;
14111308 updatedAt: string;
1412 }) =>
1413 Effect.gen(function* () {
1414 const bufferedPlan = yield* takeBufferedProposedPlan(input.planId);
1415 const bufferedMarkdown = normalizeProposedPlanMarkdown(bufferedPlan?.text);
1416 const fallbackMarkdown = normalizeProposedPlanMarkdown(input.fallbackMarkdown);
1417 const planMarkdown = bufferedMarkdown ?? fallbackMarkdown;
1418 if (!planMarkdown) {
1419 return;
1420 }
1309 }) {
1310 const bufferedPlan = Option.getOrUndefined(
1311 yield* Cache.getOption(bufferedProposedPlanById, input.planId),
1312 );
1313 const planMarkdown =
1314 normalizeProposedPlanMarkdown(bufferedPlan?.text) ??
1315 normalizeProposedPlanMarkdown(input.fallbackMarkdown);
1316 if (!planMarkdown) return yield* clearBufferedProposedPlan(input.planId);
14211317
1422 yield* upsertProposedPlan({
1423 event: input.event,
1318 const existingPlan = Option.getOrUndefined(
1319 yield* projectionThreadProposedPlans.getByPlanId({
14241320 threadId: input.threadId,
1425 threadProposedPlans: input.threadProposedPlans,
1426 planId: input.planId,
1427 ...(input.turnId ? { turnId: input.turnId } : {}),
1321 planId: OrchestrationProposedPlanId.make(input.planId),
1322 }),
1323 );
1324 yield* orchestrationEngine.dispatch({
1325 type: "thread.proposed-plan.upsert",
1326 commandId: yield* providerCommandId(input.event, "proposed-plan-upsert"),
1327 threadId: input.threadId,
1328 proposedPlan: {
1329 id: input.planId,
1330 turnId: input.turnId ?? null,
14281331 planMarkdown,
1429 createdAt:
1430 bufferedPlan?.createdAt && bufferedPlan.createdAt.length > 0
1431 ? bufferedPlan.createdAt
1432 : input.updatedAt,
1332 implementedAt: existingPlan?.implementedAt ?? null,
1333 implementationThreadId: existingPlan?.implementationThreadId ?? null,
1334 createdAt: existingPlan?.createdAt ?? (bufferedPlan?.createdAt || input.updatedAt),
14331335 updatedAt: input.updatedAt,
1434 });
1435 yield* clearBufferedProposedPlan(input.planId);
1336 },
1337 createdAt: input.updatedAt,
14361338 });
1339 yield* clearBufferedProposedPlan(input.planId);
1340 });
14371341
14381342 const clearTurnStateForSession = (threadId: ThreadId) =>
14391343 Effect.gen(function* () {
......@@ -1538,8 +1442,13 @@ const make = Effect.gen(function* () {
15381442 implementationThreadId: ThreadId,
15391443 implementedAt: string,
15401444 ) {
1541 const sourceThread = yield* resolveThreadDetail(sourceThreadId);
1542 const sourcePlan = sourceThread?.proposedPlans.find((entry) => entry.id === sourcePlanId);
1445 const sourceThread = yield* resolveThreadRuntimeContext(sourceThreadId);
1446 const sourcePlan = Option.getOrUndefined(
1447 yield* projectionThreadProposedPlans.getByPlanId({
1448 threadId: sourceThreadId,
1449 planId: sourcePlanId,
1450 }),
1451 );
15431452 if (!sourceThread || !sourcePlan || sourcePlan.implementedAt !== null) {
15441453 return;
15451454 }
......@@ -1550,9 +1459,12 @@ const make = Effect.gen(function* () {
15501459 commandId: CommandId.make(
15511460 `provider:source-proposed-plan-implemented:${implementationThreadId}:${commandUuid}`,
15521461 ),
1553 threadId: sourceThread.id,
1462 threadId: sourceThreadId,
15541463 proposedPlan: {
1555 ...sourcePlan,
1464 id: sourcePlan.planId,
1465 turnId: sourcePlan.turnId,
1466 planMarkdown: sourcePlan.planMarkdown,
1467 createdAt: sourcePlan.createdAt,
15561468 implementedAt,
15571469 implementationThreadId,
15581470 updatedAt: implementedAt,
......@@ -1571,16 +1483,6 @@ const make = Effect.gen(function* () {
15711483 const thread = yield* resolveThreadRuntimeContext(event.threadId);
15721484 if (!thread) return;
15731485
1574 let loadedThreadDetail: OrchestrationThread | null | undefined;
1575 const getLoadedThreadDetail = () =>
1576 Effect.gen(function* () {
1577 if (loadedThreadDetail !== undefined) {
1578 return loadedThreadDetail;
1579 }
1580 loadedThreadDetail = (yield* resolveThreadDetail(thread.id)) ?? null;
1581 return loadedThreadDetail;
1582 });
1583
15841486 const now = event.createdAt;
15851487 const eventTurnId = toTurnId(event.turnId);
15861488 const activeTurnId = thread.session?.activeTurnId ?? null;
......@@ -1800,7 +1702,11 @@ const make = Effect.gen(function* () {
18001702 ? toTurnId(event.turnId)
18011703 : undefined;
18021704 if (pauseForUserTurnId) {
1803 const detailedThread = yield* getLoadedThreadDetail();
1705 const hasProjectedMessage = yield* projectionThreadMessages.hasAssistantMessageForTurn({
1706 threadId: thread.id,
1707 turnId: pauseForUserTurnId,
1708 streamingOnly: true,
1709 });
18041710 const assistantDeliveryMode: AssistantDeliveryMode = yield* Effect.map(
18051711 serverSettingsService.getSettings,
18061712 (settings) => (settings.enableLegacyTokenStreaming ? "streaming" : "buffered"),
......@@ -1831,11 +1737,7 @@ const make = Effect.gen(function* () {
18311737 event.type === "request.opened"
18321738 ? "assistant-delta-finalize-on-request-opened"
18331739 : "assistant-delta-finalize-on-user-input-requested",
1834 hasProjectedMessage:
1835 detailedThread !== null &&
1836 hasAssistantMessageForTurn(detailedThread.messages, pauseForUserTurnId, {
1837 streamingOnly: true,
1838 }),
1740 hasProjectedMessage,
18391741 flushedMessageIds,
18401742 });
18411743 }
......@@ -1864,19 +1766,24 @@ const make = Effect.gen(function* () {
18641766 : undefined;
18651767
18661768 if (assistantCompletion) {
1867 const detailedThread = yield* getLoadedThreadDetail();
1868 const messages = detailedThread?.messages ?? [];
18691769 const turnId = toTurnId(event.turnId);
18701770 const activeAssistantMessageId = turnId
18711771 ? yield* getActiveAssistantMessageIdForTurn(thread.id, turnId)
18721772 : Option.none<MessageId>();
1873 const hasAssistantMessagesForTurn =
1874 turnId !== undefined ? hasAssistantMessageForTurn(messages, turnId) : false;
18751773 const assistantMessageId = Option.getOrElse(
18761774 activeAssistantMessageId,
18771775 () => assistantCompletion.messageId,
18781776 );
1879 const existingAssistantMessage = findMessageById(messages, assistantMessageId);
1777 const [existingAssistantMessage, hasAssistantMessagesForTurn] = yield* Effect.all([
1778 getThreadMessageById(thread.id, assistantMessageId),
1779 turnId === undefined
1780 ? Effect.succeed(false)
1781 : projectionThreadMessages.hasAssistantMessageForTurn({
1782 threadId: thread.id,
1783 turnId,
1784 streamingOnly: false,
1785 }),
1786 ]);
18801787 const shouldApplyFallbackCompletionText =
18811788 !existingAssistantMessage || existingAssistantMessage.text.length === 0;
18821789
......@@ -1916,11 +1823,9 @@ const make = Effect.gen(function* () {
19161823 }
19171824
19181825 if (proposedPlanCompletion) {
1919 const detailedThread = yield* getLoadedThreadDetail();
19201826 yield* finalizeBufferedProposedPlan({
19211827 event,
19221828 threadId: thread.id,
1923 threadProposedPlans: detailedThread?.proposedPlans ?? [],
19241829 planId: proposedPlanCompletion.planId,
19251830 ...(proposedPlanCompletion.turnId ? { turnId: proposedPlanCompletion.turnId } : {}),
19261831 fallbackMarkdown: proposedPlanCompletion.planMarkdown,
......@@ -1929,9 +1834,6 @@ const make = Effect.gen(function* () {
19291834 }
19301835
19311836 if (isTerminalTurn) {
1932 const detailedThread = yield* getLoadedThreadDetail();
1933 const messages = detailedThread?.messages ?? [];
1934 const proposedPlans = detailedThread?.proposedPlans ?? [];
19351837 const turnId = toTurnId(event.turnId);
19361838 if (turnId) {
19371839 const userInputActivities =
......@@ -1979,16 +1881,20 @@ const make = Effect.gen(function* () {
19791881 yield* Effect.forEach(
19801882 assistantMessageIds,
19811883 (assistantMessageId) =>
1982 finalizeAssistantMessage({
1983 event,
1984 threadId: thread.id,
1985 messageId: assistantMessageId,
1986 turnId,
1987 createdAt: now,
1988 commandTag: "assistant-complete-finalize",
1989 finalDeltaCommandTag: "assistant-delta-finalize-fallback",
1990 hasProjectedMessage: findMessageById(messages, assistantMessageId) !== undefined,
1991 }),
1884 getThreadMessageById(thread.id, assistantMessageId).pipe(
1885 Effect.flatMap((existingMessage) =>
1886 finalizeAssistantMessage({
1887 event,
1888 threadId: thread.id,
1889 messageId: assistantMessageId,
1890 turnId,
1891 createdAt: now,
1892 commandTag: "assistant-complete-finalize",
1893 finalDeltaCommandTag: "assistant-delta-finalize-fallback",
1894 hasProjectedMessage: existingMessage !== undefined,
1895 }),
1896 ),
1897 ),
19921898 { concurrency: 1 },
19931899 ).pipe(Effect.asVoid);
19941900 yield* clearAssistantMessageIdsForTurn(thread.id, turnId);
......@@ -1997,7 +1903,6 @@ const make = Effect.gen(function* () {
19971903 yield* finalizeBufferedProposedPlan({
19981904 event,
19991905 threadId: thread.id,
2000 threadProposedPlans: proposedPlans,
20011906 planId: proposedPlanIdForTurn(thread.id, turnId),
20021907 turnId,
20031908 updatedAt: now,
......@@ -2153,8 +2058,17 @@ const make = Effect.gen(function* () {
21532058 if (event.type === "task.completed") {
21542059 taskTitle = yield* lookupTaskDescription(thread.id, event.payload.taskId);
21552060 if (!taskTitle) {
2156 const threadDetail = yield* resolveThreadDetail(thread.id, TASK_TITLE_ACTIVITY_KINDS);
2157 taskTitle = findTaskTitleInActivities(threadDetail?.activities, event.payload.taskId);
2061 const taskActivity = yield* projectionThreadActivityRepository.getLatestTaskActivity({
2062 threadId: thread.id,
2063 taskId: event.payload.taskId,
2064 });
2065 taskTitle = findTaskTitleInActivities(
2066 Option.match(taskActivity, {
2067 onNone: () => undefined,
2068 onSome: (activity) => [activity],
2069 }),
2070 event.payload.taskId,
2071 );
21582072 }
21592073 }
21602074
......@@ -2172,8 +2086,9 @@ const make = Effect.gen(function* () {
21722086 DateTime.makeUnsafe(pendingTurnStart.value.requestedAt),
21732087 )
21742088 ) {
2175 const pendingMessage = (yield* getLoadedThreadDetail())?.messages.find(
2176 (message) => message.id === pendingTurnStart.value.messageId,
2089 const pendingMessage = yield* getThreadMessageById(
2090 thread.id,
2091 pendingTurnStart.value.messageId,
21772092 );
21782093 if (
21792094 pendingMessage?.role === "user" &&
......@@ -2192,11 +2107,13 @@ const make = Effect.gen(function* () {
21922107 (activityEvent.payload.beforeTokens === undefined ||
21932108 activityEvent.payload.afterTokens === undefined)
21942109 ) {
2195 const threadDetail = yield* resolveThreadDetail(thread.id, [
2196 "context-window.updated",
2197 "context-compaction",
2198 ]);
2199 const tokenCounts = compactedTokenCountsFromActivities(threadDetail?.activities);
2110 const activities = yield* projectionThreadActivityRepository.listByThreadId({
2111 threadId: thread.id,
2112 activityKinds: ["context-window.updated", "context-compaction"],
2113 // Preserve the previous thread-detail read's context-history bound.
2114 limit: 500,
2115 });
2116 const tokenCounts = compactedTokenCountsFromActivities(activities);
22002117 if (tokenCounts) {
22012118 activityEvent = {
22022119 ...activityEvent,
......@@ -2274,6 +2191,8 @@ export const ProviderRuntimeIngestionLive = Layer.effect(
22742191 ProviderRuntimeIngestionService,
22752192 make,
22762193).pipe(
2277 Layer.provide(ProjectionTurnRepositoryLive),
22782194 Layer.provide(ProjectionThreadActivityRepositoryLive),
2195 Layer.provide(ProjectionThreadMessageRepositoryLive),
2196 Layer.provide(ProjectionThreadProposedPlanRepositoryLive),
2197 Layer.provide(ProjectionTurnRepositoryLive),
22792198);
apps/server/src/persistence/Layers/ProjectionRepositories.test.ts+47-1
......@@ -1,4 +1,10 @@
1import { ProjectId, ThreadId, TurnId, ProviderInstanceId } from "@t3tools/contracts";
1import {
2 ProjectId,
3 ThreadId,
4 TurnId,
5 ProviderInstanceId,
6 OrchestrationProposedPlanId,
7} from "@t3tools/contracts";
28import { assert, it } from "@effect/vitest";
39import * as Effect from "effect/Effect";
410import * as Layer from "effect/Layer";
......@@ -229,6 +235,46 @@ projectionRepositoriesLayer("Projection repositories", (it) => {
229235 }),
230236 );
231237
238 it.effect("reads only the requested plan in its thread", () =>
239 Effect.gen(function* () {
240 const plans = yield* ProjectionThreadProposedPlanRepository;
241 const sql = yield* SqlClient.SqlClient;
242 const threadId = ThreadId.make("plan-query-thread");
243 const planId = OrchestrationProposedPlanId.make("plan-query-target");
244 yield* plans.upsert({
245 planId,
246 threadId,
247 turnId: null,
248 planMarkdown: "Keep this plan",
249 implementedAt: "2026-03-01T00:01:00.000Z",
250 implementationThreadId: ThreadId.make("implementation-thread"),
251 createdAt: "2026-03-01T00:00:00.000Z",
252 updatedAt: "2026-03-01T00:01:00.000Z",
253 });
254 // An unrelated old row must not be loaded or decoded by the exact lookup.
255 yield* sql`
256 INSERT INTO projection_thread_proposed_plans (
257 plan_id, thread_id, turn_id, plan_markdown, implemented_at,
258 implementation_thread_id, created_at, updated_at
259 ) VALUES (
260 'unrelated-plan', ${threadId}, NULL, '', NULL, NULL,
261 '2026-03-01T00:00:00.000Z', '2026-03-01T00:00:00.000Z'
262 )
263 `;
264 const plan = Option.getOrThrow(yield* plans.getByPlanId({ threadId, planId }));
265 assert.equal(plan.planMarkdown, "Keep this plan");
266 assert.equal(plan.implementedAt, "2026-03-01T00:01:00.000Z");
267 assert.isTrue(
268 Option.isNone(
269 yield* plans.getByPlanId({
270 threadId: ThreadId.make("another-thread"),
271 planId,
272 }),
273 ),
274 );
275 }),
276 );
277
232278 it.effect("stores SQL NULL for missing project model options", () =>
233279 Effect.gen(function* () {
234280 const projects = yield* ProjectionProjectRepository;
apps/server/src/persistence/Layers/ProjectionThreadActivities.test.ts created+100
......@@ -0,0 +1,100 @@
1import { EventId, ThreadId } from "@t3tools/contracts";
2import { assert, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as SqlClient from "effect/unstable/sql/SqlClient";
6
7import { ProjectionThreadActivityRepository } from "../Services/ProjectionThreadActivities.ts";
8import { ProjectionThreadActivityRepositoryLive } from "./ProjectionThreadActivities.ts";
9import { SqlitePersistenceMemory } from "./Sqlite.ts";
10
11const layer = it.layer(
12 ProjectionThreadActivityRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
13);
14
15layer("ProjectionThreadActivityRepository", (it) => {
16 it.effect("reads only the latest matching task activity", () =>
17 Effect.gen(function* () {
18 const repository = yield* ProjectionThreadActivityRepository;
19 const sql = yield* SqlClient.SqlClient;
20 const threadId = ThreadId.make("thread-latest-task-activity");
21
22 yield* sql`
23 INSERT INTO projection_thread_activities (
24 activity_id, thread_id, turn_id, tone, kind, summary, payload_json, sequence, created_at
25 )
26 VALUES
27 (
28 'activity-task-unrelated-tool', ${threadId}, NULL, 'tool', 'tool.completed',
29 'large tool output', 'not-json', 1, '2026-03-01T00:00:00.000Z'
30 ),
31 (
32 'activity-task-started', ${threadId}, NULL, 'info', 'task.started',
33 'started', '{"taskId":"task-1","title":"Initial title"}', 2,
34 '2026-03-01T00:00:01.000Z'
35 ),
36 (
37 'activity-task-progress', ${threadId}, NULL, 'info', 'task.progress',
38 'progress', '{"taskId":"task-1","title":"Updated title"}', 3,
39 '2026-03-01T00:00:02.000Z'
40 ),
41 (
42 'activity-task-other', ${threadId}, NULL, 'info', 'task.progress',
43 'other', '{"taskId":"task-2","title":"Other title"}', 4,
44 '2026-03-01T00:00:03.000Z'
45 )
46 `;
47
48 yield* repository.upsert({
49 activityId: EventId.make("activity-task-untitled"),
50 threadId,
51 turnId: null,
52 tone: "info",
53 kind: "task.progress",
54 summary: "Still running",
55 payload: { taskId: "task-1" },
56 sequence: 5,
57 createdAt: "2026-03-01T00:00:04.000Z",
58 });
59 yield* repository.upsert({
60 activityId: EventId.make("activity-task-blank-title"),
61 threadId,
62 turnId: null,
63 tone: "info",
64 kind: "task.progress",
65 summary: "Still running",
66 payload: { taskId: "task-1", title: " \t\n\u00a0" },
67 sequence: 6,
68 createdAt: "2026-03-01T00:00:05.000Z",
69 });
70
71 const recent = yield* repository.listByThreadId({
72 threadId,
73 activityKinds: ["task.progress"],
74 limit: 2,
75 });
76 assert.deepEqual(
77 recent.map((entry) => entry.activityId),
78 ["activity-task-untitled", "activity-task-blank-title"],
79 );
80
81 const activity = yield* repository.getLatestTaskActivity({
82 threadId,
83 taskId: "task-1",
84 });
85 assert.equal(activity._tag, "Some");
86 if (activity._tag === "Some") {
87 assert.equal(activity.value.activityId, EventId.make("activity-task-progress"));
88 assert.deepEqual(activity.value.payload, {
89 taskId: "task-1",
90 title: "Updated title",
91 });
92 }
93
94 assert.equal(
95 (yield* repository.getLatestTaskActivity({ threadId, taskId: "missing" }))._tag,
96 "None",
97 );
98 }),
99 );
100});
apps/server/src/persistence/Layers/ProjectionThreadActivities.ts+71-10
......@@ -3,6 +3,7 @@ import * as SqlSchema from "effect/unstable/sql/SqlSchema";
33import { NonNegativeInt } from "@t3tools/contracts";
44import * as Effect from "effect/Effect";
55import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";
67import * as Schema from "effect/Schema";
78import * as Struct from "effect/Struct";
89
......@@ -11,6 +12,7 @@ import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts";
1112import {
1213 DeleteProjectionThreadActivitiesInput,
1314 ListProjectionThreadActivitiesInput,
15 GetLatestProjectionThreadTaskActivityInput,
1416 ProjectionThreadActivity,
1517 ProjectionThreadActivityRepository,
1618 type ProjectionThreadActivityRepositoryShape,
......@@ -23,10 +25,10 @@ const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
2325 }),
2426);
2527
26const mapActivityRows = (
27 rows: ReadonlyArray<Schema.Schema.Type<typeof ProjectionThreadActivityDbRowSchema>>,
28): ReadonlyArray<ProjectionThreadActivity> =>
29 rows.map((row) => ({
28function toProjectionThreadActivity(
29 row: Schema.Schema.Type<typeof ProjectionThreadActivityDbRowSchema>,
30): ProjectionThreadActivity {
31 return {
3032 activityId: row.activityId,
3133 threadId: row.threadId,
3234 turnId: row.turnId,
......@@ -36,7 +38,8 @@ const mapActivityRows = (
3638 payload: row.payload,
3739 ...(row.sequence !== null ? { sequence: row.sequence } : {}),
3840 createdAt: row.createdAt,
39 }));
41 };
42}
4043
4144function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
4245 return (cause: unknown) =>
......@@ -45,6 +48,10 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st
4548 : toPersistenceSqlError(sqlOperation)(cause);
4649}
4750
51// Match String.trim so blank saved titles cannot hide an earlier task name.
52const taskTitleWhitespace =
53 "\u0009\u000a\u000b\u000c\u000d\u0020\u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000\ufeff";
54
4855const makeProjectionThreadActivityRepository = Effect.gen(function* () {
4956 const sql = yield* SqlClient.SqlClient;
5057
......@@ -90,7 +97,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
9097 const listProjectionThreadActivityRows = SqlSchema.findAll({
9198 Request: ListProjectionThreadActivitiesInput,
9299 Result: ProjectionThreadActivityDbRowSchema,
93 execute: ({ threadId }) =>
100 execute: ({ threadId, activityKinds, limit }) =>
94101 sql`
95102 SELECT
96103 activity_id AS "activityId",
......@@ -102,8 +109,14 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
102109 payload_json AS "payload",
103110 sequence,
104111 created_at AS "createdAt"
105 FROM projection_thread_activities
106 WHERE thread_id = ${threadId}
112 FROM (
113 SELECT *
114 FROM projection_thread_activities
115 WHERE thread_id = ${threadId}
116 ${activityKinds === undefined ? sql`` : sql`AND ${sql.in("kind", activityKinds)}`}
117 ORDER BY sequence DESC, created_at DESC, activity_id DESC
118 ${limit === undefined ? sql`` : sql`LIMIT ${limit}`}
119 ) AS recent_activities
107120 ORDER BY
108121 CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC,
109122 sequence ASC,
......@@ -142,6 +155,40 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
142155 `,
143156 });
144157
158 const getLatestProjectionThreadTaskActivityRow = SqlSchema.findOneOption({
159 Request: GetLatestProjectionThreadTaskActivityInput,
160 Result: ProjectionThreadActivityDbRowSchema,
161 execute: ({ threadId, taskId }) =>
162 sql`
163 SELECT
164 activity_id AS "activityId",
165 thread_id AS "threadId",
166 turn_id AS "turnId",
167 tone,
168 kind,
169 summary,
170 payload_json AS "payload",
171 sequence,
172 created_at AS "createdAt"
173 FROM projection_thread_activities
174 WHERE thread_id = ${threadId}
175 AND kind IN ('task.started', 'task.progress')
176 AND json_extract(payload_json, '$.taskId') = ${taskId}
177 AND length(trim(
178 CASE
179 WHEN json_type(payload_json, '$.title') = 'text'
180 THEN json_extract(payload_json, '$.title')
181 WHEN kind = 'task.started' AND json_type(payload_json, '$.detail') = 'text'
182 THEN json_extract(payload_json, '$.detail')
183 ELSE ''
184 END,
185 ${taskTitleWhitespace}
186 )) > 0
187 ORDER BY sequence DESC, created_at DESC, activity_id DESC
188 LIMIT 1
189 `,
190 });
191
145192 const deleteProjectionThreadActivityRows = SqlSchema.void({
146193 Request: DeleteProjectionThreadActivitiesInput,
147194 execute: ({ threadId }) =>
......@@ -169,7 +216,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
169216 "ProjectionThreadActivityRepository.listByThreadId:decodeRows",
170217 ),
171218 ),
172 Effect.map(mapActivityRows),
219 Effect.map((rows) => rows.map(toProjectionThreadActivity)),
173220 );
174221
175222 const listUserInputLifecycleByThreadId: ProjectionThreadActivityRepositoryShape["listUserInputLifecycleByThreadId"] =
......@@ -181,9 +228,22 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
181228 "ProjectionThreadActivityRepository.listUserInputLifecycleByThreadId:decodeRows",
182229 ),
183230 ),
184 Effect.map(mapActivityRows),
231 Effect.map((rows) => rows.map(toProjectionThreadActivity)),
185232 );
186233
234 const getLatestTaskActivity: ProjectionThreadActivityRepositoryShape["getLatestTaskActivity"] = (
235 input,
236 ) =>
237 getLatestProjectionThreadTaskActivityRow(input).pipe(
238 Effect.mapError(
239 toPersistenceSqlOrDecodeError(
240 "ProjectionThreadActivityRepository.getLatestTaskActivity:query",
241 "ProjectionThreadActivityRepository.getLatestTaskActivity:decodeRow",
242 ),
243 ),
244 Effect.map(Option.map(toProjectionThreadActivity)),
245 );
246
187247 const deleteByThreadId: ProjectionThreadActivityRepositoryShape["deleteByThreadId"] = (input) =>
188248 deleteProjectionThreadActivityRows(input).pipe(
189249 Effect.mapError(
......@@ -195,6 +255,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
195255 upsert,
196256 listByThreadId,
197257 listUserInputLifecycleByThreadId,
258 getLatestTaskActivity,
198259 deleteByThreadId,
199260 } satisfies ProjectionThreadActivityRepositoryShape;
200261});
apps/server/src/persistence/Layers/ProjectionThreadMessages.test.ts+46-1
......@@ -1,4 +1,4 @@
1import { MessageId, ThreadId } from "@t3tools/contracts";
1import { MessageId, ThreadId, TurnId } from "@t3tools/contracts";
22import { assert, it } from "@effect/vitest";
33import * as Effect from "effect/Effect";
44import * as Layer from "effect/Layer";
......@@ -231,4 +231,49 @@ layer("ProjectionThreadMessageRepository", (it) => {
231231 assert.deepEqual(rows[0]?.attachments, []);
232232 }),
233233 );
234
235 it.effect("checks assistant turn state without hydrating message text", () =>
236 Effect.gen(function* () {
237 const repository = yield* ProjectionThreadMessageRepository;
238 const threadId = ThreadId.make("thread-assistant-turn-state");
239 const turnId = TurnId.make("turn-assistant-state");
240 const createdAt = "2026-03-01T00:00:00.000Z";
241
242 yield* repository.upsert({
243 messageId: MessageId.make("message-assistant-turn-state"),
244 threadId,
245 turnId,
246 role: "assistant",
247 text: "large text that the existence query must not select",
248 isStreaming: false,
249 createdAt,
250 updatedAt: createdAt,
251 });
252
253 assert.equal(
254 yield* repository.hasAssistantMessageForTurn({
255 threadId,
256 turnId,
257 streamingOnly: false,
258 }),
259 true,
260 );
261 assert.equal(
262 yield* repository.hasAssistantMessageForTurn({
263 threadId,
264 turnId,
265 streamingOnly: true,
266 }),
267 false,
268 );
269 assert.equal(
270 yield* repository.hasAssistantMessageForTurn({
271 threadId,
272 turnId: TurnId.make("turn-assistant-state-missing"),
273 streamingOnly: false,
274 }),
275 false,
276 );
277 }),
278 );
234279});
apps/server/src/persistence/Layers/ProjectionThreadMessages.ts+31
......@@ -11,6 +11,7 @@ import { toPersistenceSqlError } from "../Errors.ts";
1111import {
1212 AppendStreamingProjectionThreadMessage,
1313 GetProjectionThreadMessageInput,
14 HasProjectionThreadAssistantMessageInput,
1415 ProjectionThreadMessageRepository,
1516 type ProjectionThreadMessageRepositoryShape,
1617 DeleteProjectionThreadMessagesInput,
......@@ -24,6 +25,7 @@ const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields(
2425 attachments: Schema.NullOr(Schema.fromJsonString(Schema.Array(ChatAttachment))),
2526 }),
2627);
28const ProjectionThreadMessageExistsDbRowSchema = Schema.Struct({ exists: Schema.Number });
2729
2830function toProjectionThreadMessage(
2931 row: Schema.Schema.Type<typeof ProjectionThreadMessageDbRowSchema>,
......@@ -161,6 +163,23 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
161163 `,
162164 });
163165
166 const hasProjectionThreadAssistantMessageRow = SqlSchema.findOne({
167 Request: HasProjectionThreadAssistantMessageInput,
168 Result: ProjectionThreadMessageExistsDbRowSchema,
169 execute: ({ threadId, turnId, streamingOnly }) =>
170 sql`
171 SELECT EXISTS (
172 SELECT 1
173 FROM projection_thread_messages
174 WHERE thread_id = ${threadId}
175 AND turn_id = ${turnId}
176 AND role = 'assistant'
177 AND (${streamingOnly ? 1 : 0} = 0 OR is_streaming = 1)
178 LIMIT 1
179 ) AS "exists"
180 `,
181 });
182
164183 const listProjectionThreadMessageRows = SqlSchema.findAll({
165184 Request: ListProjectionThreadMessagesInput,
166185 Result: ProjectionThreadMessageDbRowSchema,
......@@ -224,6 +243,17 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
224243 Effect.map(Option.map(toProjectionThreadMessage)),
225244 );
226245
246 const hasAssistantMessageForTurn: ProjectionThreadMessageRepositoryShape["hasAssistantMessageForTurn"] =
247 (input) =>
248 hasProjectionThreadAssistantMessageRow(input).pipe(
249 Effect.mapError(
250 toPersistenceSqlError(
251 "ProjectionThreadMessageRepository.hasAssistantMessageForTurn:query",
252 ),
253 ),
254 Effect.map((row) => row.exists === 1),
255 );
256
227257 const listByThreadId: ProjectionThreadMessageRepositoryShape["listByThreadId"] = (input) =>
228258 listProjectionThreadMessageRows(input).pipe(
229259 Effect.mapError(
......@@ -253,6 +283,7 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () {
253283 upsert,
254284 appendStreaming,
255285 getByMessageId,
286 hasAssistantMessageForTurn,
256287 listByThreadId,
257288 getLatestUserMessageAt,
258289 deleteByThreadId,
apps/server/src/persistence/Layers/ProjectionThreadProposedPlans.ts+27
......@@ -8,6 +8,7 @@ import { toPersistenceSqlError } from "../Errors.ts";
88import {
99 DeleteProjectionThreadProposedPlansInput,
1010 HasActionableProjectionThreadProposedPlanInput,
11 GetProjectionThreadProposedPlanInput,
1112 ListProjectionThreadProposedPlansInput,
1213 ProjectionThreadProposedPlan,
1314 ProjectionThreadProposedPlanRepository,
......@@ -52,6 +53,24 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
5253 `,
5354 });
5455
56 const getProjectionThreadProposedPlanRow = SqlSchema.findOneOption({
57 Request: GetProjectionThreadProposedPlanInput,
58 Result: ProjectionThreadProposedPlan,
59 execute: ({ threadId, planId }) => sql`
60 SELECT
61 plan_id AS "planId",
62 thread_id AS "threadId",
63 turn_id AS "turnId",
64 plan_markdown AS "planMarkdown",
65 implemented_at AS "implementedAt",
66 implementation_thread_id AS "implementationThreadId",
67 created_at AS "createdAt",
68 updated_at AS "updatedAt"
69 FROM projection_thread_proposed_plans
70 WHERE thread_id = ${threadId} AND plan_id = ${planId}
71 `,
72 });
73
5574 const listProjectionThreadProposedPlanRows = SqlSchema.findAll({
5675 Request: ListProjectionThreadProposedPlansInput,
5776 Result: ProjectionThreadProposedPlan,
......@@ -133,6 +152,13 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
133152 Effect.mapError(toPersistenceSqlError("ProjectionThreadProposedPlanRepository.upsert:query")),
134153 );
135154
155 const getByPlanId: ProjectionThreadProposedPlanRepositoryShape["getByPlanId"] = (input) =>
156 getProjectionThreadProposedPlanRow(input).pipe(
157 Effect.mapError(
158 toPersistenceSqlError("ProjectionThreadProposedPlanRepository.getByPlanId:query"),
159 ),
160 );
161
136162 const listByThreadId: ProjectionThreadProposedPlanRepositoryShape["listByThreadId"] = (input) =>
137163 listProjectionThreadProposedPlanRows(input).pipe(
138164 Effect.mapError(
......@@ -153,6 +179,7 @@ const makeProjectionThreadProposedPlanRepository = Effect.gen(function* () {
153179 upsert,
154180 listByThreadId,
155181 hasActionableByThreadId,
182 getByPlanId,
156183 deleteByThreadId,
157184 } satisfies ProjectionThreadProposedPlanRepositoryShape;
158185});
apps/server/src/persistence/Services/ProjectionThreadActivities.ts+18-1
......@@ -17,6 +17,7 @@ import {
1717import * as Schema from "effect/Schema";
1818import * as Context from "effect/Context";
1919import type * as Effect from "effect/Effect";
20import type * as Option from "effect/Option";
2021
2122import type { ProjectionRepositoryError } from "../Errors.ts";
2223
......@@ -35,9 +36,18 @@ export type ProjectionThreadActivity = typeof ProjectionThreadActivity.Type;
3536
3637export const ListProjectionThreadActivitiesInput = Schema.Struct({
3738 threadId: ThreadId,
39 activityKinds: Schema.optional(Schema.Array(Schema.String)),
40 limit: Schema.optional(NonNegativeInt),
3841});
3942export type ListProjectionThreadActivitiesInput = typeof ListProjectionThreadActivitiesInput.Type;
4043
44export const GetLatestProjectionThreadTaskActivityInput = Schema.Struct({
45 threadId: ThreadId,
46 taskId: Schema.String,
47});
48export type GetLatestProjectionThreadTaskActivityInput =
49 typeof GetLatestProjectionThreadTaskActivityInput.Type;
50
4151export const DeleteProjectionThreadActivitiesInput = Schema.Struct({
4252 threadId: ThreadId,
4353});
......@@ -61,7 +71,7 @@ export interface ProjectionThreadActivityRepositoryShape {
6171 * List projected thread activity rows for a thread.
6272 *
6373 * Returned in ascending runtime sequence order (or creation order when
64 * sequence is unavailable).
74 * sequence is unavailable). A limit selects the newest matching rows.
6575 */
6676 readonly listByThreadId: (
6777 input: ListProjectionThreadActivitiesInput,
......@@ -76,6 +86,13 @@ export interface ProjectionThreadActivityRepositoryShape {
7686 input: ListProjectionThreadActivitiesInput,
7787 ) => Effect.Effect<ReadonlyArray<ProjectionThreadActivity>, ProjectionRepositoryError>;
7888
89 /**
90 * Read the latest task-start or task-progress activity with a usable title.
91 */
92 readonly getLatestTaskActivity: (
93 input: GetLatestProjectionThreadTaskActivityInput,
94 ) => Effect.Effect<Option.Option<ProjectionThreadActivity>, ProjectionRepositoryError>;
95
7996 /**
8097 * Delete projected thread activity rows by thread.
8198 */
apps/server/src/persistence/Services/ProjectionThreadMessages.ts+15
......@@ -51,6 +51,14 @@ export const GetProjectionThreadMessageInput = Schema.Struct({
5151});
5252export type GetProjectionThreadMessageInput = typeof GetProjectionThreadMessageInput.Type;
5353
54export const HasProjectionThreadAssistantMessageInput = Schema.Struct({
55 threadId: ThreadId,
56 turnId: TurnId,
57 streamingOnly: Schema.Boolean,
58});
59export type HasProjectionThreadAssistantMessageInput =
60 typeof HasProjectionThreadAssistantMessageInput.Type;
61
5462export const DeleteProjectionThreadMessagesInput = Schema.Struct({
5563 threadId: ThreadId,
5664});
......@@ -81,6 +89,13 @@ export interface ProjectionThreadMessageRepositoryShape {
8189 input: GetProjectionThreadMessageInput,
8290 ) => Effect.Effect<Option.Option<ProjectionThreadMessage>, ProjectionRepositoryError>;
8391
92 /**
93 * Check for an assistant message in a turn without hydrating message text.
94 */
95 readonly hasAssistantMessageForTurn: (
96 input: HasProjectionThreadAssistantMessageInput,
97 ) => Effect.Effect<boolean, ProjectionRepositoryError>;
98
8499 /**
85100 * List projected thread messages for a thread.
86101 *
apps/server/src/persistence/Services/ProjectionThreadProposedPlans.ts+11
......@@ -8,6 +8,7 @@ import {
88import * as Schema from "effect/Schema";
99import * as Context from "effect/Context";
1010import type * as Effect from "effect/Effect";
11import type * as Option from "effect/Option";
1112
1213import type { ProjectionRepositoryError } from "../Errors.ts";
1314
......@@ -36,6 +37,12 @@ export const HasActionableProjectionThreadProposedPlanInput = Schema.Struct({
3637export type HasActionableProjectionThreadProposedPlanInput =
3738 typeof HasActionableProjectionThreadProposedPlanInput.Type;
3839
40export const GetProjectionThreadProposedPlanInput = Schema.Struct({
41 threadId: ThreadId,
42 planId: OrchestrationProposedPlanId,
43});
44export type GetProjectionThreadProposedPlanInput = typeof GetProjectionThreadProposedPlanInput.Type;
45
3946export const DeleteProjectionThreadProposedPlansInput = Schema.Struct({
4047 threadId: ThreadId,
4148});
......@@ -46,6 +53,10 @@ export interface ProjectionThreadProposedPlanRepositoryShape {
4653 readonly upsert: (
4754 proposedPlan: ProjectionThreadProposedPlan,
4855 ) => Effect.Effect<void, ProjectionRepositoryError>;
56 /** Read one plan without loading the thread's other plans. */
57 readonly getByPlanId: (
58 input: GetProjectionThreadProposedPlanInput,
59 ) => Effect.Effect<Option.Option<ProjectionThreadProposedPlan>, ProjectionRepositoryError>;
4960 readonly listByThreadId: (
5061 input: ListProjectionThreadProposedPlansInput,
5162 ) => Effect.Effect<ReadonlyArray<ProjectionThreadProposedPlan>, ProjectionRepositoryError>;