useagent Fleet 与 CLI 深读

25export interface FleetClientConfig {
26  /** Origin of the hosted org. Defaults to {@link DEFAULT_BASE_URL}. */
27  baseUrl?: string;
28  /** Org API key. Sent verbatim as `Authorization: Bearer <apiKey>` on every request. */
29  apiKey: string;
30  /** Injected fetch (tests / non-standard runtimes). Defaults to the ambient global. */
31  fetch?: FetchLike;
32}
33
34/** One unit of work to dispatch as a run. Mirrors the accepted POST /api/runs body. */
35export interface FleetTask {
36  prompt: string;
37  engine?: string;
38  model?: string;
39  repos?: readonly string[];
40  /** Makes a lost-response retry safe (backend dedupes on this key). */
41  idempotencyKey?: string;
42}
43
44/** A run accepted by the backend, with its web session URL for a human to open. */
45export interface DispatchedRun {
46  readonly runId: string;
47  readonly status: string;
48  readonly url: string;
49}
50
51/** Per-task result of a fan-out: never throws, so one bad task cannot sink the batch. */
52export type DispatchOutcome =
53  | { readonly ok: true; readonly task: FleetTask; readonly run: DispatchedRun }
54  | { readonly ok: false; readonly task: FleetTask; readonly error: string };
55
56export interface DispatchManyOptions {
57  /** Max in-flight dispatches (default 4). Clamped to >= 1. */
58  concurrency?: number;
59  /** When set, a task with no explicit key gets `${idempotencyPrefix}-${index}`. */
60  idempotencyPrefix?: string;
61}
62
63export interface DispatchBatchOptions {
64  /** Required batch-level key. Retrying the same request with this key is safe. */
65  idempotencyKey: string;
66}
67
68export interface FleetBatchQueue {
69  readonly state: string | null;
70  readonly reason: string | null;
71}
72
73export interface FleetBatchRun {
74  readonly ordinal: number;
75  readonly runId: string;
76  readonly status: string;
77  readonly queue: FleetBatchQueue;
78  readonly url: string;
79}
80
81export interface FleetBatch {
82  readonly batchId: string;
83  readonly status: string;
84  readonly createdAt: string;
85  /** True only when POST returned a previously accepted idempotent request. */
86  readonly replayed: boolean;
87  /** Child runs in caller-request order. */
88  readonly runs: readonly FleetBatchRun[];
89}
90
91export interface AwaitSettledOptions {
92  /** Give up after this long and return status "timeout" (default 15 min). */
93  timeoutMs?: number;
94  /** Delay between thread polls (default 3s). */
95  pollMs?: number;
96  /** Clock seam (defaults to Date.now) - injected so tests stay deterministic. */
97  now?: () => number;
98  /** Sleep seam (defaults to setTimeout) - injected so tests stay deterministic. */
99  sleep?: (ms: number) => Promise<void>;
100  /** Called once per poll with the current run row (null until it first appears).
101   *  Lets a caller stream status transitions (e.g. `useagent run --watch`). */
102  onPoll?: (run: RunSummary | null) => void;
103}

199/** Run `fn` over `items` with at most `concurrency` in flight, preserving input order. */
200async function mapWithConcurrency<T, R>(
201  items: readonly T[],
202  concurrency: number,
203  fn: (item: T, index: number) => Promise<R>,
204): Promise<R[]> {
205  const results = new Array<R>(items.length);
206  let cursor = 0;
207  const worker = async (): Promise<void> => {
208    for (;;) {
209      const index = cursor++;
210      if (index >= items.length) return;
211      results[index] = await fn(items[index]!, index);
212    }
213  };
214  const size = Math.min(Math.max(1, concurrency), items.length);
215  await Promise.all(Array.from({ length: size }, () => worker()));
216  return results;
217}

