|
| 1 | +import { Deferred, Effect } from "effect" |
| 2 | +import type { Bus } from "@/bus" |
| 3 | +import type { ActorRegistry } from "@/actor/registry" |
| 4 | +import type { Actor } from "@/actor/schema" |
| 5 | +import type { Session } from "@/session" |
| 6 | +import type { SessionID } from "@/session/schema" |
| 7 | +import { ActorStatusChanged } from "@/actor/events" |
| 8 | +import { parseReturnHeader } from "@/actor/return-header" |
| 9 | + |
| 10 | +// One member's terminal outcome within a joined dispatch group. |
| 11 | +export interface GroupMemberResult { |
| 12 | + sessionID: SessionID |
| 13 | + actorID: string |
| 14 | + description?: string |
| 15 | + agent?: string |
| 16 | + // The aggregation bucket. "unknown" = the member was never a registered actor |
| 17 | + // (bad id); it can never become terminal, so it does NOT block the barrier. |
| 18 | + outcome: "success" | "failure" | "cancelled" | "unknown" |
| 19 | + result?: string |
| 20 | + error?: string |
| 21 | + reportedStatus?: string |
| 22 | + reportedSummary?: string |
| 23 | +} |
| 24 | + |
| 25 | +export interface JoinResult { |
| 26 | + status: "complete" | "timeout" |
| 27 | + total: number |
| 28 | + counts: { success: number; failure: number; cancelled: number; unknown: number } |
| 29 | + members: GroupMemberResult[] |
| 30 | +} |
| 31 | + |
| 32 | +export interface JoinDeps { |
| 33 | + reg: ActorRegistry.Interface |
| 34 | + sessions: Session.Interface |
| 35 | + bus: Bus.Interface |
| 36 | +} |
| 37 | + |
| 38 | +const DEFAULT_TIMEOUT_MS = 600_000 |
| 39 | + |
| 40 | +// A group member is terminal once its actor row is idle WITH an outcome. Unlike |
| 41 | +// the single-actor waiter (which ignores a persistent peer's idle/success so it |
| 42 | +// can keep relaying), fan-in counts EVERY terminal outcome — success, failure, |
| 43 | +// AND cancel (all three now flow through ActorTurn.runTurn / Actor.cancel and |
| 44 | +// publish ActorStatusChanged, per T41). |
| 45 | +function terminalOutcome(entry: Actor | undefined): "success" | "failure" | "cancelled" | undefined { |
| 46 | + if (!entry) return undefined |
| 47 | + if (entry.status !== "idle") return undefined |
| 48 | + return entry.lastOutcome |
| 49 | +} |
| 50 | + |
| 51 | +type Member = { sessionID: SessionID; actorID: string } |
| 52 | +type Resolved = "success" | "failure" | "cancelled" | "unknown" |
| 53 | + |
| 54 | +// Fan-in barrier: register interest in a GROUP of child actors (each keyed by |
| 55 | +// its session id + actor id — for a peer both equal the child session id) and |
| 56 | +// resolve ONCE when EVERY member has reached a terminal state, returning an |
| 57 | +// aggregated per-member summary. Does NOT busy-wait: it subscribes to |
| 58 | +// ActorStatusChanged (the same notification path T41 drives) and re-snapshots |
| 59 | +// the group on each touching event. A member that is already terminal (or |
| 60 | +// unknown — a bad id that has no row and can never settle) at call time is |
| 61 | +// counted immediately; the barrier resolves synchronously if all are settled. |
| 62 | +// |
| 63 | +// Standalone Effect (mirrors forkQuery in tool/session.ts) rather than a Layer |
| 64 | +// service: it needs only the registry + session + bus interfaces the caller |
| 65 | +// already holds — so it adds no new layer dependency and no cycle. It takes the |
| 66 | +// INJECTED Bus.Interface (not the static Bus.subscribe) so it subscribes to the |
| 67 | +// exact same PubSub instance the ActorRegistry layer publishes ActorStatusChanged |
| 68 | +// on — the static builds its own runtime and would miss layer-emitted events. |
| 69 | +export function joinGroup( |
| 70 | + deps: JoinDeps, |
| 71 | + input: { members: Member[]; timeout_ms?: number }, |
| 72 | +): Effect.Effect<JoinResult> { |
| 73 | + return Effect.gen(function* () { |
| 74 | + const timeoutMs = input.timeout_ms ?? DEFAULT_TIMEOUT_MS |
| 75 | + |
| 76 | + // Dedup members by sessionID:actorID — a caller may list the same child |
| 77 | + // twice; the barrier must count it once. |
| 78 | + const seen = new Set<string>() |
| 79 | + const members = input.members.filter((m) => { |
| 80 | + const key = `${m.sessionID}:${m.actorID}` |
| 81 | + if (seen.has(key)) return false |
| 82 | + seen.add(key) |
| 83 | + return true |
| 84 | + }) |
| 85 | + |
| 86 | + if (members.length === 0) |
| 87 | + return { |
| 88 | + status: "complete" as const, |
| 89 | + total: 0, |
| 90 | + counts: { success: 0, failure: 0, cancelled: 0, unknown: 0 }, |
| 91 | + members: [], |
| 92 | + } |
| 93 | + |
| 94 | + const lastAssistantText = (sessionID: SessionID, actorID: string) => |
| 95 | + Effect.gen(function* () { |
| 96 | + const msgs = yield* deps.sessions |
| 97 | + .messages({ sessionID, agentID: actorID }) |
| 98 | + .pipe(Effect.orElseSucceed(() => [])) |
| 99 | + const last = msgs.findLast((m) => m.info.role === "assistant") |
| 100 | + if (!last) return undefined |
| 101 | + const textPart = last.parts.findLast( |
| 102 | + (p): p is Extract<(typeof last.parts)[number], { type: "text" }> => p.type === "text", |
| 103 | + ) |
| 104 | + return textPart?.text |
| 105 | + }) |
| 106 | + |
| 107 | + const memberResult = (m: Member, entry: Actor | undefined, outcome: Resolved) => |
| 108 | + Effect.gen(function* () { |
| 109 | + const result = outcome === "success" ? yield* lastAssistantText(m.sessionID, m.actorID) : undefined |
| 110 | + const reported = parseReturnHeader(result) |
| 111 | + return { |
| 112 | + sessionID: m.sessionID, |
| 113 | + actorID: m.actorID, |
| 114 | + ...(entry?.description !== undefined ? { description: entry.description } : {}), |
| 115 | + ...(entry?.agent !== undefined ? { agent: entry.agent } : {}), |
| 116 | + outcome, |
| 117 | + ...(result !== undefined ? { result } : {}), |
| 118 | + ...(outcome === "failure" && entry?.lastError !== undefined ? { error: entry.lastError } : {}), |
| 119 | + ...(reported.status ? { reportedStatus: reported.status } : {}), |
| 120 | + ...(reported.summary ? { reportedSummary: reported.summary } : {}), |
| 121 | + } satisfies GroupMemberResult |
| 122 | + }) |
| 123 | + |
| 124 | + // Snapshot every member's current terminal state. `settled` = terminal or |
| 125 | + // unknown (a member that can never become terminal must not block). |
| 126 | + const snapshotAll = () => |
| 127 | + Effect.forEach(members, (m) => |
| 128 | + Effect.gen(function* () { |
| 129 | + const entry = yield* deps.reg.get(m.sessionID, m.actorID) |
| 130 | + const term = terminalOutcome(entry) |
| 131 | + const outcome: Resolved | undefined = term ?? (entry ? undefined : "unknown") |
| 132 | + return { m, entry, settled: outcome !== undefined, outcome } |
| 133 | + }), |
| 134 | + ) |
| 135 | + |
| 136 | + const aggregate = ( |
| 137 | + resolved: { m: Member; entry: Actor | undefined; outcome: Resolved }[], |
| 138 | + status: "complete" | "timeout", |
| 139 | + ) => |
| 140 | + Effect.gen(function* () { |
| 141 | + const out = yield* Effect.forEach(resolved, (r) => memberResult(r.m, r.entry, r.outcome)) |
| 142 | + const counts = { |
| 143 | + success: out.filter((x) => x.outcome === "success").length, |
| 144 | + failure: out.filter((x) => x.outcome === "failure").length, |
| 145 | + cancelled: out.filter((x) => x.outcome === "cancelled").length, |
| 146 | + unknown: out.filter((x) => x.outcome === "unknown").length, |
| 147 | + } |
| 148 | + return { status, total: out.length, counts, members: out } satisfies JoinResult |
| 149 | + }) |
| 150 | + |
| 151 | + // Fast path: everyone already settled. |
| 152 | + const initial = yield* snapshotAll() |
| 153 | + if (initial.every((r) => r.settled)) |
| 154 | + return yield* aggregate( |
| 155 | + initial.map((r) => ({ m: r.m, entry: r.entry, outcome: r.outcome! })), |
| 156 | + "complete", |
| 157 | + ) |
| 158 | + |
| 159 | + const resolved = yield* Deferred.make<JoinResult>() |
| 160 | + |
| 161 | + return yield* Effect.acquireUseRelease( |
| 162 | + // On each status change touching a member, re-snapshot the whole group and |
| 163 | + // resolve only when ALL are settled. Re-snapshotting (vs a per-member |
| 164 | + // latch) keeps state authoritative against the DB and idempotent under |
| 165 | + // duplicate/out-of-order events. |
| 166 | + deps.bus.subscribeCallback(ActorStatusChanged, (evt) => { |
| 167 | + const touched = members.some( |
| 168 | + (m) => m.actorID === evt.properties.actorID && m.sessionID === evt.properties.sessionID, |
| 169 | + ) |
| 170 | + if (!touched) return |
| 171 | + Effect.runFork( |
| 172 | + Effect.gen(function* () { |
| 173 | + const snap = yield* snapshotAll() |
| 174 | + if (!snap.every((r) => r.settled)) return |
| 175 | + const agg = yield* aggregate( |
| 176 | + snap.map((r) => ({ m: r.m, entry: r.entry, outcome: r.outcome! })), |
| 177 | + "complete", |
| 178 | + ) |
| 179 | + Deferred.doneUnsafe(resolved, Effect.succeed(agg)) |
| 180 | + }).pipe(Effect.catchCause((cause) => Effect.logError(`group join snapshot failed: ${cause}`))), |
| 181 | + ) |
| 182 | + }), |
| 183 | + () => |
| 184 | + Effect.gen(function* () { |
| 185 | + // Re-check after subscribing — a member could have flipped terminal |
| 186 | + // between the initial snapshot and the bus bind (lost-wakeup guard). |
| 187 | + const recheck = yield* snapshotAll() |
| 188 | + if (recheck.every((r) => r.settled)) |
| 189 | + return yield* aggregate( |
| 190 | + recheck.map((r) => ({ m: r.m, entry: r.entry, outcome: r.outcome! })), |
| 191 | + "complete", |
| 192 | + ) |
| 193 | + const raced = yield* Deferred.await(resolved).pipe( |
| 194 | + Effect.timeout(timeoutMs), |
| 195 | + Effect.catchTag("TimeoutError", () => Effect.succeed(null)), |
| 196 | + ) |
| 197 | + if (raced === null) { |
| 198 | + // Timed out — return the partial snapshot so the caller still sees |
| 199 | + // which members settled. status: "timeout" signals not-all-done. |
| 200 | + const partial = yield* snapshotAll() |
| 201 | + return yield* aggregate( |
| 202 | + partial.map((r) => ({ m: r.m, entry: r.entry, outcome: (r.outcome ?? "unknown") as Resolved })), |
| 203 | + "timeout", |
| 204 | + ) |
| 205 | + } |
| 206 | + return raced |
| 207 | + }), |
| 208 | + (unsub) => Effect.sync(() => unsub()), |
| 209 | + ) |
| 210 | + }) |
| 211 | +} |
0 commit comments