diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index e3b18d74a9a7..7949abbb0db2 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -2086,6 +2086,125 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); + it.effect("counts an unresolved user-input request on the thread shell", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.make("evt-open-user-input-1"), + aggregateKind: "project", + aggregateId: ProjectId.make("project-open-user-input"), + occurredAt: "2026-02-26T13:00:00.000Z", + commandId: CommandId.make("cmd-open-user-input-1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-open-user-input-1"), + metadata: {}, + payload: { + projectId: ProjectId.make("project-open-user-input"), + title: "Project Open User Input", + workspaceRoot: "/tmp/project-open-user-input", + defaultModelSelection: null, + scripts: [], + createdAt: "2026-02-26T13:00:00.000Z", + updatedAt: "2026-02-26T13:00:00.000Z", + }, + }); + + yield* appendAndProject({ + type: "thread.created", + eventId: EventId.make("evt-open-user-input-2"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-open-user-input"), + occurredAt: "2026-02-26T13:00:01.000Z", + commandId: CommandId.make("cmd-open-user-input-2"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-open-user-input-2"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-open-user-input"), + projectId: ProjectId.make("project-open-user-input"), + title: "Thread Open User Input", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "approval-required", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-02-26T13:00:01.000Z", + updatedAt: "2026-02-26T13:00:01.000Z", + }, + }); + + // Noise the summary must read past: the request it counts is a handful of + // rows among a thread's whole tool timeline. + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-open-user-input-3"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-open-user-input"), + occurredAt: "2026-02-26T13:00:02.000Z", + commandId: CommandId.make("cmd-open-user-input-3"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-open-user-input-3"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-open-user-input"), + activity: { + id: EventId.make("activity-open-user-input-tool"), + tone: "tool", + kind: "tool.updated", + summary: "Tool progress", + payload: { requestId: "not-a-user-input-request" }, + turnId: null, + createdAt: "2026-02-26T13:00:02.000Z", + }, + }, + }); + + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-open-user-input-4"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-open-user-input"), + occurredAt: "2026-02-26T13:00:03.000Z", + commandId: CommandId.make("cmd-open-user-input-4"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-open-user-input-4"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-open-user-input"), + activity: { + id: EventId.make("activity-open-user-input-requested"), + tone: "info", + kind: "user-input.requested", + summary: "User input requested", + payload: { requestId: "user-input-request-open-1" }, + turnId: null, + createdAt: "2026-02-26T13:00:03.000Z", + }, + }, + }); + + const threadRows = yield* sql<{ + readonly pendingUserInputCount: number; + }>` + SELECT pending_user_input_count AS "pendingUserInputCount" + FROM projection_threads + WHERE thread_id = 'thread-open-user-input' + `; + assert.deepEqual(threadRows, [{ pendingUserInputCount: 1 }]); + }), + ); + it.effect("ignores non-stale provider approval response failures", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index e9a625dd91cf..eb7bc2eda0dc 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -130,6 +130,17 @@ function isStalePendingApprovalFailureDetail(detail: string | null): boolean { ); } +/** + * The only activity kinds `derivePendingUserInputCountFromActivities` reacts to. + * Every other kind is skipped, so the query feeding it filters on these — keep + * the two in step. + */ +const PENDING_USER_INPUT_ACTIVITY_KINDS = [ + "user-input.requested", + "user-input.resolved", + "provider.user-input.respond.failed", +] as const; + function derivePendingUserInputCountFromActivities( activities: ReadonlyArray, ): number { @@ -562,10 +573,18 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + // Only the user-input activity kinds, not the whole timeline: the single + // consumer below reduces them to one integer, while a thread's activity + // rows carry every tool payload it has produced. This runs on every event + // in the thread, so reading them in full makes each event cost the + // thread's entire history. const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([ projectionThreadMessageRepository.listByThreadId({ threadId }), projectionThreadProposedPlanRepository.listByThreadId({ threadId }), - projectionThreadActivityRepository.listByThreadId({ threadId }), + projectionThreadActivityRepository.listByThreadIdAndKinds({ + threadId, + kinds: PENDING_USER_INPUT_ACTIVITY_KINDS, + }), projectionPendingApprovalRepository.listByThreadId({ threadId }), ]); diff --git a/apps/server/src/persistence/Layers/ProjectionThreadActivities.test.ts b/apps/server/src/persistence/Layers/ProjectionThreadActivities.test.ts new file mode 100644 index 000000000000..4b9d001aa20d --- /dev/null +++ b/apps/server/src/persistence/Layers/ProjectionThreadActivities.test.ts @@ -0,0 +1,126 @@ +import { EventId, ThreadId } from "@t3tools/contracts"; +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import { SqlitePersistenceMemory } from "./Sqlite.ts"; +import { ProjectionThreadActivityRepositoryLive } from "./ProjectionThreadActivities.ts"; +import { ProjectionThreadActivityRepository } from "../Services/ProjectionThreadActivities.ts"; +import type { ProjectionThreadActivity } from "../Services/ProjectionThreadActivities.ts"; + +const threadId = ThreadId.make("thread-activity-kinds"); +const otherThreadId = ThreadId.make("thread-activity-kinds-other"); + +const activity = ( + id: string, + kind: string, + createdAt: string, + overrides: Partial = {}, +): ProjectionThreadActivity => ({ + activityId: EventId.make(id), + threadId, + turnId: null, + tone: "info", + kind, + summary: `${kind} ${id}`, + payload: { requestId: `request-${id}` }, + createdAt, + ...overrides, +}); + +const layer = it.layer( + Layer.mergeAll( + ProjectionThreadActivityRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)), + SqlitePersistenceMemory, + ), +); + +layer("ProjectionThreadActivityRepository.listByThreadIdAndKinds", (it) => { + const seed = Effect.fn("seed")(function* () { + const activities = yield* ProjectionThreadActivityRepository; + yield* activities.upsert(activity("a1", "user-input.requested", "2026-03-24T00:00:01.000Z")); + yield* activities.upsert(activity("a2", "tool.updated", "2026-03-24T00:00:02.000Z")); + yield* activities.upsert(activity("a3", "user-input.resolved", "2026-03-24T00:00:03.000Z")); + yield* activities.upsert(activity("a4", "tool.completed", "2026-03-24T00:00:04.000Z")); + yield* activities.upsert( + activity("a5", "user-input.requested", "2026-03-24T00:00:05.000Z", { + threadId: otherThreadId, + }), + ); + return activities; + }); + + it.effect("returns only the requested kinds", () => + Effect.gen(function* () { + const activities = yield* seed(); + + const rows = yield* activities.listByThreadIdAndKinds({ + threadId, + kinds: ["user-input.requested", "user-input.resolved"], + }); + + assert.deepStrictEqual( + rows.map((row) => row.activityId), + ["a1", "a3"], + ); + }), + ); + + it.effect("keeps the same ordering as the unfiltered list", () => + Effect.gen(function* () { + const activities = yield* seed(); + + const all = yield* activities.listByThreadId({ threadId }); + const filtered = yield* activities.listByThreadIdAndKinds({ + threadId, + kinds: ["user-input.requested", "user-input.resolved", "tool.updated", "tool.completed"], + }); + + assert.deepStrictEqual( + filtered.map((row) => row.activityId), + all.map((row) => row.activityId), + ); + }), + ); + + it.effect("decodes the payload the same way the unfiltered list does", () => + Effect.gen(function* () { + const activities = yield* seed(); + + const [filtered] = yield* activities.listByThreadIdAndKinds({ + threadId, + kinds: ["user-input.requested"], + }); + const all = yield* activities.listByThreadId({ threadId }); + const unfiltered = all.find((row) => row.activityId === "a1"); + + assert.deepStrictEqual(filtered, unfiltered); + }), + ); + + it.effect("does not reach across threads", () => + Effect.gen(function* () { + const activities = yield* seed(); + + const rows = yield* activities.listByThreadIdAndKinds({ + threadId: otherThreadId, + kinds: ["user-input.requested"], + }); + + assert.deepStrictEqual( + rows.map((row) => row.activityId), + ["a5"], + ); + }), + ); + + it.effect("returns nothing for an empty kind list without querying", () => + Effect.gen(function* () { + const activities = yield* seed(); + + const rows = yield* activities.listByThreadIdAndKinds({ threadId, kinds: [] }); + + assert.deepStrictEqual(rows, []); + }), + ); +}); diff --git a/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts b/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts index 2f4815f96545..7d425d10d68c 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreadActivities.ts @@ -10,6 +10,7 @@ import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts"; import { DeleteProjectionThreadActivitiesInput, + ListProjectionThreadActivitiesByKindInput, ListProjectionThreadActivitiesInput, ProjectionThreadActivity, ProjectionThreadActivityRepository, @@ -30,6 +31,20 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st : toPersistenceSqlError(sqlOperation)(cause); } +const toProjectionThreadActivity = ( + row: typeof ProjectionThreadActivityDbRowSchema.Type, +): ProjectionThreadActivity => ({ + activityId: row.activityId, + threadId: row.threadId, + turnId: row.turnId, + tone: row.tone, + kind: row.kind, + summary: row.summary, + payload: row.payload, + ...(row.sequence !== null ? { sequence: row.sequence } : {}), + createdAt: row.createdAt, +}); + const makeProjectionThreadActivityRepository = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -97,6 +112,32 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { `, }); + const listProjectionThreadActivityRowsByKind = SqlSchema.findAll({ + Request: ListProjectionThreadActivitiesByKindInput, + Result: ProjectionThreadActivityDbRowSchema, + execute: ({ threadId, kinds }) => + sql` + SELECT + activity_id AS "activityId", + thread_id AS "threadId", + turn_id AS "turnId", + tone, + kind, + summary, + payload_json AS "payload", + sequence, + created_at AS "createdAt" + FROM projection_thread_activities + WHERE thread_id = ${threadId} + AND kind IN ${sql.in(kinds)} + ORDER BY + CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC, + sequence ASC, + created_at ASC, + activity_id ASC + `, + }); + const deleteProjectionThreadActivityRows = SqlSchema.void({ Request: DeleteProjectionThreadActivitiesInput, execute: ({ threadId }) => @@ -124,21 +165,23 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { "ProjectionThreadActivityRepository.listByThreadId:decodeRows", ), ), - Effect.map((rows) => - rows.map((row) => ({ - activityId: row.activityId, - threadId: row.threadId, - turnId: row.turnId, - tone: row.tone, - kind: row.kind, - summary: row.summary, - payload: row.payload, - ...(row.sequence !== null ? { sequence: row.sequence } : {}), - createdAt: row.createdAt, - })), - ), + Effect.map((rows) => rows.map(toProjectionThreadActivity)), ); + const listByThreadIdAndKinds: ProjectionThreadActivityRepositoryShape["listByThreadIdAndKinds"] = + (input) => + input.kinds.length === 0 + ? Effect.succeed([]) + : listProjectionThreadActivityRowsByKind(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionThreadActivityRepository.listByThreadIdAndKinds:query", + "ProjectionThreadActivityRepository.listByThreadIdAndKinds:decodeRows", + ), + ), + Effect.map((rows) => rows.map(toProjectionThreadActivity)), + ); + const deleteByThreadId: ProjectionThreadActivityRepositoryShape["deleteByThreadId"] = (input) => deleteProjectionThreadActivityRows(input).pipe( Effect.mapError( @@ -149,6 +192,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () { return { upsert, listByThreadId, + listByThreadIdAndKinds, deleteByThreadId, } satisfies ProjectionThreadActivityRepositoryShape; }); diff --git a/apps/server/src/persistence/Services/ProjectionThreadActivities.ts b/apps/server/src/persistence/Services/ProjectionThreadActivities.ts index 47cb6073c479..7a49d630da00 100644 --- a/apps/server/src/persistence/Services/ProjectionThreadActivities.ts +++ b/apps/server/src/persistence/Services/ProjectionThreadActivities.ts @@ -38,6 +38,13 @@ export const ListProjectionThreadActivitiesInput = Schema.Struct({ }); export type ListProjectionThreadActivitiesInput = typeof ListProjectionThreadActivitiesInput.Type; +export const ListProjectionThreadActivitiesByKindInput = Schema.Struct({ + threadId: ThreadId, + kinds: Schema.Array(Schema.String), +}); +export type ListProjectionThreadActivitiesByKindInput = + typeof ListProjectionThreadActivitiesByKindInput.Type; + export const DeleteProjectionThreadActivitiesInput = Schema.Struct({ threadId: ThreadId, }); @@ -67,6 +74,18 @@ export interface ProjectionThreadActivityRepositoryShape { input: ListProjectionThreadActivitiesInput, ) => Effect.Effect, ProjectionRepositoryError>; + /** + * List projected thread activity rows for a thread, restricted to `kinds`. + * + * Same ordering as {@link listByThreadId}. Callers deriving a fact from a few + * activity kinds should use this: a thread's activity rows carry every tool + * payload it has produced, so reading them all to answer a narrow question + * scales with the thread's whole history. + */ + readonly listByThreadIdAndKinds: ( + input: ListProjectionThreadActivitiesByKindInput, + ) => Effect.Effect, ProjectionRepositoryError>; + /** * Delete projected thread activity rows by thread. */