[verified] fix(server): harden bounded shell summaries

t3-code[bot] committed 2 months ago

Commit 6dba7cc1 parent cb3e6e68 files · +1132 −191Browse files

Unified · Split

8 files

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +342 −11
apps/server/src/orchestration/Layers/ProjectionPipeline.ts +17 −96
apps/server/src/persistence/Layers/ProjectionThreadActivities.ts +243 −76
apps/server/src/persistence/Layers/ProjectionThreadMessages.ts +69 −8
apps/server/src/persistence/Migrations.ts +2 −0
apps/server/src/persistence/Migrations/042_ProjectionThreadShellSummaryState.test.ts +301 −0
apps/server/src/persistence/Migrations/042_ProjectionThreadShellSummaryState.ts +148 −0
apps/server/src/persistence/Services/ProjectionThreadMessages.ts +10 −0

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts+342−11

@@ -24,6 +24,8 @@ import { · 23 unmodified lines
24	SqlitePersistenceMemory,	24	SqlitePersistenceMemory,
25	} from "../../persistence/Layers/Sqlite.ts";	25	} from "../../persistence/Layers/Sqlite.ts";
26	import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";	26	import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";
		27	import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";
		28	import { ProjectionThreadMessageRepository } from "../../persistence/Services/ProjectionThreadMessages.ts";
27	import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";	29	import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";
28	import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";	30	import { OrchestrationEngineLive } from "./OrchestrationEngine.ts";
29	import {	31	import {
@@ -264,6 +266,14 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 234 unmodified lines
264	`;	266	`;
265	return rows[0]?.count;	267	return rows[0]?.count;
266	});	268	});
		269	const readLatestUserMessageAt = Effect.gen(function* () {
		270	const rows = yield* sql<{ readonly latestUserMessageAt: string \| null }>`
		271	SELECT latest_user_message_at AS "latestUserMessageAt"
		272	FROM projection_threads
		273	WHERE thread_id = ${threadId}
		274	`;
		275	return rows[0]?.latestUserMessageAt;
		276	});
267		277
268	yield* appendAndProject({	278	yield* appendAndProject({
269	type: "project.created",	279	type: "project.created",
@@ -329,18 +339,27 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 59 unmodified lines
329	summary,	339	summary,
330	payload_json,	340	payload_json,
331	sequence,	341	sequence,
332	created_at	342	created_at,
		343	user_input_request_id,
		344	user_input_state
333	)	345	)
334	SELECT	346	SELECT
335	'activity-shell-summary-history-' \|\| printf('%05d', index_value),	347	'activity-shell-summary-history-' \|\| printf('%05d', index_value),
336	${threadId},	348	${threadId},
337	NULL,	349	NULL,
338	'info',	350	'info',
339	'history.item',	351	'user-input.resolved',
340	'Historical activity',	352	'Historical user input',
341	json_object('blob', replace(hex(zeroblob(2048)), '00', 'x')),	353	json_object(
		354	'requestId',
		355	'request-shell-summary-history-' \|\| printf('%05d', index_value),
		356	'blob',
		357	replace(hex(zeroblob(2048)), '00', 'x')
		358	),
342	index_value,	359	index_value,
343	${createdAt}	360	${createdAt},
		361	'request-shell-summary-history-' \|\| printf('%05d', index_value),
		362	'cleared'
344	FROM history	363	FROM history
345	`;	364	`;
346	// If the shell update reads the collection, this sentinel makes the	365	// If the shell update reads the collection, this sentinel makes the
@@ -361,7 +380,7 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 14 unmodified lines
361	${threadId},	380	${threadId},
362	NULL,	381	NULL,
363	'info',	382	'info',
364	'history.sentinel',	383	'user-input.requested',
365	'Must not be decoded',	384	'Must not be decoded',
366	'{',	385	'{',
367	NULL,	386	NULL,
@@ -369,6 +388,13 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 1 unmodified line
369	)	388	)
370	`;	389	`;
371		390
		391	yield* sql`
		392	INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)
		393	VALUES (${threadId}, 0)
		394	ON CONFLICT (thread_id)
		395	DO UPDATE SET pending_count = 0
		396	`;
		397
372	yield* appendAndProject({	398	yield* appendAndProject({
373	type: "thread.message-sent",	399	type: "thread.message-sent",
374	eventId: EventId.make("evt-shell-summary-assistant-message"),	400	eventId: EventId.make("evt-shell-summary-assistant-message"),
@@ -393,6 +419,54 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 18 unmodified lines
393	});	419	});
394	assert.equal(yield* readPendingUserInputCount, 0);	420	assert.equal(yield* readPendingUserInputCount, 0);
395		421
		422	yield* appendAndProject({
		423	type: "thread.message-sent",
		424	eventId: EventId.make("evt-shell-summary-user-message-collision"),
		425	aggregateKind: "thread",
		426	aggregateId: threadId,
		427	occurredAt: "2026-08-19T00:00:01.100Z",
		428	commandId: CommandId.make("cmd-shell-summary-user-message-collision"),
		429	causationEventId: null,
		430	correlationId: CorrelationId.make("cmd-shell-summary-user-message-collision"),
		431	metadata: {},
		432	payload: {
		433	threadId,
		434	messageId: MessageId.make("message-shell-summary-collision"),
		435	role: "user",
		436	text: "A user message",
		437	attachments: [],
		438	turnId: null,
		439	streaming: false,
		440	createdAt: "2026-08-19T00:00:01.100Z",
		441	updatedAt: "2026-08-19T00:00:01.100Z",
		442	},
		443	});
		444	assert.equal(yield* readLatestUserMessageAt, "2026-08-19T00:00:01.100Z");
		445
		446	yield* appendAndProject({
		447	type: "thread.message-sent",
		448	eventId: EventId.make("evt-shell-summary-assistant-message-collision"),
		449	aggregateKind: "thread",
		450	aggregateId: threadId,
		451	occurredAt: "2026-08-19T00:00:01.200Z",
		452	commandId: CommandId.make("cmd-shell-summary-assistant-message-collision"),
		453	causationEventId: null,
		454	correlationId: CorrelationId.make("cmd-shell-summary-assistant-message-collision"),
		455	metadata: {},
		456	payload: {
		457	threadId,
		458	messageId: MessageId.make("message-shell-summary-collision"),
		459	role: "assistant",
		460	text: "The row was replaced",
		461	attachments: [],
		462	turnId: null,
		463	streaming: false,
		464	createdAt: "2026-08-19T00:00:01.200Z",
		465	updatedAt: "2026-08-19T00:00:01.200Z",
		466	},
		467	});
		468	assert.equal(yield* readLatestUserMessageAt, null);
		469
396	yield* appendAndProject({	470	yield* appendAndProject({
397	type: "thread.activity-appended",	471	type: "thread.activity-appended",
398	eventId: EventId.make("evt-shell-summary-routine-activity"),	472	eventId: EventId.make("evt-shell-summary-routine-activity"),
@@ -444,6 +518,58 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 45 unmodified lines
444	});	518	});
445	assert.equal(yield* readPendingUserInputCount, 1);	519	assert.equal(yield* readPendingUserInputCount, 1);
446		520
		521	yield* appendAndProject({
		522	type: "thread.activity-appended",
		523	eventId: EventId.make("evt-shell-summary-user-input-overwritten"),
		524	aggregateKind: "thread",
		525	aggregateId: threadId,
		526	occurredAt: "2026-08-19T00:00:02.500Z",
		527	commandId: CommandId.make("cmd-shell-summary-user-input-overwritten"),
		528	causationEventId: null,
		529	correlationId: CorrelationId.make("cmd-shell-summary-user-input-overwritten"),
		530	metadata: {},
		531	payload: {
		532	threadId,
		533	activity: {
		534	id: EventId.make("activity-shell-summary-user-input-requested"),
		535	tone: "info",
		536	kind: "task.progress",
		537	summary: "Request was replaced",
		538	payload: { taskId: "task-2", status: "running" },
		539	sequence: 3,
		540	turnId: null,
		541	createdAt: "2026-08-19T00:00:02.500Z",
		542	},
		543	},
		544	});
		545	assert.equal(yield* readPendingUserInputCount, 0);
		546
		547	yield* appendAndProject({
		548	type: "thread.activity-appended",
		549	eventId: EventId.make("evt-shell-summary-user-input-requested-again"),
		550	aggregateKind: "thread",
		551	aggregateId: threadId,
		552	occurredAt: "2026-08-19T00:00:02.750Z",
		553	commandId: CommandId.make("cmd-shell-summary-user-input-requested-again"),
		554	causationEventId: null,
		555	correlationId: CorrelationId.make("cmd-shell-summary-user-input-requested-again"),
		556	metadata: {},
		557	payload: {
		558	threadId,
		559	activity: {
		560	id: EventId.make("activity-shell-summary-user-input-requested"),
		561	tone: "info",
		562	kind: "user-input.requested",
		563	summary: "User input requested again",
		564	payload: { requestId: "request-shell-summary-replacement" },
		565	sequence: 4,
		566	turnId: null,
		567	createdAt: "2026-08-19T00:00:02.750Z",
		568	},
		569	},
		570	});
		571	assert.equal(yield* readPendingUserInputCount, 1);
		572
