import fs from "node:fs"; import path from "node:path"; import { ApprovalRequestId, CommandId, EventId, MessageId, ProjectId, ProviderSessionId, ProviderThreadId, ProviderTurnId, ThreadId, } from "@t3tools/contracts"; import { assert, it } from "@effect/vitest"; import { Effect, Exit, Option } from "effect"; import type { TestTurnResponse } from "./TestProviderAdapter.integration.ts"; import { gitRefExists, gitShowFileAtRef, makeOrchestrationIntegrationHarness, type OrchestrationIntegrationHarness, } from "./OrchestrationEngineHarness.integration.ts"; import { checkpointRefForThreadTurn } from "../src/checkpointing/Utils.ts"; const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asProviderSessionId = (value: string): ProviderSessionId => ProviderSessionId.makeUnsafe(value); const asProviderThreadId = (value: string): ProviderThreadId => ProviderThreadId.makeUnsafe(value); const asProviderTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asApprovalRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); const PROJECT_ID = asProjectId("project-1"); const THREAD_ID = ThreadId.makeUnsafe("thread-1"); const FIXTURE_SESSION_ID = asProviderSessionId("fixture-session"); const FIXTURE_THREAD_ID = asProviderThreadId("fixture-thread"); const FIXTURE_TURN_ID = asProviderTurnId("fixture-turn"); const APPROVAL_REQUEST_ID = asApprovalRequestId("req-approval-1"); function nowIso() { return new Date().toISOString(); } const sleep = (ms: number) => Effect.promise(() => new Promise((resolve) => setTimeout(resolve, ms))); function waitForSync( read: () => A, predicate: (value: A) => boolean, description: string, timeoutMs = 3000, ): Effect.Effect { return Effect.gen(function* () { const deadline = Date.now() + timeoutMs; while (true) { const value = read(); if (predicate(value)) { return value; } if (Date.now() >= deadline) { return yield* Effect.die(new Error(`Timed out waiting for ${description}.`)); } yield* sleep(10); } }); } function runtimeBase(eventId: string, createdAt: string) { return { eventId: asEventId(eventId), provider: "codex" as const, sessionId: FIXTURE_SESSION_ID, createdAt, }; } function withHarness(use: (harness: OrchestrationIntegrationHarness) => Effect.Effect) { return Effect.acquireUseRelease( makeOrchestrationIntegrationHarness, use, (harness) => harness.dispose, ); } const seedProjectAndThread = (harness: OrchestrationIntegrationHarness) => Effect.gen(function* () { const createdAt = nowIso(); yield* harness.engine.dispatch({ type: "project.create", commandId: CommandId.makeUnsafe("cmd-project-create"), projectId: PROJECT_ID, title: "Integration Project", workspaceRoot: harness.workspaceDir, defaultModel: "gpt-5-codex", createdAt, }); yield* harness.engine.dispatch({ type: "thread.create", commandId: CommandId.makeUnsafe("cmd-thread-create"), threadId: THREAD_ID, projectId: PROJECT_ID, title: "Integration Thread", model: "gpt-5-codex", branch: null, worktreePath: harness.workspaceDir, createdAt, }); }); const startTurn = (input: { readonly harness: OrchestrationIntegrationHarness; readonly commandId: string; readonly messageId: string; readonly text: string; }) => input.harness.engine.dispatch({ type: "thread.turn.start", commandId: CommandId.makeUnsafe(input.commandId), threadId: THREAD_ID, message: { messageId: asMessageId(input.messageId), role: "user", text: input.text, attachments: [], }, createdAt: nowIso(), }); it.live("always removes integration temp dirs when shutdown fails", () => Effect.gen(function* () { const harness = yield* makeOrchestrationIntegrationHarness; const rootDir = harness.rootDir; const originalStopAll = harness.providerService.stopAll.bind(harness.providerService); ( harness.providerService as unknown as { stopAll: typeof harness.providerService.stopAll; } ).stopAll = () => originalStopAll().pipe(Effect.andThen(Effect.fail(new Error("boom")))); const disposeExit = yield* Effect.exit(harness.dispose); assert.equal(Exit.isFailure(disposeExit), true); assert.equal(fs.existsSync(rootDir), false); }), ); it.live("runs a single turn end-to-end and persists checkpoint state in sqlite + git", () => withHarness((harness) => Effect.gen(function* () { yield* seedProjectAndThread(harness); const turnResponse: TestTurnResponse = { events: [ { type: "turn.started", ...runtimeBase("evt-single-1", "2026-02-24T10:00:00.000Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, { type: "message.delta", ...runtimeBase("evt-single-2", "2026-02-24T10:00:00.100Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, delta: "Single turn response.\n", }, { type: "turn.completed", ...runtimeBase("evt-single-3", "2026-02-24T10:00:00.200Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, status: "completed", }, ], }; yield* harness.adapterHarness.queueTurnResponseForNextSession(turnResponse); yield* startTurn({ harness, commandId: "cmd-turn-start-single", messageId: "msg-user-single", text: "Say hello", }); const thread = yield* harness.waitForThread( THREAD_ID, (entry) => entry.session?.status === "ready" && entry.messages.some( (message) => message.role === "assistant" && message.streaming === false, ) && entry.checkpoints.length === 1, ); assert.equal(thread.checkpoints[0]?.status, "ready"); assert.equal(thread.checkpoints[0]?.checkpointTurnCount, 1); const checkpointRows = yield* harness.checkpointRepository.listByThreadId({ threadId: THREAD_ID, }); assert.equal(checkpointRows.length, 1); assert.equal(checkpointRows[0]?.checkpointTurnCount, 1); assert.equal(checkpointRows[0]?.status, "ready"); assert.deepEqual(checkpointRows[0]?.files, []); yield* harness.waitForDomainEvent((event) => event.type === "thread.turn-diff-completed"); const ref0 = checkpointRefForThreadTurn(THREAD_ID, 0); const ref1 = checkpointRefForThreadTurn(THREAD_ID, 1); assert.equal(gitRefExists(harness.workspaceDir, ref0), true); assert.equal(gitRefExists(harness.workspaceDir, ref1), true); assert.equal(gitShowFileAtRef(harness.workspaceDir, ref0, "README.md"), "v1\n"); assert.equal(gitShowFileAtRef(harness.workspaceDir, ref1, "README.md"), "v1\n"); }), ), ); it.live("runs multi-turn file edits and persists checkpoint diffs", () => withHarness((harness) => Effect.gen(function* () { yield* seedProjectAndThread(harness); yield* harness.adapterHarness.queueTurnResponseForNextSession({ events: [ { type: "turn.started", ...runtimeBase("evt-multi-1", "2026-02-24T10:01:00.000Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, { type: "tool.started", ...runtimeBase("evt-multi-2", "2026-02-24T10:01:00.100Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, toolKind: "command", title: "Edit file", detail: "README.md", }, { type: "tool.completed", ...runtimeBase("evt-multi-3", "2026-02-24T10:01:00.200Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, toolKind: "command", title: "Edit file", detail: "README.md", }, { type: "message.delta", ...runtimeBase("evt-multi-4", "2026-02-24T10:01:00.300Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, delta: "Updated README to v2.\n", }, { type: "turn.completed", ...runtimeBase("evt-multi-5", "2026-02-24T10:01:00.400Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, status: "completed", }, ], mutateWorkspace: ({ cwd }) => Effect.sync(() => { fs.writeFileSync(path.join(cwd, "README.md"), "v2\n", "utf8"); }), }); yield* startTurn({ harness, commandId: "cmd-turn-start-multi-1", messageId: "msg-user-multi-1", text: "Make first edit", }); const firstTurnThread = yield* harness.waitForThread( THREAD_ID, (entry) => entry.checkpoints.length === 1 && entry.session?.providerSessionId !== null, ); const sessionId = firstTurnThread.session?.providerSessionId; assert.equal(sessionId !== null, true); if (!sessionId) { return; } yield* harness.adapterHarness.queueTurnResponse(sessionId, { events: [ { type: "turn.started", ...runtimeBase("evt-multi-6", "2026-02-24T10:02:00.000Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, { type: "message.delta", ...runtimeBase("evt-multi-7", "2026-02-24T10:02:00.100Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, delta: "Updated README to v3.\n", }, { type: "turn.completed", ...runtimeBase("evt-multi-8", "2026-02-24T10:02:00.200Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, status: "completed", }, ], mutateWorkspace: ({ cwd }) => Effect.sync(() => { fs.writeFileSync(path.join(cwd, "README.md"), "v3\n", "utf8"); }), }); yield* startTurn({ harness, commandId: "cmd-turn-start-multi-2", messageId: "msg-user-multi-2", text: "Make second edit", }); const secondTurnThread = yield* harness.waitForThread( THREAD_ID, (entry) => entry.latestTurnId === "turn-2" && entry.checkpoints.length === 2 && entry.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 2), ); const secondCheckpoint = secondTurnThread.checkpoints.find( (checkpoint) => checkpoint.checkpointTurnCount === 2, ); assert.equal( secondCheckpoint?.files.some((file) => file.path === "README.md"), true, ); const checkpointRows = yield* harness.checkpointRepository.listByThreadId({ threadId: THREAD_ID, }); assert.deepEqual( checkpointRows.map((row) => row.checkpointTurnCount), [1, 2], ); const incrementalDiff = yield* harness.checkpointStore.diffCheckpoints({ cwd: harness.workspaceDir, fromCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 1), toCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 2), fallbackFromToHead: false, }); assert.equal(incrementalDiff.includes("README.md"), true); const fullDiff = yield* harness.checkpointStore.diffCheckpoints({ cwd: harness.workspaceDir, fromCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 0), toCheckpointRef: checkpointRefForThreadTurn(THREAD_ID, 2), fallbackFromToHead: false, }); assert.equal(fullDiff.includes("README.md"), true); assert.equal( gitShowFileAtRef( harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 1), "README.md", ), "v2\n", ); assert.equal( gitShowFileAtRef( harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 2), "README.md", ), "v3\n", ); }), ), ); it.live("tracks approval requests and resolves pending approvals on user response", () => withHarness((harness) => Effect.gen(function* () { yield* seedProjectAndThread(harness); yield* harness.adapterHarness.queueTurnResponseForNextSession({ events: [ { type: "turn.started", ...runtimeBase("evt-approval-1", "2026-02-24T10:03:00.000Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, { type: "approval.requested", ...runtimeBase("evt-approval-2", "2026-02-24T10:03:00.100Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, requestId: APPROVAL_REQUEST_ID, requestKind: "command", detail: "Approve command execution", }, { type: "turn.completed", ...runtimeBase("evt-approval-3", "2026-02-24T10:03:00.200Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, status: "completed", }, ], }); yield* startTurn({ harness, commandId: "cmd-turn-start-approval", messageId: "msg-user-approval", text: "Run command needing approval", }); const thread = yield* harness.waitForThread(THREAD_ID, (entry) => entry.activities.some((activity) => activity.kind === "approval.requested"), ); assert.equal( thread.activities.some((activity) => activity.kind === "approval.requested"), true, ); const pendingRow = yield* harness.waitForPendingApproval( "req-approval-1", (row) => row.status === "pending" && row.decision === null, ); assert.equal(pendingRow.status, "pending"); yield* harness.engine.dispatch({ type: "thread.approval.respond", commandId: CommandId.makeUnsafe("cmd-approval-respond"), threadId: THREAD_ID, requestId: APPROVAL_REQUEST_ID, decision: "accept", createdAt: nowIso(), }); const resolvedRow = yield* harness.waitForPendingApproval( "req-approval-1", (row) => row.status === "resolved" && row.decision === "accept", ); assert.equal(resolvedRow.status, "resolved"); assert.equal(resolvedRow.decision, "accept"); const providerSessionId = thread.session?.providerSessionId ?? "test-session-1"; const approvalResponses = yield* waitForSync( () => harness.adapterHarness.getApprovalResponses(providerSessionId), (responses) => responses.length === 1, "provider approval response", ); assert.equal(approvalResponses.length, 1); assert.equal(approvalResponses[0]?.requestId, "req-approval-1"); assert.equal(approvalResponses[0]?.decision, "accept"); }), ), ); it.live("records failed turn runtime state and checkpoint status as error", () => withHarness((harness) => Effect.gen(function* () { yield* seedProjectAndThread(harness); yield* harness.adapterHarness.queueTurnResponseForNextSession({ events: [ { type: "turn.started", ...runtimeBase("evt-failure-1", "2026-02-24T10:04:00.000Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, { type: "message.delta", ...runtimeBase("evt-failure-2", "2026-02-24T10:04:00.100Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, delta: "Partial output before failure.\n", }, { type: "runtime.error", ...runtimeBase("evt-failure-3", "2026-02-24T10:04:00.200Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, message: "Sandbox command failed.", }, { type: "turn.completed", ...runtimeBase("evt-failure-4", "2026-02-24T10:04:00.300Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, status: "failed", errorMessage: "Sandbox command failed.", }, ], }); yield* startTurn({ harness, commandId: "cmd-turn-start-failure", messageId: "msg-user-failure", text: "Run risky command", }); const thread = yield* harness.waitForThread( THREAD_ID, (entry) => entry.session?.status === "error" && entry.session?.lastError === "Sandbox command failed." && entry.activities.some((activity) => activity.kind === "runtime.error") && entry.checkpoints.length === 1, ); assert.equal(thread.session?.status, "error"); assert.equal(thread.checkpoints[0]?.status, "error"); const checkpointRow = yield* harness.checkpointRepository.getByThreadAndTurnCount({ threadId: THREAD_ID, checkpointTurnCount: 1, }); assert.equal(Option.isSome(checkpointRow), true); if (Option.isSome(checkpointRow)) { assert.equal(checkpointRow.value.status, "error"); } assert.equal( gitRefExists(harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 1)), true, ); }), ), ); it.live("reverts to an earlier checkpoint and trims checkpoint projections + git refs", () => withHarness((harness) => Effect.gen(function* () { yield* seedProjectAndThread(harness); yield* harness.adapterHarness.queueTurnResponseForNextSession({ events: [ { type: "turn.started", ...runtimeBase("evt-revert-1", "2026-02-24T10:05:00.000Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, { type: "tool.started", ...runtimeBase("evt-revert-1-tool-started", "2026-02-24T10:05:00.025Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, toolKind: "command", title: "Edit file", detail: "README.md", }, { type: "tool.completed", ...runtimeBase("evt-revert-1-tool-completed", "2026-02-24T10:05:00.035Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, toolKind: "command", title: "Edit file", detail: "README.md", }, { type: "message.delta", ...runtimeBase("evt-revert-1a", "2026-02-24T10:05:00.050Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, delta: "Updated README to v2.\n", }, { type: "turn.completed", ...runtimeBase("evt-revert-2", "2026-02-24T10:05:00.100Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, status: "completed", }, ], mutateWorkspace: ({ cwd }) => Effect.sync(() => { fs.writeFileSync(path.join(cwd, "README.md"), "v2\n", "utf8"); }), }); yield* startTurn({ harness, commandId: "cmd-turn-start-revert-1", messageId: "msg-user-revert-1", text: "First edit", }); const firstTurnThread = yield* harness.waitForThread( THREAD_ID, (entry) => entry.session?.providerSessionId !== null && entry.checkpoints.length === 1, ); const sessionId = firstTurnThread.session?.providerSessionId; assert.equal(sessionId !== null, true); if (!sessionId) { return; } yield* harness.adapterHarness.queueTurnResponse(sessionId, { events: [ { type: "turn.started", ...runtimeBase("evt-revert-3", "2026-02-24T10:05:01.000Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, }, { type: "tool.started", ...runtimeBase("evt-revert-3-tool-started", "2026-02-24T10:05:01.025Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, toolKind: "command", title: "Edit file", detail: "README.md", }, { type: "tool.completed", ...runtimeBase("evt-revert-3-tool-completed", "2026-02-24T10:05:01.035Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, toolKind: "command", title: "Edit file", detail: "README.md", }, { type: "message.delta", ...runtimeBase("evt-revert-3a", "2026-02-24T10:05:01.050Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, delta: "Updated README to v3.\n", }, { type: "turn.completed", ...runtimeBase("evt-revert-4", "2026-02-24T10:05:01.100Z"), threadId: FIXTURE_THREAD_ID, turnId: FIXTURE_TURN_ID, status: "completed", }, ], mutateWorkspace: ({ cwd }) => Effect.sync(() => { fs.writeFileSync(path.join(cwd, "README.md"), "v3\n", "utf8"); }), }); yield* startTurn({ harness, commandId: "cmd-turn-start-revert-2", messageId: "msg-user-revert-2", text: "Second edit", }); yield* harness.waitForThread( THREAD_ID, (entry) => entry.latestTurnId === "turn-2" && entry.checkpoints.length === 2 && entry.activities.some((activity) => activity.turnId === "turn-2"), 8000, ); yield* harness.engine.dispatch({ type: "thread.checkpoint.revert", commandId: CommandId.makeUnsafe("cmd-checkpoint-revert"), threadId: THREAD_ID, turnCount: 1, createdAt: nowIso(), }); yield* harness.waitForDomainEvent((event) => event.type === "thread.reverted"); const revertedThread = yield* harness.waitForThread( THREAD_ID, (entry) => entry.checkpoints.length === 1 && entry.checkpoints[0]?.checkpointTurnCount === 1, ); assert.equal(revertedThread.checkpoints[0]?.checkpointTurnCount, 1); assert.deepEqual( revertedThread.messages.map((message) => ({ role: message.role, text: message.text })), [ { role: "user", text: "First edit" }, { role: "assistant", text: "Updated README to v2.\n" }, ], ); assert.equal( revertedThread.activities.some((activity) => activity.turnId === "turn-2"), false, ); assert.equal( revertedThread.activities.some( (activity) => activity.turnId === "turn-1" && activity.kind === "tool.started", ), true, ); assert.equal( revertedThread.activities.some( (activity) => activity.turnId === "turn-1" && activity.kind === "tool.completed", ), true, ); assert.equal(fs.readFileSync(path.join(harness.workspaceDir, "README.md"), "utf8"), "v2\n"); assert.equal( gitRefExists(harness.workspaceDir, checkpointRefForThreadTurn(THREAD_ID, 2)), false, ); assert.deepEqual(harness.adapterHarness.getRollbackCalls(sessionId), [1]); const checkpointRows = yield* harness.checkpointRepository.listByThreadId({ threadId: THREAD_ID, }); assert.equal(checkpointRows.length, 1); }), ), ); it.live( "appends checkpoint.revert.failed activity when revert is requested without an active session", () => withHarness((harness) => Effect.gen(function* () { yield* seedProjectAndThread(harness); yield* harness.engine.dispatch({ type: "thread.checkpoint.revert", commandId: CommandId.makeUnsafe("cmd-checkpoint-revert-no-session"), threadId: THREAD_ID, turnCount: 0, createdAt: nowIso(), }); const thread = yield* harness.waitForThread(THREAD_ID, (entry) => entry.activities.some( (activity) => activity.kind === "checkpoint.revert.failed" && typeof activity.payload === "object" && activity.payload !== null, ), ); const failureActivity = thread.activities.find( (activity) => activity.kind === "checkpoint.revert.failed", ); assert.equal(failureActivity !== undefined, true); assert.equal( String( (failureActivity?.payload as { readonly detail?: string } | undefined)?.detail, ).includes("No active provider session"), true, ); }), ), );