import assert from "node:assert/strict";
import { readFile } from "node:fs/promises";
import http from "node:http";
import test from "node:test";
import { setTimeout as delay } from "node:timers/promises";
import { fileURLToPath } from "node:url";

import { MeerkatRuntime, Session, isKnownEvent, KNOWN_AGENT_EVENT_TYPES } from "@rkat/web";
import * as rawWasm from "@rkat/web/wasm/meerkat_web_runtime.js";

const WEB_PACKAGE_JSON_URL = new URL("../package.json", import.meta.url);
const { version: CURRENT_WASM_VERSION } = JSON.parse(
  await readFile(WEB_PACKAGE_JSON_URL, "utf8"),
);

async function makeNodeCompatibleWasmModule() {
  const wasmUrl = import.meta.resolve("@rkat/web/wasm/meerkat_web_runtime_bg.wasm");
  const wasmBytes = await readFile(fileURLToPath(wasmUrl));
  return {
    ...rawWasm,
    default: async () => rawWasm.default({ module_or_path: wasmBytes }),
  };
}

async function withAnthropicStreamServer(chunkCount, fn) {
  const sseBody = Array.from(
    { length: chunkCount },
    (_, index) =>
      `data: ${JSON.stringify({ type: "content_block_delta", delta: { type: "text_delta", text: String(index % 10) } })}\n\n`,
  ).join("") + `data: ${JSON.stringify({ type: "message_stop" })}\n\n`;

  const server = http.createServer((req, res) => {
    if (req.method === "POST" && req.url === "/v1/messages") {
      res.writeHead(200, { "content-type": "text/event-stream" });
      res.end(sseBody);
      return;
    }
    res.writeHead(404);
    res.end("not found");
  });

  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
  const { port } = server.address();
  try {
    return await fn(`http://127.0.0.1:${port}`);
  } finally {
    await new Promise((resolve, reject) =>
      server.close((error) => (error ? reject(error) : resolve())),
    );
  }
}

test("MeerkatRuntime drives direct-session lifecycle through shipped wasm exports", async () => {
  const wasm = await makeNodeCompatibleWasmModule();
  const runtime = await MeerkatRuntime.init(wasm, {
    anthropicApiKey: "sk-test",
    model: "claude-sonnet-4-5",
  });
  try {
    const session = runtime.createSession({
      model: "claude-sonnet-4-5",
      apiKey: "sk-test",
    });
    const staged = await session.appendSystemContext({
      text: "Remember the browser-side coordinator.",
      source: "web",
      idempotencyKey: "ctx-web-1",
    });

    assert.equal(staged.status, "staged");
    assert.equal(staged.handle, session.handle);

    const state = session.getState();
    assert.equal(session.sessionId, state.session_id);
    assert.equal(state.handle, session.handle);
    assert.equal(typeof state.session_id, "string");
    assert.notEqual(state.session_id, String(session.handle));
    assert.equal(state.model, "claude-sonnet-4-5");
    assert.equal(state.run_counter, undefined);

    session.destroy();
    const archived = session.getState();
    assert.equal(archived.session_id, state.session_id);
    assert.equal(archived.is_active, false);
    await assert.rejects(
      () => session.turn("stale handles must not restart archived sessions"),
      /SESSION_NOT_FOUND|session not found/,
    );
    assert.throws(() => session.isDestroyed, /deprecated/i);
  } finally {
    runtime.destroy();
  }
});

