qmQM
Runtime Router

qm runtime router

这一页专拆 QM 如何从 Web turn 请求一路解析到具体 harness/model，并让 worker 用租约语义执行后台 run。

request.harness/model │ ▼ resolveRuntimeChoiceDurable │ approved + provider availability ▼ harness adapter runTurn │ ▼ worker complete/fail run
§ 00 · OVERVIEW

从请求到 harness 的硬边界

src/api/app-turn.ts:127 先定义 fallback harness，src/api/app-turn.ts:136 解析 org runtime，src/api/app-turn.ts:141 再解析目标 scope runtime。它不是把 UI 选项直接传给模型，而是先经过 durable config。

src/harness/harness-router.ts:63 读取 approved harnesses，src/harness/harness-router.ts:70 构造同步 view 复用同一套 resolve 逻辑，这让 durable 和内存路径不会分叉。

▸RuntimeChoice 结构很小，语义很重

src/harness/harness-router.ts · L7-54TypeScript
7export interface RuntimeChoice {
8  harnessId: HarnessId;
9  modelId: string;
10}
11
12export function resolveRuntimeChoice(
13  config: Pick<ScopedConfigStore, "getApprovedHarnesses" | "getRuntimeSelection" | "getBaseModel">,
14  orgScopeId: ScopeId,
15  scope: ScopeId,
16  fallback: RuntimeChoice,
17  requested?: Partial<RuntimeChoice>,
18): RuntimeChoice {
19  const approved = config.getApprovedHarnesses() ?? [fallback.harnessId];
20  const orgStored = config.getRuntimeSelection(orgScopeId);
21  const orgLegacy = config.getBaseModel(orgScopeId);
22  const configuredOrg =
23    orgStored && isHarnessId(orgStored.harnessId)
24      ? { harnessId: orgStored.harnessId, modelId: orgStored.modelId }
25      : { harnessId: fallback.harnessId, modelId: orgLegacy ?? fallback.modelId };
26  const firstApproved = approved.find(isHarnessId) ?? fallback.harnessId;
27  const safeFallback =
28    approved.includes(fallback.harnessId) && modelSupportedByHarness(fallback.modelId, fallback.harnessId)
29      ? fallback
30      : { harnessId: firstApproved, modelId: defaultModelForHarness(firstApproved, fallback.modelId) };
31  const org =
32    approved.includes(configuredOrg.harnessId) &&
33    modelSupportedByHarness(configuredOrg.modelId, configuredOrg.harnessId)
34      ? configuredOrg
35      : safeFallback;
36  const scopedStored = scope === orgScopeId ? null : config.getRuntimeSelection(scope);
37  const scopedLegacy = scope === orgScopeId ? null : config.getBaseModel(scope);
38  let inherited = org;
39  if (scopedStored && isHarnessId(scopedStored.harnessId)) {
40    inherited = { harnessId: scopedStored.harnessId, modelId: scopedStored.modelId };
41  } else if (scopedLegacy) {
42    inherited = { harnessId: fallback.harnessId, modelId: scopedLegacy };
43  }
44  const choice =
45    requested?.harnessId || requested?.modelId
46      ? { harnessId: requested.harnessId ?? inherited.harnessId, modelId: requested.modelId ?? inherited.modelId }
47      : inherited;
48  if (!approved.includes(choice.harnessId) || !modelSupportedByHarness(choice.modelId, choice.harnessId)) {
49    if (requested?.harnessId || requested?.modelId)
50      throw new NonRetryableTurnError(`runtime ${choice.harnessId}/${choice.modelId} is not approved`);
51    return org;
52  }
53  return choice;
54}
§ 01 · ROUTER

Adapter map 让多个 agent runtime 共存

src/harness/harness-router.ts:84 接收 adapters map，src/harness/harness-router.ts:95 在每个 turn 解析 choice，src/harness/harness-router.ts:105 把 harness/model 注入 runTurn。这个封装让 Codex、Claude、OpenCode、Pi 等 runtime 可以被同一层调度。

▸切换 harness 时主动 reset session

