Parked session never resumes after an eve start restart: continuation POST is accepted, the run is re-enqueued, and nothing else ever happens (world-local, 0.27.13) #1450

// Standalone mock OpenAI-compatible provider: POST /v1/chat/completions.
// Default reply "REPLICA_OK"; echoes a CEDAR-#### marker found in the transcript when asked.
// Logs every request to MOCK_LOG (one JSON line per request) so the driver can tell
// whether the agent server ever called the provider.
import { createServer } from "node:http";
import { appendFileSync } from "node:fs";

const MODEL = "mock-model";
const LOG = process.env.MOCK_LOG;

function partText(content) {
  if (typeof content === "string") return content;
  if (Array.isArray(content)) return content.map((p) => (typeof p === "string" ? p : (p?.text ?? ""))).join(" ");
  return "";
}
const transcriptText = (messages) => messages.map((m) => partText(m?.content)).join("\n");
function lastUserText(messages) {
  for (let i = messages.length - 1; i >= 0; i--) if (messages[i]?.role === "user") return partText(messages[i].content);
  return "";
}
function chooseResponse(body) {
  const messages = Array.isArray(body?.messages) ? body.messages : [];
  const last = lastUserText(messages);
  if (/What code did I ask you to remember/i.test(last)) {
    const match = transcriptText(messages).match(/CEDAR-\d+/);
    return match ? match[0] : "MISSING_MARKER";
  }
  if (/Remember this code/i.test(last)) return "REMEMBERED";
  return "REPLICA_OK";
}
const completionJson = (text) => ({
  id: "chatcmpl-mock",
  object: "chat.completion",
  created: 1,
  model: MODEL,
  choices: [{ index: 0, message: { role: "assistant", content: text }, finish_reason: "stop" }],
  usage: { prompt_tokens: 8, completion_tokens: 4, total_tokens: 12 },
});
function streamChunks(text) {
  const base = { id: "chatcmpl-mock", object: "chat.completion.chunk", created: 1, model: MODEL };
  return [
    { ...base, choices: [{ index: 0, delta: { role: "assistant", content: text }, finish_reason: null }] },
    { ...base, choices: [{ index: 0, delta: {}, finish_reason: "stop" }] },
    { ...base, choices: [], usage: { prompt_tokens: 8, completion_tokens: 4, total_tokens: 12 } },
  ];
}
async function readBody(req) {
  const chunks = [];
  for await (const chunk of req) chunks.push(chunk);
  return Buffer.concat(chunks).toString("utf8");
}

export async function startMockOpenAiServer() {
  const requests = [];
  const server = createServer(async (req, res) => {
    if (req.method !== "POST" || req.url !== "/v1/chat/completions") {
      res.writeHead(404, { "content-type": "application/json" });
      res.end(JSON.stringify({ error: { message: `no route: ${req.method} ${req.url}` } }));
      return;
    }
    let body;
    try {
      body = JSON.parse(await readBody(req));
    } catch {
      res.writeHead(400, { "content-type": "application/json" });
      res.end(JSON.stringify({ error: { message: "invalid JSON body" } }));
      return;
    }
    requests.push(body);
    if (LOG) appendFileSync(LOG, JSON.stringify({ at: new Date().toISOString(), last: lastUserText(body.messages ?? []) }) + "\n");
    const text = chooseResponse(body);
    if (body?.stream) {
      res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-cache", connection: "keep-alive" });
      for (const chunk of streamChunks(text)) res.write(`data: ${JSON.stringify(chunk)}\n\n`);
      res.end("data: [DONE]\n\n");
    } else {
      res.writeHead(200, { "content-type": "application/json" });
      res.end(JSON.stringify(completionJson(text)));
    }
  });
  await new Promise((resolve, reject) => {
    server.once("error", reject);
    server.listen(0, "127.0.0.1", resolve);
  });
  const { port } = server.address();
  return {
    baseUrl: `http://127.0.0.1:${port}/v1`,
    requests,
    close: () => new Promise((resolve) => server.close(resolve)),
  };
}

// Standalone mode: `node mock-openai.mjs` prints its base URL and stays up.
if (import.meta.url === `file://${process.argv[1]}`) {
  const mock = await startMockOpenAiServer();
  console.log(mock.baseUrl);
}

// Minimal reproduction for the eve 0.27.13 session-resume wedge.
//
// Scenario: park a client session, restart the eve server, then continue the parked
// session with client.session(savedState).send(...). The POST is accepted (a continuation
// token comes back) but result() never resolves and the workflow data directory stays frozen.
//
// Run: MOCK_BASE_URL is wired up automatically; just `node repro.mjs` from this directory.
import { spawn } from "node:child_process";
import { createServer } from "node:net";
import { readdirSync, statSync, existsSync, rmSync } from "node:fs";
import { join, dirname } from "node:path";
import { fileURLToPath } from "node:url";
import { startMockOpenAiServer } from "./mock-openai.mjs";

