import { ChatAttachment, IsoDateTime, MessageId, NonNegativeInt, OrchestrationCheckpointFile, OrchestrationProposedPlanId, OrchestrationReadModel, ProjectScript, ThreadId, TurnId, type OrchestrationCheckpointSummary, type OrchestrationLatestTurn, type OrchestrationMessage, type OrchestrationProposedPlan, type OrchestrationProject, type OrchestrationSession, type OrchestrationThread, type OrchestrationThreadActivity,} from "@t3tools/contracts";import { Effect, Layer, Schema, Struct } from "effect";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 { 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 { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";import { ProjectionSnapshotQuery, type ProjectionSnapshotQueryShape,} from "../Services/ProjectionSnapshotQuery.ts"; const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel);const ProjectionProjectDbRowSchema = ProjectionProject.mapFields( Struct.assign({ 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;const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( Struct.assign({ payload: Schema.fromJsonString(Schema.Unknown), sequence: Schema.NullOr(NonNegativeInt), }),);const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession;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 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 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 toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { return (cause: unknown): ProjectionRepositoryError => Schema.isSchemaError(cause) ? toPersistenceDecodeError(decodeOperation)(cause) : toPersistenceSqlError(sqlOperation)(cause);} const makeProjectionSnapshotQuery = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const listProjectRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionProjectDbRowSchema, execute: () => sql` SELECT project_id AS "projectId", title, workspace_root AS "workspaceRoot", default_model AS "defaultModel", 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, runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", deleted_at AS "deletedAt" FROM projection_threads ORDER BY created_at ASC, thread_id ASC `, }); 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, CASE WHEN sequence IS NULL THEN 0 ELSE 1 END 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_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 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 thread_id AS "threadId", turn_id AS "turnId", state, requested_at AS "requestedAt", started_at AS "startedAt", completed_at AS "completedAt", assistant_message_id AS "assistantMessageId", source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", source_proposed_plan_id AS "sourceProposedPlanId" FROM projection_turns WHERE turn_id IS NOT NULL ORDER BY thread_id ASC, requested_at DESC, turn_id DESC `, }); 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 getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () => sql .withTransaction( Effect.gen(function* () { const [ projectRows, threadRows, messageRows, proposedPlanRows, activityRows, sessionRows, checkpointRows, latestTurnRows, stateRows, ] = yield* 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", ), ), ), ]); 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"Lines 1–500 of 596. Load more
import { ChatAttachment, IsoDateTime, MessageId, NonNegativeInt, OrchestrationCheckpointFile, OrchestrationProposedPlanId, OrchestrationReadModel, ProjectScript, ThreadId, TurnId, type OrchestrationCheckpointSummary, type OrchestrationLatestTurn, type OrchestrationMessage, type OrchestrationProposedPlan, type OrchestrationProject, type OrchestrationSession, type OrchestrationThread, type OrchestrationThreadActivity, } from "@t3tools/contracts"; import { Effect, Layer, Schema, Struct } from "effect"; 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 { 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 { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts"; import { ProjectionSnapshotQuery, type ProjectionSnapshotQueryShape, } from "../Services/ProjectionSnapshotQuery.ts"; const decodeReadModel = Schema.decodeUnknownEffect(OrchestrationReadModel); const ProjectionProjectDbRowSchema = ProjectionProject.mapFields( Struct.assign({ 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; const ProjectionThreadActivityDbRowSchema = ProjectionThreadActivity.mapFields( Struct.assign({ payload: Schema.fromJsonString(Schema.Unknown), sequence: Schema.NullOr(NonNegativeInt), }), ); const ProjectionThreadSessionDbRowSchema = ProjectionThreadSession; 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 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 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 toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { return (cause: unknown): ProjectionRepositoryError => Schema.isSchemaError(cause) ? toPersistenceDecodeError(decodeOperation)(cause) : toPersistenceSqlError(sqlOperation)(cause); } const makeProjectionSnapshotQuery = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const listProjectRows = SqlSchema.findAll({ Request: Schema.Void, Result: ProjectionProjectDbRowSchema, execute: () => sql` SELECT project_id AS "projectId", title, workspace_root AS "workspaceRoot", default_model AS "defaultModel", 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, runtime_mode AS "runtimeMode", interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", deleted_at AS "deletedAt" FROM projection_threads ORDER BY created_at ASC, thread_id ASC `, }); 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, CASE WHEN sequence IS NULL THEN 0 ELSE 1 END 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_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 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 thread_id AS "threadId", turn_id AS "turnId", state, requested_at AS "requestedAt", started_at AS "startedAt", completed_at AS "completedAt", assistant_message_id AS "assistantMessageId", source_proposed_plan_thread_id AS "sourceProposedPlanThreadId", source_proposed_plan_id AS "sourceProposedPlanId" FROM projection_turns WHERE turn_id IS NOT NULL ORDER BY thread_id ASC, requested_at DESC, turn_id DESC `, }); 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 getSnapshot: ProjectionSnapshotQueryShape["getSnapshot"] = () => sql .withTransaction( Effect.gen(function* () { const [ projectRows, threadRows, messageRows, proposedPlanRows, activityRows, sessionRows, checkpointRows, latestTurnRows, stateRows, ] = yield* 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", ), ), ), ]); 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"