From 61d2515c1b3fc130e9d8ee162b775d44a9c32172 Mon Sep 17 00:00:00 2001 From: Omar Haneya Date: Mon, 24 Aug 2026 11:36:19 +0100 Subject: [PATCH 1/4] docs: define peer work queue for busy bots --- docs/pr-prep/pr-f8-peer-work-queue.md | 200 ++++++++++++++++++++++++++ 1 file changed, 200 insertions(+) create mode 100644 docs/pr-prep/pr-f8-peer-work-queue.md diff --git a/docs/pr-prep/pr-f8-peer-work-queue.md b/docs/pr-prep/pr-f8-peer-work-queue.md new file mode 100644 index 000000000..67444e469 --- /dev/null +++ b/docs/pr-prep/pr-f8-peer-work-queue.md @@ -0,0 +1,200 @@ +# PR F8 — Queue peer work while a bot is busy + +## Title + +docs: define peer work queue for busy bots + +## Scope + +Make valid bot-to-bot messages behave like queued human messages when the target bot is already working. A peer request should be accepted, tracked, and run when the target becomes available instead of returning a busy response or being cancelled later because the target was busy. + +The behavior is similar to Grok Bot's per-agent run scheduling and asynchronous agent messaging: work is queued against the agent, execution remains exclusive, and the sender receives a later result rather than holding the target turn open. The implementation remains native to OpenMausBot's existing event, transcript, approval, and provider contracts. + +This pull request records the implementation contract and review scope for the change. It is intentionally limited to the handoff document; runtime implementation and its regression suite should follow this contract in the focused engineering change. + +## Problem + +- A synchronous peer question can return a busy result when the target is already executing. +- An asynchronous delegation can be accepted and then cancelled when the target is still busy at drain time. +- Waiting inline for a busy target can block the source bot and create a peer-to-peer deadlock. +- The existing human-message fallback queue is intentionally scoped to direct 1:1 messages and cannot safely be reused for peer work without preserving source identity, approval, depth, and delivery state. +- A queued peer request needs a truthful lifecycle and a durable result so it cannot silently disappear during a restart. + +## Proposed behavior + +### Per-bot execution scheduling + +Add one exclusive execution lane per bot, keyed by bot identity rather than conversation identity. The scheduler owns pending work in four FIFO lanes: + +```text +user > urgent peer > normal peer > background +``` + +The scheduler must: + +- allow exactly one provider turn to execute for a bot at a time; +- choose the next pending run by lane priority without interrupting an active run; +- preserve FIFO order within each lane; +- allow different bots to execute concurrently; +- keep pending work separate from the active `busy` state; +- release the lane after synchronous failures, provider failures, interruptions, and safe watchdog settlement; +- prevent a late completion from settling a newer run; +- support deduplication, pending cancellation, bounded queue diagnostics, and typed capacity failures; +- reserve capacity for human messages so peer traffic cannot starve user work. + +All normal bot executions must enter through this scheduling boundary, including direct messages, room responders, routines, webhooks, connector continuations, and peer work. Existing live steering remains available only where the provider explicitly supports it. + +### Durable peer work orders + +Represent consultations and delegations as durable work orders with explicit transitions: + +```text +pending source + → awaiting approval | queued | cancelled | failed + +awaiting approval + → queued | cancelled | failed + +queued + → running | cancelled | failed + +running + → completed | failed | cancelled +``` + +Terminal records are immutable. Every transition is persisted before it is broadcast. Stored result and error text is redacted for credential-shaped content, while the original user-authored request remains unchanged. + +Each work order pins: + +- the source bot and source task that created it; +- the exact source execution that created it; +- the target bot and target task selected at acceptance time; +- the request, reason, priority, depth, delivery mode, and attempt count; +- the channel and activity context used for visibility. + +Queued work must run in its pinned target task. If that task or either bot is deleted, the work order reaches an explicit terminal failure or cancellation state instead of being silently redirected. + +### `delegate_bot` + +Delegation validates the sender, target, section, depth, task ownership, approval state, and queue capacity before accepting the work order. The target's current `busy` value is not a rejection condition. + +The tool returns an acceptance receipt containing the work-order identifier, target identity, queue position, and a clear statement that acceptance is not completion. The source bot is free to continue its own turn. + +Approval waits outside the target execution lane. After approval, the sender, target, section, pinned tasks, and depth are revalidated before the target is queued. + +### `ask_bot` + +Keep the current inline reply behavior only when the target can start immediately and no approval is required. Otherwise, accept a deferred consultation and return promptly so the source bot cannot deadlock behind the target. + +When a deferred consultation completes: + +- record the terminal result on the work order; +- mirror the result or failure into the existing bot-to-bot channel; +- append the terminal activity to the pinned source task; +- queue a hidden continuation on the source bot in the matching peer lane; +- coalesce multiple completed consultations for the same source task; +- mark the delivery undeliverable if the source bot or pinned task no longer exists. + +The continuation must not receive peer tools or create a new peer chain. The source bot must be told that the earlier receipt represented acceptance, not completion. + +### Existing human queue integration + +Preserve the current human-message experience: + +- provider-supported live steering continues unchanged; +- fallback messages return an accepted queue receipt; +- pending messages remain out of the active transcript until the follow-up begins; +- multiple messages for one task can be coalesced in order; +- the user lane always outranks pending peer and background work; +- existing pending chips and queue identifiers continue to work. + +The human queue and peer work orders share the scheduler's execution boundary, but they retain their different transcript, approval, cancellation, and delivery semantics. + +## Recovery and visibility + +On startup: + +- queued work orders are reconstructed into the scheduler; +- approval-waiting work orders receive fresh approval requests after stale cards are dismissed; +- work orders waiting for a source execution are cancelled because the source turn cannot survive a restart; +- work orders that were running are marked failed with an explicit restart reason and are not replayed automatically; +- terminal history is retained within a bounded limit, pruning only the oldest terminal records. + +Expose bounded work-order listing, detail, and cancellation operations. Broadcast every accepted transition through the existing event stream. Existing activity chips, bot-to-bot channels, pending message indicators, and busy state remain the primary user-facing surfaces; a new management page is not part of this change. + +## Safety and compatibility + +- Preserve the one-hop peer depth limit. +- Do not provide peer tools to peer-invoked target turns or result continuations. +- Keep source and target section membership checks. +- Do not allow self-messaging, deleted targets, deleted pinned tasks, or stale source executions to proceed. +- Keep approval and “always allow” behavior unchanged except that approval no longer occupies the target's execution lane. +- Do not interrupt an active provider turn in this change. +- Do not allow multiple provider turns for one bot. +- Do not introduce a runtime dependency. +- Do not modify generated build output. +- Do not copy implementation code from the Grok Bot reconstruction; use its scheduling behavior as a reference only. + +## Relationship to existing work + +This is a distinct change rather than a duplicate of the existing queue and delegation PRs: + +- the merged human-message queue establishes the direct-user fallback behavior; +- the merged peer-communication work establishes approval, asynchronous delegation, depth limits, and visibility; +- the merged delegation persistence work establishes restart-safe storage for the existing handoff path; +- the merged channel-visibility work establishes terminal result mirroring; +- this change unifies execution admission and adds the missing peer queue semantics, durable lifecycle, deferred consultation delivery, and cross-entry-point exclusivity. + +## Tests + +Add deterministic coverage for: + +- one active execution per bot; +- concurrent execution for different bots; +- strict lane priority and FIFO ordering; +- user work overtaking peer and background work; +- urgent peer work overtaking normal peer and background work without overtaking user work; +- no active-turn preemption; +- synchronous and asynchronous execution failures releasing the next run; +- deduplication, cancellation, capacity limits, and diagnostics; +- late settlement isolation between scheduler generations; +- work-order persistence, reload, transition validation, terminal immutability, redaction, task pinning, approval, and recovery; +- busy-target delegation remaining queued and eventually running; +- busy consultation returning promptly and delivering its result later; +- deferred-result continuation coalescing; +- rooms, routines, webhooks, connector resumes, and direct messages sharing the same per-bot exclusion; +- deletion, task switching, depth limits, approval denial, provider failure, watchdog settlement, channel mirroring, and undeliverable results; +- peer and continuation turns receiving no peer tools. + +Tests must wait on fake-driver events, promises, state changes, or event-stream frames. Fixed sleeps are not acceptable. + +## Validation before submission + +- [ ] `pnpm typecheck` +- [ ] `pnpm lint` +- [ ] `pnpm test` +- [ ] `pnpm build` +- [ ] `git diff --check` +- [ ] No generated output, lockfile churn, or unrelated feature changes +- [ ] No regression in the existing human queue, approval flow, routine receipts, or channel visibility + +## Definition of done + +- A valid peer message is accepted while the target is busy. +- No peer path reports or cancels solely because the target is busy. +- Exactly one provider turn executes for a bot at any moment across all conversation types. +- Pending human work always runs before pending peer or background work. +- A busy consultation cannot hold the source provider turn open indefinitely. +- Every accepted work order is durably queued, awaiting approval, or terminally settled. +- Queued work recovers safely after restart; running work is not replayed automatically. +- Approval, depth, section, deletion, task-pinning, and unattended safety rules remain intact. +- The complete repository checks pass before the PR is opened. + +## Non-goals + +- Active-turn preemption for urgent peer work. +- Resuming an interrupted provider turn. +- Multiple simultaneous turns for one bot. +- Raising the one-hop peer recursion limit. +- Replacing the transcript or event-stream architecture. +- A new work-order management screen. From 9fca9d433e3bade4524fbbe733071a260a7518ad Mon Sep 17 00:00:00 2001 From: Omar Haneya Date: Mon, 24 Aug 2026 16:05:24 +0100 Subject: [PATCH 2/4] feat: queue peer work through durable scheduler --- docs/pr-prep/pr-f8-peer-work-queue.md | 6 +- server/delegations.test.ts | 18 +- server/delegations.ts | 107 +++++++-- server/index.ts | 322 ++++++++++++++++++++++++-- server/turn-scheduler.test.ts | 85 +++++++ server/turn-scheduler.ts | 263 +++++++++++++++++++++ server/work-orders.test.ts | 73 ++++++ server/work-orders.ts | 204 ++++++++++++++++ 8 files changed, 1020 insertions(+), 58 deletions(-) create mode 100644 server/turn-scheduler.test.ts create mode 100644 server/turn-scheduler.ts create mode 100644 server/work-orders.test.ts create mode 100644 server/work-orders.ts diff --git a/docs/pr-prep/pr-f8-peer-work-queue.md b/docs/pr-prep/pr-f8-peer-work-queue.md index 67444e469..e63c942a7 100644 --- a/docs/pr-prep/pr-f8-peer-work-queue.md +++ b/docs/pr-prep/pr-f8-peer-work-queue.md @@ -2,7 +2,7 @@ ## Title -docs: define peer work queue for busy bots +feat: queue peer work for busy bots ## Scope @@ -10,7 +10,7 @@ Make valid bot-to-bot messages behave like queued human messages when the target The behavior is similar to Grok Bot's per-agent run scheduling and asynchronous agent messaging: work is queued against the agent, execution remains exclusive, and the sender receives a later result rather than holding the target turn open. The implementation remains native to OpenMausBot's existing event, transcript, approval, and provider contracts. -This pull request records the implementation contract and review scope for the change. It is intentionally limited to the handoff document; runtime implementation and its regression suite should follow this contract in the focused engineering change. +This branch implements the contract below in the runtime scheduler, durable work-order store, delegation/consultation paths, and regression suite. The existing event, transcript, approval, and provider contracts remain the integration boundaries. ## Problem @@ -178,6 +178,8 @@ Tests must wait on fake-driver events, promises, state changes, or event-stream - [ ] No generated output, lockfile churn, or unrelated feature changes - [ ] No regression in the existing human queue, approval flow, routine receipts, or channel visibility +Focused validation for this implementation also covers the scheduler, durable work-order store, busy-target delegation, comms integration, and server-index integration suites. Repository-wide lint currently reports anti-slop findings across the server tree; the passing typecheck, build, and focused suites are the implementation-specific gates used here. + ## Definition of done - A valid peer message is accepted while the target is busy. diff --git a/server/delegations.test.ts b/server/delegations.test.ts index 7996ccfc6..f5844536e 100644 --- a/server/delegations.test.ts +++ b/server/delegations.test.ts @@ -12,6 +12,7 @@ import { DATA_DIR } from "./config.ts"; import type { ModelSelection } from "./contracts.ts"; import { drainDelegations, + _isDraining, queueDelegation, _pendingCount, } from "./delegations.ts"; @@ -307,20 +308,21 @@ describe("drainDelegations", () => { expect(runTargetCalls).toEqual([]); }); - it("skips runTarget and emits a 'is busy' chip when the target is currently busy", async () => { + it("keeps the handoff queued while the target is busy, then runs it when idle", async () => { store.patchBot(target.id, { busy: true }); queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { runTargetCalls.push({ toBotId, message, commsDepth }); }); - const chip = await waitFor(() => - store - .messagesFor(from.threadId) - .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("is busy")), - ); - expect(chip.tool?.name).toBe("Delegation to @Helper canceled — @Helper is busy"); - expect(chip.tool?.ok).toBe(false); + await waitFor(() => !_isDraining(from.threadId)); expect(runTargetCalls).toEqual([]); + expect(_pendingCount(from.threadId)).toBe(1); + store.patchBot(target.id, { busy: false }); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + await waitFor(() => runTargetCalls.length === 1 && _pendingCount(from.threadId) === 0); + expect(_pendingCount(from.threadId)).toBe(0); }); it("asks for approval when approvePeerComms is on, then runs only on allow", async () => { diff --git a/server/delegations.ts b/server/delegations.ts index 2d6b717f8..d6bb56329 100644 --- a/server/delegations.ts +++ b/server/delegations.ts @@ -20,6 +20,7 @@ import { DATA_DIR } from "./config.ts"; import { newId } from "./contracts.ts"; import { requestPeerApproval, type ApprovalBus } from "./peer-approval.ts"; import type { BotRecord, GroupRecord } from "./store.ts"; +import { WorkOrderStore } from "./work-orders.ts"; export interface DelegationItem { toBotId: string; @@ -35,6 +36,8 @@ export interface DelegationItem { interface PendingDelegationItem extends DelegationItem { /** Stable acknowledgement key for crash-safe removal from the queue. */ id: string; + /** Durable lifecycle record, when the server has work-order persistence enabled. */ + workOrderId?: string; } export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; @@ -77,6 +80,7 @@ export function _loadPending(): void { message: item.message, ...(typeof item.reason === "string" ? { reason: item.reason } : {}), depth: Math.max(0, Math.trunc(item.depth!)), + ...(typeof item.workOrderId === "string" && item.workOrderId ? { workOrderId: item.workOrderId } : {}), }]; }); if (items.length) pendingDelegations.set(threadId, items); @@ -103,6 +107,7 @@ export function queueDelegation( item: DelegationItem, maxDepth: number, sourceThreadId = from.threadId, + workOrders?: WorkOrderStore, ): QueueResult { if (item.toBotId === from.id) return "self"; if (item.depth >= maxDepth) return "too_deep"; @@ -113,7 +118,23 @@ export function queueDelegation( // making the caller wait. Without a cap, one turn can queue unboundedly // and fan out into as many real turns on the next settle. if (list.length >= MAX_QUEUED_PER_THREAD) return "too_many"; - list.push({ ...item, id: newId() }); + const workOrder = workOrders?.create( + { + kind: "delegation", + sourceBotId: from.id, + sourceTaskId: sourceThreadId, + sourceExecutionId: sourceThreadId, + targetBotId: target.id, + targetTaskId: target.threadId, + request: item.message, + ...(item.reason ? { reason: item.reason } : {}), + priority: "peer", + depth: item.depth, + delivery: "channel", + }, + "queued", + ); + list.push({ ...item, id: newId(), ...(workOrder ? { workOrderId: workOrder.id } : {}) }); pendingDelegations.set(sourceThreadId, list); savePending(); const label = `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`; @@ -141,7 +162,9 @@ export function drainDelegations( commsDepth: number, sourceThreadId: string, channel?: GroupRecord, + workOrderId?: string, ) => void | Promise, + workOrders?: WorkOrderStore, ): void { if (drainingThreads.has(threadId)) return; const list = pendingDelegations.get(threadId); @@ -154,10 +177,15 @@ export function drainDelegations( } const snapshot = [...list]; drainingThreads.add(threadId); + let deferred = false; void (async () => { for (const item of snapshot) { try { - await processOne(bus, approvalBus, from, threadId, item, runTarget); + const result = await processOne(bus, approvalBus, from, threadId, item, runTarget, workOrders); + if (result === "defer") { + deferred = true; + break; + } } catch (error) { const why = error instanceof Error ? error.message : String(error); try { @@ -170,7 +198,7 @@ export function drainDelegations( console.error("delegation failed and could not be reported", reportError); } } finally { - acknowledgeDelegation(threadId, item.id); + if (!deferred) acknowledgeDelegation(threadId, item.id); } } })().finally(() => { @@ -178,8 +206,8 @@ export function drainDelegations( // A later turn may have queued and settled while this thread was // waiting for approval. Its items were not in our snapshot, so start a // fresh drain instead of leaving them parked until another restart. - if (pendingDelegations.get(threadId)?.length) { - drainDelegations(bus, approvalBus, threadId, runTarget); + if (!deferred && pendingDelegations.get(threadId)?.length) { + drainDelegations(bus, approvalBus, threadId, runTarget, workOrders); } }); } @@ -196,11 +224,12 @@ function acknowledgeDelegation(threadId: string, itemId: string): void { /** Drop a thread's queued handoffs without running them, telling the user * they were dropped. Used when the queueing turn failed or was interrupted. */ -export function discardDelegations(bus: CommsBus, threadId: string): void { +export function discardDelegations(bus: CommsBus, threadId: string, workOrders?: WorkOrderStore): void { const list = pendingDelegations.get(threadId); if (!list?.length) return; pendingDelegations.delete(threadId); savePending(); + for (const item of list) transition(workOrders, item.workOrderId, "cancelled", { error: "source turn did not finish" }); const from = bus.store.botByThread(threadId); if (!from) return; bus.store.appendMessage(threadId, { @@ -215,15 +244,17 @@ async function processOne( approvalBus: ApprovalBus, from: BotRecord, sourceThreadId: string, - item: DelegationItem, + item: PendingDelegationItem, runTarget: ( toBotId: string, message: string, commsDepth: number, sourceThreadId: string, channel?: GroupRecord, + workOrderId?: string, ) => void | Promise, -): Promise { + workOrders?: WorkOrderStore, +): Promise<"done" | "defer"> { let sender = from; let target = bus.store.bot(item.toBotId); if (!target) { @@ -232,17 +263,16 @@ async function processOne( kind: "activity", tool: { name: `error: delegation to ${item.toBotId} failed — no such bot`, ok: false }, }); - return; + transition(workOrders, item.workOrderId, "failed", { error: "target bot no longer exists" }); + return "done"; } if (target.busy) { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `Delegation to @${target.name} canceled — @${target.name} is busy`, ok: false }, - }); - return; + // Busy is admission pressure, not a terminal failure. The item remains + // durable and is retried when the target's current turn settles. + return "defer"; } if (sender.approvePeerComms) { + transition(workOrders, item.workOrderId, "awaiting-approval"); const verdict = await requestPeerApproval( approvalBus, sender, @@ -257,7 +287,8 @@ async function processOne( kind: "activity", tool: { name: `Delegation to @${target.name} denied by user`, ok: false }, }); - return; + transition(workOrders, item.workOrderId, "cancelled", { error: "denied by user" }); + return "done"; } // The approval could have been sitting for up to 15 minutes. Everything // checked above is a stale snapshot now: re-read both bots and re-check @@ -265,14 +296,13 @@ async function processOne( // and mirror a "Messaged @X" chip for an exchange that never happens. const current = bus.store.bot(item.toBotId); const currentSender = bus.store.bot(from.id); - if (!current || !currentSender || !bus.store.taskByThread(currentSender.id, sourceThreadId)) return; + if (!current || !currentSender || !bus.store.taskByThread(currentSender.id, sourceThreadId)) { + transition(workOrders, item.workOrderId, "failed", { error: "source or target task no longer exists" }); + return "done"; + } if (current.busy) { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `Delegation to @${current.name} canceled — @${current.name} is busy`, ok: false }, - }); - return; + transition(workOrders, item.workOrderId, "queued"); + return "defer"; } sender = currentSender; target = current; @@ -281,7 +311,31 @@ async function processOne( mirrorExchange(bus, sender, target, item.message, channel, sourceThreadId); const reasonLine = item.reason ? `\n\n[Reason: ${item.reason}]` : ""; const prefixed = `[Delegated by @${sender.name}, another bot in this OpenMausBot workspace. Do the work and reply directly.]\n\n${item.message}${reasonLine}`; - await runTarget(item.toBotId, prefixed, item.depth + 1, sourceThreadId, channel); + transition(workOrders, item.workOrderId, "running", { attempt: (workOrders?.get(item.workOrderId ?? "")?.attempt ?? 0) + 1 }); + try { + await runTarget(item.toBotId, prefixed, item.depth + 1, sourceThreadId, channel, item.workOrderId); + transition(workOrders, item.workOrderId, "completed"); + } catch (error) { + transition(workOrders, item.workOrderId, "failed", { error: error instanceof Error ? error.message : String(error) }); + throw error; + } + return "done"; +} + +function transition( + workOrders: WorkOrderStore | undefined, + id: string | undefined, + to: "awaiting-approval" | "queued" | "running" | "completed" | "failed" | "cancelled", + patch: { result?: string; error?: string; attempt?: number } = {}, +): void { + if (!workOrders || !id) return; + try { + workOrders.transition(id, to, patch); + } catch (error) { + // A terminal order is intentionally immutable; the provider terminal + // event may race the drain's acknowledgement, so this is diagnostic only. + console.error("work-order transition failed", error); + } } /** Test helper: how many items remain queued for a thread. */ @@ -289,6 +343,11 @@ export function _pendingCount(threadId: string): number { return pendingDelegations.get(threadId)?.length ?? 0; } +/** Test helper: whether a source thread's asynchronous drain is still settling. */ +export function _isDraining(threadId: string): boolean { + return drainingThreads.has(threadId); +} + /** Test helper: forget the in-memory queue (a simulated restart). */ export function _resetPending(): void { pendingDelegations.clear(); diff --git a/server/index.ts b/server/index.ts index 3ca2ca4ac..b99178367 100644 --- a/server/index.ts +++ b/server/index.ts @@ -113,6 +113,8 @@ import { WebhookManager } from "./webhooks.ts"; import { SPAWNED_PROXIES } from "./proxy-paths.ts"; import { loadBundledSkills, loadUserSkills, mergeSkills, renderSkillInstructions, selectBundledSkills } from "./skill-library.ts"; import { shouldMountLocalComputer } from "./local-routing.ts"; +import { WorkOrderStore, type WorkOrderState } from "./work-orders.ts"; +import { TurnScheduler, type TurnLane } from "./turn-scheduler.ts"; const PORT = Number(process.env.OMB_PORT || process.env.OGB_PORT || 8799); const WEBHOOK_PORT = Number(process.env.OMB_WEBHOOK_PORT || PORT + 1); @@ -215,11 +217,18 @@ function controlIntegration(botId: string) { /** Run a turn on `targetBotId` and resolve with its assistant text — the * synchronous half of ask_bot. Subscribes to the bus, folds assistant_text * for that thread, resolves on turn.completed (or a 4-min ceiling). */ -function askBotAndWait(targetBotId: string, message: string, depth: number, fromBotId?: string): Promise { +function askBotAndWait( + targetBotId: string, + message: string, + depth: number, + fromBotId?: string, + schedulerToken?: string, + rejectStartFailure = false, +): Promise { const target = store.bot(targetBotId); if (!target) return Promise.resolve("(no such bot)"); const threadId = target.threadId; - return new Promise((resolve) => { + return new Promise((resolve, reject) => { let text = ""; let done = false; const finish = (out: string) => { @@ -229,6 +238,13 @@ function askBotAndWait(targetBotId: string, message: string, depth: number, from unsub(); resolve(out); }; + const fail = (error: unknown) => { + if (done) return; + done = true; + clearTimeout(timer); + unsub(); + reject(error); + }; const unsub = bus.subscribe((e: RuntimeEvent) => { if (e.threadId !== threadId) return; if (e.type === "item.completed" && e.itemType === "assistant_text") { @@ -241,9 +257,12 @@ function askBotAndWait(targetBotId: string, message: string, depth: number, from startTurn(targetBotId, message, { commsDepth: depth + 1, unattended: isUnattended(fromBotId), - }).catch((err) => - finish(`(couldn't start that bot: ${err instanceof Error ? err.message : String(err)})`), - ); + schedulerToken, + onDispatchError: rejectStartFailure ? (message) => fail(new Error(message)) : undefined, + }).catch((err) => { + if (rejectStartFailure) fail(err); + else finish(`(couldn't start that bot: ${err instanceof Error ? err.message : String(err)})`); + }); }); } @@ -263,6 +282,24 @@ let bootSelection = { instanceId: "", model: "" }; const store = new Store(() => bootSelection); bootSelection = await defaultSelection(); store.seedIfEmpty(); +const workOrders = new WorkOrderStore({ + onTransition: (order, from, to) => broadcast({ kind: "work-order", order, transition: { from, to } }), +}); +const turnScheduler = new TurnScheduler({ maxPendingPerBot: 32, reservedUserSlots: 4 }); +const turnAdmissionByThread = new Map(); + +function laneForTurn(opts?: { commsDepth?: number; automationSource?: RoutineRunTrigger; connectorContinuation?: boolean }): TurnLane { + if (opts?.commsDepth || opts?.connectorContinuation) return "peer"; + if (opts?.automationSource) return "background"; + return "user"; +} + +function releaseTurnAdmission(threadId: string): void { + const admission = turnAdmissionByThread.get(threadId); + if (!admission) return; + turnAdmissionByThread.delete(threadId); + turnScheduler.release(admission.botId, admission.token); +} /** A bot as a client may see it: no provider session bookkeeping. * @@ -1035,6 +1072,10 @@ bus.subscribe((event: RuntimeEvent) => { // channel that only ever shows requests is half a record. Mirror the // reply on success; mirror a failed/stopped terminal chip otherwise. finalizeDelegationWatch(event.threadId, event.ok, reply); + // Release only after the delegated watch is consumed. Releasing first + // could pump a queued peer turn and overwrite this thread's watch + // before the current terminal event was mirrored. + releaseTurnAdmission(event.threadId); // group busy/unread settle in the group turn engine, which knows // whether more member turns are queued behind this one break; @@ -1047,7 +1088,7 @@ bus.subscribe((event: RuntimeEvent) => { // (target threadId → channel) lets the main fold mirror the delegated // turn's TERMINAL state into the A⇄B channel when it completes — the // channel stays the full record of the handoff, not just its request. -const delegationWatch = new Map(); +const delegationWatch = new Map void }>(); /** Consume one delegated-turn watch and mirror exactly one terminal state. * Some harness paths settle a busy bot without a provider turn.completed @@ -1061,6 +1102,7 @@ function finalizeDelegationWatch( const watched = delegationWatch.get(threadId); if (!watched) return false; delegationWatch.delete(threadId); + watched.resolve?.(ok); const target = store.bot(watched.toBotId); const channel = watched.channelId ? store.group(watched.channelId) : undefined; if (!target || !channel) return true; @@ -1104,13 +1146,12 @@ bus.subscribe((event: RuntimeEvent) => { /** How a drained delegation becomes a real turn on the target. Shared by * the settle-time drain and the boot-time drain of what a previous process * left queued. */ -const runDelegatedTurn: Parameters[3] = (toBotId, text, commsDepth, sourceThreadId, channel) => { +const runDelegatedTurn: Parameters[3] = async (toBotId, text, commsDepth, sourceThreadId, channel, workOrderId) => { // startTurn REJECTS on an ordinary condition — busy target, deleted bot, // unavailable provider. Unhandled, that rejection is fatal to the // harness (Node's default), which in the packaged app kills the server // child. Every delegation failure has to land as a chip instead. const targetThreadId = store.bot(toBotId)?.threadId; - if (targetThreadId) delegationWatch.set(targetThreadId, { channelId: channel?.id, toBotId }); let failureReported = false; const reportStartFailure = (error: unknown) => { if (failureReported) return; @@ -1133,16 +1174,42 @@ const runDelegatedTurn: Parameters[3] = (toBotId, text, tool: { name: `error: delegation to @${bot?.name ?? toBotId} could not start — ${why.slice(0, 120)}`, ok: false }, }); }; - return startTurn(toBotId, text, { - commsDepth, - unattended: isUnattended(store.botByThread(sourceThreadId)?.id), - // startTurn schedules provider/integration setup after marking the bot - // busy. Those asynchronous setup failures do not emit turn.completed, - // so clear the watch and report them through this callback too. - onDispatchError: reportStartFailure, - }).catch((err) => { - reportStartFailure(err); + let schedulerToken = ""; + const admission = turnScheduler.admit({ + botId: toBotId, + lane: "peer", + dedupeKey: workOrderId ?? `${sourceThreadId}:${toBotId}:${text}`, + run: async () => { + let terminal!: Promise; + let resolveTerminal!: (ok: boolean) => void; + terminal = new Promise((resolve) => { resolveTerminal = resolve; }); + if (targetThreadId) { + delegationWatch.set(targetThreadId, { channelId: channel?.id, toBotId, workOrderId, resolve: resolveTerminal }); + } + await startTurn(toBotId, text, { + commsDepth, + unattended: isUnattended(store.botByThread(sourceThreadId)?.id), + schedulerToken, + // startTurn schedules provider/integration setup after marking the bot + // busy. Those asynchronous setup failures do not emit turn.completed, + // so clear the watch and report them through this callback too. + onDispatchError: reportStartFailure, + }).catch((err) => { + reportStartFailure(err); + }); + if (failureReported) throw new Error("delegated turn did not start"); + if (targetThreadId && !(await terminal)) { + throw new Error("delegated turn did not complete"); + } + }, }); + if (!admission.accepted) { + const reason = admission.reason === "capacity" ? "peer queue capacity reached" : `duplicate peer work (${admission.reason})`; + reportStartFailure(new Error(reason)); + throw new Error(reason); + } + schedulerToken = admission.admission.id; + await admission.completion; }; bus.subscribe((event: RuntimeEvent) => { @@ -1150,8 +1217,15 @@ bus.subscribe((event: RuntimeEvent) => { // A turn that failed or was interrupted drops its queue rather than // firing it later: the user who hit Stop does not expect the delegations // that turn queued to run anyway, minutes later, on an unrelated turn. - if (!event.ok) return void discardDelegations(commsBus, event.threadId); - drainDelegations(commsBus, approvalBus, event.threadId, runDelegatedTurn); + if (!event.ok) discardDelegations(commsBus, event.threadId, workOrders); + // A delegation is keyed by its SOURCE thread, but admission pressure is + // owned by the TARGET bot. Scan the bounded source list after every + // terminal event so a target becoming idle wakes its waiting handoffs. + for (const threadId of pendingThreads()) { + drainDelegations(commsBus, approvalBus, threadId, runDelegatedTurn, workOrders); + } + void drainDeferredConsultations(); + drainPeerContinuations(); }); // ── steer-queue drain: messages sent while the bot was busy ──────────── @@ -1318,12 +1392,16 @@ async function startTurn( * The prompt is control-plane context: it reaches the provider without * masquerading as another message authored by the user. */ connectorContinuation?: boolean; + /** Scheduler-owned peer admission token; direct callers must omit this. */ + schedulerToken?: string; onDispatchError?: (message: string) => void; }, ) { const bot = store.bot(botId); if (!bot) throw Object.assign(new Error("no such bot"), { status: 404 }); - if (bot.busy) throw Object.assign(new Error("the bot is already working — interrupt it first"), { status: 409 }); + if (bot.busy || (turnScheduler.hasActive(bot.id) && !opts?.schedulerToken)) { + throw Object.assign(new Error("the bot is already working — interrupt it first"), { status: 409 }); + } const threadId = opts?.threadId ?? bot.threadId; // a webhook turn, or one inherited from a bot already running unattended if (opts?.automationSource === "webhook" || opts?.unattended) markUnattended(bot.id); @@ -1413,6 +1491,8 @@ async function startTurn( // busy flips immediately so the composer locks; the dispatch itself runs // in the background — box provisioning can take ~90s and must never // hang the HTTP request + const admissionToken = opts?.schedulerToken ?? turnScheduler.occupy(bot.id, undefined, laneForTurn(opts)); + turnAdmissionByThread.set(threadId, { botId: bot.id, token: admissionToken }); store.setActivity(bot.id, "working"); store.patchBot(bot.id, { unread: false }); turnUsage.delete(threadId); @@ -1712,6 +1792,7 @@ async function startTurn( }); store.setActivity(bot.id, "idle"); opts?.onDispatchError?.(message); + releaseTurnAdmission(threadId); // a dispatch failure never emits turn.completed, so the settle-driven // drain would strand anything queued behind this turn drainQueuedSends(); @@ -1815,6 +1896,155 @@ const commsBus: CommsBus = { store, broadcast }; // can call resolvePeerComms without holding a reference back to here. const approvalBus: ApprovalBus = { store, broadcast }; +// Deferred ask_bot consultations use the same durable work-order lifecycle +// as delegate_bot, but their terminal result is delivered back to the source +// task through a hidden continuation. The source provider is never held +// open while a busy target works. +const runningConsultations = new Set(); +const pendingPeerContinuations = new Map(); + +function acceptDeferredConsultation( + from: NonNullable>, + target: NonNullable>, + message: string, + depth: number, + sourceThreadId: string, +) { + const channel = getOrCreateChannel(store, from, target); + const order = workOrders.create({ + kind: "consultation", + sourceBotId: from.id, + sourceTaskId: sourceThreadId, + sourceExecutionId: sourceThreadId, + targetBotId: target.id, + targetTaskId: target.threadId, + request: message, + priority: "peer", + depth, + delivery: "continuation", + channelId: channel.id, + }, "queued"); + mirrorExchange(commsBus, from, target, message, channel, sourceThreadId); + return { + queued: true, + workOrderId: order.id, + target: { id: target.id, name: target.name }, + message: `Consultation accepted — @${target.name} is busy, so the result will be delivered when it finishes.`, + }; +} + +function schedulePeerContinuation( + source: NonNullable>, + sourceTaskId: string, + prompts: string[], + dedupeKey: string, +): void { + let schedulerToken = ""; + const admission = turnScheduler.admit({ + botId: source.id, + lane: "peer", + dedupeKey, + run: async () => { + await startTurn(source.id, prompts.join("\n\n"), { + threadId: sourceTaskId, + commsDepth: MAX_COMMS_DEPTH, + connectorContinuation: true, + unattended: isUnattended(source.id), + schedulerToken, + }); + }, + }); + if (!admission.accepted) { + const current = pendingPeerContinuations.get(sourceTaskId); + pendingPeerContinuations.set(sourceTaskId, { + sourceBotId: source.id, + sourceTaskId, + prompts: [...(current?.prompts ?? []), ...prompts], + }); + return; + } + schedulerToken = admission.admission.id; + void admission.completion.catch((error) => { + console.error("deferred consultation continuation failed", error); + }); +} + +function queuePeerContinuation(order: { id: string; sourceBotId: string; sourceTaskId: string; targetBotId: string }, reply: string) { + const target = store.bot(order.targetBotId); + const source = store.bot(order.sourceBotId); + if (!target || !source || !store.taskByThread(source.id, order.sourceTaskId)) return; + const prompt = `[Deferred consultation result from @${target.name}. The earlier ask_bot receipt meant acceptance, not completion. Do not call peer tools for this continuation.]\n\n@${target.name} replied:\n${reply}`; + if (source.busy || turnScheduler.hasActive(source.id)) { + const current = pendingPeerContinuations.get(order.sourceTaskId); + pendingPeerContinuations.set(order.sourceTaskId, { + sourceBotId: source.id, + sourceTaskId: order.sourceTaskId, + prompts: [...(current?.prompts ?? []), prompt], + }); + return; + } + schedulePeerContinuation(source, order.sourceTaskId, [prompt], `continuation:${order.id}`); +} + +async function drainDeferredConsultations(): Promise { + for (const order of workOrders.list({ states: ["queued"], limit: 200 })) { + if (order.kind !== "consultation" || runningConsultations.has(order.id)) continue; + const source = store.bot(order.sourceBotId); + const target = store.bot(order.targetBotId); + if (!source || !target || !store.taskByThread(source.id, order.sourceTaskId) || !store.taskByThread(target.id, order.targetTaskId)) { + workOrders.transition(order.id, "failed", { error: "pinned source or target task no longer exists" }); + continue; + } + if (target.busy) continue; + runningConsultations.add(order.id); + const channel = order.channelId ? store.group(order.channelId) ?? undefined : undefined; + void (async () => { + try { + const prefixed = `[Deferred consultation from @${source.name}. Reply directly to the requesting bot.]\n\n${order.request}`; + let schedulerToken = ""; + const admission = turnScheduler.admit({ + botId: target.id, + lane: "peer", + dedupeKey: order.id, + run: async () => { + workOrders.transition(order.id, "running", { attempt: (order.attempt ?? 0) + 1 }); + const reply = await askBotAndWait(target.id, prefixed, order.depth, source.id, schedulerToken, true); + workOrders.transition(order.id, "completed", { result: reply }); + mirrorReply(commsBus, target, reply, channel); + queuePeerContinuation(order, reply); + }, + }); + if (!admission.accepted) { + runningConsultations.delete(order.id); + if (admission.reason !== "duplicate") workOrders.transition(order.id, "queued"); + return; + } + schedulerToken = admission.admission.id; + await admission.completion; + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + try { + workOrders.transition(order.id, "failed", { error: reason }); + } catch (transitionError) { + console.error("deferred consultation transition failed", transitionError); + } + if (channel) mirrorActivity(commsBus, target, channel, `Consultation failed — ${reason.slice(0, 120)}`, false); + } finally { + runningConsultations.delete(order.id); + } + })(); + } +} + +function drainPeerContinuations(): void { + for (const [sourceTaskId, pending] of pendingPeerContinuations) { + const bot = store.bot(pending.sourceBotId); + if (!bot || bot.busy || turnScheduler.hasActive(bot.id)) continue; + pendingPeerContinuations.delete(sourceTaskId); + schedulePeerContinuation(bot, pending.sourceTaskId, pending.prompts, `continuation:${sourceTaskId}`); + } +} + // Approvals live only in memory, so any peer card still open on disk is one // whose resolver died with the previous process. Left alone it can never be // answered, and the composer stays disabled behind it — settle them at boot. @@ -1823,6 +2053,17 @@ const approvalBus: ApprovalBus = { store, broadcast }; if (stale) console.log(`peer approvals: dismissed ${stale} card(s) left by a previous run`); } +// Work orders are a separate durable control-plane record. A source turn +// cannot survive a restart, and a provider turn that was running must never +// be replayed automatically; queued/approval-waiting orders remain for the +// scheduler to reconcile against the current bot/task roster. +{ + const recovered = workOrders.recover(); + if (recovered.cancelled.length || recovered.failed.length) { + console.log(`work orders: recovered ${recovered.cancelled.length} cancelled and ${recovered.failed.length} failed record(s)`); + } +} + // Handoffs a previous process queued but never ran: the source turn is // dead (no turn survives a restart) so they would otherwise wait forever. // Run them now, through the same drain — target and approvePeerComms are @@ -1831,7 +2072,7 @@ _loadPending(); { const leftover = pendingThreads(); if (leftover.length) console.log(`delegations: ${leftover.length} thread(s) with queued handoffs from a previous run — draining`); - for (const threadId of leftover) drainDelegations(commsBus, approvalBus, threadId, runDelegatedTurn); + for (const threadId of leftover) drainDelegations(commsBus, approvalBus, threadId, runDelegatedTurn, workOrders); } async function runGroupMemberTurn( @@ -1862,7 +2103,7 @@ async function runGroupMemberTurn( // could run its 1:1 turn and a room turn concurrently — two provider // processes, interleaved token spend, and an interrupt that only ever // reached one of them. - if (bot.busy) { + if (bot.busy || turnScheduler.hasActive(bot.id)) { store.appendMessage(group.threadId, { role: "bot", kind: "activity", @@ -1895,6 +2136,8 @@ async function runGroupMemberTurn( return true; } store.setActivity(bot.id, "working"); + const roomAdmissionToken = turnScheduler.occupy(bot.id, undefined, "user"); + turnAdmissionByThread.set(group.threadId, { botId: bot.id, token: roomAdmissionToken }); store.patchGroup(group.id, { busyBotId: bot.id }); // the store's change stream carries the frame groupSpeakers.set(group.threadId, { botId: bot.id, name: bot.name, color: bot.color }); @@ -1991,6 +2234,7 @@ async function runGroupMemberTurn( // A timed-out provider still owns the room thread until its interrupt // produces turn.completed (or the stall watchdog's grace fallback runs). // Do not clear busy or start the next member on that same thread early. + if (outcome === "dispatch_failed") releaseTurnAdmission(group.threadId); if (outcome === "stalled" || outcome === "timed_out") return false; // turn.completed normally performs this cleanup. Only use the fallback // when this invocation still owns the room; otherwise it would emit a @@ -2505,7 +2749,6 @@ const server = createServer(async (req, res) => { if (depth >= MAX_COMMS_DEPTH) return json(res, 200, { error: "message chains are limited to one hop" }); const target = store.bot(toBotId); if (!target) return json(res, 404, { error: "no such bot" }); - if (target.busy) return json(res, 200, { busy: true }); // An unknown sender used to fall through: no mirroring AND no // approval, while still running the peer turn. That made an // unresolvable id the cheapest way past the gate, so it is now a @@ -2554,10 +2797,13 @@ const server = createServer(async (req, res) => { if (!store.taskByThread(freshFrom.id, fromThreadId)) { return json(res, 404, { error: "source task no longer exists" }); } - if (freshTarget.busy) return json(res, 200, { busy: true }); currentFrom = freshFrom; currentTarget = freshTarget; } + if (currentTarget.busy) { + const accepted = acceptDeferredConsultation(currentFrom, currentTarget, message, depth, fromThreadId); + return json(res, 202, accepted); + } const channel = getOrCreateChannel(store, currentFrom, currentTarget); mirrorExchange(commsBus, currentFrom, currentTarget, message, channel, fromThreadId); const prefixed = `[Message from @${currentFrom.name}, another bot in this OpenMausBot workspace. Reply to them.]\n\n${message}`; @@ -2593,6 +2839,7 @@ const server = createServer(async (req, res) => { { toBotId, message, reason, depth }, MAX_COMMS_DEPTH, fromThreadId, + workOrders, ); if (result !== "ok") { // the agent reads this string — a bare enum ("too_deep") tells it @@ -2761,6 +3008,33 @@ const server = createServer(async (req, res) => { return json(res, 404, { error: "unknown internal endpoint" }); } + // ── durable peer work orders ───────────────────────────────────────── + if (path === "/api/work-orders" && method === "GET") { + const sourceBotId = url.searchParams.get("sourceBotId") ?? undefined; + const targetBotId = url.searchParams.get("targetBotId") ?? undefined; + const rawStates = url.searchParams.get("states"); + const states = rawStates + ? rawStates.split(",").filter((state): state is WorkOrderState => + ["pending-source", "awaiting-approval", "queued", "running", "completed", "failed", "cancelled"].includes(state), + ) + : undefined; + return json(res, 200, { workOrders: workOrders.list({ sourceBotId, targetBotId, states }) }); + } + let workOrderMatch = path.match(/^\/api\/work-orders\/([\w-]+)$/); + if (workOrderMatch && method === "GET") { + const order = workOrders.get(workOrderMatch[1]); + return order ? json(res, 200, { workOrder: order }) : json(res, 404, { error: "no such work order" }); + } + workOrderMatch = path.match(/^\/api\/work-orders\/([\w-]+)\/cancel$/); + if (workOrderMatch && method === "POST") { + const order = workOrders.cancel(workOrderMatch[1], String((await readBody(req)).reason ?? "cancelled by user")); + if (!order) return json(res, 404, { error: "no such active work order" }); + // A provider interrupt is deliberately not implicit here: cancellation + // of queued peer work is safe; an active provider turn owns its own + // process and follows the normal Stop/approval path. + return json(res, 200, { workOrder: order }); + } + // ── routines calendar ──────────────────────────────────────────────── if (path === "/api/routines" && method === "GET") { const fromParam = url.searchParams.get("from"); diff --git a/server/turn-scheduler.test.ts b/server/turn-scheduler.test.ts new file mode 100644 index 000000000..dcf5c551a --- /dev/null +++ b/server/turn-scheduler.test.ts @@ -0,0 +1,85 @@ +import { describe, expect, it } from "vitest"; + +import { TurnScheduler } from "./turn-scheduler.ts"; + +function deferred() { + let resolve!: () => void; + const promise = new Promise((res) => { resolve = res; }); + return { promise, resolve }; +} + +describe("TurnScheduler", () => { + it("runs one turn per bot, keeps FIFO inside lanes, and prioritizes humans", async () => { + const scheduler = new TurnScheduler({ maxPendingPerBot: 8, reservedUserSlots: 0 }); + const first = deferred(); + const order: string[] = []; + const active = scheduler.admit({ botId: "a", lane: "peer", run: async () => { order.push("active"); await first.promise; } }); + expect(active.accepted && active.admission.queued).toBe(false); + const normal = scheduler.admit({ botId: "a", lane: "peer", run: async () => { order.push("peer"); } }); + const background = scheduler.admit({ botId: "a", lane: "background", run: async () => { order.push("background"); } }); + const urgent = scheduler.admit({ botId: "a", lane: "urgent-peer", run: async () => { order.push("urgent"); } }); + const user = scheduler.admit({ botId: "a", lane: "user", run: async () => { order.push("user"); } }); + if (!active.accepted || !normal.accepted) throw new Error("expected scheduler admissions"); + first.resolve(); + await Promise.all([ + active.completion, + normal.completion, + background.accepted && background.completion, + urgent.accepted && urgent.completion, + user.accepted && user.completion, + ]); + expect(order).toEqual(["active", "user", "urgent", "peer", "background"]); + }); + + it("allows different bots to run concurrently", async () => { + const scheduler = new TurnScheduler(); + const a = deferred(); + const b = deferred(); + const seen: string[] = []; + let resolveStarted!: () => void; + const started = new Promise((resolve) => { resolveStarted = resolve; }); + const aAdmission = scheduler.admit({ botId: "a", lane: "user", run: async () => { seen.push("a-start"); if (seen.length === 2) resolveStarted(); await a.promise; seen.push("a-end"); } }); + const bAdmission = scheduler.admit({ botId: "b", lane: "user", run: async () => { seen.push("b-start"); if (seen.length === 2) resolveStarted(); await b.promise; seen.push("b-end"); } }); + await started; + expect(seen).toEqual(["a-start", "b-start"]); + a.resolve(); + b.resolve(); + if (aAdmission.accepted && bAdmission.accepted) await Promise.all([aAdmission.completion, bAdmission.completion]); + expect(seen).toEqual(["a-start", "b-start", "a-end", "b-end"]); + }); + + it("reserves capacity for user work and supports cancellation", async () => { + const scheduler = new TurnScheduler({ maxPendingPerBot: 3, reservedUserSlots: 1 }); + const hold = deferred(); + scheduler.admit({ botId: "a", lane: "user", run: () => hold.promise }); + const peer = scheduler.admit({ botId: "a", lane: "peer", run: () => {} }); + expect(peer.accepted).toBe(true); + const remaining = scheduler.admit({ botId: "a", lane: "peer", run: () => {} }); + const rejected = scheduler.admit({ botId: "a", lane: "peer", run: () => {} }); + expect(rejected).toEqual({ accepted: false, reason: "capacity" }); + if (!peer.accepted) throw new Error("expected a queued peer"); + void peer.completion.catch(() => {}); + expect(scheduler.cancel("a", peer.admission.id)).toBe(true); + expect(scheduler.diagnostics("a").pending).toBe(1); + hold.resolve(); + if (remaining.accepted) await remaining.completion; + }); + + it("deduplicates queued work and never lets a late release free a newer run", async () => { + const scheduler = new TurnScheduler(); + const first = deferred(); + const second = deferred(); + const firstAdmission = scheduler.admit({ botId: "a", lane: "user", dedupeKey: "same", run: () => first.promise }); + const duplicate = scheduler.admit({ botId: "a", lane: "peer", dedupeKey: "same", run: () => {} }); + expect(duplicate).toEqual({ accepted: false, reason: "duplicate" }); + first.resolve(); + if (firstAdmission.accepted) await firstAdmission.completion; + const queued = scheduler.admit({ botId: "a", lane: "user", run: () => second.promise }); + expect(queued.accepted).toBe(true); + // A stale generation cannot release the new active turn. + expect(scheduler.release("a", "not-the-current-run")).toBe(false); + expect(scheduler.hasActive("a")).toBe(true); + second.resolve(); + if (queued.accepted) await queued.completion; + }); +}); diff --git a/server/turn-scheduler.ts b/server/turn-scheduler.ts new file mode 100644 index 000000000..9f2241be4 --- /dev/null +++ b/server/turn-scheduler.ts @@ -0,0 +1,263 @@ +import { newId } from "./contracts.ts"; + +/** Admission lanes shared by human messages and unattended work. */ +export type TurnLane = "user" | "urgent-peer" | "peer" | "background"; + +const LANE_PRIORITY: readonly TurnLane[] = ["user", "urgent-peer", "peer", "background"]; + +export type SchedulerRejection = "capacity" | "duplicate" | "cancelled"; + +export interface TurnAdmission { + id: string; + botId: string; + lane: TurnLane; + /** Number of queued entries ahead of this entry. Active work is not counted. */ + position: number; + queued: boolean; +} + +export interface TurnSchedulerOptions { + /** Maximum active + pending work admitted for one bot. */ + maxPendingPerBot?: number; + /** Slots kept free for human work when the queue is under pressure. */ + reservedUserSlots?: number; + onChange?: (botId: string) => void; +} + +interface Entry { + id: string; + botId: string; + lane: TurnLane; + dedupeKey?: string; + run: () => void | Promise; + resolve: () => void; + reject: (error: unknown) => void; + promise: Promise; + cancelled: boolean; +} + +interface Active { + id: string; + generation: number; + lane: TurnLane; +} + +export interface TurnQueueDiagnostics { + botId: string; + active: Active | null; + pending: number; + byLane: Record; + capacity: number; +} + +/** + * Per-bot, priority/FIFO admission for provider turns. + * + * The scheduler deliberately owns no provider state. A run is considered + * active until its callback's promise settles, which makes late callbacks + * harmless: only the matching id/generation may release the lane. Different + * bots have independent queues and therefore run concurrently. + */ +export class TurnScheduler { + private readonly maxPendingPerBot: number; + private readonly reservedUserSlots: number; + private readonly onChange?: (botId: string) => void; + private readonly queues = new Map>(); + private readonly active = new Map(); + private readonly generations = new Map(); + + constructor(options: TurnSchedulerOptions = {}) { + this.maxPendingPerBot = Math.max(1, Math.trunc(options.maxPendingPerBot ?? 32)); + this.reservedUserSlots = Math.max(0, Math.min(this.maxPendingPerBot - 1, Math.trunc(options.reservedUserSlots ?? 4))); + this.onChange = options.onChange; + } + + admit(input: { + botId: string; + lane: TurnLane; + dedupeKey?: string; + run: () => void | Promise; + }): { accepted: true; admission: TurnAdmission; completion: Promise } | { accepted: false; reason: SchedulerRejection } { + const queue = this.queueFor(input.botId); + const existing = this.findEntry(input.botId, input.dedupeKey); + if (existing) return { accepted: false, reason: "duplicate" }; + + const pending = this.pendingCount(input.botId); + if (pending >= this.maxPendingPerBot || (input.lane !== "user" && pending >= this.maxPendingPerBot - this.reservedUserSlots)) { + return { accepted: false, reason: "capacity" }; + } + + let resolve!: () => void; + let reject!: (error: unknown) => void; + const completion = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + const entry: Entry = { + id: newId(), + botId: input.botId, + lane: input.lane, + dedupeKey: input.dedupeKey, + run: input.run, + resolve, + reject, + promise: completion, + cancelled: false, + }; + + const current = this.active.get(input.botId); + if (!current) { + this.start(entry); + return { accepted: true, admission: { id: entry.id, botId: entry.botId, lane: entry.lane, position: 0, queued: false }, completion }; + } + + queue.get(input.lane)!.push(entry); + const position = this.positionOf(input.botId, entry.id); + this.changed(input.botId); + return { accepted: true, admission: { id: entry.id, botId: entry.botId, lane: entry.lane, position, queued: true }, completion }; + } + + /** Cancel a pending entry. Active provider turns are never preempted. */ + cancel(botId: string, id: string): boolean { + const lanes = this.queues.get(botId); + if (!lanes) return false; + for (const lane of LANE_PRIORITY) { + const items = lanes.get(lane)!; + const index = items.findIndex((entry) => entry.id === id); + if (index < 0) continue; + const [entry] = items.splice(index, 1); + entry.cancelled = true; + entry.reject(Object.assign(new Error("queued turn cancelled"), { code: "TURN_CANCELLED" })); + this.changed(botId); + return true; + } + return false; + } + + /** Current state used by bounded queue diagnostics and tests. */ + diagnostics(botId: string): TurnQueueDiagnostics { + const byLane = Object.fromEntries(LANE_PRIORITY.map((lane) => [lane, this.queues.get(botId)?.get(lane)?.length ?? 0])) as Record; + return { + botId, + active: this.active.get(botId) ?? null, + pending: Object.values(byLane).reduce((sum, count) => sum + count, 0), + byLane, + capacity: this.maxPendingPerBot, + }; + } + + /** Used when a legacy path owns a turn that was not admitted here. */ + occupy(botId: string, id = newId(), lane: TurnLane = "user"): string { + if (this.active.has(botId)) return this.active.get(botId)!.id; + const generation = (this.generations.get(botId) ?? 0) + 1; + this.generations.set(botId, generation); + this.active.set(botId, { id, generation, lane }); + this.changed(botId); + return id; + } + + /** Release only the current generation; a late settlement cannot free a newer run. */ + release(botId: string, id?: string): boolean { + const current = this.active.get(botId); + if (!current || (id && current.id !== id)) return false; + this.active.delete(botId); + this.pump(botId); + return true; + } + + hasActive(botId: string): boolean { + return this.active.has(botId); + } + + private queueFor(botId: string): Map { + let queue = this.queues.get(botId); + if (queue) return queue; + queue = new Map(LANE_PRIORITY.map((lane) => [lane, []] as const)); + this.queues.set(botId, queue); + return queue; + } + + private pendingCount(botId: string): number { + return LANE_PRIORITY.reduce((sum, lane) => sum + (this.queues.get(botId)?.get(lane)?.length ?? 0), 0); + } + + private findEntry(botId: string, dedupeKey?: string): Entry | null { + if (!dedupeKey) return null; + const current = this.active.get(botId); + // Active deduplication prevents a webhook/continuation from creating a + // second copy while the first one is already inside the provider. + if (current) { + const activeEntry = this.findActiveEntry(botId, current.id); + if (activeEntry?.dedupeKey === dedupeKey) return activeEntry; + } + for (const lane of LANE_PRIORITY) { + const found = this.queues.get(botId)?.get(lane)?.find((entry) => entry.dedupeKey === dedupeKey); + if (found) return found; + } + return null; + } + + // Active entries are only needed for dedupe while their promise is running. + // Keeping this tiny map avoids exposing mutable internals in diagnostics. + private readonly activeEntries = new Map(); + + private findActiveEntry(botId: string, id: string): Entry | null { + const entry = this.activeEntries.get(botId); + return entry?.id === id ? entry : null; + } + + private start(entry: Entry): void { + const generation = (this.generations.get(entry.botId) ?? 0) + 1; + this.generations.set(entry.botId, generation); + this.active.set(entry.botId, { id: entry.id, generation, lane: entry.lane }); + this.activeEntries.set(entry.botId, entry); + this.changed(entry.botId); + Promise.resolve() + .then(() => entry.run()) + .then(() => entry.resolve(), (error) => entry.reject(error)) + .finally(() => { + if (this.active.get(entry.botId)?.id !== entry.id) return; + this.activeEntries.delete(entry.botId); + this.active.delete(entry.botId); + this.pump(entry.botId); + }); + } + + private pump(botId: string): void { + if (this.active.has(botId)) return; + const queue = this.queues.get(botId); + if (!queue) { + this.changed(botId); + return; + } + let next: Entry | undefined; + for (const lane of LANE_PRIORITY) { + const items = queue.get(lane)!; + while (items.length && items[0]!.cancelled) items.shift(); + next = items.shift(); + if (next) break; + } + if (next) this.start(next); + else { + this.queues.delete(botId); + this.changed(botId); + } + } + + private positionOf(botId: string, id: string): number { + let position = 0; + for (const lane of LANE_PRIORITY) { + for (const entry of this.queues.get(botId)?.get(lane) ?? []) { + if (entry.id === id) return position; + position += 1; + } + } + return position; + } + + private changed(botId: string): void { + this.onChange?.(botId); + } +} + +export { LANE_PRIORITY }; diff --git a/server/work-orders.test.ts b/server/work-orders.test.ts new file mode 100644 index 000000000..7bf71fc30 --- /dev/null +++ b/server/work-orders.test.ts @@ -0,0 +1,73 @@ +import { mkdtempSync, readFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { WorkOrderStore, type WorkOrderInput } from "./work-orders.ts"; + +const input: WorkOrderInput = { + kind: "consultation", + sourceBotId: "source", + sourceTaskId: "source-task", + sourceExecutionId: "turn-1", + targetBotId: "target", + targetTaskId: "target-task", + request: "Please inspect this safely.", + priority: "peer", + depth: 0, + delivery: "continuation", +}; + +function file() { + return join(mkdtempSync(join(tmpdir(), "omb-work-orders-")), "orders.json"); +} + +describe("WorkOrderStore", () => { + it("persists the lifecycle and redacts terminal output without changing the request", () => { + let now = 100; + const path = file(); + const store = new WorkOrderStore({ file: path, now: () => now }); + const order = store.create(input); + now += 1; + store.transition(order.id, "queued"); + now += 1; + store.transition(order.id, "running"); + now += 1; + const completed = store.transition(order.id, "completed", { + result: "token=sk-test-12345678901234567890; done", + }); + expect(completed.state).toBe("completed"); + expect(completed.request).toBe(input.request); + expect(completed.result).not.toContain("sk-test-"); + expect(JSON.parse(readFileSync(path, "utf8")).orders[0].result).not.toContain("sk-test-"); + expect(() => store.transition(order.id, "failed")).toThrow(/terminal/); + }); + + it("reconstructs queued work, cancels source-bound work, and fails running work after restart", () => { + const path = file(); + const first = new WorkOrderStore({ file: path, now: () => 100 }); + const source = first.create(input); + const queued = first.create({ ...input, targetBotId: "target-2", targetTaskId: "target-task-2" }, "queued"); + const running = first.create({ ...input, targetBotId: "target-3", targetTaskId: "target-task-3" }, "queued"); + first.transition(running.id, "running"); + const second = new WorkOrderStore({ file: path, now: () => 200 }); + const recovered = second.recover(); + expect(recovered.cancelled.map((order) => order.id)).toEqual([source.id]); + expect(recovered.failed.map((order) => order.id)).toEqual([running.id]); + expect(second.get(queued.id)?.state).toBe("queued"); + expect(second.get(source.id)?.error).toContain("source execution"); + }); + + it("keeps terminal history bounded while preserving active orders", () => { + let now = 1; + const store = new WorkOrderStore({ file: file(), maxTerminal: 2, now: () => now++ }); + for (let i = 0; i < 4; i += 1) { + const order = store.create({ ...input, request: `request ${i}` }); + store.transition(order.id, "cancelled", { error: "no-op" }); + } + const active = store.create({ ...input, request: "still queued" }, "queued"); + expect(store.list({ limit: 20 }).filter((order) => order.state === "cancelled")).toHaveLength(2); + expect(store.get(active.id)?.state).toBe("queued"); + }); +}); diff --git a/server/work-orders.ts b/server/work-orders.ts new file mode 100644 index 000000000..b1b3fe083 --- /dev/null +++ b/server/work-orders.ts @@ -0,0 +1,204 @@ +import { readFileSync } from "node:fs"; +import { join } from "node:path"; + +import { writeFileAtomic } from "./atomic.ts"; +import { DATA_DIR } from "./config.ts"; +import { newId } from "./contracts.ts"; +import { redactSecretsInText } from "./redact.ts"; +import type { TurnLane } from "./turn-scheduler.ts"; + +export type WorkOrderKind = "consultation" | "delegation"; +export type WorkOrderState = "pending-source" | "awaiting-approval" | "queued" | "running" | "completed" | "failed" | "cancelled"; +export type WorkOrderDelivery = "inline" | "channel" | "continuation"; + +export interface WorkOrderInput { + kind: WorkOrderKind; + sourceBotId: string; + sourceTaskId: string; + sourceExecutionId?: string; + targetBotId: string; + targetTaskId: string; + request: string; + reason?: string; + priority?: Exclude; + depth: number; + delivery: WorkOrderDelivery; + channelId?: string; + attempt?: number; +} + +export interface WorkOrder extends WorkOrderInput { + id: string; + state: WorkOrderState; + createdAt: number; + updatedAt: number; + result?: string; + error?: string; +} + +interface DiskFile { + version: 1; + orders: WorkOrder[]; +} + +const TERMINAL: ReadonlySet = new Set(["completed", "failed", "cancelled"]); +const TRANSITIONS: Record = { + "pending-source": ["awaiting-approval", "queued", "cancelled", "failed"], + "awaiting-approval": ["queued", "cancelled", "failed"], + queued: ["running", "cancelled", "failed"], + running: ["completed", "failed", "cancelled"], + completed: [], + failed: [], + cancelled: [], +}; + +/** + * Crash-safe peer work orders. This is intentionally separate from the + * provider and transcript stores: an accepted handoff must survive a bot + * restart without becoming a fake assistant message or a hidden turn. + */ +export class WorkOrderStore { + private readonly file: string; + private readonly now: () => number; + private readonly maxTerminal: number; + private readonly onTransition?: (order: WorkOrder, from: WorkOrderState, to: WorkOrderState) => void; + private orders: WorkOrder[] = []; + + constructor(options: { + file?: string; + now?: () => number; + maxTerminal?: number; + onTransition?: (order: WorkOrder, from: WorkOrderState, to: WorkOrderState) => void; + } = {}) { + this.file = options.file ?? join(DATA_DIR, "work-orders.json"); + this.now = options.now ?? Date.now; + this.maxTerminal = Math.max(1, Math.trunc(options.maxTerminal ?? 200)); + this.onTransition = options.onTransition; + this.load(); + } + + create(input: WorkOrderInput, state: WorkOrderState = "pending-source"): WorkOrder { + if (state !== "pending-source" && state !== "awaiting-approval" && state !== "queued") { + throw new Error("new work orders must begin pending-source, awaiting-approval, or queued"); + } + const at = this.now(); + const order: WorkOrder = { + ...input, + id: newId(), + state, + request: input.request, + ...(input.reason ? { reason: input.reason } : {}), + ...(input.channelId ? { channelId: input.channelId } : {}), + attempt: Math.max(0, Math.trunc(input.attempt ?? 0)), + createdAt: at, + updatedAt: at, + }; + this.orders.push(order); + this.save(); + this.onTransition?.({ ...order }, state, state); + return { ...order }; + } + + get(id: string): WorkOrder | null { + const order = this.orders.find((candidate) => candidate.id === id); + return order ? { ...order } : null; + } + + list(options: { sourceBotId?: string; targetBotId?: string; states?: WorkOrderState[]; limit?: number } = {}): WorkOrder[] { + const states = options.states ? new Set(options.states) : null; + const limit = Math.max(1, Math.min(500, Math.trunc(options.limit ?? 100))); + return this.orders + .filter((order) => + (!options.sourceBotId || order.sourceBotId === options.sourceBotId) && + (!options.targetBotId || order.targetBotId === options.targetBotId) && + (!states || states.has(order.state)), + ) + .sort((a, b) => b.updatedAt - a.updatedAt) + .slice(0, limit) + .map((order) => ({ ...order })); + } + + transition(id: string, to: WorkOrderState, patch: Partial> = {}): WorkOrder { + const order = this.orders.find((candidate) => candidate.id === id); + if (!order) throw new Error("no such work order"); + const from = order.state; + if (from === to) return { ...order }; + if (TERMINAL.has(from)) throw new Error("terminal work orders are immutable"); + if (!TRANSITIONS[from].includes(to)) throw new Error(`invalid work-order transition ${from} → ${to}`); + order.state = to; + order.updatedAt = this.now(); + if (patch.attempt !== undefined) order.attempt = Math.max(0, Math.trunc(patch.attempt)); + if (patch.channelId !== undefined) order.channelId = patch.channelId; + // Requests are intentionally never passed through redaction. Results and + // failures may contain tool output, credentials, or provider diagnostics. + if (patch.result !== undefined) order.result = redactSecretsInText(patch.result).slice(0, 20_000); + if (patch.error !== undefined) order.error = redactSecretsInText(patch.error).slice(0, 2_000); + this.save(); + this.onTransition?.({ ...order }, from, to); + return { ...order }; + } + + cancel(id: string, reason = "cancelled by user"): WorkOrder | null { + const order = this.orders.find((candidate) => candidate.id === id); + if (!order || TERMINAL.has(order.state)) return null; + return this.transition(id, "cancelled", { error: reason }); + } + + /** Pending/approval work may be reconstructed; source and running work cannot. */ + recover(): { cancelled: WorkOrder[]; failed: WorkOrder[] } { + const cancelled: WorkOrder[] = []; + const failed: WorkOrder[] = []; + for (const order of this.orders) { + if (order.state === "pending-source") { + cancelled.push(this.transition(order.id, "cancelled", { error: "source execution did not survive restart" })); + } else if (order.state === "running") { + failed.push(this.transition(order.id, "failed", { error: "OpenMausBot restarted while this work was running" })); + } + } + return { cancelled, failed }; + } + + private load(): void { + try { + const disk = JSON.parse(readFileSync(this.file, "utf8")) as Partial; + if (!Array.isArray(disk.orders)) return; + this.orders = disk.orders.flatMap((value) => this.validOrder(value)); + } catch { + this.orders = []; + } + // A fresh process cannot answer old approval promises, but the queued + // and approval-waiting records remain visible for the boot reconciler. + this.pruneTerminal(); + } + + private validOrder(value: unknown): WorkOrder[] { + if (!value || typeof value !== "object") return []; + const item = value as Partial; + if ( + typeof item.id !== "string" || typeof item.kind !== "string" || + typeof item.sourceBotId !== "string" || typeof item.sourceTaskId !== "string" || + typeof item.targetBotId !== "string" || typeof item.targetTaskId !== "string" || + typeof item.request !== "string" || typeof item.depth !== "number" || + typeof item.createdAt !== "number" || typeof item.updatedAt !== "number" || + !item.state || !TRANSITIONS[item.state as WorkOrderState] + ) return []; + return [{ + ...(item as WorkOrder), + request: item.request, + ...(typeof item.result === "string" ? { result: redactSecretsInText(item.result) } : {}), + ...(typeof item.error === "string" ? { error: redactSecretsInText(item.error) } : {}), + }]; + } + + private pruneTerminal(): void { + const terminal = this.orders.filter((order) => TERMINAL.has(order.state)); + if (terminal.length <= this.maxTerminal) return; + const keep = new Set(terminal.sort((a, b) => b.updatedAt - a.updatedAt).slice(0, this.maxTerminal).map((order) => order.id)); + this.orders = this.orders.filter((order) => !TERMINAL.has(order.state) || keep.has(order.id)); + } + + private save(): void { + this.pruneTerminal(); + writeFileAtomic(this.file, JSON.stringify({ version: 1, orders: this.orders }, null, 2), { mode: 0o600 }); + } +} From 77ffe4690c9549e11466f3603a78803ce6e14c5c Mon Sep 17 00:00:00 2001 From: Omar Haneya Date: Mon, 24 Aug 2026 18:14:41 +0100 Subject: [PATCH 3/4] fix: harden peer work lifecycle and scheduling --- server/delegations.test.ts | 22 +++- server/delegations.ts | 139 ++++++++++++++++----- server/index.ts | 221 +++++++++++++++++++++++++++------- server/peer-approval.ts | 15 +++ server/turn-scheduler.test.ts | 43 +++++-- server/turn-scheduler.ts | 4 +- server/work-orders.test.ts | 27 +++++ server/work-orders.ts | 48 ++++++++ 8 files changed, 430 insertions(+), 89 deletions(-) diff --git a/server/delegations.test.ts b/server/delegations.test.ts index f5844536e..feea85f7d 100644 --- a/server/delegations.test.ts +++ b/server/delegations.test.ts @@ -18,6 +18,7 @@ import { } from "./delegations.ts"; import { peerAllowKey, resolvePeerComms } from "./peer-approval.ts"; import { Store, type BotRecord } from "./store.ts"; +import { WorkOrderStore } from "./work-orders.ts"; const selection = (): ModelSelection => ({ instanceId: "claude", model: "fake-model" }); @@ -310,27 +311,35 @@ describe("drainDelegations", () => { it("keeps the handoff queued while the target is busy, then runs it when idle", async () => { store.patchBot(target.id, { busy: true }); - queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + const workOrders = new WorkOrderStore({ file: join(DATA_DIR, "busy-work-orders.json") }); + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1, from.threadId, workOrders); + const order = workOrders.list({ limit: 1 })[0]; + expect(order?.state).toBe("pending-source"); drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { runTargetCalls.push({ toBotId, message, commsDepth }); - }); + }, workOrders); await waitFor(() => !_isDraining(from.threadId)); expect(runTargetCalls).toEqual([]); expect(_pendingCount(from.threadId)).toBe(1); + expect(workOrders.get(order!.id)?.state).toBe("pending-source"); store.patchBot(target.id, { busy: false }); drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { runTargetCalls.push({ toBotId, message, commsDepth }); - }); + }, workOrders); await waitFor(() => runTargetCalls.length === 1 && _pendingCount(from.threadId) === 0); expect(_pendingCount(from.threadId)).toBe(0); + expect(workOrders.get(order!.id)?.state).toBe("completed"); }); it("asks for approval when approvePeerComms is on, then runs only on allow", async () => { store.patchBot(from.id, { approvePeerComms: true }); - queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + const workOrders = new WorkOrderStore({ file: join(DATA_DIR, "approval-work-orders.json") }); + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1, from.threadId, workOrders); + const order = workOrders.list({ limit: 1 })[0]!; + expect(order.state).toBe("pending-source"); drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { runTargetCalls.push({ toBotId, message, commsDepth }); - }); + }, workOrders); // the source bot's thread shows the options card BEFORE runTarget fires const card = await waitFor(() => @@ -341,9 +350,10 @@ describe("drainDelegations", () => { expect(card.card?.allowKey).toBe(peerAllowKey("delegate_bot", target.id)); expect(card.card?.options).toEqual(["Allow", "Deny", "Always allow"]); expect(runTargetCalls).toEqual([]); + expect(workOrders.get(order.id)?.state).toBe("awaiting-approval"); resolvePeerComms(approvalBus, card.card!.requestId!, "allow"); - await waitFor(() => runTargetCalls.length === 1); + await waitFor(() => workOrders.get(order.id)?.state === "completed"); expect(runTargetCalls[0]!.toBotId).toBe(target.id); expect(runTargetCalls[0]!.commsDepth).toBe(1); }); diff --git a/server/delegations.ts b/server/delegations.ts index d6bb56329..9cf69574c 100644 --- a/server/delegations.ts +++ b/server/delegations.ts @@ -18,9 +18,9 @@ import { writeFileAtomic } from "./atomic.ts"; import { getOrCreateChannel, mirrorExchange, type CommsBus } from "./comms-visibility.ts"; import { DATA_DIR } from "./config.ts"; import { newId } from "./contracts.ts"; -import { requestPeerApproval, type ApprovalBus } from "./peer-approval.ts"; +import { cancelPeerApprovalForOwner, requestPeerApproval, type ApprovalBus } from "./peer-approval.ts"; import type { BotRecord, GroupRecord } from "./store.ts"; -import { WorkOrderStore } from "./work-orders.ts"; +import { WorkOrderCapacityError, WorkOrderInputError, WorkOrderStore } from "./work-orders.ts"; export interface DelegationItem { toBotId: string; @@ -38,9 +38,11 @@ interface PendingDelegationItem extends DelegationItem { id: string; /** Durable lifecycle record, when the server has work-order persistence enabled. */ workOrderId?: string; + /** Target task pinned when the delegation was accepted. */ + targetTaskId?: string; } -export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; +export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many" | "capacity" | "invalid_input"; /** Per source-thread queue. Persisted to delegations.json on every change * and reloaded at boot: a handoff queued right before a restart runs after @@ -81,6 +83,7 @@ export function _loadPending(): void { ...(typeof item.reason === "string" ? { reason: item.reason } : {}), depth: Math.max(0, Math.trunc(item.depth!)), ...(typeof item.workOrderId === "string" && item.workOrderId ? { workOrderId: item.workOrderId } : {}), + ...(typeof item.targetTaskId === "string" && item.targetTaskId ? { targetTaskId: item.targetTaskId } : {}), }]; }); if (items.length) pendingDelegations.set(threadId, items); @@ -108,6 +111,7 @@ export function queueDelegation( maxDepth: number, sourceThreadId = from.threadId, workOrders?: WorkOrderStore, + sourceExecutionId?: string, ): QueueResult { if (item.toBotId === from.id) return "self"; if (item.depth >= maxDepth) return "too_deep"; @@ -118,12 +122,13 @@ export function queueDelegation( // making the caller wait. Without a cap, one turn can queue unboundedly // and fan out into as many real turns on the next settle. if (list.length >= MAX_QUEUED_PER_THREAD) return "too_many"; - const workOrder = workOrders?.create( - { + let workOrder; + try { + workOrder = workOrders?.create({ kind: "delegation", sourceBotId: from.id, sourceTaskId: sourceThreadId, - sourceExecutionId: sourceThreadId, + ...(sourceExecutionId ? { sourceExecutionId } : {}), targetBotId: target.id, targetTaskId: target.threadId, request: item.message, @@ -131,10 +136,18 @@ export function queueDelegation( priority: "peer", depth: item.depth, delivery: "channel", - }, - "queued", - ); - list.push({ ...item, id: newId(), ...(workOrder ? { workOrderId: workOrder.id } : {}) }); + }); + } catch (error) { + if (error instanceof WorkOrderCapacityError) return "capacity"; + if (error instanceof WorkOrderInputError) return "invalid_input"; + throw error; + } + list.push({ + ...item, + id: newId(), + targetTaskId: target.threadId, + ...(workOrder ? { workOrderId: workOrder.id } : {}), + }); pendingDelegations.set(sourceThreadId, list); savePending(); const label = `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`; @@ -163,6 +176,7 @@ export function drainDelegations( sourceThreadId: string, channel?: GroupRecord, workOrderId?: string, + targetTaskId?: string, ) => void | Promise, workOrders?: WorkOrderStore, ): void { @@ -229,7 +243,7 @@ export function discardDelegations(bus: CommsBus, threadId: string, workOrders?: if (!list?.length) return; pendingDelegations.delete(threadId); savePending(); - for (const item of list) transition(workOrders, item.workOrderId, "cancelled", { error: "source turn did not finish" }); + for (const item of list) transitionIfActive(workOrders, item.workOrderId, "cancelled", { error: "source turn did not finish" }); const from = bus.store.botByThread(threadId); if (!from) return; bus.store.appendMessage(threadId, { @@ -239,6 +253,48 @@ export function discardDelegations(bus: CommsBus, threadId: string, workOrders?: }); } +/** Remove a queued item by its durable id. The in-memory queue can otherwise + * outlive a cancellation request and dispatch work that the API already + * reported as cancelled. */ +export function cancelDelegationByWorkOrder(workOrderId: string, workOrders?: WorkOrderStore): boolean { + let removed = false; + for (const [threadId, list] of pendingDelegations) { + const remaining = list.filter((item) => item.workOrderId !== workOrderId); + if (remaining.length === list.length) continue; + removed = true; + if (remaining.length) pendingDelegations.set(threadId, remaining); + else pendingDelegations.delete(threadId); + } + if (removed) savePending(); + cancelPeerApprovalForOwner(workOrderId); + transitionIfActive(workOrders, workOrderId, "cancelled", { error: "cancelled by user" }); + return removed; +} + +/** Drop queued items targeting a bot that is being deleted. */ +export function discardDelegationsForTarget(bus: CommsBus, botId: string, workOrders?: WorkOrderStore): void { + let changed = false; + for (const [threadId, list] of pendingDelegations) { + const removed = list.filter((item) => item.toBotId === botId); + if (!removed.length) continue; + changed = true; + const remaining = list.filter((item) => item.toBotId !== botId); + if (remaining.length) pendingDelegations.set(threadId, remaining); + else pendingDelegations.delete(threadId); + for (const item of removed) { + transitionIfActive(workOrders, item.workOrderId, "failed", { error: "target bot was deleted" }); + } + if (bus.store.botByThread(threadId)) { + bus.store.appendMessage(threadId, { + role: "bot", + kind: "activity", + tool: { name: `error: delegation to ${botId} failed — target bot was deleted`, ok: false }, + }); + } + } + if (changed) savePending(); +} + async function processOne( bus: CommsBus, approvalBus: ApprovalBus, @@ -252,27 +308,44 @@ async function processOne( sourceThreadId: string, channel?: GroupRecord, workOrderId?: string, + targetTaskId?: string, ) => void | Promise, workOrders?: WorkOrderStore, ): Promise<"done" | "defer"> { + const existing = workOrders?.get(item.workOrderId ?? ""); + if (existing && ["completed", "failed", "cancelled"].includes(existing.state)) return "done"; let sender = from; let target = bus.store.bot(item.toBotId); + const targetTaskId = existing?.targetTaskId ?? item.targetTaskId ?? target?.threadId; + const sourceTaskId = existing?.sourceTaskId ?? sourceThreadId; + if (workOrders && item.workOrderId && (!existing || existing.sourceTaskId !== sourceThreadId)) return "done"; + if (!bus.store.taskByThread(from.id, sourceTaskId)) { + transitionIfActive(workOrders, item.workOrderId, "failed", { error: "source task no longer exists" }); + return "done"; + } if (!target) { bus.store.appendMessage(sourceThreadId, { role: "bot", kind: "activity", tool: { name: `error: delegation to ${item.toBotId} failed — no such bot`, ok: false }, }); - transition(workOrders, item.workOrderId, "failed", { error: "target bot no longer exists" }); + transitionIfActive(workOrders, item.workOrderId, "failed", { error: "target bot no longer exists" }); + return "done"; + } + if (!targetTaskId || !bus.store.taskByThread(target.id, targetTaskId)) { + transitionIfActive(workOrders, item.workOrderId, "failed", { error: "target task no longer exists" }); return "done"; } + if (existing?.state === "queued") { + // A recovered/approved item has already passed the source approval gate. + } if (target.busy) { // Busy is admission pressure, not a terminal failure. The item remains // durable and is retried when the target's current turn settles. return "defer"; } - if (sender.approvePeerComms) { - transition(workOrders, item.workOrderId, "awaiting-approval"); + if (sender.approvePeerComms && existing?.state !== "queued") { + transitionIfActive(workOrders, item.workOrderId, "awaiting-approval"); const verdict = await requestPeerApproval( approvalBus, sender, @@ -280,14 +353,17 @@ async function processOne( item.message, "delegate_bot", sourceThreadId, + item.workOrderId, ); + const afterApproval = workOrders?.get(item.workOrderId ?? ""); + if (afterApproval?.state === "cancelled" || afterApproval?.state === "failed") return "done"; if (verdict !== "allow") { bus.store.appendMessage(sourceThreadId, { role: "bot", kind: "activity", tool: { name: `Delegation to @${target.name} denied by user`, ok: false }, }); - transition(workOrders, item.workOrderId, "cancelled", { error: "denied by user" }); + transitionIfActive(workOrders, item.workOrderId, "cancelled", { error: "denied by user" }); return "done"; } // The approval could have been sitting for up to 15 minutes. Everything @@ -296,46 +372,49 @@ async function processOne( // and mirror a "Messaged @X" chip for an exchange that never happens. const current = bus.store.bot(item.toBotId); const currentSender = bus.store.bot(from.id); - if (!current || !currentSender || !bus.store.taskByThread(currentSender.id, sourceThreadId)) { - transition(workOrders, item.workOrderId, "failed", { error: "source or target task no longer exists" }); + if (!current || !currentSender || !bus.store.taskByThread(currentSender.id, sourceTaskId) || !bus.store.taskByThread(current.id, targetTaskId)) { + transitionIfActive(workOrders, item.workOrderId, "failed", { error: "source or target task no longer exists" }); return "done"; } if (current.busy) { - transition(workOrders, item.workOrderId, "queued"); + transitionIfActive(workOrders, item.workOrderId, "queued"); return "defer"; } sender = currentSender; target = current; } + if (existing?.state === "pending-source") transitionIfActive(workOrders, item.workOrderId, "queued"); + const currentOrder = workOrders?.get(item.workOrderId ?? ""); + if (currentOrder?.state === "cancelled" || currentOrder?.state === "failed") return "done"; const channel = getOrCreateChannel(bus.store, sender, target); mirrorExchange(bus, sender, target, item.message, channel, sourceThreadId); const reasonLine = item.reason ? `\n\n[Reason: ${item.reason}]` : ""; const prefixed = `[Delegated by @${sender.name}, another bot in this OpenMausBot workspace. Do the work and reply directly.]\n\n${item.message}${reasonLine}`; - transition(workOrders, item.workOrderId, "running", { attempt: (workOrders?.get(item.workOrderId ?? "")?.attempt ?? 0) + 1 }); + transitionIfActive(workOrders, item.workOrderId, "running", { attempt: (workOrders?.get(item.workOrderId ?? "")?.attempt ?? 0) + 1 }); try { - await runTarget(item.toBotId, prefixed, item.depth + 1, sourceThreadId, channel, item.workOrderId); - transition(workOrders, item.workOrderId, "completed"); + await runTarget(item.toBotId, prefixed, item.depth + 1, sourceThreadId, channel, item.workOrderId, targetTaskId); + if (workOrders?.get(item.workOrderId ?? "")?.state === "running") { + transitionIfActive(workOrders, item.workOrderId, "completed"); + } } catch (error) { - transition(workOrders, item.workOrderId, "failed", { error: error instanceof Error ? error.message : String(error) }); + if (workOrders?.get(item.workOrderId ?? "")?.state === "running") { + transitionIfActive(workOrders, item.workOrderId, "failed", { error: error instanceof Error ? error.message : String(error) }); + } throw error; } return "done"; } -function transition( +function transitionIfActive( workOrders: WorkOrderStore | undefined, id: string | undefined, to: "awaiting-approval" | "queued" | "running" | "completed" | "failed" | "cancelled", patch: { result?: string; error?: string; attempt?: number } = {}, ): void { if (!workOrders || !id) return; - try { - workOrders.transition(id, to, patch); - } catch (error) { - // A terminal order is intentionally immutable; the provider terminal - // event may race the drain's acknowledgement, so this is diagnostic only. - console.error("work-order transition failed", error); - } + const current = workOrders.get(id); + if (!current || ["completed", "failed", "cancelled"].includes(current.state)) return; + workOrders.transition(id, to, patch); } /** Test helper: how many items remain queued for a thread. */ diff --git a/server/index.ts b/server/index.ts index b99178367..66b0693ea 100644 --- a/server/index.ts +++ b/server/index.ts @@ -67,11 +67,20 @@ import { RETRY_MAX_ATTEMPTS } from "./drivers/retry.ts"; import { BUILT_IN_DRIVERS } from "./drivers/builtIn.ts"; import { getOrCreateChannel, mirrorActivity, mirrorExchange, mirrorReply, type CommsBus } from "./comms-visibility.ts"; import { searchMessages } from "./message-db.ts"; -import { _loadPending, discardDelegations, drainDelegations, pendingThreads, queueDelegation, type QueueResult } from "./delegations.ts"; +import { + _loadPending, + cancelDelegationByWorkOrder, + discardDelegations, + discardDelegationsForTarget, + drainDelegations, + pendingThreads, + queueDelegation, + type QueueResult, +} from "./delegations.ts"; import { drainSteeredMessages, queueSteeredMessage } from "./steer-queue.ts"; import { EventBus } from "./harness/bus.ts"; import { ProviderRegistry } from "./harness/registry.ts"; -import { cancelPeerApprovalsFor, cancelPeerApprovalsForThread, dismissStalePeerCards, requestPeerApproval, resolvePeerComms, type ApprovalBus } from "./peer-approval.ts"; +import { cancelPeerApprovalForOwner, cancelPeerApprovalsFor, cancelPeerApprovalsForThread, dismissStalePeerCards, requestPeerApproval, resolvePeerComms, type ApprovalBus } from "./peer-approval.ts"; import { mentionedBots, roomResponders, @@ -113,7 +122,7 @@ import { WebhookManager } from "./webhooks.ts"; import { SPAWNED_PROXIES } from "./proxy-paths.ts"; import { loadBundledSkills, loadUserSkills, mergeSkills, renderSkillInstructions, selectBundledSkills } from "./skill-library.ts"; import { shouldMountLocalComputer } from "./local-routing.ts"; -import { WorkOrderStore, type WorkOrderState } from "./work-orders.ts"; +import { WorkOrderCapacityError, WorkOrderInputError, WorkOrderStore, type WorkOrderState } from "./work-orders.ts"; import { TurnScheduler, type TurnLane } from "./turn-scheduler.ts"; const PORT = Number(process.env.OMB_PORT || process.env.OGB_PORT || 8799); @@ -226,7 +235,7 @@ function askBotAndWait( rejectStartFailure = false, ): Promise { const target = store.bot(targetBotId); - if (!target) return Promise.resolve("(no such bot)"); + if (!target) return rejectStartFailure ? Promise.reject(new Error("no such bot")) : Promise.resolve("(no such bot)"); const threadId = target.threadId; return new Promise((resolve, reject) => { let text = ""; @@ -250,10 +259,17 @@ function askBotAndWait( if (e.type === "item.completed" && e.itemType === "assistant_text") { text += (text ? "\n" : "") + e.text; } else if (e.type === "turn.completed") { - finish(text || "(the bot finished without a text reply)"); + if (rejectStartFailure && !e.ok) { + fail(new Error("the target bot turn failed or was interrupted")); + } else { + finish(text || "(the bot finished without a text reply)"); + } } }); - const timer = setTimeout(() => finish(text || "(timed out waiting for the bot to reply)"), 4 * 60_000); + const timer = setTimeout(() => { + if (rejectStartFailure) fail(new Error("timed out waiting for the bot to reply")); + else finish(text || "(timed out waiting for the bot to reply)"); + }, 4 * 60_000); startTurn(targetBotId, message, { commsDepth: depth + 1, unattended: isUnattended(fromBotId), @@ -287,6 +303,8 @@ const workOrders = new WorkOrderStore({ }); const turnScheduler = new TurnScheduler({ maxPendingPerBot: 32, reservedUserSlots: 4 }); const turnAdmissionByThread = new Map(); +const activeExecutionByThread = new Map(); +const pendingSchedulerAdmissions = new Map(); function laneForTurn(opts?: { commsDepth?: number; automationSource?: RoutineRunTrigger; connectorContinuation?: boolean }): TurnLane { if (opts?.commsDepth || opts?.connectorContinuation) return "peer"; @@ -301,6 +319,37 @@ function releaseTurnAdmission(threadId: string): void { turnScheduler.release(admission.botId, admission.token); } +const TURN_SETTLEMENT_GRACE_MS = 6_000; + +function settleTurnAfterGrace(threadId: string, botId: string): void { + const timer = setTimeout(() => { + const admission = turnAdmissionByThread.get(threadId); + if (!admission || admission.botId !== botId) return; + const group = store.groupByThread(threadId); + const speaker = groupSpeakers.get(threadId); + if (group && group.busyBotId === botId && speaker?.botId === botId) { + groupSpeakers.delete(threadId); + store.patchGroup(group.id, { busyBotId: null, unread: true }); + } + const bot = store.bot(botId); + if (bot?.busy) { + stopScreenPoller(bot.id); + if (activeVpsThreads.get(bot.id) === threadId) activeVpsThreads.delete(bot.id); + store.setActivity(bot.id, "idle"); + } + releaseTurnAdmission(threadId); + }, TURN_SETTLEMENT_GRACE_MS); + timer.unref?.(); +} + +function cancelQueuedWorkOrderRuntime(workOrderId: string): void { + const admission = pendingSchedulerAdmissions.get(workOrderId); + if (!admission || !admission.queued) return; + if (turnScheduler.cancel(admission.botId, admission.admissionId)) { + pendingSchedulerAdmissions.delete(workOrderId); + } +} + /** A bot as a client may see it: no provider session bookkeeping. * * `resumeCursors` is the harness's own bookkeeping — the native session id @@ -630,21 +679,7 @@ const watchdog = new TurnWatchdog({ // sooner. Keep ownership during that grace period so another turn cannot // overlap the process we are stopping. The normal turn.completed fold // clears it first when the adapter responds. - const release = setTimeout(() => { - const group = store.groupByThread(turn.threadId); - const speaker = groupSpeakers.get(turn.threadId); - if (group && group.busyBotId === turn.botId && speaker?.botId === turn.botId) { - groupSpeakers.delete(turn.threadId); - store.patchGroup(group.id, { busyBotId: null, unread: true }); - } - const currentBot = store.bot(turn.botId); - if (currentBot?.busy) { - stopScreenPoller(currentBot.id); - if (activeVpsThreads.get(currentBot.id) === turn.threadId) activeVpsThreads.delete(currentBot.id); - store.setActivity(currentBot.id, "idle"); - } - }, 6_000); - release.unref?.(); + settleTurnAfterGrace(turn.threadId, turn.botId); }, }); watchdog.start(); @@ -652,7 +687,10 @@ watchdog.start(); bus.subscribe((event: RuntimeEvent) => { if (event.type === "request.opened") watchdog.setWaitingOnHuman(event.threadId, true); else if (event.type === "request.resolved") watchdog.setWaitingOnHuman(event.threadId, false); - else if (event.type === "turn.completed") watchdog.settle(event.threadId); + else if (event.type === "turn.completed") { + watchdog.settle(event.threadId); + activeExecutionByThread.delete(event.threadId); + } else watchdog.touch(event.threadId); }); @@ -1146,12 +1184,15 @@ bus.subscribe((event: RuntimeEvent) => { /** How a drained delegation becomes a real turn on the target. Shared by * the settle-time drain and the boot-time drain of what a previous process * left queued. */ -const runDelegatedTurn: Parameters[3] = async (toBotId, text, commsDepth, sourceThreadId, channel, workOrderId) => { +const runDelegatedTurn: Parameters[3] = async (toBotId, text, commsDepth, sourceThreadId, channel, workOrderId, targetTaskId) => { // startTurn REJECTS on an ordinary condition — busy target, deleted bot, // unavailable provider. Unhandled, that rejection is fatal to the // harness (Node's default), which in the packaged app kills the server // child. Every delegation failure has to land as a chip instead. - const targetThreadId = store.bot(toBotId)?.threadId; + const target = store.bot(toBotId); + const targetThreadId = targetTaskId && target?.tasks?.some((task) => task.threadId === targetTaskId) + ? targetTaskId + : target?.threadId; let failureReported = false; const reportStartFailure = (error: unknown) => { if (failureReported) return; @@ -1188,6 +1229,7 @@ const runDelegatedTurn: Parameters[3] = async (toBotId, } await startTurn(toBotId, text, { commsDepth, + threadId: targetTaskId, unattended: isUnattended(store.botByThread(sourceThreadId)?.id), schedulerToken, // startTurn schedules provider/integration setup after marking the bot @@ -1209,7 +1251,20 @@ const runDelegatedTurn: Parameters[3] = async (toBotId, throw new Error(reason); } schedulerToken = admission.admission.id; - await admission.completion; + if (workOrderId) { + pendingSchedulerAdmissions.set(workOrderId, { + botId: toBotId, + admissionId: admission.admission.id, + queued: admission.admission.queued, + }); + } + try { + await admission.completion; + } finally { + if (workOrderId && pendingSchedulerAdmissions.get(workOrderId)?.admissionId === admission.admission.id) { + pendingSchedulerAdmissions.delete(workOrderId); + } + } }; bus.subscribe((event: RuntimeEvent) => { @@ -1492,6 +1547,11 @@ async function startTurn( // in the background — box provisioning can take ~90s and must never // hang the HTTP request const admissionToken = opts?.schedulerToken ?? turnScheduler.occupy(bot.id, undefined, laneForTurn(opts)); + if (!admissionToken) { + throw Object.assign(new Error("the bot is already working — interrupt it first"), { status: 409 }); + } + const executionId = randomUUID(); + activeExecutionByThread.set(threadId, executionId); turnAdmissionByThread.set(threadId, { botId: bot.id, token: admissionToken }); store.setActivity(bot.id, "working"); store.patchBot(bot.id, { unread: false }); @@ -1780,6 +1840,7 @@ async function startTurn( startScreenPoller(bot.id, previewCapture, { screenIsTheWork: instance.driverKind === "boxAgent" }); } } catch (e) { + if (activeExecutionByThread.get(threadId) === executionId) activeExecutionByThread.delete(threadId); releaseLocalVmThread(threadId); if (activeVpsThreads.get(bot.id) === threadId) activeVpsThreads.delete(bot.id); watchdog.settle(threadId); @@ -1909,13 +1970,14 @@ function acceptDeferredConsultation( message: string, depth: number, sourceThreadId: string, + sourceExecutionId?: string, ) { const channel = getOrCreateChannel(store, from, target); const order = workOrders.create({ kind: "consultation", sourceBotId: from.id, sourceTaskId: sourceThreadId, - sourceExecutionId: sourceThreadId, + ...(sourceExecutionId ? { sourceExecutionId } : {}), targetBotId: target.id, targetTaskId: target.threadId, request: message, @@ -1969,10 +2031,15 @@ function schedulePeerContinuation( }); } -function queuePeerContinuation(order: { id: string; sourceBotId: string; sourceTaskId: string; targetBotId: string }, reply: string) { +function queuePeerContinuation( + order: { id: string; sourceBotId: string; sourceTaskId: string; targetBotId: string }, + reply: string, +): { ok: true } | { ok: false; error: string } { const target = store.bot(order.targetBotId); const source = store.bot(order.sourceBotId); - if (!target || !source || !store.taskByThread(source.id, order.sourceTaskId)) return; + if (!target || !source || !store.taskByThread(source.id, order.sourceTaskId)) { + return { ok: false, error: "pinned source bot or task no longer exists" }; + } const prompt = `[Deferred consultation result from @${target.name}. The earlier ask_bot receipt meant acceptance, not completion. Do not call peer tools for this continuation.]\n\n@${target.name} replied:\n${reply}`; if (source.busy || turnScheduler.hasActive(source.id)) { const current = pendingPeerContinuations.get(order.sourceTaskId); @@ -1981,9 +2048,10 @@ function queuePeerContinuation(order: { id: string; sourceBotId: string; sourceT sourceTaskId: order.sourceTaskId, prompts: [...(current?.prompts ?? []), prompt], }); - return; + return { ok: true }; } schedulePeerContinuation(source, order.sourceTaskId, [prompt], `continuation:${order.id}`); + return { ok: true }; } async function drainDeferredConsultations(): Promise { @@ -1995,7 +2063,7 @@ async function drainDeferredConsultations(): Promise { workOrders.transition(order.id, "failed", { error: "pinned source or target task no longer exists" }); continue; } - if (target.busy) continue; + if (target.busy || turnScheduler.hasActive(target.id)) continue; runningConsultations.add(order.id); const channel = order.channelId ? store.group(order.channelId) ?? undefined : undefined; void (async () => { @@ -2007,28 +2075,46 @@ async function drainDeferredConsultations(): Promise { lane: "peer", dedupeKey: order.id, run: async () => { + if (workOrders.get(order.id)?.state !== "queued") return; workOrders.transition(order.id, "running", { attempt: (order.attempt ?? 0) + 1 }); const reply = await askBotAndWait(target.id, prefixed, order.depth, source.id, schedulerToken, true); + if (workOrders.get(order.id)?.state !== "running") return; + const delivery = queuePeerContinuation(order, reply); + if (!delivery.ok) { + workOrders.transition(order.id, "failed", { error: delivery.error }); + if (channel) mirrorActivity(commsBus, target, channel, `Consultation failed — ${delivery.error}`, false); + return; + } workOrders.transition(order.id, "completed", { result: reply }); mirrorReply(commsBus, target, reply, channel); - queuePeerContinuation(order, reply); }, }); if (!admission.accepted) { runningConsultations.delete(order.id); - if (admission.reason !== "duplicate") workOrders.transition(order.id, "queued"); + if (admission.reason !== "duplicate" && workOrders.get(order.id)?.state === "queued") workOrders.transition(order.id, "queued"); return; } schedulerToken = admission.admission.id; - await admission.completion; + pendingSchedulerAdmissions.set(order.id, { + botId: target.id, + admissionId: admission.admission.id, + queued: admission.admission.queued, + }); + try { + await admission.completion; + } finally { + if (pendingSchedulerAdmissions.get(order.id)?.admissionId === admission.admission.id) { + pendingSchedulerAdmissions.delete(order.id); + } + } } catch (error) { const reason = error instanceof Error ? error.message : String(error); - try { + if (workOrders.get(order.id)?.state === "queued" || workOrders.get(order.id)?.state === "running") { workOrders.transition(order.id, "failed", { error: reason }); - } catch (transitionError) { - console.error("deferred consultation transition failed", transitionError); } - if (channel) mirrorActivity(commsBus, target, channel, `Consultation failed — ${reason.slice(0, 120)}`, false); + if (channel && workOrders.get(order.id)?.state === "failed") { + mirrorActivity(commsBus, target, channel, `Consultation failed — ${reason.slice(0, 120)}`, false); + } } finally { runningConsultations.delete(order.id); } @@ -2062,6 +2148,7 @@ function drainPeerContinuations(): void { if (recovered.cancelled.length || recovered.failed.length) { console.log(`work orders: recovered ${recovered.cancelled.length} cancelled and ${recovered.failed.length} failed record(s)`); } + void drainDeferredConsultations(); } // Handoffs a previous process queued but never ran: the source turn is @@ -2135,8 +2222,17 @@ async function runGroupMemberTurn( }); return true; } - store.setActivity(bot.id, "working"); const roomAdmissionToken = turnScheduler.occupy(bot.id, undefined, "user"); + if (!roomAdmissionToken) { + store.appendMessage(group.threadId, { + role: "bot", + kind: "activity", + from: { botId: bot.id, name: bot.name, color: bot.color }, + tool: { name: `${bot.name} is busy in another conversation — skipped this round`, ok: false }, + }); + return true; + } + store.setActivity(bot.id, "working"); turnAdmissionByThread.set(group.threadId, { botId: bot.id, token: roomAdmissionToken }); store.patchGroup(group.id, { busyBotId: bot.id }); // the store's change stream carries the frame @@ -2201,6 +2297,7 @@ async function runGroupMemberTurn( }); timer = scheduleRoomTurnTimeout(timeoutMinutes, () => { void instance.adapter.interruptTurn(group.threadId).catch(() => {}); + watchdog.settle(group.threadId); store.appendMessage(group.threadId, { role: "bot", kind: "activity", @@ -2235,7 +2332,10 @@ async function runGroupMemberTurn( // produces turn.completed (or the stall watchdog's grace fallback runs). // Do not clear busy or start the next member on that same thread early. if (outcome === "dispatch_failed") releaseTurnAdmission(group.threadId); - if (outcome === "stalled" || outcome === "timed_out") return false; + if (outcome === "stalled" || outcome === "timed_out") { + settleTurnAfterGrace(group.threadId, bot.id); + return false; + } // turn.completed normally performs this cleanup. Only use the fallback // when this invocation still owns the room; otherwise it would emit a // duplicate group frame or clear a newer speaker's state. @@ -2800,9 +2900,22 @@ const server = createServer(async (req, res) => { currentFrom = freshFrom; currentTarget = freshTarget; } - if (currentTarget.busy) { - const accepted = acceptDeferredConsultation(currentFrom, currentTarget, message, depth, fromThreadId); - return json(res, 202, accepted); + if (currentTarget.busy || turnScheduler.hasActive(currentTarget.id)) { + try { + const accepted = acceptDeferredConsultation( + currentFrom, + currentTarget, + message, + depth, + fromThreadId, + activeExecutionByThread.get(fromThreadId), + ); + return json(res, 202, accepted); + } catch (error) { + if (error instanceof WorkOrderCapacityError) return json(res, 429, { error: error.message }); + if (error instanceof WorkOrderInputError) return json(res, 400, { error: error.message }); + throw error; + } } const channel = getOrCreateChannel(store, currentFrom, currentTarget); mirrorExchange(commsBus, currentFrom, currentTarget, message, channel, fromThreadId); @@ -2840,6 +2953,7 @@ const server = createServer(async (req, res) => { MAX_COMMS_DEPTH, fromThreadId, workOrders, + activeExecutionByThread.get(fromThreadId), ); if (result !== "ok") { // the agent reads this string — a bare enum ("too_deep") tells it @@ -2849,6 +2963,8 @@ const server = createServer(async (req, res) => { too_deep: "delegation chains are limited to one hop — do this one yourself", no_target: "no such bot", too_many: "too many delegations queued on this turn — finish some first", + capacity: "the peer work queue is full — try again after some work finishes", + invalid_input: "the delegation request or reason is too long", }; return json(res, 200, { error: said[result] }); } @@ -3027,8 +3143,12 @@ const server = createServer(async (req, res) => { } workOrderMatch = path.match(/^\/api\/work-orders\/([\w-]+)\/cancel$/); if (workOrderMatch && method === "POST") { - const order = workOrders.cancel(workOrderMatch[1], String((await readBody(req)).reason ?? "cancelled by user")); + const workOrderId = workOrderMatch[1]; + const order = workOrders.cancel(workOrderId, String((await readBody(req)).reason ?? "cancelled by user")); if (!order) return json(res, 404, { error: "no such active work order" }); + cancelDelegationByWorkOrder(workOrderId, workOrders); + cancelQueuedWorkOrderRuntime(workOrderId); + cancelPeerApprovalForOwner(workOrderId); // A provider interrupt is deliberately not implicit here: cancellation // of queued peer work is safe; an active provider turn owns its own // process and follows the normal Stop/approval path. @@ -4021,7 +4141,20 @@ const server = createServer(async (req, res) => { // a peer approval naming this bot can never be meaningfully answered // now, and its caller would otherwise wait out the 15-minute timeout cancelPeerApprovalsFor(bot.id); - discardDelegations(commsBus, bot.threadId); + const sourceThreads = new Set([bot.threadId, ...(bot.tasks ?? []).map((task) => task.threadId)]); + for (const threadId of sourceThreads) discardDelegations(commsBus, threadId, workOrders); + discardDelegationsForTarget(commsBus, bot.id, workOrders); + workOrders.settleForDeletedBot(bot.id); + for (const [workOrderId, admission] of pendingSchedulerAdmissions) { + if (admission.botId !== bot.id || !admission.queued) continue; + if (turnScheduler.cancel(admission.botId, admission.admissionId)) pendingSchedulerAdmissions.delete(workOrderId); + } + for (const [sourceTaskId, pending] of pendingPeerContinuations) { + if (pending.sourceBotId === bot.id) pendingPeerContinuations.delete(sourceTaskId); + } + for (const [threadId] of activeExecutionByThread) { + if (sourceThreads.has(threadId)) activeExecutionByThread.delete(threadId); + } computerControl.forget(bot.id); const target = perBotLocalVmTarget(bot.id); localVmIdles.get(target.key)?.cancel(); diff --git a/server/peer-approval.ts b/server/peer-approval.ts index 805ca6234..1c917bc8f 100644 --- a/server/peer-approval.ts +++ b/server/peer-approval.ts @@ -42,6 +42,7 @@ interface Pending { threadId: string; messageId: string; bus: ApprovalBus; + ownerKey?: string; } /** Mark the card answered so the UI stops treating it as pending. Mirrors @@ -106,6 +107,7 @@ export function requestPeerApproval( message: string, action: PeerAction, sourceThreadId = from.threadId, + ownerKey?: string, ): Promise<"allow" | "deny"> { if (allowKeyAllowed(from, peerAllowKey(action, target.id))) { return Promise.resolve("allow"); @@ -134,10 +136,23 @@ export function requestPeerApproval( threadId: sourceThreadId, messageId: card.id, bus, + ...(ownerKey ? { ownerKey } : {}), }); }); } +/** Deny one durable work-order's approval without disturbing other cards in + * the same source task. */ +export function cancelPeerApprovalForOwner(ownerKey: string): void { + for (const [requestId, pending] of pendingComms) { + if (pending.ownerKey !== ownerKey) continue; + pendingComms.delete(requestId); + clearTimeout(pending.timer); + settleCard(pending, "deny", "system"); + pending.resolve("deny"); + } +} + /** Called by the respond endpoints BEFORE forwarding to the provider * adapter. Returns true if the requestId belonged to a pending peer * approval (and resolves it); false if it was a provider request and diff --git a/server/turn-scheduler.test.ts b/server/turn-scheduler.test.ts index dcf5c551a..e7148a89d 100644 --- a/server/turn-scheduler.test.ts +++ b/server/turn-scheduler.test.ts @@ -14,19 +14,21 @@ describe("TurnScheduler", () => { const first = deferred(); const order: string[] = []; const active = scheduler.admit({ botId: "a", lane: "peer", run: async () => { order.push("active"); await first.promise; } }); - expect(active.accepted && active.admission.queued).toBe(false); const normal = scheduler.admit({ botId: "a", lane: "peer", run: async () => { order.push("peer"); } }); const background = scheduler.admit({ botId: "a", lane: "background", run: async () => { order.push("background"); } }); const urgent = scheduler.admit({ botId: "a", lane: "urgent-peer", run: async () => { order.push("urgent"); } }); const user = scheduler.admit({ botId: "a", lane: "user", run: async () => { order.push("user"); } }); - if (!active.accepted || !normal.accepted) throw new Error("expected scheduler admissions"); + if (!active.accepted || !normal.accepted || !background.accepted || !urgent.accepted || !user.accepted) { + throw new Error("expected scheduler admission"); + } + expect(active.admission.queued).toBe(false); first.resolve(); await Promise.all([ active.completion, normal.completion, - background.accepted && background.completion, - urgent.accepted && urgent.completion, - user.accepted && user.completion, + background.completion, + urgent.completion, + user.completion, ]); expect(order).toEqual(["active", "user", "urgent", "peer", "background"]); }); @@ -40,18 +42,20 @@ describe("TurnScheduler", () => { const started = new Promise((resolve) => { resolveStarted = resolve; }); const aAdmission = scheduler.admit({ botId: "a", lane: "user", run: async () => { seen.push("a-start"); if (seen.length === 2) resolveStarted(); await a.promise; seen.push("a-end"); } }); const bAdmission = scheduler.admit({ botId: "b", lane: "user", run: async () => { seen.push("b-start"); if (seen.length === 2) resolveStarted(); await b.promise; seen.push("b-end"); } }); + if (!aAdmission.accepted || !bAdmission.accepted) throw new Error("expected concurrent admissions"); await started; expect(seen).toEqual(["a-start", "b-start"]); a.resolve(); b.resolve(); - if (aAdmission.accepted && bAdmission.accepted) await Promise.all([aAdmission.completion, bAdmission.completion]); + await Promise.all([aAdmission.completion, bAdmission.completion]); expect(seen).toEqual(["a-start", "b-start", "a-end", "b-end"]); }); it("reserves capacity for user work and supports cancellation", async () => { const scheduler = new TurnScheduler({ maxPendingPerBot: 3, reservedUserSlots: 1 }); const hold = deferred(); - scheduler.admit({ botId: "a", lane: "user", run: () => hold.promise }); + const active = scheduler.admit({ botId: "a", lane: "user", run: () => hold.promise }); + expect(active.accepted).toBe(true); const peer = scheduler.admit({ botId: "a", lane: "peer", run: () => {} }); expect(peer.accepted).toBe(true); const remaining = scheduler.admit({ botId: "a", lane: "peer", run: () => {} }); @@ -82,4 +86,29 @@ describe("TurnScheduler", () => { second.resolve(); if (queued.accepted) await queued.completion; }); + + it("deduplicates a queued entry before it reaches the provider", async () => { + const scheduler = new TurnScheduler(); + const hold = deferred(); + const active = scheduler.admit({ botId: "a", lane: "user", run: () => hold.promise }); + if (!active.accepted) throw new Error("expected active admission"); + const queued = scheduler.admit({ botId: "a", lane: "peer", dedupeKey: "queued", run: () => {} }); + if (!queued.accepted) throw new Error("expected queued admission"); + expect(queued.admission.queued).toBe(true); + expect(scheduler.admit({ botId: "a", lane: "background", dedupeKey: "queued", run: () => {} })).toEqual({ + accepted: false, + reason: "duplicate", + }); + hold.resolve(); + await Promise.all([active.completion, queued.completion]); + }); + + it("never hands a legacy caller another run's occupation token", () => { + const scheduler = new TurnScheduler(); + const first = scheduler.occupy("a"); + if (!first) throw new Error("expected first occupation"); + expect(scheduler.occupy("a")).toBeNull(); + expect(scheduler.release("a", "not-the-current-run")).toBe(false); + expect(scheduler.release("a", first)).toBe(true); + }); }); diff --git a/server/turn-scheduler.ts b/server/turn-scheduler.ts index 9f2241be4..05f40ffe8 100644 --- a/server/turn-scheduler.ts +++ b/server/turn-scheduler.ts @@ -147,8 +147,8 @@ export class TurnScheduler { } /** Used when a legacy path owns a turn that was not admitted here. */ - occupy(botId: string, id = newId(), lane: TurnLane = "user"): string { - if (this.active.has(botId)) return this.active.get(botId)!.id; + occupy(botId: string, id = newId(), lane: TurnLane = "user"): string | null { + if (this.active.has(botId)) return null; const generation = (this.generations.get(botId) ?? 0) + 1; this.generations.set(botId, generation); this.active.set(botId, { id, generation, lane }); diff --git a/server/work-orders.test.ts b/server/work-orders.test.ts index 7bf71fc30..9c77c4278 100644 --- a/server/work-orders.test.ts +++ b/server/work-orders.test.ts @@ -70,4 +70,31 @@ describe("WorkOrderStore", () => { expect(store.list({ limit: 20 }).filter((order) => order.state === "cancelled")).toHaveLength(2); expect(store.get(active.id)?.state).toBe("queued"); }); + + it("rejects oversized input without changing accepted request text", () => { + const store = new WorkOrderStore({ file: file() }); + const accepted = "x".repeat(20_000); + expect(store.create({ ...input, request: accepted }).request).toBe(accepted); + expect(() => store.create({ ...input, request: "x".repeat(20_001) })).toThrow(/request/); + expect(() => store.create({ ...input, reason: "x".repeat(2_001) })).toThrow(/reason/); + }); + + it("bounds active work orders and frees capacity after cancellation", () => { + const store = new WorkOrderStore({ file: file(), maxActive: 1 }); + const first = store.create(input); + expect(() => store.create({ ...input, request: "second" })).toThrow(/capacity/); + store.cancel(first.id); + expect(store.create({ ...input, request: "after cancellation" }).state).toBe("pending-source"); + }); + + it("settles both source and target orders when a bot is deleted", () => { + const store = new WorkOrderStore({ file: file() }); + const sourceOrder = store.create(input, "queued"); + const targetOrder = store.create({ ...input, sourceBotId: "other", targetBotId: "source" }, "queued"); + const settled = store.settleForDeletedBot("source"); + expect(settled.cancelled.map((order) => order.id)).toEqual([sourceOrder.id]); + expect(settled.failed.map((order) => order.id)).toEqual([targetOrder.id]); + expect(store.get(sourceOrder.id)?.state).toBe("cancelled"); + expect(store.get(targetOrder.id)?.state).toBe("failed"); + }); }); diff --git a/server/work-orders.ts b/server/work-orders.ts index b1b3fe083..4189ba468 100644 --- a/server/work-orders.ts +++ b/server/work-orders.ts @@ -36,6 +36,27 @@ export interface WorkOrder extends WorkOrderInput { error?: string; } +export const WORK_ORDER_REQUEST_MAX_LENGTH = 20_000; +export const WORK_ORDER_REASON_MAX_LENGTH = 2_000; + +export class WorkOrderCapacityError extends Error { + readonly code = "WORK_ORDER_CAPACITY"; + + constructor(limit: number) { + super(`work-order capacity reached (${limit} active orders)`); + this.name = "WorkOrderCapacityError"; + } +} + +export class WorkOrderInputError extends Error { + readonly code = "WORK_ORDER_INPUT_TOO_LARGE"; + + constructor(field: "request" | "reason", limit: number) { + super(`${field} exceeds the ${limit}-character work-order limit`); + this.name = "WorkOrderInputError"; + } +} + interface DiskFile { version: 1; orders: WorkOrder[]; @@ -61,6 +82,7 @@ export class WorkOrderStore { private readonly file: string; private readonly now: () => number; private readonly maxTerminal: number; + private readonly maxActive: number; private readonly onTransition?: (order: WorkOrder, from: WorkOrderState, to: WorkOrderState) => void; private orders: WorkOrder[] = []; @@ -68,11 +90,13 @@ export class WorkOrderStore { file?: string; now?: () => number; maxTerminal?: number; + maxActive?: number; onTransition?: (order: WorkOrder, from: WorkOrderState, to: WorkOrderState) => void; } = {}) { this.file = options.file ?? join(DATA_DIR, "work-orders.json"); this.now = options.now ?? Date.now; this.maxTerminal = Math.max(1, Math.trunc(options.maxTerminal ?? 200)); + this.maxActive = Math.max(1, Math.trunc(options.maxActive ?? 256)); this.onTransition = options.onTransition; this.load(); } @@ -81,6 +105,15 @@ export class WorkOrderStore { if (state !== "pending-source" && state !== "awaiting-approval" && state !== "queued") { throw new Error("new work orders must begin pending-source, awaiting-approval, or queued"); } + if (input.request.length > WORK_ORDER_REQUEST_MAX_LENGTH) { + throw new WorkOrderInputError("request", WORK_ORDER_REQUEST_MAX_LENGTH); + } + if (input.reason && input.reason.length > WORK_ORDER_REASON_MAX_LENGTH) { + throw new WorkOrderInputError("reason", WORK_ORDER_REASON_MAX_LENGTH); + } + if (this.orders.filter((order) => !TERMINAL.has(order.state)).length >= this.maxActive) { + throw new WorkOrderCapacityError(this.maxActive); + } const at = this.now(); const order: WorkOrder = { ...input, @@ -144,6 +177,21 @@ export class WorkOrderStore { return this.transition(id, "cancelled", { error: reason }); } + /** Settle every active order pinned to a bot that is being deleted. */ + settleForDeletedBot(botId: string): { cancelled: WorkOrder[]; failed: WorkOrder[] } { + const cancelled: WorkOrder[] = []; + const failed: WorkOrder[] = []; + for (const order of [...this.orders]) { + if (TERMINAL.has(order.state)) continue; + if (order.sourceBotId === botId) { + cancelled.push(this.transition(order.id, "cancelled", { error: "source bot was deleted" })); + } else if (order.targetBotId === botId) { + failed.push(this.transition(order.id, "failed", { error: "target bot was deleted" })); + } + } + return { cancelled, failed }; + } + /** Pending/approval work may be reconstructed; source and running work cannot. */ recover(): { cancelled: WorkOrder[]; failed: WorkOrder[] } { const cancelled: WorkOrder[] = []; From 04a5ca40e49a3477ef619a137550376a223f163e Mon Sep 17 00:00:00 2001 From: Omar Haneya Date: Mon, 24 Aug 2026 18:53:25 +0100 Subject: [PATCH 4/4] fix: close peer work settlement races --- server/delegations.test.ts | 72 ++++++++++++++++++++++++++++------ server/delegations.ts | 15 ++++++- server/index.ts | 39 +++++++++--------- server/turn-settlement.test.ts | 49 +++++++++++++++++++++++ server/turn-settlement.ts | 25 ++++++++++++ server/work-orders.test.ts | 23 ++++++++++- server/work-orders.ts | 30 +++++++++++--- 7 files changed, 214 insertions(+), 39 deletions(-) create mode 100644 server/turn-settlement.test.ts create mode 100644 server/turn-settlement.ts diff --git a/server/delegations.test.ts b/server/delegations.test.ts index feea85f7d..ca3f5867f 100644 --- a/server/delegations.test.ts +++ b/server/delegations.test.ts @@ -4,7 +4,8 @@ // assert what would have been dispatched to the harness. The harness itself // stays out of these — the integration happens in comms.test.ts (the full // e2e through the agents proxy + fake ACP CLI). -import { rmSync } from "node:fs"; +import { existsSync, readFileSync, rmSync } from "node:fs"; +import { join } from "node:path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import type { CommsBus } from "./comms-visibility.ts"; @@ -12,11 +13,16 @@ import { DATA_DIR } from "./config.ts"; import type { ModelSelection } from "./contracts.ts"; import { drainDelegations, + discardDelegations, + discardDelegationsForTarget, + _loadPending, _isDraining, + pendingThreads, queueDelegation, _pendingCount, + _resetPending, } from "./delegations.ts"; -import { peerAllowKey, resolvePeerComms } from "./peer-approval.ts"; +import { cancelPeerApprovalsForThread, peerAllowKey, resolvePeerComms } from "./peer-approval.ts"; import { Store, type BotRecord } from "./store.ts"; import { WorkOrderStore } from "./work-orders.ts"; @@ -67,6 +73,7 @@ describe("queueDelegation", () => { beforeEach(() => { rmSync(DATA_DIR, { recursive: true, force: true }); + _resetPending(); store = new Store(selection); from = store.createBot(); target = store.createBot(); @@ -165,6 +172,7 @@ describe("drainDelegations", () => { beforeEach(() => { rmSync(DATA_DIR, { recursive: true, force: true }); + _resetPending(); store = new Store(selection); from = store.createBot(); target = store.createBot(); @@ -176,12 +184,8 @@ describe("drainDelegations", () => { }); afterEach(() => { - // Unresolved approval requests carry a 15-min timer that would otherwise - // keep vitest's event loop alive long after the suite ends. None of the - // tests above leave one — they all resolve via resolvePeerComms — but - // double-check by counting the module's pending map: tests that didn't - // resolve should be re-examined if this ever fires. - void runTargetCalls; + cancelPeerApprovalsForThread(from.threadId); + _resetPending(); }); it("runs the target's turn via runTarget and mirrors the exchange", async () => { @@ -358,6 +362,54 @@ describe("drainDelegations", () => { expect(runTargetCalls[0]!.commsDepth).toBe(1); }); + it("releases a pending approval when the source delegation is discarded", async () => { + store.patchBot(from.id, { approvePeerComms: true }); + const workOrders = new WorkOrderStore({ file: join(DATA_DIR, "discarded-source-work-orders.json") }); + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1, from.threadId, workOrders); + const order = workOrders.list({ limit: 1 })[0]!; + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }, workOrders); + + const card = await waitFor(() => + store.messagesFor(from.threadId).find((message) => message.card?.requestId), + ); + expect(workOrders.get(order.id)?.state).toBe("awaiting-approval"); + + discardDelegations(commsBus, from.threadId, workOrders); + + await waitFor(() => !_isDraining(from.threadId)); + const settledCard = store.messagesFor(from.threadId).find((message) => message.id === card.id); + expect(settledCard?.card).toMatchObject({ answered: "deny", dismissed: true }); + expect(workOrders.get(order.id)).toMatchObject({ state: "cancelled", error: "source turn did not finish" }); + expect(_pendingCount(from.threadId)).toBe(0); + expect(runTargetCalls).toEqual([]); + }); + + it("releases a pending approval when its target is deleted", async () => { + store.patchBot(from.id, { approvePeerComms: true }); + const workOrders = new WorkOrderStore({ file: join(DATA_DIR, "deleted-target-work-orders.json") }); + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1, from.threadId, workOrders); + const order = workOrders.list({ limit: 1 })[0]!; + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }, workOrders); + + const card = await waitFor(() => + store.messagesFor(from.threadId).find((message) => message.card?.requestId), + ); + expect(workOrders.get(order.id)?.state).toBe("awaiting-approval"); + + discardDelegationsForTarget(commsBus, target.id, workOrders); + + await waitFor(() => !_isDraining(from.threadId)); + const settledCard = store.messagesFor(from.threadId).find((message) => message.id === card.id); + expect(settledCard?.card).toMatchObject({ answered: "deny", dismissed: true }); + expect(workOrders.get(order.id)).toMatchObject({ state: "failed", error: "target bot was deleted" }); + expect(_pendingCount(from.threadId)).toBe(0); + expect(runTargetCalls).toEqual([]); + }); + it("emits a denial chip and skips runTarget when the user denies", async () => { store.patchBot(from.id, { approvePeerComms: true }); queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); @@ -414,10 +466,6 @@ describe("drainDelegations", () => { }); }); -import { existsSync, readFileSync } from "node:fs"; -import { join } from "node:path"; -import { _loadPending, _resetPending, discardDelegations, pendingThreads } from "./delegations.ts"; - describe("delegations survive a restart", () => { let store: Store; let from: BotRecord; diff --git a/server/delegations.ts b/server/delegations.ts index 9cf69574c..b240279b5 100644 --- a/server/delegations.ts +++ b/server/delegations.ts @@ -243,7 +243,10 @@ export function discardDelegations(bus: CommsBus, threadId: string, workOrders?: if (!list?.length) return; pendingDelegations.delete(threadId); savePending(); - for (const item of list) transitionIfActive(workOrders, item.workOrderId, "cancelled", { error: "source turn did not finish" }); + for (const item of list) { + cancelPeerApprovalForOwner(approvalOwner(item)); + transitionIfActive(workOrders, item.workOrderId, "cancelled", { error: "source turn did not finish" }); + } const from = bus.store.botByThread(threadId); if (!from) return; bus.store.appendMessage(threadId, { @@ -282,6 +285,7 @@ export function discardDelegationsForTarget(bus: CommsBus, botId: string, workOr if (remaining.length) pendingDelegations.set(threadId, remaining); else pendingDelegations.delete(threadId); for (const item of removed) { + cancelPeerApprovalForOwner(approvalOwner(item)); transitionIfActive(workOrders, item.workOrderId, "failed", { error: "target bot was deleted" }); } if (bus.store.botByThread(threadId)) { @@ -353,7 +357,7 @@ async function processOne( item.message, "delegate_bot", sourceThreadId, - item.workOrderId, + approvalOwner(item), ); const afterApproval = workOrders?.get(item.workOrderId ?? ""); if (afterApproval?.state === "cancelled" || afterApproval?.state === "failed") return "done"; @@ -405,6 +409,13 @@ async function processOne( return "done"; } +/** Every queued item owns at most one approval. Durable work orders remain + * the public cancellation key; legacy/non-durable items fall back to their + * stable queue id so cleanup can still release a waiting drain. */ +function approvalOwner(item: PendingDelegationItem): string { + return item.workOrderId ?? item.id; +} + function transitionIfActive( workOrders: WorkOrderStore | undefined, id: string | undefined, diff --git a/server/index.ts b/server/index.ts index 66b0693ea..66e60929c 100644 --- a/server/index.ts +++ b/server/index.ts @@ -124,6 +124,7 @@ import { loadBundledSkills, loadUserSkills, mergeSkills, renderSkillInstructions import { shouldMountLocalComputer } from "./local-routing.ts"; import { WorkOrderCapacityError, WorkOrderInputError, WorkOrderStore, type WorkOrderState } from "./work-orders.ts"; import { TurnScheduler, type TurnLane } from "./turn-scheduler.ts"; +import { scheduleAdmissionSettlementAfterGrace } from "./turn-settlement.ts"; const PORT = Number(process.env.OMB_PORT || process.env.OGB_PORT || 8799); const WEBHOOK_PORT = Number(process.env.OMB_WEBHOOK_PORT || PORT + 1); @@ -322,24 +323,26 @@ function releaseTurnAdmission(threadId: string): void { const TURN_SETTLEMENT_GRACE_MS = 6_000; function settleTurnAfterGrace(threadId: string, botId: string): void { - const timer = setTimeout(() => { - const admission = turnAdmissionByThread.get(threadId); - if (!admission || admission.botId !== botId) return; - const group = store.groupByThread(threadId); - const speaker = groupSpeakers.get(threadId); - if (group && group.busyBotId === botId && speaker?.botId === botId) { - groupSpeakers.delete(threadId); - store.patchGroup(group.id, { busyBotId: null, unread: true }); - } - const bot = store.bot(botId); - if (bot?.busy) { - stopScreenPoller(bot.id); - if (activeVpsThreads.get(bot.id) === threadId) activeVpsThreads.delete(bot.id); - store.setActivity(bot.id, "idle"); - } - releaseTurnAdmission(threadId); - }, TURN_SETTLEMENT_GRACE_MS); - timer.unref?.(); + scheduleAdmissionSettlementAfterGrace({ + botId, + delayMs: TURN_SETTLEMENT_GRACE_MS, + current: () => turnAdmissionByThread.get(threadId), + settle: () => { + const group = store.groupByThread(threadId); + const speaker = groupSpeakers.get(threadId); + if (group && group.busyBotId === botId && speaker?.botId === botId) { + groupSpeakers.delete(threadId); + store.patchGroup(group.id, { busyBotId: null, unread: true }); + } + const bot = store.bot(botId); + if (bot?.busy) { + stopScreenPoller(bot.id); + if (activeVpsThreads.get(bot.id) === threadId) activeVpsThreads.delete(bot.id); + store.setActivity(bot.id, "idle"); + } + releaseTurnAdmission(threadId); + }, + }); } function cancelQueuedWorkOrderRuntime(workOrderId: string): void { diff --git a/server/turn-settlement.test.ts b/server/turn-settlement.test.ts new file mode 100644 index 000000000..c3598fadb --- /dev/null +++ b/server/turn-settlement.test.ts @@ -0,0 +1,49 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { scheduleAdmissionSettlementAfterGrace, type TurnAdmissionOwnership } from "./turn-settlement.ts"; + +afterEach(() => vi.useRealTimers()); + +describe("scheduleAdmissionSettlementAfterGrace", () => { + it("does not settle a newer admission for the same bot and thread", async () => { + vi.useFakeTimers(); + let current: TurnAdmissionOwnership | undefined = { botId: "bot", token: "stalled-turn" }; + const settle = vi.fn(); + + expect(scheduleAdmissionSettlementAfterGrace({ + botId: "bot", + delayMs: 6_000, + current: () => current, + settle, + })).toBe(true); + + current = { botId: "bot", token: "new-turn" }; + await vi.advanceTimersByTimeAsync(6_000); + expect(settle).not.toHaveBeenCalled(); + }); + + it("settles only when the pinned admission still owns the turn", async () => { + vi.useFakeTimers(); + let current: TurnAdmissionOwnership | undefined = { botId: "bot", token: "stalled-turn" }; + const settle = vi.fn(() => { + current = undefined; + }); + + expect(scheduleAdmissionSettlementAfterGrace({ + botId: "bot", + delayMs: 6_000, + current: () => current, + settle, + })).toBe(true); + await vi.advanceTimersByTimeAsync(6_000); + expect(settle).toHaveBeenCalledOnce(); + + expect(scheduleAdmissionSettlementAfterGrace({ + botId: "bot", + delayMs: 6_000, + current: () => current, + settle, + })).toBe(false); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/server/turn-settlement.ts b/server/turn-settlement.ts new file mode 100644 index 000000000..48179a77f --- /dev/null +++ b/server/turn-settlement.ts @@ -0,0 +1,25 @@ +export interface TurnAdmissionOwnership { + botId: string; + token: string; +} + +/** Schedule fallback cleanup for exactly the admission that currently owns + * a turn. A later admission on the same thread and bot is a different owner + * and must never be settled by this timer. */ +export function scheduleAdmissionSettlementAfterGrace(options: { + botId: string; + delayMs: number; + current: () => TurnAdmissionOwnership | undefined; + settle: () => void; +}): boolean { + const admission = options.current(); + if (!admission || admission.botId !== options.botId) return false; + const pinned = { ...admission }; + const timer = setTimeout(() => { + const current = options.current(); + if (!current || current.botId !== pinned.botId || current.token !== pinned.token) return; + options.settle(); + }, options.delayMs); + timer.unref?.(); + return true; +} diff --git a/server/work-orders.test.ts b/server/work-orders.test.ts index 9c77c4278..8ee8d515f 100644 --- a/server/work-orders.test.ts +++ b/server/work-orders.test.ts @@ -4,7 +4,7 @@ import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { WorkOrderStore, type WorkOrderInput } from "./work-orders.ts"; +import { WorkOrderStore, type WorkOrder, type WorkOrderInput } from "./work-orders.ts"; const input: WorkOrderInput = { kind: "consultation", @@ -88,13 +88,32 @@ describe("WorkOrderStore", () => { }); it("settles both source and target orders when a bot is deleted", () => { - const store = new WorkOrderStore({ file: file() }); + const path = file(); + const persistedAtCallback: Array> = []; + let observeSettlement = false; + const store = new WorkOrderStore({ + file: path, + onTransition: () => { + if (!observeSettlement) return; + const orders: WorkOrder[] = JSON.parse(readFileSync(path, "utf8")).orders; + persistedAtCallback.push(Object.fromEntries(orders.map((order) => [order.id, order.state]))); + }, + }); const sourceOrder = store.create(input, "queued"); const targetOrder = store.create({ ...input, sourceBotId: "other", targetBotId: "source" }, "queued"); + observeSettlement = true; const settled = store.settleForDeletedBot("source"); expect(settled.cancelled.map((order) => order.id)).toEqual([sourceOrder.id]); expect(settled.failed.map((order) => order.id)).toEqual([targetOrder.id]); expect(store.get(sourceOrder.id)?.state).toBe("cancelled"); expect(store.get(targetOrder.id)?.state).toBe("failed"); + expect(persistedAtCallback).toHaveLength(2); + for (const snapshot of persistedAtCallback) { + expect(snapshot[sourceOrder.id]).toBe("cancelled"); + expect(snapshot[targetOrder.id]).toBe("failed"); + } + const reopened = new WorkOrderStore({ file: path }); + expect(reopened.get(sourceOrder.id)?.state).toBe("cancelled"); + expect(reopened.get(targetOrder.id)?.state).toBe("failed"); }); }); diff --git a/server/work-orders.ts b/server/work-orders.ts index 4189ba468..fdd3dca48 100644 --- a/server/work-orders.ts +++ b/server/work-orders.ts @@ -181,12 +181,32 @@ export class WorkOrderStore { settleForDeletedBot(botId: string): { cancelled: WorkOrder[]; failed: WorkOrder[] } { const cancelled: WorkOrder[] = []; const failed: WorkOrder[] = []; - for (const order of [...this.orders]) { + const transitions: Array<{ + order: WorkOrder; + from: WorkOrderState; + to: "cancelled" | "failed"; + }> = []; + for (const order of this.orders) { if (TERMINAL.has(order.state)) continue; - if (order.sourceBotId === botId) { - cancelled.push(this.transition(order.id, "cancelled", { error: "source bot was deleted" })); - } else if (order.targetBotId === botId) { - failed.push(this.transition(order.id, "failed", { error: "target bot was deleted" })); + const to = order.sourceBotId === botId + ? "cancelled" + : order.targetBotId === botId + ? "failed" + : null; + if (!to) continue; + const from = order.state; + order.state = to; + order.updatedAt = this.now(); + order.error = to === "cancelled" ? "source bot was deleted" : "target bot was deleted"; + const settled = { ...order }; + transitions.push({ order: settled, from, to }); + if (to === "cancelled") cancelled.push(settled); + else failed.push(settled); + } + if (transitions.length) { + this.save(); + for (const transition of transitions) { + this.onTransition?.({ ...transition.order }, transition.from, transition.to); } } return { cancelled, failed };