import * as NodeServices from "@effect/platform-node/NodeServices";import {  ProjectId,  ThreadId,  TurnId,  ProviderDriverKind,  ProviderInstanceId,} from "@t3tools/contracts";import * as Clock from "effect/Clock";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Layer from "effect/Layer";import * as ManagedRuntime from "effect/ManagedRuntime";import * as Option from "effect/Option";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import { afterEach, describe, expect, it, vi } from "vite-plus/test"; 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; async function waitFor(  predicate: () => boolean | Promise<boolean>,  timeoutMs = 2_000,): Promise<void> {  const deadline = (await Effect.runPromise(Clock.currentTimeMillis)) + timeoutMs;  const poll = async (): Promise<void> => {    if (await predicate()) {      return;    }    if ((await Effect.runPromise(Clock.currentTimeMillis)) >= deadline) {      throw new Error("Timed out waiting for expectation.");    }    await Effect.runPromise(Effect.yieldNow);    return poll();  };   return poll();} const drainFibers = Effect.forEach(Array.from({ length: 10 }), () => Effect.yieldNow, {  discard: true,}); const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; 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;    readonly backgroundLiveness?: "working" | "monitoring" | null;  }>,) {  const now = "2026-01-01T00:00:00.000Z";  const projectId = ProjectId.make("project-provider-session-reaper");   return {    snapshotSequence: 0,    updatedAt: now,    projects: [      {        id: projectId,        title: "Provider Reaper Project",        workspaceRoot: "/tmp/provider-reaper-project",        defaultModelSelection,        scripts: [],        createdAt: now,        updatedAt: now,        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,      pullRequests: [],      createdAt: now,      updatedAt: now,      archivedAt: null,      settledOverride: null,      settledAt: null,      latestUserMessageAt: null,      hasPendingApprovals: false,      hasPendingUserInput: false,      hasActionableProposedPlan: false,      latestTurn: null,      messages: [],      session: thread.session,      backgroundLiveness: thread.backgroundLiveness ?? null,      activities: [],      proposedPlans: [],      checkpoints: [],      deletedAt: null,    })),  };} describe("ProviderSessionReaper", () => {  let runtime: ManagedRuntime.ManagedRuntime<    ProviderSessionReaper | ProviderSessionRuntime.ProviderSessionRuntimeRepository,    unknown  > | null = null;  let scope: Scope.Closeable | null = null;   afterEach(async () => {    if (scope) {      await Effect.runPromise(Scope.close(scope, Exit.void));    }    scope = null;    if (runtime) {      await runtime.dispose();    }    runtime = null;  });   // Shared start sequence so each test adds no manual Effect runners  // (no-manual-effect-runtime-in-tests tracks this file's legacy count).  async function startReaper() {    await runtime!.runPromise(      Effect.gen(function* () {        const reaper = yield* ProviderSessionReaper;        scope = yield* Scope.make("sequential");        yield* reaper.start().pipe(Scope.provide(scope));      }),    );  }   async function sweepAt(nowMs: number) {    await runtime!.runPromise(      Effect.gen(function* () {        const reaper = yield* ProviderSessionReaper;        const clock = yield* Clock.Clock;        const swept = yield* Deferred.make<void>();        yield* reaper.start().pipe(          Effect.provideService(Clock.Clock, {            currentTimeMillis: Effect.succeed(nowMs),            currentTimeMillisUnsafe: () => nowMs,            currentTimeNanos: Effect.succeed(BigInt(nowMs) * 1_000_000n),            currentTimeNanosUnsafe: () => BigInt(nowMs) * 1_000_000n,            monotonicTimeNanos: clock.monotonicTimeNanos,            monotonicTimeNanosUnsafe: () => clock.monotonicTimeNanosUnsafe(),            // Reaching the next scheduled sleep proves this sweep has finished.            sleep: () => Deferred.succeed(swept, undefined).pipe(Effect.andThen(Effect.never)),          }),        );        yield* Deferred.await(swept);      }).pipe(Effect.scoped),    );  }   async 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 providerService: ProviderServiceShape = {      startSession: () => unsupported(),      sendTurn: () => unsupported(),      compactThread: () => unsupported(),      interruptTurn: () => unsupported(),      respondToRequest: () => unsupported(),      respondToUserInput: () => unsupported(),      stopSession,      listSessions: () => Effect.succeed([]),      getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }),      assertConversationRollbackSupported: () => unsupported(),      getInstanceInfo: (instanceId) => {        const driverKind = ProviderDriverKind.make(String(instanceId));        return Effect.succeed({          instanceId,          driverKind,          displayName: undefined,          enabled: true,          continuationIdentity: {            driverKind,            continuationKey: `${driverKind}:instance:${instanceId}`,          },        });      },      rollbackConversation: () => unsupported(),      uploadFeedback: () => unsupported(),      streamEvents: Stream.empty,    };     const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(      Layer.provide(SqlitePersistenceMemory),    );    const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(      Layer.provide(runtimeRepositoryLayer),    );    const layer = makeProviderSessionReaperLive({      inactivityThresholdMs: 1_000,      sweepIntervalMs: 60_000,    }).pipe(      Layer.provideMerge(providerSessionDirectoryLayer),      Layer.provideMerge(runtimeRepositoryLayer),      Layer.provideMerge(Layer.succeed(ProviderService, providerService)),      Layer.provideMerge(        Layer.succeed(ProjectionSnapshotQuery, {          getUserInputActivity: () => Effect.die("unused"),          getCommandReadModel: () => Effect.die("unused"),          getSnapshot: () => Effect.die("unused"),          getShellSnapshot: () => Effect.die("unused"),          getArchivedShellSnapshot: () => Effect.die("unused"),          getSnapshotSequence: () =>            Effect.succeed({ snapshotSequence: input.readModel.snapshotSequence }),          getCounts: () => Effect.die("unused"),          getEventReplayStats: () => Effect.die("unused"),          getActiveProjectByWorkspaceRoot: () => Effect.die("unused"),          getProjectShellById: () => Effect.die("unused"),          getFirstActiveThreadIdByProjectId: () => Effect.die("unused"),          getImportedAgentSessionSources: () => Effect.die("unused"),          getThreadCheckpointContext: () => Effect.die("unused"),          getFullThreadDiffContext: () => Effect.die("unused"),          getThreadRuntimeContext: () => Effect.die("unused"),          getTurnStartMessage: () => Effect.die("unused"),          getThreadShellById: (threadId) =>            Effect.succeed(              input.readModel.threads.find((thread) => thread.id === threadId)                ? Option.some(input.readModel.threads.find((thread) => thread.id === threadId)!)                : Option.none(),            ),          getThreadDetailById: () => Effect.die("unused"),          getThreadDetailSnapshot: () => Effect.die("unused"),          searchThreads: () => Effect.succeed({ matches: [] }),        }),      ),      Layer.provideMerge(NodeServices.layer),    );     runtime = ManagedRuntime.make(layer);    return { stopSession, stoppedThreadIds };  }   it("reaps stale persisted sessions without active turns", async () => {    const threadId = ThreadId.make("thread-reaper-stale");    const now = "2026-01-01T00:00:00.000Z";    const harness = await createHarness({      readModel: makeReadModel([        {          id: threadId,          session: {            threadId,            status: "ready",            providerName: "claudeAgent",            runtimeMode: "full-access",            activeTurnId: null,            lastError: null,            updatedAt: now,          },        },      ]),    });    const repository = await runtime!.runPromise(      Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),    );     await runtime!.runPromise(      repository.upsert({        threadId,        providerName: "claudeAgent",        providerInstanceId: null,        adapterKey: "claudeAgent",        runtimeMode: "full-access",        status: "running",        lastSeenAt: "2026-04-14T00:00:00.000Z",        resumeCursor: {          opaque: "resume-stale",        },        runtimePayload: null,      }),    );     await startReaper();     await waitFor(() => harness.stopSession.mock.calls.length === 1);     expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ threadId });    expect(harness.stoppedThreadIds.has(threadId)).toBe(true);  });   it("skips stale sessions when the thread still has an active turn", async () => {    const threadId = ThreadId.make("thread-reaper-active-turn");    const turnId = TurnId.make("turn-reaper-active");    const now = "2026-01-01T00:00:00.000Z";    const harness = await createHarness({      readModel: makeReadModel([        {          id: threadId,          session: {            threadId,            status: "running",            providerName: "claudeAgent",            runtimeMode: "full-access",            activeTurnId: turnId,            lastError: null,            updatedAt: now,          },        },      ]),    });    const repository = await runtime!.runPromise(      Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),    );     await runtime!.runPromise(      repository.upsert({        threadId,        providerName: "claudeAgent",        providerInstanceId: null,        adapterKey: "claudeAgent",        runtimeMode: "full-access",        status: "running",        lastSeenAt: "2026-04-14T00:00:00.000Z",        resumeCursor: {          opaque: "resume-active-turn",        },        runtimePayload: null,      }),    );     await startReaper();    await Effect.runPromise(drainFibers);     expect(harness.stopSession).not.toHaveBeenCalled();    const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId }));    expect(Option.isSome(remaining)).toBe(true);  });   it("skips stale sessions while background work is still live", async () => {    const threadId = ThreadId.make("thread-reaper-background-work");    const now = "2026-01-01T00:00:00.000Z";    const harness = await createHarness({      readModel: makeReadModel([        {          id: threadId,          session: {            threadId,            status: "ready",            providerName: "claudeAgent",            runtimeMode: "full-access",            activeTurnId: null,            lastError: null,            updatedAt: now,          },          backgroundLiveness: "working",        },      ]),    });    const repository = await runtime!.runPromise(      Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),    );     await runtime!.runPromise(      repository.upsert({        threadId,        providerName: "claudeAgent",        providerInstanceId: null,        adapterKey: "claudeAgent",        runtimeMode: "full-access",        status: "running",        lastSeenAt: "2026-04-14T00:00:00.000Z",        resumeCursor: {          opaque: "resume-background-work",        },        runtimePayload: null,      }),    );     await startReaper();    await Effect.runPromise(drainFibers);     expect(harness.stopSession).not.toHaveBeenCalled();    const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId }));    expect(Option.isSome(remaining)).toBe(true);  });   it.each(["ready", "interrupted", "error"] as const)(    "gives a long turn a full idle window after becoming %s",    async (status) => {      const threadId = ThreadId.make(`thread-reaper-long-turn-${status}`);      const startedAt = "2026-04-14T00:00:00.000Z";      const completedAt = "2026-04-14T01:00:00.000Z";      const completedAtMs = Date.parse(completedAt);      const readModel = makeReadModel([        {          id: threadId,          session: {            threadId,            status: "running",            providerName: "claudeAgent",            runtimeMode: "full-access",            activeTurnId: TurnId.make("turn-reaper-long"),            lastError: null,            updatedAt: startedAt,          },        },      ]);      const harness = await createHarness({ readModel });      const repository = await runtime!.runPromise(        Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),      );      await runtime!.runPromise(        repository.upsert({          threadId,          providerName: "claudeAgent",          providerInstanceId: null,          adapterKey: "claudeAgent",          runtimeMode: "full-access",          status: "running",          lastSeenAt: startedAt,          resumeCursor: { opaque: "resume-long-turn" },          runtimePayload: null,        }),      );       await sweepAt(completedAtMs);      expect(harness.stopSession).not.toHaveBeenCalled();       // Ingestion changes the timestamp and clears the active turn in the same session row.      readModel.threads[0]!.session = {        ...readModel.threads[0]!.session!,        status,        activeTurnId: null,        updatedAt: completedAt,      };      await sweepAt(completedAtMs);      expect(harness.stopSession).not.toHaveBeenCalled();      await sweepAt(completedAtMs + 999);      expect(harness.stopSession).not.toHaveBeenCalled();      await sweepAt(completedAtMs + 1_000);      expect(harness.stopSession).toHaveBeenCalledExactlyOnceWith({ threadId });    },  );   it.each([true, false])(    "uses the binding idle window when the session is older or missing, hasSession=%s",    async (hasSession) => {      const threadId = ThreadId.make("thread-reaper-fresh");      const now = "2026-04-14T01:00:00.000Z";      const nowMs = Date.parse(now);      const harness = await createHarness({        readModel: makeReadModel([          {            id: threadId,            session: hasSession              ? {                  threadId,                  status: "ready",                  providerName: "claudeAgent",                  runtimeMode: "full-access",                  activeTurnId: null,                  lastError: null,                  updatedAt: "2026-04-14T00:00:00.000Z",                }              : null,          },        ]),      });      const repository = await runtime!.runPromise(        Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),      );      await runtime!.runPromise(        repository.upsert({

Lines 1–500 of 733. Load more