// @effect-diagnostics nodeBuiltinImport:off -- fake systemd binaries live in a temp dir.
import * as NodeChildProcess from "node:child_process";
import * as NodeCrypto from "node:crypto";
import * as NodeFS from "node:fs";
import * as NodeOS from "node:os";
import * as NodePath from "node:path";

import * as NodeServices from "@effect/platform-node/NodeServices";
import { assert, describe, expect, it } from "@effect/vitest";
import { ProviderSessionId, ThreadId } from "@t3tools/contracts";
import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
import * as AcpSessionRuntime from "@t3tools/provider-acp/server/AcpSessionRuntime";
import { AgentScope } from "@t3tools/shared/AgentScope";
import * as HostProcess from "@t3tools/shared/HostProcess";
import * as Deferred from "effect/Deferred";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import type * as Scope from "effect/Scope";
import * as Schema from "effect/Schema";
import { ChildProcess, ChildProcessSpawner } from "effect/process";

import * as ClaudeAdapterV2 from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";
import * as ServerConfig from "../config.ts";
import { makeCodexAppServerSpawnCommand } from "../orchestration-v2/Adapters/CodexAdapterV2.ts";
import * as ProcessRunner from "../processRunner.ts";
import * as ProviderProcesses from "./providerProcesses.ts";
import { classifyScope, layer, make, ownedAgentSlice, type ScopeState } from "./agentScope.ts";

const running: ScopeState = {
  loadState: "loaded",
  activeState: "active",
  result: "success",
  oomKills: 0,
  populated: true,
};

describe("classifyScope", () => {
  it("reports an OOM kill only once systemd records it", () => {
    expect(classifyScope({ ...running, activeState: "failed", result: "oom-kill" })).toBe(
      "oom-killed",
    );
    expect(classifyScope({ ...running, oomKills: 1 })).toBe("stopping");
    expect(classifyScope({ ...running, populated: false })).toBe("stopping");
    expect(classifyScope({ ...running, activeState: "deactivating" })).toBe("stopping");
  });
});

const withFakeSystemd = <A, E>(
  body: (input: {
    readonly scope: Effect.Success<typeof make>;
    readonly units: Map<string, { activeState: string; result: string }>;
    readonly calls: ReadonlyArray<ReadonlyArray<string>>;
    readonly stateDir: string;
  }) => Effect.Effect<A, E>,
  {
    oomPolicy = true,
    leftovers = false,
    liveOwner = false,
  }: {
    readonly oomPolicy?: boolean;
    readonly leftovers?: boolean;
    readonly liveOwner?: boolean;
  } = {},
) =>
  Effect.gen(function* () {
    const bin = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-agent-scope-"));
    for (const name of ["systemd-run", "systemctl", "agent"]) {
      NodeFS.writeFileSync(NodePath.join(bin, name), "", { mode: 0o755 });
    }
    const units = new Map<string, { activeState: string; result: string }>();
    const calls: Array<ReadonlyArray<string>> = [];
    const output = (stdout: string, code = 0): ProcessRunner.ProcessRunOutput => ({
      stdout,
      stderr: "",
      code: ChildProcessSpawner.ExitCode(code),
      timedOut: false,
      stdoutTruncated: false,
      stderrTruncated: false,
      stdoutInvalidUtf8: false,
      stderrInvalidUtf8: false,
    });
    const runner = ProcessRunner.ProcessRunner.of({
      run: ({ args }) =>
        Effect.sync(() => {
          calls.push(args);
          if (!oomPolicy && args.some((arg) => arg.startsWith("--property=OOMPolicy="))) {
            return output("", 1);
          }
          const [, verb, unit = ""] = args;
          if (verb === "reset-failed") units.delete(unit);
          if (verb !== "show") return output("");
          const state = units.get(unit);
          return output(
            state === undefined
              ? "LoadState=not-found\nActiveState=inactive\nResult=success\n"
              : `LoadState=loaded\nActiveState=${state.activeState}\nResult=${state.result}\n`,
          );
        }),
    });
    const { stateDir } = yield* ServerConfig.ServerConfig;
    if (leftovers)
      units.set(ownedAgentSlice(stateDir), { activeState: "active", result: "success" });
    const scope = yield* make.pipe(
      Effect.provideService(ProviderProcesses.ProviderProcesses, { hasOtherLiveOwner: liveOwner }),
      Effect.provideService(ProcessRunner.ProcessRunner, runner),
      Effect.provideService(HostProcess.Platform, "linux"),
      Effect.provideService(HostProcess.Environment, { PATH: bin, XDG_RUNTIME_DIR: "/run/user/1" }),
    );
    return yield* body({ scope, units, calls, stateDir }).pipe(
      Effect.ensuring(Effect.sync(() => NodeFS.rmSync(bin, { recursive: true, force: true }))),
    );
  }).pipe(
    Effect.scoped,
    Effect.provide(
      ServerConfig.layerTest(process.cwd(), { prefix: "t3-agent-scope-" }).pipe(
        Layer.provideMerge(NodeServices.layer),
      ),
    ),
  );

