cueloop

1type Workflow = "plan" | "reply" | "prototype" | "diff" | "review" | "refine";
2type ToolCall = {
3  tool: string;
4  tool_use_id: string;
5  plan?: string;
6  workflow?: Workflow;
7  content?: string;
8  proposal?: string;
9  pullRequestReference?: string;
10  title?: string;
11};
12type ToolResult = { deny: string } | { result: string };
13type OpenThreadResult = { output: ToolResult; approvedRetry: boolean };
14type SessionStart = { cwd: string; isInteractive: boolean };
15type SessionEnd = { sessionId: string; reason: string };
16type SessionCompact = { trigger: string };
17type Engine = {
18  session: { id(): Promise<string>; cwd(): Promise<string> };
19  prompt: { submit(input: { text: string }): Promise<object> };
20  tool: {
21    register(input: {
22      name: string;
23      description: string;
24      inputSchema: {
25        type: "object";
26        properties: Record<string, { type: "string"; enum?: readonly string[] }>;
27        required?: string[];
28      };
29    }): Promise<object>;
30  };
31  process: {
32    run(
33      argv: readonly string[],
34      options?: { cwd?: string; stdin?: string; timeoutMs?: number },
35    ): Promise<{ exitCode: number; stdout: string; stderr: string }>;
36  };
37  env: { get(name: string): Promise<string | undefined> };
38  store: {
39    get(key: string): Promise<boolean | undefined>;
40    set(key: string, value: boolean): Promise<void>;
41  };
42  clock: { every(ms: number, fn: () => void): { cancel(): void } };
43};
44type On = {
45  (
46    event: "session.start",
47    hook: (
48      engine: Engine,
49      input: SessionStart,
50      next: (input: SessionStart) => Promise<{ cwd: string }>,
51    ) => Promise<{ cwd: string }>,
52  ): void;
53  (
54    event: "tool.call",
55    hook: (
56      engine: Engine,
57      input: ToolCall,
58      next: (input: ToolCall) => Promise<ToolResult>,
59    ) => Promise<ToolResult>,
60  ): void;
61  (
62    event: "session.compact",
63    hook: (
64      engine: Engine,
65      input: SessionCompact,
66      next: (input: SessionCompact) => Promise<{ messages: object[] }>,
67    ) => Promise<{ messages: object[] }>,
68  ): void;
69  (
70    event: "session.end",
71    hook: (
72      engine: Engine,
73      input: SessionEnd,
74      next: (input: SessionEnd) => Promise<{ sessionId: string }>,
75    ) => Promise<{ sessionId: string }>,
76  ): void;
77};
78type BridgeRequest =
79  | {
80      operation: "open";
81      harness: "claude-code";
82      harnessSessionId: string;
83      cwd: string;
84      workflow: Workflow;
85      content?: string;
86      proposal?: string;
87      pullRequestReference?: string;
88      title?: string;
89    }
90  | { operation: "pending"; harness: "claude-code"; harnessSessionId: string }
91  | { operation: "ack"; bindingId: string; deliveryId: string; messageId: string }
92  | { operation: "refine" };
93type BridgeDelivery = {
94  bindingId: string;
95  threadId: string;
96  deliveryId: string;
97  message: { id: string };
98  wakeText: string;
99};
100type BridgeResponse =
101  | { operation: "open"; threadId: string; approvedRetry: boolean; manualOpenCommand?: string }
102  | { operation: "pending"; deliveries: BridgeDelivery[]; pendingThreadIds: string[] }
103  | { operation: "ack"; deliveryId: string }
104  | { operation: "refine"; report: string };
105
106/* oxlint-disable type-evidence/no-unknown-parameters, type-evidence/no-unsafe-dictionary-type, type-evidence/no-runtime-typeof -- Claude Mod cannot import valibot, so this parser validates the subprocess boundary locally. */
107function bridgeObject(value: unknown): Record<string, unknown> {
108  if (!value || typeof value !== "object" || Array.isArray(value))
109    throw new Error("Claude Mod cueloop bridge returned a malformed object");
110
111  // SAFETY: the preceding check excludes null, arrays, and non-object values.
112  return value as Record<string, unknown>;
113}
114
115function bridgeString(value: unknown): string {
116  if (typeof value !== "string")
117    throw new Error("Claude Mod cueloop bridge returned a malformed string");
118
119  return value;
120}
121
122function bridgeBoolean(value: unknown): boolean {
123  if (typeof value !== "boolean")
124    throw new Error("Claude Mod cueloop bridge returned a malformed boolean");
125
126  return value;
127}
128
129function parseBridgeResponse(
130  stdout: string,
131  operation: BridgeRequest["operation"],
132): BridgeResponse {
133  const parsed: unknown = JSON.parse(stdout);
134  const raw = bridgeObject(parsed);
135
136  if (raw.operation !== operation)
137    throw new Error("Claude Mod cueloop harness bridge returned the wrong operation");
138
139  if (operation === "open") {
140    const manualOpenCommand =
141      raw.manualOpenCommand === undefined ? undefined : bridgeString(raw.manualOpenCommand);
142
143    return {
144      operation,
145      threadId: bridgeString(raw.threadId),
146      approvedRetry: bridgeBoolean(raw.approvedRetry),
147      manualOpenCommand,
148    };
149  }
150
151  if (operation === "pending") {
152    if (!Array.isArray(raw.pendingThreadIds) || !Array.isArray(raw.deliveries))
153      throw new Error("Claude Mod cueloop bridge returned malformed pending deliveries");
154
155    const pendingThreadIds = raw.pendingThreadIds.map(bridgeString);
156    const deliveries = raw.deliveries.map((rawDelivery: unknown) => {
157      const delivery = bridgeObject(rawDelivery);
158      const message = bridgeObject(delivery.message);
159
160      return {
161        bindingId: bridgeString(delivery.bindingId),
162        threadId: bridgeString(delivery.threadId),
163        deliveryId: bridgeString(delivery.deliveryId),
164        message: { id: bridgeString(message.id) },
165        wakeText: bridgeString(delivery.wakeText),
166      };
167    });
168
169    return { operation, deliveries, pendingThreadIds };
170  }
171
172  if (operation === "ack") {
173    return { operation, deliveryId: bridgeString(raw.deliveryId) };
174  }
175
176  return { operation, report: bridgeString(raw.report) };
177}
178/* oxlint-enable type-evidence/no-unknown-parameters, type-evidence/no-unsafe-dictionary-type, type-evidence/no-runtime-typeof */
179
180export type ClaudeModEngine = Engine;
181
182export type ClaudeModOn = On;
183
184const READ_ONLY_TOOLS = new Set(["Read", "Grep", "Glob", "LS", "WebSearch", "WebFetch"]);
185const OPEN_THREAD_TOOL = "mcp__cueloop__open_thread";
186const REFINE_CORPUS_TOOL = "mcp__cueloop__refine_corpus";
187
188async function callBridge(engine: Engine, request: BridgeRequest): Promise<BridgeResponse> {
189  const entry = await engine.env.get("CUELOOP_HARNESS_ENTRY");
190  const argv = entry ? ["bun", "run", entry, "harness"] : ["cueloop", "harness"];
191  const result = await engine.process.run(argv, {
192    stdin: JSON.stringify(request),
193    timeoutMs: 30_000,
194  });
195
196  if (result.exitCode !== 0)
197    throw new Error(`Claude Mod cueloop harness bridge failed: ${result.stderr.trim()}`);
198
199  return parseBridgeResponse(result.stdout, request.operation);
200}
201
202type ModState = {
203  active: boolean;
204  sessionId: string | null;
205  pendingThreadIds: string[];
206  available: boolean;
207  timer: { cancel(): void } | null;
208  polling: Promise<void> | null;
209};
210
211async function pollMessages(engine: Engine, state: ModState): Promise<void> {
212  if (!state.sessionId || state.polling) {
213    return state.polling ?? undefined;
214  }
215
216  state.polling = (async () => {
217    const response = await callBridge(engine, {
218      operation: "pending",
219      harness: "claude-code",
220      harnessSessionId: state.sessionId!,
221    });
222
223    if (response.operation !== "pending")
224      throw new Error("Claude Mod cueloop bridge returned no pending deliveries");
225
226    state.pendingThreadIds = response.pendingThreadIds;
227    for (const delivery of response.deliveries) {
228      const storeKey = `cueloop-message-${delivery.message.id}`;
229
230      if ((await engine.store.get(storeKey)) !== true) {
231        await engine.prompt.submit({ text: delivery.wakeText });
232        await engine.store.set(storeKey, true);
233      }
234
235      await callBridge(engine, {
236        operation: "ack",
237        bindingId: delivery.bindingId,
238        deliveryId: delivery.deliveryId,
239        messageId: delivery.message.id,
240      });
241    }
242    state.available = true;
243  })()
244    .catch(() => {
245      state.available = false;
246    })
247    .finally(() => {
248      state.polling = null;
249    });
250
251  return state.polling;
252}
253
254async function openThread(
255  engine: Engine,
256  state: ModState,
257  input: ToolCall,
258): Promise<OpenThreadResult> {
259  if (!input.workflow) {
260    return { output: { deny: "Claude Mod cueloop workflow is missing" }, approvedRetry: false };
261  }
262
263  if (input.workflow === "review" && !input.pullRequestReference) {
264    return {
265      output: { deny: "Claude Mod cueloop review needs pullRequestReference" },
266      approvedRetry: false,
267    };
268  }
269
270  if (input.workflow === "refine" && !input.proposal) {
271    return { output: { deny: "Claude Mod cueloop refine needs proposal" }, approvedRetry: false };
272  }
273
274  if (input.workflow !== "review" && input.workflow !== "refine" && !input.content) {
275    return {
276      output: { deny: `Claude Mod cueloop ${input.workflow} needs content` },
277      approvedRetry: false,
278    };
279  }
280
281  try {
282    const activeSessionId = await engine.session.id();
283    const response = await callBridge(engine, {
284      operation: "open",
285      harness: "claude-code",
286      harnessSessionId: activeSessionId,
287      cwd: await engine.session.cwd(),
288      workflow: input.workflow,
289      content: input.content,
290      proposal: input.proposal,
291      pullRequestReference: input.pullRequestReference,
292      title: input.title,
293    });
294
295    if (response.operation !== "open") {
296      throw new Error("Claude Mod cueloop bridge did not open");
297    }
298
299    if (response.approvedRetry) {
300      state.pendingThreadIds = state.pendingThreadIds.filter((id) => id !== response.threadId);
301    } else if (!state.pendingThreadIds.includes(response.threadId)) {
302      state.pendingThreadIds.push(response.threadId);
303    }
304
305    return {
306      approvedRetry: response.approvedRetry,
307      output: {
308        result: response.approvedRetry
309          ? "The unchanged approved plan may proceed."
310          : `Thread ${response.threadId} is pending. End this turn and wait for its Message.` +
311            (response.manualOpenCommand ? ` ${response.manualOpenCommand}` : ""),
312      },
313    };
314  } catch (error) {
315    state.available = false;
316
317    return {
318      output: { deny: `Claude Mod cueloop Thread unavailable: ${String(error)}` },
319      approvedRetry: false,
320    };
321  }
322}
323
324export function register(on: On): void {
325  const state: ModState = {
326    active: false,
327    sessionId: null,
328    pendingThreadIds: [],
329    available: false,
330    timer: null,
331    polling: null,
332  };
333
334  on("session.start", async (engine, input, next) => {
335    try {
336      if (
337        (await engine.env.get("CLAUDE_CODE_ENABLE_FUNCTION_HOOKS")) !== "1" ||
338        (await engine.env.get("CUELOOP_DISABLE")) === "1"
339      )
340        return next(input);
341
342      state.sessionId = await engine.session.id();
343      await engine.tool.register({
344        name: "open_thread",
345        description:
346          "Open or revise a cueloop Thread for plan, reply, prototype, diff, review, or refine.",
347        inputSchema: {
348          type: "object",
349          properties: {
350            workflow: {
351              type: "string",
352              enum: ["plan", "reply", "prototype", "diff", "review", "refine"],
353            },
354            content: { type: "string" },
355            proposal: { type: "string" },
356            pullRequestReference: { type: "string" },
357            title: { type: "string" },
358          },
359          required: ["workflow"],
360        },
361      });
362      await engine.tool.register({
363        name: "refine_corpus",
364        description: "Analyze resolved cueloop Threads before drafting a refine proposal.",
365        inputSchema: { type: "object", properties: {} },
366      });
367      state.active = true;
368      await pollMessages(engine, state);
369      state.timer?.cancel();
370      state.timer = engine.clock.every(1_000, () => {
371        void pollMessages(engine, state);
372      });
373    } catch {
374      state.active = false;
375      state.available = false;
376      state.timer?.cancel();
377      state.timer = null;
378
379      return next(input);
380    }
381
382    return next(input);
383  });
384
385  on("tool.call", async (engine, input, next) => {
386    if (!state.active) return next(input);
387
388    if (input.tool === "ExitPlanMode") {
389      if (!state.available) return next(input);
390
391      if (!input.plan) {
392        return { deny: "Claude Mod could not read the plan; cueloop gate stayed closed." };
393      }
394
395      const result = await openThread(engine, state, {
396        ...input,
397        workflow: "plan",
398        content: input.plan,
399      });
400
401      if (result.approvedRetry) {
402        return next(input);
403      }
404
405      if ("deny" in result.output) return next(input);
406
407      return { deny: result.output.result };
408    }
409
410    if (input.tool === OPEN_THREAD_TOOL) {
411      const opened = await openThread(engine, state, input);
412
413      return opened.output;
414    }
415
416    if (input.tool === REFINE_CORPUS_TOOL) {
417      try {
418        const response = await callBridge(engine, { operation: "refine" });
419
420        return response.operation === "refine"
421          ? { result: response.report }
422          : { deny: "Claude Mod cueloop refine bridge returned the wrong result" };
423      } catch (error) {
424        return { deny: `Claude Mod cueloop refine unavailable: ${String(error)}` };
425      }
426    }
427
428    if (state.available && state.pendingThreadIds.length > 0 && !READ_ONLY_TOOLS.has(input.tool)) {
429      return { deny: `cueloop Threads pending: ${state.pendingThreadIds.join(", ")}` };
430    }
431
432    return next(input);
433  });
434
435  on("session.compact", async (engine, input, next) => {
436    if (state.active) await pollMessages(engine, state);
437
438    return next(input);
439  });
440  on("session.end", async (_engine, input, next) => {
441    state.timer?.cancel();
442    state.timer = null;
443    state.sessionId = null;
444
445    return next(input);
446  });
447}
448