Add shell snapshot queries for orchestration state

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 

apps/server/src/checkpointing/Layers/CheckpointDiffQuery.test.ts

@@ -82,10 +82,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );
@@ -136,10 +141,15 @@ describe("CheckpointDiffQueryLive", () => {
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () =>
  Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"),
+ getShellSnapshot: () =>
+ Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
  getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  );

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -144,10 +144,20 @@ describe("OrchestrationEngine", () => {
  Layer.provide(
  Layer.succeed(ProjectionSnapshotQuery, {
  getSnapshot: () => Effect.succeed(projectionSnapshot),
+ getShellSnapshot: () =>
+ Effect.succeed({
+ snapshotSequence: projectionSnapshot.snapshotSequence,
+ projects: [],
+ threads: [],
+ updatedAt: projectionSnapshot.updatedAt,
+ }),
  getCounts: () => Effect.succeed({ projectCount: 1, threadCount: 1 }),
  getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
+ getProjectShellById: () => Effect.succeed(Option.none()),
  getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),
  getThreadCheckpointContext: () => Effect.succeed(Option.none()),
+ getThreadShellById: () => Effect.succeed(Option.none()),
+ getThreadDetailById: () => Effect.succeed(Option.none()),
  }),
  ),
  Layer.provide(

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -2,6 +2,7 @@ import {
  ApprovalRequestId,
  type ChatAttachment,
  type OrchestrationEvent,
+ ThreadId,
 } from "@t3tools/contracts";
 import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";
 import * as SqlClient from "effect/unstable/sql/SqlClient";
@@ -89,6 +90,77 @@ function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {
  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;
 }
 
+function derivePendingUserInputCountFromActivities(
+ activities: ReadonlyArray<ProjectionThreadActivity>,
+): number {
+ const openRequestIds = new Set<string>();
+ const ordered = [...activities].toSorted(
+ (left, right) =>
+ left.createdAt.localeCompare(right.createdAt) ||
+ left.activityId.localeCompare(right.activityId),
+ );
+
+ for (const activity of ordered) {
+ const requestId = extractActivityRequestId(activity.payload);
+ if (requestId === null) {
+ continue;
+ }
+ const payload =
+ typeof activity.payload === "object" && activity.payload !== null
+ ? (activity.payload as Record<string, unknown>)
+ : null;
+ const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;
+
+ if (activity.kind === "user-input.requested") {
+ openRequestIds.add(requestId);
+ continue;
+ }
+
+ if (activity.kind === "user-input.resolved") {
+ openRequestIds.delete(requestId);
+ continue;
+ }
+
+ if (
+ activity.kind === "provider.user-input.respond.failed" &&
+ detail !== null &&
+ (detail.includes("stale pending user-input request") ||
+ detail.includes("unknown pending user-input request"))
+ ) {
+ openRequestIds.delete(requestId);
+ }
+ }
+
+ return openRequestIds.size;
+}
+
+function deriveHasActionableProposedPlan(input: {
+ readonly latestTurnId: string | null;
+ readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
+}): boolean {
+ const sorted = [...input.proposedPlans].toSorted(
+ (left, right) =>
+ left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),
+ );
+
+ let latestForTurn: ProjectionThreadProposedPlan | null = null;
+ if (input.latestTurnId !== null) {
+ for (let index = sorted.length - 1; index >= 0; index -= 1) {
+ const plan = sorted[index];
+ if (plan?.turnId === input.latestTurnId) {
+ latestForTurn = plan;
+ break;
+ }
+ }
+ }
+ if (latestForTurn !== null) {
+ return latestForTurn.implementedAt === null;
+ }
+
+ const latestPlan = sorted.at(-1) ?? null;
+ return latestPlan !== null && latestPlan.implementedAt === null;
+}
+
 function retainProjectionMessagesAfterRevert(
  messages: ReadonlyArray<ProjectionThreadMessage>,
  turns: ReadonlyArray<ProjectionTurn>,
@@ -432,6 +504,48 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  }
  });
 
+ const refreshThreadShellSummary = Effect.fn("refreshThreadShellSummary")(function* (
+ threadId: ThreadId,
+ ) {
+ const existingRow = yield* projectionThreadRepository.getById({
+ threadId,
+ });
+ if (Option.isNone(existingRow)) {
+ return;
+ }
+
+ const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
+ projectionThreadMessageRepository.listByThreadId({ threadId }),
+ projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
+ projectionThreadActivityRepository.listByThreadId({ threadId }),
+ projectionPendingApprovalRepository.listByThreadId({ threadId }),
+ ]);
+
+ const latestUserMessageAt =
+ messages
+ .filter((message) => message.role === "user")
+ .map((message) => message.createdAt)
+ .toSorted()
+ .at(-1) ?? null;
+
+ const pendingApprovalCount = pendingApprovals.filter(
+ (approval) => approval.status === "pending",
+ ).length;
+ const pendingUserInputCount = derivePendingUserInputCountFromActivities(activities);
+ const hasActionableProposedPlan = deriveHasActionableProposedPlan({
+ latestTurnId: existingRow.value.latestTurnId,
+ proposedPlans,
+ });
+
+ yield* projectionThreadRepository.upsert({
+ ...existingRow.value,
+ latestUserMessageAt,
+ pendingApprovalCount,
+ pendingUserInputCount,
+ hasActionableProposedPlan: hasActionableProposedPlan ? 1 : 0,
+ });
+ });
+
  const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(
  "applyThreadsProjection",
  )(function* (event, attachmentSideEffects) {
@@ -450,6 +564,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  createdAt: event.payload.createdAt,
  updatedAt: event.payload.updatedAt,
  archivedAt: null,
+ latestUserMessageAt: null,
+ pendingApprovalCount: 0,
+ pendingUserInputCount: 0,
+ hasActionableProposedPlan: 0,
  deletedAt: null,
  });
  return;
@@ -554,7 +672,9 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
 
  case "thread.message-sent":
  case "thread.proposed-plan-upserted":
- case "thread.activity-appended": {
+ case "thread.activity-appended":
+ case "thread.approval-response-requested":
+ case "thread.user-input-response-requested": {
  const existingRow = yield* projectionThreadRepository.getById({
  threadId: event.payload.threadId,
  });
@@ -565,6 +685,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  ...existingRow.value,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -580,6 +701,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.session.activeTurnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -595,6 +717,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: event.payload.turnId,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }
 
@@ -610,6 +733,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
  latestTurnId: null,
  updatedAt: event.occurredAt,
  });
+ yield* refreshThreadShellSummary(event.payload.threadId);
  return;
  }