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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187// Node fixtures for Claude's Effect programs. Each test owns its runtime,
// queues and scope; shutdown finishes before the next fixture is released.
import * as Exit from "effect/Exit";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as ManagedRuntime from "effect/ManagedRuntime";
import * as Scope from "effect/Scope";
import * as Queue from "effect/Queue";
import * as Stream from "effect/Stream";
import * as z from "zod";
import assert from "node:assert/strict";
import { test as nodeTest } from "node:test";
import type { TestContext } from "node:test";
import { AsyncLocalStorage } from "node:async_hooks";
import { mkdtempSync, readFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { liveLayer, runPlugin } from "convergence/effect";
import type { PluginServices } from "convergence/effect";
import type { Api } from "convergence";
import { apiFiles } from "./files.ts";
import { diskFs, fakeApi, FakePeer } from "../sdk/testing.ts";
import { LineProcess } from "../sdk/process.ts";
import { SessionStates } from "./state.ts";
import { registerAccounts } from "./main.ts";
import type { ClaudeAgent } from "./agent.ts";
import type { Job } from "./session.ts";
import type { AgentEventKind, TranscriptItem, ToolContent } from "./types.ts";

interface Fixture {
  api: Api;
  resources: Map<Api, Resources>;
  disposers: (() => Promise<void>)[];
}
interface Resources {
  scope: Scope.Closeable;
  jobs: Queue.Queue<Job>;
  runtime: ManagedRuntime.ManagedRuntime<PluginServices | Scope.Scope, never>;
}
const current = new AsyncLocalStorage<Fixture>();
function fixtureState() {
  const state = current.getStore();
  if (!state) throw new Error("use a Claude test fixture");
  return state;
}
export function test(name: string, body: (context: TestContext) => unknown) {
  return nodeTest(name, async (context) => {
    const state: Fixture = { api: fakeApi({ fs: diskFs() }), resources: new Map(), disposers: [] };
    await current.run(state, async () => {
      try {
        await body(context);
      } finally {
        for (const dispose of state.disposers.reverse()) await dispose();
        for (const resource of state.resources.values()) {
          await Effect.runPromise(Scope.close(resource.scope, Exit.void));
          await resource.runtime.dispose();
        }
      }
    });
  });
}
export function useApi(api: Api) {
  fixtureState().api = api;
  return api;
}
export function resources(api: Api = fixtureState().api): Resources {
  const state = fixtureState();
  const existing = state.resources.get(api);
  if (existing) return existing;
  const scope = Effect.runSync(Scope.make());
  const jobs = Effect.runSync(Queue.make<Job>());
  const runtime: Resources["runtime"] = ManagedRuntime.make(
    Layer.mergeAll(
      liveLayer(
        api,
        (effect, signal) => runtime.runPromise(effect, { signal }),
        (effect) => {
          void runtime.runPromise(effect);
        },
      ),
      Layer.succeed(Scope.Scope, scope),
    ),
  );
  const resource = { scope, jobs, runtime };
  state.resources.set(api, resource);
  void runtime.runPromise(
    Stream.runForEach(Stream.fromQueue(jobs), (job) => job.pipe(Effect.forkIn(scope))).pipe(Effect.forkIn(scope)),
  );
  return resource;
}
export function run<A, E>(effect: Effect.Effect<A, E, PluginServices | Scope.Scope>, api?: Api) {
  return resources(api).runtime.runPromise(effect);
}
export function activateAccounts(api: Api) {
  useApi(api);
  const agents: ClaudeAgent[] = [];
  fixtureState().disposers.push(runPlugin(SessionStates.layer, registerAccounts(agents))(api));
  return { agents };
}

export const WF = "toolu_01VtcZ7X6iVcHJxYYKo3ShgL";
export const RUN = "wf_47a42fba-b24";
const here = new URL(".", import.meta.url).pathname;
export const fixture = (name: string) => readFileSync(join(here, "testdata", name), "utf8");
export function streamLines(dir: string, output: string) {
  return fixture("workflow_stream.jsonl")
    .split("\n")
    .filter((line) => line.trim())
    .map((line) => line.replaceAll("$DIR", dir).replaceAll("$OUTPUT", output));
}
export const tempDir = () => mkdtempSync(join(tmpdir(), "claude-plugin-"));
export const files = () => apiFiles();

const text = z.string().optional();
const mcpResult = z.looseObject({
  tools: z.array(z.looseObject({ name: z.string() })).optional(),
  content: z.array(z.looseObject({ text })).optional(),
});
export const writtenSchema = z.looseObject({
  type: text,
  uuid: text,
  priority: text,
  request_id: text,
  answered: z.boolean().optional(),
  request: z
    .looseObject({
      subtype: text,
      mode: text,
      sdkMcpServers: z.array(z.string()).optional(),
      sdkMcpServerConfigs: z.record(z.string(), z.object({ timeout: z.number() })).optional(),
      appendSystemPrompt: text,
      settings: z.record(z.string(), z.unknown()).optional(),
    })
    .optional(),
  response: z
    .looseObject({
      subtype: text,
      request_id: text,
      error: text,
      response: z
        .looseObject({ behavior: text, mcp_response: z.looseObject({ result: mcpResult.optional() }).optional() })
        .optional(),
    })
    .optional(),
});
export type Written = z.infer<typeof writtenSchema>;
export class RecordingProcess extends LineProcess {
  constructor(readonlyWritten: Written[]) {
    super(new FakePeer({ program: "claude", args: [], options: {}, pid: 1 }).child, { name: "claude" });
    this.written = readonlyWritten;
  }
  written: Written[];
  override send(message: unknown) {
    this.written.push(writtenSchema.parse(typeof message === "string" ? JSON.parse(message) : message));
  }
}

// Assertions narrow protocol unions while keeping fixture expectations explicit.

export function required<A>(value: A | null | undefined): A {
  assert.ok(value !== null && value !== undefined);
  return value;
}
export function eventAs<K extends AgentEventKind["event"]>(
  value: AgentEventKind,
  kind: K,
): Extract<AgentEventKind, { event: K }>;
export function eventAs(value: AgentEventKind, kind: AgentEventKind["event"]) {
  assert.equal(value.event, kind);
  return value;
}
export function itemAs<K extends TranscriptItem["role"]>(
  value: TranscriptItem,
  role: K,
): Extract<TranscriptItem, { role: K }>;
export function itemAs(value: TranscriptItem, role: TranscriptItem["role"]) {
  assert.equal(value.role, role);
  return value;
}
export function contentAs<K extends ToolContent["type"]>(
  value: ToolContent,
  type: K,
): Extract<ToolContent, { type: K }>;
export function contentAs(value: ToolContent, type: ToolContent["type"]) {
  assert.equal(value.type, type);
  return value;
}