claude

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36// Claude's line transport uses the SDK's ordered writer and pipe reader.
// Acquisition and graceful shutdown belong to the provider's Effect scope.
import * as Effect from "effect/Effect";
import { Process } from "convergence/effect";
import { LineProcess } from "../sdk/process.ts";
import type { LineProcessOptions } from "../sdk/process.ts";
import { TransportFailed } from "../sdk/effect.ts";
import { errorMessage } from "../sdk/errors.ts";

const attempt = Effect.fnUntraced(function* <A>(run: () => Promise<A>) {
  return yield* Effect.tryPromise({
    try: run,
    catch: (cause) => new TransportFailed({ message: errorMessage(cause), cause }),
  });
});
export const stop = Effect.fn("Claude.transport.stop")(function* (peer: LineProcess, grace = 2000) {
  peer.closedReason ??= "shut down";
  if (peer.exitStatus !== null) return yield* attempt(() => peer.ended);
  // Closed input or an already exited peer needs no further cleanup.
  yield* attempt(() => peer.child.stdin.close()).pipe(Effect.catchTag("TransportFailed", () => Effect.void));
  yield* attempt(() => peer.kill("SIGTERM")).pipe(Effect.catchTag("TransportFailed", () => Effect.void));
  const exited = yield* Effect.raceFirst(
    attempt(() => peer.exited).pipe(Effect.as(true)),
    Effect.sleep(grace).pipe(Effect.as(false)),
  );
  if (!exited) yield* attempt(() => peer.kill("SIGKILL")).pipe(Effect.catchTag("TransportFailed", () => Effect.void));
  // Exit and stdout completion are separate: drain before releasing the line queue.
  else yield* attempt(() => peer.ended);
});
export const spawn = Effect.fn("Claude.transport.spawn")(function* (args: string[], options: LineProcessOptions) {
  const child = yield* (yield* Process).spawn("claude", args, { cwd: options.cwd, env: options.env });
  return yield* Effect.acquireRelease(
    Effect.sync(() => new LineProcess(child, options)),
    (peer) => stop(peer).pipe(Effect.orDie),
  );
});