test("Session.destroy does not cache destroyed state when the underlying wasm destroy throws", async () => {
  let destroyAttempts = 0;
  let stateReads = 0;
  const session = new Session(
    7,
    async () => {
      throw new Error("SESSION_NOT_FOUND: session not found");
    },
    () => {
      stateReads += 1;
      return JSON.stringify({
        handle: 7,
        session_id: "sess_busy",
        mob_id: "",
        model: "claude-sonnet-4-5",
        usage: { input_tokens: 0, output_tokens: 0 },
        message_count: 0,
        is_active: true,
        last_assistant_text: null,
      });
    },
    () => {
      destroyAttempts += 1;
      if (destroyAttempts === 1) {
        throw new Error("SESSION_BUSY: turn still active");
      }
    },
    () => "[]",
    async () => JSON.stringify({ handle: 7, status: "staged" }),
  );

  assert.throws(() => session.destroy(), /SESSION_BUSY/);
  assert.equal(session.sessionId, "sess_busy");
  assert.equal(stateReads, 1);

  session.destroy();
  assert.equal(session.getState().session_id, "sess_busy");
  assert.equal(stateReads, 2);
  await assert.rejects(
    () => session.turn("after canonical destroy"),
    /SESSION_NOT_FOUND|session not found/,
  );
  assert.throws(() => session.isDestroyed, /deprecated/i);
});

test("Session.destroy treats runtime teardown as canonical absence without poisoning the handle", () => {
  let destroyAttempts = 0;
  let stateReads = 0;
  const session = new Session(
    8,
    async () => "{}",
    () => {
      stateReads += 1;
      throw new Error("not_initialized: runtime not initialized");
    },
    () => {
      destroyAttempts += 1;
      throw new Error("not_initialized: runtime not initialized");
    },
    () => "[]",
    async () => JSON.stringify({ handle: 8, status: "staged" }),
  );

  assert.doesNotThrow(() => session.destroy());
  assert.equal(destroyAttempts, 1);
  assert.doesNotThrow(() => session.destroy());
  assert.equal(destroyAttempts, 2);
  assert.throws(() => session.getState(), /not_initialized/i);
  assert.equal(stateReads, 1);
});

test("Session.destroy swallows typed not_initialized without caching lifecycle state", () => {
  let destroyAttempts = 0;
  const session = new Session(
    9,
    async () => "{}",
    () => JSON.stringify({
      handle: 9,
      session_id: "sess_runtime_code_gone",
      mob_id: "",
      model: "claude-sonnet-4-5",
      usage: { input_tokens: 0, output_tokens: 0 },
      message_count: 0,
      is_active: false,
      last_assistant_text: null,
    }),
    () => {
      destroyAttempts += 1;
      const error = new Error("runtime not initialized");
      error.code = "not_initialized";
      throw error;
    },
    () => "[]",
    () => JSON.stringify({ handle: 9, status: "staged" }),
  );

  assert.doesNotThrow(() => session.destroy());
  assert.equal(destroyAttempts, 1);
  assert.doesNotThrow(() => session.destroy());
  assert.equal(destroyAttempts, 2);
  assert.equal(session.getState().session_id, "sess_runtime_code_gone");
  assert.throws(() => session.isDestroyed, /deprecated/i);
});

test("Session observation remains a canonical wasm call after destroy", async () => {
  let pollAttempts = 0;
  let appendAttempts = 0;
  const session = new Session(
    10,
    async () => "{}",
    () => JSON.stringify({
      handle: 10,
      session_id: "sess_projection",
      mob_id: "",
      model: "claude-sonnet-4-5",
      usage: { input_tokens: 0, output_tokens: 0 },
      message_count: 0,
      is_active: false,
      last_assistant_text: null,
    }),
    () => {},
    () => {
      pollAttempts += 1;
      return JSON.stringify([{ type: "text_complete", text: "from wasm" }]);
    },
    async () => {
      appendAttempts += 1;
      throw new Error("SESSION_NOT_FOUND: session not found");
    },
  );

  session.destroy();

  assert.deepEqual(session.pollEvents(), [{ type: "text_complete", text: "from wasm" }]);
  assert.equal(pollAttempts, 1);
  await assert.rejects(
    () => session.appendSystemContext({ text: "after destroy" }),
    /SESSION_NOT_FOUND|session not found/,
  );
  assert.equal(appendAttempts, 1);
});