447	yield* appendAndProject({	573	yield* appendAndProject({
448	type: "thread.activity-appended",	574	type: "thread.activity-appended",
449	eventId: EventId.make("evt-shell-summary-user-input-resolved"),	575	eventId: EventId.make("evt-shell-summary-user-input-resolved"),
@@ -461,7 +587,7 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 11 unmodified lines
461	tone: "info",	587	tone: "info",
462	kind: "user-input.resolved",	588	kind: "user-input.resolved",
463	summary: "User input resolved",	589	summary: "User input resolved",
464	payload: { requestId: "request-shell-summary-cost" },	590	payload: { requestId: "request-shell-summary-replacement" },
465	sequence: 1,	591	sequence: 1,
466	turnId: null,	592	turnId: null,
467	createdAt: "2026-08-19T00:00:03.000Z",	593	createdAt: "2026-08-19T00:00:03.000Z",
@@ -470,12 +596,192 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-shel · 2 unmodified lines
470	});	596	});
471	assert.equal(yield* readPendingUserInputCount, 0);	597	assert.equal(yield* readPendingUserInputCount, 0);
472		598
		599	const recoveryActivity = {
		600	id: EventId.make("activity-shell-summary-recovery"),
		601	tone: "info" as const,
		602	kind: "user-input.requested",
		603	summary: "Recovery request",
		604	payload: { requestId: "request-shell-summary-recovery" },
		605	sequence: 5,
		606	turnId: null,
		607	createdAt: "2026-08-19T00:00:04.000Z",
		608	};
		609	yield* appendAndProject({
		610	type: "thread.activity-appended",
		611	eventId: EventId.make("evt-shell-summary-recovery-requested"),
		612	aggregateKind: "thread",
		613	aggregateId: threadId,
		614	occurredAt: "2026-08-19T00:00:04.000Z",
		615	commandId: CommandId.make("cmd-shell-summary-recovery-requested"),
		616	causationEventId: null,
		617	correlationId: CorrelationId.make("cmd-shell-summary-recovery-requested"),
		618	metadata: {},
		619	payload: { threadId, activity: recoveryActivity },
		620	});
		621	assert.equal(yield* readPendingUserInputCount, 1);
		622
		623	yield* sql`
		624	DELETE FROM projection_thread_user_input_summaries
		625	WHERE thread_id = ${threadId}
		626	`;
		627	yield* appendAndProject({
		628	type: "thread.activity-appended",
		629	eventId: EventId.make("evt-shell-summary-recovery-replayed"),
		630	aggregateKind: "thread",
		631	aggregateId: threadId,
		632	occurredAt: "2026-08-19T00:00:04.500Z",
		633	commandId: CommandId.make("cmd-shell-summary-recovery-replayed"),
		634	causationEventId: null,
		635	correlationId: CorrelationId.make("cmd-shell-summary-recovery-replayed"),
		636	metadata: {},
		637	payload: { threadId, activity: recoveryActivity },
		638	});
		639	assert.equal(yield* readPendingUserInputCount, 1);
		640
473	const activityRows = yield* sql<{ readonly count: number }>`	641	const activityRows = yield* sql<{ readonly count: number }>`
474	SELECT COUNT(*) AS "count"	642	SELECT COUNT(*) AS "count"
475	FROM projection_thread_activities	643	FROM projection_thread_activities
476	WHERE thread_id = ${threadId}	644	WHERE thread_id = ${threadId}
477	`;	645	`;
478	assert.deepEqual(activityRows, [{ count: 10004 }]);	646	assert.deepEqual(activityRows, [{ count: 10005 }]);
		647	}),
		648	);
		649
		650	it.effect("refreshes previous owners after cross-thread id replacement", () =>
		651	Effect.gen(function* () {
		652	const sql = yield* SqlClient.SqlClient;
		653	const messages = yield* ProjectionThreadMessageRepository;
		654	const activities = yield* ProjectionThreadActivityRepository;
		655	const threadA = ThreadId.make("thread-shell-summary-owner-a");
		656	const threadB = ThreadId.make("thread-shell-summary-owner-b");
		657
		658	yield* sql`
		659	INSERT INTO projection_threads (
		660	thread_id,
		661	project_id,
		662	title,
		663	model_selection_json,
		664	runtime_mode,
		665	interaction_mode,
		666	branch,
		667	worktree_path,
		668	latest_turn_id,
		669	created_at,
		670	updated_at,
		671	archived_at,
		672	latest_user_message_at,
		673	pending_approval_count,
		674	pending_user_input_count,
		675	has_actionable_proposed_plan,
		676	deleted_at
		677	) VALUES
		678	(
		679	${threadA},
		680	'project-shell-summary-owner',
		681	'Owner A',
		682	'{"provider":"codex","model":"gpt-5-codex"}',
		683	'approval-required',
		684	'default',
		685	NULL,
		686	NULL,
		687	NULL,
		688	'2026-08-19T00:00:00.000Z',
		689	'2026-08-19T00:00:00.000Z',
		690	NULL,
		691	NULL,
		692	0,
		693	0,
		694	0,
		695	NULL
		696	),
		697	(
		698	${threadB},
		699	'project-shell-summary-owner',
		700	'Owner B',
		701	'{"provider":"codex","model":"gpt-5-codex"}',
		702	'approval-required',
		703	'default',
		704	NULL,
		705	NULL,
		706	NULL,
		707	'2026-08-19T00:00:00.000Z',
		708	'2026-08-19T00:00:00.000Z',
		709	NULL,
		710	NULL,
		711	0,
		712	0,
		713	0,
		714	NULL
		715	)
		716	`;
		717
		718	yield* messages.upsert({
		719	messageId: MessageId.make("message-shell-summary-owner-move"),
		720	threadId: threadA,
		721	turnId: null,
		722	role: "user",
		723	text: "Move me",
		724	isStreaming: false,
		725	createdAt: "2026-08-19T00:00:01.000Z",
		726	updatedAt: "2026-08-19T00:00:01.000Z",
		727	});
		728	yield* activities.upsert({
		729	activityId: EventId.make("activity-shell-summary-owner-move"),
		730	threadId: threadA,
		731	turnId: null,
		732	tone: "info",
		733	kind: "user-input.requested",
		734	summary: "Move request",
		735	payload: { requestId: "request-shell-summary-owner-move" },
		736	createdAt: "2026-08-19T00:00:02.000Z",
		737	});
		738
		739	yield* messages.upsert({
		740	messageId: MessageId.make("message-shell-summary-owner-move"),
		741	threadId: threadB,
		742	turnId: null,
		743	role: "assistant",
		744	text: "Moved",
		745	isStreaming: false,
		746	createdAt: "2026-08-19T00:00:03.000Z",
		747	updatedAt: "2026-08-19T00:00:03.000Z",
		748	});
		749	yield* activities.upsert({
		750	activityId: EventId.make("activity-shell-summary-owner-move"),
		751	threadId: threadB,
		752	turnId: null,
		753	tone: "info",
		754	kind: "task.progress",
		755	summary: "Moved request",
		756	payload: { taskId: "task-owner-move", status: "running" },
		757	createdAt: "2026-08-19T00:00:04.000Z",
		758	});
		759
		760	const rows = yield* sql<{
		761	readonly threadId: string;
		762	readonly latestUserMessageAt: string \| null;
		763	readonly pendingUserInputCount: number;
		764	}>`
		765	SELECT
		766	thread_id AS "threadId",
		767	latest_user_message_at AS "latestUserMessageAt",
		768	pending_user_input_count AS "pendingUserInputCount"
		769	FROM projection_threads
		770	WHERE thread_id IN (${threadA}, ${threadB})
		771	ORDER BY thread_id
		772	`;
		773	assert.deepStrictEqual(rows, [
		774	{
		775	threadId: threadA,
		776	latestUserMessageAt: null,
		777	pendingUserInputCount: 0,
		778	},
		779	{
		780	threadId: threadB,
		781	latestUserMessageAt: null,
		782	pendingUserInputCount: 0,
		783	},
		784	]);
479	}),	785	}),
480	);	786	);
481	},	787	},
@@ -2261,7 +2567,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { · 1779 unmodified lines
2261	payload: {	2567	payload: {
2262	threadId: ThreadId.make("thread-stale-user-input"),	2568	threadId: ThreadId.make("thread-stale-user-input"),
2263	activity: {	2569	activity: {
2264	id: EventId.make("activity-stale-user-input-requested"),	2570	id: EventId.make("A"),
2265	tone: "info",	2571	tone: "info",
2266	kind: "user-input.requested",	2572	kind: "user-input.requested",
2267	summary: "User input requested",	2573	summary: "User input requested",
@@ -2300,7 +2606,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { · 32 unmodified lines
2300	payload: {	2606	payload: {
2301	threadId: ThreadId.make("thread-stale-user-input"),	2607	threadId: ThreadId.make("thread-stale-user-input"),
2302	activity: {	2608	activity: {
2303	id: EventId.make("activity-stale-user-input-failed"),	2609	id: EventId.make("a"),
2304	tone: "error",	2610	tone: "error",
2305	kind: "provider.user-input.respond.failed",	2611	kind: "provider.user-input.respond.failed",
2306	summary: "Provider user input response failed",	2612	summary: "Provider user input response failed",
@@ -2310,7 +2616,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { · 3 unmodified lines
2310	"Provider adapter request failed (codex) for item/tool/requestUserInput: Unknown pending Codex user input request: user-input-request-stale-1",	2616	"Provider adapter request failed (codex) for item/tool/requestUserInput: Unknown pending Codex user input request: user-input-request-stale-1",
2311	},	2617	},
2312	turnId: null,	2618	turnId: null,
2313	createdAt: "2026-02-26T12:35:03.000Z",	2619	createdAt: "2026-02-26T12:35:02.000Z",
2314	},	2620	},
2315	},	2621	},
2316	});	2622	});
@@ -2323,6 +2629,31 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { · 6 unmodified lines
2323	WHERE thread_id = 'thread-stale-user-input'	2629	WHERE thread_id = 'thread-stale-user-input'
2324	`;	2630	`;
2325	assert.deepEqual(threadRows, [{ pendingUserInputCount: 0 }]);	2631	assert.deepEqual(threadRows, [{ pendingUserInputCount: 0 }]);
		2632
		2633	yield* appendAndProject({
		2634	type: "thread.reverted",
		2635	eventId: EventId.make("evt-stale-user-input-5"),
		2636	aggregateKind: "thread",
		2637	aggregateId: ThreadId.make("thread-stale-user-input"),
		2638	occurredAt: "2026-02-26T12:35:04.000Z",
		2639	commandId: CommandId.make("cmd-stale-user-input-5"),
		2640	causationEventId: null,
		2641	correlationId: CorrelationId.make("cmd-stale-user-input-5"),
		2642	metadata: {},
		2643	payload: {
		2644	threadId: ThreadId.make("thread-stale-user-input"),
		2645	turnCount: 0,
		2646	},
		2647	});
		2648
		2649	const revertedThreadRows = yield* sql<{
		2650	readonly pendingUserInputCount: number;
		2651	}>`
		2652	SELECT pending_user_input_count AS "pendingUserInputCount"
		2653	FROM projection_threads
		2654	WHERE thread_id = 'thread-stale-user-input'
		2655	`;
		2656	assert.deepEqual(revertedThreadRows, [{ pendingUserInputCount: 0 }]);
2326	}),	2657	}),
2327	);	2658	);
2328		2659

