useagent：把 AI coworker 落到可恢复的运行时账本

25/** Provider-native capability detection (a subset is meaningful in Stage 1;
26 *  fields describe what the HARNESS supports natively, not what useAgent already
27 *  projects). */
28export interface HarnessCapabilities {
29  resume: boolean;
30  cancel: boolean;
31  streaming: "none" | "text" | "parts";
32  authoritativeHistory: boolean;
33  childSessions: boolean;
34  approvals: boolean;
35  questions: boolean;
36  reasoning: boolean;
37  todos: boolean;
38  patches: boolean;
39  usage: boolean;
40}

42/** Enough native identity to address one live session. No provider SDK types -
43 *  the adapter resolves these strings to a concrete sandbox/server. */
44export interface HarnessSessionHandle {
45  provider: string;
46  /** Native harness session id (opencode `ses_*`). */
47  sessionId: string;
48  /** Sandbox instance holding the resident harness. */
49  sandboxId: string;
50  /** Durable protocol/generation authority. Optional only for compatibility
51   * callers; current recovery/control paths must provide both. */
52  protocol?: string;
53  generation?: number;
54  /** Credential epoch recorded with the session and the independently resolved
55   * current epoch. Both are required for typed out-of-turn control. */
56  authEpoch?: string | null;
57  currentAuthEpoch?: string | null;
58}

109/** The Stage-1 minimal harness boundary. Fuller methods (createSession,
110 *  submitTurn, subscribe, snapshot, resolveInteraction) are deferred - the
111 *  existing EngineAdapter.run drives turns today; this seam adds the typed
112 *  control/observability surface the product layer needs. */
113export interface HarnessAdapter {
114  readonly provider: string;
115  /** Provider capability detection. Pass a live handle when session/runtime
116   *  metadata is available; adapters without per-session routing may ignore it. */
117  capabilities(handle?: HarnessSessionHandle): HarnessCapabilities;
118  /** Ask the native session to abort. Records product intent at the caller. */
119  cancel(
120    handle: HarnessSessionHandle,
121    reason: string,
122  ): Promise<HarnessOperationResult>;
123  /** Probe native history after an interruption (see north star Crash Recovery).
124   *  Bounded and non-throwing; unreachable is a normal outcome, not an error. */
125  reconcile(
126    handle: HarnessSessionHandle,
127    checkpoint?: HarnessCheckpoint,
128  ): Promise<HarnessReconciliation>;
129}

8/** Bump only when a native bridge changes its command/frame wire shape. */
9export const NATIVE_BRIDGE_PROTOCOL_VERSION = 4 as const;
10
11export type NativeBridgeCommand =
12  | { readonly kind: "prompt"; readonly text: string; readonly model?: string }
13  | { readonly kind: "steer"; readonly text: string }
14  | { readonly kind: "follow_up"; readonly text: string }
15  | { readonly kind: "cancel"; readonly reason: string };
16
17type NativeBridgeFramePayload =
18  | { readonly kind: "turn.started" }
19  | { readonly kind: "turn.completed"; readonly stopReason?: string }
20  | { readonly kind: "turn.failed"; readonly error: string; readonly stopReason?: string }
21  | { readonly kind: "message.started"; readonly messageId: string }
22  | { readonly kind: "message.delta"; readonly messageId: string; readonly text: string; readonly segment?: number; readonly authoritative?: boolean }
23  | { readonly kind: "message.authoritative"; readonly messageId: string; readonly text: string }
24  | { readonly kind: "message.completed"; readonly messageId: string }
25  | { readonly kind: "reasoning.delta"; readonly messageId: string; readonly text: string; readonly segment?: number; readonly authoritative?: boolean }
26  | { readonly kind: "reasoning.authoritative"; readonly messageId: string; readonly text: string }
27  | { readonly kind: "plan.updated"; readonly entries: readonly CanonicalPlanEntry[] }
28  | { readonly kind: "commands.updated"; readonly commands: readonly CanonicalCommand[] }

86export interface NativeBridgeFrame {
87  readonly protocolVersion: typeof NATIVE_BRIDGE_PROTOCOL_VERSION;
88  readonly sessionId: string;
89  readonly seq: number;
90  readonly ts: number;
91  readonly body: NativeBridgeFrameBody;
92}
93
94/** Assigns one monotonic sequence to every native frame before product mapping. */
95export class NativeBridgeSequencer {
96  #seq = 0;
97
98  constructor(
99    private readonly sessionId: string,
100    private readonly now: () => number = Date.now,
101  ) {}
102
103  frame(body: NativeBridgeFrameBody): NativeBridgeFrame {
104    return {
105      protocolVersion: NATIVE_BRIDGE_PROTOCOL_VERSION,
106      sessionId: this.sessionId,
107      seq: ++this.#seq,
108      ts: this.now(),
109      body,
110    };
111  }
112}

