subagents

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// Offer the tool immediately, then follow live catalogs after agent bursts settle.
// The activation scope owns listeners, debounce fibers and replaceable tool scopes.
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
import * as z from "zod";
import { Events, Plugin, Tools, promise } from "convergence/effect";
import { mentionsFor, toolSpec } from "./lib.ts";
import type { CatalogAgent, SubagentType } from "./lib.ts";
import { spawn } from "./spawn.ts";

const SETTLE_MS = 1500;
export const program = Effect.gen(function* () {
  const { api } = yield* Plugin;
  const tools = yield* Tools;
  const events = yield* Events;
  const scope = yield* Effect.scope;
  let catalog: readonly CatalogAgent[] = [];
  let types: readonly SubagentType[] = [];
  let registered: string | undefined;
  let tool: Scope.Closeable | undefined;
  let timer: Fiber.Fiber<void> | undefined;
  let running = false;
  let again = false;

  const register = Effect.fn("Subagents.register")(function* () {
    const spec = toolSpec(catalog, types);
    const key = JSON.stringify({ ...spec, input: z.toJSONSchema(spec.input) });
    if (key === registered) return;
    registered = key;
    if (tool) yield* Scope.close(tool, Exit.void);
    tool = yield* Scope.fork(scope);
    yield* tools
      .register<typeof spec.input, z.ZodUnknown>({ ...spec, run: (input, ctx) => spawn(api, input, ctx) })
      .pipe(Scope.provide(tool));
  });
  const refresh: () => Effect.Effect<void> = Effect.fn("Subagents.refresh")(function* () {
    if (running) {
      again = true;
      return;
    }
    running = true;
    // Catalog errors retain the previous catalog and warn, as before.
    yield* promise("models.list", () => api.models.list()).pipe(
      Effect.tap((result) =>
        Effect.sync(() => {
          catalog = result.agents;
          api.mentions.set(mentionsFor(catalog));
        }),
      ),
      Effect.catch((error) =>
        Effect.sync(() => console.warn(`The subagents plugin could not list the agents' models: ${error.message}`)),
      ),
    );
    yield* promise("subagents.types", () => api.subagents.types()).pipe(
      Effect.tap((result) =>
        Effect.sync(() => {
          types = result;
        }),
      ),
      Effect.catch((error) =>
        Effect.sync(() => console.warn(`The subagents plugin could not list subagent types: ${error.message}`)),
      ),
    );
    yield* register();
    running = false;
    if (again) {
      again = false;
      yield* schedule();
    }
  });
  const schedule: () => Effect.Effect<void> = Effect.fn("Subagents.schedule")(function* () {
    if (timer) yield* Fiber.interrupt(timer);
    // Clear the timer before refreshing: a new burst must not interrupt an active refresh.
    timer = yield* Effect.sleep(SETTLE_MS).pipe(
      Effect.andThen(
        Effect.sync(() => {
          timer = undefined;
        }),
      ),
      Effect.andThen(refresh()),
      Effect.forkIn(scope, { startImmediately: true }),
    );
  });
  yield* register();
  yield* Effect.forkScoped(refresh(), { startImmediately: true });
  for (const event of ["agents_changed", "settings_changed", "workspaces_changed", "plugins_changed"] as const) {
    yield* events.stream(event).pipe(
      Stream.runForEach(() => schedule()),
      Effect.forkScoped({ startImmediately: true }),
    );
  }
});