From 887f906b8c7ef1da25dfb6eda66d319da96e9cc9 Mon Sep 17 00:00:00 2001 From: ishu86 <112744711+ishu86@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:02:02 +0530 Subject: [PATCH 1/9] Keep Drive turns inside the model window and retry 429s correctly. Turn-aware history trim plus OpenAI compaction stop a long thread from 400ing; rate-limit helpers stop a 429 from being treated as an unsupported-param fallback. Co-authored-by: Cursor --- web/package.json | 2 +- web/web-ui/case-tools.mjs | 42 +++++++--- web/web-ui/serve.mjs | 149 ++++++++++++++++++++++++++++----- web/web-ui/test_rate_retry.mjs | 90 ++++++++++++++++++++ web/web-ui/test_serve.mjs | 134 ++++++++++++++++++----------- 5 files changed, 332 insertions(+), 85 deletions(-) create mode 100644 web/web-ui/test_rate_retry.mjs diff --git a/web/package.json b/web/package.json index 4840ec4..3a13020 100644 --- a/web/package.json +++ b/web/package.json @@ -4,7 +4,7 @@ "type": "module", "scripts": { "start": "node web-ui/serve.mjs", - "test": "node web-ui/test_serve.mjs && node web-ui/test_http.mjs && node web-ui/test_phone.mjs && node web-ui/test_ntfy.mjs && node web-ui/test_telegram.mjs && node web-ui/test_nav.mjs && node web-ui/test_deploy.mjs" + "test": "node web-ui/test_serve.mjs && node web-ui/test_rate_retry.mjs && node web-ui/test_http.mjs && node web-ui/test_phone.mjs && node web-ui/test_ntfy.mjs && node web-ui/test_telegram.mjs && node web-ui/test_nav.mjs && node web-ui/test_deploy.mjs" }, "dependencies": { "@anthropic-ai/sdk": "^0.117.1", diff --git a/web/web-ui/case-tools.mjs b/web/web-ui/case-tools.mjs index 6f09a26..c5a8a15 100644 --- a/web/web-ui/case-tools.mjs +++ b/web/web-ui/case-tools.mjs @@ -372,9 +372,29 @@ function clipJson(v, n = 8000) { return s.length > n ? s.slice(0, n) + '…' : s; } +/** Is this the provider saying "too fast" rather than "bad request"? Callers with + * their own error fallbacks (unsupported summary/effort) must ask first: a 429 + * misread as an unsupported-param error retries with no backoff and degrades the + * request for nothing. */ +export function isRateLimited(err) { + const status = err?.status ?? err?.response?.status; + return status === 429 || status === 529 + || /rate limit|overloaded/i.test(err?.message || ''); +} + +/** Seconds to wait before attempt `a`: the server's own hint ("try again in Xs" + * or retry-after) if it gave one, else exponential. Padded, clamped to 1..60s. */ +export function rateWaitS(err, a) { + const m = /try again in ([\d.]+)s/i.exec(err?.message || ''); + const hdr = Number(err?.headers?.['retry-after'] + ?? err?.response?.headers?.get?.('retry-after')); + const wait = m ? Number(m[1]) : Number.isFinite(hdr) && hdr > 0 ? hdr : 2 ** a; + return Math.min(Math.max(wait + 0.5, 1), 60); +} + /** Retry a provider round on rate limits (429/529), honoring the server's - * suggested wait ("try again in Xs" / retry-after), capped at 60s. History is - * only mutated after a round completes, so replaying a failed round is safe. */ + * suggested wait. History is only mutated after a round completes, so replaying + * a failed round is safe. `signal` cancels the backoff sleep on STOP. */ function abortError(signal) { if (signal?.reason instanceof Error) return signal.reason; const err = new Error(signal?.reason ? String(signal.reason) : 'stopped by user'); @@ -407,16 +427,10 @@ export async function withRateRetry(fn, emit, tries = 5, signal) { try { return await fn(); } catch (err) { if (signal?.aborted) throw err; - const status = err?.status ?? err?.response?.status; - const limited = status === 429 || status === 529 - || /rate limit|overloaded/i.test(err?.message || ''); - if (!limited || a >= tries - 1) throw err; - const m = /try again in ([\d.]+)s/i.exec(err?.message || ''); - const hdr = Number(err?.headers?.['retry-after'] - ?? err?.response?.headers?.get?.('retry-after')); - let wait = m ? Number(m[1]) : Number.isFinite(hdr) && hdr > 0 ? hdr : 2 ** a; - wait = Math.min(Math.max(wait + 0.5, 1), 60); - emit?.({ type: 'think', text: `rate limited — retrying in ${Math.ceil(wait)}s` }); + if (!isRateLimited(err) || a >= tries - 1) throw err; + const wait = rateWaitS(err, a); + // `rate: true` so a non-UI consumer can pick the wait out of the think stream. + emit?.({ type: 'think', rate: true, text: `rate limited — retrying in ${Math.ceil(wait)}s` }); await abortableDelay(wait * 1000, signal); } } @@ -479,7 +493,9 @@ export async function anthropicToolLoop({ result = await withRateRetry(() => round(params), emit, 5, signal); } catch (err) { if (signal?.aborted) throw err; - if (!params.output_config) throw err; + // This fallback is for models that reject output_config — not for a rate + // limit whose retries already ran dry, which would only buy 5 more waits. + if (!params.output_config || isRateLimited(err)) throw err; const rest = { ...params }; delete rest.output_config; result = await withRateRetry(() => round(rest), emit, 5, signal); diff --git a/web/web-ui/serve.mjs b/web/web-ui/serve.mjs index aa87c93..fad2c9a 100644 --- a/web/web-ui/serve.mjs +++ b/web/web-ui/serve.mjs @@ -20,7 +20,7 @@ import path from 'node:path'; import crypto from 'node:crypto'; import { fileURLToPath } from 'node:url'; import OpenAI from 'openai'; -import { CASE_TOOLS, caseCall, caseToolPlan, runCaseTool, streamEventToNdjson, tracesFromOutput, chatAuth, envDriveAuth, resolveChatModel, histToAnthropicMessages, anthropicToolLoop, withRateRetry } from './case-tools.mjs'; +import { CASE_TOOLS, caseCall, caseToolPlan, runCaseTool, streamEventToNdjson, tracesFromOutput, chatAuth, envDriveAuth, resolveChatModel, histToAnthropicMessages, anthropicToolLoop, withRateRetry, isRateLimited } from './case-tools.mjs'; import * as ntfy from './ntfy.mjs'; import { PHONE_THREAD_ID, routePhone } from './phone.mjs'; import * as telegram from './telegram.mjs'; @@ -548,8 +548,16 @@ function actFor(name, args, id) { const ROUNDS = 200; // ROUNDS bounds steps, not spend: history is re-sent every round, so cost is quadratic -// in rounds. This bounds the money — cumulative input tokens for one turn. +// in rounds. This bounds the money — cumulative *billed* input for one turn, i.e. +// eff = (in - cached) + 0.1*cached. Counting raw `in` killed turns at ~260k eff +// (97% cache hit → 7.6x inflated); see the `drive turn` log line. const TURN_TOKEN_BUDGET = Number(process.env.CASE_TURN_TOKENS || 2_000_000); +const BUDGET_WARN = 'Turn budget nearly spent — a few tool steps remain. Append your progress and the exact next step to a file under /home/agent/reports now, then stop and say where you stopped.'; +const isBudgetWarn = (it) => Array.isArray(it?.content) && it.content.some((c) => c?.text === BUDGET_WARN); +// Window guard. OpenAI compacts server-side once the rendered context passes this +// (opaque `compaction` item we carry forward); `truncation:'auto'` is the floor if a +// model lacks compaction. 200k fits every window in the list (272k–1M). 0 = off. +const COMPACT_AT = Number(process.env.CASE_COMPACT_AT ?? 200_000); // Threads: the sidebar's unit of navigation, each with its own conversation memory. // The Responses API runs stateless here (store:false), so the item list IS the // memory. agent stays '' until the run first needs hands (a tool call executes) — @@ -649,24 +657,78 @@ function threadsRoute(req, res, url) { return json(res, 405, { error: 'method' }); } function finishTurn(hist, thread) { - hist.items = histCloseOpenCalls(hist.items); + // The budget warning is turn-scoped: left in, the next turn opens with + // "nearly spent" and wraps up on its first round. + hist.items = histCloseOpenCalls(hist.items).filter((it) => !isBudgetWarn(it)); histTrim(hist); thread.updated = Date.now(); saveThreads(); } const HIST_MAX = 240_000; +// A turn opener is the prompt the user typed: string content. Screenshots, steers +// and "screen unchanged" notes are role:user too but carry content arrays — they +// are NOT boundaries (cutting at one deleted the task and kept the tool spam). +const isTurnStart = (it) => it?.role === 'user' && typeof it.content === 'string'; +const turnStarts = (items) => items.map((it, i) => (isTurnStart(it) ? i : -1)).filter((i) => i >= 0); +const PROMPT_KEEP = 20_000; // the user's own words: elide only a genuinely huge paste +const TAIL_KEEP = 40_000; // recent observations carried into the next turn +/** The last `budget` chars of a turn, whole call/output pairs only — a tail that + * opens on an orphaned function_call_output 400s every later request. */ +function turnTail(items, budget) { + const out = []; + let n = 0; + for (let i = items.length - 1; i >= 0; i--) { + n += JSON.stringify(items[i]).length; + if (n > budget && out.length) break; + out.unshift(items[i]); + } + const calls = new Set(out.filter((it) => it.type === 'function_call').map((it) => it.call_id)); + while (out.length && out[0].type === 'function_call_output' && !calls.has(out[0].call_id)) out.shift(); + return out; +} +/** Over budget, turns collapse to [prompt, last reply]: the task and the model's own + * "stopped at X" survive forever, tool observations die. Only if every turn is + * already collapsed and it is still over does the oldest go. + * The newest turn collapses too — every caller runs this after the turn has ended, + * so nothing in flight is cut, and a 159-round turn left whole is what the *next* + * turn pays to re-send on every round. A short turn is nowhere near `max`, so a + * follow-up like "click the blue one" still has its snapshot to work from. */ export function histTrim(h, max = HIST_MAX) { - // Turn boundaries are derived, not tracked: the only user-role items are the ones - // that open a turn, so they survive any filtering of the array. - let starts = h.items.map((it, i) => (it.role === 'user' ? i : -1)).filter((i) => i >= 0); - let size = JSON.stringify(h.items).length; - while (starts.length > 1 && size > max) { - const cut = starts[1]; - size -= JSON.stringify(h.items.slice(0, cut)).length; - h.items.splice(0, cut); - starts = starts.slice(1).map((i) => i - cut); + const over = () => JSON.stringify(h.items).length > max; + let starts = turnStarts(h.items); + for (let k = 0; k < starts.length && over(); k++) { + const [a, b] = [starts[k], starts[k + 1] ?? h.items.length]; + const turn = h.items.slice(a, b); + const body = turn.slice(1); + let keep; + if (k === starts.length - 1) { + // The newest turn is the one "continue" resumes into, so it keeps a bounded + // tail of real observations, not just its closing line. Older turns are + // already answered — their reply is the summary. + keep = turnTail(body, TAIL_KEEP); + } else { + const last = [...body].reverse().find((it) => it.type === 'message' && it.role === 'assistant'); + keep = last ? [{ ...last, content: last.content.map((c) => (c.type === 'output_text' ? { ...c, text: clip(c.text, 2000) } : c)) }] : []; + } + // A compaction item is the only copy of everything summarized away for it. + const comp = body.filter((it) => it.type === 'compaction' && !keep.includes(it)); + const kept = [{ ...turn[0], content: clip(turn[0].content, PROMPT_KEEP) }, ...comp, ...keep]; + if (kept.length === turn.length) continue; + h.items.splice(a, b - a, ...kept); + starts = turnStarts(h.items); } + while (starts.length > 1 && over()) { + h.items.splice(0, starts[1]); + starts = turnStarts(h.items); + } +} +/** A server-side compaction item carries everything before it. Drop that — except + * the turn openers, so the task is never only inside an opaque blob. */ +export function histApplyCompaction(items) { + const ci = items.findLastIndex((it) => it?.type === 'compaction'); + if (ci < 0) return items; + return [...items.slice(0, ci).filter(isTurnStart), ...items.slice(ci)]; } /** Drop stale reasoning and close any function_call that has no output. * OpenAI 400s "No tool output found for function call …" otherwise. */ @@ -1032,7 +1094,9 @@ export async function runTurn({ const client = new OpenAI({ apiKey: auth.key }); let text = ''; const spend = { in: 0, cached: 0, out: 0 }; + const effSoFar = () => Math.round((spend.in - spend.cached) + 0.1 * spend.cached); let summary = 'detailed'; + let compact = COMPACT_AT > 0; const round = async () => { // The SDK leaves its abort listener on the signal after the round ends; over a // 200-round turn that is 200 dead listeners on one signal. A per-round @@ -1049,15 +1113,31 @@ export async function runTurn({ // exactly the shape the prefix cache wants. A stable key is required for // reliable matching; cached input bills at 0.1x. prompt_cache_key: thread.id, + truncation: 'auto', + ...(compact ? { context_management: [{ type: 'compaction', compact_threshold: COMPACT_AT }] } : {}), }; + // Two params can be rejected independently: context_management, and a + // 'detailed' reasoning summary. Drop whichever the error names and go round + // again. Each branch flips a one-way flag, so this runs at most three times. let stream; - try { stream = await client.responses.create(params, { signal: rc.signal }); } - catch (err) { - // Only the summary-unsupported fallback retries; an abort must not. - if (summary !== 'auto' && !gone.signal.aborted) { - summary = 'auto'; - stream = await client.responses.create({ ...params, reasoning: { effort, summary } }, { signal: rc.signal }); - } else throw err; + for (;;) { + try { stream = await client.responses.create(params, { signal: rc.signal }); break; } + catch (err) { + // Only param fallbacks retry here. An abort must not — and neither may a + // rate limit: `summary`/`compact` outlive the round, so treating a 429 as + // "unsupported" retries with no backoff AND thins reasoning / drops the + // window guard for every later round. Let withRateRetry have it. + if (gone.signal.aborted) throw err; + if (isRateLimited(err)) throw err; + if (compact && /context_management|compaction/i.test(err?.message || '')) { + console.log(`drive turn ${thread.id}: compaction rejected — ${err?.message || 'no message'}`); + compact = false; + delete params.context_management; + } else if (summary !== 'auto') { + summary = 'auto'; + params.reasoning = { effort, summary }; + } else throw err; + } } let response = null; let thinkDelta = false; @@ -1102,9 +1182,16 @@ export async function runTurn({ let finished = false; const shots = new Set(); // screenshot hashes already in this turn's history const snaps = { last: '' }; // hash of the most recent snapshot's element list - const overBudget = () => spend.in > TURN_TOKEN_BUDGET; + const overBudget = () => effSoFar() > TURN_TOKEN_BUDGET; + let warned = ''; + let compactions = 0; let i = 0; for (; i < ROUNDS && !finished && !stopped() && !overBudget(); i++) { + if (!warned && (effSoFar() > 0.8 * TURN_TOKEN_BUDGET || i >= 0.8 * ROUNDS)) { + warned = effSoFar() > 0.8 * TURN_TOKEN_BUDGET ? 'budget' : 'rounds'; + pushSteerItems(hist.items, [BUDGET_WARN]); + emit({ type: 'think', text: `[80% of the turn ${warned} — told the model to wrap up]` }); + } const nudges = takeSteers(thread.id); if (nudges.length) { for (const n of nudges) { @@ -1124,6 +1211,10 @@ export async function runTurn({ // only in the stream: reloads show bare prompts and the model never // sees what it already said. hist.items.push(...response.output); + if (compact && response.output.some((it) => it.type === 'compaction')) { + compactions++; + hist.items = histApplyCompaction(hist.items); + } if (text && !textDelta) emit({ type: 'text', text }); break; } @@ -1145,6 +1236,17 @@ export async function runTurn({ if (image_b64) images.push(image_b64); } hist.items = histCloseOpenCalls(hist.items, { keepReasoning: true }); + if (compact && response.output.some((it) => it.type === 'compaction')) { + compactions++; + hist.items = histApplyCompaction(hist.items); + // Both dedup caches point at items compaction just deleted. Left set, the + // next screenshot comes back "identical to an earlier one this turn" with + // no earlier one in the input, and snapshotElide promises refs are "still + // valid" from a snapshot the model can no longer see. + shots.clear(); + snaps.last = ''; + emit({ type: 'think', text: '[context compacted server-side — carrying the summary forward]' }); + } for (const b64 of images) pushShot(hist.items, shots, b64); } if (!finished && !stopped()) { @@ -1152,17 +1254,18 @@ export async function runTurn({ // "it just stopped" — and the carried history makes "continue" actually resume. text = (text ? text + '\n\n' : '') + (overBudget() - ? `**Out of budget.** Stopped after ${spend.in.toLocaleString('en-US')} input tokens with the task unfinished. Say **continue** and I pick up from here.` + ? `**Out of budget.** Stopped after ${effSoFar().toLocaleString('en-US')} billed input tokens with the task unfinished. Say **continue** and I pick up from here.` : `**Out of steps.** Stopped after ${ROUNDS} tool calls with the task unfinished. Say **continue** and I pick up from here.`); emit({ type: 'text', text }); } // Reasoning items are only valid inside the turn that produced them; carrying // them forward bloats the payload and some models reject stale ones. finishTurn(hist, thread); - const eff = Math.round((spend.in - spend.cached) + 0.1 * spend.cached); + const eff = effSoFar(); console.log(`drive turn ${thread.id}: in=${spend.in} cached=${spend.cached}` + ` (${spend.in ? Math.round((100 * spend.cached) / spend.in) : 0}%)` - + ` eff=${eff} out=${spend.out} rounds=${i}`); + + ` eff=${eff} out=${spend.out} rounds=${i} warn=${warned || 'none'}` + + ` hist=${JSON.stringify(hist.items).length} compactions=${compactions}`); spend.eff = eff; emit({ type: 'done', text, computer_id: id, thread_id: thread.id, spend, rounds: i }); return { text, computerId: id, threadId: thread.id }; diff --git a/web/web-ui/test_rate_retry.mjs b/web/web-ui/test_rate_retry.mjs new file mode 100644 index 0000000..2bb4730 --- /dev/null +++ b/web/web-ui/test_rate_retry.mjs @@ -0,0 +1,90 @@ +#!/usr/bin/env node +// SPDX-License-Identifier: MIT +// Pure-unit checks for the rate-limit retry. Run: node web/web-ui/test_rate_retry.mjs +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import { isRateLimited, rateWaitS, withRateRetry } from './case-tools.mjs'; + +// --- what counts as "too fast" ------------------------------------------- +assert.ok(isRateLimited({ status: 429 })); +assert.ok(isRateLimited({ status: 529 })); +assert.ok(isRateLimited({ response: { status: 429 } })); +assert.ok(isRateLimited({ message: 'Rate limit reached for gpt-5.6' })); +assert.ok(isRateLimited({ message: 'Overloaded' })); +assert.ok(!isRateLimited({ status: 400, message: "unsupported value: 'detailed'" })); +assert.ok(!isRateLimited({ status: 401 })); +assert.ok(!isRateLimited(undefined)); + +// --- how long to wait ---------------------------------------------------- +// the server's own hint wins over the exponential guess +assert.equal(rateWaitS({ message: 'try again in 12s' }, 0), 12.5); +assert.equal(rateWaitS({ headers: { 'retry-after': '9' } }, 0), 9.5); +assert.equal(rateWaitS({ response: { headers: new Map([['retry-after', '4']]) } }, 0), 4.5); +// no hint: exponential in the attempt number +assert.equal(rateWaitS({ status: 429 }, 0), 1.5); +assert.equal(rateWaitS({ status: 429 }, 3), 8.5); +// clamped both ends — never busy-loop, never park for an hour +assert.equal(rateWaitS({ message: 'try again in 0.01s' }, 0), 1); +assert.equal(rateWaitS({ message: 'try again in 3600s' }, 0), 60); +assert.equal(rateWaitS({ status: 429 }, 20), 60); + +// --- the loop ------------------------------------------------------------ +// a non-rate-limit error is not retried: one call, straight out +let calls = 0; +await assert.rejects( + () => withRateRetry(async () => { calls++; throw Object.assign(new Error('bad request'), { status: 400 }); }), + /bad request/); +assert.equal(calls, 1, 'a 400 must not be retried'); + +// a rate limit is retried and the flow continues where it left off +calls = 0; +const seen = []; +const limited = Object.assign(new Error('rate limit — try again in 0.01s'), { status: 429 }); +const out = await withRateRetry(async () => { + calls++; + if (calls < 3) throw limited; + return 'round done'; +}, (ev) => seen.push(ev)); +assert.equal(out, 'round done'); +assert.equal(calls, 3); +assert.equal(seen.length, 2, 'one notice per wait'); +assert.ok(seen.every((e) => e.type === 'think' && e.rate === true), 'notices carry rate:true so headless consumers can log them'); +assert.match(seen[0].text, /rate limited — retrying in 1s/); + +// retries do run dry — the caller has to see the error, not hang forever +calls = 0; +await assert.rejects(() => withRateRetry(async () => { + calls++; + throw limited; +}, null, 3), /rate limit/); +assert.equal(calls, 3, 'tries is a hard cap'); + +// STOP / disconnect must cancel the backoff sleep, not wait it out +{ + const ctl = new AbortController(); + let n = 0; + const started = Date.now(); + await assert.rejects( + withRateRetry(async () => { + n += 1; + const err = new Error('rate limited'); + err.status = 429; + throw err; + }, () => ctl.abort(), 5, ctl.signal), + (err) => err?.name === 'AbortError', + ); + assert.equal(n, 1, 'disconnect stops retries before another provider request'); + assert.ok(Date.now() - started < 500, 'disconnect interrupts the backoff sleep'); +} + +// --- the two error fallbacks must not eat a 429 -------------------------- +// Both retry a round with a param removed when the model rejects it. Neither may +// fire on a rate limit: no backoff, and serve.mjs's `summary` is sticky for the turn. +const serve = fs.readFileSync(new URL('./serve.mjs', import.meta.url), 'utf8'); +assert.match(serve, /if \(gone\.signal\.aborted\) throw err;/); +assert.match(serve, /if \(isRateLimited\(err\)\) throw err;/); +assert.match(serve, /else if \(summary !== 'auto'\)/); +const tools = fs.readFileSync(new URL('./case-tools.mjs', import.meta.url), 'utf8'); +assert.match(tools, /!params\.output_config \|\| isRateLimited\(err\)/); + +console.log('rate retry ok'); diff --git a/web/web-ui/test_serve.mjs b/web/web-ui/test_serve.mjs index 35a95f8..a72c027 100644 --- a/web/web-ui/test_serve.mjs +++ b/web/web-ui/test_serve.mjs @@ -7,11 +7,11 @@ import fs from 'node:fs'; import os from 'node:os'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; -import { shq, pathOk, parseErr, parseFind, mimeFor, histTrim, histCloseOpenCalls, normHost, threadTurns, parseCaseUrl, liveCid, liveDestPath, livePathHasDotDot, tokenMatches, liveHeaders, hostOf, browserOk, extraPlan, isLocalMode, pageFile, clip, snapshotElide, stashShot, pushShot, hydrateShots, migrateShots, stashAttach, resolveAttach, hydrateAttaches, attachKind, ATTACH_MAX, sseEvents } from './serve.mjs'; +import { shq, pathOk, parseErr, parseFind, mimeFor, histTrim, histApplyCompaction, histCloseOpenCalls, normHost, threadTurns, parseCaseUrl, liveCid, liveDestPath, livePathHasDotDot, tokenMatches, liveHeaders, hostOf, browserOk, extraPlan, isLocalMode, pageFile, clip, snapshotElide, stashShot, pushShot, hydrateShots, migrateShots, stashAttach, resolveAttach, hydrateAttaches, attachKind, ATTACH_MAX, sseEvents } from './serve.mjs'; import { CASE_TOOLS, chatAuth, resolveChatModel, openaiToolsToAnthropic, newAnthropicStreamCtx, anthropicEventToNdjson, tracesFromAnthropicMessage, - histToAnthropicMessages, anthropicThinkingFor, caseToolPlan, withRateRetry, + histToAnthropicMessages, anthropicThinkingFor, caseToolPlan, } from './case-tools.mjs'; const html = fs.readFileSync(fileURLToPath(new URL('./index.html', import.meta.url)), 'utf8'); @@ -95,28 +95,91 @@ assert.equal(parseErr(Buffer.from('502'), 'read failed'), 'read fai assert.ok(!/fonts\.(googleapis|gstatic)/.test(html), 'no font CDN on the page'); } -// histTrim: conversation memory drops WHOLE turns, never splitting a function_call -// from its output (an orphan of either kind 400s every later request). +// histTrim: over budget, old turns collapse to [prompt, last reply] — the task and +// the model's own "stopped at X" survive, tool observations die. Never an orphan +// function_call / output (either kind 400s every later request). +const reply = (n) => ({ type: 'message', role: 'assistant', content: [{ type: 'output_text', text: `did ${n}` }] }); const turn = (n, pad) => [ { role: 'user', content: `ask ${n}${pad}` }, { type: 'function_call', call_id: `c${n}`, name: 'computer_snapshot', arguments: '{}' }, - { type: 'function_call_output', call_id: `c${n}`, output: 'ok' }, + { type: 'function_call_output', call_id: `c${n}`, output: 'ok' + pad }, + reply(n), ]; const h = { items: [...turn(1, 'x'.repeat(400)), ...turn(2, ''), ...turn(3, '')] }; -histTrim(h, 500); // 938 chars over three turns; dropping the fat first turn leaves 359 -assert.deepEqual(h.items.map((i) => i.call_id || i.content), ['ask 2', 'c2', 'c2', 'ask 3', 'c3', 'c3']); +histTrim(h, 1200); // 1602 chars; collapsing turn 1 leaves 1057 +assert.deepEqual(h.items.map((i) => i.call_id || (Array.isArray(i.content) ? i.content[0].text : i.content)), + ['ask 1' + 'x'.repeat(400), 'did 1', 'ask 2', 'c2', 'c2', 'did 2', 'ask 3', 'c3', 'c3', 'did 3']); for (const it of h.items.filter((i) => i.type === 'function_call')) { assert.ok(h.items.some((o) => o.type === 'function_call_output' && o.call_id === it.call_id), `call ${it.call_id} lost its output`); } -// the newest turn is never trimmed away, however far over budget it is -const solo = { items: turn(9, 'y'.repeat(5000)) }; -histTrim(solo, 10); -assert.equal(solo.items.length, 3); -// under budget: untouched +// screenshots, steers and notes are role:user too — they are NOT turn boundaries. +// The cut must never land inside a turn and take the task with it. +const tmpShots = fs.mkdtempSync(path.join(os.tmpdir(), 'shots-')); +const shotTurn = { items: [ + { role: 'user', content: 'ORIGINAL TASK' }, + { type: 'function_call', call_id: 'c1', name: 'computer_navigate', arguments: '{}' }, + { type: 'function_call_output', call_id: 'c1', output: 'x'.repeat(3000) }, + stashShot('aGVsbG8=', tmpShots), + { role: 'user', content: [{ type: 'input_text', text: 'steer: faster' }] }, + { type: 'function_call', call_id: 'c2', name: 'computer_click', arguments: '{}' }, + { type: 'function_call_output', call_id: 'c2', output: 'ok' }, + reply('shot'), + { role: 'user', content: 'continue' }, + { type: 'function_call', call_id: 'c3', name: 'computer_click', arguments: '{}' }, + { type: 'function_call_output', call_id: 'c3', output: 'ok' }, +] }; +histTrim(shotTurn, 500); +assert.deepEqual(shotTurn.items.map((i) => i.call_id || (Array.isArray(i.content) ? i.content[0].text : i.content)), + ['ORIGINAL TASK', 'did shot', 'continue', 'c3', 'c3']); +fs.rmSync(tmpShots, { recursive: true, force: true }); +// still over after every turn is collapsed: drop the oldest collapsed ones. The +// newest turn keeps its tail (small enough to fit TAIL_KEEP whole). +const many = { items: [...turn(1, 'x'.repeat(300)), ...turn(2, 'y'.repeat(300)), ...turn(3, '')] }; +histTrim(many, 800); +assert.deepEqual(many.items.map((i) => i.call_id || (Array.isArray(i.content) ? i.content[0].text : i.content)), + ['ask 2' + 'y'.repeat(300), 'did 2', 'ask 3', 'c3', 'c3', 'did 3']); +// A lone giant turn collapses too: histTrim only runs after a turn has ended, and +// left whole it is what every round of the NEXT turn re-sends. It keeps its prompt +// and a bounded tail, so "continue" resumes with real observations rather than blind. +const giant = { items: [{ role: 'user', content: 'THE TASK' }] }; +for (let n = 0; n < 60; n++) { + giant.items.push({ type: 'function_call', call_id: `g${n}`, name: 'computer_eval', arguments: '{}' }); + giant.items.push({ type: 'function_call_output', call_id: `g${n}`, output: 'z'.repeat(4000) }); +} +giant.items.push(reply('giant')); +const wasBig = JSON.stringify(giant.items).length; +histTrim(giant); +const nowSmall = JSON.stringify(giant.items).length; +assert.ok(wasBig > 240_000 && nowSmall < 60_000, `collapsed ${wasBig} -> ${nowSmall}`); +assert.equal(giant.items[0].content, 'THE TASK', 'the task survives its own turn collapsing'); +assert.ok(giant.items.length > 3, 'a tail of real observations survives, not just the prompt'); +const tailCalls = new Set(giant.items.filter((i) => i.type === 'function_call').map((i) => i.call_id)); +for (const it of giant.items.filter((i) => i.type === 'function_call_output')) { + assert.ok(tailCalls.has(it.call_id), `orphan output ${it.call_id} would 400 every later request`); +} +histTrim(giant, 10); +assert.equal(giant.items[0].content, 'THE TASK'); const small = { items: turn(1, '') }; histTrim(small, 100000); -assert.equal(small.items.length, 3); +assert.equal(small.items.length, 4); + +// histApplyCompaction: after a server-side compaction item, everything before it +// goes except the turn openers — the task survives, the compaction carries the rest. +const comp = histApplyCompaction([ + { role: 'user', content: 'task A' }, + { type: 'function_call', call_id: 'a1', name: 'computer_snapshot', arguments: '{}' }, + { type: 'function_call_output', call_id: 'a1', output: 'big' }, + { role: 'user', content: 'continue' }, + { type: 'function_call', call_id: 'b1', name: 'computer_snapshot', arguments: '{}' }, + { type: 'function_call_output', call_id: 'b1', output: 'big' }, + { role: 'user', content: [{ type: 'input_text', text: 'steer' }] }, + { type: 'compaction', id: 'cmp_1', encrypted_content: 'opaque' }, + { type: 'function_call', call_id: 'b2', name: 'computer_click', arguments: '{}' }, + { type: 'function_call_output', call_id: 'b2', output: 'ok' }, +]); +assert.deepEqual(comp.map((i) => i.call_id || i.type || i.content), ['task A', 'continue', 'compaction', 'b2', 'b2']); +assert.deepEqual(histApplyCompaction(small.items), small.items, 'no compaction item: untouched'); const closed = histCloseOpenCalls([ { type: 'reasoning', summary: [] }, @@ -374,7 +437,13 @@ assert.equal(pageFile('/deploy.html'), '/deploy.html'); assert.match(chatFn, /res\.on\('close', \(\) => \{ clientGone\(\); gone\.abort\(\); \}\)/); assert.match(loopFn, /responses\.create\(params, \{ signal: rc\.signal \}\)/); assert.ok(!/responses\.create\(params\)/.test(loopFn), 'every round is abortable'); - assert.match(loopFn, /summary !== 'auto' && !gone\.signal\.aborted/, 'an abort never retries as a summary fallback'); + assert.match(loopFn, /if \(gone\.signal\.aborted\) throw err;/, 'an abort never retries as a param fallback'); + assert.match(loopFn, /if \(isRateLimited\(err\)\) throw err;/); + assert.match(loopFn, /else if \(summary !== 'auto'\)/); + assert.ok(loopFn.lastIndexOf('histApplyCompaction') > loopFn.indexOf("histCloseOpenCalls(hist.items, { keepReasoning: true })"), + 'the tool round applies compaction only after closing its open calls'); + assert.equal([...loopFn.matchAll(/histApplyCompaction/g)].length, 2, 'both the answering and tool rounds compact'); + assert.match(serveSrc, /\.filter\(\(it\) => !isBudgetWarn\(it\)\)/, 'the budget warning does not outlive its turn'); assert.match(loopFn, /gone\.signal\.removeEventListener\('abort', relay\)/, 'round listener is unlinked'); assert.match(serveSrc, /takeSteers\(thread\.id\)/, 'steer inbox drained in the loop'); assert.match(serveSrc, /type: 'steer'/, 'steer emits to the stream'); @@ -396,8 +465,9 @@ assert.equal(pageFile('/deploy.html'), '/deploy.html'); /cache_control: \{ type: 'ephemeral' \}/, 'Anthropic path requests prompt cache'); assert.match(loopFn, /hydrateShots\(hydrateAttaches\(hist\.items\)\)/); assert.match(chatFn, /attachment not found/, 'a missing file is an error, not a silent drop'); - assert.ok(!/truncation:\s*['"]auto['"]/.test(loopFn), 'no truncation:auto'); - assert.ok(!/compactHistory|SUMMARIZE_PROMPT|CASE_COMPACT_AT/.test(serveSrc), 'no compaction'); + assert.match(loopFn, /truncation: 'auto'/); + assert.match(serveSrc, /CASE_COMPACT_AT/); + assert.ok(!/compactHistory|SUMMARIZE_PROMPT/.test(serveSrc), 'no client-side summarizer'); assert.match(serveSrc, /try \{ computerId = await cid\(\); \}/); assert.ok(!/drive ntfy chat on \$\{cfg\.url\}\/\$\{cfg\.topic\}/.test(serveSrc)); } @@ -507,38 +577,6 @@ assert.equal(pageFile('/deploy.html'), '/deploy.html'); fs.rmSync(dir, { recursive: true, force: true }); } -{ - let n = 0; - const out = await withRateRetry(async () => { - n += 1; - if (n < 3) { - const err = new Error('rate limit: try again in 0s'); - err.status = 429; - throw err; - } - return 'ok'; - }, () => {}, 5); - assert.equal(out, 'ok'); - assert.equal(n, 3); -} - -{ - const ctl = new AbortController(); - let n = 0; - const started = Date.now(); - await assert.rejects( - withRateRetry(async () => { - n += 1; - const err = new Error('rate limited'); - err.status = 429; - throw err; - }, () => ctl.abort(), 5, ctl.signal), - (err) => err?.name === 'AbortError', - ); - assert.equal(n, 1, 'disconnect stops retries before another provider request'); - assert.ok(Date.now() - started < 500, 'disconnect interrupts the backoff sleep'); -} - { const run = (cmd) => { const plan = caseToolPlan('computer_exec', { command: cmd }, 'c_1'); From b64dbc99ce9685372e1619abe14c5d7526f90d7d Mon Sep 17 00:00:00 2001 From: ishu86 <112744711+ishu86@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:04:36 +0530 Subject: [PATCH 2/9] Let daily schedules keep a timezone and skip when the box is short on RAM. Daily HH:MM can be an IANA zone instead of box-local time. A tight RAM budget is a skip, not a failure. Compose can point CASE_BRAIN_URL at Drive instead of requiring a second brain binary. Co-authored-by: Cursor --- control-plane/config.py | 3 + control-plane/scheduler.py | 72 +++++++++++++--- control-plane/store.py | 10 ++- mcp/case_mcp.py | 10 ++- tests/test_scheduler.py | 163 +++++++++++++++++++++++++++++++++++++ 5 files changed, 239 insertions(+), 19 deletions(-) diff --git a/control-plane/config.py b/control-plane/config.py index 7c4eefa..51d9797 100644 --- a/control-plane/config.py +++ b/control-plane/config.py @@ -67,6 +67,9 @@ def _desk_resolution(): # contain {prompt}. Unset = stock claude. The template carries no --allowedTools clamp: # a template harness runs with the box's full privileges, only use one you trust. BRAIN_CMD = os.environ.get("CASE_BRAIN_CMD", "") +# HTTP brain (compose default): POST {computer_id, prompt} to Drive. Used only +# when BRAIN_CMD is empty. Precedence: CASE_BRAIN_CMD > CASE_BRAIN_URL > claude. +BRAIN_URL = os.environ.get("CASE_BRAIN_URL", "") MCP_CONFIG = os.environ.get( "CASE_MCP_CONFIG", os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "case-mcp.json")) diff --git a/control-plane/scheduler.py b/control-plane/scheduler.py index 01a0a24..184975a 100644 --- a/control-plane/scheduler.py +++ b/control-plane/scheduler.py @@ -16,8 +16,11 @@ import subprocess import threading from datetime import datetime, timedelta, timezone +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError -from config import (BRAIN_BIN, BRAIN_CMD, BRAIN_TIMEOUT, MAX_RUNNING, MCP_CONFIG, +import requests + +from config import (BRAIN_BIN, BRAIN_CMD, BRAIN_TIMEOUT, BRAIN_URL, MAX_RUNNING, MCP_CONFIG, RUNS_DIR, log) from deskclient import desk_json, screenshot_bytes from errors import ApiError @@ -31,20 +34,28 @@ _LOCK = threading.Lock() # guards the check-then-add on SCHED_RUNNING (sweeper vs run-now) -def compute_next(kind, spec, jitter_s): - """Next fire time as UTC ISO. Lexicographic order == chronological (zero-padded, Z).""" +def compute_next(kind, spec, jitter_s, tz=None): + """Next fire time as UTC ISO. Lexicographic order == chronological (zero-padded, Z). + Daily HH:MM is wall clock in tz (IANA). Empty tz = box local (MCP / old callers).""" j = random.randint(0, int(jitter_s or 0)) if kind == "interval": if int(spec) < 60: raise ApiError(400, "bad_request", "interval must be at least 60 seconds") nxt = datetime.now(timezone.utc) + timedelta(seconds=int(spec) + j) elif kind == "daily": - local = datetime.now() + name = str(tz).strip() if tz else "" + if name: + try: + local = datetime.now(ZoneInfo(name)) + except (ZoneInfoNotFoundError, ValueError): + raise ApiError(400, "bad_tz", f"unknown timezone {name}") + else: + local = datetime.now().astimezone() hh, mm = (int(x) for x in str(spec).split(":")) t = local.replace(hour=hh, minute=mm, second=0, microsecond=0) if t <= local: t += timedelta(days=1) - nxt = (t + timedelta(seconds=j)).astimezone(timezone.utc) # naive→aware picks that date's offset + nxt = (t + timedelta(seconds=j)).astimezone(timezone.utc) else: raise ApiError(400, "bad_kind", "kind must be 'interval' or 'daily'") return nxt.strftime("%Y-%m-%dT%H:%M:%SZ") @@ -67,8 +78,37 @@ def brain_argv(full_prompt): "--allowedTools", "mcp__case__*"] +def _run_brain_url(cid, prompt): + """POST {computer_id, prompt} to Drive. Returns the same (code, summary) as the argv path.""" + token = (os.environ.get("CASE_TOKEN") or "").strip() + headers = {"Authorization": f"Bearer {token}"} if token else {} + try: + r = requests.post(BRAIN_URL, json={"computer_id": cid, "prompt": prompt}, + headers=headers, timeout=BRAIN_TIMEOUT) + except requests.Timeout: + return -1, "brain run timed out" + except requests.RequestException: + return 127, f"schedule brain unreachable at {BRAIN_URL}" + try: + body = r.json() if r.content else {} + except ValueError: + body = {} + if not isinstance(body, dict): + body = {} + if r.status_code == 503: + return 2, str(body.get("error") or "schedule brain unavailable") + if body.get("ok"): + return (0 if body.get("finished") else 3), str(body.get("text") or "") + if body.get("error"): + return 1, str(body["error"]) + return 1, (r.text or f"HTTP {r.status_code}")[-800:] + + def run_brain(cid, prompt): - """Invoke the headless brain against this computer via Case MCP. Returns (code, summary).""" + """Invoke the headless brain against this computer. Returns (code, summary). + Precedence: CASE_BRAIN_CMD > CASE_BRAIN_URL > stock claude on PATH.""" + if not BRAIN_CMD and BRAIN_URL: + return _run_brain_url(cid, prompt) try: argv = brain_argv(f"On Case computer {cid}: {prompt}") except ValueError as e: @@ -134,8 +174,10 @@ def run_schedule(sid): s = store.get_schedule(sid, enabled_only=True) if not s: return + # sqlite3.Row has no dict.get — index like every other column. + tz = s["tz"] if "tz" in s.keys() else None # Reschedule FIRST so a hung/crashed run never wedges the slot. - store.set_schedule_next(sid, compute_next(s["kind"], s["spec"], s["jitter_s"])) + store.set_schedule_next(sid, compute_next(s["kind"], s["spec"], s["jitter_s"], tz)) cid, rid, started = s["computer_id"], new_id("run"), now() code, summary, status, artifact = -1, "", "fail", None # Only the run that woke an asleep box may put it back, never borrow a live session @@ -154,6 +196,9 @@ def run_schedule(sid): if e.code == "too_many_running": status = "skipped" summary = f"another computer is running (max {MAX_RUNNING} on this box)" + elif e.code == "not_enough_ram": + status = "skipped" + summary = f"not enough free RAM on this box ({e.message})" else: summary = f"{e.code}: {e.message}" log.exception("schedule %s run failed", sid) @@ -182,9 +227,11 @@ def fire_due_schedules(spawn): def schedule_json(row): - return {k: row[k] for k in ("id", "computer_id", "name", "prompt", "kind", "spec", - "jitter_s", "enabled", "next_run_at", "last_run_at", - "last_status", "created_at")} + out = {k: row[k] for k in ("id", "computer_id", "name", "prompt", "kind", "spec", + "jitter_s", "enabled", "next_run_at", "last_run_at", + "last_status", "created_at")} + out["tz"] = row["tz"] if "tz" in row.keys() else None + return out def create_schedule(cid, body): @@ -192,16 +239,17 @@ def create_schedule(cid, body): if "prompt" not in body or "spec" not in body: raise ApiError(400, "bad_request", "prompt and spec are required") kind = body.get("kind", "daily") + tz = str(body.get("tz") or "").strip() or None try: jitter = int(body.get("jitter_s", 300)) - nxt = compute_next(kind, body["spec"], jitter) # also validates kind/spec + nxt = compute_next(kind, body["spec"], jitter, tz) # also validates kind/spec/tz except (TypeError, ValueError): raise ApiError(400, "bad_request", "spec must be seconds (interval) or HH:MM (daily); " "jitter_s must be an integer") sid = new_id("sch") store.insert_schedule(sid, cid, str(body.get("name") or sid), body["prompt"], - kind, str(body["spec"]), jitter, nxt) + kind, str(body["spec"]), jitter, nxt, tz) return schedule_json(store.get_schedule(sid)) diff --git a/control-plane/store.py b/control-plane/store.py index 302beec..c8e4525 100644 --- a/control-plane/store.py +++ b/control-plane/store.py @@ -66,7 +66,8 @@ CREATE TABLE IF NOT EXISTS schedules ( id TEXT PRIMARY KEY, computer_id TEXT, name TEXT, prompt TEXT, kind TEXT, spec TEXT, jitter_s INTEGER, enabled INTEGER, - next_run_at TEXT, last_run_at TEXT, last_status TEXT, created_at TEXT + next_run_at TEXT, last_run_at TEXT, last_status TEXT, created_at TEXT, + tz TEXT ); CREATE TABLE IF NOT EXISTS runs ( id TEXT PRIMARY KEY, schedule_id TEXT, computer_id TEXT, @@ -131,6 +132,7 @@ def __init__(self, home=None): ("credentials", "probe_url", "TEXT"), ("credentials", "proof_spec", "TEXT"), ("credentials", "verification_hosts", "TEXT"), + ("schedules", "tz", "TEXT"), ] # Active (non-terminal) auth-attempt statuses, kept here so the partial unique @@ -541,10 +543,10 @@ def prune_expired_assist_tokens(self): (ts, ts)).rowcount # ---- schedules ---- - def insert_schedule(self, sid, cid, name, prompt, kind, spec, jitter_s, next_run_at): + def insert_schedule(self, sid, cid, name, prompt, kind, spec, jitter_s, next_run_at, tz=None): self.q("INSERT INTO schedules (id,computer_id,name,prompt,kind,spec,jitter_s,enabled," - "next_run_at,last_run_at,last_status,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", - (sid, cid, name, prompt, kind, spec, jitter_s, 1, next_run_at, None, None, now())) + "next_run_at,last_run_at,last_status,created_at,tz) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + (sid, cid, name, prompt, kind, spec, jitter_s, 1, next_run_at, None, None, now(), tz)) def get_schedule(self, sid, enabled_only=False): if enabled_only: diff --git a/mcp/case_mcp.py b/mcp/case_mcp.py index 5256ffa..f646712 100644 --- a/mcp/case_mcp.py +++ b/mcp/case_mcp.py @@ -601,12 +601,16 @@ def handoff_get(handoff_id: str) -> dict: if os.environ.get("CASE_MCP_SCHEDULES") == "1": @mcp.tool() def schedule_create(computer_id: str, prompt: str, kind: str = "daily", - spec: str = "09:00", name: str = "", jitter_s: int = 300) -> dict: - """Create a recurring schedule on a computer. kind=daily (spec HH:MM local) or - interval (spec seconds as string). Fires unattended using the host brain credential.""" + spec: str = "09:00", name: str = "", jitter_s: int = 300, + tz: str = "") -> dict: + """Create a recurring schedule on a computer. kind=daily (spec HH:MM in tz, + or box local if tz is empty) or interval (spec seconds as string). Fires + unattended using the host brain credential.""" body = {"prompt": prompt, "kind": kind, "spec": spec, "jitter_s": jitter_s} if name: body["name"] = name + if tz: + body["tz"] = tz return call("POST", f"/computers/{computer_id}/schedules", json=body).json() diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 00abcb7..d952460 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -44,6 +44,37 @@ def test_daily_fires_at_requested_local_time(): assert f"{local.hour:02d}:{local.minute:02d}" == spec, (spec, local.isoformat()) +def test_daily_kolkata_is_0330_utc(): + nxt = _dt(compute_next("daily", "09:00", 0, "Asia/Kolkata")) + assert nxt.hour == 3 and nxt.minute == 30, nxt.isoformat() + + +def test_bad_tz_raises(): + from errors import ApiError + try: + compute_next("daily", "09:00", 0, "Not/AZone") + assert False, "expected bad_tz" + except ApiError as e: + assert e.code == "bad_tz", e + + +def test_sqlite_row_has_no_get_but_tz_index_works(): + store.q("DELETE FROM schedules") + store.insert_schedule("sch_tz", "c_1", "n", "p", "daily", "09:00", 0, + "2026-08-30T03:30:00Z", "Asia/Kolkata") + s = store.get_schedule("sch_tz") + assert not hasattr(s, "get"), type(s) + tz = s["tz"] if "tz" in s.keys() else None + assert tz == "Asia/Kolkata", tz + nxt = compute_next(s["kind"], s["spec"], s["jitter_s"], tz) + assert nxt[11:16] == "03:30", nxt + + +def test_schedules_tz_column_exists(): + cols = [r["name"] for r in store.db.execute("PRAGMA table_info(schedules)")] + assert "tz" in cols + + def test_jitter_stays_bounded(): base = datetime.now(timezone.utc) for _ in range(20): @@ -293,6 +324,138 @@ def active_attempt_exists(self, cid): assert rec.get("status") == "ok", rec +def test_ram_tight_box_is_a_skip_too(): + import scheduler + from errors import ApiError + rec = {} + + class _Store: + def get_schedule(self, sid, enabled_only=False): + return {"id": sid, "computer_id": "c_1", "name": "nightly", "prompt": "go", + "kind": "interval", "spec": "3600", "jitter_s": 0} + def set_schedule_next(self, *a): pass + def insert_run(self, rid, sid, cid, started, ended, code, summary, artifact, status): + rec["summary"] = summary + def set_schedule_result(self, sid, at, status): + rec["status"] = status + + def _tight(cid): + raise ApiError(409, "not_enough_ram", "3072 MB in use of 4096") + + old = (scheduler.store, scheduler.do_wake, scheduler.do_sleep, scheduler.get_computer, + scheduler.notifier, scheduler.emit) + try: + scheduler.store = _Store() + scheduler.get_computer = lambda cid: {"id": cid, "state": "asleep"} + scheduler.do_wake = _tight + scheduler.do_sleep = lambda cid: None + scheduler.notifier = type("N", (), {"push": lambda self, m: None})() + scheduler.emit = lambda *a, **k: None + scheduler.run_schedule("sch_x") + finally: + (scheduler.store, scheduler.do_wake, scheduler.do_sleep, scheduler.get_computer, + scheduler.notifier, scheduler.emit) = old + assert rec["status"] == "skipped", rec + assert "not enough free RAM" in rec["summary"], rec + assert "ApiError" not in rec["summary"], rec + + +def test_run_brain_url_finished(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + resp = mock.Mock(status_code=200, content=b'{"ok":true}', text="ok") + resp.json.return_value = {"ok": True, "finished": True, "text": "done"} + with mock.patch("scheduler.requests.post", return_value=resp) as post: + code, text = scheduler.run_brain("c_1", "hello") + assert (code, text) == (0, "done") + assert post.call_args.args[0] == "http://ui:4174/api/brain" + assert post.call_args.kwargs["json"] == {"computer_id": "c_1", "prompt": "hello"} + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_unfinished(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + resp = mock.Mock(status_code=200, content=b'{"ok":true}', text="") + resp.json.return_value = {"ok": True, "finished": False, "text": "stopped mid-task"} + with mock.patch("scheduler.requests.post", return_value=resp): + code, text = scheduler.run_brain("c_1", "hello") + assert code == 3 and text == "stopped mid-task" + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_503(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + resp = mock.Mock(status_code=503, content=b'{"error":"no key"}', text="") + resp.json.return_value = {"error": "set CASE_DRIVE_API_KEY in .env"} + with mock.patch("scheduler.requests.post", return_value=resp): + code, text = scheduler.run_brain("c_1", "hello") + assert code == 2 + assert "CASE_DRIVE_API_KEY" in text + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_connection_error(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + with mock.patch("scheduler.requests.post", side_effect=scheduler.requests.ConnectionError()): + code, text = scheduler.run_brain("c_1", "hello") + assert code == 127 + assert "http://ui:4174/api/brain" in text + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_url_timeout(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + with mock.patch("scheduler.requests.post", side_effect=scheduler.requests.Timeout()): + code, text = scheduler.run_brain("c_1", "hello") + assert code == -1 + assert "timed out" in text + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + +def test_run_brain_cmd_wins_over_url(): + import unittest.mock as mock + import scheduler + old_cmd, old_url = scheduler.BRAIN_CMD, scheduler.BRAIN_URL + try: + scheduler.BRAIN_CMD = "definitely-not-a-brain-bin {prompt}" + scheduler.BRAIN_URL = "http://ui:4174/api/brain" + with mock.patch("scheduler.requests.post") as post: + code, _ = scheduler.run_brain("c_1", "hello") + assert post.call_count == 0 + assert code == 127 + finally: + scheduler.BRAIN_CMD, scheduler.BRAIN_URL = old_cmd, old_url + + if __name__ == "__main__": for name, fn in sorted(globals().items()): if name.startswith("test_"): From 648bbac26518fb890f7740971319d80b40488256 Mon Sep 17 00:00:00 2001 From: ishu86 <112744711+ishu86@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:06:33 +0530 Subject: [PATCH 3/9] Let the scheduler call Drive's existing brain over HTTP. Compose points CASE_BRAIN_URL at POST /api/brain, which reuses runTurn and the box key. Scheduled runs show up as Drive threads. No second runtime on the cased image. Co-authored-by: Cursor --- compose.yaml | 2 ++ web/web-ui/serve.mjs | 57 ++++++++++++++++++++++++++++++++++++---- web/web-ui/test_http.mjs | 32 +++++++++++++++++++++- 3 files changed, 85 insertions(+), 6 deletions(-) diff --git a/compose.yaml b/compose.yaml index 09e7e47..9380c5e 100644 --- a/compose.yaml +++ b/compose.yaml @@ -41,6 +41,8 @@ services: CASE_BIND: "0.0.0.0" CASE_IMAGE: ${CASE_IMAGE:-case-desk:0.1} CASE_TOKEN: ${CASE_TOKEN:-} + CASE_BRAIN_URL: ${CASE_BRAIN_URL:-http://ui:4174/api/brain} + CASE_BRAIN_CMD: ${CASE_BRAIN_CMD:-} CASE_MAX_RUNNING: ${CASE_MAX_RUNNING:-4} # Total RAM the awake desktops may hold. Unset = 75% of what the engine sees, # which on a Mac is the Docker VM, not the Mac. diff --git a/web/web-ui/serve.mjs b/web/web-ui/serve.mjs index fad2c9a..14247bd 100644 --- a/web/web-ui/serve.mjs +++ b/web/web-ui/serve.mjs @@ -381,7 +381,7 @@ async function power(res, req, action) { } // Chat: same NDJSON contract as web/serve.mjs, hands always local REST. -// Tool names + semantics match mcp/case_mcp.py (prod default surface; no schedules). +// Tool names + semantics match mcp/case_mcp.py. const EXTRA_TOOLS = [ { type: 'function', name: 'computer_list', description: 'List all computers with state, resources and credential names. Reuse an existing computer — only computer_create for an identity that should stay separate.', parameters: { type: 'object', properties: {}, additionalProperties: false } }, { type: 'function', name: 'computer_create', description: 'Create a persistent computer (Linux desktop + Chromium). Blocks until running. Computers are durable: logins, cookies and files survive sleep. Check computer_list first.', parameters: { type: 'object', properties: { name: { type: 'string' } }, additionalProperties: false } }, @@ -1089,7 +1089,7 @@ export async function runTurn({ } finishTurn(hist, thread); emit({ type: 'done', text, computer_id: id, thread_id: thread.id }); - return { text, computerId: id, threadId: thread.id }; + return { text, computerId: id, threadId: thread.id, finished }; } const client = new OpenAI({ apiKey: auth.key }); let text = ''; @@ -1268,7 +1268,7 @@ export async function runTurn({ + ` hist=${JSON.stringify(hist.items).length} compactions=${compactions}`); spend.eff = eff; emit({ type: 'done', text, computer_id: id, thread_id: thread.id, spend, rounds: i }); - return { text, computerId: id, threadId: thread.id }; + return { text, computerId: id, threadId: thread.id, finished }; } catch (err) { // Keep the turn even on provider errors: tools already ran, that work is // real. histCloseOpenCalls synthesizes outputs for any dangling @@ -1276,7 +1276,7 @@ export async function runTurn({ // inside the round; landing here means retries ran dry or a real fault.) finishTurn(hist, thread); if (!stopped()) emit({ type: 'error', error: (err?.message || 'provider error') + ' — say continue, I pick up where I stopped.' }); - return { text: '', computerId: id, threadId: thread.id, error: err?.message || 'provider error' }; + return { text: '', computerId: id, threadId: thread.id, finished: false, error: err?.message || 'provider error' }; } finally { const leftover = takeSteers(thread.id); if (leftover.length) { @@ -1287,6 +1287,51 @@ export async function runTurn({ } } +// Mutable so HTTP tests can stub the provider loop without a live key. +export const driveLoop = { turn: runTurn }; + +export async function brainRoute(req, res) { + const buf = await readBody(req, res); + if (!buf) return; + let body; + try { body = JSON.parse(buf.toString('utf8') || '{}'); } + catch { return json(res, 400, { error: 'bad json' }); } + const computerId = String(body.computer_id || '').trim(); + const prompt = String(body.prompt || '').slice(0, 32000); + if (!computerId || !prompt) return json(res, 400, { error: 'computer_id and prompt required' }); + const auth = envDriveAuth(); + if (!auth.key) { + return json(res, 503, { error: 'set CASE_DRIVE_API_KEY (and CASE_DRIVE_PROVIDER) in .env' }); + } + const model = resolveChatModel(process.env.CASE_DRIVE_MODEL || '', auth.provider); + const thread = newThread('sched · ' + prompt, computerId); + if (CHAT_BUSY.has(thread.id)) return json(res, 409, { error: 'this thread is still running a turn' }); + CHAT_BUSY.add(thread.id); + let text = ''; + let errText = ''; + let finished = false; + const emit = (obj) => { + if (obj?.type === 'done') text = obj.text || text; + if (obj?.type === 'text' && obj.text) text = obj.text; + if (obj?.type === 'error') errText = obj.error || 'provider error'; + }; + try { + const result = await driveLoop.turn({ + thread, inputText: prompt, attaches: [], auth, computerId, + model, effort: 'medium', emit, stopped: () => false, + }); + if (result?.error) errText = result.error; + if (result?.text) text = result.text; + finished = !errText && !!result?.finished; + } catch (err) { + errText = err.message || 'turn failed'; + } finally { + CHAT_BUSY.delete(thread.id); + } + if (errText) return json(res, 200, { ok: false, error: errText }); + return json(res, 200, { ok: true, finished, text }); +} + async function chat(req, res) { const buf = await readBody(req, res); if (!buf) return; @@ -1677,9 +1722,10 @@ export const server = http.createServer(async (req, res) => { running: Number(h.json?.running) || 0, computers: Number(h.json?.computers) || 0, docker: !!h.json?.docker, + brain_key: !!envDriveAuth().key, }); } catch { - return json(res, 200, { ok: true, live: CASE.hostname, up: false, local: LOCAL, max_running: 0, running: 0 }); + return json(res, 200, { ok: true, live: CASE.hostname, up: false, local: LOCAL, max_running: 0, running: 0, brain_key: !!envDriveAuth().key }); } } try { @@ -1690,6 +1736,7 @@ export const server = http.createServer(async (req, res) => { if (p === '/api/creds') return creds(req, res, url); if (p === '/api/threads') return threadsRoute(req, res, url); if (req.method === 'GET' && p === '/api/file') return fsFile(res, url); + if (req.method === 'POST' && p === '/api/brain') return brainRoute(req, res); if (req.method === 'POST' && p === '/api/chat') return chat(req, res); if (req.method === 'POST' && p === '/api/chat/steer') return steer(req, res); if (req.method === 'POST' && p === '/api/attach') return attach(req, res); diff --git a/web/web-ui/test_http.mjs b/web/web-ui/test_http.mjs index 8af6f35..6909472 100644 --- a/web/web-ui/test_http.mjs +++ b/web/web-ui/test_http.mjs @@ -5,10 +5,17 @@ import assert from 'node:assert/strict'; import http from 'node:http'; process.env.CASE_TOKEN = 'tok'; -const { server } = await import('./serve.mjs'); +delete process.env.CASE_DRIVE_API_KEY; +delete process.env.CASE_DRIVE_PROVIDER; +const serve = await import('./serve.mjs'); +const { server } = serve; await new Promise((r) => server.listen(0, '127.0.0.1', r)); const base = `http://127.0.0.1:${server.address().port}`; const get = (p, headers = {}) => fetch(base + p, { headers, redirect: 'manual' }); +const post = (p, body, headers = {}) => fetch(base + p, { + method: 'POST', headers: { 'content-type': 'application/json', ...headers }, + body: typeof body === 'string' ? body : JSON.stringify(body), +}); assert.equal((await get('/api/threads')).status, 401); assert.equal((await get('/api/threads', { authorization: 'Bearer tok' })).status, 200); @@ -19,5 +26,28 @@ const status = await new Promise((r) => http.get({ host: '127.0.0.1', port: serv headers: { authorization: 'Bearer tok', host: 'evil.example' } }, (res) => { res.resume(); r(res.statusCode); })); assert.equal(status, 403); +assert.equal((await post('/api/brain', { computer_id: 'c_1', prompt: 'hi' })).status, 401); +assert.equal((await post('/api/brain', {}, { authorization: 'Bearer tok' })).status, 400); +assert.equal((await post('/api/brain', { computer_id: 'c_1', prompt: 'hi' }, { authorization: 'Bearer tok' })).status, 503); +{ + const h = await (await get('/api/health', { authorization: 'Bearer tok' })).json(); + assert.equal(h.brain_key, false); +} +const origTurn = serve.driveLoop.turn; +process.env.CASE_DRIVE_API_KEY = 'sk-test'; +process.env.CASE_DRIVE_PROVIDER = 'openai'; +serve.driveLoop.turn = async () => ({ text: 'did it', finished: true }); +try { + const r = await post('/api/brain', { computer_id: 'c_1', prompt: 'hi' }, { authorization: 'Bearer tok' }); + assert.equal(r.status, 200); + assert.deepEqual(await r.json(), { ok: true, finished: true, text: 'did it' }); + const h = await (await get('/api/health', { authorization: 'Bearer tok' })).json(); + assert.equal(h.brain_key, true); +} finally { + serve.driveLoop.turn = origTurn; + delete process.env.CASE_DRIVE_API_KEY; + delete process.env.CASE_DRIVE_PROVIDER; +} + server.close(); console.log('test_http: ok'); From 625eee0e7afff3ff3c0a9d2463b3bc922adb940b Mon Sep 17 00:00:00 2001 From: ishu86 <112744711+ishu86@users.noreply.github.com> Date: Mon, 7 Sep 2026 16:08:36 +0530 Subject: [PATCH 4/9] Add a Drive SCHEDULES modal that talks to the existing scheduler. List, create, run, and delete from the sidebar. Daily times keep the browser timezone. The box key from .env is what the run actually uses. Co-authored-by: Cursor --- web/web-ui/index.html | 176 +++++++++++++++++++++++++++++++++++++- web/web-ui/serve.mjs | 42 +++++++++ web/web-ui/test_http.mjs | 1 + web/web-ui/test_serve.mjs | 7 ++ 4 files changed, 225 insertions(+), 1 deletion(-) diff --git a/web/web-ui/index.html b/web/web-ui/index.html index f54ce31..6fa6704 100644 --- a/web/web-ui/index.html +++ b/web/web-ui/index.html @@ -471,6 +471,34 @@ .modal-card input:focus{border-color:var(--accent)} .modal-actions{display:flex;gap:8px;margin-top:12px;justify-content:flex-end} .modal-head{display:flex;align-items:stretch;gap:8px;margin-bottom:14px} +.mbar{display:none;align-items:center;gap:10px;padding:9px 12px;background:var(--field);border-bottom:1px solid var(--soft)} +.mbar #schedBtnM{margin-left:auto} +.mbar #keyBtnM{margin-left:6px} +.sched-card{width:min(520px,100%)} +.sched-card h3{margin:0} +.sched-list{margin:0 0 16px;max-height:40vh;overflow:auto} +.sched-row{ + display:grid;grid-template-columns:1fr auto auto;gap:8px;align-items:start; + padding:10px 0;border-bottom:1px solid var(--soft); +} +.sched-row .sn{font-size:13px;font-weight:600} +.sched-row .sm{font-size:11px;color:var(--faint);margin-top:2px} +.sched-row .sp{font-size:12px;color:var(--mut);margin-top:4px;white-space:pre-wrap} +.sched-empty{padding:12px 0;font-size:13px;color:var(--faint)} +.sched-card label{display:block;font-size:11px;color:var(--faint);margin:10px 0 4px} +.sched-card textarea,.sched-card input[type=text],.sched-card input[type=time],.sched-card input[type=number]{ + width:100%;border:1px solid var(--soft);background:var(--field);padding:8px 10px; + font-family:var(--sans);font-size:13px;outline:none; +} +.sched-card textarea:focus,.sched-card input:focus{border-color:var(--accent);background:var(--paper)} +.sched-when{display:flex;flex-wrap:wrap;gap:10px;align-items:center;margin-top:8px;font-size:13px} +.sched-when label{margin:0;color:var(--ink);font-size:13px} +.sched-when input[type=time],.sched-when input[type=number]{width:auto;max-width:8rem} +.sched-when select{border:1px solid var(--soft);background:var(--field);padding:7px 8px;font-size:12px;max-width:16rem} +.sched-when.interval #schedTz,.sched-when.interval #schedDaily{opacity:.4;pointer-events:none} +.sched-err{font-size:12px;color:var(--warn);min-height:14px;margin-top:8px} +#schedFuel{font-size:12px;color:var(--warn);margin:0 0 10px} +#schedFuel[hidden]{display:none} .key-tabs{display:flex;flex:1;border:1px solid var(--ink);min-width:0} .key-tab{ flex:1;border:0;background:transparent;padding:8px; @@ -489,6 +517,7 @@ @media(max-width:900px){ body{grid-template-columns:1fr} .side,#sideSplit{display:none} + .mbar{display:flex} .stage{flex-direction:column} #railSplit{flex-basis:8px;cursor:row-resize} #railSplit::before{inset:3px 0} @@ -508,12 +537,14 @@
no computer
+
+
@@ -636,6 +667,35 @@
+ +