const APP = dirname(fileURLToPath(import.meta.url));
const MARKER = "CEDAR-4729";
const HEALTH_TIMEOUT_MS = 90_000;
const TURN_TIMEOUT_MS = 60_000;
const WORKFLOW_DIR = join(APP, ".eve", ".workflow-data");

const logs = [];
const note = (line) => {
  logs.push(line);
  if (logs.length > 600) logs.shift();
  if (process.env.VERBOSE === "1") console.error(line);
};
const say = (line) => console.log(`[repro] ${line}`);

async function freePort() {
  return new Promise((resolve, reject) => {
    const srv = createServer();
    srv.once("error", reject);
    srv.listen(0, "127.0.0.1", () => {
      const { port } = srv.address();
      srv.close(() => resolve(port));
    });
  });
}

function run(cmd, args, env) {
  return new Promise((resolve, reject) => {
    const child = spawn(cmd, args, { cwd: APP, env, stdio: ["ignore", "pipe", "pipe"] });
    const capture = (buf) => String(buf).split("\n").filter(Boolean).forEach(note);
    child.stdout.on("data", capture);
    child.stderr.on("data", capture);
    child.on("error", reject);
    child.on("exit", (code) => (code === 0 ? resolve() : reject(new Error(`${cmd} ${args.join(" ")} exited with ${code}`))));
  });
}

function startEve(env, port) {
  const child = spawn(
    process.execPath,
    [join(APP, "node_modules/eve/bin/eve.js"), "start", "--host", "127.0.0.1", "--port", String(port)],
    { cwd: APP, env, detached: true, stdio: ["ignore", "pipe", "pipe"] },
  );
  const capture = (buf) => String(buf).split("\n").filter(Boolean).forEach(note);
  child.stdout.on("data", capture);
  child.stderr.on("data", capture);
  return child;
}

async function stopEve(child) {
  if (!child || child.exitCode !== null) return;
  const gone = new Promise((resolve) => child.once("exit", resolve));
  try {
    process.kill(-child.pid, "SIGTERM");
  } catch {
    return;
  }
  const timer = new Promise((resolve) => setTimeout(resolve, 15_000, "timeout"));
  if ((await Promise.race([gone, timer])) === "timeout") {
    try {
      process.kill(-child.pid, "SIGKILL");
    } catch {
      /* already dead */
    }
    await gone;
  }
}

async function waitForHealth(port, child) {
  const deadline = Date.now() + HEALTH_TIMEOUT_MS;
  while (Date.now() < deadline) {
    if (child.exitCode !== null) throw new Error(`eve exited with ${child.exitCode} before becoming healthy`);
    try {
      const res = await fetch(`http://127.0.0.1:${port}/`, { signal: AbortSignal.timeout(2000) });
      if (res.ok) return;
    } catch {
      /* not up yet */
    }
    await new Promise((r) => setTimeout(r, 250));
  }
  throw new Error(`eve did not become healthy within ${HEALTH_TIMEOUT_MS / 1000}s`);
}

// Recursive snapshot of .eve/.workflow-data: relative path -> mtimeMs.
function snapshotWorkflowData(dir = WORKFLOW_DIR, base = WORKFLOW_DIR, out = new Map()) {
  if (!existsSync(dir)) return out;
  for (const entry of readdirSync(dir, { withFileTypes: true })) {
    const full = join(dir, entry.name);
    if (entry.isDirectory()) snapshotWorkflowData(full, base, out);
    else out.set(full.slice(base.length + 1), statSync(full).mtimeMs);
  }
  return out;
}

function diffSnapshots(before, after) {
  const added = [...after.keys()].filter((k) => !before.has(k));
  const changed = [...after.keys()].filter((k) => before.has(k) && before.get(k) !== after.get(k));
  return { added, changed };
}

const accepted = { send: false, token: null };