src/harness/harness-router.ts · L84-116TypeScript
84export function createHarnessRouter(
85  adapters: ReadonlyMap<HarnessId, Harness>,
86  utility: Harness,
87  resolve: (input: HarnessTurnInput) => RuntimeChoice | Promise<RuntimeChoice>,
88): Harness {
89  const lastHarness = new Map<string, HarnessId>();
90  return {
91    profile: utility.profile,
92    models: utility.models,
93    tools: utility.tools,
94    turns: {
95      async runTurn(input) {
96        const choice = await resolve(input);
97        const adapter = adapters.get(choice.harnessId);
98        if (!adapter) throw new Error(`harness ${choice.harnessId} is unavailable`);
99        const prior = lastHarness.get(input.session.id);
100        if (prior && prior !== choice.harnessId) {
101          await adapters.get(prior)?.turns.resetSession?.(input.session.id);
102          await adapter.turns.resetSession?.(input.session.id);
103        }
104        lastHarness.set(input.session.id, choice.harnessId);
105        return adapter.turns.runTurn({ ...input, harness: choice.harnessId, model: choice.modelId });
106      },
107      async resetSession(sessionId) {
108        lastHarness.delete(sessionId);
109        await Promise.all([...adapters.values()].map((adapter) => adapter.turns.resetSession?.(sessionId)));
110      },
111      async close() {
112        await Promise.all([...new Set(adapters.values())].map((adapter) => adapter.turns.close?.()));
113      },
114    },
115  };
116}

src/harness/harness-router.ts:99 读取上一轮 harness，src/harness/harness-router.ts:101 reset prior，src/harness/harness-router.ts:102 reset current。这个小细节防止模型运行时之间共享不兼容的 session 状态。

§ 02 · WEB TURN

Web 入口同时检查成员、模型和 provider

src/api/app-turn.ts:108 处理 web thread，src/api/app-turn.ts:116 防止同一个 threadRef 被不同 scope 复用，src/api/app-turn.ts:151 拒绝非法 harness 字符串。这些都是模型调用前的系统边界。

▸Web turn runtime gate

src/api/app-turn.ts · L122-190TypeScript
122        const org = scopeId("org", orgIdOf());
123        const targetScope =
124          req.conversation.kind === "dm"
125            ? scopeId("personal", actor.id)
126            : scopeId(req.conversation.kind, req.conversation.channelRef ?? threadRef);
127        const fallbackHarness = isHarnessId(deps.harnessId) ? deps.harnessId : "pi";
128        const runtimeFallback = deps.runtimeFallback ?? {
129          harnessId: fallbackHarness,
130          modelId: defaultModelForHarness(fallbackHarness),
131        };
132        let orgRuntime;
133        let configuredRuntime;
134        let runtime;
135        try {
136          orgRuntime = await resolveRuntimeChoiceDurable(deps.config, org, org, runtimeFallback);
137          configuredRuntime =
138            targetScope === org
139              ? orgRuntime
140              : await resolveRuntimeChoiceDurable(deps.config, org, targetScope, runtimeFallback);
141          runtime =
142            req.harness || req.model
143              ? await resolveRuntimeChoiceDurable(deps.config, org, targetScope, runtimeFallback, {
144                  ...(req.harness && isHarnessId(req.harness) ? { harnessId: req.harness } : {}),
145                  ...(req.model ? { modelId: req.model } : {}),
146                })
147              : configuredRuntime;
148        } catch (error) {
149          return { status: "refused", reason: errMessage(error) };
150        }
151        if (req.harness && !isHarnessId(req.harness)) {
152          return { status: "refused", reason: `runtime ${req.harness} is not approved` };
153        }
154        const configuredKeys = deps.providerKeys ??
155          deps.modelProviders ?? { anthropic: false, openai: false, openrouter: false };
156        let providers = deps.modelProviders;
157        if (deps.modelCredentials) {
158          providers = modelProviderAvailabilityFor(
159            runtime.harnessId,
160            configuredKeys,
161            await deps.modelCredentials.availability(),
162          );
163        } else if (deps.providerKeys) {
164          providers = modelProviderAvailabilityFor(runtime.harnessId, configuredKeys);
165        }
166        if (providers && !modelServiceable(runtime.modelId, providers)) {
167          return {
168            status: "refused",
169            reason: "that model isn't available on this deployment (its provider isn't configured)",
170          };
171        }
172        const configuredWebuiModels = await deps.config.getWebuiModelsDurable(org);
173        let enabledWebuiModels: string[] | null = null;
174        if (configuredWebuiModels?.length) {
175          enabledWebuiModels = [...new Set([...configuredWebuiModels, orgRuntime.modelId])];
176        } else if (providers?.openrouter) {
177          enabledWebuiModels = [
178            ...new Set([
179              ...selectableCatalogForHarness(
180                await selectableModelCatalog(deps.modelCredentialFetch),
181                runtime.harnessId,
182              ).map((model) => model.id),
183              ...(orgRuntime.harnessId === runtime.harnessId ? [orgRuntime.modelId] : []),
184            ]),
185          ];
186        }
187        const invalidModelOption =
188          validateWebTurnModelOptions(req, enabledWebuiModels, providers) ??
189          webTurnRuntimeModelRefusal(runtime.modelId, orgRuntime.modelId, configuredWebuiModels);
190        if (invalidModelOption) return { status: "refused", reason: invalidModelOption };