129export class NativeBridgeDeltaAccumulator {
130  #messageText = new Map<string, DurableTextSegment>();
131  #reasoningText = new Map<string, DurableTextSegment>();
132  #encoder = new TextEncoder();
133
134  #key(body: { readonly messageId: string; readonly ownerChildId?: string }): string {
135    return `${body.ownerChildId ?? "root"}\0${body.messageId}`;
136  }
137
138  #append(
139    body: Extract<NativeBridgeFrameBody, { kind: "message.delta" | "reasoning.delta" }>,
140    target: Map<string, DurableTextSegment>,
141  ): readonly NativeBridgeFrameBody[] {
142    const key = this.#key(body);
143    let segment = target.get(key) ?? { index: 0, text: "", bytes: 0 };
144    const output: NativeBridgeFrameBody[] = [];
145    let changed = false;
146    for (const character of body.text) {
147      const bytes = this.#encoder.encode(character).byteLength;
148      if (segment.bytes > 0 && segment.bytes + bytes > NATIVE_BRIDGE_DURABLE_TEXT_BYTES) {
149        if (changed) output.push({ ...body, text: segment.text, segment: segment.index });
150        segment = { index: segment.index + 1, text: "", bytes: 0 };
151        changed = false;
152      }
153      segment.text += character;
154      segment.bytes += bytes;
155      changed = true;
156    }
157    if (changed) output.push({ ...body, text: segment.text, segment: segment.index });
158    target.set(key, segment);
159    return output;
160  }
161
162  #replace(
163    body: Extract<NativeBridgeFrameBody, { kind: "message.authoritative" | "reasoning.authoritative" }>,
164    target: Map<string, DurableTextSegment>,
165  ): readonly NativeBridgeFrameBody[] {
166    const key = this.#key(body);
167    const previousLastIndex = target.get(key)?.index ?? -1;
168    target.delete(key);
169    const kind = body.kind === "message.authoritative" ? "message.delta" : "reasoning.delta";
170    const replacement = this.#append(
171      {
172        kind,
173        messageId: body.messageId,
174        text: body.text,
175        authoritative: true,
176        ...(body.ownerChildId ? { ownerChildId: body.ownerChildId } : {}),
177      },
178      target,
179    );
180    const authoritativeReplacement = replacement.length > 0
181      ? replacement
182      : [{
183          kind,
184          messageId: body.messageId,
185          text: "",
186          segment: 0,
187          authoritative: true,
188          ...(body.ownerChildId ? { ownerChildId: body.ownerChildId } : {}),
189        } as const];
190    const replacementLastIndex = target.get(key)?.index ?? -1;
191    const tombstones: NativeBridgeFrameBody[] = [];
192    for (let index = replacementLastIndex + 1; index <= previousLastIndex; index++) {
193      tombstones.push({
194        kind,
195        messageId: body.messageId,
196        text: "",
197        segment: index,
198        authoritative: true,
199        ...(body.ownerChildId ? { ownerChildId: body.ownerChildId } : {}),
200      });
201    }
202    return [...authoritativeReplacement, ...tombstones];
203  }
204
205  durable(body: NativeBridgeFrameBody): readonly NativeBridgeFrameBody[] {
206    if (body.kind === "message.delta") return this.#append(body, this.#messageText);
207    if (body.kind === "reasoning.delta") return this.#append(body, this.#reasoningText);
208    if (body.kind === "message.authoritative") return this.#replace(body, this.#messageText);
209    if (body.kind === "reasoning.authoritative") return this.#replace(body, this.#reasoningText);
210    return [body];
211  }
212}