test("isKnownEvent recognizes the full current canonical event surface", () => {
  assert.equal(isKnownEvent({ type: "run_started", session_id: "s", prompt: "p" }), true);
  assert.equal(
    isKnownEvent({
      type: "hook_started",
      hook_id: "hook-1",
      point: "before_tool",
    }),
    true,
  );
  assert.equal(
    isKnownEvent({
      type: "budget_warning",
      budget_type: "time",
      used: 10,
      limit: 20,
      percent: 50,
    }),
    true,
  );
  assert.equal(
    isKnownEvent({
      type: "tool_config_changed",
      payload: {
        operation: "reload",
        target: "external",
        status: "applied",
        persisted: true,
        applied_at_turn: 4,
      },
    }),
    true,
  );
  assert.equal(
    isKnownEvent({
      type: "stream_truncated",
      reason: "backpressure",
    }),
    true,
  );
});

test("web known event types stay aligned with the canonical contracts artifact", async () => {
  const eventsJsonPath = new URL("../../../artifacts/schemas/events.json", import.meta.url);
  const raw = JSON.parse(await readFile(eventsJsonPath, "utf8"));
  const canonicalTypes = raw.WireEvent?.known_event_types;

  assert.ok(Array.isArray(canonicalTypes), "contracts events.json must expose known_event_types");
  assert.deepEqual(
    [...KNOWN_AGENT_EVENT_TYPES].sort(),
    [...canonicalTypes].sort(),
  );
});

test("MeerkatRuntime opens and closes a public mob subscription through the shipped package surface", async () => {
  let subscribedHandle;
  const wasm = {
    ...(await makeNodeCompatibleWasmModule()),
    async mob_subscribe_events(mobId) {
      const handle = await rawWasm.mob_subscribe_events(mobId);
      subscribedHandle = handle;
      return handle;
    },
  };
  const runtime = await MeerkatRuntime.init(wasm, {
    anthropicApiKey: "sk-test",
    model: "claude-sonnet-4-5",
  });
  try {
    const mob = await runtime.createMob({
      id: "mob-web-phase-0",
      profiles: {
        worker: {
          model: "claude-sonnet-4-5",
          tools: { comms: true },
        },
      },
    });

    const subscription = await mob.subscribeEvents();
    assert.deepEqual(subscription.poll(), []);
    subscription.close();
    assert.equal(subscription.isClosed, true);
    assert.notEqual(subscribedHandle, undefined);
    assert.throws(
      () => rawWasm.poll_subscription(subscribedHandle),
      /unknown subscription handle/i,
    );
  } finally {
    runtime.destroy();
  }
});

test("MeerkatRuntime exposes runtime-scoped tool registration on the instance surface", async () => {
  const calls = [];
  const wasm = {
    async default() {},
    runtime_version() {
      return CURRENT_WASM_VERSION;
    },
    init_runtime_from_config() {
      return JSON.stringify({ status: "initialized", model: "claude-sonnet-4-5", providers: ["anthropic"] });
    },
    register_tool_callback(name, description, schemaJson, callback) {
      calls.push(["callback", name, description, JSON.parse(schemaJson), typeof callback]);
    },
    register_js_tool(name, description, schemaJson) {
      calls.push(["fire_and_forget", name, description, JSON.parse(schemaJson)]);
    },
    clear_tool_callbacks() {
      calls.push(["clear"]);
    },
    destroy_runtime() {},
  };

  const runtime = await MeerkatRuntime.init(wasm, {
    anthropicApiKey: "sk-test",
    model: "claude-sonnet-4-5",
  });
  try {
    runtime.registerTool(
      "echo_browser",
      "Echo browser payloads",
      {
        type: "object",
        properties: {
          value: { type: "string" },
        },
        required: ["value"],
      },
      async (args) => ({ content: args, is_error: false }),
    );
    runtime.registerFireAndForgetTool(
      "notify_browser",
      "Emit a browser-side notification",
      {
        type: "object",
        properties: {
          message: { type: "string" },
        },
        required: ["message"],
      },
    );
    runtime.clearTools();

    assert.deepEqual(calls[0], [
      "callback",
      "echo_browser",
      "Echo browser payloads",
      {
        type: "object",
        properties: {
          value: { type: "string" },
        },
        required: ["value"],
      },
      "function",
    ]);
    assert.deepEqual(calls[1], [
      "fire_and_forget",
      "notify_browser",
      "Emit a browser-side notification",
      {
        type: "object",
        properties: {
          message: { type: "string" },
        },
        required: ["message"],
      },
    ]);
    assert.deepEqual(calls[2], ["clear"]);
  } finally {
    runtime.destroy();
  }
});