const unitOf = (launch: { readonly args: ReadonlyArray<string> }) =>
  launch.args.find((arg) => arg.startsWith("--unit="))?.slice("--unit=".length) ?? "";

const sliceOf = (launch: { readonly args: ReadonlyArray<string> }) =>
  launch.args.find((arg) => arg.startsWith("--slice="))?.slice("--slice=".length) ?? "";

const launchCodex = (scope: Effect.Success<typeof make>, threadId: string) =>
  makeCodexAppServerSpawnCommand({ command: "agent", args: ["app-server"], threadId }).pipe(
    Effect.provideService(AgentScope, scope),
    Effect.provideService(HostProcess.Platform, "linux"),
    Effect.map((command) => {
      assert(ChildProcess.isStandardCommand(command));
      return command;
    }),
  );

describe("AgentScope service", () => {
  it.effect("labels a thread from its newest scope until its next session clears it", () =>
    withFakeSystemd(({ scope, units }) =>
      Effect.gen(function* () {
        const first = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
        units.set(unitOf(first), { activeState: "failed", result: "oom-kill" });
        const second = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
        units.set(unitOf(second), { activeState: "active", result: "success" });
        expect(yield* scope.oomKilled("a")).toBe(false);

        units.set(unitOf(second), { activeState: "failed", result: "oom-kill" });
        expect(yield* scope.oomKilled("a")).toBe(true);
        units.delete(unitOf(second));
        expect(yield* scope.oomKilled("a")).toBe(true);

        yield* scope.clear("a");
        expect(yield* scope.oomKilled("a")).toBe(false);
      }),
    ),
  );

  it.effect(
    "keeps shared Codex scopes running without attributing their OOM to the launch thread",
    () =>
      withFakeSystemd(({ scope, units, calls }) =>
        Effect.gen(function* () {
          const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
          expect(agent.args).toContain("--property=OOMPolicy=stop");
          units.set(unitOf(agent), { activeState: "failed", result: "oom-kill" });
          expect(yield* scope.oomKilled("a")).toBe(true);

          const codex = yield* launchCodex(scope, "a");
          expect(codex.args).toContain("--property=OOMPolicy=continue");
          expect(codex.args).not.toContain("--property=OOMPolicy=stop");
          expect(codex.args).toContain("--property=TimeoutStopSec=10s");
          expect(codex.args).toContain("--collect");
          units.set(unitOf(codex), { activeState: "active", result: "success" });
          expect(yield* scope.oomKilled("a")).toBe(false);
          units.set(unitOf(codex), { activeState: "failed", result: "oom-kill" });
          expect(yield* scope.oomKilled("a")).toBe(false);
          expect(
            calls.filter(([, verb, unit]) => verb === "show" && unit === unitOf(codex)),
          ).toEqual([]);
        }),
      ),
  );

  it.effect("keeps a running scope while hundreds of other threads launch", () =>
    withFakeSystemd(({ scope, units }) =>
      Effect.gen(function* () {
        const live = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "live" });
        units.set(unitOf(live), { activeState: "active", result: "success" });
        for (let index = 0; index < 600; index++) {
          yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: `other-${index}` });
        }
        units.set(unitOf(live), { activeState: "failed", result: "oom-kill" });
        expect(yield* scope.oomKilled("live")).toBe(true);
      }),
    ),
  );

  it.effect("stops agents a killed server left during acquisition before the first launch", () =>
    withFakeSystemd(
      ({ scope, calls, stateDir }) =>
        Effect.gen(function* () {
          const stops = calls.filter(([, verb]) => verb === "stop");
          expect(stops).toEqual([["--user", "stop", ownedAgentSlice(stateDir)]]);
          const agent = yield* scope.wrap({ command: "agent", args: [], name: "t", threadId: "a" });
          expect(sliceOf(agent)).toBe(ownedAgentSlice(stateDir));
          expect(agent.args).toContain("--property=TimeoutStopSec=10s");
          const terminal = yield* scope.wrap({
            command: "agent",
            args: [],
            name: "terminal",
            survivesRestart: true,
          });
          expect(sliceOf(terminal)).toBe("app-t3code-agents.slice");
          expect(calls.filter(([, verb]) => verb === "stop")).toHaveLength(1);
        }),
      { leftovers: true },
    ),
  );

  it.effect("does not stop the owned slice while another recorded server is alive", () =>
    withFakeSystemd(
      ({ calls }) =>
        Effect.sync(() => {
          expect(calls.filter(([, verb]) => verb === "stop")).toEqual([]);
        }),
      { leftovers: true, liveOwner: true },
    ),
  );
});