apps/server/src/orchestration/Layers/ProjectionPipeline.ts+17−96

@@ -130,60 +130,6 @@ function isStalePendingApprovalFailureDetail(detail: string \| null): boolean { · 129 unmodified lines
130	);	130	);
131	}	131	}
132		132
133	function isStalePendingUserInputFailureDetail(detail: string \| null): boolean {
134	if (detail === null) {
135	return false;
136	}
137	return (
138	detail.includes("stale pending user-input request") \|\|
139	detail.includes("unknown pending user-input request") \|\|
140	detail.includes("unknown pending user input request") \|\|
141	detail.includes("unknown pending codex user input request")
142	);
143	}
144
145	function derivePendingUserInputCountFromActivities(
146	activities: ReadonlyArray<ProjectionThreadActivity>,
147	): number {
148	const openRequestIds = new Set<string>();
149	const ordered = [...activities].toSorted(
150	(left, right) =>
151	left.createdAt.localeCompare(right.createdAt) \|\|
152	left.activityId.localeCompare(right.activityId),
153	);
154
155	for (const activity of ordered) {
156	const requestId = extractActivityRequestId(activity.payload);
157	if (requestId === null) {
158	continue;
159	}
160	const payload =
161	typeof activity.payload === "object" && activity.payload !== null
162	? (activity.payload as Record<string, unknown>)
163	: null;
164	const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
165
166	if (activity.kind === "user-input.requested") {
167	openRequestIds.add(requestId);
168	continue;
169	}
170
171	if (activity.kind === "user-input.resolved") {
172	openRequestIds.delete(requestId);
173	continue;
174	}
175
176	if (
177	activity.kind === "provider.user-input.respond.failed" &&
178	isStalePendingUserInputFailureDetail(detail)
179	) {
180	openRequestIds.delete(requestId);
181	}
182	}
183
184	return openRequestIds.size;
185	}
186
187	function deriveHasActionableProposedPlan(input: {	133	function deriveHasActionableProposedPlan(input: {
188	readonly latestTurnId: string \| null;	134	readonly latestTurnId: string \| null;
189	readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;	135	readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
@@ -570,27 +516,18 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti · 380 unmodified lines
570	return;	516	return;
571	}	517	}
572		518
573	const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([	519	const [latestUserMessageAtOption, proposedPlans, pendingUserInputCount, pendingApprovals] =
574	projectionThreadMessageRepository.listByThreadId({ threadId }),	520	yield* Effect.all([
575	projectionThreadProposedPlanRepository.listByThreadId({ threadId }),	521	projectionThreadMessageRepository.getLatestUserMessageAtByThreadId({ threadId }),
576	projectionThreadActivityRepository.listByThreadId({ threadId }),	522	projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
577	projectionPendingApprovalRepository.listByThreadId({ threadId }),	523	projectionThreadActivityRepository.countPendingUserInputsByThreadId({ threadId }),
578	]);	524	projectionPendingApprovalRepository.listByThreadId({ threadId }),
579		525	]);
580	let latestUserMessageAt: string \| null = null;
581	for (const message of messages) {
582	if (
583	message.role === "user" &&
584	(latestUserMessageAt === null \|\| message.createdAt > latestUserMessageAt)
585	) {
586	latestUserMessageAt = message.createdAt;
587	}
588	}
589		526
		527	const latestUserMessageAt = Option.getOrNull(latestUserMessageAtOption);
590	const pendingApprovalCount = pendingApprovals.filter(	528	const pendingApprovalCount = pendingApprovals.filter(
591	(approval) => approval.status === "pending",	529	(approval) => approval.status === "pending",
592	).length;	530	).length;
593	const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
594	const hasActionableProposedPlan = deriveHasActionableProposedPlan({	531	const hasActionableProposedPlan = deriveHasActionableProposedPlan({
595	latestTurnId: existingRow.value.latestTurnId,	532	latestTurnId: existingRow.value.latestTurnId,
596	proposedPlans,	533	proposedPlans,
@@ -877,20 +814,11 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti · 280 unmodified lines
877		814
878	switch (event.type) {	815	switch (event.type) {
879	case "thread.message-sent": {	816	case "thread.message-sent": {
880	if (event.payload.role !== "user") {	817	latestUserMessageAt = Option.getOrNull(
881	break;	818	yield* projectionThreadMessageRepository.getLatestUserMessageAtByThreadId({
882	}	819	threadId: event.payload.threadId,
883	const projectedMessage = yield* projectionThreadMessageRepository.getByMessageId({	820	}),
884	messageId: event.payload.messageId,	821	);
885	});
886	if (
887	Option.isSome(projectedMessage) &&
888	projectedMessage.value.role === "user" &&
889	(latestUserMessageAt === null \|\|
890	projectedMessage.value.createdAt > latestUserMessageAt)
891	) {
892	latestUserMessageAt = projectedMessage.value.createdAt;
893	}
894	break;	822	break;
895	}	823	}
896		824
@@ -931,17 +859,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti · 34 unmodified lines
931	).length;	859	).length;
932	}	860	}
933		861
934	if (	862	pendingUserInputCount =
935	kind === "user-input.requested" \|\|	863	yield* projectionThreadActivityRepository.countPendingUserInputsByThreadId({
936	kind === "user-input.resolved" \|\|	864	threadId: event.payload.threadId,
937	(kind === "provider.user-input.respond.failed" &&	865	});
938	isStalePendingUserInputFailureDetail(detail))
939	) {
940	pendingUserInputCount =
941	yield* projectionThreadActivityRepository.countPendingUserInputsByThreadId({
942	threadId: event.payload.threadId,
943	});
944	}
945	break;	866	break;
946	}	867	}
947		868

apps/server/src/persistence/Layers/ProjectionThreadActivities.ts+243−76

@@ -1,8 +1,9 @@
1	import * as SqlClient from "effect/unstable/sql/SqlClient";	1	import * as SqlClient from "effect/unstable/sql/SqlClient";
2	import * as SqlSchema from "effect/unstable/sql/SqlSchema";	2	import * as SqlSchema from "effect/unstable/sql/SqlSchema";
3	import { NonNegativeInt } from "@t3tools/contracts";	3	import { EventId, NonNegativeInt, ThreadId } from "@t3tools/contracts";
4	import * as Effect from "effect/Effect";	4	import * as Effect from "effect/Effect";
5	import * as Layer from "effect/Layer";	5	import * as Layer from "effect/Layer";
		6	import * as Option from "effect/Option";
6	import * as Schema from "effect/Schema";	7	import * as Schema from "effect/Schema";
7	import * as Struct from "effect/Struct";	8	import * as Struct from "effect/Struct";
8		9
@@ -40,6 +41,58 @@ const PendingUserInputCountRowSchema = Schema.Struct({ · 31 unmodified lines
40	count: NonNegativeInt,	41	count: NonNegativeInt,
41	});	42	});
42		43
		44	const ExistingUserInputActivityRowSchema = Schema.Struct({
		45	threadId: ThreadId,
		46	requestId: Schema.NullOr(Schema.String),
		47	});
		48
		49	const UserInputActivityKeySchema = Schema.Struct({
		50	threadId: ThreadId,
		51	requestId: Schema.String,
		52	});
		53	type UserInputActivityKey = typeof UserInputActivityKeySchema.Type;
		54
		55	const UserInputStateRowSchema = Schema.Struct({
		56	state: Schema.Literals(["pending", "cleared"]),
		57	});
		58	type UserInputState = typeof UserInputStateRowSchema.Type.state;
		59
		60	function userInputStateFromActivity(activity: {
		61	readonly kind: string;
		62	readonly payload: unknown;
		63	}): { readonly requestId: string; readonly state: UserInputState } \| null {
		64	if (typeof activity.payload !== "object" \|\| activity.payload === null) {
		65	return null;
		66	}
		67	const payload = activity.payload as Record<string, unknown>;
		68	if (typeof payload.requestId !== "string") {
		69	return null;
		70	}
		71	if (activity.kind === "user-input.requested") {
		72	return { requestId: payload.requestId, state: "pending" };
		73	}
		74	if (activity.kind === "user-input.resolved") {
		75	return { requestId: payload.requestId, state: "cleared" };
		76	}
		77	if (activity.kind !== "provider.user-input.respond.failed") {
		78	return null;
		79	}
		80	const detail = typeof payload.detail === "string" ? payload.detail.toLowerCase() : "";
		81	if (
		82	!detail.includes("stale pending user-input request") &&
		83	!detail.includes("unknown pending user-input request") &&
		84	!detail.includes("unknown pending user input request") &&
		85	!detail.includes("unknown pending codex user input request")
		86	) {
		87	return null;
		88	}
		89	return { requestId: payload.requestId, state: "cleared" };
		90	}
		91
		92	function userInputKeyId(key: UserInputActivityKey): string {
		93	return `${key.threadId.length}:${key.threadId}${key.requestId}`;
		94	}
		95