test("MeerkatRuntime surfaces lagged subscription signals through the shipped package surface", async () => {
  await withAnthropicStreamServer(305, async (anthropicBaseUrl) => {
    const wasm = await makeNodeCompatibleWasmModule();
    const runtime = await MeerkatRuntime.init(wasm, {
      anthropicApiKey: "sk-test",
      anthropicBaseUrl,
      model: "claude-sonnet-4-5",
    });
    try {
      const mob = await runtime.createMob({
        id: "mob-web-phase-4",
        profiles: {
          worker: {
            model: "claude-sonnet-4-5",
            external_addressable: true,
            tools: { comms: true },
          },
        },
      });

      const spawned = await mob.spawn([
        {
          profile: "worker",
          agent_identity: "worker-1",
          runtime_mode: "turn_driven",
        },
      ]);
      assert.equal(spawned[0].agent_identity, "worker-1");
      assert.ok(
        typeof spawned[0].member_ref === "string" && spawned[0].member_ref.length > 0,
        "spawn should return an opaque member_ref",
      );

      const subscription = await mob.member("worker-1").subscribe();
      await mob.member("worker-1").send("Trigger a long streamed response.");

      let items = [];
      for (let attempt = 0; attempt < 20; attempt += 1) {
        await delay(25);
        items = subscription.poll();
        if (items.some((item) => item.type === "lagged")) {
          break;
        }
      }

      const lagged = items.find((item) => item.type === "lagged");
      assert.ok(lagged, "expected a lagged signal from the shipped subscription path");
      assert.ok(lagged.skipped >= 1);
      const survivingEvent = items.find((item) => item.payload?.type === "text_delta");
      assert.ok(survivingEvent, "expected a surviving text_delta event after lag");
      subscription.close();
    } finally {
      runtime.destroy();
    }
  });
});