132export function runtimeAdapterEnabled(
133  env: Readonly<Record<string, string | undefined>> = process.env,
134): boolean {
135  const value = operatorEnv(env, "RUNTIME_RUN_ADAPTER_ENABLED", "T3_RUN_ADAPTER_ENABLED")
136    ?.trim()
137    .toLowerCase();
138  return value === "1" || value === "true";
139}
140
141export function runtimeAdapterEngineSelected(
142  engine: string,
143  env: Readonly<Record<string, string | undefined>> = process.env,
144): boolean {
145  const configured = operatorEnv(env, "RUNTIME_RUN_ADAPTER_ENGINES", "T3_RUN_ADAPTER_ENGINES")?.trim();
146  if (!configured) return engine === "codex" || engine === "opencode";
147  return configured
148    .split(",")
149    .map((value) => value.trim().toLowerCase())
150    .filter(Boolean)
151    .includes(engine.toLowerCase());
152}
153
154export type RuntimeAdapterMode = "canary" | "all";
155
156export function runtimeAdapterMode(
157  env: Readonly<Record<string, string | undefined>> = process.env,
158): RuntimeAdapterMode {
159  const mode = operatorEnv(env, "RUNTIME_RUN_ADAPTER_MODE", "T3_RUN_ADAPTER_MODE")
160    ?.trim()
161    .toLowerCase() || "canary";
162  if (mode !== "canary" && mode !== "all") {
163    throw new Error("RUNTIME_RUN_ADAPTER_MODE (legacy T3_RUN_ADAPTER_MODE) must be canary or all");
164  }
165  return mode;
166}
167
168export function runtimeAdapterSelected(
169  ctx: Pick<EngineRunContext, "runId" | "threadId">,
170  env: Readonly<Record<string, string | undefined>> = process.env,
171): boolean {
172  if (!runtimeAdapterEnabled(env)) return false;
173  if (runtimeAdapterMode(env) === "all") return true;
174  const allowlist = new Set(
175    (operatorEnv(env, "RUNTIME_CANARY_THREAD_IDS", "T3_CANARY_THREAD_IDS") ?? "")
176      .split(",")
177      .map((value) => value.trim())
178      .filter(Boolean),
179  );
180  return allowlist.has(ctx.threadId ?? "") || allowlist.has(ctx.runId);

519export function makeRuntimeAdapter(engine: RuntimeEngineId, driver: ProviderDriver): EngineAdapter {
520  return {
521    id: engine,
522    async run(ctx): Promise<void> {
523      if (!providerGatewayWired()) {
524        throw new Error("Engine requires a configured provider gateway");
525      }
526      const startedAt = Date.now();
527      await ctx.emit({
528        kind: "task",
529        label: "Preparing runtime and integrations…",
530        chip: `runtime:${engine}`,
531      });
532      const prepared = await prepareSandboxTurn(ctx, {
533        snapshot: runtimeRunSnapshot(),
534        chip: `runtime:${engine}`,
535        warmPool: RUNTIME_CUBE_WARM_POOL_NAME,
536        labels: { [RUNTIME_GENERATION_LABEL]: RUNTIME_GENERATION },
537        requiredLabels: { [RUNTIME_GENERATION_LABEL]: RUNTIME_GENERATION },
538        // Frozen timing prefix: hosted cutover canaries read these values.
539        timingPrefix: "t3",
540        async prepareProvider(sandbox, workdir) {
541          return await prepareRuntimeProviderBridge(sandbox, ctx, engine, workdir);
542        },
543      });
544      const { sandbox, workdir, redact } = prepared;
545      const providerBridgeLease: RuntimeProviderBridgeLease = prepared.providerState;
546
547      try {
548        // Claude also patches T3 settings.json above. The explicit provider
549        // instance carries a unique display marker, so the cache probe proves
550        // T3 applied the gateway-backed wrapper rather than merely observing
551        // that the settings file exists.
552        if (providerBridgeLease.readiness) {
553          const endBarrier = ctx.timing?.begin("t3.prepare.runtime_barrier");
554          try {
555            await ensureRuntimeProviderReadyForTurn({
556              sandbox,
557              signal: ctx.signal,
558              readiness: providerBridgeLease.readiness,
559              barrierDeadlineMs: CLAUDE_BARRIER_DEADLINE_MS,
560              verifyDeadlineMs: CLAUDE_VERIFY_DEADLINE_MS,
561              providerLabel: "Claude",
562            });
563          } finally {
564            endBarrier?.();
565          }
566        }
567
568        // Codex subscription patches its per-run relay config into the sandbox's
569        // T3 settings.json above (provider_bridge). T3 only applies settings via
570        // an asynchronous settings-watch reconcile, so a turn dispatched before
571        // that reconcile binds to the pre-reconcile, relay-less codex instance and
572        // falls back to a local, unauthenticated app-server (no first activity).
573        // Scoped to subscription Codex. Provider-gateway Codex does not create
574        // a per-run instance, and Claude has its own marker barrier above. The
575        // no-first-activity watchdog below remains the final safety net.
576        if (providerBridgeLease?.authPath === "subscription") {
577          const endBarrier = ctx.timing?.begin("t3.prepare.runtime_barrier");
578          try {
579            // (B) Barrier: wait for the reconcile to publish the subscription
580            // (relay-backed) codex instance into its status cache. Content, not
581            // mtime: health refreshes rewrite the cache for the legacy instance
582            // too. Fast path, no restart cost.
583            if (
584              !(await awaitCodexProviderReady(sandbox, ctx.signal, CODEX_BARRIER_DEADLINE_MS))
585            ) {
586              // (A) Fallback: the reconcile did not land in time. Bounce T3 so boot
587              // reads the relay config synchronously and builds the remote instance
588              // from the start, then verify once before steering. Honest error if
589              // the runtime never reports ready.
590              await restartRuntimeEnvironment(sandbox, ctx.signal);
591              invalidateRuntimeEnvironmentAccess(sandbox);
592              if (
593                !(await awaitCodexProviderReady(sandbox, ctx.signal, CODEX_VERIFY_DEADLINE_MS))
594              ) {
595                throw new Error("Codex runtime did not become ready after restart");
596              }
597            }
598          } finally {
599            endBarrier?.();
600          }
601        }
602
603        const endShell = ctx.timing?.begin("t3.shell");
604        const shell = await requestRuntimeEnvironment<RuntimeShellSnapshot>(
605          sandbox,
606          { method: "GET", path: "/api/orchestration/shell" },
607          ctx.signal,
608        );
609        endShell?.();
610        const threadId = runtimeThreadId(ctx);
611        const threadExists = shell.threads.some((thread) => thread.id === threadId);
612        const createdAt = new Date().toISOString();
613        const runtimeMode = configuredRuntimeMode();
614        const negotiatedCapabilities = sessionCapabilities(engine, {
615          desktop: false,
616          knowledgeTools: true,
617          runtimeOrchestration: true,
618        });
619        const executionCapabilities = buildExecutionCapabilitySnapshot({
620          runtime: "sandbox",
621          workspaceRoot: workdir,
622          gatewayAvailable: true,
623          desktopAvailability: "on_demand",
624        });
625        const established = await establishProviderSession({
626          driver,
627          ctx,
628          runtime: { kind: "sandbox", id: sandbox.id },
629          capabilities: negotiatedCapabilities,
630          executionCapabilities,
631          generation: T3_SESSION_GENERATION,
632          authEpoch: providerBridgeLease.authEpoch,
633          priorSessionId: threadExists ? threadId : undefined,
634          startMetadata: { workspaceRoot: workdir, runtimeMode, createdAt },
635          persistSession: async (providerSession) => {
636            if (!ctx.saveProviderSession) {
637              throw new Error("Session persistence is unavailable");
638            }
639            await ctx.saveProviderSession(providerSession, providerBridgeLease.authEpoch);
640          },

415async function waitForRuntimeTurn(
416  ctx: EngineRunContext,
417  sandbox: Awaited<ReturnType<typeof acquireThreadSandbox>>["sandbox"],
418  preExistingActivities: ReadonlyMap<string, string>,
419  priorTurnId: string | null,
420  redact: ReturnType<typeof createSecretRedactor>,
421): Promise<string> {
422  const activityRevisions = new Map(preExistingActivities);
423  const activitySteps = new Map<string, string>();
424  // Single owner of the turn-stream no-progress bound: a provider retry storm
425  // (only runtime.warning activities, no tool/text progress) must terminate
426  // the run with the real provider reason instead of running forever.
427  const watchdog = createNoProgressWatchdog(runtimeNoProgressTimeoutMs(), redact.text);
428  // A long-running tool emits no new activity revisions while it executes, so
429  // the event stream goes silent even though the turn is making real progress.
430  // While the latest snapshot shows an open tool call, tick the watchdog on a
431  // timer; provider stalls (no tool running, no text) stay fully guarded.
432  let toolInFlight = false;
433  const toolHeartbeat = setInterval(() => {
434    if (toolInFlight) {
435      watchdog.observeProgress();
436      ctx.reportActivity?.();
437    }
438  }, 15_000);
439  toolHeartbeat.unref?.();
440  let publishedText = "";
441  let finalText = "";
442  const applySnapshot = async (snapshot: RuntimeThreadSnapshot): Promise<boolean> => {
443    toolInFlight = hasOpenRuntimeToolCall(snapshot.thread.activities);
444    for (const activity of snapshot.thread.activities) {
445      const revision = runtimeActivityRevision(activity);
446      if (activityRevisions.get(activity.id) === revision) continue;
447      activityRevisions.set(activity.id, revision);
448      await recordProviderEvent(runtimeActivityProviderEvent(
449        ctx,
450        runtimeThreadId(ctx),
451        activity,
452        redact,
453      ), {
454        critical:
455          activity.kind === "user-input.requested" || activity.kind === "approval.requested",
456      });
457      watchdog.observeActivity(activity);
458      if (!shouldProjectRuntimeActivity(activity, snapshot.thread.activities)) continue;
459      const step = redact.unknown(activityStep(activity, runtimeThreadId(ctx)));
460      const activityStepKey = runtimeActivityStepKey(activity);
461      const priorStepId = activitySteps.get(activityStepKey);
462      if (priorStepId && ctx.updateStep) {
463        await ctx.updateStep(priorStepId, step.code_json ?? null);
464      } else {
465        const stepId = await ctx.emit(step);
466        if (stepId) activitySteps.set(activityStepKey, stepId);
467      }
468    }
469
470    const text = redact.text(assistantText(snapshot));
471    const settled = runtimeTurnSettled(snapshot);
472    const projection = projectRuntimeAssistantText({ publishedText, finalText }, text, settled);
473    const delta = projection.delta;
474    if (delta) {
475      ctx.publishDelta?.(delta);
476      watchdog.observeProgress();
477    }
478    publishedText = projection.publishedText;
479    finalText = projection.finalText;
480
481    const error = runtimeTurnError(snapshot);
482    if (error) throw new Error(redact.text(error));
483    return !settled;
484  };
485  // Dispatch commits before the read projection necessarily observes the new
486  // turn. Poll only this short projection hand-off; all subsequent updates use
487  // T3's native replayable websocket stream.
488  try {
489    const initial = await waitForNewRuntimeTurnSnapshot(ctx, sandbox, priorTurnId);
490    if (await applySnapshot(initial)) {
491      await subscribeRuntimeThread(
492        sandbox,
493        runtimeThreadId(ctx),
494        initial.snapshotSequence,
495        AbortSignal.any([ctx.signal, watchdog.signal]),
496        async (item) => {
497          if (item.kind === "synchronized") return true;
498          return await applySnapshot(await readThreadSnapshot(ctx, sandbox));
499        },
500      );
501    }
502  } finally {
503    clearInterval(toolHeartbeat);
504    watchdog.dispose();
505  }
506  if (watchdog.signal.aborted) throw watchdog.signal.reason;
507  ctx.signal.throwIfAborted();
508  return await drainRuntimeTerminalOutput({
509    initialText: finalText,
510    fallbackText: publishedText,
511    signal: ctx.signal,
512    readAndApplySnapshot: async (drainSignal) => {
513      await applySnapshot(await readRuntimeTerminalSnapshot(ctx, sandbox, drainSignal));
514      return finalText;
515    },
516  });
517}

20export const CHILD_SESSION_TOOLS = [
21  {
22    name: "child_session_create",
23    description:
24      "Create a durable child session under the current live run. Children do NOT run in parallel with this turn: each child is queued as a deferred serial thread turn that starts only after the current turn settles, so plan to finish this turn and read child results in a later turn (via child_session_list/events/gather). Identity, thread, engine, model, repositories, and memory scope are derived only from the signed gateway capability and current run. Creation is idempotent by idempotencyKey.",
25    inputSchema: {
26      type: "object",
27      properties: {
28        idempotencyKey: {
29          type: "string",
30          description:
31            "Stable caller-chosen key for this child task within the current run.",
32        },
33        prompt: {
34          type: "string",
35          description: `Child task prompt, bounded to ${MAX_PROMPT_CHARS} characters.`,
36        },
37      },
38      required: ["idempotencyKey", "prompt"],
39      additionalProperties: false,
40    },
41  },
42  {
43    name: "child_session_list",
44    description:
45      "List durable child sessions for this live thread. Results are bounded and include event references for reconnect/reload inspection.",
46    inputSchema: {
47      type: "object",
48      properties: {
49        limit: {
50          type: "integer",
51          minimum: 1,
52          maximum: 20,
53          description: "Maximum child sessions to return.",
54        },
55      },
56      additionalProperties: false,
57    },
58  },
59  {
60    name: "child_session_events",
61    description:
62      "Read a bounded page of durable native events for one child session in this live thread.",
63    inputSchema: {
64      type: "object",
65      properties: {
66        childRunId: {
67          type: "string",
68          description: "Child run id returned by child_session_create/list.",
69        },
70        cursor: {
71          type: "integer",
72          description:
73            "Last native event seq already seen. Omit or use -1 for the first page.",
74        },
75        limit: {
76          type: "integer",
77          minimum: 1,
78          maximum: 50,
79          description: "Maximum native events to return.",
80        },
81      },
82      required: ["childRunId"],
83      additionalProperties: false,
84    },
85  },
86  {
87    name: "child_session_gather",
88    description:
89      "Gather a bounded status summary for durable child sessions in this live thread, returning event counts and references rather than full transcripts.",
90    inputSchema: {
91      type: "object",
92      properties: {
93        limit: {
94          type: "integer",
95          minimum: 1,
96          maximum: 20,
97          description: "Maximum child sessions to summarize.",
98        },
99      },
100      additionalProperties: false,
101    },
102  },
103] as const;

97export async function createChildSession(input: {
98  readonly orgId: string;
99  readonly actorId: string | null;
100  readonly parentRunId: string;
101  readonly threadId: string;
102  readonly prompt: string;
103  readonly engine: EngineId;
104  readonly model: string;
105  readonly repos: readonly string[];
106  readonly memoryScope: MemoryScope;
107  readonly idempotencyKey: string;
108}): Promise<{ readonly status: "created" | "replayed"; readonly child: ChildSessionSummary } | { readonly status: "conflict" }> {
109  const runId = crypto.randomUUID();
110  const idempotencyKey = childKey(
111    input.threadId,
112    input.parentRunId,
113    input.idempotencyKey,
114  );
115  const intent: RunCommandIntent = {
116    prompt: input.prompt,
117    model: input.model,
118    engine: input.engine,
119    parentRunId: input.parentRunId,
120    // Child sessions inherit the parent's already-authorized resources. They
121    // cannot make an explicit repository selection of their own.
122    requestedRepos: [],
123    requestedResources: [],
124    attachmentIds: [],
125    memoryScope: input.memoryScope,
126    skillId: null,
127    skillVersion: null,
128    commandName: null,
129    commandProvider: null,
130    commandSessionId: null,
131    commandCatalogRevision: null,
132  };
133  const parent = await getRunForOrg(input.orgId, input.parentRunId);
134  if (!parent || parent.threadId !== input.threadId) {
135    throw new Error("child session parent is not available in this thread");
136  }
137  const internalOrigin = isInternalRunOrigin(parent.origin) ? parent.origin : null;
138  let accepted = internalOrigin
139    ? await preflightInternalRunCommandReplay({
140      orgId: input.orgId,
141      idempotencyKey,
142      intent,
143      origin: internalOrigin,
144    })
145    : await preflightRunCommandReplay({
146      orgId: input.orgId,
147      idempotencyKey,
148      intent,
149    });
150  if (accepted?.status === "conflict") return { status: "conflict" };
151
152  if (!accepted) {
153    const inheritedResources =
154      parent.resolvedResources.length > 0
155        ? parent.resolvedResources
156        : legacyParentResources(parent.repos, "api");
157    const intake = await resolveRunIntake(
158      {
159        source: "api",
160        // Child prompts are agent-authored delegation text, not direct user
161        // input, so they cannot discover or widen resource scope.
162        text: "",
163        inheritedResources,
164      },
165      { authorize: createRunResourceAuthorization(input.orgId) },
166    );
167    const commandInput = {
168      idempotencyKey,
169      orgId: input.orgId,
170      actorId: input.actorId,
171      intent,
172      run: {
173        id: runId,
174        prompt: input.prompt,
175        model: input.model,
176        engine: input.engine,
177        parentRunId: input.parentRunId,
178        threadId: input.threadId,
179        repos: [...intake.repos],
180        resolvedResources: intake.resources,
181        attachmentIds: [],
182        memoryScope: input.memoryScope,
183        skillId: null,
184        skillVersion: null,
185        skillContentHash: null,
186        commandName: null,
187        commandProvider: null,
188        commandSessionId: null,
189        commandCatalogRevision: null,
190      },
191    };
192    accepted = internalOrigin
193      ? await acceptInternalRunCommand({ ...commandInput, origin: internalOrigin })
194      : await acceptRunCommand(commandInput);
195  }
196  if (accepted.status === "conflict") return { status: "conflict" };
197  const child = await getChildSession(input.orgId, input.threadId, accepted.runId);
198  if (!child) throw new Error(`Accepted child session ${accepted.runId} was not readable`);
199  return { status: accepted.status, child };

202export async function listChildSessions(input: {
203  readonly orgId: string;
204  readonly threadId: string;
205  readonly limit?: unknown;
206}): Promise<readonly ChildSessionSummary[]> {
207  const rows = await db
208    .select({
209      id: runs.id,
210      parentRunId: runs.parentRunId,
211      threadId: runs.threadId,
212      status: runs.status,
213      prompt: runs.prompt,
214      engine: runs.engine,
215      model: runs.model,
216      createdAt: runs.createdAt,
217      updatedAt: runs.updatedAt,
218    })
219    .from(commands)
220    .innerJoin(runs, eq(commands.runId, runs.id))
221    .where(
222      and(
223        eq(commands.orgId, input.orgId),
224        eq(commands.threadId, input.threadId),
225        like(commands.idempotencyKey, `${childKeyPrefix(input.threadId)}%`),
226      ),
227    )
228    .orderBy(desc(runs.createdAt), desc(runs.id))
229    .limit(childSessionLimit(input.limit));
230  return rows.map(toSummary);
231}
232
233export async function getChildSession(
234  orgId: string,
235  threadId: string,
236  childRunId: string,
237): Promise<ChildSessionSummary | null> {
238  const [row] = await db
239    .select({
240      id: runs.id,
241      parentRunId: runs.parentRunId,
242      threadId: runs.threadId,
243      status: runs.status,
244      prompt: runs.prompt,
245      engine: runs.engine,
246      model: runs.model,
247      createdAt: runs.createdAt,
248      updatedAt: runs.updatedAt,
249    })
250    .from(commands)
251    .innerJoin(runs, eq(commands.runId, runs.id))
252    .where(
253      and(
254        eq(commands.orgId, orgId),
255        eq(commands.threadId, threadId),
256        eq(commands.runId, childRunId),
257        like(commands.idempotencyKey, `${childKeyPrefix(threadId)}%`),
258      ),
259    )
260    .limit(1);
261  return row ? toSummary(row) : null;
262}
263
264export async function listChildSessionEvents(input: {
265  readonly orgId: string;
266  readonly threadId: string;
267  readonly childRunId: string;
268  readonly cursor?: unknown;
269  readonly limit?: unknown;
270}): Promise<ChildSessionEventPage | null> {
271  const child = await getChildSession(input.orgId, input.threadId, input.childRunId);
272  if (!child) return null;
273  const cursor = typeof input.cursor === "number" && Number.isInteger(input.cursor) ? input.cursor : -1;
274  const limit = childSessionEventLimit(input.limit);
275  const rows = (await getNativeFramesSince(input.childRunId, cursor)).slice(0, limit + 1);
276  const hasMore = rows.length > limit;
277  const events = rows.slice(0, limit);
278  const last = events.at(-1);
279  return {
280    childRunId: input.childRunId,
281    events,
282    nextCursor: hasMore && last ? last.seq : null,
283    eventRef: child.eventRef,
284  };
285}
286
287export async function gatherChildSessions(input: {
288  readonly orgId: string;
289  readonly threadId: string;
290  readonly limit?: unknown;
291}): Promise<ReadonlyArray<ChildSessionSummary & {
292  readonly eventCount: number;
293  readonly latestEventTypes: readonly string[];
294}>> {
295  const children = await listChildSessions(input);
296  if (children.length === 0) return [];
297  const rows = await db
298    .select({
299      runId: providerEvents.runId,
300      count: sql<number>`count(*)::int`,
301      latestTypes: sql<string[]>`(array_remove(array_agg(${providerEvents.eventType} order by ${providerEvents.seq} desc), null))[1:5]`,
302    })
303    .from(providerEvents)
304    .where(inArray(providerEvents.runId, children.map((child) => child.id)))
305    .groupBy(providerEvents.runId)
306    .orderBy(asc(providerEvents.runId));
307  const byRun = new Map(rows.map((row) => [row.runId, row]));
308  return children.map((child) => {
309    const row = byRun.get(child.id);
310    return {
311      ...child,
312      eventCount: row?.count ?? 0,
313      latestEventTypes: row?.latestTypes ?? [],
314    };
315  });
316}

135export interface FleetClient {
136  readonly baseUrl: string;
137  /** Submit one task as a run (POST /api/runs). Throws AgentClientError on failure. */
138  dispatch(task: FleetTask): Promise<DispatchedRun>;
139  /** Fan tasks out with bounded concurrency. Resolves per-task; never rejects wholesale. */
140  dispatchMany(tasks: readonly FleetTask[], options?: DispatchManyOptions): Promise<DispatchOutcome[]>;
141  /** Atomically accept 1-20 tasks as one durable server-owned batch. */
142  dispatchBatch(tasks: readonly FleetTask[], options: DispatchBatchOptions): Promise<FleetBatch>;
143  /** Read the durable batch and its ordered child queue state. */
144  getBatch(batchId: string): Promise<FleetBatch>;
145  /** One thread read: the run's current status + answer, without waiting. */
146  getRun(runId: string): Promise<RunSnapshot>;
147  /** Poll the thread endpoint until the run is completed/failed (or times out). */
148  awaitSettled(runId: string, options?: AwaitSettledOptions): Promise<SettledRun>;
149  /** Post a QC reply run asking `verifierPrompt`, settle it, and parse its verdict. */
150  verify(runId: string, verifierPrompt: string, options?: AwaitSettledOptions): Promise<VerifyResult>;
151  /** Recent run summaries (newest first) for the authenticated org. */
152  listRecent(limit?: number): Promise<readonly ApiRunSummary[]>;
153  /** The web session URL for a run id. */
154  urlFor(runId: string): string;
155}
156
157/** The hosted org origin used when no baseUrl / USEAGENT_BASE_URL is given. The ONE
158 *  place this domain is written - the CLI, the MCP server, and the README all resolve
159 *  through here, so moving the product domain is a single-line change. */
160export const DEFAULT_BASE_URL = "https://app.useagent.org";
161export const DEFAULT_CONCURRENCY = 4;
162export const MAX_FLEET_CONCURRENCY = 20;
163export const MAX_FLEET_TASKS = 100;
164export const MAX_DURABLE_BATCH_TASKS = 20;
165export const DEFAULT_SETTLE_TIMEOUT_MS = 15 * 60 * 1000;
166export const DEFAULT_POLL_MS = 3 * 1000;
167
168export function validateFleetConcurrency(value: number): number {
169  if (!Number.isInteger(value) || value < 1 || value > MAX_FLEET_CONCURRENCY) {
170    throw new RangeError(`concurrency must be an integer between 1 and ${MAX_FLEET_CONCURRENCY}`);
171  }
172  return value;
173}
174
175/** The run's web session URL: `<baseUrl>/session/<runId>` (a root run threads under its
176 *  own id, so runId == threadId). Pure; mirrors the backend `sessionUrl` shape. */
177export function fleetRunUrl(baseUrl: string, runId: string): string {
178  return `${baseUrl.replace(/\/+$/, "")}/session/${runId}`;

72function normalizeTask(value: unknown): FleetBatchTaskInput | null {
73  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
74  const task = value as Record<string, unknown>;
75  if (!hasOnlyKeys(task, TASK_KEYS)) return null;
76
77  const prompt = typeof task.prompt === "string" ? task.prompt.trim() : "";
78  if (!prompt) return null;
79
80  let engine: EngineId | null = null;
81  if (task.engine !== undefined && task.engine !== null && task.engine !== "") {
82    if (typeof task.engine !== "string" || !(ENGINE_IDS as readonly string[]).includes(task.engine)) {
83      return null;
84    }
85    engine = task.engine as EngineId;
86  }
87
88  const model = task.model === undefined || task.model === null || task.model === ""
89    ? null
90    : typeof task.model === "string" && task.model.trim()
91      ? task.model.trim()
92      : null;
93  if (task.model !== undefined && task.model !== null && task.model !== "" && model === null) {
94    return null;
95  }
96
97  const rawRepos = task.repos ?? [];
98  if (!Array.isArray(rawRepos)) return null;
99  const repos = rawRepos.map((repo) => typeof repo === "string" ? repo.trim() : "");
100  if (repos.some((repo) => !repo) || new Set(repos).size !== repos.length) return null;
101
102  return { prompt, engine, model, repos };
103}
104
105/** Parse the narrow public request shape and fingerprint only caller-owned
106 * intent. Server trust fields are rejected by the exact key allowlists. */
107export function validateFleetBatchBody(body: unknown): FleetBatchValidationResult {
108  if (!body || typeof body !== "object" || Array.isArray(body)) {
109    return { ok: false, status: 400, body: { error: "invalid_batch_body" } };
110  }
111  const record = body as Record<string, unknown>;
112  if (!hasOnlyKeys(record, BODY_KEYS) || !Array.isArray(record.tasks)) {
113    return { ok: false, status: 400, body: { error: "invalid_batch_body" } };
114  }
115  if (record.tasks.length < 1 || record.tasks.length > MAX_BATCH_ITEMS) {
116    return { ok: false, status: 400, body: { error: "batch_size_must_be_1_to_20" } };
117  }
118
119  const tasks: FleetBatchTaskInput[] = [];
120  for (const [index, raw] of record.tasks.entries()) {
121    const parsed = normalizeTask(raw);
122    if (!parsed) {
123      return { ok: false, status: 400, body: { error: "invalid_batch_task", index } };
124    }
125    tasks.push(parsed);
126  }
127  const canonical = JSON.stringify(tasks.map((task) => [
128    task.prompt,
129    task.engine,
130    task.model,
131    task.repos,
132  ]));
133  const fingerprint = new Bun.CryptoHasher("sha256").update(canonical).digest("hex");
134  return { ok: true, tasks, fingerprint };
135}

61export async function fanCommand(client: FleetClient, args: FanArgs, io: CommandIO): Promise<number> {
62  const tasks = parseTasksJsonl(await io.readFile(args.file));
63  io.err(`dispatching ${tasks.length} task(s), parallelism ${args.parallel}${args.qc ? ", QC on" : ""}...`);
64
65  const outcomes = await client.dispatchMany(tasks, { concurrency: args.parallel });
66  const results: FanResultLine[] = await Promise.all(
67    outcomes.map(async (outcome): Promise<FanResultLine> => {
68      if (!outcome.ok) {
69        return { prompt: outcome.task.prompt, runId: null, status: "dispatch_error", answer: "", url: null, error: outcome.error };
70      }
71      let settled: Awaited<ReturnType<FleetClient["awaitSettled"]>>;
72      try {
73        settled = await client.awaitSettled(outcome.run.runId);
74      } catch (error) {
75        return {
76          prompt: outcome.task.prompt,
77          runId: outcome.run.runId,
78          status: "settle_error",
79          answer: "",
80          url: outcome.run.url,
81          error: errorMessage(error),
82        };
83      }
84      const base: FanResultLine = {
85        prompt: outcome.task.prompt,
86        runId: outcome.run.runId,
87        status: settled.status,
88        answer: settled.answer,
89        url: settled.url,
90      };
91      if (!args.qc) return base;
92      // QC is a real reply run recorded in the SAME thread as the work it judges.
93      try {
94        const verified = await client.verify(outcome.run.runId, args.qc);
95        return { ...base, verdict: verified.verdict };
96      } catch (error) {
97        return {
98          ...base,
99          status: "verification_error",
100          verdict: "unknown",
101          error: errorMessage(error),
102        };
103      }
104    }),
105  );
106
107  const jsonl = serializeResults(results);
108  if (args.out) {
109    await io.writeFile(args.out, jsonl);
110    io.err(`wrote ${results.length} result(s) to ${args.out}`);
111  } else {
112    io.out(jsonl.trimEnd());
113  }
114  io.err("");
115  io.err(formatFanSummary(results));
116
117  const allGood = results.every(
118    (r) => r.status === "completed" && r.error === undefined && r.verdict !== "fail",
119  );
120  return allGood ? 0 : 1;
121}