// These fixtures only signal PIDs captured at spawn; no server or provider credentials are used.
const ownedProcessFixture = Effect.gen(function* () {
  const ready = yield* Deferred.make<void>();
  const exited = yield* Deferred.make<void>();
  const child = yield* Effect.acquireRelease(
    Effect.sync(() => {
      const child = NodeChildProcess.spawn(
        process.execPath,
        ["-e", 'process.stdin.resume(); process.send("ready");'],
        { stdio: ["pipe", "ignore", "ignore", "ipc"] },
      );
      child.once("message", () => Deferred.doneUnsafe(ready, Effect.void));
      child.once("error", (error) => Deferred.doneUnsafe(ready, Effect.die(error)));
      child.once("exit", () => Deferred.doneUnsafe(exited, Effect.void));
      return child;
    }),
    (child) =>
      Effect.sync(() => child.kill("SIGKILL")).pipe(Effect.andThen(Deferred.await(exited))),
  );
  yield* Deferred.await(ready);
  assert(child.pid !== undefined);
  const controller = AcpSessionRuntime.makePosixProcessTreeController("linux");
  const pid = child.pid;
  const identity = controller.identity(pid);
  assert(identity !== undefined);
  const boot = NodeFS.readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim();
  return {
    pid,
    startTime: `${boot}:${identity.startTime}`,
    pgid: 0,
    running: () => {
      const observed = controller.identity(pid);
      return observed !== undefined && !AcpSessionRuntime.posixProcessIsZombie(observed);
    },
    stop: Effect.sync(() => child.kill("SIGKILL")).pipe(Effect.andThen(Deferred.await(exited))),
  };
});

const withRestartOwnership = <A, E, R>(
  body: (input: {
    readonly record: (
      owner: { readonly pid: number; readonly startTime: string },
      processes: ReadonlyArray<{
        readonly pid: number;
        readonly startTime: string;
        readonly pgid: number;
      }>,
    ) => void;
    readonly restart: Effect.Effect<void, never, Scope.Scope>;
  }) => Effect.Effect<A, E, R>,
) =>
  Effect.gen(function* () {
    const config = yield* ServerConfig.ServerConfig;
    const { stateDir } = config;
    const directory = NodePath.join(stateDir, "provider-processes");
    NodeFS.mkdirSync(directory, { recursive: true });
    const record = (
      owner: { readonly pid: number; readonly startTime: string },
      processes: ReadonlyArray<{
        readonly pid: number;
        readonly startTime: string;
        readonly pgid: number;
      }>,
    ) => {
      const instanceId = NodeCrypto.randomUUID();
      NodeFS.writeFileSync(
        NodePath.join(directory, `${instanceId}.json`),
        JSON.stringify({
          version: 1,
          instanceId,
          hostname: NodeOS.hostname(),
          stateDir,
          owner,
          processes,
        }),
      );
    };
    // Acquiring the production layer is the barrier before orchestration recovery may run.
    const restart = Layer.build(layer).pipe(
      Effect.provideService(ServerConfig.ServerConfig, config),
      Effect.provide(NodeServices.layer),
      Effect.asVoid,
    );
    return yield* body({ record, restart });
  }).pipe(
    Effect.scoped,
    Effect.provideService(HostProcess.Platform, "linux"),
    Effect.provideService(HostProcess.Environment, { PATH: process.env.PATH }),
    Effect.provide(
      ServerConfig.layerTest(process.cwd(), { prefix: "t3-restart-ownership-" }).pipe(
        Layer.provideMerge(NodeServices.layer),
      ),
    ),
  );