test("MeerkatRuntime forwards canonical mob status/helper methods through the wasm binding surface", async () => {
  const calls = [];
  const wasm = {
    async default() {},
    runtime_version() {
      return CURRENT_WASM_VERSION;
    },
    init_runtime_from_config() {
      return JSON.stringify({ status: "initialized", model: "claude-sonnet-4-5", providers: ["anthropic"] });
    },
    register_tool_callback() {},
    register_js_tool() {},
    clear_tool_callbacks() {},
    destroy_runtime() {},
    async mob_create() {
      return "mob-web-parity";
    },
    async mob_status(mobId) {
      return JSON.stringify({ mob_id: mobId, status: "running" });
    },
    async mob_list() {
      return JSON.stringify({ mobs: [] });
    },
    async mob_lifecycle(mobId, action) {
      calls.push(["lifecycle", mobId, action]);
    },
    async mob_events() {
      return JSON.stringify([]);
    },
    async mob_spawn(mobId, specsJson) {
      calls.push(["spawn", mobId, JSON.parse(specsJson)]);
      return JSON.stringify([
        {
          status: "spawned",
          result: {
            agent_identity: "worker-1",
            member_ref: "ref-worker-1",
          },
        },
      ]);
    },
    async mob_retire(mobId, agentIdentity) {
      calls.push(["retire", mobId, agentIdentity]);
    },
    async mob_wire(mobId, member, peer) {
      calls.push(["wire", mobId, member, peer]);
    },
    async mob_unwire(mobId, member, peer) {
      calls.push(["unwire", mobId, member, peer]);
    },
    async mob_wire_target(mobId, member, targetJson) {
      calls.push(["wire_target", mobId, member, JSON.parse(targetJson)]);
    },
    async mob_unwire_target(mobId, member, targetJson) {
      calls.push(["unwire_target", mobId, member, JSON.parse(targetJson)]);
    },
    async mob_list_members() {
      return JSON.stringify([
        {
          agent_identity: "worker-1",
          member_ref: "ref-worker-1",
          profile: "worker",
        },
      ]);
    },
    async mob_append_system_context(_mobId, agentIdentity) {
      return JSON.stringify({
        mob_id: "mob-web-parity",
        agent_identity: agentIdentity,
        status: "staged",
      });
    },
    async mob_member_send(_mobId, agentIdentity, requestJson) {
      calls.push(["member_send", agentIdentity, JSON.parse(requestJson)]);
      return JSON.stringify({
        mob_id: _mobId,
        agent_identity: agentIdentity,
        member_ref: `ref-${agentIdentity}`,
        handling_mode: "queue",
      });
    },
    async mob_member_status(_mobId, agentIdentity) {
      return JSON.stringify({
        status: "active",
        agent_runtime_id: { identity: agentIdentity, generation: 1 },
        fence_token: 1,
        tokens_used: 7,
        is_final: false,
      });
    },
    async mob_respawn(_mobId, agentIdentity) {
      return JSON.stringify({
        status: "completed",
        receipt: {
          identity: agentIdentity,
          member_ref: `ref-${agentIdentity}`,
        },
      });
    },
    async mob_force_cancel(mobId, agentIdentity) {
      calls.push(["force_cancel", mobId, agentIdentity]);
    },
    async mob_spawn_helper(mobId, requestJson) {
      calls.push(["spawn_helper", mobId, JSON.parse(requestJson)]);
      return JSON.stringify({
        output: "helper complete",
        tokens_used: 11,
        agent_identity: "helper-1",
        member_ref: "ref-helper-1",
      });
    },
    async mob_fork_helper(mobId, requestJson) {
      calls.push(["fork_helper", mobId, JSON.parse(requestJson)]);
      return JSON.stringify({
        output: "fork complete",
        tokens_used: 13,
        agent_identity: "fork-1",
        member_ref: "ref-fork-1",
      });
    },
    async mob_run_flow() {
      return "run-1";
    },
    async mob_flow_status() {
      return JSON.stringify({ run: { run_id: "run-1", status: "running" } });
    },
    async mob_cancel_flow(mobId, runId) {
      calls.push(["cancel_flow", mobId, runId]);
    },
    async mob_member_subscribe() {
      return 1;
    },
    async mob_subscribe_events() {
      return 2;
    },
    poll_subscription() {
      return "[]";
    },
    close_subscription(handle) {
      calls.push(["close_subscription", handle]);
    },
  };

  const runtime = await MeerkatRuntime.init(wasm, {
    anthropicApiKey: "sk-test",
    model: "claude-sonnet-4-5",
  });
  try {
    const mob = await runtime.createMob({
      id: "mob-web-parity",
      profiles: {
        worker: {
          model: "claude-sonnet-4-5",
        },
      },
    });

    const spawned = await mob.spawn([
      {
        profile: "worker",
        agent_identity: "worker-1",
        generation: 7,
      },
    ]);
    assert.equal(spawned[0].agent_identity, "worker-1");
    assert.equal(spawned[0].member_ref, "ref-worker-1");
    assert.equal(
      spawned[0].agent_runtime_id,
      undefined,
      "binding-era agent_runtime_id must not leak to app-facing spawn result",
    );
    assert.deepEqual(calls[0], [
      "spawn",
      "mob-web-parity",
      [
        {
          profile: "worker",
          agent_identity: "worker-1",
        },
      ],
    ]);

    const listed = await mob.listMembers();
    assert.equal(listed[0].agent_identity, "worker-1");
    assert.equal(listed[0].member_ref, "ref-worker-1");
    assert.equal(listed[0].agent_runtime_id, undefined);

    const receipt = await mob.member("worker-1").send("hello");
    assert.equal(receipt.agent_identity, "worker-1");
    assert.equal(receipt.member_ref, "ref-worker-1");
    assert.equal(receipt.agent_runtime_id, undefined);

    const snapshot = await mob.memberStatus("worker-1");
    assert.ok(snapshot.member_ref);
    assert.equal(snapshot.agent_runtime_id, undefined);
    assert.equal(snapshot.fence_token, undefined);
    assert.equal(snapshot.incarnation_ref, undefined);

    const appended = await mob.appendSystemContext("worker-1", {
      text: "Remember [BRIDGE-CTX].",
      source: "web-parity",
      idempotencyKey: "bridge-ctx",
    });
    assert.equal(appended.agent_identity, "worker-1");

    const respawn = await mob.respawn("worker-1");
    assert.equal(respawn.receipt.agent_identity, "worker-1");
    assert.equal(respawn.receipt.member_ref, "ref-worker-1");
    assert.equal(respawn.receipt.agent_runtime_id, undefined);
    assert.equal(respawn.receipt.fence_token, undefined);
    assert.equal(respawn.receipt.previous_fence_token, undefined);

    mob.bindings.mob_respawn = async (_mobId, agentIdentity) =>
      JSON.stringify({
        status: "topology_restore_failed",
        receipt: {
          identity: agentIdentity,
          member_ref: `ref-${agentIdentity}`,
        },
        failed_peer_ids: ["peer-a", "peer-b"],
      });
    const topologyFailure = await mob.respawn("worker-1");
    assert.equal(topologyFailure.status, "topology_restore_failed");
    assert.equal(topologyFailure.receipt.agent_identity, "worker-1");
    assert.equal(topologyFailure.receipt.member_ref, "ref-worker-1");
    assert.deepEqual(topologyFailure.failed_peer_ids, ["peer-a", "peer-b"]);
    assert.equal(topologyFailure.receipt.agent_runtime_id, undefined);
    assert.equal(topologyFailure.receipt.fence_token, undefined);

    await mob.forceCancel("worker-1");

    const helper = await mob.spawnHelper("Summarize the thread.", {
      agentIdentity: "helper-1",
      profileName: "worker",
    });
    assert.equal(helper.agent_identity, "helper-1");
    assert.equal(helper.member_ref, "ref-helper-1");
    assert.equal(helper.agent_runtime_id, undefined);

    const fork = await mob.forkHelper("worker-1", "Review the draft.", {
      agentIdentity: "fork-1",
      profileName: "worker",
      forkContext: { mode: "full_history" },
    });
    assert.equal(fork.agent_identity, "fork-1");
    assert.equal(fork.member_ref, "ref-fork-1");
    assert.equal(fork.agent_runtime_id, undefined);

    assert.deepEqual(
      calls.filter(([name]) =>
        ["member_send", "force_cancel", "spawn_helper", "fork_helper"].includes(name),
      ),
      [
        [
          "member_send",
          "worker-1",
          { content: "hello", handling_mode: "queue" },
        ],
        ["force_cancel", "mob-web-parity", "worker-1"],
        [
          "spawn_helper",
          "mob-web-parity",
          {
            prompt: "Summarize the thread.",
            agent_identity: "helper-1",
            profile_name: "worker",
          },
        ],
        [
          "fork_helper",
          "mob-web-parity",
          {
            source_member_id: "worker-1",
            prompt: "Review the draft.",
            agent_identity: "fork-1",
            profile_name: "worker",
            fork_context: { mode: "full_history" },
          },
        ],
      ],
    );
  } finally {
    runtime.destroy();
  }
});