137export async function resolveFleetBatchTasks(
138  orgId: string,
139  tasks: readonly FleetBatchTaskInput[],
140): Promise<FleetBatchResolveResult> {
141  const resolved: ResolvedFleetBatchTask[] = [];
142  for (const [index, task] of tasks.entries()) {
143    const engineResolution = resolveAcceptedEngine(task.engine);
144    if (!engineResolution.ok) {
145      return {
146        ok: false,
147        status: engineResolution.status,
148        body: { ...engineResolutionErrorBody(engineResolution), index },
149      };
150    }
151    const engine = engineResolution.engine;
152    const model = task.model ?? defaultModelForEngine(engine);
153    if (!isModelAllowedForEngine(engine, model)) {
154      return { ok: false, status: 400, body: { error: "model_not_allowed", engine, model, index } };
155    }
156    if (!modelProviderReadyForEngine(engine, model)) {
157      return {
158        ok: false,
159        status: 403,
160        body: { ...modelProviderReadinessErrorBody(engine, model), index },
161      };
162    }
163
164    try {
165      const intake = await resolveRunIntake(
166        {
167          source: "web",
168          text: task.prompt,
169          explicitResources: explicitRepositoryResources(task.repos),
170        },
171        { authorize: createRunResourceAuthorization(orgId) },
172      );
173      resolved.push({
174        input: task,
175        engine,
176        model,
177        repos: [...intake.repos],
178        resources: intake.resources,
179      });
180    } catch (error) {
181      if (error instanceof RunIntakeError) {
182        return {
183          ok: false,
184          status: error.code === "resource_unauthorized" ? 403 : 400,
185          body: { error: error.code, ...error.diagnostic, index },
186        };
187      }
188      throw error;
189    }
190  }
191  return { ok: true, tasks: resolved };
192}

256/** Commit the complete fan-out manifest and all accepted roots atomically. */
257export async function acceptFleetBatch(input: {
258  readonly orgId: string;
259  readonly actorId: string;
260  readonly idempotencyKey: string;
261  readonly requestFingerprint: string;
262  readonly tasks: readonly ResolvedFleetBatchTask[];
263}): Promise<FleetBatchAcceptance> {
264  const result = await db.transaction(async (tx) => {
265    await lockOrgBatchQueue(tx, input.orgId);
266    const replay = await preflightFleetBatch({
267      orgId: input.orgId,
268      actorId: input.actorId,
269      idempotencyKey: input.idempotencyKey,
270      requestFingerprint: input.requestFingerprint,
271      itemCount: input.tasks.length,
272    }, tx);
273    if (replay) return { created: false, batch: replay } as const;
274
275    await assertRunAdmissionOpen(tx);
276    const runIds = input.tasks.map(() => crypto.randomUUID());
277    for (const [index, task] of input.tasks.entries()) {
278      const run = acceptedRun(task, runIds[index]!);
279      const intent = commandIntent(task);
280      const payload = JSON.stringify({
281        prompt: run.prompt,
282        model: run.model,
283        engine: run.engine,
284        parentRunId: null,
285        threadId: run.threadId,
286        repos: run.repos,
287        resolvedResources: run.resolvedResources,
288        attachmentIds: [],
289        memoryScope: run.memoryScope,
290        intent,
291      }).slice(0, MAX_COMMAND_PAYLOAD);
292      await insertCommandWithRun({
293        commandId: crypto.randomUUID(),
294        idempotencyKey: null,
295        orgId: input.orgId,
296        actorId: input.actorId,
297        payloadFingerprint: runIntentFingerprint(intent),
298        payload,
299        run,
300        origin: null,
301        priority: 0,
302      }, tx);
303    }
304    return ensureFleetBatch({
305      orgId: input.orgId,
306      actorId: input.actorId,
307      idempotencyKey: input.idempotencyKey,
308      requestFingerprint: input.requestFingerprint,
309      runIds,
310    }, tx);
311  });
312
313  if (result.created) {
314    for (const run of result.batch.runs) {
315      publishRunLifecycleChange({
316        orgId: input.orgId,
317        threadId: run.runId,
318        runId: run.runId,
319        kind: "created",
320      });
321    }
322  }
323  return result;
324}

24export async function runCommand(client: FleetClient, args: RunArgs, io: CommandIO): Promise<number> {
25  const run = await client.dispatch({
26    prompt: args.prompt,
27    engine: args.engine,
28    model: args.model,
29    repos: args.repos,
30  });
31  io.out(`run ${run.runId}`);
32  io.out(run.url);
33  if (!args.watch) return 0;
34
35  let lastStatus = "";
36  const settled = await client.awaitSettled(run.runId, {
37    onPoll: (r) => {
38      if (r && r.status !== lastStatus) {
39        lastStatus = r.status;
40        io.err(`... ${r.status}`);
41      }
42    },
43  });
44  io.err(`status: ${settled.status}`);
45  if (settled.answer) io.out(settled.answer);
46  return settled.status === "completed" ? 0 : 1;
47}
48
49export async function statusCommand(client: FleetClient, args: StatusArgs, io: CommandIO): Promise<number> {
50  const snapshot = await client.getRun(args.runId);
51  if (snapshot.status === "unknown") {
52    io.err(`run ${args.runId} not found`);
53    return 1;
54  }
55  io.err(`status: ${snapshot.status}`);
56  io.out(snapshot.url);
57  if (snapshot.answer) io.out(snapshot.answer);
58  return snapshot.status === "failed" ? 1 : 0;
59}

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}