describe.runIf(HostProcess.Platform.defaultValue() === "linux")(
  "provider restart ownership without systemd",
  () => {
    it.live(
      "stops both recorded survivors before startup completes and leaves an unrecorded process alone",
      () =>
        withRestartOwnership(({ record, restart }) =>
          Effect.gen(function* () {
            const owner = yield* ownedProcessFixture;
            const first = yield* ownedProcessFixture;
            const second = yield* ownedProcessFixture;
            const unrelated = yield* ownedProcessFixture;
            record(owner, [first]);
            record(owner, [second]);
            yield* owner.stop;

            yield* restart;

            expect([first.running(), second.running()]).toEqual([false, false]);
            expect(unrelated.running()).toBe(true);
          }),
        ),
    );

    it.live(
      "preserves a live server's provider and a reused provider PID while reaping a matching orphan",
      () =>
        withRestartOwnership(({ record, restart }) =>
          Effect.gen(function* () {
            const deadOwner = yield* ownedProcessFixture;
            const liveOwner = yield* ownedProcessFixture;
            const liveProvider = yield* ownedProcessFixture;
            const reusedProvider = yield* ownedProcessFixture;
            const orphan = yield* ownedProcessFixture;
            record(liveOwner, [liveProvider]);
            record(deadOwner, [
              { ...reusedProvider, startTime: reusedProvider.startTime.replace(/:\d+$/, ":1") },
            ]);
            record(deadOwner, [orphan]);
            yield* deadOwner.stop;

            yield* restart;

            expect(liveProvider.running()).toBe(true);
            expect(reusedProvider.running()).toBe(true);
            expect(orphan.running()).toBe(false);
          }),
        ),
    );

    it.live("does not mistake a reused owner PID for a live owning server", () =>
      withRestartOwnership(({ record, restart }) =>
        Effect.gen(function* () {
          const reusedOwner = yield* ownedProcessFixture;
          const orphan = yield* ownedProcessFixture;
          record({ ...reusedOwner, startTime: reusedOwner.startTime.replace(/:\d+$/, ":1") }, [
            orphan,
          ]);

          yield* restart;

          expect(reusedOwner.running()).toBe(true);
          expect(orphan.running()).toBe(false);
        }),
      ),
    );
  },
);

const LaunchObservation = Schema.Struct({
  pid: Schema.Number,
  record: Schema.Struct({
    version: Schema.Literal(1),
    owner: Schema.Struct({ pid: Schema.Number, startTime: Schema.String }),
    processes: Schema.Array(Schema.Struct({ pid: Schema.Number, startTime: Schema.String })),
  }),
});

// The fake CLI reports ownership at entry and answers SDK control requests.
// It never starts a provider or reads provider credentials.
const decodeLaunchObservation = Schema.decodeEffect(Schema.fromJsonString(LaunchObservation));

const FAKE_CLI = `
import * as fs from "node:fs";
import * as path from "node:path";
import * as readline from "node:readline";
const directory = process.env.T3_TEST_PROCESS_DIRECTORY;
const ready = process.env.T3_TEST_PROCESS_READY;
const records = fs.readdirSync(directory)
  .filter((file) => file.endsWith(".json"))
  .map((file) => JSON.parse(fs.readFileSync(path.join(directory, file), "utf8")));
const record = records.find((record) => record.processes.some((entry) => entry.pid === process.pid));
fs.writeFileSync(ready + ".pending", JSON.stringify({ pid: process.pid, record }));
fs.renameSync(ready + ".pending", ready);
readline.createInterface({ input: process.stdin }).on("line", (line) => {
  const frame = JSON.parse(line);
  if (frame.type !== "control_request") return;
  process.stdout.write(JSON.stringify({
    type: "control_response",
    response: { subtype: "success", request_id: frame.request_id, response: {} },
  }) + "\\n");
});
`;