src/api/app-turn.ts:154 合并 provider keys，src/api/app-turn.ts:157 查询 model credential availability，src/api/app-turn.ts:166 拒绝 provider 未配置的模型。它把“用户想用哪个模型”和“部署真的能服务哪个模型”分开处理。

§ 03 · WORKER

Worker 把 runtime choice 放进可靠执行循环

src/runs/worker.ts:59 计算 queueMs，src/runs/worker.ts:60 调 orchestrator.handleTurn，src/runs/worker.ts:71 complete run；异常路径在 src/runs/worker.ts:75 fail 并设置 retry。

▸processRun 的 heartbeat/cancel 结构

src/runs/worker.ts · L22-82TypeScript
22export async function processRun(deps: ProcessDeps, run: Run, opts?: { background?: boolean }): Promise<TurnResult> {
23  const token = run.leaseToken;
24  if (token === null) throw new Error(`processRun called with an unleased run ${run.id}`);
25  const intervalMs = deps.heartbeatIntervalMs ?? Math.max(1_000, Math.floor(deps.leaseTtlMs / 3));
26  const cancel = new AbortController();
27  let consecutiveLost = 0;
28  let leaseLost = false;
29  const beat = setInterval(() => {
30    void deps.runs
31      .heartbeat(run.id, token, deps.leaseTtlMs)
32      .then((alive) => {
33        if (alive) {
34          consecutiveLost = 0;
35          return;
36        }
37        consecutiveLost += 1;
38        if (consecutiveLost >= LEASE_LOST_CONSECUTIVE && !leaseLost) {
39          leaseLost = true;
40          clearInterval(beat);
41          console.warn(
42            `[worker] run ${run.id} lost its lease after ${consecutiveLost} consecutive beats; cancelling the in-process turn`,
43          );
44          cancel.abort();
45        }
46      })
47      .catch((err: unknown) => {
48        consecutiveLost = 0;
49        console.warn(`[worker] heartbeat failed for run ${run.id} (transient, ignored): ${errMessage(err)}`);
50      });
51  }, intervalMs);
52  let beatStopped = false;
53  const stopBeat = (): void => {
54    if (beatStopped) return;
55    beatStopped = true;
56    clearInterval(beat);
57  };
58  try {
59    const queueMs = run.startedAt !== null ? Math.max(0, run.startedAt - run.createdAt) : undefined;
60    const result = await deps.orchestrator.handleTurn({
61      ...run.request,
62      origin: resolveTurnOrigin(run.request),
63      runId: run.id,
64      attempt: run.attempts,
65      finalAttempt: errorParks(run, deps.runs.maxClaims),
66      background: opts?.background ?? false,
67      cancel: cancel.signal,
68      ...(queueMs !== undefined ? { queueMs } : {}),
69    });
70    stopBeat();
71    await deps.runs.complete(run.id, token, result);
72    return result;
73  } catch (err) {
74    stopBeat();
75    await deps.runs.fail(run.id, token, errMessage(err), {
76      retry: !(err instanceof NonRetryableTurnError),
77    });
78    throw err;
79  } finally {
80    stopBeat();
81  }
82}

src/runs/worker.ts:29 的 heartbeat loop 和 src/runs/worker.ts:44 的 cancel.abort 让 lease 丢失能中止本地 turn。对后台自动化来说，这比“进程还活着就继续跑”安全得多。

▸claim loop 的退避和释放