async function turn(session, prompt, timeoutMs = TURN_TIMEOUT_MS) {
  let timer;
  accepted.send = false;
  accepted.token = null;
  try {
    const result = await Promise.race([
      session.send(prompt).then((r) => {
        accepted.send = true;
        accepted.token = session.state?.continuationToken ?? null;
        note(`[repro] send accepted (sessionId=${session.state?.sessionId ?? "new"}) token=${accepted.token}`);
        return r.result();
      }),
      new Promise((_, reject) => {
        timer = setTimeout(
          () => reject(new Error(`turn timed out after ${timeoutMs / 1000}s (POST accepted: ${accepted.send}, continuationToken: ${accepted.token})`)),
          timeoutMs,
        );
      }),
    ]);
    note(`[repro] turn done: status=${JSON.stringify(result?.status)}`);
    if (result.status === "failed" || !result.message) throw new Error(`turn failed: status=${result.status}`);
    return result.message.trim();
  } finally {
    clearTimeout(timer);
  }
}

async function main() {
  if (process.env.KEEP_STATE !== "1") rmSync(join(APP, ".eve"), { recursive: true, force: true });
  const mock = await startMockOpenAiServer();
  const port = await freePort();
  const env = { ...process.env, MOCK_BASE_URL: mock.baseUrl, PORT: String(port), NODE_ENV: "production" };
  let eve = null;
  let reproduced = false;
  try {
    say("build");
    await run(process.execPath, [join(APP, "node_modules/eve/bin/eve.js"), "build"], env);

    say(`start (port ${port})`);
    eve = startEve(env, port);
    await waitForHealth(port, eve);

    const { Client } = await import("eve/client");
    const client = new Client({ host: `http://127.0.0.1:${port}` });

    say("turn 1 on a fresh session");
    const session = client.session();
    const first = await turn(session, "Reply with a status word.");
    say(`  reply: ${JSON.stringify(first)}`);

    say("turn 2: seed the marker");
    const seeded = await turn(session, `Remember this code: ${MARKER}`);
    say(`  reply: ${JSON.stringify(seeded)}`);
    say("control: resume from saved state WITHOUT restarting (proves the saved state is usable)");
    const rehydrated = client.session(session.state);
    const preRestart = await turn(rehydrated, "Reply with a status word.");
    say(`  reply: ${JSON.stringify(preRestart)}`);

    const savedState = rehydrated.state;
    say(`  saved session state: ${JSON.stringify(savedState)}`);

    say("restart eve (SIGTERM to the process group, then start again)");
    await stopEve(eve);
    eve = startEve(env, port);
    await waitForHealth(port, eve);

    say("control: a fresh session after the restart");
    const fresh = await turn(client.session(), "Reply with a status word.");
    say(`  reply: ${JSON.stringify(fresh)} (fresh sessions work)`);

    // Let the control run's writes settle so the diff below only reflects the resume attempt.
    await new Promise((r) => setTimeout(r, 5000));
    const mockBefore = mock.requests.length;
    const before = snapshotWorkflowData();
    say(`.workflow-data before resume: ${before.size} files`);

    say("resume the parked session across the restart");
    let resumeError = null;
    let resumed = null;
    try {
      resumed = await turn(client.session(savedState), "What code did I ask you to remember? Reply with the code only.");
    } catch (err) {
      resumeError = err;
    }
    const after = snapshotWorkflowData();
    const { added, changed } = diffSnapshots(before, after);
    const mockCalls = mock.requests.length - mockBefore;

    say("--- resume result ---");
    if (resumeError) {
      reproduced = true;
      say(`  REPRODUCED: ${resumeError.message}`);
    } else {
      say(`  resume returned ${JSON.stringify(resumed)} (expected ${MARKER})`);
      if (resumed !== MARKER) reproduced = true;
    }
    say(`  provider calls during resume: ${mockCalls}`);
    const parked = savedState.sessionId;
    const forParked = (list) => list.filter((f) => f.includes(parked));
    say(`  .workflow-data new files: ${added.length}${added.length ? ` -> ${added.slice(0, 20).join(", ")}` : ""}`);
    say(`  .workflow-data touched files: ${changed.length}${changed.length ? ` -> ${changed.slice(0, 20).join(", ")}` : ""}`);
    say(`  ...of those, belonging to the parked run ${parked}: ${forParked(added).length} new, ${forParked(changed).length} touched`);
    for (const f of forParked(added)) say(`      new: ${f}`);
    for (const f of forParked(changed)) say(`      touched: ${f}`);
    say(`  total provider calls this run: ${mock.requests.length}`);
    say(reproduced ? "VERDICT: session-resume wedge REPRODUCED on bare eve" : "VERDICT: resume worked, NOT reproduced");
  } catch (err) {
    console.error(`[repro] FAILED: ${err?.message ?? err}`);
    process.exitCode = 1;
  } finally {
    console.error("--- last server output ---");
    for (const line of logs.slice(-80)) console.error(line);
    await stopEve(eve);
    await mock.close();
  }
  if (reproduced) process.exitCode = 2;
}

await main();