describe.runIf(HostProcess.Platform.defaultValue() === "linux")("Claude process ownership", () => {
  it.live("records the production SDK child before it executes and forgets it after exit", () =>
    Effect.gen(function* () {
      const config = yield* ServerConfig.ServerConfig;
      const directory = NodePath.join(config.stateDir, "provider-processes");
      const readyPath = NodePath.join(config.stateDir, "ready.json");
      const cliPath = NodePath.join(config.stateDir, "fake-cli.mjs");
      NodeFS.writeFileSync(cliPath, FAKE_CLI);
      const ready = yield* Deferred.make<void>();
      const removed = yield* Deferred.make<void>();
      yield* Effect.acquireRelease(
        Effect.sync(() =>
          NodeFS.watch(config.stateDir, (_, file) => {
            if (file === "ready.json" && NodeFS.existsSync(readyPath)) {
              Deferred.doneUnsafe(ready, Effect.void);
            }
          }),
        ),
        (watcher) => Effect.sync(() => watcher.close()),
      );
      yield* Effect.gen(function* () {
        const runner = yield* ClaudeAdapterV2.ClaudeAgentSdkQueryRunner;
        yield* Effect.acquireRelease(
          Effect.sync(() =>
            NodeFS.watch(directory, (_, file) => {
              if (
                file !== null &&
                file.endsWith(".json") &&
                !NodeFS.existsSync(NodePath.join(directory, file))
              ) {
                Deferred.doneUnsafe(removed, Effect.void);
              }
            }),
          ),
          (watcher) => Effect.sync(() => watcher.close()),
        );
        yield* Effect.gen(function* () {
          yield* Effect.acquireRelease(
            runner.open({
              threadId: ThreadId.make("ownership-thread"),
              providerSessionId: ProviderSessionId.make("ownership-session"),
              options: {
                ...ClaudeAdapterV2.makeClaudeQueryOptions({
                  modelSelection: {
                    instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID,
                    model: "claude-sonnet-4-6",
                  },
                  nativeThreadId: NodeCrypto.randomUUID(),
                  resume: false,
                  cwd: config.stateDir,
                }),
                pathToClaudeCodeExecutable: cliPath,
                env: {
                  PATH: `${NodePath.dirname(process.execPath)}:/usr/bin:/bin`,
                  T3_TEST_PROCESS_DIRECTORY: directory,
                  T3_TEST_PROCESS_READY: readyPath,
                },
              },
            }),
            (session) => session.close.pipe(Effect.orDie),
          );
          yield* Deferred.await(ready);
          const observation = yield* decodeLaunchObservation(
            NodeFS.readFileSync(readyPath, "utf8"),
          );
          assert.equal(observation.record.owner.pid, process.pid);
          assert.deepEqual(
            observation.record.processes.map(({ pid }) => pid),
            [observation.pid],
          );
          const boot = NodeFS.readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim();
          assert.isTrue(observation.record.processes[0]?.startTime.startsWith(`${boot}:`));
          const files = NodeFS.readdirSync(directory).filter((file) => file.endsWith(".json"));
          // One record belongs to the server; the other belongs to this CLI.
          assert.lengthOf(files, 2);
        }).pipe(Effect.scoped);
        yield* Deferred.await(removed);
      }).pipe(
        Effect.provide(ClaudeAdapterV2.layerQueryRunner),
        Effect.provideService(
          ProviderEventLoggers.ProviderEventLoggers,
          ProviderEventLoggers.NoOpProviderEventLoggers,
        ),
        Effect.provide(layer),
      );
    }).pipe(
      Effect.scoped,
      Effect.provideService(HostProcess.Platform, "linux"),
      Effect.provideService(HostProcess.Environment, { PATH: process.env.PATH }),
      Effect.provide(
        ServerConfig.layerTest(process.cwd(), { prefix: "t3-process-launch-" }).pipe(
          Layer.provideMerge(NodeServices.layer),
        ),
      ),
    ),
  );
});