import {
ApprovalRequestId,
ChatAttachment,
CheckpointRef,
IsoDateTime,
MessageId,
NonNegativeInt,
OrchestrationCheckpointFile,
OrchestrationProposedPlanId,
OrchestrationReadModel,
OrchestrationThreadSearchSource,
OrchestrationShellSnapshot,
OrchestrationThread,
OrchestrationThreadDetailSnapshot,
ProjectScript,
ProjectIconOverride,
TurnId,
type OrchestrationCheckpointSummary,
type OrchestrationLatestTurn,
type OrchestrationMessage,
type OrchestrationProjectShell,
type OrchestrationProposedPlan,
type OrchestrationProject,
type OrchestrationSession,
type OrchestrationThreadActivity,
type OrchestrationThreadShell,
ModelSelection,
ProjectId,
ThreadLinkedPullRequest,
ThreadId,
} from "@t3tools/contracts";
import * as Arr from "effect/Array";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Result from "effect/Result";
import * as Schema from "effect/Schema";
import * as Struct from "effect/Struct";
import * as SqlClient from "effect/unstable/sql/SqlClient";
import * as SqlSchema from "effect/unstable/sql/SqlSchema";
import {
isPersistenceError,
toPersistenceDecodeError,
toPersistenceSqlError,
type ProjectionRepositoryError,
} from "../../persistence/Errors.ts";
import { ProjectionCheckpoint } from "../../persistence/Services/ProjectionCheckpoints.ts";
import { ThreadBackgroundLivenessService } from "../ThreadBackgroundLiveness.ts";
import { ThreadPlanProgressService } from "../ThreadPlanProgress.ts";
import { ProjectionProject } from "../../persistence/Services/ProjectionProjects.ts";
import { ProjectionState } from "../../persistence/Services/ProjectionState.ts";
import { ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";
import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionThreadMessages.ts";
import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";
import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";
import {
decodeThreadDetailPageCursor,
encodeThreadDetailPageCursor,
} from "../threadDetailCursor.ts";
import { projectActivityPayload } from "../ActivityPayloadProjection.ts";
import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";
import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
import {
ProjectionSnapshotQuery,
type ProjectionEventReplayStats,
type ProjectionFullThreadDiffContext,
type ProjectionSnapshotCounts,
type ProjectionThreadCheckpointContext,
type ProjectionThreadDetailQuery,
type ProjectionSnapshotQueryShape,
} from "../Services/ProjectionSnapshotQuery.ts";
const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel);
const decodeShellSnapshot = Schema.decodeUnknownEffect(OrchestrationShellSnapshot);
const decodeThread = Schema.decodeUnknownEffect(OrchestrationThread);
// Keep detail reads consistent with the in-memory projector's retained
// activity window. Applying the limit in SQL avoids decoding an unbounded
// payload_json set before the projector can enforce that invariant.
const THREAD_DETAIL_ACTIVITY_LIMIT = 500;
// Snapshot payloads are decoded and projected in small sequential batches so
// one client read does not retain the raw payloads for the full activity window.
const THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE = 25;
const ProjectionProjectDbRowSchema = ProjectionProject.mapFields(
Struct.assign({
defaultModelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)),
autoPull: Schema.Number,
projectIcon: Schema.NullOr(Schema.fromJsonString(ProjectIconOverride)),
scripts: Schema.fromJsonString(Schema.Array(ProjectScript)),
}),
);
const ProjectionThreadMessageDbRowSchema = ProjectionThreadMessage.mapFields(
Struct.assign({
isStreaming: Schema.Number,
attachments: Schema.NullOr(Schema.fromJsonString(Schema.Array(ChatAttachment))),
}),
);
const ProjectionThreadProposedPlanDbRowSchema = ProjectionThreadProposedPlan;
const ProjectionThreadDbRowSchema = ProjectionThread.mapFields(
Struct.assign({
modelSelection: Schema.fromJsonString(ModelSelection),
linkedPullRequest: Schema.NullOr(Schema.fromJsonString(ThreadLinkedPullRequest)),
}),
);
const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields(
Struct.assign({
payload: Schema.fromJsonString(Schema.Unknown),
sequence: Schema.NullOr(NonNegativeInt),
}),
);
const ProjectionThreadActivityIdRowSchema = Schema.Struct({
activityId: ProjectionThreadActivity.fields.activityId,
});
const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;
const ProjectionThreadRuntimeContextDbRowSchema = Schema.Struct({
id: ThreadId,
title: Schema.String,
session: Schema.NullOr(ProjectionThreadSessionDbRowSchema),
});
const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields(
Struct.assign({
files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)),
}),
);
const ProjectionLatestTurnDbRowSchema = Schema.Struct({
threadId: ProjectionThread.fields.threadId,
turnId: TurnId,
state: Schema.String,
requestedAt: IsoDateTime,
startedAt: Schema.NullOr(IsoDateTime),
completedAt: Schema.NullOr(IsoDateTime),
assistantMessageId: Schema.NullOr(MessageId),
sourceProposedPlanThreadId: Schema.NullOr(ThreadId),
sourceProposedPlanId: Schema.NullOr(OrchestrationProposedPlanId),
});
const ProjectionStateDbRowSchema = ProjectionState;
const ProjectionCountsRowSchema = Schema.Struct({
projectCount: Schema.Number,
threadCount: Schema.Number,
});
const EventReplayStatsInput = Schema.Struct({
fromSequenceExclusive: NonNegativeInt,
toSequenceInclusive: NonNegativeInt,
});
const EventReplayStatsRowSchema = Schema.Struct({
eventCount: Schema.Number,
payloadBytes: Schema.Number,
});
const ProjectionThreadSearchRequest = Schema.Struct({
pattern: Schema.String,
limit: Schema.Int,
});
const ProjectionThreadSearchRow = Schema.Struct({
threadId: ThreadId,
projectId: ProjectId,
source: OrchestrationThreadSearchSource,
matchText: Schema.String,
messageCreatedAt: Schema.NullOr(IsoDateTime),
});
const WorkspaceRootLookupInput = Schema.Struct({
workspaceRoot: Schema.String,
});
const ProjectIdLookupInput = Schema.Struct({
projectId: ProjectId,
});
const ThreadIdLookupInput = Schema.Struct({
threadId: ThreadId,
});
const ThreadActivityKindsLookupInput = Schema.Struct({
threadId: ThreadId,
activityKinds: Schema.Array(Schema.String),
});
const ThreadActivityIdsLookupInput = Schema.Struct({
activityIds: Schema.Array(ProjectionThreadActivity.fields.activityId),
});
// Windowed reads order turns by the stable keyset (anchor, turn key), where
// anchor is requested_at and turn key is
// COALESCE(turn_id, ''). Both are event-derived, so cursors survive the
// revert projector's row-id rewrite and full projection rebuilds.
const ThreadTurnWindowLookupInput = Schema.Struct({
threadId: ThreadId,
// Exclusive keyset upper bound. Sentinels "~"/"" mean unbounded ("~" sorts
// after every ISO timestamp).
beforeAnchorAt: Schema.String,
beforeTurnKey: Schema.String,
userTurnLimit: Schema.Number,
maxRawTurns: Schema.Number,
});
const ProjectionTurnWindowRowSchema = Schema.Struct({
// The turn's timeline anchor, used to bound rows that have no turn linkage
// (user messages and turnless activities) to the same page window.
anchorAt: Schema.String,
turnKey: Schema.String,
});
const ThreadTurnRangeLookupInput = Schema.Struct({
threadId: ThreadId,
// Turn-linked rows are bounded by the keyset range [min, before) over
// (anchor, turn key); turnless rows by the matching [minAnchorAt,
// beforeAnchorAt) time range. Unbounded ends use sentinels: "" for the
// lower bound, "~" (sorts after ISO dates) for the upper bound.
minAnchorAt: Schema.String,
minTurnKey: Schema.String,
beforeAnchorAt: Schema.String,
beforeTurnKey: Schema.String,
});
const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema;
const ProjectionThreadIdLookupRowSchema = Schema.Struct({
threadId: ThreadId,
});
const ProjectionThreadCheckpointContextThreadRowSchema = Schema.Struct({
threadId: ThreadId,
projectId: ProjectId,
workspaceRoot: Schema.String,
worktreePath: Schema.NullOr(Schema.String),
});
const FullThreadDiffContextLookupInput = Schema.Struct({
threadId: ThreadId,
checkpointTurnCount: NonNegativeInt,
});
const ProjectionFullThreadDiffContextRowSchema = Schema.Struct({
threadId: ThreadId,
projectId: ProjectId,
workspaceRoot: Schema.String,
worktreePath: Schema.NullOr(Schema.String),
latestCheckpointTurnCount: Schema.NullOr(NonNegativeInt),
toCheckpointRef: Schema.NullOr(CheckpointRef),
});
const REQUIRED_SNAPSHOT_PROJECTORS = [
ORCHESTRATION_PROJECTOR_NAMES.projects,
ORCHESTRATION_PROJECTOR_NAMES.threads,
ORCHESTRATION_PROJECTOR_NAMES.threadMessages,
ORCHESTRATION_PROJECTOR_NAMES.threadProposedPlans,
ORCHESTRATION_PROJECTOR_NAMES.threadActivities,
ORCHESTRATION_PROJECTOR_NAMES.threadSessions,
ORCHESTRATION_PROJECTOR_NAMES.checkpoints,
] as const;
function maxIso(left: string | null, right: string): string {
if (left === null) {
return right;
}
return left > right ? left : right;
}
function escapeLikePattern(value: string): string {
return value.replaceAll("!", "!!").replaceAll("%", "!%").replaceAll("_", "!_");
}
function foldAsciiCase(value: string): string {
return value.replace(/[A-Z]/g, (character) => character.toLowerCase());
}
function buildSearchSnippet(text: string, query: string): string {
const normalizedText = text.replace(/\s+/g, " ").trim();
if (normalizedText.length <= 240) {
return normalizedText;
}
const normalizedQuery = foldAsciiCase(query.replace(/\s+/g, " ").trim());
const matchIndex = foldAsciiCase(normalizedText).indexOf(normalizedQuery);
const bodyLength = 236;
const idealStart = Math.max(0, matchIndex - 72);
const start = Math.min(idealStart, normalizedText.length - bodyLength);
const end = Math.min(normalizedText.length, start + bodyLength);
return `${start > 0 ? "…" : ""}${normalizedText.slice(start, end)}${
end < normalizedText.length ? "…" : ""
}`;
}
function computeSnapshotSequence(
stateRows: ReadonlyArray<Schema.Schema.Type<typeof ProjectionStateDbRowSchema>>,
): number {
if (stateRows.length === 0) {
return 0;
}
const sequenceByProjector = new Map(
stateRows.map((row) => [row.projector, row.lastAppliedSequence] as const),
);
let minSequence = Number.POSITIVE_INFINITY;
for (const projector of REQUIRED_SNAPSHOT_PROJECTORS) {
const sequence = sequenceByProjector.get(projector);
if (sequence === undefined) {
return 0;
}
if (sequence < minSequence) {
minSequence = sequence;
}
}
return Number.isFinite(minSequence) ? minSequence : 0;
}
function mapLatestTurn(
row: Schema.Schema.Type<typeof ProjectionLatestTurnDbRowSchema>,
): OrchestrationLatestTurn {
return {
turnId: row.turnId,
state:
row.state === "error"
? "error"
: row.state === "interrupted"
? "interrupted"
: row.state === "completed"
? "completed"
: "running",
requestedAt: row.requestedAt,
startedAt: row.startedAt,
completedAt: row.completedAt,
assistantMessageId: row.assistantMessageId,
...(row.sourceProposedPlanThreadId !== null && row.sourceProposedPlanId !== null
? {
sourceProposedPlan: {
threadId: row.sourceProposedPlanThreadId,
planId: row.sourceProposedPlanId,
},
}
: {}),
};
}
function mapTitleRegeneration(row: Schema.Schema.Type<typeof ProjectionThreadDbRowSchema>) {
return row.titleRegenerationRequestId != null && row.titleRegenerationStartedAt != null
? {
requestId: row.titleRegenerationRequestId,
startedAt: row.titleRegenerationStartedAt,
}
: null;
}
function mapSessionRow(
row: Schema.Schema.Type<typeof ProjectionThreadSessionDbRowSchema>,
): OrchestrationSession {
return {
threadId: row.threadId,
status: row.status,
providerName: row.providerName,
...(row.providerInstanceId !== null ? { providerInstanceId: row.providerInstanceId } : {}),
runtimeMode: row.runtimeMode,
activeTurnId: row.activeTurnId,
lastError: row.lastError,
updatedAt: row.updatedAt,
};
}
function mapProjectShellRow(
row: Schema.Schema.Type<typeof ProjectionProjectDbRowSchema>,
repositoryIdentity: OrchestrationProject["repositoryIdentity"],
): OrchestrationProjectShell {
return {
id: row.projectId,
title: row.title,
workspaceRoot: row.workspaceRoot,
repositoryIdentity,
defaultModelSelection: row.defaultModelSelection,
defaultThreadEnvMode: row.defaultThreadEnvMode,
autoPull: row.autoPull === 1,
faviconPath: row.faviconPath ?? null,
projectIcon: row.projectIcon ?? null,
scripts: row.scripts,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
}
function mapProposedPlanRow(
row: Schema.Schema.Type<typeof ProjectionThreadProposedPlanDbRowSchema>,
): OrchestrationProposedPlan {
return {
id: row.planId,
turnId: row.turnId,
planMarkdown: row.planMarkdown,
implementedAt: row.implementedAt,
implementationThreadId: row.implementationThreadId,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
}
function mapThreadActivityRow(
row: Schema.Schema.Type<typeof ProjectionThreadActivityDbRowSchema>,
): OrchestrationThreadActivity {
return {
id: row.activityId,
tone: row.tone,
kind: row.kind,
summary: row.summary,
payload: row.payload,
turnId: row.turnId,
createdAt: row.createdAt,
...(row.sequence !== null ? { sequence: row.sequence } : {}),
};
}
function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) {
return (cause: unknown): ProjectionRepositoryError =>
Schema.isSchemaError(cause)
? toPersistenceDecodeError(decodeOperation)(cause)
: toPersistenceSqlError(sqlOperation)(cause);
}
const makeProjectionSnapshotQuery = Effect.gen(function* () {
const threadBackgroundLiveness = yield* ThreadBackgroundLivenessService;
const threadPlanProgress = yield* ThreadPlanProgressService;
const sql = yield* SqlClient.SqlClient;
const repositoryIdentityResolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
const repositoryIdentityResolutionConcurrency = 4;
const resolveRepositoryIdentitiesForProjects = Effect.fn(
"ProjectionSnapshotQuery.resolveRepositoryIdentitiesForProjects",
)(function* (
projectRows: ReadonlyArray<Schema.Schema.Type<typeof ProjectionProjectDbRowSchema>>,
options?: {
readonly includeDeleted?: boolean;
},
) {
const filteredProjectRows =
options?.includeDeleted === true
? projectRows
: projectRows.filter((row) => row.deletedAt === null);
const uniqueWorkspaceRoots = [...new Set(filteredProjectRows.map((row) => row.workspaceRoot))];
const repositoryIdentityByWorkspaceRoot = new Map(
yield* Effect.forEach(
uniqueWorkspaceRoots,
(workspaceRoot) =>
repositoryIdentityResolver
.resolve(workspaceRoot)
.pipe(Effect.map((identity) => [workspaceRoot, identity] as const)),
{ concurrency: repositoryIdentityResolutionConcurrency },
),
);
return new Map(
filteredProjectRows.map((row) => [
row.projectId,
repositoryIdentityByWorkspaceRoot.get(row.workspaceRoot) ?? null,
]),
);
});
const listProjectRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionProjectDbRowSchema,
execute: () =>
sql`
SELECT
project_id AS "projectId",
title,
workspace_root AS "workspaceRoot",
default_model_selection_json AS "defaultModelSelection",
default_thread_env_mode AS "defaultThreadEnvMode",
auto_pull AS "autoPull",
favicon_path AS "faviconPath",
project_icon_json AS "projectIcon",
scripts_json AS "scripts",
created_at AS "createdAt",
updated_at AS "updatedAt",
deleted_at AS "deletedAt"
FROM projection_projects
ORDER BY created_at ASC, project_id ASC
`,
});
const listThreadRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
project_id AS "projectId",
title,
model_selection_json AS "modelSelection",
runtime_mode AS "runtimeMode",
interaction_mode AS "interactionMode",
branch,
worktree_path AS "worktreePath",
linked_pull_request_json AS "linkedPullRequest",
latest_turn_id AS "latestTurnId",
created_at AS "createdAt",
updated_at AS "updatedAt",
archived_at AS "archivedAt",
settled_override AS "settledOverride",
settled_at AS "settledAt",
unsettled_at AS "unsettledAt",
snoozed_until AS "snoozedUntil",
snoozed_at AS "snoozedAt",
pinned_at AS "pinnedAt",
pin_order_key AS "pinOrderKey",
title_regeneration_request_id AS "titleRegenerationRequestId",
title_regeneration_started_at AS "titleRegenerationStartedAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
has_actionable_proposed_plan AS "hasActionableProposedPlan",
deleted_at AS "deletedAt"
FROM projection_threads
ORDER BY created_at ASC, thread_id ASC
`,
});
const listActiveThreadRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
project_id AS "projectId",
title,
model_selection_json AS "modelSelection",
runtime_mode AS "runtimeMode",
interaction_mode AS "interactionMode",
branch,
worktree_path AS "worktreePath",
linked_pull_request_json AS "linkedPullRequest",
latest_turn_id AS "latestTurnId",
created_at AS "createdAt",
updated_at AS "updatedAt",
archived_at AS "archivedAt",
settled_override AS "settledOverride",
settled_at AS "settledAt",
unsettled_at AS "unsettledAt",
snoozed_until AS "snoozedUntil",
snoozed_at AS "snoozedAt",
pinned_at AS "pinnedAt",
pin_order_key AS "pinOrderKey",
title_regeneration_request_id AS "titleRegenerationRequestId",
title_regeneration_started_at AS "titleRegenerationStartedAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
has_actionable_proposed_plan AS "hasActionableProposedPlan",
deleted_at AS "deletedAt"
FROM projection_threads
WHERE deleted_at IS NULL
AND archived_at IS NULL
ORDER BY project_id ASC, created_at ASC, thread_id ASC
`,
});
const listArchivedThreadRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
project_id AS "projectId",
title,
model_selection_json AS "modelSelection",
runtime_mode AS "runtimeMode",
interaction_mode AS "interactionMode",
branch,
worktree_path AS "worktreePath",
linked_pull_request_json AS "linkedPullRequest",
latest_turn_id AS "latestTurnId",
created_at AS "createdAt",
updated_at AS "updatedAt",
archived_at AS "archivedAt",
settled_override AS "settledOverride",
settled_at AS "settledAt",
unsettled_at AS "unsettledAt",
snoozed_until AS "snoozedUntil",
snoozed_at AS "snoozedAt",
pinned_at AS "pinnedAt",
pin_order_key AS "pinOrderKey",
title_regeneration_request_id AS "titleRegenerationRequestId",
title_regeneration_started_at AS "titleRegenerationStartedAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
has_actionable_proposed_plan AS "hasActionableProposedPlan",
deleted_at AS "deletedAt"
FROM projection_threads
WHERE deleted_at IS NULL
AND archived_at IS NOT NULL
ORDER BY project_id ASC, archived_at DESC, thread_id DESC
`,
});
const listThreadMessageRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadMessageDbRowSchema,
execute: () =>
sql`
SELECT
message_id AS "messageId",
thread_id AS "threadId",
turn_id AS "turnId",
role,
text,
attachments_json AS "attachments",
is_streaming AS "isStreaming",
created_at AS "createdAt",
updated_at AS "updatedAt"
FROM projection_thread_messages
ORDER BY thread_id ASC, created_at ASC, message_id ASC
`,
});
const listThreadProposedPlanRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadProposedPlanDbRowSchema,
execute: () =>
sql`
SELECT
plan_id AS "planId",
thread_id AS "threadId",
turn_id AS "turnId",
plan_markdown AS "planMarkdown",
implemented_at AS "implementedAt",
implementation_thread_id AS "implementationThreadId",
created_at AS "createdAt",
updated_at AS "updatedAt"
FROM projection_thread_proposed_plans
ORDER BY thread_id ASC, created_at ASC, plan_id ASC
`,
});
const listThreadActivityRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadActivityDbRowSchema,
execute: () =>
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
ORDER BY
thread_id ASC,
sequence ASC,
created_at ASC,
activity_id ASC
`,
});
const listThreadSessionRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadSessionDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
status,
provider_name AS "providerName",
provider_instance_id AS "providerInstanceId",
provider_session_id AS "providerSessionId",
provider_thread_id AS "providerThreadId",
runtime_mode AS "runtimeMode",
active_turn_id AS "activeTurnId",
last_error AS "lastError",
updated_at AS "updatedAt"
FROM projection_thread_sessions
ORDER BY thread_id ASC
`,
});
const listActiveThreadSessionRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadSessionDbRowSchema,
execute: () =>
sql`
SELECT
sessions.thread_id AS "threadId",
sessions.status,
sessions.provider_name AS "providerName",
sessions.provider_instance_id AS "providerInstanceId",
sessions.provider_session_id AS "providerSessionId",
sessions.provider_thread_id AS "providerThreadId",
sessions.runtime_mode AS "runtimeMode",
sessions.active_turn_id AS "activeTurnId",
sessions.last_error AS "lastError",
sessions.updated_at AS "updatedAt"
FROM projection_thread_sessions sessions
INNER JOIN projection_threads threads
ON threads.thread_id = sessions.thread_id
WHERE threads.deleted_at IS NULL
AND threads.archived_at IS NULL
ORDER BY sessions.thread_id ASC
`,
});
const listArchivedThreadSessionRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionThreadSessionDbRowSchema,
execute: () =>
sql`
SELECT
sessions.thread_id AS "threadId",
sessions.status,
sessions.provider_name AS "providerName",
sessions.provider_instance_id AS "providerInstanceId",
sessions.provider_session_id AS "providerSessionId",
sessions.provider_thread_id AS "providerThreadId",
sessions.runtime_mode AS "runtimeMode",
sessions.active_turn_id AS "activeTurnId",
sessions.last_error AS "lastError",
sessions.updated_at AS "updatedAt"
FROM projection_thread_sessions sessions
INNER JOIN projection_threads threads
ON threads.thread_id = sessions.thread_id
WHERE threads.deleted_at IS NULL
AND threads.archived_at IS NOT NULL
ORDER BY sessions.thread_id ASC
`,
});
const listCheckpointRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionCheckpointDbRowSchema,
execute: () =>
sql`
SELECT
thread_id AS "threadId",
turn_id AS "turnId",
checkpoint_turn_count AS "checkpointTurnCount",
checkpoint_ref AS "checkpointRef",
checkpoint_status AS "status",
checkpoint_files_json AS "files",
assistant_message_id AS "assistantMessageId",
completed_at AS "completedAt"
FROM projection_turns
WHERE checkpoint_turn_count IS NOT NULL
ORDER BY thread_id ASC, checkpoint_turn_count ASC
`,
});
const listLatestTurnRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionLatestTurnDbRowSchema,
execute: () =>
sql`
SELECT
turns.thread_id AS "threadId",
turns.turn_id AS "turnId",
turns.state,
turns.requested_at AS "requestedAt",
turns.started_at AS "startedAt",
turns.completed_at AS "completedAt",
turns.assistant_message_id AS "assistantMessageId",
turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
turns.source_proposed_plan_id AS "sourceProposedPlanId"
FROM projection_threads threads
JOIN projection_turns turns
ON turns.thread_id = threads.thread_id
AND turns.turn_id = threads.latest_turn_id
WHERE threads.latest_turn_id IS NOT NULL
ORDER BY turns.thread_id ASC
`,
});
const listActiveLatestTurnRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionLatestTurnDbRowSchema,
execute: () =>
sql`
SELECT
turns.thread_id AS "threadId",
turns.turn_id AS "turnId",
turns.state,
turns.requested_at AS "requestedAt",
turns.started_at AS "startedAt",
turns.completed_at AS "completedAt",
turns.assistant_message_id AS "assistantMessageId",
turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
turns.source_proposed_plan_id AS "sourceProposedPlanId"
FROM projection_threads threads
JOIN projection_turns turns
ON turns.thread_id = threads.thread_id
AND turns.turn_id = threads.latest_turn_id
WHERE threads.deleted_at IS NULL
AND threads.archived_at IS NULL
AND threads.latest_turn_id IS NOT NULL
ORDER BY turns.thread_id ASC
`,
});
const listArchivedLatestTurnRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionLatestTurnDbRowSchema,
execute: () =>
sql`
SELECT
turns.thread_id AS "threadId",
turns.turn_id AS "turnId",
turns.state,
turns.requested_at AS "requestedAt",
turns.started_at AS "startedAt",
turns.completed_at AS "completedAt",
turns.assistant_message_id AS "assistantMessageId",
turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
turns.source_proposed_plan_id AS "sourceProposedPlanId"
FROM projection_threads threads
JOIN projection_turns turns
ON turns.thread_id = threads.thread_id
AND turns.turn_id = threads.latest_turn_id
WHERE threads.deleted_at IS NULL
AND threads.archived_at IS NOT NULL
AND threads.latest_turn_id IS NOT NULL
ORDER BY turns.thread_id ASC
`,
});
const listProjectionStateRows = SqlSchema.findAll({
Request: Schema.Void,
Result: ProjectionStateDbRowSchema,
execute: () =>
sql`
SELECT
projector,
last_applied_sequence AS "lastAppliedSequence",
updated_at AS "updatedAt"
FROM projection_state
`,
});
const readProjectionCounts = SqlSchema.findOne({
Request: Schema.Void,
Result: ProjectionCountsRowSchema,
execute: () =>
sql`
SELECT
(SELECT COUNT(*) FROM projection_projects) AS "projectCount",
(SELECT COUNT(*) FROM projection_threads) AS "threadCount"
`,
});
const readEventReplayStats = SqlSchema.findOne({
Request: EventReplayStatsInput,
Result: EventReplayStatsRowSchema,
execute: ({ fromSequenceExclusive, toSequenceInclusive }) =>
sql`
SELECT
COUNT(*) AS "eventCount",
COALESCE(SUM(octet_length(payload_json)), 0) AS "payloadBytes"
FROM orchestration_events
WHERE sequence > ${fromSequenceExclusive}
AND sequence <= ${toSequenceInclusive}
`,
});
const searchActiveThreadRows = SqlSchema.findAll({
Request: ProjectionThreadSearchRequest,
Result: ProjectionThreadSearchRow,
execute: ({ pattern, limit }) =>
sql`
WITH ranked AS (
SELECT
threads.thread_id AS thread_id,
threads.project_id AS project_id,
CASE messages.role
WHEN 'user' THEN 'user'
ELSE 'assistant'
END AS source,
messages.text AS match_text,
messages.created_at AS message_created_at,
CASE messages.role
WHEN 'user' THEN 0
ELSE 1
END AS match_rank,
threads.updated_at AS thread_updated_at,
ROW_NUMBER() OVER (
PARTITION BY threads.thread_id
ORDER BY
CASE messages.role
WHEN 'user' THEN 0
ELSE 1
END ASC,
messages.created_at DESC,
messages.message_id ASC
) AS thread_match_rank
FROM projection_thread_messages AS messages
INNER JOIN projection_threads AS threads
ON threads.thread_id = messages.thread_id
INNER JOIN projection_projects AS projects
ON projects.project_id = threads.project_id
WHERE threads.deleted_at IS NULL
AND threads.archived_at IS NULL
AND projects.deleted_at IS NULL
AND messages.is_streaming = 0
AND (
messages.role = 'user'
OR (
messages.role = 'assistant'
AND messages.message_id IN (
SELECT turns.assistant_message_id
FROM projection_turns AS turns
WHERE turns.assistant_message_id IS NOT NULL
)
)
)
AND messages.text LIKE ${pattern} ESCAPE '!'
)
SELECT
thread_id AS "threadId",
project_id AS "projectId",
source,
match_text AS "matchText",
message_created_at AS "messageCreatedAt"
FROM ranked
WHERE thread_match_rank = 1
ORDER BY
match_rank ASC,
thread_updated_at DESC,
thread_id ASC
LIMIT ${limit}
`,
});
const getActiveProjectRowByWorkspaceRoot = SqlSchema.findOneOption({
Request: WorkspaceRootLookupInput,
Result: ProjectionProjectLookupRowSchema,
execute: ({ workspaceRoot }) =>
sql`
SELECT
project_id AS "projectId",
title,
workspace_root AS "workspaceRoot",
default_model_selection_json AS "defaultModelSelection",
default_thread_env_mode AS "defaultThreadEnvMode",
auto_pull AS "autoPull",
favicon_path AS "faviconPath",
project_icon_json AS "projectIcon",
scripts_json AS "scripts",
created_at AS "createdAt",
updated_at AS "updatedAt",
deleted_at AS "deletedAt"
FROM projection_projects
WHERE workspace_root = ${workspaceRoot}
AND deleted_at IS NULL
ORDER BY created_at ASC, project_id ASC
LIMIT 1
`,
});
const getActiveProjectRowById = SqlSchema.findOneOption({
Request: ProjectIdLookupInput,
Result: ProjectionProjectLookupRowSchema,
execute: ({ projectId }) =>
sql`
SELECT
project_id AS "projectId",
title,
workspace_root AS "workspaceRoot",
default_model_selection_json AS "defaultModelSelection",
default_thread_env_mode AS "defaultThreadEnvMode",
auto_pull AS "autoPull",
favicon_path AS "faviconPath",
project_icon_json AS "projectIcon",
scripts_json AS "scripts",
created_at AS "createdAt",
updated_at AS "updatedAt",
deleted_at AS "deletedAt"
FROM projection_projects
WHERE project_id = ${projectId}
AND deleted_at IS NULL
LIMIT 1
`,
});
const getFirstActiveThreadIdByProject = SqlSchema.findOneOption({
Request: ProjectIdLookupInput,
Result: ProjectionThreadIdLookupRowSchema,
execute: ({ projectId }) =>
sql`
SELECT
thread_id AS "threadId"
FROM projection_threads
WHERE project_id = ${projectId}
AND deleted_at IS NULL
AND archived_at IS NULL
ORDER BY created_at ASC, thread_id ASC
LIMIT 1
`,
});
const getThreadCheckpointContextThreadRow = SqlSchema.findOneOption({
Request: ThreadIdLookupInput,
Result: ProjectionThreadCheckpointContextThreadRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
threads.thread_id AS "threadId",
threads.project_id AS "projectId",
projects.workspace_root AS "workspaceRoot",
threads.worktree_path AS "worktreePath"
FROM projection_threads AS threads
INNER JOIN projection_projects AS projects
ON projects.project_id = threads.project_id
WHERE threads.thread_id = ${threadId}
AND threads.deleted_at IS NULL
LIMIT 1
`,
});
const getActiveThreadRowById = SqlSchema.findOneOption({
Request: ThreadIdLookupInput,
Result: ProjectionThreadDbRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
thread_id AS "threadId",
project_id AS "projectId",
title,
model_selection_json AS "modelSelection",
runtime_mode AS "runtimeMode",
interaction_mode AS "interactionMode",
branch,
worktree_path AS "worktreePath",
linked_pull_request_json AS "linkedPullRequest",
latest_turn_id AS "latestTurnId",
created_at AS "createdAt",
updated_at AS "updatedAt",
archived_at AS "archivedAt",
settled_override AS "settledOverride",
settled_at AS "settledAt",
unsettled_at AS "unsettledAt",
snoozed_until AS "snoozedUntil",
snoozed_at AS "snoozedAt",
pinned_at AS "pinnedAt",
pin_order_key AS "pinOrderKey",
title_regeneration_request_id AS "titleRegenerationRequestId",
title_regeneration_started_at AS "titleRegenerationStartedAt",
latest_user_message_at AS "latestUserMessageAt",
pending_approval_count AS "pendingApprovalCount",
pending_user_input_count AS "pendingUserInputCount",
has_actionable_proposed_plan AS "hasActionableProposedPlan",
deleted_at AS "deletedAt"
FROM projection_threads
WHERE thread_id = ${threadId}
AND deleted_at IS NULL
AND archived_at IS NULL
LIMIT 1
`,
});
const getThreadRuntimeContextRow = SqlSchema.findOneOption({
Request: ThreadIdLookupInput,
Result: ProjectionThreadRuntimeContextDbRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
threads.thread_id AS id,
threads.title,
sessions.thread_id AS "threadId",
sessions.status,
sessions.provider_name AS "providerName",
sessions.provider_instance_id AS "providerInstanceId",
sessions.runtime_mode AS "runtimeMode",
sessions.active_turn_id AS "activeTurnId",
sessions.last_error AS "lastError",
sessions.updated_at AS "updatedAt"
FROM projection_threads AS threads
LEFT JOIN projection_thread_sessions AS sessions
ON sessions.thread_id = threads.thread_id
WHERE threads.thread_id = ${threadId}
AND threads.deleted_at IS NULL
AND threads.archived_at IS NULL
LIMIT 1
`.pipe(
Effect.map((rows) =>
rows.map((row) => ({
id: row.id,
title: row.title,
session: row.threadId === null ? null : row,
})),
),
),
});
const listThreadMessageRowsByThread = SqlSchema.findAll({
Request: ThreadIdLookupInput,
Result: ProjectionThreadMessageDbRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
message_id AS "messageId",
thread_id AS "threadId",
turn_id AS "turnId",
role,
text,
attachments_json AS "attachments",
is_streaming AS "isStreaming",
created_at AS "createdAt",
updated_at AS "updatedAt"
FROM projection_thread_messages
WHERE thread_id = ${threadId}
ORDER BY created_at ASC, message_id ASC
`,
});
const listThreadProposedPlanRowsByThread = SqlSchema.findAll({
Request: ThreadIdLookupInput,
Result: ProjectionThreadProposedPlanDbRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
plan_id AS "planId",
thread_id AS "threadId",
turn_id AS "turnId",
plan_markdown AS "planMarkdown",
implemented_at AS "implementedAt",
implementation_thread_id AS "implementationThreadId",
created_at AS "createdAt",
updated_at AS "updatedAt"
FROM projection_thread_proposed_plans
WHERE thread_id = ${threadId}
ORDER BY created_at ASC, plan_id ASC
`,
});
const listThreadActivityRowsByThread = SqlSchema.findAll({
Request: ThreadIdLookupInput,
Result: ProjectionThreadActivityDbRowSchema,
execute: ({ threadId }) =>
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 (
SELECT
activity_id,
thread_id,
turn_id,
tone,
kind,
summary,
payload_json,
sequence,
created_at
FROM projection_thread_activities
WHERE thread_id = ${threadId}
ORDER BY
sequence DESC,
created_at DESC,
activity_id DESC
LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
) AS recent_activities
ORDER BY
sequence ASC,
created_at ASC,
activity_id ASC
`,
});
const getUserInputActivityRow = SqlSchema.findOneOption({
Request: Schema.Struct({ threadId: ThreadId, requestId: ApprovalRequestId }),
Result: ProjectionThreadActivityDbRowSchema,
execute: ({ threadId, requestId }) => 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 ('user-input.requested', 'user-input.resolved')
AND json_extract(payload_json, '$.requestId') = ${requestId}
ORDER BY sequence DESC, created_at DESC, activity_id DESC
LIMIT 1
`,
});
const getUserInputActivity: ProjectionSnapshotQueryShape["getUserInputActivity"] = (input) =>
getUserInputActivityRow(input).pipe(
Effect.map(Option.map(mapThreadActivityRow)),
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getUserInputActivity:query",
"ProjectionSnapshotQuery.getUserInputActivity:decodeRow",
),
),
);
const listThreadActivityIdsByThread = SqlSchema.findAll({
Request: ThreadIdLookupInput,
Result: ProjectionThreadActivityIdRowSchema,
execute: ({ threadId }) =>
sql`
SELECT activity_id AS "activityId"
FROM projection_thread_activities
WHERE thread_id = ${threadId}
ORDER BY
sequence DESC,
created_at DESC,
activity_id DESC
LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
`,
});
const listThreadActivityRowsByIds = SqlSchema.findAll({
Request: ThreadActivityIdsLookupInput,
Result: ProjectionThreadActivityDbRowSchema,
execute: ({ activityIds }) =>
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
-- The selectors already scoped these globally unique ids to the
-- thread inside this transaction. Keep this as a primary-key lookup.
WHERE ${sql.in("activity_id", activityIds)}
`,
});
const listThreadActivityRowsByThreadAndKinds = SqlSchema.findAll({
Request: ThreadActivityKindsLookupInput,
Result: ProjectionThreadActivityDbRowSchema,
execute: ({ threadId, activityKinds }) =>
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 (
SELECT
activity_id,
thread_id,
turn_id,
tone,
kind,
summary,
payload_json,
sequence,
created_at
FROM projection_thread_activities
WHERE thread_id = ${threadId}
AND ${sql.in("kind", activityKinds)}
ORDER BY
sequence DESC,
created_at DESC,
activity_id DESC
LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
) AS recent_activities
ORDER BY
sequence ASC,
created_at ASC,
activity_id ASC
`,
});
const getThreadSessionRowByThread = SqlSchema.findOneOption({
Request: ThreadIdLookupInput,
Result: ProjectionThreadSessionDbRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
thread_id AS "threadId",
status,
provider_name AS "providerName",
provider_instance_id AS "providerInstanceId",
runtime_mode AS "runtimeMode",
active_turn_id AS "activeTurnId",
last_error AS "lastError",
updated_at AS "updatedAt"
FROM projection_thread_sessions
WHERE thread_id = ${threadId}
LIMIT 1
`,
});
const getLatestTurnRowByThread = SqlSchema.findOneOption({
Request: ThreadIdLookupInput,
Result: ProjectionLatestTurnDbRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
turns.thread_id AS "threadId",
turns.turn_id AS "turnId",
turns.state,
turns.requested_at AS "requestedAt",
turns.started_at AS "startedAt",
turns.completed_at AS "completedAt",
turns.assistant_message_id AS "assistantMessageId",
turns.source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
turns.source_proposed_plan_id AS "sourceProposedPlanId"
FROM projection_threads threads
JOIN projection_turns turns
ON turns.thread_id = threads.thread_id
AND turns.turn_id = threads.latest_turn_id
WHERE threads.thread_id = ${threadId}
AND threads.deleted_at IS NULL
AND threads.archived_at IS NULL
LIMIT 1
`,
});
const listCheckpointRowsByThread = SqlSchema.findAll({
Request: ThreadIdLookupInput,
Result: ProjectionCheckpointDbRowSchema,
execute: ({ threadId }) =>
sql`
SELECT
thread_id AS "threadId",
turn_id AS "turnId",
checkpoint_turn_count AS "checkpointTurnCount",
checkpoint_ref AS "checkpointRef",
checkpoint_status AS "status",
checkpoint_files_json AS "files",
assistant_message_id AS "assistantMessageId",
completed_at AS "completedAt"
FROM projection_turns
WHERE thread_id = ${threadId}
AND checkpoint_turn_count IS NOT NULL
ORDER BY checkpoint_turn_count ASC
`,
});
// Resolves a page of recent turns for a windowed thread detail read. Walks
// back from the exclusive (beforeAnchorAt, beforeTurnKey) keyset boundary
// (sentinels "~"/"" mean unbounded, i.e. the first page) until it has seen
// `userTurnLimit` user-anchored turns — turns whose pending message is a
// user message; subagent/fan-out turns between them ride along — or hits the
// `maxRawTurns` ceiling that bounds pathological fan-out. The `candidates`
// CTE applies the keyset bound and LIMIT before the window functions run;
// its ORDER BY uses raw columns so the migration-037
// (thread_id, requested_at, turn_id) index serves both range and order with
// no temp B-tree — the scan is genuinely bounded by the LIMIT. (Raw
// turn_id DESC places NULLs exactly where COALESCE-to-'' would, below every
// real id.) The caller derives the continuation cursor from the oldest
// returned row.
// Highest thread-DETAIL event sequence for this thread that the projection
// has applied (bounded by the global snapshot sequence read in the same
// transaction). This is the thread-scoped watermark a windowed page carries
// so clients can defer merging until their live subscription has caught up;
// the global sequence is not waitable per-thread. The event_type filter
// must match ws.ts's isThreadDetailEvent exactly: the subscription only
// delivers these types, so a watermark counting any other event could
// never be reached by the client and would park the page forever. Served
// by the event store's (aggregate_kind, stream_id, sequence) index.
const getThreadEventWatermarkRow = SqlSchema.findOneOption({
Request: Schema.Struct({ threadId: ThreadId, maxSequence: Schema.Number }),
Result: Schema.Struct({ threadSequence: Schema.NullOr(Schema.Number) }),
execute: ({ threadId, maxSequence }) =>
sql`
SELECT MAX(sequence) AS "threadSequence"
FROM orchestration_events
WHERE aggregate_kind = 'thread'
AND stream_id = ${threadId}
AND sequence <= ${maxSequence}
AND event_type IN (
'thread.message-sent',
'thread.proposed-plan-upserted',
'thread.activity-appended',
'thread.turn-diff-completed',
'thread.reverted',
'thread.session-set'
)
`,
});
const listTurnWindowRows = SqlSchema.findAll({
Request: ThreadTurnWindowLookupInput,
Result: ProjectionTurnWindowRowSchema,
execute: ({ threadId, beforeAnchorAt, beforeTurnKey, userTurnLimit, maxRawTurns }) =>
sql`
WITH candidates AS (
SELECT
turns.requested_at AS anchor_at,
COALESCE(turns.turn_id, '') AS turn_key,
turns.pending_message_id
FROM projection_turns AS turns
WHERE turns.thread_id = ${threadId}
AND (
turns.requested_at < ${beforeAnchorAt}
OR (
turns.requested_at = ${beforeAnchorAt}
AND COALESCE(turns.turn_id, '') < ${beforeTurnKey}
)
)
ORDER BY turns.requested_at DESC, turns.turn_id DESC
LIMIT ${maxRawTurns}
),
walked AS (
SELECT
candidates.anchor_at,
candidates.turn_key,
CASE WHEN messages.role = 'user' THEN 1 ELSE 0 END AS is_user_turn,
SUM(CASE WHEN messages.role = 'user' THEN 1 ELSE 0 END) OVER (
ORDER BY candidates.anchor_at DESC, candidates.turn_key DESC
) AS user_turns_seen
FROM candidates
LEFT JOIN projection_thread_messages AS messages
ON messages.message_id = candidates.pending_message_id
)
SELECT
anchor_at AS "anchorAt",
turn_key AS "turnKey"
FROM walked
WHERE user_turns_seen < ${userTurnLimit}
OR (user_turns_seen = ${userTurnLimit} AND is_user_turn = 1)
ORDER BY anchor_at ASC, turn_key ASC
`,
});
// Windowed variants of the two heavy collections. Turn-linked rows are
// bounded by the page's (anchor, turn key) keyset range over
// projection_turns; rows with no turn linkage (user messages always, and
// turnless activities like pre-turn context-window updates) are bounded by
// the matching turn-anchor time range so they land on the same page as the
// turns around them. Proposed plans and checkpoints stay unwindowed: they
// are metadata-scale.
const listThreadMessageRowsByThreadWindow = SqlSchema.findAll({
Request: ThreadTurnRangeLookupInput,
Result: ProjectionThreadMessageDbRowSchema,
execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) =>
sql`
SELECT
message_id AS "messageId",
thread_id AS "threadId",
turn_id AS "turnId",
role,
text,
attachments_json AS "attachments",
is_streaming AS "isStreaming",
created_at AS "createdAt",
updated_at AS "updatedAt"
FROM projection_thread_messages
WHERE thread_id = ${threadId}
AND (
turn_id IN (
SELECT turn_id FROM projection_turns
WHERE thread_id = ${threadId}
AND turn_id IS NOT NULL
AND (
requested_at > ${minAnchorAt}
OR (
requested_at = ${minAnchorAt}
AND turn_id >= ${minTurnKey}
)
)
AND (
requested_at < ${beforeAnchorAt}
OR (
requested_at = ${beforeAnchorAt}
AND turn_id < ${beforeTurnKey}
)
)
)
OR (
turn_id IS NULL
AND created_at >= ${minAnchorAt}
AND created_at < ${beforeAnchorAt}
)
)
ORDER BY created_at ASC, message_id ASC
`,
});
const pinnedThreadActivityIdsCte = (threadId: string) => sql`
pending_approval_requests AS (
SELECT request_id, thread_id
FROM projection_pending_approvals
WHERE thread_id = ${threadId}
AND status = 'pending'
),
pending_approval_activities AS (
SELECT
activity.activity_id,
ROW_NUMBER() OVER (
PARTITION BY pending.request_id
ORDER BY activity.created_at DESC, activity.activity_id DESC
) AS request_order
FROM pending_approval_requests AS pending
CROSS JOIN projection_thread_activities AS activity
WHERE activity.thread_id = pending.thread_id
AND activity.kind = 'approval.requested'
AND json_extract(activity.payload_json, '$.requestId') = pending.request_id
),
pending_user_input_thread AS (
SELECT thread_id
FROM projection_threads
WHERE thread_id = ${threadId}
AND pending_user_input_count > 0
),
user_input_lifecycle AS (
SELECT
activity.activity_id,
activity.kind,
ROW_NUMBER() OVER (
PARTITION BY json_extract(activity.payload_json, '$.requestId')
ORDER BY activity.created_at DESC, activity.activity_id DESC
) AS request_order
FROM pending_user_input_thread AS pending
CROSS JOIN projection_thread_activities AS activity
WHERE activity.thread_id = pending.thread_id
AND (
activity.kind IN ('user-input.requested', 'user-input.resolved')
OR (
activity.kind = 'provider.user-input.respond.failed'
AND (
lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
LIKE '%stale pending user-input request%'
OR lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
LIKE '%unknown pending user-input request%'
OR lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
LIKE '%unknown pending user input request%'
OR lower(COALESCE(json_extract(activity.payload_json, '$.detail'), ''))
LIKE '%unknown pending codex user input request%'
)
)
)
AND json_extract(activity.payload_json, '$.requestId') IS NOT NULL
),
pinned_activity_ids AS (
SELECT activity_id
FROM pending_approval_activities
WHERE request_order = 1
UNION ALL
SELECT activity_id
FROM user_input_lifecycle
WHERE request_order = 1
AND kind = 'user-input.requested'
)
`;
// Blocking request payloads must remain available even if they predate the
// recent activity window. Each CTE returns at most one unresolved row per
// request, so the merge below stays bounded by actionable work.
const listPinnedThreadActivityRowsByThread = SqlSchema.findAll({
Request: ThreadIdLookupInput,
Result: ProjectionThreadActivityDbRowSchema,
execute: ({ threadId }) =>
sql`
WITH ${pinnedThreadActivityIdsCte(threadId)}
SELECT
activity.activity_id AS "activityId",
activity.thread_id AS "threadId",
activity.turn_id AS "turnId",
activity.tone,
activity.kind,
activity.summary,
activity.payload_json AS "payload",
activity.sequence,
activity.created_at AS "createdAt"
FROM pinned_activity_ids AS pinned
INNER JOIN projection_thread_activities AS activity
ON activity.activity_id = pinned.activity_id
ORDER BY activity.created_at ASC, activity.activity_id ASC
`,
});
const listPinnedThreadActivityIdsByThread = SqlSchema.findAll({
Request: ThreadIdLookupInput,
Result: ProjectionThreadActivityIdRowSchema,
execute: ({ threadId }) =>
sql`
WITH ${pinnedThreadActivityIdsCte(threadId)}
SELECT activity_id AS "activityId"
FROM pinned_activity_ids
`,
});
const listThreadActivityRowsByThreadWindow = SqlSchema.findAll({
Request: ThreadTurnRangeLookupInput,
Result: ProjectionThreadActivityDbRowSchema,
execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) =>
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 (
SELECT
activity_id,
thread_id,
turn_id,
tone,
kind,
summary,
payload_json,
sequence,
created_at
FROM projection_thread_activities
WHERE thread_id = ${threadId}
AND (
turn_id IN (
SELECT turn_id FROM projection_turns
WHERE thread_id = ${threadId}
AND turn_id IS NOT NULL
AND (
requested_at > ${minAnchorAt}
OR (
requested_at = ${minAnchorAt}
AND turn_id >= ${minTurnKey}
)
)
AND (
requested_at < ${beforeAnchorAt}
OR (
requested_at = ${beforeAnchorAt}
AND turn_id < ${beforeTurnKey}
)
)
)
OR (
turn_id IS NULL
AND created_at >= ${minAnchorAt}
AND created_at < ${beforeAnchorAt}
)
)
ORDER BY
sequence DESC,
created_at DESC,
activity_id DESC
LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
) AS recent_activities
ORDER BY
sequence ASC,
created_at ASC,
activity_id ASC
`,
});
const listThreadActivityIdsByThreadWindow = SqlSchema.findAll({
Request: ThreadTurnRangeLookupInput,
Result: ProjectionThreadActivityIdRowSchema,
execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) =>
sql`
SELECT activity_id AS "activityId"
FROM projection_thread_activities
WHERE thread_id = ${threadId}
AND (
turn_id IN (
SELECT turn_id FROM projection_turns
WHERE thread_id = ${threadId}
AND turn_id IS NOT NULL
AND (
requested_at > ${minAnchorAt}
OR (
requested_at = ${minAnchorAt}
AND turn_id >= ${minTurnKey}
)
)
AND (
requested_at < ${beforeAnchorAt}
OR (
requested_at = ${beforeAnchorAt}
AND turn_id < ${beforeTurnKey}
)
)
)
OR (
turn_id IS NULL
AND created_at >= ${minAnchorAt}
AND created_at < ${beforeAnchorAt}
)
)
ORDER BY
sequence DESC,
created_at DESC,
activity_id DESC
LIMIT ${THREAD_DETAIL_ACTIVITY_LIMIT}
`,
});
const getFullThreadDiffContextRow = SqlSchema.findOneOption({
Request: FullThreadDiffContextLookupInput,
Result: ProjectionFullThreadDiffContextRowSchema,
execute: ({ threadId, checkpointTurnCount }) =>
sql`
SELECT
threads.thread_id AS "threadId",
threads.project_id AS "projectId",
projects.workspace_root AS "workspaceRoot",
threads.worktree_path AS "worktreePath",
(
SELECT MAX(turns.checkpoint_turn_count)
FROM projection_turns AS turns
WHERE turns.thread_id = threads.thread_id
AND turns.checkpoint_turn_count IS NOT NULL
) AS "latestCheckpointTurnCount",
(
SELECT turns.checkpoint_ref
FROM projection_turns AS turns
WHERE turns.thread_id = threads.thread_id
AND turns.checkpoint_turn_count = ${checkpointTurnCount}
LIMIT 1
) AS "toCheckpointRef"
FROM projection_threads AS threads
INNER JOIN projection_projects AS projects
ON projects.project_id = threads.project_id
WHERE threads.thread_id = ${threadId}
AND threads.deleted_at IS NULL
LIMIT 1
`,
});
const getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () =>
sql
.withTransaction(
Effect.all([
listProjectRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listProjects:query",
"ProjectionSnapshotQuery.getSnapshot:listProjects:decodeRows",
),
),
),
listThreadRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreads:query",
"ProjectionSnapshotQuery.getSnapshot:listThreads:decodeRows",
),
),
),
listThreadMessageRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreadMessages:query",
"ProjectionSnapshotQuery.getSnapshot:listThreadMessages:decodeRows",
),
),
),
listThreadProposedPlanRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:query",
"ProjectionSnapshotQuery.getSnapshot:listThreadProposedPlans:decodeRows",
),
),
),
listThreadActivityRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreadActivities:query",
"ProjectionSnapshotQuery.getSnapshot:listThreadActivities:decodeRows",
),
),
),
listThreadSessionRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listThreadSessions:query",
"ProjectionSnapshotQuery.getSnapshot:listThreadSessions:decodeRows",
),
),
),
listCheckpointRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listCheckpoints:query",
"ProjectionSnapshotQuery.getSnapshot:listCheckpoints:decodeRows",
),
),
),
listLatestTurnRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listLatestTurns:query",
"ProjectionSnapshotQuery.getSnapshot:listLatestTurns:decodeRows",
),
),
),
listProjectionStateRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshot:listProjectionState:query",
"ProjectionSnapshotQuery.getSnapshot:listProjectionState:decodeRows",
),
),
),
]),
)
.pipe(
Effect.flatMap(
([
projectRows,
threadRows,
messageRows,
proposedPlanRows,
activityRows,
sessionRows,
checkpointRows,
latestTurnRows,
stateRows,
]) =>
Effect.gen(function* () {
const messagesByThread = new Map<string, Array<OrchestrationMessage>>();
const proposedPlansByThread = new Map<string, Array<OrchestrationProposedPlan>>();
const activitiesByThread = new Map<string, Array<OrchestrationThreadActivity>>();
const checkpointsByThread = new Map<string, Array<OrchestrationCheckpointSummary>>();
const sessionsByThread = new Map<string, OrchestrationSession>();
const latestTurnByThread = new Map<string, OrchestrationLatestTurn>();
let updatedAt: string | null = null;
for (const row of projectRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of threadRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of stateRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of messageRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
const threadMessages = messagesByThread.get(row.threadId) ?? [];
threadMessages.push({
id: row.messageId,
role: row.role,
text: row.text,
...(row.attachments !== null ? { attachments: row.attachments } : {}),
turnId: row.turnId,
streaming: row.isStreaming === 1,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
});
messagesByThread.set(row.threadId, threadMessages);
}
for (const row of proposedPlanRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? [];
threadProposedPlans.push({
id: row.planId,
turnId: row.turnId,
planMarkdown: row.planMarkdown,
implementedAt: row.implementedAt,
implementationThreadId: row.implementationThreadId,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
});
proposedPlansByThread.set(row.threadId, threadProposedPlans);
}
for (const row of activityRows) {
updatedAt = maxIso(updatedAt, row.createdAt);
const threadActivities = activitiesByThread.get(row.threadId) ?? [];
threadActivities.push({
id: row.activityId,
tone: row.tone,
kind: row.kind,
summary: row.summary,
payload: row.payload,
turnId: row.turnId,
...(row.sequence !== null ? { sequence: row.sequence } : {}),
createdAt: row.createdAt,
});
activitiesByThread.set(row.threadId, threadActivities);
}
for (const row of checkpointRows) {
updatedAt = maxIso(updatedAt, row.completedAt);
const threadCheckpoints = checkpointsByThread.get(row.threadId) ?? [];
threadCheckpoints.push({
turnId: row.turnId,
checkpointTurnCount: row.checkpointTurnCount,
checkpointRef: row.checkpointRef,
status: row.status,
files: row.files,
assistantMessageId: row.assistantMessageId,
completedAt: row.completedAt,
});
checkpointsByThread.set(row.threadId, threadCheckpoints);
}
for (const row of latestTurnRows) {
updatedAt = maxIso(updatedAt, row.requestedAt);
if (row.startedAt !== null) {
updatedAt = maxIso(updatedAt, row.startedAt);
}
if (row.completedAt !== null) {
updatedAt = maxIso(updatedAt, row.completedAt);
}
if (latestTurnByThread.has(row.threadId)) {
continue;
}
latestTurnByThread.set(row.threadId, {
turnId: row.turnId,
state:
row.state === "error"
? "error"
: row.state === "interrupted"
? "interrupted"
: row.state === "completed"
? "completed"
: "running",
requestedAt: row.requestedAt,
startedAt: row.startedAt,
completedAt: row.completedAt,
assistantMessageId: row.assistantMessageId,
...(row.sourceProposedPlanThreadId !== null && row.sourceProposedPlanId !== null
? {
sourceProposedPlan: {
threadId: row.sourceProposedPlanThreadId,
planId: row.sourceProposedPlanId,
},
}
: {}),
});
}
for (const row of sessionRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
sessionsByThread.set(row.threadId, {
threadId: row.threadId,
status: row.status,
providerName: row.providerName,
...(row.providerInstanceId !== null
? { providerInstanceId: row.providerInstanceId }
: {}),
runtimeMode: row.runtimeMode,
activeTurnId: row.activeTurnId,
lastError: row.lastError,
updatedAt: row.updatedAt,
});
}
const repositoryIdentities = yield* resolveRepositoryIdentitiesForProjects(
projectRows,
{ includeDeleted: true },
);
const projects: ReadonlyArray<OrchestrationProject> = projectRows.map((row) => ({
id: row.projectId,
title: row.title,
workspaceRoot: row.workspaceRoot,
repositoryIdentity: repositoryIdentities.get(row.projectId) ?? null,
defaultModelSelection: row.defaultModelSelection,
defaultThreadEnvMode: row.defaultThreadEnvMode,
autoPull: row.autoPull === 1,
faviconPath: row.faviconPath ?? null,
projectIcon: row.projectIcon ?? null,
scripts: row.scripts,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
deletedAt: row.deletedAt,
}));
const threads: ReadonlyArray<OrchestrationThread> = threadRows.map((row) => ({
id: row.threadId,
projectId: row.projectId,
title: row.title,
modelSelection: row.modelSelection,
runtimeMode: row.runtimeMode,
interactionMode: row.interactionMode,
branch: row.branch,
worktreePath: row.worktreePath,
...(row.linkedPullRequest === null
? {}
: { linkedPullRequest: row.linkedPullRequest }),
latestTurn: latestTurnByThread.get(row.threadId) ?? null,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
settledOverride: row.settledOverride,
settledAt: row.settledAt,
unsettledAt: row.unsettledAt,
snoozedUntil: row.snoozedUntil,
snoozedAt: row.snoozedAt,
pinnedAt: row.pinnedAt,
pinOrderKey: row.pinOrderKey ?? null,
titleRegeneration: mapTitleRegeneration(row),
deletedAt: row.deletedAt,
messages: messagesByThread.get(row.threadId) ?? [],
proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
activities: activitiesByThread.get(row.threadId) ?? [],
checkpoints: checkpointsByThread.get(row.threadId) ?? [],
session: sessionsByThread.get(row.threadId) ?? null,
}));
const snapshot = {
snapshotSequence: computeSnapshotSequence(stateRows),
projects,
threads,
updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z",
};
return yield* decodeReadModel(snapshot).pipe(
Effect.mapError(
toPersistenceDecodeError("ProjectionSnapshotQuery.getSnapshot:decodeReadModel"),
),
);
}),
),
Effect.mapError((error) => {
if (isPersistenceError(error)) {
return error;
}
return toPersistenceSqlError("ProjectionSnapshotQuery.getSnapshot:query")(error);
}),
);
const getCommandReadModel: ProjectionSnapshotQueryShape["getCommandReadModel"] = () =>
sql
.withTransaction(
Effect.all([
listProjectRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getCommandReadModel:listProjects:query",
"ProjectionSnapshotQuery.getCommandReadModel:listProjects:decodeRows",
),
),
),
listThreadRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getCommandReadModel:listThreads:query",
"ProjectionSnapshotQuery.getCommandReadModel:listThreads:decodeRows",
),
),
),
listThreadProposedPlanRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getCommandReadModel:listThreadProposedPlans:query",
"ProjectionSnapshotQuery.getCommandReadModel:listThreadProposedPlans:decodeRows",
),
),
),
listThreadSessionRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getCommandReadModel:listThreadSessions:query",
"ProjectionSnapshotQuery.getCommandReadModel:listThreadSessions:decodeRows",
),
),
),
listLatestTurnRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getCommandReadModel:listLatestTurns:query",
"ProjectionSnapshotQuery.getCommandReadModel:listLatestTurns:decodeRows",
),
),
),
listProjectionStateRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getCommandReadModel:listProjectionState:query",
"ProjectionSnapshotQuery.getCommandReadModel:listProjectionState:decodeRows",
),
),
),
]),
)
.pipe(
Effect.flatMap(
([projectRows, threadRows, proposedPlanRows, sessionRows, latestTurnRows, stateRows]) =>
Effect.sync(() => {
let updatedAt: string | null = null;
const projects: OrchestrationProject[] = [];
const threads: OrchestrationThread[] = [];
for (let index = 0; index < projectRows.length; index += 1) {
const row = projectRows[index];
if (!row) {
continue;
}
updatedAt = maxIso(updatedAt, row.updatedAt);
projects.push({
id: row.projectId,
title: row.title,
workspaceRoot: row.workspaceRoot,
defaultModelSelection: row.defaultModelSelection,
defaultThreadEnvMode: row.defaultThreadEnvMode,
autoPull: row.autoPull === 1,
faviconPath: row.faviconPath ?? null,
projectIcon: row.projectIcon ?? null,
scripts: row.scripts,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
deletedAt: row.deletedAt,
});
}
for (let index = 0; index < threadRows.length; index += 1) {
const row = threadRows[index];
if (!row) {
continue;
}
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (let index = 0; index < proposedPlanRows.length; index += 1) {
const row = proposedPlanRows[index];
if (!row) {
continue;
}
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (let index = 0; index < sessionRows.length; index += 1) {
const row = sessionRows[index];
if (!row) {
continue;
}
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (let index = 0; index < latestTurnRows.length; index += 1) {
const row = latestTurnRows[index];
if (!row) {
continue;
}
updatedAt = maxIso(updatedAt, row.requestedAt);
if (row.startedAt !== null) {
updatedAt = maxIso(updatedAt, row.startedAt);
}
if (row.completedAt !== null) {
updatedAt = maxIso(updatedAt, row.completedAt);
}
}
for (let index = 0; index < stateRows.length; index += 1) {
const row = stateRows[index];
if (!row) {
continue;
}
updatedAt = maxIso(updatedAt, row.updatedAt);
}
const latestTurnByThread = new Map<string, OrchestrationLatestTurn>();
for (let index = 0; index < latestTurnRows.length; index += 1) {
const row = latestTurnRows[index];
if (!row) {
continue;
}
latestTurnByThread.set(row.threadId, mapLatestTurn(row));
}
const proposedPlansByThread = new Map<string, Array<OrchestrationProposedPlan>>();
const sessionByThread = new Map<string, OrchestrationSession>();
for (let index = 0; index < sessionRows.length; index += 1) {
const row = sessionRows[index];
if (!row) {
continue;
}
sessionByThread.set(row.threadId, mapSessionRow(row));
}
for (let index = 0; index < proposedPlanRows.length; index += 1) {
const row = proposedPlanRows[index];
if (!row) {
continue;
}
const threadProposedPlans = proposedPlansByThread.get(row.threadId) ?? [];
threadProposedPlans.push(mapProposedPlanRow(row));
proposedPlansByThread.set(row.threadId, threadProposedPlans);
}
for (let index = 0; index < threadRows.length; index += 1) {
const row = threadRows[index];
if (!row) {
continue;
}
threads.push({
id: row.threadId,
projectId: row.projectId,
title: row.title,
modelSelection: row.modelSelection,
runtimeMode: row.runtimeMode,
interactionMode: row.interactionMode,
branch: row.branch,
worktreePath: row.worktreePath,
...(row.linkedPullRequest === null
? {}
: { linkedPullRequest: row.linkedPullRequest }),
latestTurn: latestTurnByThread.get(row.threadId) ?? null,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
settledOverride: row.settledOverride,
settledAt: row.settledAt,
unsettledAt: row.unsettledAt,
snoozedUntil: row.snoozedUntil,
snoozedAt: row.snoozedAt,
pinnedAt: row.pinnedAt,
pinOrderKey: row.pinOrderKey ?? null,
titleRegeneration: mapTitleRegeneration(row),
deletedAt: row.deletedAt,
messages: [],
proposedPlans: proposedPlansByThread.get(row.threadId) ?? [],
activities: [],
checkpoints: [],
session: sessionByThread.get(row.threadId) ?? null,
});
}
return {
snapshotSequence: computeSnapshotSequence(stateRows),
projects,
threads,
updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z",
} satisfies OrchestrationReadModel;
}),
),
Effect.mapError((error) => {
if (isPersistenceError(error)) {
return error;
}
return toPersistenceSqlError("ProjectionSnapshotQuery.getCommandReadModel:query")(error);
}),
);
const getShellSnapshot: ProjectionSnapshotQueryShape["getShellSnapshot"] = () =>
sql
.withTransaction(
Effect.all([
listProjectRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getShellSnapshot:listProjects:query",
"ProjectionSnapshotQuery.getShellSnapshot:listProjects:decodeRows",
),
),
),
listActiveThreadRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getShellSnapshot:listThreads:query",
"ProjectionSnapshotQuery.getShellSnapshot:listThreads:decodeRows",
),
),
),
listActiveThreadSessionRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getShellSnapshot:listThreadSessions:query",
"ProjectionSnapshotQuery.getShellSnapshot:listThreadSessions:decodeRows",
),
),
),
listActiveLatestTurnRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getShellSnapshot:listLatestTurns:query",
"ProjectionSnapshotQuery.getShellSnapshot:listLatestTurns:decodeRows",
),
),
),
listProjectionStateRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getShellSnapshot:listProjectionState:query",
"ProjectionSnapshotQuery.getShellSnapshot:listProjectionState:decodeRows",
),
),
),
]),
)
.pipe(
Effect.flatMap(([projectRows, threadRows, sessionRows, latestTurnRows, stateRows]) =>
Effect.gen(function* () {
let updatedAt: string | null = null;
for (const row of projectRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of threadRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of sessionRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of latestTurnRows) {
updatedAt = maxIso(updatedAt, row.requestedAt);
if (row.startedAt !== null) {
updatedAt = maxIso(updatedAt, row.startedAt);
}
if (row.completedAt !== null) {
updatedAt = maxIso(updatedAt, row.completedAt);
}
}
for (const row of stateRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
const repositoryIdentities = yield* resolveRepositoryIdentitiesForProjects(projectRows);
const latestTurnByThread = new Map(
latestTurnRows.map((row) => [row.threadId, mapLatestTurn(row)] as const),
);
const sessionByThread = new Map(
sessionRows.map((row) => [row.threadId, mapSessionRow(row)] as const),
);
const snapshot = {
snapshotSequence: computeSnapshotSequence(stateRows),
projects: Arr.filterMap(projectRows, (row) =>
row.deletedAt === null
? Result.succeed(
mapProjectShellRow(row, repositoryIdentities.get(row.projectId) ?? null),
)
: Result.failVoid,
),
threads: Arr.filterMap(threadRows, (row) =>
row.deletedAt === null
? Result.succeed({
id: row.threadId,
projectId: row.projectId,
title: row.title,
modelSelection: row.modelSelection,
runtimeMode: row.runtimeMode,
interactionMode: row.interactionMode,
branch: row.branch,
worktreePath: row.worktreePath,
...(row.linkedPullRequest === null
? {}
: { linkedPullRequest: row.linkedPullRequest }),
latestTurn: latestTurnByThread.get(row.threadId) ?? null,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
settledOverride: row.settledOverride,
settledAt: row.settledAt,
unsettledAt: row.unsettledAt,
snoozedUntil: row.snoozedUntil,
snoozedAt: row.snoozedAt,
pinnedAt: row.pinnedAt,
pinOrderKey: row.pinOrderKey ?? null,
titleRegeneration: mapTitleRegeneration(row),
session: sessionByThread.get(row.threadId) ?? null,
latestUserMessageAt: row.latestUserMessageAt,
hasPendingApprovals: row.pendingApprovalCount > 0,
hasPendingUserInput: row.pendingUserInputCount > 0,
hasActionableProposedPlan: row.hasActionableProposedPlan > 0,
backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness(
row.threadId,
),
planProgress: threadPlanProgress.getThreadPlanProgress(row.threadId),
} satisfies OrchestrationThreadShell)
: Result.failVoid,
),
updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z",
};
return yield* decodeShellSnapshot(snapshot).pipe(
Effect.mapError(
toPersistenceDecodeError(
"ProjectionSnapshotQuery.getShellSnapshot:decodeShellSnapshot",
),
),
);
}),
),
Effect.mapError((error) => {
if (isPersistenceError(error)) {
return error;
}
return toPersistenceSqlError("ProjectionSnapshotQuery.getShellSnapshot:query")(error);
}),
);
const getArchivedShellSnapshot: ProjectionSnapshotQueryShape["getArchivedShellSnapshot"] = () =>
sql
.withTransaction(
Effect.all([
listProjectRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjects:query",
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjects:decodeRows",
),
),
),
listArchivedThreadRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreads:query",
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreads:decodeRows",
),
),
),
listArchivedThreadSessionRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreadSessions:query",
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listThreadSessions:decodeRows",
),
),
),
listArchivedLatestTurnRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listLatestTurns:query",
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listLatestTurns:decodeRows",
),
),
),
listProjectionStateRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjectionState:query",
"ProjectionSnapshotQuery.getArchivedShellSnapshot:listProjectionState:decodeRows",
),
),
),
]),
)
.pipe(
Effect.flatMap(([projectRows, threadRows, sessionRows, latestTurnRows, stateRows]) =>
Effect.gen(function* () {
let updatedAt: string | null = null;
for (const row of projectRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of threadRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of sessionRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
for (const row of latestTurnRows) {
updatedAt = maxIso(updatedAt, row.requestedAt);
if (row.startedAt !== null) {
updatedAt = maxIso(updatedAt, row.startedAt);
}
if (row.completedAt !== null) {
updatedAt = maxIso(updatedAt, row.completedAt);
}
}
for (const row of stateRows) {
updatedAt = maxIso(updatedAt, row.updatedAt);
}
const activeProjectIds = new Set(threadRows.map((row) => row.projectId));
const repositoryIdentities = yield* resolveRepositoryIdentitiesForProjects(
projectRows.filter((row) => activeProjectIds.has(row.projectId)),
);
const latestTurnByThread = new Map(
latestTurnRows.map((row) => [row.threadId, mapLatestTurn(row)] as const),
);
const sessionByThread = new Map(
sessionRows.map((row) => [row.threadId, mapSessionRow(row)] as const),
);
const snapshot = {
snapshotSequence: computeSnapshotSequence(stateRows),
projects: Arr.filterMap(projectRows, (row) =>
row.deletedAt === null && activeProjectIds.has(row.projectId)
? Result.succeed(
mapProjectShellRow(row, repositoryIdentities.get(row.projectId) ?? null),
)
: Result.failVoid,
),
threads: threadRows.map((row): OrchestrationThreadShell => ({
id: row.threadId,
projectId: row.projectId,
title: row.title,
modelSelection: row.modelSelection,
runtimeMode: row.runtimeMode,
interactionMode: row.interactionMode,
branch: row.branch,
worktreePath: row.worktreePath,
...(row.linkedPullRequest === null
? {}
: { linkedPullRequest: row.linkedPullRequest }),
latestTurn: latestTurnByThread.get(row.threadId) ?? null,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
archivedAt: row.archivedAt,
settledOverride: row.settledOverride,
settledAt: row.settledAt,
unsettledAt: row.unsettledAt,
snoozedUntil: row.snoozedUntil,
snoozedAt: row.snoozedAt,
pinnedAt: row.pinnedAt,
pinOrderKey: row.pinOrderKey ?? null,
titleRegeneration: mapTitleRegeneration(row),
session: sessionByThread.get(row.threadId) ?? null,
latestUserMessageAt: row.latestUserMessageAt,
hasPendingApprovals: row.pendingApprovalCount > 0,
hasPendingUserInput: row.pendingUserInputCount > 0,
hasActionableProposedPlan: row.hasActionableProposedPlan > 0,
backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness(
row.threadId,
),
planProgress: threadPlanProgress.getThreadPlanProgress(row.threadId),
})),
updatedAt: updatedAt ?? "1970-01-01T00:00:00.000Z",
};
return yield* decodeShellSnapshot(snapshot).pipe(
Effect.mapError(
toPersistenceDecodeError(
"ProjectionSnapshotQuery.getArchivedShellSnapshot:decodeShellSnapshot",
),
),
);
}),
),
Effect.mapError((error) => {
if (isPersistenceError(error)) {
return error;
}
return toPersistenceSqlError("ProjectionSnapshotQuery.getArchivedShellSnapshot:query")(
error,
);
}),
);
const getSnapshotSequence: ProjectionSnapshotQueryShape["getSnapshotSequence"] = () =>
listProjectionStateRows(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getSnapshotSequence:query",
"ProjectionSnapshotQuery.getSnapshotSequence:decodeRows",
),
),
Effect.map((stateRows) => ({
snapshotSequence: computeSnapshotSequence(stateRows),
})),
);
const getCounts: ProjectionSnapshotQueryShape["getCounts"] = () =>
readProjectionCounts(undefined).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getCounts:query",
"ProjectionSnapshotQuery.getCounts:decodeRow",
),
),
Effect.map((row): ProjectionSnapshotCounts => ({
projectCount: row.projectCount,
threadCount: row.threadCount,
})),
);
const getEventReplayStats: ProjectionSnapshotQueryShape["getEventReplayStats"] = (input) =>
readEventReplayStats(input).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getEventReplayStats:query",
"ProjectionSnapshotQuery.getEventReplayStats:decodeRow",
),
),
Effect.map((row): ProjectionEventReplayStats => ({
eventCount: row.eventCount,
payloadBytes: row.payloadBytes,
})),
);
const searchThreads: ProjectionSnapshotQueryShape["searchThreads"] = Effect.fn(
"ProjectionSnapshotQuery.searchThreads",
)(function* (input) {
const escapedQuery = escapeLikePattern(input.query);
const rows = yield* searchActiveThreadRows({
pattern: `%${escapedQuery}%`,
limit: input.limit ?? 50,
}).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.searchThreads:query",
"ProjectionSnapshotQuery.searchThreads:decodeRows",
),
),
);
return {
matches: rows.map((row) => ({
threadId: row.threadId,
projectId: row.projectId,
source: row.source,
snippet: buildSearchSnippet(row.matchText, input.query),
messageCreatedAt: row.messageCreatedAt,
})),
};
});
const getActiveProjectByWorkspaceRoot: ProjectionSnapshotQueryShape["getActiveProjectByWorkspaceRoot"] =
(workspaceRoot) =>
getActiveProjectRowByWorkspaceRoot({ workspaceRoot }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getActiveProjectByWorkspaceRoot:query",
"ProjectionSnapshotQuery.getActiveProjectByWorkspaceRoot:decodeRow",
),
),
Effect.flatMap((option) =>
Option.isNone(option)
? Effect.succeed(Option.none<OrchestrationProject>())
: repositoryIdentityResolver.resolve(option.value.workspaceRoot).pipe(
Effect.map((repositoryIdentity) =>
Option.some({
id: option.value.projectId,
title: option.value.title,
workspaceRoot: option.value.workspaceRoot,
repositoryIdentity,
defaultModelSelection: option.value.defaultModelSelection,
defaultThreadEnvMode: option.value.defaultThreadEnvMode,
autoPull: option.value.autoPull === 1,
faviconPath: option.value.faviconPath ?? null,
projectIcon: option.value.projectIcon ?? null,
scripts: option.value.scripts,
createdAt: option.value.createdAt,
updatedAt: option.value.updatedAt,
deletedAt: option.value.deletedAt,
} satisfies OrchestrationProject),
),
),
),
);
const getProjectShellById: ProjectionSnapshotQueryShape["getProjectShellById"] = (projectId) =>
getActiveProjectRowById({ projectId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getProjectShellById:query",
"ProjectionSnapshotQuery.getProjectShellById:decodeRow",
),
),
Effect.flatMap((option) =>
Option.isNone(option)
? Effect.succeed(Option.none<OrchestrationProjectShell>())
: repositoryIdentityResolver
.resolve(option.value.workspaceRoot)
.pipe(
Effect.map((repositoryIdentity) =>
Option.some(mapProjectShellRow(option.value, repositoryIdentity)),
),
),
),
);
const getFirstActiveThreadIdByProjectId: ProjectionSnapshotQueryShape["getFirstActiveThreadIdByProjectId"] =
(projectId) =>
getFirstActiveThreadIdByProject({ projectId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getFirstActiveThreadIdByProjectId:query",
"ProjectionSnapshotQuery.getFirstActiveThreadIdByProjectId:decodeRow",
),
),
Effect.map(Option.map((row) => row.threadId)),
);
const getThreadCheckpointContext: ProjectionSnapshotQueryShape["getThreadCheckpointContext"] = (
threadId,
) =>
Effect.gen(function* () {
const threadRow = yield* getThreadCheckpointContextThreadRow({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadCheckpointContext:getThread:query",
"ProjectionSnapshotQuery.getThreadCheckpointContext:getThread:decodeRow",
),
),
);
if (Option.isNone(threadRow)) {
return Option.none<ProjectionThreadCheckpointContext>();
}
const checkpointRows = yield* listCheckpointRowsByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:query",
"ProjectionSnapshotQuery.getThreadCheckpointContext:listCheckpoints:decodeRows",
),
),
);
return Option.some({
threadId: threadRow.value.threadId,
projectId: threadRow.value.projectId,
workspaceRoot: threadRow.value.workspaceRoot,
worktreePath: threadRow.value.worktreePath,
checkpoints: checkpointRows.map((row): OrchestrationCheckpointSummary => ({
turnId: row.turnId,
checkpointTurnCount: row.checkpointTurnCount,
checkpointRef: row.checkpointRef,
status: row.status,
files: row.files,
assistantMessageId: row.assistantMessageId,
completedAt: row.completedAt,
})),
});
});
const getFullThreadDiffContext: NonNullable<
ProjectionSnapshotQueryShape["getFullThreadDiffContext"]
> = (threadId, toTurnCount) =>
Effect.gen(function* () {
const row = yield* getFullThreadDiffContextRow({
threadId,
checkpointTurnCount: toTurnCount,
}).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getFullThreadDiffContext:query",
"ProjectionSnapshotQuery.getFullThreadDiffContext:decodeRow",
),
),
);
if (Option.isNone(row)) {
return Option.none<ProjectionFullThreadDiffContext>();
}
return Option.some({
threadId: row.value.threadId,
projectId: row.value.projectId,
workspaceRoot: row.value.workspaceRoot,
worktreePath: row.value.worktreePath,
latestCheckpointTurnCount: row.value.latestCheckpointTurnCount ?? 0,
toCheckpointRef: row.value.toCheckpointRef,
});
});
const getThreadShellById: ProjectionSnapshotQueryShape["getThreadShellById"] = (threadId) =>
Effect.gen(function* () {
const [threadRow, latestTurnRow, sessionRow] = yield* Effect.all([
getActiveThreadRowById({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadShellById:getThread:query",
"ProjectionSnapshotQuery.getThreadShellById:getThread:decodeRow",
),
),
),
getLatestTurnRowByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadShellById:getLatestTurn:query",
"ProjectionSnapshotQuery.getThreadShellById:getLatestTurn:decodeRow",
),
),
),
getThreadSessionRowByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadShellById:getSession:query",
"ProjectionSnapshotQuery.getThreadShellById:getSession:decodeRow",
),
),
),
]);
if (Option.isNone(threadRow)) {
return Option.none<OrchestrationThreadShell>();
}
return Option.some({
id: threadRow.value.threadId,
projectId: threadRow.value.projectId,
title: threadRow.value.title,
modelSelection: threadRow.value.modelSelection,
runtimeMode: threadRow.value.runtimeMode,
interactionMode: threadRow.value.interactionMode,
branch: threadRow.value.branch,
worktreePath: threadRow.value.worktreePath,
...(threadRow.value.linkedPullRequest === null
? {}
: { linkedPullRequest: threadRow.value.linkedPullRequest }),
latestTurn: Option.isSome(latestTurnRow) ? mapLatestTurn(latestTurnRow.value) : null,
createdAt: threadRow.value.createdAt,
updatedAt: threadRow.value.updatedAt,
archivedAt: threadRow.value.archivedAt,
settledOverride: threadRow.value.settledOverride,
settledAt: threadRow.value.settledAt,
unsettledAt: threadRow.value.unsettledAt,
snoozedUntil: threadRow.value.snoozedUntil,
snoozedAt: threadRow.value.snoozedAt,
pinnedAt: threadRow.value.pinnedAt,
pinOrderKey: threadRow.value.pinOrderKey ?? null,
titleRegeneration: mapTitleRegeneration(threadRow.value),
session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null,
latestUserMessageAt: threadRow.value.latestUserMessageAt,
hasPendingApprovals: threadRow.value.pendingApprovalCount > 0,
hasPendingUserInput: threadRow.value.pendingUserInputCount > 0,
hasActionableProposedPlan: threadRow.value.hasActionableProposedPlan > 0,
backgroundLiveness: threadBackgroundLiveness.getThreadBackgroundLiveness(
threadRow.value.threadId,
),
planProgress: threadPlanProgress.getThreadPlanProgress(threadRow.value.threadId),
} satisfies OrchestrationThreadShell);
});
const getThreadRuntimeContext: ProjectionSnapshotQueryShape["getThreadRuntimeContext"] =
Effect.fn("ProjectionSnapshotQuery.getThreadRuntimeContext")(function* (threadId) {
const context = yield* getThreadRuntimeContextRow({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadRuntimeContext:query",
"ProjectionSnapshotQuery.getThreadRuntimeContext:decodeRow",
),
),
);
return Option.map(context, (row) => ({
id: row.id,
title: row.title,
session: row.session === null ? null : mapSessionRow(row.session),
}));
});
// Contiguous turn range bounding a windowed detail read; undefined loads the
// full thread. Resolved from a window request inside the snapshot
// transaction (see getThreadDetailSnapshot).
interface ThreadDetailBounds {
readonly minAnchorAt: string;
readonly minTurnKey: string;
readonly beforeAnchorAt: string;
readonly beforeTurnKey: string;
}
type ThreadDetailActivityRead =
| {
readonly mode: "raw";
readonly query?: ProjectionThreadDetailQuery;
}
| {
readonly mode: "client";
};
const listProjectedThreadActivities = Effect.fn(
"ProjectionSnapshotQuery.listProjectedThreadActivities",
)(function* (threadId: ThreadId, bounds: ThreadDetailBounds | undefined) {
const [activityIdRows, pinnedActivityIdRows] = yield* Effect.all([
(bounds === undefined
? listThreadActivityIdsByThread({ threadId })
: listThreadActivityIdsByThreadWindow({ threadId, ...bounds })
).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listActivityIds:query",
"ProjectionSnapshotQuery.getThreadDetailById:listActivityIds:decodeRows",
),
),
),
listPinnedThreadActivityIdsByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivityIds:query",
"ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivityIds:decodeRows",
),
),
),
]);
const activityIds = [
...new Set([...activityIdRows, ...pinnedActivityIdRows].map(({ activityId }) => activityId)),
];
const activities: OrchestrationThreadActivity[] = [];
for (
let offset = 0;
offset < activityIds.length;
offset += THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE
) {
const batchIds = activityIds.slice(
offset,
offset + THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE,
);
const batchRows = yield* listThreadActivityRowsByIds({ activityIds: batchIds }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listActivityPayloadBatch:query",
"ProjectionSnapshotQuery.getThreadDetailById:listActivityPayloadBatch:decodeRows",
),
),
);
for (const row of batchRows) {
activities.push(projectActivityPayload(mapThreadActivityRow(row)));
}
}
return activities.toSorted(
(left, right) =>
(left.sequence ?? -1) - (right.sequence ?? -1) ||
left.createdAt.localeCompare(right.createdAt) ||
left.id.localeCompare(right.id),
);
});
const getThreadDetailByIdBounded = (
threadId: ThreadId,
bounds: ThreadDetailBounds | undefined,
activityRead: ThreadDetailActivityRead = { mode: "raw" },
) =>
Effect.gen(function* () {
const activitiesEffect =
activityRead.mode === "client"
? listProjectedThreadActivities(threadId, bounds)
: Effect.all([
(activityRead.query?.activityKinds === undefined
? bounds === undefined
? listThreadActivityRowsByThread({ threadId })
: listThreadActivityRowsByThreadWindow({ threadId, ...bounds })
: activityRead.query.activityKinds.length === 0
? Effect.succeed([])
: listThreadActivityRowsByThreadAndKinds({
threadId,
activityKinds: activityRead.query.activityKinds,
})
).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
"ProjectionSnapshotQuery.getThreadDetailById:listActivities:decodeRows",
),
),
),
activityRead.query?.activityKinds === undefined
? listPinnedThreadActivityRowsByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:query",
"ProjectionSnapshotQuery.getThreadDetailById:listPinnedActivities:decodeRows",
),
),
)
: Effect.succeed([]),
]).pipe(
Effect.map(([activityRows, pinnedActivityRows]) =>
[
...new Map(
[...activityRows, ...pinnedActivityRows].map(
(row) => [row.activityId, row] as const,
),
).values(),
]
.toSorted(
(left, right) =>
(left.sequence ?? -1) - (right.sequence ?? -1) ||
left.createdAt.localeCompare(right.createdAt) ||
left.activityId.localeCompare(right.activityId),
)
.map(mapThreadActivityRow),
),
);
const [
threadRow,
messageRows,
proposedPlanRows,
activities,
checkpointRows,
latestTurnRow,
sessionRow,
] = yield* Effect.all([
getActiveThreadRowById({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:getThread:query",
"ProjectionSnapshotQuery.getThreadDetailById:getThread:decodeRow",
),
),
),
(bounds === undefined
? listThreadMessageRowsByThread({ threadId })
: listThreadMessageRowsByThreadWindow({ threadId, ...bounds })
).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listMessages:query",
"ProjectionSnapshotQuery.getThreadDetailById:listMessages:decodeRows",
),
),
),
listThreadProposedPlanRowsByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listPlans:query",
"ProjectionSnapshotQuery.getThreadDetailById:listPlans:decodeRows",
),
),
),
activitiesEffect,
listCheckpointRowsByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:listCheckpoints:query",
"ProjectionSnapshotQuery.getThreadDetailById:listCheckpoints:decodeRows",
),
),
),
getLatestTurnRowByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:getLatestTurn:query",
"ProjectionSnapshotQuery.getThreadDetailById:getLatestTurn:decodeRow",
),
),
),
getThreadSessionRowByThread({ threadId }).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailById:getSession:query",
"ProjectionSnapshotQuery.getThreadDetailById:getSession:decodeRow",
),
),
),
]);
if (Option.isNone(threadRow)) {
return Option.none<OrchestrationThread>();
}
const thread = {
id: threadRow.value.threadId,
projectId: threadRow.value.projectId,
title: threadRow.value.title,
modelSelection: threadRow.value.modelSelection,
runtimeMode: threadRow.value.runtimeMode,
interactionMode: threadRow.value.interactionMode,
branch: threadRow.value.branch,
worktreePath: threadRow.value.worktreePath,
...(threadRow.value.linkedPullRequest === null
? {}
: { linkedPullRequest: threadRow.value.linkedPullRequest }),
latestTurn: Option.isSome(latestTurnRow) ? mapLatestTurn(latestTurnRow.value) : null,
createdAt: threadRow.value.createdAt,
updatedAt: threadRow.value.updatedAt,
archivedAt: threadRow.value.archivedAt,
settledOverride: threadRow.value.settledOverride,
settledAt: threadRow.value.settledAt,
unsettledAt: threadRow.value.unsettledAt,
snoozedUntil: threadRow.value.snoozedUntil,
snoozedAt: threadRow.value.snoozedAt,
pinnedAt: threadRow.value.pinnedAt,
pinOrderKey: threadRow.value.pinOrderKey ?? null,
titleRegeneration: mapTitleRegeneration(threadRow.value),
deletedAt: null,
messages: messageRows.map((row) => {
const message = {
id: row.messageId,
role: row.role,
text: row.text,
turnId: row.turnId,
streaming: row.isStreaming === 1,
createdAt: row.createdAt,
updatedAt: row.updatedAt,
};
if (row.attachments !== null) {
return Object.assign(message, { attachments: row.attachments });
}
return message;
}),
proposedPlans: proposedPlanRows.map(mapProposedPlanRow),
activities,
checkpoints: checkpointRows.map((row) => ({
turnId: row.turnId,
checkpointTurnCount: row.checkpointTurnCount,
checkpointRef: row.checkpointRef,
status: row.status,
files: row.files,
assistantMessageId: row.assistantMessageId,
completedAt: row.completedAt,
})),
session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null,
};
return Option.some(
yield* decodeThread(thread).pipe(
Effect.mapError(
toPersistenceDecodeError("ProjectionSnapshotQuery.getThreadDetailById:decodeThread"),
),
),
);
});
const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (
threadId,
query,
) =>
getThreadDetailByIdBounded(threadId, undefined, {
mode: "raw",
...(query === undefined ? {} : { query }),
});
// Bounds pathological fan-out: one user turn that spawned hundreds of
// subagent turns still pages in bounded chunks, at the cost of splitting the
// fan-out group across pages (the cursor continues the same group). Also
// structurally bounds the window scan via the candidates CTE's LIMIT.
const THREAD_DETAIL_MAX_RAW_TURNS_PER_PAGE = 150;
// Sentinels for unbounded keyset ends; "~" sorts after any ISO timestamp.
const ANCHOR_UNBOUNDED = "~";
const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = (
threadId,
window,
) =>
// Read the thread detail and the snapshot sequence within a single
// transaction so the sequence is consistent with the returned state; a
// projector update landing between two separate reads could otherwise return
// a sequence ahead of the thread detail, causing the client to resume from
// too far and drop events. Window resolution runs inside the same
// transaction so the page boundary is consistent with the returned rows.
sql
.withTransaction(
Effect.gen(function* () {
if (window?.turnLimit === undefined) {
const thread = yield* getThreadDetailByIdBounded(threadId, undefined, {
mode: "client",
});
if (Option.isNone(thread)) {
return Option.none<OrchestrationThreadDetailSnapshot>();
}
const { snapshotSequence } = yield* getSnapshotSequence();
return Option.some({ snapshotSequence, thread: thread.value });
}
// A malformed or foreign-thread cursor falls back to the first page
// rather than failing: the client's stale cursor after a revert or
// reconnect should degrade to "reload recent history", not error.
const decodedCursor =
window.beforeCursor === undefined
? null
: decodeThreadDetailPageCursor(window.beforeCursor);
const cursor = decodedCursor?.threadId === threadId ? decodedCursor : null;
const windowRows = yield* listTurnWindowRows({
threadId,
beforeAnchorAt: cursor?.beforeAnchorAt ?? ANCHOR_UNBOUNDED,
beforeTurnKey: cursor?.beforeTurnId ?? "",
userTurnLimit: window.turnLimit,
maxRawTurns: THREAD_DETAIL_MAX_RAW_TURNS_PER_PAGE,
}).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailSnapshot:listTurnWindow:query",
"ProjectionSnapshotQuery.getThreadDetailSnapshot:listTurnWindow:decodeRows",
),
),
);
const oldest = windowRows[0];
// An empty window (no turns before the cursor, or a thread with no
// turns at all) still returns thread metadata with empty collections
// for turn-linked rows; turnless rows are bounded to the same empty
// range. The first page of a turnless thread stays unwindowed so
// pre-turn content (e.g. a just-created thread) is not hidden.
const bounds: ThreadDetailBounds | undefined =
oldest === undefined && cursor === null
? undefined
: {
minAnchorAt: oldest?.anchorAt ?? "",
minTurnKey: oldest?.turnKey ?? "",
beforeAnchorAt: cursor?.beforeAnchorAt ?? ANCHOR_UNBOUNDED,
beforeTurnKey: cursor?.beforeTurnId ?? "",
};
// Empty window behind a cursor: nothing older remains.
const emptyBounds =
oldest === undefined && cursor !== null
? { minAnchorAt: "", minTurnKey: "", beforeAnchorAt: "", beforeTurnKey: "" }
: undefined;
const thread = yield* getThreadDetailByIdBounded(threadId, emptyBounds ?? bounds, {
mode: "client",
});
if (Option.isNone(thread)) {
return Option.none<OrchestrationThreadDetailSnapshot>();
}
const hasMore =
oldest !== undefined &&
(yield* listTurnWindowRows({
threadId,
beforeAnchorAt: oldest.anchorAt,
beforeTurnKey: oldest.turnKey,
userTurnLimit: 1,
maxRawTurns: 1,
}).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailSnapshot:probeOlder:query",
"ProjectionSnapshotQuery.getThreadDetailSnapshot:probeOlder:decodeRows",
),
),
)).length > 0;
const { snapshotSequence } = yield* getSnapshotSequence();
const watermarkRow = yield* getThreadEventWatermarkRow({
threadId,
maxSequence: snapshotSequence,
}).pipe(
Effect.mapError(
toPersistenceSqlOrDecodeError(
"ProjectionSnapshotQuery.getThreadDetailSnapshot:threadWatermark:query",
"ProjectionSnapshotQuery.getThreadDetailSnapshot:threadWatermark:decodeRow",
),
),
);
const threadSequence = Option.match(watermarkRow, {
onNone: () => 0,
onSome: (row) => row.threadSequence ?? 0,
});
return Option.some({
snapshotSequence,
thread: thread.value,
page: {
beforeCursor:
hasMore && oldest !== undefined
? encodeThreadDetailPageCursor({
threadId,
beforeAnchorAt: oldest.anchorAt,
beforeTurnId: oldest.turnKey,
})
: null,
hasMore,
snapshotSequence,
threadSequence,
},
});
}),
)
.pipe(
Effect.mapError((error) =>
isPersistenceError(error)
? error
: toPersistenceSqlError("ProjectionSnapshotQuery.getThreadDetailSnapshot:transaction")(
error,
),
),
);
return {
getCommandReadModel,
getUserInputActivity,
getSnapshot,
getShellSnapshot,
getArchivedShellSnapshot,
searchThreads,
getSnapshotSequence,
getCounts,
getEventReplayStats,
getActiveProjectByWorkspaceRoot,
getProjectShellById,
getFirstActiveThreadIdByProjectId,
getThreadCheckpointContext,
getFullThreadDiffContext,
getThreadShellById,
getThreadRuntimeContext,
getThreadDetailById,
getThreadDetailSnapshot,
} satisfies ProjectionSnapshotQueryShape;
});
export const OrchestrationProjectionSnapshotQueryLive = Layer.effect(
ProjectionSnapshotQuery,
makeProjectionSnapshotQuery,
);
