import * as NodeServices from "@effect/platform-node/NodeServices";import { assert, describe, it, vi } from "@effect/vitest";import { ProjectId, ProviderInstanceId, ThreadId, TurnId } from "@t3tools/contracts";import * as DateTime from "effect/DateTime";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as TestClock from "effect/testing/TestClock"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts";import { ProviderValidationError } from "../Errors.ts";import { ProviderSessionReaper } from "../Services/ProviderSessionReaper.ts";import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";import { makeProviderSessionReaperLive } from "./ProviderSessionReaper.ts"; const defaultModelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex",} as const; const TEST_NOW = DateTime.makeUnsafe("2026-04-14T00:30:00.000Z");const TEST_NOW_ISO = DateTime.formatIso(TEST_NOW);const STALE_LAST_SEEN_AT = DateTime.formatIso( DateTime.subtractDuration(TEST_NOW, Duration.seconds(2)),);const FRESH_LAST_SEEN_AT = DateTime.formatIso( DateTime.subtractDuration(TEST_NOW, Duration.millis(500)),); const drainFibers = Effect.forEach(Array.from({ length: 10 }), () => Effect.yieldNow, { discard: true,}); function makeReadModel( threads: ReadonlyArray<{ readonly id: ThreadId; readonly session: { readonly threadId: ThreadId; readonly status: "starting" | "running" | "ready" | "interrupted" | "stopped" | "error"; readonly providerName: "codex" | "claudeAgent"; readonly runtimeMode: "approval-required" | "full-access" | "auto-accept-edits"; readonly activeTurnId: TurnId | null; readonly lastError: string | null; readonly updatedAt: string; } | null; }>,) { const projectId = ProjectId.make("project-provider-session-reaper"); return { snapshotSequence: 0, updatedAt: TEST_NOW_ISO, projects: [ { id: projectId, title: "Provider Reaper Project", workspaceRoot: "/tmp/provider-reaper-project", defaultModelSelection, scripts: [], createdAt: TEST_NOW_ISO, updatedAt: TEST_NOW_ISO, deletedAt: null, }, ], threads: threads.map((thread) => ({ id: thread.id, projectId, title: `Thread ${thread.id}`, modelSelection: defaultModelSelection, interactionMode: "default" as const, runtimeMode: "full-access" as const, branch: null, worktreePath: null, createdAt: TEST_NOW_ISO, updatedAt: TEST_NOW_ISO, archivedAt: null, latestUserMessageAt: null, hasPendingApprovals: false, hasPendingUserInput: false, hasActionableProposedPlan: false, latestTurn: null, messages: [], session: thread.session, activities: [], proposedPlans: [], checkpoints: [], deletedAt: null, })), };} function createHarness(input: { readonly readModel: ReturnType<typeof makeReadModel>; readonly stopSessionImplementation?: (input: { readonly threadId: ThreadId; }) => ReturnType<ProviderServiceShape["stopSession"]>;}) { const stoppedThreadIds = new Set<ThreadId>(); const stopSession = vi.fn<ProviderServiceShape["stopSession"]>( (request) => (input.stopSessionImplementation ? input.stopSessionImplementation(request) : Effect.sync(() => { stoppedThreadIds.add(request.threadId); })) as ReturnType<ProviderServiceShape["stopSession"]>, ); const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( Layer.provide(runtimeRepositoryLayer), ); const layer = makeProviderSessionReaperLive({ inactivityThreshold: Duration.seconds(1), sweepInterval: Duration.seconds(60), }).pipe( Layer.provideMerge(providerSessionDirectoryLayer), Layer.provideMerge(runtimeRepositoryLayer), Layer.provideMerge( Layer.mock(ProviderService)({ stopSession, }), ), Layer.provideMerge( Layer.mock(ProjectionSnapshotQuery)({ getThreadShellById: (threadId) => Effect.succeed( Option.fromNullishOr(input.readModel.threads.find((thread) => thread.id === threadId)), ), }), ), Layer.provideMerge(NodeServices.layer), Layer.provideMerge(TestClock.layer()), ); return { layer, stopSession, stoppedThreadIds };} function seedSession(input: { readonly threadId: ThreadId; readonly providerName: "codex" | "claudeAgent"; readonly status?: "starting" | "running" | "stopped" | "error"; readonly lastSeenAt: string; readonly resumeOpaque: string;}) { return Effect.gen(function* () { const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; yield* repository.upsert({ threadId: input.threadId, providerName: input.providerName, providerInstanceId: null, adapterKey: input.providerName, runtimeMode: "full-access", status: input.status ?? "running", lastSeenAt: input.lastSeenAt, resumeCursor: { opaque: input.resumeOpaque, }, runtimePayload: null, }); });} const runInitialSweep = Effect.gen(function* () { const reaper = yield* ProviderSessionReaper; yield* reaper.start(); yield* drainFibers;}); function runWithHarness<A, E, R>( harness: ReturnType<typeof createHarness>, effect: Effect.Effect<A, E, R>,) { return Effect.gen(function* () { yield* TestClock.setTime(TEST_NOW.epochMilliseconds); return yield* effect; }).pipe(Effect.scoped, Effect.provide(harness.layer));} describe("ProviderSessionReaper", () => { it.effect("reaps stale persisted sessions without active turns", () => { const threadId = ThreadId.make("thread-reaper-stale"); const harness = createHarness({ readModel: makeReadModel([ { id: threadId, session: { threadId, status: "ready", providerName: "claudeAgent", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: TEST_NOW_ISO, }, }, ]), }); return runWithHarness( harness, Effect.gen(function* () { yield* seedSession({ threadId, providerName: "claudeAgent", lastSeenAt: STALE_LAST_SEEN_AT, resumeOpaque: "resume-stale", }); yield* runInitialSweep; assert.deepStrictEqual(harness.stopSession.mock.calls[0]?.[0], { threadId }); assert.isTrue(harness.stoppedThreadIds.has(threadId)); }), ); }); it.effect("skips stale sessions when the thread still has an active turn", () => { const threadId = ThreadId.make("thread-reaper-active-turn"); const turnId = TurnId.make("turn-reaper-active"); const harness = createHarness({ readModel: makeReadModel([ { id: threadId, session: { threadId, status: "running", providerName: "claudeAgent", runtimeMode: "full-access", activeTurnId: turnId, lastError: null, updatedAt: TEST_NOW_ISO, }, }, ]), }); return runWithHarness( harness, Effect.gen(function* () { yield* seedSession({ threadId, providerName: "claudeAgent", lastSeenAt: STALE_LAST_SEEN_AT, resumeOpaque: "resume-active-turn", }); const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; yield* runInitialSweep; assert.equal(harness.stopSession.mock.calls.length, 0); const remaining = yield* repository.getByThreadId({ threadId }); assert.isTrue(Option.isSome(remaining)); }), ); }); it.effect("does not reap sessions that are still within the inactivity threshold", () => { const threadId = ThreadId.make("thread-reaper-fresh"); const harness = createHarness({ readModel: makeReadModel([ { id: threadId, session: { threadId, status: "ready", providerName: "claudeAgent", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: TEST_NOW_ISO, }, }, ]), }); return runWithHarness( harness, Effect.gen(function* () { yield* seedSession({ threadId, providerName: "claudeAgent", lastSeenAt: FRESH_LAST_SEEN_AT, resumeOpaque: "resume-fresh", }); const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; yield* runInitialSweep; assert.equal(harness.stopSession.mock.calls.length, 0); const remaining = yield* repository.getByThreadId({ threadId }); assert.isTrue(Option.isSome(remaining)); }), ); }); it.effect("skips persisted sessions that are already marked stopped", () => { const threadId = ThreadId.make("thread-reaper-stopped"); const harness = createHarness({ readModel: makeReadModel([ { id: threadId, session: { threadId, status: "stopped", providerName: "claudeAgent", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: TEST_NOW_ISO, }, }, ]), }); return runWithHarness( harness, Effect.gen(function* () { yield* seedSession({ threadId, providerName: "claudeAgent", status: "stopped", lastSeenAt: STALE_LAST_SEEN_AT, resumeOpaque: "resume-stopped", }); const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; yield* runInitialSweep; assert.equal(harness.stopSession.mock.calls.length, 0); const remaining = yield* repository.getByThreadId({ threadId }); assert.isTrue(Option.isSome(remaining)); }), ); }); it.effect("continues reaping other sessions when one stop attempt fails", () => { const failedThreadId = ThreadId.make("thread-reaper-stop-failure"); const reapedThreadId = ThreadId.make("thread-reaper-stop-success"); const harness = createHarness({ readModel: makeReadModel([ { id: failedThreadId, session: { threadId: failedThreadId, status: "ready", providerName: "claudeAgent", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: TEST_NOW_ISO, }, }, { id: reapedThreadId, session: { threadId: reapedThreadId, status: "ready", providerName: "codex", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: TEST_NOW_ISO, }, }, ]), stopSessionImplementation: (request) => request.threadId === failedThreadId ? Effect.fail( new ProviderValidationError({ operation: "ProviderSessionReaper.test", issue: "simulated stop failure", }), ) : Effect.void, }); return runWithHarness( harness, Effect.gen(function* () { yield* seedSession({ threadId: failedThreadId, providerName: "claudeAgent", lastSeenAt: STALE_LAST_SEEN_AT, resumeOpaque: "resume-failure", }); yield* seedSession({ threadId: reapedThreadId, providerName: "codex", lastSeenAt: STALE_LAST_SEEN_AT, resumeOpaque: "resume-success", }); yield* runInitialSweep; assert.equal(harness.stopSession.mock.calls.length, 2); assert.deepStrictEqual( new Set(harness.stopSession.mock.calls.map(([request]) => request.threadId)), new Set([failedThreadId, reapedThreadId]), ); }), ); }); it.effect("continues reaping other sessions when one stop attempt defects", () => { const defectThreadId = ThreadId.make("thread-reaper-stop-defect"); const reapedThreadId = ThreadId.make("thread-reaper-stop-after-defect"); const harness = createHarness({ readModel: makeReadModel([ { id: defectThreadId, session: { threadId: defectThreadId, status: "ready", providerName: "claudeAgent", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: TEST_NOW_ISO, }, }, { id: reapedThreadId, session: { threadId: reapedThreadId, status: "ready", providerName: "codex", runtimeMode: "full-access", activeTurnId: null, lastError: null, updatedAt: TEST_NOW_ISO, }, }, ]), stopSessionImplementation: (request) => request.threadId === defectThreadId ? Effect.die(new Error("simulated stop defect")) : Effect.void, }); return runWithHarness( harness, Effect.gen(function* () { yield* seedSession({ threadId: defectThreadId, providerName: "claudeAgent", lastSeenAt: STALE_LAST_SEEN_AT, resumeOpaque: "resume-defect", }); yield* seedSession({ threadId: reapedThreadId, providerName: "codex", lastSeenAt: STALE_LAST_SEEN_AT, resumeOpaque: "resume-after-defect", }); yield* runInitialSweep; assert.equal(harness.stopSession.mock.calls.length, 2); assert.deepStrictEqual( new Set(harness.stopSession.mock.calls.map(([request]) => request.threadId)), new Set([defectThreadId, reapedThreadId]), ); }), ); });});
import * as NodeServices from "@effect/platform-node/NodeServices";
import { assert, describe, it, vi } from "@effect/vitest";
import { ProjectId, ProviderInstanceId, ThreadId, TurnId } from "@t3tools/contracts";
import * as DateTime from "effect/DateTime";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as TestClock from "effect/testing/TestClock";
import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";
import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts";
import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts";
import { ProviderValidationError } from "../Errors.ts";
import { ProviderSessionReaper } from "../Services/ProviderSessionReaper.ts";
import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";
import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";
import { makeProviderSessionReaperLive } from "./ProviderSessionReaper.ts";
const defaultModelSelection = {
instanceId: ProviderInstanceId.make("codex"),
model: "gpt-5-codex",
} as const;
const TEST_NOW = DateTime.makeUnsafe("2026-04-14T00:30:00.000Z");
const TEST_NOW_ISO = DateTime.formatIso(TEST_NOW);
const STALE_LAST_SEEN_AT = DateTime.formatIso(
DateTime.subtractDuration(TEST_NOW, Duration.seconds(2)),
);
const FRESH_LAST_SEEN_AT = DateTime.formatIso(
DateTime.subtractDuration(TEST_NOW, Duration.millis(500)),
);
const drainFibers = Effect.forEach(Array.from({ length: 10 }), () => Effect.yieldNow, {
discard: true,
});
function makeReadModel(
threads: ReadonlyArray<{
readonly id: ThreadId;
readonly session: {
readonly threadId: ThreadId;
readonly status: "starting" | "running" | "ready" | "interrupted" | "stopped" | "error";
readonly providerName: "codex" | "claudeAgent";
readonly runtimeMode: "approval-required" | "full-access" | "auto-accept-edits";
readonly activeTurnId: TurnId | null;
readonly lastError: string | null;
readonly updatedAt: string;
} | null;
}>,
) {
const projectId = ProjectId.make("project-provider-session-reaper");
return {
snapshotSequence: 0,
updatedAt: TEST_NOW_ISO,
projects: [
{
id: projectId,
title: "Provider Reaper Project",
workspaceRoot: "/tmp/provider-reaper-project",
defaultModelSelection,
scripts: [],
createdAt: TEST_NOW_ISO,
updatedAt: TEST_NOW_ISO,
deletedAt: null,
},
],
threads: threads.map((thread) => ({
id: thread.id,
projectId,
title: `Thread ${thread.id}`,
modelSelection: defaultModelSelection,
interactionMode: "default" as const,
runtimeMode: "full-access" as const,
branch: null,
worktreePath: null,
createdAt: TEST_NOW_ISO,
updatedAt: TEST_NOW_ISO,
archivedAt: null,
latestUserMessageAt: null,
hasPendingApprovals: false,
hasPendingUserInput: false,
hasActionableProposedPlan: false,
latestTurn: null,
messages: [],
session: thread.session,
activities: [],
proposedPlans: [],
checkpoints: [],
deletedAt: null,
})),
};
}
function createHarness(input: {
readonly readModel: ReturnType<typeof makeReadModel>;
readonly stopSessionImplementation?: (input: {
readonly threadId: ThreadId;
}) => ReturnType<ProviderServiceShape["stopSession"]>;
}) {
const stoppedThreadIds = new Set<ThreadId>();
const stopSession = vi.fn<ProviderServiceShape["stopSession"]>(
(request) =>
(input.stopSessionImplementation
? input.stopSessionImplementation(request)
: Effect.sync(() => {
stoppedThreadIds.add(request.threadId);
})) as ReturnType<ProviderServiceShape["stopSession"]>,
);
const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(
Layer.provide(SqlitePersistenceMemory),
);
const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(
Layer.provide(runtimeRepositoryLayer),
);
const layer = makeProviderSessionReaperLive({
inactivityThreshold: Duration.seconds(1),
sweepInterval: Duration.seconds(60),
}).pipe(
Layer.provideMerge(providerSessionDirectoryLayer),
Layer.provideMerge(runtimeRepositoryLayer),
Layer.provideMerge(
Layer.mock(ProviderService)({
stopSession,
}),
),
Layer.provideMerge(
Layer.mock(ProjectionSnapshotQuery)({
getThreadShellById: (threadId) =>
Effect.succeed(
Option.fromNullishOr(input.readModel.threads.find((thread) => thread.id === threadId)),
),
}),
),
Layer.provideMerge(NodeServices.layer),
Layer.provideMerge(TestClock.layer()),
);
return { layer, stopSession, stoppedThreadIds };
}
function seedSession(input: {
readonly threadId: ThreadId;
readonly providerName: "codex" | "claudeAgent";
readonly status?: "starting" | "running" | "stopped" | "error";
readonly lastSeenAt: string;
readonly resumeOpaque: string;
}) {
return Effect.gen(function* () {
const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
yield* repository.upsert({
threadId: input.threadId,
providerName: input.providerName,
providerInstanceId: null,
adapterKey: input.providerName,
runtimeMode: "full-access",
status: input.status ?? "running",
lastSeenAt: input.lastSeenAt,
resumeCursor: {
opaque: input.resumeOpaque,
},
runtimePayload: null,
});
});
}
const runInitialSweep = Effect.gen(function* () {
const reaper = yield* ProviderSessionReaper;
yield* reaper.start();
yield* drainFibers;
});
function runWithHarness<A, E, R>(
harness: ReturnType<typeof createHarness>,
effect: Effect.Effect<A, E, R>,
) {
return Effect.gen(function* () {
yield* TestClock.setTime(TEST_NOW.epochMilliseconds);
return yield* effect;
}).pipe(Effect.scoped, Effect.provide(harness.layer));
}
describe("ProviderSessionReaper", () => {
it.effect("reaps stale persisted sessions without active turns", () => {
const threadId = ThreadId.make("thread-reaper-stale");
const harness = createHarness({
readModel: makeReadModel([
{
id: threadId,
session: {
threadId,
status: "ready",
providerName: "claudeAgent",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
]),
});
return runWithHarness(
harness,
Effect.gen(function* () {
yield* seedSession({
threadId,
providerName: "claudeAgent",
lastSeenAt: STALE_LAST_SEEN_AT,
resumeOpaque: "resume-stale",
});
yield* runInitialSweep;
assert.deepStrictEqual(harness.stopSession.mock.calls[0]?.[0], { threadId });
assert.isTrue(harness.stoppedThreadIds.has(threadId));
}),
);
});
it.effect("skips stale sessions when the thread still has an active turn", () => {
const threadId = ThreadId.make("thread-reaper-active-turn");
const turnId = TurnId.make("turn-reaper-active");
const harness = createHarness({
readModel: makeReadModel([
{
id: threadId,
session: {
threadId,
status: "running",
providerName: "claudeAgent",
runtimeMode: "full-access",
activeTurnId: turnId,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
]),
});
return runWithHarness(
harness,
Effect.gen(function* () {
yield* seedSession({
threadId,
providerName: "claudeAgent",
lastSeenAt: STALE_LAST_SEEN_AT,
resumeOpaque: "resume-active-turn",
});
const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
yield* runInitialSweep;
assert.equal(harness.stopSession.mock.calls.length, 0);
const remaining = yield* repository.getByThreadId({ threadId });
assert.isTrue(Option.isSome(remaining));
}),
);
});
it.effect("does not reap sessions that are still within the inactivity threshold", () => {
const threadId = ThreadId.make("thread-reaper-fresh");
const harness = createHarness({
readModel: makeReadModel([
{
id: threadId,
session: {
threadId,
status: "ready",
providerName: "claudeAgent",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
]),
});
return runWithHarness(
harness,
Effect.gen(function* () {
yield* seedSession({
threadId,
providerName: "claudeAgent",
lastSeenAt: FRESH_LAST_SEEN_AT,
resumeOpaque: "resume-fresh",
});
const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
yield* runInitialSweep;
assert.equal(harness.stopSession.mock.calls.length, 0);
const remaining = yield* repository.getByThreadId({ threadId });
assert.isTrue(Option.isSome(remaining));
}),
);
});
it.effect("skips persisted sessions that are already marked stopped", () => {
const threadId = ThreadId.make("thread-reaper-stopped");
const harness = createHarness({
readModel: makeReadModel([
{
id: threadId,
session: {
threadId,
status: "stopped",
providerName: "claudeAgent",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
]),
});
return runWithHarness(
harness,
Effect.gen(function* () {
yield* seedSession({
threadId,
providerName: "claudeAgent",
status: "stopped",
lastSeenAt: STALE_LAST_SEEN_AT,
resumeOpaque: "resume-stopped",
});
const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
yield* runInitialSweep;
assert.equal(harness.stopSession.mock.calls.length, 0);
const remaining = yield* repository.getByThreadId({ threadId });
assert.isTrue(Option.isSome(remaining));
}),
);
});
it.effect("continues reaping other sessions when one stop attempt fails", () => {
const failedThreadId = ThreadId.make("thread-reaper-stop-failure");
const reapedThreadId = ThreadId.make("thread-reaper-stop-success");
const harness = createHarness({
readModel: makeReadModel([
{
id: failedThreadId,
session: {
threadId: failedThreadId,
status: "ready",
providerName: "claudeAgent",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
{
id: reapedThreadId,
session: {
threadId: reapedThreadId,
status: "ready",
providerName: "codex",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
]),
stopSessionImplementation: (request) =>
request.threadId === failedThreadId
? Effect.fail(
new ProviderValidationError({
operation: "ProviderSessionReaper.test",
issue: "simulated stop failure",
}),
)
: Effect.void,
});
return runWithHarness(
harness,
Effect.gen(function* () {
yield* seedSession({
threadId: failedThreadId,
providerName: "claudeAgent",
lastSeenAt: STALE_LAST_SEEN_AT,
resumeOpaque: "resume-failure",
});
yield* seedSession({
threadId: reapedThreadId,
providerName: "codex",
lastSeenAt: STALE_LAST_SEEN_AT,
resumeOpaque: "resume-success",
});
yield* runInitialSweep;
assert.equal(harness.stopSession.mock.calls.length, 2);
assert.deepStrictEqual(
new Set(harness.stopSession.mock.calls.map(([request]) => request.threadId)),
new Set([failedThreadId, reapedThreadId]),
);
}),
);
});
it.effect("continues reaping other sessions when one stop attempt defects", () => {
const defectThreadId = ThreadId.make("thread-reaper-stop-defect");
const reapedThreadId = ThreadId.make("thread-reaper-stop-after-defect");
const harness = createHarness({
readModel: makeReadModel([
{
id: defectThreadId,
session: {
threadId: defectThreadId,
status: "ready",
providerName: "claudeAgent",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
{
id: reapedThreadId,
session: {
threadId: reapedThreadId,
status: "ready",
providerName: "codex",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: TEST_NOW_ISO,
},
},
]),
stopSessionImplementation: (request) =>
request.threadId === defectThreadId
? Effect.die(new Error("simulated stop defect"))
: Effect.void,
});
return runWithHarness(
harness,
Effect.gen(function* () {
yield* seedSession({
threadId: defectThreadId,
providerName: "claudeAgent",
lastSeenAt: STALE_LAST_SEEN_AT,
resumeOpaque: "resume-defect",
});
yield* seedSession({
threadId: reapedThreadId,
providerName: "codex",
lastSeenAt: STALE_LAST_SEEN_AT,
resumeOpaque: "resume-after-defect",
});
yield* runInitialSweep;
assert.equal(harness.stopSession.mock.calls.length, 2);
assert.deepStrictEqual(
new Set(harness.stopSession.mock.calls.map(([request]) => request.threadId)),
new Set([defectThreadId, reapedThreadId]),
);
}),
);
});
});