import { assert, it, vi } from "@effect/vitest";import {  MessageId,  NodeId,  ProviderDriverKind,  ProviderInstanceId,  ProviderSessionId,  ProviderThreadId,  ProviderTurnId,  RunAttemptId,  RunId,  RuntimeRequestId,  ThreadId,  TurnItemId,  type OrchestrationV2ThreadProjection,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Ref from "effect/Ref"; import * as EffectWorker from "./EffectWorker.ts";import * as EffectOutbox from "./EffectOutbox.ts";import * as EventSink from "./EventSink.ts";import * as IdAllocator from "./IdAllocator.ts";import * as ProjectionStore from "./ProjectionStore.ts";import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";import * as ServerSettings from "../serverSettings.ts"; it.effect("drains durable effects before reporting recovery complete", () =>  Effect.gen(function* () {    const runs = yield* Ref.make(0);    const layer = ProviderRuntimeRecovery.layer.pipe(      Layer.provide(ServerSettings.layerTest()),      Layer.provide(        Layer.mergeAll(          Layer.mock(ProjectionStore.ProjectionStoreV2)({            getRecoveryThreadIds: () => Effect.succeed([]),          }),          Layer.mock(EventSink.EventSinkV2)({}),          IdAllocator.layer,          Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({            runRecoveryOnce: Ref.getAndUpdate(runs, (count) => count + 1).pipe(              Effect.map((count) => count < 2),            ),          }),          Layer.mock(EffectOutbox.EffectOutboxV2)({            reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),          }),        ),      ),    );    const summary = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService.pipe(      Effect.flatMap((recovery) => recovery.recover),      Effect.provide(layer),    );    assert.deepEqual(summary, {      terminalizedRuns: 0,      stoppedSessions: 0,      closedRequests: 0,      retiredEffects: 0,      requeuedEffects: 0,      executedEffects: 2,    });  }),); it.effect("expires orphaned runtime requests before command readiness", () => {  const threadId = ThreadId.make("thread_recovery_requests");  let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =    null;  const committed = vi.fn(    (input: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0]) => {      committedInput = input;      return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);    },  );  const projection = {    thread: { id: threadId },    runtimeRequests: [      {        id: RuntimeRequestId.make("request_orphaned"),        nodeId: NodeId.make("node_orphaned"),        status: "pending",        responseCapability: { type: "not_resumable", reason: "old process" },      },    ],    providerSessions: [],    providerThreads: [],    runs: [],    nodes: [],  } as unknown as OrchestrationV2ThreadProjection;  const layer = ProviderRuntimeRecovery.layer.pipe(    Layer.provide(ServerSettings.layerTest()),    Layer.provide(      Layer.mergeAll(        Layer.mock(ProjectionStore.ProjectionStoreV2)({          getRecoveryThreadIds: () => Effect.succeed([threadId]),          getThreadProjection: () => Effect.succeed(projection),        }),        Layer.mock(EventSink.EventSinkV2)({ commitCommand: committed }),        IdAllocator.layer,        Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({          runRecoveryOnce: Effect.succeed(false),        }),        Layer.mock(EffectOutbox.EffectOutboxV2)({          reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),        }),      ),    ),  );  return Effect.gen(function* () {    yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).recover;    const command = committedInput;    assert.isNotNull(command);    if (command === null) return;    assert.equal(command?.events[0]?.type, "runtime-request.updated");    if (command?.events[0]?.type === "runtime-request.updated") {      assert.equal(command.events[0].payload.status, "expired");      assert.equal(command.events[0].payload.responseCapability.type, "not_resumable");    }  }).pipe(Effect.provide(layer));}); it.effect("preserves async questions across startup and shutdown", () => {  const threadId = ThreadId.make("async-recovery-thread");  const nodeId = NodeId.make("async-recovery-node");  const projection = {    thread: { id: threadId },    runtimeRequests: [      {        id: RuntimeRequestId.make("async-recovery-request"),        nodeId,        status: "pending",        responseCapability: { type: "message" },      },    ],    providerSessions: [],    providerThreads: [],    runs: [],    nodes: [],    turnItems: [],  } as unknown as OrchestrationV2ThreadProjection;  const commitCommand = vi.fn(() => Effect.die("an async question needs no process-loss write"));  const layer = ProviderRuntimeRecovery.layer.pipe(    Layer.provide(ServerSettings.layerTest()),    Layer.provide(      Layer.mergeAll(        Layer.mock(ProjectionStore.ProjectionStoreV2)({          getRecoveryThreadIds: () => Effect.succeed([threadId]),          getThreadProjection: () => Effect.succeed(projection),        }),        Layer.mock(EventSink.EventSinkV2)({ commitCommand }),        IdAllocator.layer,        Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({          runRecoveryOnce: Effect.succeed(false),        }),        Layer.mock(EffectOutbox.EffectOutboxV2)({          reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),          cancelUnsettled: () => Effect.succeed([]),          signalCancellations: () => Effect.void,        }),      ),    ),  );  return Effect.gen(function* () {    const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;    assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0);    assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0);    assert.isFalse(commitCommand.mock.calls.length > 0);  }).pipe(Effect.provide(layer));}); it.effect("uses the same reconciliation path to cancel runtime requests during shutdown", () => {  const threadId = ThreadId.make("thread_shutdown_requests");  let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =    null;  const projection = {    thread: { id: threadId },    runtimeRequests: [      {        id: RuntimeRequestId.make("request_shutdown"),        nodeId: NodeId.make("node_shutdown"),        status: "pending",        responseCapability: { type: "live" },      },    ],    providerSessions: [],    providerThreads: [],    runs: [],    nodes: [],  } as unknown as OrchestrationV2ThreadProjection;  const layer = ProviderRuntimeRecovery.layer.pipe(    Layer.provide(ServerSettings.layerTest()),    Layer.provide(      Layer.mergeAll(        Layer.mock(ProjectionStore.ProjectionStoreV2)({          getRecoveryThreadIds: () => Effect.succeed([threadId]),          getThreadProjection: () => Effect.succeed(projection),        }),        Layer.mock(EventSink.EventSinkV2)({          commitCommand: (input) => {            committedInput = input;            return Effect.succeed({ committed: true, cancelledEffectCount: 1 } as never);          },        }),        IdAllocator.layer,        Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({          runRecoveryOnce: Effect.succeed(false),        }),        Layer.mock(EffectOutbox.EffectOutboxV2)({          reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),        }),      ),    ),  );   return Effect.gen(function* () {    const summary =      yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("shutdown");    assert.equal(summary.closedRequests, 1);    assert.equal(summary.retiredEffects, 1);    const requestEvent = committedInput?.events[0];    assert.equal(requestEvent?.type, "runtime-request.updated");    if (requestEvent?.type === "runtime-request.updated") {      assert.equal(requestEvent.payload.status, "cancelled");      assert.equal(requestEvent.payload.responseCapability.type, "not_resumable");      if (requestEvent.payload.responseCapability.type === "not_resumable") {        assert.match(requestEvent.payload.responseCapability.reason, /shut down/);      }    }  }).pipe(Effect.provide(layer));}); it.effect(  "preserves a replayable waiting run while cancelling its process-bound background work",  () => {    const threadId = ThreadId.make("thread_waiting_checkpoint");    const runId = RunId.make("run_waiting_checkpoint");    const providerThreadId = ProviderThreadId.make("provider_thread_waiting_checkpoint");    const providerInstanceId = ProviderInstanceId.make("claude");    const backgroundItemId = TurnItemId.make("turn_item_waiting_checkpoint");    let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =      null;    const projection = {      thread: { id: threadId },      runtimeRequests: [],      providerSessions: [],      providerThreads: [        {          id: providerThreadId,          driver: ProviderDriverKind.make("claude"),          providerInstanceId,          status: "idle",          pendingBackgroundTasks: [            { taskId: String(backgroundItemId), description: "Finish background task" },          ],        },      ],      runs: [{ id: runId, status: "waiting", providerInstanceId }],      attempts: [],      nodes: [],      subagents: [],      messages: [],      turnItems: [        {          id: backgroundItemId,          runId,          nodeId: null,          providerThreadId,          type: "command_execution",          status: "running",        },      ],    } as unknown as OrchestrationV2ThreadProjection;    const layer = ProviderRuntimeRecovery.layer.pipe(      Layer.provide(ServerSettings.layerTest()),      Layer.provide(        Layer.mergeAll(          Layer.mock(ProjectionStore.ProjectionStoreV2)({            getRecoveryThreadIds: () => Effect.succeed([threadId]),            getThreadProjection: () => Effect.succeed(projection),          }),          Layer.mock(EventSink.EventSinkV2)({            commitCommand: (input) => {              committedInput = input;              return Effect.succeed({                committed: true,                cancelledEffectCount: 0,              } as never);            },          }),          IdAllocator.layer,          Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({            runRecoveryOnce: Effect.succeed(false),          }),          Layer.mock(EffectOutbox.EffectOutboxV2)({            listByCommandId: () =>              Effect.succeed([                {                  request: { type: "checkpoint.capture", runId },                  status: "running",                },              ] as never),            reconcileAfterProcessLoss: Effect.succeed({ requeued: 1, cancelled: 0 }),          }),        ),      ),    );     return Effect.gen(function* () {      const summary =        yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");      assert.equal(summary.terminalizedRuns, 0);      assert.equal(summary.requeuedEffects, 1);       const command = committedInput;      assert.isNotNull(command);      if (command === null) return;      assert.isFalse(command.events.some((event) => event.type === "run.updated"));      assert.isTrue(        command.events.some(          (event) =>            event.type === "turn-item.updated" &&            event.payload.id === backgroundItemId &&            event.payload.status === "cancelled",        ),      );      assert.isTrue(        command.events.some(          (event) =>            event.type === "provider-thread.updated" &&            event.payload.id === providerThreadId &&            event.payload.pendingBackgroundTasks?.length === 0,        ),      );    }).pipe(Effect.provide(layer));  },); it.effect("cancels a stale waiting run when no checkpoint capture can finish it", () => {  const threadId = ThreadId.make("thread_stale_waiting");  const runId = RunId.make("run_stale_waiting");  let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =    null;  const projection = {    thread: { id: threadId },    runtimeRequests: [],    providerSessions: [],    providerThreads: [],    providerTurns: [],    runs: [      {        id: runId,        status: "waiting",        providerInstanceId: ProviderInstanceId.make("codex"),      },    ],    attempts: [],    nodes: [],    subagents: [],    messages: [],    turnItems: [],  } as unknown as OrchestrationV2ThreadProjection;  const layer = ProviderRuntimeRecovery.layer.pipe(    Layer.provide(ServerSettings.layerTest()),    Layer.provide(      Layer.mergeAll(        Layer.mock(ProjectionStore.ProjectionStoreV2)({          getRecoveryThreadIds: () => Effect.succeed([threadId]),          getThreadProjection: () => Effect.succeed(projection),        }),        Layer.mock(EventSink.EventSinkV2)({          commitCommand: (input) => {            committedInput = input;            return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);          },        }),        IdAllocator.layer,        Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({          runRecoveryOnce: Effect.succeed(false),        }),        Layer.mock(EffectOutbox.EffectOutboxV2)({          listByCommandId: () => Effect.succeed([]),          reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),        }),      ),    ),  );   return Effect.gen(function* () {    const summary =      yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");    assert.equal(summary.terminalizedRuns, 1);    const runEvent = committedInput?.events.find((event) => event.type === "run.updated");    assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.status : null, "cancelled");  }).pipe(Effect.provide(layer));}); it.effect("cancels accepted queued work instead of replaying it after restart", () => {  const threadId = ThreadId.make("thread_queued_restart");  const runId = RunId.make("run_queued_restart");  const attemptId = RunAttemptId.make("attempt_queued_restart");  const rootNodeId = NodeId.make("node_queued_restart");  let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =    null;  const projection = {    thread: { id: threadId },    runtimeRequests: [],    providerSessions: [],    providerThreads: [],    providerTurns: [],    runs: [      {        id: runId,        status: "queued",        queuePosition: 1,        providerInstanceId: ProviderInstanceId.make("codex"),      },    ],    attempts: [      {        id: attemptId,        runId,        rootNodeId,        status: "pending",      },    ],    nodes: [      {        id: rootNodeId,        runId,        status: "pending",      },    ],    subagents: [],    messages: [],    turnItems: [],  } as unknown as OrchestrationV2ThreadProjection;  const layer = ProviderRuntimeRecovery.layer.pipe(    Layer.provide(ServerSettings.layerTest()),    Layer.provide(      Layer.mergeAll(        Layer.mock(ProjectionStore.ProjectionStoreV2)({          getRecoveryThreadIds: () => Effect.succeed([threadId]),          getThreadProjection: () => Effect.succeed(projection),        }),        Layer.mock(EventSink.EventSinkV2)({          commitCommand: (input) => {            committedInput = input;            return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);          },        }),        IdAllocator.layer,        Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({          runRecoveryOnce: Effect.succeed(false),        }),        Layer.mock(EffectOutbox.EffectOutboxV2)({          reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),        }),      ),    ),  );   return Effect.gen(function* () {    const summary =      yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");    assert.equal(summary.terminalizedRuns, 1);    const command = committedInput;    assert.isNotNull(command);    if (command === null) return;    const runEvent = command.events.find((event) => event.type === "run.updated");    const attemptEvent = command.events.find((event) => event.type === "run-attempt.updated");    const nodeEvent = command.events.find((event) => event.type === "node.updated");    assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.status : null, "cancelled");    assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queuePosition : 1, null);    assert.equal(      attemptEvent?.type === "run-attempt.updated" ? attemptEvent.payload.status : null,      "cancelled",    );    assert.equal(nodeEvent?.type === "node.updated" ? nodeEvent.payload.status : null, "cancelled");  }).pipe(Effect.provide(layer));}); it.effect(  "cancels the complete in-flight subtree and stops its persisted session without reopening it",  () => {    const threadId = ThreadId.make("thread_recovery_cancel");    const runId = RunId.make("run_recovery_cancel");    const attemptId = RunAttemptId.make("attempt_recovery_cancel");    const rootNodeId = NodeId.make("node_recovery_cancel");    const providerThreadId = ProviderThreadId.make("provider_thread_recovery_cancel");    const providerTurnId = ProviderTurnId.make("provider_turn_recovery_cancel");    const providerSessionId = ProviderSessionId.make("provider_session_recovery_cancel");    let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =      null;    const projection = {      thread: { id: threadId },      runtimeRequests: [],      providerSessions: [        {

Lines 1–500 of 1094. Load more