43	function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {	96	function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
44	return (cause: unknown) =>	97	return (cause: unknown) =>
45	Schema.isSchemaError(cause)	98	Schema.isSchemaError(cause)
@@ -50,10 +103,41 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st · 4 unmodified lines
50	const makeProjectionThreadActivityRepository = Effect.gen(function* () {	103	const makeProjectionThreadActivityRepository = Effect.gen(function* () {
51	const sql = yield* SqlClient.SqlClient;	104	const sql = yield* SqlClient.SqlClient;
52		105
		106	const getExistingUserInputActivity = SqlSchema.findOneOption({
		107	Request: Schema.Struct({ activityId: EventId }),
		108	Result: ExistingUserInputActivityRowSchema,
		109	execute: ({ activityId }) =>
		110	sql`
		111	SELECT
		112	thread_id AS "threadId",
		113	user_input_request_id AS "requestId"
		114	FROM projection_thread_activities
		115	WHERE activity_id = ${activityId}
		116	AND user_input_state IS NOT NULL
		117	LIMIT 1
		118	`,
		119	});
		120
		121	const getLatestUserInputState = SqlSchema.findOneOption({
		122	Request: UserInputActivityKeySchema,
		123	Result: UserInputStateRowSchema,
		124	execute: ({ threadId, requestId }) =>
		125	sql`
		126	SELECT user_input_state AS "state"
		127	FROM projection_thread_activities
		128	WHERE thread_id = ${threadId}
		129	AND user_input_request_id = ${requestId}
		130	AND user_input_state IS NOT NULL
		131	ORDER BY created_at DESC, activity_id DESC
		132	LIMIT 1
		133	`,
		134	});
		135
53	const upsertProjectionThreadActivityRow = SqlSchema.void({	136	const upsertProjectionThreadActivityRow = SqlSchema.void({
54	Request: ProjectionThreadActivity,	137	Request: ProjectionThreadActivity,
55	execute: (row) =>	138	execute: (row) => {
56	sql`	139	const userInputState = userInputStateFromActivity(row);
		140	return sql`
57	INSERT INTO projection_thread_activities (	141	INSERT INTO projection_thread_activities (
58	activity_id,	142	activity_id,
59	thread_id,	143	thread_id,
@@ -63,7 +147,9 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { · 3 unmodified lines
63	summary,	147	summary,
64	payload_json,	148	payload_json,
65	sequence,	149	sequence,
66	created_at	150	created_at,
		151	user_input_request_id,
		152	user_input_state
67	)	153	)
68	VALUES (	154	VALUES (
69	${row.activityId},	155	${row.activityId},
@@ -74,7 +160,9 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { · 4 unmodified lines
74	${row.summary},	160	${row.summary},
75	${JSON.stringify(row.payload)},	161	${JSON.stringify(row.payload)},
76	${row.sequence ?? null},	162	${row.sequence ?? null},
77	${row.createdAt}	163	${row.createdAt},
		164	${userInputState?.requestId ?? null},
		165	${userInputState?.state ?? null}
78	)	166	)
79	ON CONFLICT (activity_id)	167	ON CONFLICT (activity_id)
80	DO UPDATE SET	168	DO UPDATE SET
@@ -85,8 +173,11 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { · 4 unmodified lines
85	summary = excluded.summary,	173	summary = excluded.summary,
86	payload_json = excluded.payload_json,	174	payload_json = excluded.payload_json,
87	sequence = excluded.sequence,	175	sequence = excluded.sequence,
88	created_at = excluded.created_at	176	created_at = excluded.created_at,
89	`,	177	user_input_request_id = excluded.user_input_request_id,
		178	user_input_state = excluded.user_input_state
		179	`;
		180	},
90	});	181	});
91		182
92	const listProjectionThreadActivityRows = SqlSchema.findAll({	183	const listProjectionThreadActivityRows = SqlSchema.findAll({
@@ -114,68 +205,60 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { · 21 unmodified lines
114	`,	205	`,
115	});	206	});
116		207
117	const readPendingUserInputCount = SqlSchema.findOne({	208	const readPendingUserInputCount = SqlSchema.findOneOption({
118	Request: ListProjectionThreadActivitiesInput,	209	Request: ListProjectionThreadActivitiesInput,
119	Result: PendingUserInputCountRowSchema,	210	Result: PendingUserInputCountRowSchema,
120	execute: ({ threadId }) =>	211	execute: ({ threadId }) =>
121	sql`	212	sql`
122	WITH extracted AS (	213	SELECT pending_count AS "count"
123	SELECT	214	FROM projection_thread_user_input_summaries
124	activity_id,	215	WHERE thread_id = ${threadId}
125	kind,	216	LIMIT 1
126	created_at,	217	`,
127	CASE	218	});
128	WHEN json_type(payload_json, '$.requestId') = 'text'	219
129	THEN json_extract(payload_json, '$.requestId')	220	const rebuildPendingUserInputCount = SqlSchema.findOne({
130	ELSE NULL	221	Request: ListProjectionThreadActivitiesInput,
131	END AS request_id,	222	Result: PendingUserInputCountRowSchema,
132	lower(	223	execute: ({ threadId }) =>
133	CASE	224	sql`
134	WHEN json_type(payload_json, '$.detail') = 'text'	225	WITH ranked AS (
135	THEN json_extract(payload_json, '$.detail')
136	ELSE ''
137	END
138	) AS detail
139	FROM projection_thread_activities
140	WHERE thread_id = ${threadId}
141	AND kind IN (
142	'user-input.requested',
143	'user-input.resolved',
144	'provider.user-input.respond.failed'
145	)
146	),
147	latest_state_change AS (
148	SELECT	226	SELECT
149	kind,	227	user_input_state,
150	ROW_NUMBER() OVER (	228	ROW_NUMBER() OVER (
151	PARTITION BY request_id	229	PARTITION BY user_input_request_id
152	ORDER BY	230	ORDER BY created_at DESC, activity_id DESC
153	created_at DESC,
154	activity_id DESC
155	) AS row_number	231	) AS row_number
156	FROM extracted	232	FROM projection_thread_activities
157	WHERE request_id IS NOT NULL	233	WHERE thread_id = ${threadId}
158	AND length(request_id) > 0	234	AND user_input_request_id IS NOT NULL
159	AND (	235	AND user_input_state IS NOT NULL
160	kind IN ('user-input.requested', 'user-input.resolved')
161	OR (
162	kind = 'provider.user-input.respond.failed'
163	AND (
164	detail LIKE '%stale pending user-input request%'
165	OR detail LIKE '%unknown pending user-input request%'
166	OR detail LIKE '%unknown pending user input request%'
167	OR detail LIKE '%unknown pending codex user input request%'
168	)
169	)
170	)
171	)	236	)
172	SELECT COUNT(*) AS "count"	237	SELECT COUNT(*) AS "count"
173	FROM latest_state_change	238	FROM ranked
174	WHERE row_number = 1	239	WHERE row_number = 1
175	AND kind = 'user-input.requested'	240	AND user_input_state = 'pending'
176	`,	241	`,
177	});	242	});
178		243
		244	const getOrRebuildPendingUserInputCount = Effect.fn(
		245	"ProjectionThreadActivityRepository.getOrRebuildPendingUserInputCount",
		246	)(function* (input: ListProjectionThreadActivitiesInput) {
		247	const existing = yield* readPendingUserInputCount(input);
		248	if (Option.isSome(existing)) {
		249	return existing.value.count;
		250	}
		251
		252	const rebuilt = yield* rebuildPendingUserInputCount(input);
		253	yield* sql`
		254	INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)
		255	VALUES (${input.threadId}, ${rebuilt.count})
		256	ON CONFLICT (thread_id)
		257	DO UPDATE SET pending_count = excluded.pending_count
		258	`;
		259	return rebuilt.count;
		260	});
		261
179	const deleteProjectionThreadActivityRows = SqlSchema.void({	262	const deleteProjectionThreadActivityRows = SqlSchema.void({
180	Request: DeleteProjectionThreadActivitiesInput,	263	Request: DeleteProjectionThreadActivitiesInput,
181	execute: ({ threadId }) =>	264	execute: ({ threadId }) =>
@@ -185,15 +268,86 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { · 3 unmodified lines
185	`,	268	`,
186	});	269	});
187		270
		271	const resetPendingUserInputSummary = SqlSchema.void({
		272	Request: DeleteProjectionThreadActivitiesInput,
		273	execute: ({ threadId }) =>
		274	sql`
		275	INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)
		276	VALUES (${threadId}, 0)
		277	ON CONFLICT (thread_id)
		278	DO UPDATE SET pending_count = 0
		279	`,
		280	});
		281
188	const upsert: ProjectionThreadActivityRepositoryShape["upsert"] = (row) =>	282	const upsert: ProjectionThreadActivityRepositoryShape["upsert"] = (row) =>
189	upsertProjectionThreadActivityRow(row).pipe(	283	sql
190	Effect.mapError(	284	.withTransaction(
191	toPersistenceSqlOrDecodeError(	285	Effect.gen(function* () {
192	"ProjectionThreadActivityRepository.upsert:query",	286	yield* getOrRebuildPendingUserInputCount({ threadId: row.threadId });
193	"ProjectionThreadActivityRepository.upsert:encodeRequest",	287	const previous = yield* getExistingUserInputActivity({ activityId: row.activityId });
		288	if (Option.isSome(previous) && previous.value.threadId !== row.threadId) {
		289	yield* getOrRebuildPendingUserInputCount({ threadId: previous.value.threadId });
		290	}
		291	const next = userInputStateFromActivity(row);
		292	const keys = new Map<string, UserInputActivityKey>();
		293	if (Option.isSome(previous) && previous.value.requestId !== null) {
		294	const key = {
		295	threadId: previous.value.threadId,
		296	requestId: previous.value.requestId,
		297	};
		298	keys.set(userInputKeyId(key), key);
		299	}
		300	if (next !== null) {
		301	const key = { threadId: row.threadId, requestId: next.requestId };
		302	keys.set(userInputKeyId(key), key);
		303	}
		304
		305	const before = new Map<string, boolean>();
		306	for (const [keyId, key] of keys) {
		307	const state = yield* getLatestUserInputState(key);
		308	before.set(keyId, Option.isSome(state) && state.value.state === "pending");
		309	}
		310
		311	yield* upsertProjectionThreadActivityRow(row);
		312
		313	const deltas = new Map<string, number>();
		314	for (const [keyId, key] of keys) {
		315	const state = yield* getLatestUserInputState(key);
		316	const wasPending = before.get(keyId) ?? false;
		317	const isPending = Option.isSome(state) && state.value.state === "pending";
		318	const delta = Number(isPending) - Number(wasPending);
		319	if (delta !== 0) {
		320	deltas.set(key.threadId, (deltas.get(key.threadId) ?? 0) + delta);
		321	}
		322	}
		323	for (const [threadId, delta] of deltas) {
		324	yield* sql`
		325	INSERT INTO projection_thread_user_input_summaries (thread_id, pending_count)
		326	VALUES (${threadId}, ${Math.max(delta, 0)})
		327	ON CONFLICT (thread_id)
		328	DO UPDATE SET pending_count = pending_count + ${delta}
		329	`;
		330	}
		331
		332	const affectedThreadIds = new Set(Array.from(keys.values(), (key) => key.threadId));
		333	for (const threadId of affectedThreadIds) {
		334	const pendingUserInputCount = yield* getOrRebuildPendingUserInputCount({ threadId });
		335	yield* sql`
		336	UPDATE projection_threads
		337	SET pending_user_input_count = ${pendingUserInputCount}
		338	WHERE thread_id = ${threadId}
		339	`;
		340	}
		341	}),
		342	)
		343	.pipe(
		344	Effect.mapError(
		345	toPersistenceSqlOrDecodeError(
		346	"ProjectionThreadActivityRepository.upsert:query",
		347	"ProjectionThreadActivityRepository.upsert:encodeRequest",
		348	),
194	),	349	),
195	),	350	);
196	);
197		351
198	const listByThreadId: ProjectionThreadActivityRepositoryShape["listByThreadId"] = (input) =>	352	const listByThreadId: ProjectionThreadActivityRepositoryShape["listByThreadId"] = (input) =>
199	listProjectionThreadActivityRows(input).pipe(	353	listProjectionThreadActivityRows(input).pipe(
@@ -208,22 +362,35 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { · 8 unmodified lines
208		362
209	const countPendingUserInputsByThreadId: ProjectionThreadActivityRepositoryShape["countPendingUserInputsByThreadId"] =	363	const countPendingUserInputsByThreadId: ProjectionThreadActivityRepositoryShape["countPendingUserInputsByThreadId"] =
210	(input) =>	364	(input) =>
211	readPendingUserInputCount(input).pipe(	365	sql
212	Effect.mapError(	366	.withTransaction(getOrRebuildPendingUserInputCount(input))
213	toPersistenceSqlOrDecodeError(	367	.pipe(
214	"ProjectionThreadActivityRepository.countPendingUserInputsByThreadId:query",	368	Effect.mapError(
215	"ProjectionThreadActivityRepository.countPendingUserInputsByThreadId:decodeRow",	369	toPersistenceSqlOrDecodeError(
		370	"ProjectionThreadActivityRepository.countPendingUserInputsByThreadId:query",
		371	"ProjectionThreadActivityRepository.countPendingUserInputsByThreadId:decodeRow",
		372	),
216	),	373	),
217	),	374	);
218	Effect.map((row) => row.count),
219	);
220		375
221	const deleteByThreadId: ProjectionThreadActivityRepositoryShape["deleteByThreadId"] = (input) =>	376	const deleteByThreadId: ProjectionThreadActivityRepositoryShape["deleteByThreadId"] = (input) =>
222	deleteProjectionThreadActivityRows(input).pipe(	377	sql
223	Effect.mapError(	378	.withTransaction(
224	toPersistenceSqlError("ProjectionThreadActivityRepository.deleteByThreadId:query"),	379	Effect.gen(function* () {
225	),	380	yield* deleteProjectionThreadActivityRows(input);
226	);	381	yield* resetPendingUserInputSummary(input);
		382	yield* sql`
		383	UPDATE projection_threads
		384	SET pending_user_input_count = 0
		385	WHERE thread_id = ${input.threadId}
		386	`;
		387	}),
		388	)
		389	.pipe(
		390	Effect.mapError(
		391	toPersistenceSqlError("ProjectionThreadActivityRepository.deleteByThreadId:query"),
		392	),
		393	);
227		394
228	return {	395	return {
229	upsert,	396	upsert,

apps/server/src/persistence/Layers/ProjectionThreadMessages.ts+69−8

@@ -40,6 +40,10 @@ function toProjectionThreadMessage( · 39 unmodified lines
40	};	40	};
41	}	41	}
42		42
		43	const LatestUserMessageAtRowSchema = Schema.Struct({
		44	createdAt: ProjectionThreadMessage.fields.createdAt,
		45	});
		46
43	const makeProjectionThreadMessageRepository = Effect.gen(function* () {	47	const makeProjectionThreadMessageRepository = Effect.gen(function* () {
44	const sql = yield* SqlClient.SqlClient;	48	const sql = yield* SqlClient.SqlClient;
45		49
@@ -137,6 +141,20 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () { · 91 unmodified lines
137	`,	141	`,
138	});	142	});
139		143
		144	const getLatestUserMessageAtRow = SqlSchema.findOneOption({
		145	Request: ListProjectionThreadMessagesInput,
		146	Result: LatestUserMessageAtRowSchema,
		147	execute: ({ threadId }) =>
		148	sql`
		149	SELECT created_at AS "createdAt"
		150	FROM projection_thread_messages
		151	WHERE thread_id = ${threadId}
		152	AND role = 'user'
		153	ORDER BY created_at DESC, message_id DESC
		154	LIMIT 1
		155	`,
		156	});
		157
140	const deleteProjectionThreadMessageRows = SqlSchema.void({	158	const deleteProjectionThreadMessageRows = SqlSchema.void({
141	Request: DeleteProjectionThreadMessagesInput,	159	Request: DeleteProjectionThreadMessagesInput,
142	execute: ({ threadId }) =>	160	execute: ({ threadId }) =>
@@ -147,9 +165,29 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () { · 4 unmodified lines
147	});	165	});
148		166
149	const upsert: ProjectionThreadMessageRepositoryShape["upsert"] = (row) =>	167	const upsert: ProjectionThreadMessageRepositoryShape["upsert"] = (row) =>
150	upsertProjectionThreadMessageRow(row).pipe(	168	sql
151	Effect.mapError(toPersistenceSqlError("ProjectionThreadMessageRepository.upsert:query")),	169	.withTransaction(
152	);	170	Effect.gen(function* () {
		171	const previous = yield* getProjectionThreadMessageRow({ messageId: row.messageId });
		172	yield* upsertProjectionThreadMessageRow(row);
		173
		174	const affectedThreadIds = new Set([row.threadId]);
		175	if (Option.isSome(previous)) {
		176	affectedThreadIds.add(previous.value.threadId);
		177	}
		178	for (const threadId of affectedThreadIds) {
		179	const latest = yield* getLatestUserMessageAtRow({ threadId });
		180	yield* sql`
		181	UPDATE projection_threads
		182	SET latest_user_message_at = ${Option.isSome(latest) ? latest.value.createdAt : null}
		183	WHERE thread_id = ${threadId}
		184	`;
		185	}
		186	}),
		187	)
		188	.pipe(
		189	Effect.mapError(toPersistenceSqlError("ProjectionThreadMessageRepository.upsert:query")),
		190	);
153		191
154	const getByMessageId: ProjectionThreadMessageRepositoryShape["getByMessageId"] = (input) =>	192	const getByMessageId: ProjectionThreadMessageRepositoryShape["getByMessageId"] = (input) =>
155	getProjectionThreadMessageRow(input).pipe(	193	getProjectionThreadMessageRow(input).pipe(
@@ -167,17 +205,40 @@ const makeProjectionThreadMessageRepository = Effect.gen(function* () { · 11 unmodified lines
167	Effect.map((rows) => rows.map(toProjectionThreadMessage)),	205	Effect.map((rows) => rows.map(toProjectionThreadMessage)),
168	);	206	);
169		207
		208	const getLatestUserMessageAtByThreadId: ProjectionThreadMessageRepositoryShape["getLatestUserMessageAtByThreadId"] =
		209	(input) =>
		210	getLatestUserMessageAtRow(input).pipe(
		211	Effect.mapError(
		212	toPersistenceSqlError(
		213	"ProjectionThreadMessageRepository.getLatestUserMessageAtByThreadId:query",
		214	),
		215	),
		216	Effect.map(Option.map((row) => row.createdAt)),
		217	);
		218
170	const deleteByThreadId: ProjectionThreadMessageRepositoryShape["deleteByThreadId"] = (input) =>	219	const deleteByThreadId: ProjectionThreadMessageRepositoryShape["deleteByThreadId"] = (input) =>
171	deleteProjectionThreadMessageRows(input).pipe(	220	sql
172	Effect.mapError(	221	.withTransaction(
173	toPersistenceSqlError("ProjectionThreadMessageRepository.deleteByThreadId:query"),	222	Effect.gen(function* () {
174	),	223	yield* deleteProjectionThreadMessageRows(input);
175	);	224	yield* sql`
		225	UPDATE projection_threads
		226	SET latest_user_message_at = NULL
		227	WHERE thread_id = ${input.threadId}
		228	`;
		229	}),
		230	)
		231	.pipe(
		232	Effect.mapError(
		233	toPersistenceSqlError("ProjectionThreadMessageRepository.deleteByThreadId:query"),
		234	),
		235	);
176		236
177	return {	237	return {
178	upsert,	238	upsert,
179	getByMessageId,	239	getByMessageId,
180	listByThreadId,	240	listByThreadId,
		241	getLatestUserMessageAtByThreadId,
181	deleteByThreadId,	242	deleteByThreadId,
182	} satisfies ProjectionThreadMessageRepositoryShape;	243	} satisfies ProjectionThreadMessageRepositoryShape;
183	});	244	});

apps/server/src/persistence/Migrations.ts+2−0

 import Migration0039 from "./Migrations/039_ProjectionProjectsDefaultThreadEnvMode.ts";
 import Migration0040 from "./Migrations/040_ProjectionProjectFaviconPath.ts";
 import Migration0041 from "./Migrations/041_ProjectionThreadUserInputLifecycleIndex.ts";
+import Migration0042 from "./Migrations/042_ProjectionThreadShellSummaryState.ts";
 /**
  * Migration loader with all migrations defined inline.
   [39, "ProjectionProjectsDefaultThreadEnvMode", Migration0039],
   [40, "ProjectionProjectFaviconPath", Migration0040],
   [41, "ProjectionThreadUserInputLifecycleIndex", Migration0041],
+  [42, "ProjectionThreadShellSummaryState", Migration0042],
 ] as const;
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/042_ProjectionThreadShellSummaryState.test.ts+301−0

@@ -0,0 +1,301 @@
	1	import { assert, it } from "@effect/vitest";
	2	import * as Effect from "effect/Effect";
	3	import * as Layer from "effect/Layer";
	4	import * as SqlClient from "effect/unstable/sql/SqlClient";
	5
	6	import { runMigrations } from "../Migrations.ts";
	7	import * as NodeSqliteClient from "../NodeSqliteClient.ts";
	8
	9	const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory()));
	10
	11	layer("042_ProjectionThreadShellSummaryState", (it) => {
	12	it.effect("upgrades recorded migration 41 and reconciles shell summaries", () =>
	13	Effect.gen(function* () {
	14	const sql = yield* SqlClient.SqlClient;
	15
	16	yield* runMigrations({ toMigrationInclusive: 41 });
	17	yield* sql`
	18	INSERT INTO projection_threads (
	19	thread_id,
	20	project_id,
	21	title,
	22	model_selection_json,
	23	runtime_mode,
	24	interaction_mode,
	25	branch,
	26	worktree_path,
	27	latest_turn_id,
	28	created_at,
	29	updated_at,
	30	archived_at,
	31	latest_user_message_at,
	32	pending_approval_count,
	33	pending_user_input_count,
	34	has_actionable_proposed_plan,
	35	deleted_at
	36	) VALUES
	37	(
	38	'thread-cleared',
	39	'project-1',
	40	'Cleared',
	41	'{"provider":"codex","model":"gpt-5-codex"}',
	42	'approval-required',
	43	'default',
	44	NULL,
	45	NULL,
	46	NULL,
	47	'2026-08-19T00:00:00.000Z',
	48	'2026-08-19T00:00:00.000Z',
	49	NULL,
	50	'2026-08-19T00:00:05.000Z',
	51	0,
	52	1,
	53	0,
	54	NULL
	55	),
	56	(
	57	'thread-pending',
	58	'project-1',
	59	'Pending',
	60	'{"provider":"codex","model":"gpt-5-codex"}',
	61	'approval-required',
	62	'default',
	63	NULL,
	64	NULL,
	65	NULL,
	66	'2026-08-19T00:00:00.000Z',
	67	'2026-08-19T00:00:00.000Z',
	68	NULL,
	69	NULL,
	70	0,
	71	0,
	72	0,
	73	NULL
	74	)
	75	`;
	76
	77	yield* sql`
	78	INSERT INTO projection_thread_activities (
	79	activity_id,
	80	thread_id,
	81	turn_id,
	82	tone,
	83	kind,
	84	summary,
	85	payload_json,
	86	sequence,
	87	created_at
	88	) VALUES
	89	(
	90	'A',
	91	'thread-cleared',
	92	NULL,
	93	'info',
	94	'user-input.requested',
	95	'Tie requested',
	96	'{"requestId":"request-tie"}',
	97	NULL,
	98	'2026-08-19T00:00:02.000Z'
	99	),
	100	(
	101	'a',
	102	'thread-cleared',
	103	NULL,
	104	'info',
	105	'user-input.resolved',
	106	'Tie resolved',
	107	'{"requestId":"request-tie"}',
	108	NULL,
	109	'2026-08-19T00:00:02.000Z'
	110	),
	111	(
	112	'activity-pending',
	113	'thread-pending',
	114	NULL,
	115	'info',
	116	'user-input.requested',
	117	'Pending request',
	118	'{"requestId":"request-pending"}',
	119	NULL,
	120	'2026-08-19T00:00:01.000Z'
	121	),
	122	(
	123	'activity-object-detail',
	124	'thread-pending',
	125	NULL,
	126	'error',
	127	'provider.user-input.respond.failed',
	128	'Object detail',
	129	'{"requestId":"request-pending","detail":{"message":"stale pending user-input request"}}',
	130	NULL,
	131	'2026-08-19T00:00:02.000Z'
	132	),
	133	(
	134	'activity-malformed',
	135	'thread-pending',
	136	NULL,
	137	'info',
	138	'user-input.requested',
	139	'Malformed',
	140	'{',
	141	NULL,
	142	'2026-08-19T00:00:03.000Z'
	143	)
	144	`;
	145
	146	yield* sql`
	147	INSERT INTO projection_thread_messages (
	148	message_id,
	149	thread_id,
	150	turn_id,
	151	role,
	152	text,
	153	attachments_json,
	154	is_streaming,
	155	created_at,
	156	updated_at
	157	) VALUES
	158	(
	159	'message-assistant',
	160	'thread-cleared',
	161	NULL,
	162	'assistant',
	163	'No user message remains',
	164	NULL,
	165	0,
	166	'2026-08-19T00:00:05.000Z',
	167	'2026-08-19T00:00:05.000Z'
	168	),
	169	(
	170	'message-user',
	171	'thread-pending',
	172	NULL,
	173	'user',
	174	'Latest user message',
	175	NULL,
	176	0,
	177	'2026-08-19T00:00:06.000Z',
	178	'2026-08-19T00:00:06.000Z'
	179	)
	180	`;
	181
	182	yield* runMigrations({ toMigrationInclusive: 42 });
	183
	184	const normalizedRows = yield* sql<{
	185	readonly activityId: string;
	186	readonly requestId: string \| null;
	187	readonly state: string \| null;
	188	}>`
	189	SELECT
	190	activity_id AS "activityId",
	191	user_input_request_id AS "requestId",
	192	user_input_state AS "state"
	193	FROM projection_thread_activities
	194	WHERE activity_id IN ('activity-malformed', 'activity-object-detail')
	195	ORDER BY activity_id
	196	`;
	197	assert.deepStrictEqual(normalizedRows, [
	198	{ activityId: "activity-malformed", requestId: null, state: null },
	199	{ activityId: "activity-object-detail", requestId: null, state: null },
	200	]);
	201
	202	const summaryRows = yield* sql<{
	203	readonly threadId: string;
	204	readonly pendingCount: number;
	205	}>`
	206	SELECT
	207	thread_id AS "threadId",
	208	pending_count AS "pendingCount"
	209	FROM projection_thread_user_input_summaries
	210	ORDER BY thread_id
	211	`;
	212	assert.deepStrictEqual(summaryRows, [
	213	{ threadId: "thread-cleared", pendingCount: 0 },
	214	{ threadId: "thread-pending", pendingCount: 1 },
	215	]);
	216
	217	const threadRows = yield* sql<{
	218	readonly threadId: string;
	219	readonly latestUserMessageAt: string \| null;
	220	readonly pendingUserInputCount: number;
	221	}>`
	222	SELECT
	223	thread_id AS "threadId",
	224	latest_user_message_at AS "latestUserMessageAt",
	225	pending_user_input_count AS "pendingUserInputCount"
	226	FROM projection_threads
	227	ORDER BY thread_id
	228	`;
	229	assert.deepStrictEqual(threadRows, [
	230	{
	231	threadId: "thread-cleared",
	232	latestUserMessageAt: null,
	233	pendingUserInputCount: 0,
	234	},
	235	{
	236	threadId: "thread-pending",
	237	latestUserMessageAt: "2026-08-19T00:00:06.000Z",
	238	pendingUserInputCount: 1,
	239	},
	240	]);
	241
	242	const legacyIndexes = yield* sql<{ readonly name: string }>`
	243	SELECT name
	244	FROM sqlite_master
	245	WHERE type = 'index'
	246	AND name = 'idx_projection_thread_activities_thread_user_input_lifecycle'
	247	`;
	248	assert.deepStrictEqual(legacyIndexes, []);
	249
	250	const lifecycleIndexColumns = yield* sql<{ readonly name: string }>`
	251	PRAGMA index_info('idx_projection_thread_activities_user_input_state')
	252	`;
	253	assert.deepStrictEqual(
	254	lifecycleIndexColumns.map((column) => column.name),
	255	["thread_id", "user_input_request_id", "created_at", "activity_id"],
	256	);
	257
	258	const lifecyclePlan = yield* sql<{ readonly detail: string }>`
	259	EXPLAIN QUERY PLAN
	260	SELECT user_input_state
	261	FROM projection_thread_activities
	262	WHERE thread_id = 'thread-pending'
	263	AND user_input_request_id = 'request-pending'
	264	AND user_input_state IS NOT NULL
	265	ORDER BY created_at DESC, activity_id DESC
	266	LIMIT 1
	267	`;
	268	const lifecyclePlanText = lifecyclePlan.map((row) => row.detail).join("\n");
	269	assert.match(
	270	lifecyclePlanText,
	271	/USING INDEX idx_projection_thread_activities_user_input_state/,
	272	);
	273	assert.notMatch(lifecyclePlanText, /USE TEMP B-TREE/);
	274
	275	const summaryPlan = yield* sql<{ readonly detail: string }>`
	276	EXPLAIN QUERY PLAN
	277	SELECT pending_count
	278	FROM projection_thread_user_input_summaries
	279	WHERE thread_id = 'thread-pending'
	280	LIMIT 1
	281	`;
	282	assert.match(summaryPlan.map((row) => row.detail).join("\n"), /USING INDEX/);
	283
	284	const messagePlan = yield* sql<{ readonly detail: string }>`
	285	EXPLAIN QUERY PLAN
	286	SELECT created_at
	287	FROM projection_thread_messages
	288	WHERE thread_id = 'thread-pending'
	289	AND role = 'user'
	290	ORDER BY created_at DESC, message_id DESC
	291	LIMIT 1
	292	`;
	293	const messagePlanText = messagePlan.map((row) => row.detail).join("\n");
	294	assert.match(
	295	messagePlanText,
	296	/USING COVERING INDEX idx_projection_thread_messages_latest_user/,
	297	);
	298	assert.notMatch(messagePlanText, /USE TEMP B-TREE/);
	299	}),
	300	);
	301	});

apps/server/src/persistence/Migrations/042_ProjectionThreadShellSummaryState.ts+148−0

@@ -0,0 +1,148 @@
	1	import * as SqlClient from "effect/unstable/sql/SqlClient";
	2	import * as Effect from "effect/Effect";
	3
	4	export default Effect.gen(function* () {
	5	const sql = yield* SqlClient.SqlClient;
	6
	7	yield* sql`
	8	ALTER TABLE projection_thread_activities
	9	ADD COLUMN user_input_request_id TEXT
	10	`;
	11
	12	yield* sql`
	13	ALTER TABLE projection_thread_activities
	14	ADD COLUMN user_input_state TEXT
	15	CHECK (user_input_state IN ('pending', 'cleared'))
	16	`;
	17
	18	yield* sql`
	19	UPDATE projection_thread_activities
	20	SET
	21	user_input_request_id = CASE
	22	WHEN json_valid(payload_json) THEN CASE
	23	WHEN json_type(payload_json, '$.requestId') = 'text'
	24	THEN json_extract(payload_json, '$.requestId')
	25	ELSE NULL
	26	END
	27	ELSE NULL
	28	END,
	29	user_input_state = CASE
	30	WHEN kind = 'user-input.requested' THEN 'pending'
	31	WHEN kind = 'user-input.resolved' THEN 'cleared'
	32	WHEN kind = 'provider.user-input.respond.failed'
	33	AND json_valid(payload_json)
	34	AND json_type(payload_json, '$.detail') = 'text'
	35	THEN CASE
	36	WHEN lower(json_extract(payload_json, '$.detail'))
	37	LIKE '%stale pending user-input request%'
	38	OR lower(json_extract(payload_json, '$.detail'))
	39	LIKE '%unknown pending user-input request%'
	40	OR lower(json_extract(payload_json, '$.detail'))
	41	LIKE '%unknown pending user input request%'
	42	OR lower(json_extract(payload_json, '$.detail'))
	43	LIKE '%unknown pending codex user input request%'
	44	THEN 'cleared'
	45	ELSE NULL
	46	END
	47	ELSE NULL
	48	END
	49	WHERE kind IN (
	50	'user-input.requested',
	51	'user-input.resolved',
	52	'provider.user-input.respond.failed'
	53	)
	54	`;
	55
	56	yield* sql`
	57	UPDATE projection_thread_activities
	58	SET user_input_state = NULL
	59	WHERE user_input_request_id IS NULL
	60	`;
	61
	62	yield* sql`
	63	UPDATE projection_thread_activities
	64	SET user_input_request_id = NULL
	65	WHERE user_input_state IS NULL
	66	`;
	67
	68	yield* sql`
	69	CREATE INDEX IF NOT EXISTS idx_projection_thread_activities_user_input_state
	70	ON projection_thread_activities(
	71	thread_id,
	72	user_input_request_id,
	73	created_at DESC,
	74	activity_id DESC
	75	)
	76	WHERE user_input_state IS NOT NULL
	77	`;
	78
	79	yield* sql`
	80	DROP INDEX IF EXISTS idx_projection_thread_activities_thread_user_input_lifecycle
	81	`;
	82
	83	yield* sql`
	84	CREATE TABLE IF NOT EXISTS projection_thread_user_input_summaries (
	85	thread_id TEXT PRIMARY KEY,
	86	pending_count INTEGER NOT NULL CHECK (pending_count >= 0)
	87	)
	88	`;
	89
	90	yield* sql`
	91	WITH ranked AS (
	92	SELECT
	93	thread_id,
	94	user_input_request_id,
	95	user_input_state,
	96	ROW_NUMBER() OVER (
	97	PARTITION BY thread_id, user_input_request_id
	98	ORDER BY created_at DESC, activity_id DESC
	99	) AS row_number
	100	FROM projection_thread_activities
	101	WHERE user_input_state IS NOT NULL
	102	AND user_input_request_id IS NOT NULL
	103	)
	104	INSERT INTO projection_thread_user_input_summaries (
	105	thread_id,
	106	pending_count
	107	)
	108	SELECT
	109	thread_id,
	110	SUM(CASE WHEN user_input_state = 'pending' THEN 1 ELSE 0 END)
	111	FROM ranked
	112	WHERE row_number = 1
	113	GROUP BY thread_id
	114	`;
	115
	116	yield* sql`
	117	INSERT OR IGNORE INTO projection_thread_user_input_summaries (
	118	thread_id,
	119	pending_count
	120	)
	121	SELECT thread_id, 0
	122	FROM projection_threads
	123	`;
	124
	125	yield* sql`
	126	CREATE INDEX IF NOT EXISTS idx_projection_thread_messages_latest_user
	127	ON projection_thread_messages(thread_id, created_at DESC, message_id DESC)
	128	WHERE role = 'user'
	129	`;
	130
	131	yield* sql`
	132	UPDATE projection_threads
	133	SET
	134	pending_user_input_count = COALESCE((
	135	SELECT summary.pending_count
	136	FROM projection_thread_user_input_summaries AS summary
	137	WHERE summary.thread_id = projection_threads.thread_id
	138	), 0),
	139	latest_user_message_at = (
	140	SELECT message.created_at
	141	FROM projection_thread_messages AS message
	142	WHERE message.thread_id = projection_threads.thread_id
	143	AND message.role = 'user'
	144	ORDER BY message.created_at DESC, message.message_id DESC
	145	LIMIT 1
	146	)
	147	`;
	148	});

apps/server/src/persistence/Services/ProjectionThreadMessages.ts+10−0

@@ -78,6 +78,16 @@ export interface ProjectionThreadMessageRepositoryShape { · 77 unmodified lines
78	input: ListProjectionThreadMessagesInput,	78	input: ListProjectionThreadMessagesInput,
79	) => Effect.Effect<ReadonlyArray<ProjectionThreadMessage>, ProjectionRepositoryError>;	79	) => Effect.Effect<ReadonlyArray<ProjectionThreadMessage>, ProjectionRepositoryError>;
80		80
		81	/**
		82	* Read the latest projected user-message timestamp for a thread.
		83	*/
		84	readonly getLatestUserMessageAtByThreadId: (
		85	input: ListProjectionThreadMessagesInput,
		86	) => Effect.Effect<
		87	Option.Option<ProjectionThreadMessage["createdAt"]>,
		88	ProjectionRepositoryError
		89	>;
		90
81	/**	91	/**
82	* Delete projected thread messages by thread.	92	* Delete projected thread messages by thread.
83	*/	93	*/