import { type CheckpointRef, OrchestrationGetTurnDiffResult, type ThreadId, type OrchestrationGetFullThreadDiffResult, type OrchestrationGetTurnDiffResult as OrchestrationGetTurnDiffResultType,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Schema from "effect/Schema"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";import { CheckpointInvariantError, CheckpointUnavailableError } from "../Errors.ts";import { checkpointRefForThreadTurn } from "../Utils.ts";import { CheckpointStore } from "../Services/CheckpointStore.ts";import { CheckpointDiffQuery, type CheckpointDiffQueryShape,} from "../Services/CheckpointDiffQuery.ts"; const isTurnDiffResult = Schema.is(OrchestrationGetTurnDiffResult); function buildTurnDiffResult( input: { readonly threadId: ThreadId; readonly fromTurnCount: number; readonly toTurnCount: number; }, diff: string,): OrchestrationGetTurnDiffResultType { return { threadId: input.threadId, fromTurnCount: input.fromTurnCount, toTurnCount: input.toTurnCount, diff, };} const make = Effect.gen(function* () { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; const checkpointStore = yield* CheckpointStore; const getTurnDiff: CheckpointDiffQueryShape["getTurnDiff"] = Effect.fn("getTurnDiff")( function* (input) { const operation = "CheckpointDiffQuery.getTurnDiff"; const ignoreWhitespace = input.ignoreWhitespace ?? true; yield* Effect.annotateCurrentSpan({ "checkpoint.thread_id": input.threadId, "checkpoint.from_turn_count": input.fromTurnCount, "checkpoint.to_turn_count": input.toTurnCount, "checkpoint.ignore_whitespace": ignoreWhitespace, }); if (input.fromTurnCount === input.toTurnCount) { const emptyDiff: OrchestrationGetTurnDiffResultType = { threadId: input.threadId, fromTurnCount: input.fromTurnCount, toTurnCount: input.toTurnCount, diff: "", }; if (!isTurnDiffResult(emptyDiff)) { return yield* new CheckpointInvariantError({ operation, detail: "Computed turn diff result does not satisfy contract schema.", }); } return emptyDiff; } const threadContext = yield* projectionSnapshotQuery .getThreadCheckpointContext(input.threadId) .pipe(Effect.withSpan("checkpoint.turnDiff.lookupContext")); if (Option.isNone(threadContext)) { return yield* new CheckpointInvariantError({ operation, detail: `Thread '${input.threadId}' not found.`, }); } const maxTurnCount = threadContext.value.checkpoints.reduce( (max, checkpoint) => Math.max(max, checkpoint.checkpointTurnCount), 0, ); if (input.toTurnCount > maxTurnCount) { return yield* new CheckpointUnavailableError({ threadId: input.threadId, turnCount: input.toTurnCount, detail: `Turn diff range exceeds current turn count: requested ${input.toTurnCount}, current ${maxTurnCount}.`, }); } const workspaceCwd = threadContext.value.worktreePath ?? threadContext.value.workspaceRoot; if (!workspaceCwd) { return yield* new CheckpointInvariantError({ operation, detail: `Workspace path missing for thread '${input.threadId}' when computing turn diff.`, }); } const fromCheckpointRef = input.fromTurnCount === 0 ? checkpointRefForThreadTurn(input.threadId, 0) : threadContext.value.checkpoints.find( (checkpoint) => checkpoint.checkpointTurnCount === input.fromTurnCount, )?.checkpointRef; if (!fromCheckpointRef) { return yield* new CheckpointUnavailableError({ threadId: input.threadId, turnCount: input.fromTurnCount, detail: `Checkpoint ref is unavailable for turn ${input.fromTurnCount}.`, }); } const toCheckpointRef = threadContext.value.checkpoints.find( (checkpoint) => checkpoint.checkpointTurnCount === input.toTurnCount, )?.checkpointRef; if (!toCheckpointRef) { return yield* new CheckpointUnavailableError({ threadId: input.threadId, turnCount: input.toTurnCount, detail: `Checkpoint ref is unavailable for turn ${input.toTurnCount}.`, }); } const diff = yield* checkpointStore .diffCheckpoints({ cwd: workspaceCwd, fromCheckpointRef, toCheckpointRef, fallbackFromToHead: false, ignoreWhitespace, }) .pipe(Effect.withSpan("checkpoint.turnDiff.diffCheckpoints")); const turnDiff = buildTurnDiffResult(input, diff); if (!isTurnDiffResult(turnDiff)) { return yield* new CheckpointInvariantError({ operation, detail: "Computed turn diff result does not satisfy contract schema.", }); } return turnDiff; }, ); const getFullThreadDiff: CheckpointDiffQueryShape["getFullThreadDiff"] = Effect.fn( "CheckpointDiffQuery.getFullThreadDiff", )(function* (input) { const operation = "CheckpointDiffQuery.getFullThreadDiff"; const ignoreWhitespace = input.ignoreWhitespace ?? true; yield* Effect.annotateCurrentSpan({ "checkpoint.thread_id": input.threadId, "checkpoint.from_turn_count": 0, "checkpoint.to_turn_count": input.toTurnCount, "checkpoint.ignore_whitespace": ignoreWhitespace, "checkpoint.diff_kind": "full-thread", }); if (input.toTurnCount === 0) { const emptyDiff = buildTurnDiffResult( { threadId: input.threadId, fromTurnCount: 0, toTurnCount: 0, }, "", ); if (!isTurnDiffResult(emptyDiff)) { return yield* new CheckpointInvariantError({ operation, detail: "Computed full thread diff result does not satisfy contract schema.", }); } return emptyDiff satisfies OrchestrationGetFullThreadDiffResult; } const threadContext = yield* projectionSnapshotQuery .getFullThreadDiffContext(input.threadId, input.toTurnCount) .pipe(Effect.withSpan("checkpoint.fullThread.lookupContext")); if (Option.isNone(threadContext)) { return yield* new CheckpointInvariantError({ operation, detail: `Thread '${input.threadId}' not found.`, }); } if (input.toTurnCount > threadContext.value.latestCheckpointTurnCount) { return yield* new CheckpointUnavailableError({ threadId: input.threadId, turnCount: input.toTurnCount, detail: `Turn diff range exceeds current turn count: requested ${input.toTurnCount}, current ${threadContext.value.latestCheckpointTurnCount}.`, }); } const workspaceCwd = threadContext.value.worktreePath ?? threadContext.value.workspaceRoot; if (!workspaceCwd) { return yield* new CheckpointInvariantError({ operation, detail: `Workspace path missing for thread '${input.threadId}' when computing full thread diff.`, }); } if (!threadContext.value.toCheckpointRef) { return yield* new CheckpointUnavailableError({ threadId: input.threadId, turnCount: input.toTurnCount, detail: `Checkpoint ref is unavailable for turn ${input.toTurnCount}.`, }); } const diff = yield* checkpointStore .diffCheckpoints({ cwd: workspaceCwd, fromCheckpointRef: checkpointRefForThreadTurn(input.threadId, 0), toCheckpointRef: threadContext.value.toCheckpointRef as CheckpointRef, fallbackFromToHead: false, ignoreWhitespace, }) .pipe(Effect.withSpan("checkpoint.fullThread.diffCheckpoints")); const turnDiff = buildTurnDiffResult( { threadId: input.threadId, fromTurnCount: 0, toTurnCount: input.toTurnCount, }, diff, ); if (!isTurnDiffResult(turnDiff)) { return yield* new CheckpointInvariantError({ operation, detail: "Computed full thread diff result does not satisfy contract schema.", }); } return turnDiff satisfies OrchestrationGetFullThreadDiffResult; }); return { getTurnDiff, getFullThreadDiff, } satisfies CheckpointDiffQueryShape;}); export const CheckpointDiffQueryLive = Layer.effect(CheckpointDiffQuery, make);
import {
type CheckpointRef,
OrchestrationGetTurnDiffResult,
type ThreadId,
type OrchestrationGetFullThreadDiffResult,
type OrchestrationGetTurnDiffResult as OrchestrationGetTurnDiffResultType,
} from "@t3tools/contracts";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
import { CheckpointInvariantError, CheckpointUnavailableError } from "../Errors.ts";
import { checkpointRefForThreadTurn } from "../Utils.ts";
import { CheckpointStore } from "../Services/CheckpointStore.ts";
import {
CheckpointDiffQuery,
type CheckpointDiffQueryShape,
} from "../Services/CheckpointDiffQuery.ts";
const isTurnDiffResult = Schema.is(OrchestrationGetTurnDiffResult);
function buildTurnDiffResult(
input: {
readonly threadId: ThreadId;
readonly fromTurnCount: number;
readonly toTurnCount: number;
},
diff: string,
): OrchestrationGetTurnDiffResultType {
return {
threadId: input.threadId,
fromTurnCount: input.fromTurnCount,
toTurnCount: input.toTurnCount,
diff,
};
}
const make = Effect.gen(function* () {
const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
const checkpointStore = yield* CheckpointStore;
const getTurnDiff: CheckpointDiffQueryShape["getTurnDiff"] = Effect.fn("getTurnDiff")(
function* (input) {
const operation = "CheckpointDiffQuery.getTurnDiff";
const ignoreWhitespace = input.ignoreWhitespace ?? true;
yield* Effect.annotateCurrentSpan({
"checkpoint.thread_id": input.threadId,
"checkpoint.from_turn_count": input.fromTurnCount,
"checkpoint.to_turn_count": input.toTurnCount,
"checkpoint.ignore_whitespace": ignoreWhitespace,
});
if (input.fromTurnCount === input.toTurnCount) {
const emptyDiff: OrchestrationGetTurnDiffResultType = {
threadId: input.threadId,
fromTurnCount: input.fromTurnCount,
toTurnCount: input.toTurnCount,
diff: "",
};
if (!isTurnDiffResult(emptyDiff)) {
return yield* new CheckpointInvariantError({
operation,
detail: "Computed turn diff result does not satisfy contract schema.",
});
}
return emptyDiff;
}
const threadContext = yield* projectionSnapshotQuery
.getThreadCheckpointContext(input.threadId)
.pipe(Effect.withSpan("checkpoint.turnDiff.lookupContext"));
if (Option.isNone(threadContext)) {
return yield* new CheckpointInvariantError({
operation,
detail: `Thread '${input.threadId}' not found.`,
});
}
const maxTurnCount = threadContext.value.checkpoints.reduce(
(max, checkpoint) => Math.max(max, checkpoint.checkpointTurnCount),
0,
);
if (input.toTurnCount > maxTurnCount) {
return yield* new CheckpointUnavailableError({
threadId: input.threadId,
turnCount: input.toTurnCount,
detail: `Turn diff range exceeds current turn count: requested ${input.toTurnCount}, current ${maxTurnCount}.`,
});
}
const workspaceCwd = threadContext.value.worktreePath ?? threadContext.value.workspaceRoot;
if (!workspaceCwd) {
return yield* new CheckpointInvariantError({
operation,
detail: `Workspace path missing for thread '${input.threadId}' when computing turn diff.`,
});
}
const fromCheckpointRef =
input.fromTurnCount === 0
? checkpointRefForThreadTurn(input.threadId, 0)
: threadContext.value.checkpoints.find(
(checkpoint) => checkpoint.checkpointTurnCount === input.fromTurnCount,
)?.checkpointRef;
if (!fromCheckpointRef) {
return yield* new CheckpointUnavailableError({
threadId: input.threadId,
turnCount: input.fromTurnCount,
detail: `Checkpoint ref is unavailable for turn ${input.fromTurnCount}.`,
});
}
const toCheckpointRef = threadContext.value.checkpoints.find(
(checkpoint) => checkpoint.checkpointTurnCount === input.toTurnCount,
)?.checkpointRef;
if (!toCheckpointRef) {
return yield* new CheckpointUnavailableError({
threadId: input.threadId,
turnCount: input.toTurnCount,
detail: `Checkpoint ref is unavailable for turn ${input.toTurnCount}.`,
});
}
const diff = yield* checkpointStore
.diffCheckpoints({
cwd: workspaceCwd,
fromCheckpointRef,
toCheckpointRef,
fallbackFromToHead: false,
ignoreWhitespace,
})
.pipe(Effect.withSpan("checkpoint.turnDiff.diffCheckpoints"));
const turnDiff = buildTurnDiffResult(input, diff);
if (!isTurnDiffResult(turnDiff)) {
return yield* new CheckpointInvariantError({
operation,
detail: "Computed turn diff result does not satisfy contract schema.",
});
}
return turnDiff;
},
);
const getFullThreadDiff: CheckpointDiffQueryShape["getFullThreadDiff"] = Effect.fn(
"CheckpointDiffQuery.getFullThreadDiff",
)(function* (input) {
const operation = "CheckpointDiffQuery.getFullThreadDiff";
const ignoreWhitespace = input.ignoreWhitespace ?? true;
yield* Effect.annotateCurrentSpan({
"checkpoint.thread_id": input.threadId,
"checkpoint.from_turn_count": 0,
"checkpoint.to_turn_count": input.toTurnCount,
"checkpoint.ignore_whitespace": ignoreWhitespace,
"checkpoint.diff_kind": "full-thread",
});
if (input.toTurnCount === 0) {
const emptyDiff = buildTurnDiffResult(
{
threadId: input.threadId,
fromTurnCount: 0,
toTurnCount: 0,
},
"",
);
if (!isTurnDiffResult(emptyDiff)) {
return yield* new CheckpointInvariantError({
operation,
detail: "Computed full thread diff result does not satisfy contract schema.",
});
}
return emptyDiff satisfies OrchestrationGetFullThreadDiffResult;
}
const threadContext = yield* projectionSnapshotQuery
.getFullThreadDiffContext(input.threadId, input.toTurnCount)
.pipe(Effect.withSpan("checkpoint.fullThread.lookupContext"));
if (Option.isNone(threadContext)) {
return yield* new CheckpointInvariantError({
operation,
detail: `Thread '${input.threadId}' not found.`,
});
}
if (input.toTurnCount > threadContext.value.latestCheckpointTurnCount) {
return yield* new CheckpointUnavailableError({
threadId: input.threadId,
turnCount: input.toTurnCount,
detail: `Turn diff range exceeds current turn count: requested ${input.toTurnCount}, current ${threadContext.value.latestCheckpointTurnCount}.`,
});
}
const workspaceCwd = threadContext.value.worktreePath ?? threadContext.value.workspaceRoot;
if (!workspaceCwd) {
return yield* new CheckpointInvariantError({
operation,
detail: `Workspace path missing for thread '${input.threadId}' when computing full thread diff.`,
});
}
if (!threadContext.value.toCheckpointRef) {
return yield* new CheckpointUnavailableError({
threadId: input.threadId,
turnCount: input.toTurnCount,
detail: `Checkpoint ref is unavailable for turn ${input.toTurnCount}.`,
});
}
const diff = yield* checkpointStore
.diffCheckpoints({
cwd: workspaceCwd,
fromCheckpointRef: checkpointRefForThreadTurn(input.threadId, 0),
toCheckpointRef: threadContext.value.toCheckpointRef as CheckpointRef,
fallbackFromToHead: false,
ignoreWhitespace,
})
.pipe(Effect.withSpan("checkpoint.fullThread.diffCheckpoints"));
const turnDiff = buildTurnDiffResult(
{
threadId: input.threadId,
fromTurnCount: 0,
toTurnCount: input.toTurnCount,
},
diff,
);
if (!isTurnDiffResult(turnDiff)) {
return yield* new CheckpointInvariantError({
operation,
detail: "Computed full thread diff result does not satisfy contract schema.",
});
}
return turnDiff satisfies OrchestrationGetFullThreadDiffResult;
});
return {
getTurnDiff,
getFullThreadDiff,
} satisfies CheckpointDiffQueryShape;
});
export const CheckpointDiffQueryLive = Layer.effect(CheckpointDiffQuery, make);