src/runs/worker.ts · L101-187TypeScript
101export function createWorker(deps: WorkerDeps): Worker {
102  const workerId = deps.workerId ?? `w-${randomUUID().slice(0, 8)}`;
103  const pollMs = deps.pollMs ?? 50;
104  let stopped = false;
105  let loopDone: Promise<void> | null = null;
106  let inFlight: { runId: string; leaseToken: string; threadRef: string } | null = null;
107  let releasedLeaseToken: string | null = null;
108  let releasing: Promise<void> | null = null;
109
110  async function loop(): Promise<void> {
111    let claimFailures = 0;
112    while (!stopped) {
113      if (deps.canClaim && !deps.canClaim()) {
114        await sleep(pollMs);
115        continue;
116      }
117      let run: Run | null;
118      try {
119        run = await deps.runs.claim(workerId, deps.leaseTtlMs);
120        claimFailures = 0;
121      } catch (e) {
122        claimFailures += 1;
123        if (claimFailures >= CLAIM_FAIL_CRASH_CONSECUTIVE) throw e;
124        swallow("worker: claim failed (transient, retrying)", e);
125        await sleep(Math.min(pollMs * 2 ** Math.min(claimFailures, 5), 5_000));
126        continue;
127      }
128      if (!run) {
129        await sleep(pollMs);
130        continue;
131      }
132      if (stopped) {
133        if (run.leaseToken !== null)
134          await deps.runs
135            .releaseLease(run.id, run.leaseToken)
136            .catch((e) => swallow("worker: post-stop claim handback failed", e));
137        break;
138      }
139      inFlight =
140        run.leaseToken !== null ? { runId: run.id, leaseToken: run.leaseToken, threadRef: run.sessionId } : null;
141      deps.onClaimed?.();
142      try {
143        await processRun(deps, run, { background: true });
144      } catch (e) {
145        swallow("worker: background run crashed", e);
146      } finally {
147        inFlight = null;
148      }
149    }
150  }
151
152  return {
153    start() {
154      if (loopDone) return;
155      stopped = false;
156      loopDone = loop();
157    },
158    busy() {
159      return inFlight !== null;
160    },
161    releaseInFlight() {
162      const held = inFlight;
163      if (!held || held.leaseToken === releasedLeaseToken) return Promise.resolve();
164      if (releasing) return releasing;
165      releasing = (async () => {
166        try {
167          if (await deps.runs.heartbeat(held.runId, held.leaseToken, deps.leaseTtlMs)) {
168            const session = await deps.sessions.getByThread(held.threadRef);
169            if (session) await deps.sessions.forceReleaseLease(session.id);
170            await deps.runs.releaseLease(held.runId, held.leaseToken);
171          }
172          releasedLeaseToken = held.leaseToken;
173        } catch (e) {
174          swallow("worker: releaseInFlight failed", e);
175        } finally {
176          releasing = null;
177        }
178      })();
179      return releasing;
180    },
181    async stop(drainMs = STOP_DRAIN_MS) {
182      stopped = true;
183      await Promise.race([loopDone, sleep(drainMs, { unref: true })]);
184      loopDone = null;
185    },
186  };
187}

src/runs/worker.ts:119 claim，src/runs/worker.ts:125 exponential backoff，src/runs/worker.ts:161 releaseInFlight。三段合起来让 worker 更像队列消费者，而不是脆弱的定时轮询器。

§ 04 · EDGE

边界条件

▸非法 runtime 是非重试错误

src/harness/harness-router.ts:49 在 requested runtime 非法时抛 NonRetryableTurnError，避免队列重复重试一个永远不会成功的用户选择。

▸Web thread 绑定 scope

src/api/app-turn.ts:110 读取 existing session，src/api/app-turn.ts:116 对比 claimed scope，能防止同一 browser thread 被拿来穿越 scope。

§ 05 · HIGHLIGHTS

为什么这套路由值得学

▸模型选择是治理对象

src/api/app-turn.ts:172 读取 Web UI 模型配置，src/api/app-turn.ts:187 验证请求选项，这让模型列表可以面向部署动态变化。

§ 06 · RISKS

风险

▸状态多，排障要靠可观测性

src/runs/worker.ts:41 只有 console.warn，真实生产中还需要把 lease lost、claim fail 和 runtime refusal 全部接到指标与审计。

Generated · 2026-08-03Source · yc-software/qm← 返回 Trending 索引↑ 返回主页面