openrig/packages/daemon/test/environment-lifecycle.test.ts

import { afterEach, describe, expect, it, vi } from "vitest";
import { mkdtempSync, realpathSync, writeFileSync, readFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { EnvironmentLifecycle } from "../src/domain/environment-lifecycle.js";
import { QueueRepository } from "../src/domain/queue-repository.js";
import { EventBus } from "../src/domain/event-bus.js";
import { OutboxHandler } from "../src/domain/outbox-handler.js";
import { EnvironmentPlatform, sha256 } from "../src/domain/environment-platform.js";
import { createFullTestDb, createTestApp } from "./helpers/test-app.js";

const cleanup: Array<() => void> = [];
afterEach(() => { for (const close of cleanup.splice(0).reverse()) close(); });
async function fixture(attached = false, automatic = true, beforeGeneration = false) {
  const root = realpathSync(mkdtempSync(join(tmpdir(), "environment-lifecycle-"))); cleanup.push(() => rmSync(root, { recursive: true, force: true }));
  const db = createFullTestDb(); cleanup.push(() => db.close());
  const resource = { id: "compute-one", name: "worker-one", workspace: "workspace", phase: "Ready", operationId: "create-one" };
  let present = attached, running = true, unavailable = false, loseStop = false, stopCalls = 0, deleteCalls = 0, creates = 0;
  let dependents = { workload: false, supervisor: false, volumes: false };
  const files = new Map<string, Buffer>();
  const native = { nativeId: "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee", nativeType: "community:openshell-claude", generation: "generation-one" };
  const history = Buffer.from(JSON.stringify({ sessionId: native.nativeId, type: "user", message: { content: "retained context" } }) + "\n");
  files.set(`/sandbox/openrig/native/projects/-sandbox-openrig-work/${native.nativeId}.jsonl`, history);
  const driver = { version: 1 as const,
    list: async () => { if (unavailable) throw new Error("provider unavailable"); return { complete: true, resources: present ? [{ ...resource }] : [] }; },
    create: async (input: { operationId: string }) => { creates++; present = true; resource.operationId = input.operationId; return { ...resource }; },
    inspect: async () => { if (unavailable || !present) throw new Error("lookup unavailable"); return { ...resource }; },
    upload: async (_b: unknown, source: string, destination: string) => { files.set(destination, readFileSync(source)); },
    read: async (_b: unknown, destination: string) => { const data = files.get(destination); if (!data) throw new Error("missing"); return data; },
    download: async (_b: unknown, source: string, destination: string) => { writeFileSync(destination, files.get(source)!); },
    nativeIdentity: async () => ({ ...native, running }),
    stopNative: async () => { stopCalls++; running = false; if (loseStop) throw new Error("reply lost"); },
    stopCompute: async () => { stopCalls++; running = false; resource.phase = "Stopped"; if (loseStop) throw new Error("reply lost"); },
    deleteEnvironment: async () => { deleteCalls++; present = false; },
    cleanupState: async () => ({ ...dependents }),
    checkpointFiles: async () => [{ path: `native/projects/-sandbox-openrig-work/${native.nativeId}.jsonl`, bytes: history.length, sha256: (await import("../src/domain/environment-platform.js")).sha256(history) }],
  };
  const queueEvents = new EventBus(db), queue = new QueueRepository(db, queueEvents); queue.attachOutbox(new OutboxHandler(db));
  const f = createTestApp(db, { appDeps: { eventBus: automatic ? queueEvents : new EventBus(db), queueRepo: queue, environmentDriver: driver, environmentArtifactRoot: join(root, "artifacts"), environmentStateRoot: join(root, "retained") } as never });
  const rig = f.rigRepo.createRig("lifecycle"), node = f.rigRepo.addNode(rig.id, "worker", { runtime: "openshell-claude", cwd: root });
  const request = async (path: string, body?: unknown) => { const res = await f.app.request(`/api/environments${path}`, { method: body ? "POST" : "GET", headers: { "content-type": "application/json", "x-openrig-session": "coordinator@lifecycle" }, ...(body ? { body: JSON.stringify(body) } : {}) }); return { status: res.status, body: await res.json() as Record<string, any> }; };
  const policy = join(root, "policy.yaml"); writeFileSync(policy, "version: 1\n");
  const input = { version: 1, bindingId: "binding-one", operationId: "create-one", nodeId: beforeGeneration ? undefined : node.id, seat: "worker@lifecycle", baseRuntime: "claude-code", guestRoot: "/sandbox/openrig", guestBaseUrl: "http://guest.test:7433", placement: { gateway: "gateway", workspace: "workspace", name: "worker-one", image: `image@sha256:${"a".repeat(64)}`, provider: "provider", policy } };
  const created = await request(attached ? "/attach" : "/create", { ...input, ...(attached ? { resourceId: resource.id } : {}) }); expect(created.status, JSON.stringify(created.body)).toBeLessThan(300);
  if (!beforeGeneration) {
  db.prepare("UPDATE environment_bindings SET generation='generation-one' WHERE binding_id='binding-one'").run();
  const receipt = { version: 1, bindingId: "binding-one", nodeId: node.id, resourceId: resource.id, ...native, activityGrantId: "grant-one" };
  db.prepare("INSERT INTO environment_operations(operation_id,binding_id,kind,request_json,state,receipt_json) VALUES ('launch-generation-one','binding-one','launch',?,'confirmed',?)").run(JSON.stringify(receipt), JSON.stringify(receipt));
  }
  const effect = (name: string, operationId = "effect-one") => request("/binding-one/effect", { operationId, generation: "generation-one", effect: name });
  const lifecycle = new EnvironmentLifecycle(new EnvironmentPlatform(db, driver), queue, join(root, "retained"));
  return { root, db, driver, queue, lifecycle, request, effect, resource, native, files, history, capture: () => f.snapshotCapture.captureSnapshot(rig.id, "manual"), disappear: () => { present = false; }, counts: () => ({ stopCalls, deleteCalls, creates }), loseStop: () => { loseStop = true; }, unavailable: () => { unavailable = true; }, dirtyCleanup: () => { dependents.volumes = true; } };
}
describe("owned and attached environment effects", () => {
  async function failedProjection(attached = false) {
    const f = await fixture(attached, true, true), platform = new EnvironmentPlatform(f.db, f.driver);
    vi.spyOn(f.driver, "read").mockRejectedValueOnce(new Error("guest root read denied"));
    const source = join(f.root, "policy.yaml");
    await expect(platform.project("binding-one", { operationId: "projection-failed", generation: "not-established", files: [
      { path: "work/role.md", source, sha256: sha256(readFileSync(source)), required: true },
    ] })).rejects.toThrow("guest root read denied");
    expect(platform.get("binding-one")).toMatchObject({ generation: null, resource: { id: f.resource.id } });
    expect(platform.get("binding-one")!.nodeId).toBeUndefined();
    return { ...f, cleanup: () => f.request("/binding-one/effect", { operationId: "cleanup-failed-create", effect: "delete-owned", generation: null, resourceId: f.resource.id }) };
  }
  it("cleans the exact owned create after projection fails before node and generation binding", async () => {
    const f = await failedProjection(), result = await f.cleanup();
    expect(result.status, JSON.stringify(result.body)).toBe(200);
    expect(result.body).toMatchObject({ state: "confirmed", generation: null, resourceId: "compute-one", effect: "delete-owned" });
    expect(f.counts()).toMatchObject({ creates: 1, deleteCalls: 1, stopCalls: 0 });
    expect((await f.cleanup()).body).toEqual(result.body); expect(f.counts().deleteCalls).toBe(1);
  });
  it("reconciles uncertain pre-generation cleanup without another provider delete", async () => {
    const f = await failedProjection();
    let observe = false;
    vi.spyOn(f.driver, "deleteEnvironment").mockImplementationOnce(async () => { f.disappear(); throw new Error("lost delete reply"); });
    const original = f.driver.cleanupState;
    vi.spyOn(f.driver, "cleanupState").mockImplementation(async () => { if (!observe) throw new Error("engine unavailable"); return original(); });
    expect((await f.cleanup()).body.state).toBe("unconfirmed");
    observe = true;
    expect((await f.cleanup()).body.state).toBe("confirmed");
    expect(f.driver.deleteEnvironment).toHaveBeenCalledTimes(1);
  });
  it("preserves an attached resource after projection fails before a generation exists", async () => {
    const f = await failedProjection(true), result = await f.cleanup();
    expect(result.status).toBe(409); expect(result.body.code).toBe("environment_attached_preserved");
    expect(f.counts()).toMatchObject({ creates: 0, deleteCalls: 0, stopCalls: 0 });
  });

  it("requires the exact resource for pre-generation cleanup and never infers a native generation", async () => {
    const f = await failedProjection();
    for (const body of [
      { effect: "delete-owned", generation: null },
      { effect: "delete-owned", generation: null, resourceId: "another-resource" },
      { effect: "stop-compute", generation: null, resourceId: f.resource.id },
    ]) {
      const result = await f.request("/binding-one/effect", { operationId: "bad-selection", ...body });
      expect(result.status).toBeGreaterThanOrEqual(400);
    }
    expect(f.counts()).toMatchObject({ deleteCalls: 0, stopCalls: 0 });
  });
  it("refuses generation-free cleanup once a generation exists", async () => {
    const f = await fixture();
    const result = await f.request("/binding-one/effect", { operationId: "bad-selection", effect: "delete-owned", generation: null, resourceId: f.resource.id });
    expect(result.status).toBe(409); expect(result.body.code).toBe("environment_generation_changed");
    expect(f.counts().deleteCalls).toBe(0);
  });
  it("does not dispatch pre-generation deletion if projection establishes a generation during observation", async () => {
    const f = await failedProjection(), list = f.driver.list;
    vi.spyOn(f.driver, "list").mockImplementationOnce(async () => {
      const result = await list();
      f.db.prepare("UPDATE environment_bindings SET generation='new-generation' WHERE binding_id='binding-one'").run();
      return result;
    });
    const result = await f.cleanup();
    expect(result.status).toBe(409); expect(result.body.code).toBe("environment_generation_changed");
    expect(f.counts().deleteCalls).toBe(0);
  });
  it("stops only the selected native and reconciles a lost acknowledgement without a second signal", async () => {
    const f = await fixture(); f.loseStop();
    const first = await f.effect("stop-native"); expect(first.body.state).toBe("confirmed");
    expect(first.body.effect).toBe("stop-native");
    const replay = await f.effect("stop-native"); expect(replay.body).toEqual(first.body); expect(f.counts().stopCalls).toBe(1);
    expect(f.resource.phase).toBe("Ready"); expect(f.counts().deleteCalls).toBe(0);
  });
  it("observes stopped compute independently from native and viewer state", async () => {
    const f = await fixture(); const result = await f.effect("stop-compute");
    expect(result.body).toMatchObject({ state: "confirmed", effect: "stop-compute", resourceId: "compute-one" });
    expect(f.resource.phase).toBe("Stopped"); expect(f.counts().deleteCalls).toBe(0);
  });
  it.each(["stop-native", "stop-compute", "delete-owned"])("preserves attached resources for %s", async effect => {
    const f = await fixture(true), result = await f.effect(effect);
    expect(result.status).toBe(409); expect(result.body.code).toBe("environment_attached_preserved");
    expect(f.counts()).toMatchObject({ stopCalls: 0, deleteCalls: 0, creates: 0 });
  });
  it("does not call provider absence complete until workload, supervisor and volumes are all observed absent", async () => {
    const f = await fixture(); f.dirtyCleanup();
    const result = await f.effect("delete-owned"); expect(result.body.state).toBe("unconfirmed");
    expect(result.body.remaining.volumes).toBe(true); expect(f.counts().deleteCalls).toBe(1);
    await f.effect("delete-owned"); expect(f.counts().deleteCalls).toBe(1);
  });
  it("keeps the original operation pending when a lookup fails and never infers absence", async () => {
    const f = await fixture(); f.unavailable();
    const result = await f.effect("delete-owned"); expect(result.body.state).toBe("unconfirmed");
    expect(f.counts().deleteCalls).toBe(0);
  });
  it("refuses the wrong generation before any provider mutation", async () => {
    const f = await fixture(), result = await f.request("/binding-one/effect", { operationId: "stale", generation: "older", effect: "stop-compute" });
    expect(result.status).toBe(409); expect(f.counts().stopCalls).toBe(0);
  });
});
describe("exact contained checkpoint and restore", () => {
  it("retains verified private history bytes and nonsecret exact native/compute references", async () => {
    const f = await fixture(), result = await f.request("/binding-one/checkpoint", { operationId: "checkpoint-one", generation: "generation-one" });
    expect(result.status).toBe(200); expect(result.body).toMatchObject({ version: 1, bindingId: "binding-one", nativeId: f.native.nativeId, nativeType: "community:openshell-claude", predecessorResourceId: "compute-one" });
    expect(result.body.files).toHaveLength(1); expect(readFileSync(result.body.files[0].source)).toEqual(f.history);
    expect(JSON.stringify(result.body)).not.toContain("retained context");
  });
  it("carries nonsecret environment/native/checkpoint references in the ordinary snapshot", async () => {
    const f = await fixture(); await f.request("/binding-one/checkpoint", { operationId: "checkpoint-one", generation: "generation-one" });
    const snapshot = f.capture(), references = Object.values(snapshot.data.environmentBindingsByNode ?? {});
    expect(references).toHaveLength(1);
    expect(references[0]).toMatchObject({ bindingId: "binding-one", resourceId: "compute-one", checkpointId: "checkpoint-one", nativeId: f.native.nativeId, nativeType: f.native.nativeType, nativeState: "/sandbox/openrig/native" });
    expect(JSON.stringify(snapshot.data.environmentBindingsByNode)).not.toContain("retained context");
  });
  it("rejects missing checkpoint state before creating successor compute", async () => {
    const f = await fixture(), result = await f.request("/binding-one/recreate", { operationId: "recreate-one", generation: "generation-two", checkpointId: "missing", resumeType: "community:openshell-claude", resumeToken: f.native.nativeId, placement: { ...f.resource, name: "worker-two" } });
    expect(result.status).toBe(409); expect(f.counts().creates).toBe(1);
  });
  it("recreates only after verified predecessor absence and carries the exact retained native id", async () => {
    const f = await fixture(), checkpoint = await f.request("/binding-one/checkpoint", { operationId: "checkpoint-one", generation: "generation-one" });
    expect(checkpoint.status).toBe(200); f.disappear();
    const policy = join(f.root, "policy.yaml");
    // The provider fixture returns a different immutable ID on explicit successor creation.
    f.resource.id = "compute-two"; f.resource.name = "worker-two";
    const result = await f.request("/binding-one/recreate", { operationId: "recreate-one", generation: "generation-one", checkpointId: "checkpoint-one", resumeType: "community:openshell-claude", resumeToken: f.native.nativeId,
      placement: { gateway: "gateway", workspace: "workspace", name: "worker-two", image: `image@sha256:${"a".repeat(64)}`, provider: "new-provider", policy } });
    expect(result.status).toBe(200); expect(result.body).toMatchObject({ state: "confirmed", predecessorResourceId: "compute-one", successorResourceId: "compute-two", nativeId: f.native.nativeId });
  });
  it("refuses changed private history before any successor create", async () => {
    const f = await fixture(), checkpoint = await f.request("/binding-one/checkpoint", { operationId: "checkpoint-one", generation: "generation-one" });
    expect(checkpoint.status).toBe(200); writeFileSync(checkpoint.body.files[0].source, "changed history"); f.disappear();
    const result = await f.request("/binding-one/recreate", { operationId: "recreate-one", generation: "generation-one", checkpointId: "checkpoint-one", resumeType: "community:openshell-claude", resumeToken: f.native.nativeId, placement: {} });
    expect(result.status).toBe(409); expect(f.counts().creates).toBe(1);
  });
  it("refuses another native type without starting a fresh conversation", async () => {
    const f = await fixture(), result = await f.request("/binding-one/recreate", { operationId: "recreate-one", generation: "generation-two", checkpointId: "missing", resumeType: "codex_id", resumeToken: "foreign" });
    expect(result.status).toBe(409); expect(f.counts().creates).toBe(1);
  });
});

describe("finite completion is selected work plus a verified actual export", () => {
  async function selected(automatic = true) {
    const f = await fixture(false, automatic);
    await f.queue.create({ qitemId: "task-one", sourceSession: "coordinator@lifecycle", destinationSession: "worker@lifecycle", body: "Produce the report", nudge: false });
    f.lifecycle.selectCompletion("binding-one", { operationId: "completion-one", generation: "generation-one", taskId: "task-one", exportOperationId: "export-one", effect: "stop-native" });
    const publish = () => {
      const path = join(f.root, "verified-report.json"), bytes = Buffer.from("actual report bytes"); writeFileSync(path, bytes);
      const receipt = { bindingId: "binding-one", generation: "generation-one", taskId: "task-one", path, bytes: bytes.length, sha256: sha256(bytes) };
      f.db.prepare("INSERT INTO environment_operations(operation_id,binding_id,kind,request_json,state,receipt_json) VALUES ('export-one','binding-one','export','{}','confirmed',?)").run(JSON.stringify(receipt));
      return path;
    };
    const done = () => f.queue.update({ qitemId: "task-one", actorSession: "worker@lifecycle", state: "done", closureReason: "no-follow-on", transitionNote: "Verified result complete" });
    return { ...f, publish, done };
  }
  it("does not stop on an empty/unsettled queue or a pending export", async () => {
    const f = await selected(); await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(0);
    f.done(); await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(0);
    f.publish(); await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(1);
    await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(1);
  });
  it("the actual queue event schedules selected completion without another manual check", async () => {
    const f = await selected(); f.publish(); f.done();
    for (let turn = 0; turn < 20 && !f.counts().stopCalls; turn++) await new Promise(resolve => setTimeout(resolve, 5));
    expect(f.counts().stopCalls).toBe(1);
  });
  it("another seat's same-state note does not undo a settled selected task", async () => {
    const f = await selected(); f.done();
    f.queue.update({ qitemId: "task-one", actorSession: "observer@lifecycle", transitionNote: "Result noted" });
    f.publish(); await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(1);
  });
  it("keeps failed work and another task's export from triggering completion", async () => {
    const f = await selected(); f.publish();
    f.queue.update({ qitemId: "task-one", actorSession: "worker@lifecycle", state: "failed", transitionNote: "Task failed" });
    await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(0);
  });
  it("preserves execution when the returned artifact changed after its receipt", async () => {
    const f = await selected(), path = f.publish(); f.done(); writeFileSync(path, "changed host artifact");
    await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(0);
  });
  it("does not stop a successor generation for an old task's successful close", async () => {
    const f = await selected(); f.publish(); f.done();
    f.db.prepare("UPDATE environment_bindings SET generation='generation-two' WHERE binding_id='binding-one'").run();
    await f.lifecycle.checkCompletions(); expect(f.counts().stopCalls).toBe(0);
  });
  it.each(["stop-native", "stop-compute"] as const)("rechecks selected completion %s after awaited provider observation", async effect => {
    for (const change of ["task", "artifact", "generation"] as const) {
      const f = await selected(false), artifact = f.publish();
      f.db.prepare("UPDATE environment_operations SET request_json=json_set(request_json,'$.effect',?) WHERE operation_id='completion-one'").run(effect);
      f.done();
      let enter!: () => void, release!: () => void;
      const entered = new Promise<void>(resolve => { enter = resolve; }), gate = new Promise<void>(resolve => { release = resolve; });
      const inspect = f.driver.inspect;
      f.driver.inspect = async () => { enter(); await gate; return inspect(); };
      const attempt = f.lifecycle.checkCompletions(); await entered;
      if (change === "task") f.queue.update({ qitemId: "task-one", actorSession: "worker@lifecycle", state: "in-progress", reopen: true, transitionNote: "Reopen selected work" });
      if (change === "artifact") writeFileSync(artifact, "changed while observing provider");
      if (change === "generation") f.db.prepare("UPDATE environment_bindings SET generation='generation-two' WHERE binding_id='binding-one'").run();
      release(); await attempt; expect(f.counts().stopCalls, change).toBe(0);
      const op = f.db.prepare("SELECT state FROM environment_operations WHERE operation_id LIKE 'completion-stop-%'").get() as { state: string } | undefined;
      expect(op?.state, change).not.toBe("requested");
    }
  });
  it("bounds pending completion observations and retains an inspectable last attempt", async () => {
    const f = await selected(false); f.publish(); f.done();
    let clock = Date.now(); const now = vi.spyOn(Date, "now").mockImplementation(() => clock); cleanup.push(() => now.mockRestore());
    let inspections = 0, signals = 0;
    const inspect = f.driver.inspect;
    f.driver.inspect = async () => { inspections++; return inspect(); };
    f.driver.stopNative = async () => { signals++; }; // Native remains present after its one TERM.
    await f.lifecycle.checkCompletions(); const first = inspections;
    for (let event = 0; event < 5; event++) await f.lifecycle.checkCompletions();
    expect(inspections).toBe(first); expect(signals).toBe(1);
    const op = f.db.prepare("SELECT state,receipt_json FROM environment_operations WHERE operation_id='completion-one'").get() as { state: string; receipt_json: string };
    expect(op.state).toBe("requested"); expect(JSON.parse(op.receipt_json)).toMatchObject({ state: "pending", lastAttemptAt: clock });
    clock += 60_001; await f.lifecycle.checkCompletions(); expect(inspections).toBeGreaterThan(first); expect(signals).toBe(1);
  });

});

it.each(["provider-list", "cleanup-inventory"])("reports the actual pre-dispatch stage and error without claiming a delete: %s", async stage => {
  const f = await fixture(); const error = Object.assign(new Error("fixture engine connection refused"), { code: "ECONNREFUSED" });
  if (stage === "provider-list") vi.spyOn(f.driver, "list").mockRejectedValue(error);
  else Object.assign(f.driver, { captureCleanup: async () => { throw error; } });
  const request = { effect: "delete-owned", operationId: "delete-before-dispatch", generation: "generation-one", engineSocket: "/fixture/engine.sock" };
  for (let observation = 0; observation < 2; observation++) {
    const result = await f.request("/binding-one/effect", request);
    expect(result.body).toMatchObject({ state: "unconfirmed", stage, dispatched: false, error: { code: "ECONNREFUSED", message: "fixture engine connection refused" } });
    expect(f.counts().deleteCalls).toBe(0);
    const row = f.db.prepare("SELECT state FROM environment_operations WHERE operation_id=?").get(request.operationId) as { state: string };
    expect(row.state).toBe("intent");
  }
});

it("keeps an uncertain delete requested and reconciles the same operation without another mutation", async () => {
  const f = await fixture(); const deletion = vi.spyOn(f.driver, "deleteEnvironment").mockRejectedValue(new Error("fixture reply lost at delete boundary"));
  const request = { effect: "delete-owned", operationId: "delete-uncertain", generation: "generation-one" };
  const first = await f.request("/binding-one/effect", request);
  expect(first.body).toMatchObject({ state: "unconfirmed", stage: "delete-dispatch", dispatched: "unknown", error: { message: "fixture reply lost at delete boundary" } });
  expect((f.db.prepare("SELECT state FROM environment_operations WHERE operation_id=?").get(request.operationId) as { state: string }).state).toBe("requested");
  await f.request("/binding-one/effect", request);
  expect(deletion).toHaveBeenCalledTimes(1);
});