import { assert, describe, expect, it } from "@effect/vitest";import * as DateTime from "effect/DateTime";import {  DEFAULT_SERVER_SETTINGS,  ProjectId,  ProviderInstanceId,  ThreadId,  type OrchestrationProjectShell,  type OrchestrationV2Command,  type OrchestrationV2ShellSnapshot,  type OrchestrationV2ThreadShell,  type PullRequestSummary,  type ServerSettings,  type ServerSettingsPatch,} from "@t3tools/contracts";import { applyServerSettingsPatch } from "@t3tools/shared/serverSettings";import * as Crypto from "effect/Crypto";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as PubSub from "effect/PubSub";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Stream from "effect/Stream";import { TestClock } from "effect/testing"; import { GitManager } from "../git/GitManager.ts";import {  PullRequestService,  type PullRequestMergeEvent,} from "../pullRequest/PullRequestService.ts";import { ServerActivation } from "../serverActivation.ts";import { ServerSettingsService } from "../serverSettings.ts";import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";import { OrchestratorV2, type OrchestratorV2Shape } from "./Orchestrator.ts";import { ProjectionStoreV2 } from "./ProjectionStore.ts";import * as ThreadSettlementService from "./ThreadSettlementService.ts"; import {  isAutoSettlementCandidate,  QUEUED_TURN_START_GRACE_MS,  resolveAutoSettlementAt,  threadHasQueuedTurnStart,} from "./ThreadSettlementService.ts"; const NOW_MS = Date.parse("2026-06-10T12:00:00.000Z");const DAY_MS = 24 * 60 * 60 * 1_000; function at(offsetMs: number): DateTime.Utc {  return DateTime.makeUnsafe(NOW_MS + offsetMs);} function shell(overrides: Partial<OrchestrationV2ThreadShell> = {}): OrchestrationV2ThreadShell {  return {    id: ThreadId.make("thread-1"),    projectId: ProjectId.make("project-1"),    title: "Thread",    providerInstanceId: ProviderInstanceId.make("codex"),    modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },    runtimeMode: "full-access",    interactionMode: "default",    worktreePath: null,    activeProviderThreadId: null,    lineage: {      rootThreadId: ThreadId.make("thread-1"),      parentThreadId: null,      relationshipToParent: null,    },    forkedFrom: null,    createdBy: "user",    creationSource: "web",    activeRunId: null,    latestVisibleMessage: null,    hasActionableProposedPlan: false,    itemCount: 0,    visibleItemCount: 0,    lastVisitedAt: null,    deletedAt: null,    branch: null,    linkedPullRequest: null,    status: "idle",    activityRunStatus: null,    pendingRuntimeRequest: null,    pendingBackgroundTasks: [],    latestRunId: null,    latestRunRequestedAt: null,    latestRunStartedAt: null,    latestRunCompletedAt: null,    latestUserMessageAt: null,    createdAt: at(-30 * DAY_MS),    updatedAt: at(-10 * DAY_MS),    archivedAt: null,    settledOverride: null,    settledAt: null,    snoozedUntil: null,    snoozedAt: null,    pinnedAt: null,    ...overrides,  };} describe("isAutoSettlementCandidate", () => {  it("excludes overridden, pinned, blocked, and working threads", () => {    expect(isAutoSettlementCandidate(shell(), NOW_MS)).toBe(true);    expect(isAutoSettlementCandidate(shell({ archivedAt: at(-1) }), NOW_MS)).toBe(false);    expect(isAutoSettlementCandidate(shell({ settledOverride: "settled" }), NOW_MS)).toBe(false);    expect(isAutoSettlementCandidate(shell({ settledOverride: "active" }), NOW_MS)).toBe(false);    expect(isAutoSettlementCandidate(shell({ pinnedAt: at(-1) }), NOW_MS)).toBe(false);    expect(isAutoSettlementCandidate(shell({ activityRunStatus: "running" }), NOW_MS)).toBe(false);    expect(      isAutoSettlementCandidate(        shell({          pendingRuntimeRequest: { kind: "approval" } as never,        }),        NOW_MS,      ),    ).toBe(false);    expect(      isAutoSettlementCandidate(        shell({ pendingBackgroundTasks: [{ label: "task" }] as never }),        NOW_MS,      ),    ).toBe(false);  });   it("keeps snoozed threads parked until they wake early on error or completion", () => {    const snoozed = shell({      snoozedUntil: at(60 * 60 * 1_000),      snoozedAt: at(-60 * 60 * 1_000),    });    expect(isAutoSettlementCandidate(snoozed, NOW_MS)).toBe(false);    expect(      isAutoSettlementCandidate(        shell({ ...snoozed, status: "failed", latestRunCompletedAt: at(-30 * 60 * 1_000) }),        NOW_MS,      ),    ).toBe(true);    expect(      isAutoSettlementCandidate(        shell({ ...snoozed, latestRunCompletedAt: at(-30 * 60 * 1_000) }),        NOW_MS,      ),    ).toBe(true);    expect(      isAutoSettlementCandidate(        shell({ ...snoozed, status: "failed", latestRunCompletedAt: at(-2 * 60 * 60 * 1_000) }),        NOW_MS,      ),    ).toBe(false);    expect(      isAutoSettlementCandidate(        shell({ ...snoozed, status: "failed", latestRunCompletedAt: snoozed.snoozedAt }),        NOW_MS,      ),    ).toBe(false);    expect(      isAutoSettlementCandidate(        shell({ ...snoozed, status: "failed", latestRunCompletedAt: null }),        NOW_MS,      ),    ).toBe(false);    // Expired snooze is no longer a park.    expect(isAutoSettlementCandidate(shell({ ...snoozed, snoozedUntil: at(-1) }), NOW_MS)).toBe(      true,    );  });}); describe("threadHasQueuedTurnStart", () => {  it("holds a fresh unadopted user message inside the grace window only", () => {    const fresh = shell({ latestUserMessageAt: at(-1_000) });    expect(threadHasQueuedTurnStart(fresh, NOW_MS)).toBe(true);    // Adoption stamps the run with the message time, clearing the hold.    expect(      threadHasQueuedTurnStart(        shell({          latestUserMessageAt: at(-1_000),          latestRunId: "run-1" as never,          latestRunRequestedAt: at(-500),        }),        NOW_MS,      ),    ).toBe(false);    // Outside the grace window the stale message no longer blocks.    expect(      threadHasQueuedTurnStart(        shell({ latestUserMessageAt: at(-QUEUED_TURN_START_GRACE_MS - 1) }),        NOW_MS,      ),    ).toBe(false);    // Client clocks ahead of the server must not extend the hold.    expect(      threadHasQueuedTurnStart(        shell({ latestUserMessageAt: at(QUEUED_TURN_START_GRACE_MS + 1) }),        NOW_MS,      ),    ).toBe(false);    // A failed start clears the hold immediately.    expect(      threadHasQueuedTurnStart(        shell({ latestUserMessageAt: at(-1_000), status: "failed" }),        NOW_MS,      ),    ).toBe(false);  });}); describe("resolveAutoSettlementAt", () => {  it("uses the latest activity time when the inactivity window elapses", () => {    const idle = shell({      latestUserMessageAt: at(-4 * DAY_MS),      latestRunRequestedAt: at(-4 * DAY_MS),      latestRunStartedAt: at(-4 * DAY_MS),      latestRunCompletedAt: at(-3 * DAY_MS),    });    const input = {      thread: idle,      pullRequest: null,      nowMs: NOW_MS,      autoSettleAfterDays: 2,      autoSettleOnMerge: true,    };    expect(resolveAutoSettlementAt(input)).toEqual(at(-3 * DAY_MS));    expect(resolveAutoSettlementAt({ ...input, autoSettleAfterDays: 5 })).toBeNull();    expect(resolveAutoSettlementAt({ ...input, autoSettleAfterDays: null })).toBeNull();    expect(resolveAutoSettlementAt({ ...input, thread: shell() })).toBeNull();  });   it("settles on merge only after the user's last action and preserves the activity time", () => {    const thread = shell({      latestUserMessageAt: at(-2 * 60 * 60 * 1_000),      latestRunCompletedAt: at(-90 * 60 * 1_000),    });    const input = {      thread,      pullRequest: {        state: "merged" as const,        updatedAt: DateTime.formatIso(at(-60 * 60 * 1_000)),      },      nowMs: NOW_MS,      autoSettleAfterDays: null,      autoSettleOnMerge: true,    };    expect(resolveAutoSettlementAt(input)).toEqual(at(-90 * 60 * 1_000));    expect(resolveAutoSettlementAt({ ...input, autoSettleOnMerge: false })).toBeNull();    expect(      resolveAutoSettlementAt({        ...input,        pullRequest: { state: "merged", updatedAt: DateTime.formatIso(at(-3 * 60 * 60 * 1_000)) },      }),    ).toBeNull();    // A thread without messages still has a stable timestamp when its PR closes.    expect(      resolveAutoSettlementAt({        ...input,        thread: shell(),        pullRequest: { state: "closed", updatedAt: DateTime.formatIso(at(-1)) },        autoSettleOnMerge: false,      }),    ).toEqual(shell().createdAt);  });   it("settles inactive threads even when their pull request remains open", () => {    const input = {      thread: shell({ latestUserMessageAt: at(-30 * DAY_MS) }),      pullRequest: { state: "open" as const, updatedAt: DateTime.formatIso(at(0)) },      nowMs: NOW_MS,      autoSettleAfterDays: 2,      autoSettleOnMerge: true,    };    expect(resolveAutoSettlementAt(input)).toEqual(at(-30 * DAY_MS));    expect(resolveAutoSettlementAt({ ...input, autoSettleAfterDays: null })).toBeNull();    expect(      resolveAutoSettlementAt({        ...input,        thread: shell({ latestUserMessageAt: at(-DAY_MS) }),      }),    ).toBeNull();  });}); const NOW = "2026-08-28T12:00:00.000Z";const PROJECT_ID = ProjectId.make("settlement-project"); type AutoSettleCommand = Extract<OrchestrationV2Command, { readonly type: "thread.auto-settle" }>; const testCrypto = Crypto.make({  randomBytes: (size) => new Uint8Array(size).fill(1),  digest: (_algorithm, data) => Effect.succeed(data),}); function makeProject(  id: ProjectId = PROJECT_ID,  workspaceRoot = "/workspace/project",): OrchestrationProjectShell {  return {    id,    title: `Project ${id}`,    workspaceRoot,    defaultModelSelection: null,    scripts: [],    createdAt: "2026-08-01T00:00:00.000Z",    updatedAt: NOW,  };} function makeThread(  id: string,  overrides: Partial<OrchestrationV2ThreadShell> = {},): OrchestrationV2ThreadShell {  return shell({    id: ThreadId.make(id),    projectId: PROJECT_ID,    title: id,    createdAt: DateTime.makeUnsafe("2026-08-01T00:00:00.000Z"),    updatedAt: DateTime.makeUnsafe("2026-08-20T00:00:00.000Z"),    latestUserMessageAt: DateTime.makeUnsafe("2026-08-20T00:00:00.000Z"),    ...overrides,  });} function makeSnapshot(  threads: ReadonlyArray<OrchestrationV2ThreadShell>,  projects: ReadonlyArray<OrchestrationProjectShell> = [makeProject()],): OrchestrationV2ShellSnapshot {  return {    schemaVersion: 1,    snapshotSequence: 1,    projects,    threads,    archivedThreads: [],  };} function makePullRequestSummary(input: {  readonly projectId: ProjectId;  readonly repository: string;  readonly number: number;  readonly state: "open" | "closed" | "merged";  readonly updatedAt?: string;}): PullRequestSummary {  return {    provider: "github",    projectId: input.projectId,    repository: input.repository,    number: input.number,    title: "Pull request",    url: `https://example.test/${input.repository}/pull/${input.number}`,    state: input.state,    headBranch: "feature",    baseBranch: "main",    updatedAt: input.updatedAt ?? NOW,  };} interface HarnessOptions {  readonly snapshot: OrchestrationV2ShellSnapshot;  readonly settings?: ServerSettings;  readonly branchPullRequest?: GitManager["Service"]["branchPullRequest"];  readonly pullRequestSummary?: PullRequestService["Service"]["summary"];  readonly existingWorktreePaths?: ReadonlyArray<string>;  readonly onDispatch?: (command: AutoSettleCommand) => Effect.Effect<void>;} const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: HarnessOptions) {  const activation = yield* Deferred.make<void>();  const snapshots = yield* Ref.make(options.snapshot);  const snapshotReadCount = yield* Ref.make(0);  const snapshotReads = yield* Queue.unbounded<number>();  const settings = yield* Ref.make(options.settings ?? DEFAULT_SERVER_SETTINGS);  const settingsChanges = yield* PubSub.unbounded<ServerSettings>();  const mergedPullRequests = yield* PubSub.unbounded<PullRequestMergeEvent>();  const commands = yield* Ref.make<ReadonlyArray<AutoSettleCommand>>([]);  const branchCalls = yield* Ref.make<    ReadonlyArray<{ readonly cwd: string; readonly branch: string }>  >([]);  const summaryCalls = yield* Ref.make<    ReadonlyArray<{      readonly projectId: ProjectId;      readonly repository: string;      readonly number: number;    }>  >([]);  const summaryRecovery = yield* Ref.make<ReadonlyArray<boolean | undefined>>([]);  const invalidatedCwds = yield* Ref.make<ReadonlyArray<string>>([]);   const updateSettings = (patch: ServerSettingsPatch) =>    Effect.gen(function* () {      const next = applyServerSettingsPatch(yield* Ref.get(settings), patch);      yield* Ref.set(settings, next);      yield* PubSub.publish(settingsChanges, next);      return next;    });   const branchPullRequest: GitManager["Service"]["branchPullRequest"] = (input) =>    Ref.update(branchCalls, (calls) => [...calls, input]).pipe(      Effect.andThen(options.branchPullRequest?.(input) ?? Effect.succeed(null)),    );  const pullRequestSummary: PullRequestService["Service"]["summary"] = (input, readOptions) =>    Effect.gen(function* () {      yield* Ref.update(summaryCalls, (calls) => [...calls, input]);      yield* Ref.update(summaryRecovery, (values) => [        ...values,        readOptions?.recoverTransientFailure,      ]);      return yield* (        options.pullRequestSummary?.(input, readOptions) ??          Effect.succeed(            makePullRequestSummary({              ...input,              state: "open",            }),          )      );    });   const dispatch: OrchestratorV2Shape["dispatch"] = (command) => {    if (command.type !== "thread.auto-settle") {      return Effect.die(new Error(`Unexpected command: ${command.type}`));    }    return Ref.update(commands, (recorded) => [...recorded, command]).pipe(      Effect.andThen(options.onDispatch?.(command) ?? Effect.void),      Effect.as({ sequence: 1, storedEvents: [] }),    );  };   const serverSettings = ServerSettingsService.of({    start: Effect.void,    ready: Effect.void,    getSettings: Ref.get(settings),    updateSettings,    streamChanges: Stream.fromPubSub(settingsChanges),    subscribeChanges: PubSub.subscribe(settingsChanges).pipe(      Effect.map((subscription) => Stream.fromSubscription(subscription)),    ),  });   const dependencies = Layer.mergeAll(    Layer.mock(ProjectionSnapshotQuery)({      getProjectShellsWithoutEnrichment: () =>        Ref.get(snapshots).pipe(Effect.map((snapshot) => snapshot.projects)),    }),    Layer.mock(ProjectionStoreV2)({      getSettlementCandidates: () =>        Ref.updateAndGet(snapshotReadCount, (count) => count + 1).pipe(          Effect.tap((count) => Queue.offer(snapshotReads, count)),          Effect.andThen(Ref.get(snapshots)),          Effect.map((snapshot) => snapshot.threads),        ),    }),    Layer.mock(OrchestratorV2)({      dispatch,    }),    Layer.mock(GitManager)({      branchPullRequest,      invalidateStatus: (cwd) => Ref.update(invalidatedCwds, (cwds) => [...cwds, cwd]),    }),    Layer.mock(PullRequestService)({      summary: pullRequestSummary,      subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(        Effect.map((subscription) => Stream.fromSubscription(subscription)),      ),    }),    Layer.succeed(ServerSettingsService, serverSettings),    Layer.succeed(ServerActivation, Deferred.await(activation)),    Layer.succeed(Crypto.Crypto, testCrypto),    FileSystem.layerNoop({      exists: (path) => Effect.succeed(options.existingWorktreePaths?.includes(path) ?? false),    }),  );   return {    activation,    snapshots,    snapshotReadCount,    snapshotReads,    commands,    branchCalls,    summaryCalls,    summaryRecovery,    invalidatedCwds,    updateSettings,    publishMerge: PubSub.publish(mergedPullRequests, {      projectId: PROJECT_ID,      repository: "owner/repository",      number: 42,      mergedAt: NOW,    }),    layer: ThreadSettlementService.layer.pipe(Layer.provide(dependencies)),  };}); const startHarness = Effect.fn("startThreadSettlementHarness")(function* (  reactor: ThreadSettlementService.ThreadSettlementServiceV2["Service"],  activation: Deferred.Deferred<void>,  snapshotReads: Queue.Queue<number>,) {  yield* reactor.start();  yield* Deferred.succeed(activation, undefined);

Lines 1–500 of 739. Load more