diff --git a/robot-docs/cache-warming.md b/robot-docs/cache-warming.md new file mode 100644 index 00000000..fe9fd59c --- /dev/null +++ b/robot-docs/cache-warming.md @@ -0,0 +1,48 @@ +# cache warming + +an idle session waiting on work that will wake it still owns the most expensive thing it has: a warm prompt cache. the warmer re-sends the session's last request, with only the output budget lowered, just before the provider's cache would expire, so the next real turn reads its context from cache instead of rewriting it. + +first case: an orchestrator whose children are still running. it sits idle while its cache decays. warming keeps it, and the same eligibility test is one small function (`warm.wanted` in `src/albedo/daemon/warm.gleam`), so persistent agents can add their own reason later. + +## when it warms + +a session is eligible while: + +- it is idle — no turn, compaction or warm ping in flight. a ping holds the session exactly as a compaction run does, so a submit queues behind it and starts when the ping ends; it is short. +- at least one of its children is running (`bus.is_running`, the same answer `agents.self.children()` gives). +- the last request's cached prefix is worth a round trip: at least `minCachedTokens` (default 1024) cached tokens on the turn it repeats — reads plus writes, so a first write-only turn counts. +- a clock can be beat. with cache marks (Claude), the shortest `ttlSeconds` among them. with none — the OpenAI protocols cache on their own — the cache table's entry: `refresh`/`fixed` policy, its first tier's `seconds`. `evict`/`unknown`, or no entry, means no warming (robot-docs/cache-ttl.md). + +warming starts when the turn that will be repeated ends, and again after every ping. a child that starts running while the session is already idle does not wake warming until the session's next turn. any new activity — a submit, a turn, a compaction, a model or profile change — cancels the pending ping through a generation counter; a stale tick is dropped when it arrives. + +## what a ping sends + +exactly the request the turn actually sent: same instructions, tools, inputs, options, cache marks. it is captured, never rebuilt — a rebuild runs `prepare`, which can compact or call the summarizer and change the prefix. when a turn call succeeds, `loop.call` reports the request, its prefix identity, its usage and timing to the session, which keeps the latest one in memory only. a restart simply stops warming. + +the only change is the output budget: `max_output_tokens` 1, or 16 on the Responses protocol, whose minimum is higher. + +## the stop rule and its arithmetic + +ping at `min(0.9 × ttl, ttl − 10)` seconds after the previous send — from the send's start when the entry's clock is `request` (Anthropic counts lifetime from request start), else from its finish. + +each ping costs about `read × cached` input; letting the cache go cold costs one rewrite of about `write × cached` — `write` the matching tier's write multiplier (default 1.0), `read` the entry's read multiplier (default 0.1). so at most `floor(write / read)` consecutive pings per idle stretch, then it stops and lets the cache go cold: ski rental, with the break-even at Claude's 5-minute tier at 1.25 / 0.1 = 12 pings ≈ 54 minutes. a new turn resets the budget. + +warming also stops when: + +- a ping comes back with no cached input tokens although the turn it repeats read or wrote cache — the TTL model was wrong, and the ledger keeps the evidence. +- the kernel was released, or the ping failed or was cancelled. +- settings say `enabled: false`. + +## the ledger kind + +every ping is a ledger row of kind `warm`, with the same prefix identity (`headHash`, `inputs`, `replaced`, `projectionHash`) and `cacheMarks` as the turn it repeats, and its own usage. its `cachedInputTokens` says whether the cache was still there, which makes every ping a free TTL measurement for phase 2. a ping never attaches a seq, never commits to the transcript, and never publishes stream events; the agents bus does not flap for it (robot-docs/usage-ledger.md). + +## settings + +`$ALBEDO_HOME/extensions.json` under `"warm"`, re-read each time a ping is scheduled: + +```json +{ "warm": { "enabled": true, "minCachedTokens": 1024 } } +``` + +malformed settings fall back to these defaults. the cache table itself (its layers, refresh, local override) is documented in robot-docs/cache-ttl.md. diff --git a/robot-docs/usage-ledger.md b/robot-docs/usage-ledger.md index 8fb5843f..c86e9d01 100644 --- a/robot-docs/usage-ledger.md +++ b/robot-docs/usage-ledger.md @@ -7,7 +7,7 @@ Rows are written by the worker that streams the call, one per upstream attempt: ## What a row holds - `session`, `seq` — the conversation, and the transcript row the call produced: the first assistant row of the response that turn committed (for a tool-call turn, the row carrying the call itself, not the tool output after it). `seq` is null when the call produced no transcript row: a failed call, or a compaction summary, which writes strategy state instead of the transcript. -- `kind` — `turn` for the model/tool loop, `summarizer` for the compaction summary call. +- `kind` — `turn` for the model/tool loop, `summarizer` for the compaction summary call, `warm` for a cache-warming ping that repeats the last turn's request with a tiny output budget. A `warm` row never attaches a seq: pings commit nothing. - `profile`, `provider`, `account`, `model` — the saved profile the request went through, `protocol:endpoint` as it was reached, the non-secret label of the account that served (an account id, email, or short key hash; null when the provider has no account pool), and the model. - `startedMs`, `finishedMs` — when the attempt ran. - `outcome`, `status`, `error` — `ok`, or `error` with the HTTP status and a truncated error body, so a 429 is a quota reading. diff --git a/src/albedo/daemon/ledger.gleam b/src/albedo/daemon/ledger.gleam index 8c0431ac..d765f157 100644 --- a/src/albedo/daemon/ledger.gleam +++ b/src/albedo/daemon/ledger.gleam @@ -26,16 +26,19 @@ import sqlight pub const schema = "CREATE TABLE IF NOT EXISTS request_ledger(id INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),seq INTEGER,kind TEXT NOT NULL,profile TEXT NOT NULL,provider TEXT NOT NULL,account TEXT,model TEXT NOT NULL,started_ms INTEGER NOT NULL,finished_ms INTEGER NOT NULL,outcome TEXT NOT NULL,status INTEGER,error TEXT,input_tokens INTEGER,cached_input_tokens INTEGER,cache_creation_tokens INTEGER,cache_write_5m_tokens INTEGER,cache_write_1h_tokens INTEGER,output_tokens INTEGER,reasoning_tokens INTEGER,head_hash TEXT NOT NULL,inputs INTEGER NOT NULL,replaced INTEGER,projection_hash TEXT,strategy TEXT,cache_marks TEXT NOT NULL DEFAULT '[]'); CREATE INDEX IF NOT EXISTS request_ledger_session ON request_ledger(session,id);" /// What the call was for. A turn is the model loop; a summarizer is the -/// compaction summary call `loop.summarize` makes. +/// compaction summary call `loop.summarize` makes; a warm is a cache-warming +/// ping that repeats the last turn's request with a tiny output budget. pub type Kind { Turn Summarizer + Warm } fn kind_name(kind: Kind) -> String { case kind { Turn -> "turn" Summarizer -> "summarizer" + Warm -> "warm" } } diff --git a/src/albedo/daemon/session.gleam b/src/albedo/daemon/session.gleam index 9c369191..ba4359ef 100644 --- a/src/albedo/daemon/session.gleam +++ b/src/albedo/daemon/session.gleam @@ -7,6 +7,7 @@ import albedo/daemon/event_buffer import albedo/daemon/events as view import albedo/daemon/family import albedo/daemon/images +import albedo/daemon/ledger import albedo/daemon/mail import albedo/daemon/session_extensions import albedo/daemon/session_history @@ -17,6 +18,7 @@ import albedo/daemon/session_state import albedo/daemon/session_submission import albedo/daemon/turn.{type Submission, Submission} import albedo/daemon/usage +import albedo/daemon/warm import albedo/harness/command import albedo/harness/extension import albedo/harness/extensions/python/kernel as python @@ -130,6 +132,13 @@ pub type Message { RecordContext(String, context_snapshot.Snapshot, Bool, Subject(Nil)) RecordUsage(String, usage.Metadata, Subject(Result(Nil, String))) ReportPin(String, Option(Int), Subject(Nil)) + /// The worker's turn call succeeded; its request is what a warm ping + /// repeats. Fire-and-forget, like the pin report. + ReportSent(String, warm.Sent, Subject(Nil)) + /// A scheduled warm ping is due, under the generation it was scheduled at. + WarmTick(Int) + /// A warm ping finished: its usage says whether the cache was still there. + WarmFinished(String, Result(Option(types.Usage), String)) Compact(Option(String), Subject(Result(json.Json, String))) RefreshData(Subject(Result(json.Json, String))) DrainSteering(String, Subject(Result(List(types.Input), String))) @@ -216,6 +225,7 @@ pub fn start( None, now_ms(), None, + warm.fresh(), ) // Background jobs wake this session through the kernel's jobs route; the // registered closure lands a completion notice as an ordinary submit, so @@ -494,6 +504,7 @@ fn handle(state: State, message: Message) { } } ChangeWorkspace(cwd, reply) -> { + let state = stirred(state) let changed = case turn.running(state.activity), directory(cwd) { Some(_), _ -> Error("session must be idle to change workspace") None, False -> Error("workspace must be an existing absolute directory") @@ -536,10 +547,10 @@ fn handle(state: State, message: Message) { runtime.extension_summaries(state.host, state.info.id), ) ChangeExtension(change, reply) -> - transition(reply, session_extensions.change(state, change)) + transition(reply, session_extensions.change(stirred(state), change)) ChangeModel(model, provider_name, effort, reply) -> { let #(state, outcome) = - session_provider.select(state, model, provider_name, effort) + session_provider.select(stirred(state), model, provider_name, effort) answer( state, reply, @@ -550,7 +561,7 @@ fn handle(state: State, message: Message) { ReadEffort(reply) -> answer(state, reply, session_provider.read_effort(state)) ChangeEffort(level, reply) -> - transition(reply, session_provider.change_effort(state, level)) + transition(reply, session_provider.change_effort(stirred(state), level)) RefreshData(reply) -> transition(reply, session_extensions.refresh(state)) Compact(strategy, reply) -> @@ -753,6 +764,30 @@ fn handle(state: State, message: Message) { _, _, _ -> actor.continue(state) } } + ReportSent(id, sent, reply) -> { + process.send(reply, Nil) + case turn.live(state.activity, id) { + True -> + actor.continue( + session_state.State(..state, warm: warm.captured(state.warm, sent)), + ) + False -> actor.continue(state) + } + } + WarmTick(generation) -> + case generation == state.warm.generation, turn.running(state.activity) { + False, _ | True, Some(_) -> actor.continue(state) + True, None -> actor.continue(warm_tick(state)) + } + WarmFinished(id, outcome) -> + case turn.owner(state.activity, id) { + Some(run) -> + case run.work { + turn.Warm -> warm_finish(state, run, outcome) + _ -> actor.continue(state) + } + None -> actor.continue(state) + } RecordContext(id, snapshot, compacted, reply) -> answer( case turn.live(state.activity, id) { @@ -805,11 +840,16 @@ fn handle(state: State, message: Message) { Down(process.ProcessDown(_, pid, _)) -> case turn.running(state.activity) { Some(run) if run.pid == pid -> - finish_run( - state, - run, - Error("worker stopped; execution may have had effects"), - ) + case run.work { + turn.Warm -> + warm_finish(state, run, Error("warm ping worker stopped")) + _ -> + finish_run( + state, + run, + Error("worker stopped; execution may have had effects"), + ) + } _ -> actor.continue(state) } Down(_) -> actor.continue(state) @@ -1292,6 +1332,9 @@ fn admit( submission: Submission, reply: Subject(Result(Bool, SubmissionError)), ) -> actor.Next(State, Message) { + // Anything a client submits counts as attention: a pending warm ping is + // cancelled, and the turn that follows re-captures the request to repeat. + let state = stirred(state) case turn.admit(state.activity, submission, list.length(state.steering)) { turn.Reject(turn.Busy) -> answer(state, reply, Error(Busy)) turn.Reject(turn.Oversized) -> @@ -1505,6 +1548,19 @@ fn finish_run( state: State, run: turn.Run, outcome: Result(Nil, String), +) -> actor.Next(State, Message) { + case run.work { + // A ping holds the session like a compaction run does, but commits + // nothing and never reaches the transcript. + turn.Warm -> warm_finish(state, run, result.replace(outcome, None)) + _ -> finish_turn(state, run, outcome) + } +} + +fn finish_turn( + state: State, + run: turn.Run, + outcome: Result(Nil, String), ) -> actor.Next(State, Message) { process.demonitor_process(run.monitor) let persisted = @@ -1550,10 +1606,121 @@ fn finish_run( turn.Turn(_), False -> answer_parent(state, outcome) _, _ -> state } + let state = warm_after_run(state, run) process.send(state.self, Collect) actor.continue(start_queued(state)) } +/// A scheduled ping is due under the current generation: start it, or stop +/// warming for this idle stretch. Settings are re-read here, so an edit to +/// extensions.json takes effect without a restart. +fn warm_tick(state: State) -> State { + case state.warm.sent { + None -> state + Some(sent) -> { + let decision = + warm.decision(state.home, state.info.provider, state.info.model, sent) + case + warm.next( + decision, + warm.wanted(runtime.ledger(state.host), state.info.id), + state.warm.pings, + ) + { + None -> state + Some(plan) -> { + let #(with_kernel, kernel) = session_namespace.ready(state) + case kernel, session_provider.configured_client(with_kernel) { + // A released kernel ends warming: a restart simply stops it. + Some(live), Ok(#(primed, client)) -> { + let started = ledger.now() + session_state.State( + ..session_run.start_warm( + primed, + live, + client, + sent, + WarmFinished, + ), + warm: warm.pinging(state.warm, started, plan), + ) + } + _, _ -> state + } + } + } + } + } +} + +/// A warm ping ended. It never touched the transcript, so nothing commits and +/// nothing is announced; the submissions it held back start now, and the next +/// ping is scheduled only while the idle stretch still holds. +fn warm_finish( + state: State, + run: turn.Run, + outcome: Result(Option(types.Usage), String), +) -> actor.Next(State, Message) { + process.demonitor_process(run.monitor) + let now = ledger.now() + // The next ping is decided from the run that just ended, before the state + // records that no ping is in flight any more. + let again = warm.reschedule(state.warm, run.cancelled, outcome, now, now) + let state = + session_state.State( + ..state, + activity: turn.Resting, + warm: warm.settled(state.warm), + ) + // A webhook or wake refused while the ping held the session can be + // admitted now. + mail.waiting() + case again { + Some(#(delay, generation)) -> { + let _ = process.send_after(state.self, delay, WarmTick(generation)) + Nil + } + None -> Nil + } + actor.continue(start_queued(state)) +} + +/// A run that held the session ended: a turn's request is what the next idle +/// stretch can warm, while a compaction rewrote the prefix, so nothing is +/// warmable until the next turn captures a request again. +fn warm_after_run(state: State, run: turn.Run) -> State { + case run.work, run.cancelled { + turn.Turn(_), False -> { + case + warm.first_ping( + state.home, + state.info.provider, + state.info.model, + state.warm, + warm.wanted(runtime.ledger(state.host), state.info.id), + ledger.now(), + ) + { + Some(#(delay, generation)) -> { + let _ = process.send_after(state.self, delay, WarmTick(generation)) + Nil + } + None -> Nil + } + state + } + turn.Compaction, _ -> + session_state.State(..state, warm: warm.reset(state.warm)) + _, _ -> state + } +} + +/// Any new activity: a pending warm tick scheduled under an older generation +/// is dropped when it arrives. +fn stirred(state: State) -> State { + session_state.State(..state, warm: warm.stirred(state.warm)) +} + fn start_queued(state: State) -> State { case turn.starts_turn(state.steering), kernel_or_park(state, StartQueued) { False, _ -> state @@ -1603,6 +1770,7 @@ fn start_worker( work: turn.Work, ) -> State { bus.running(state.info.id, True) + let state = stirred(state) session_run.start( state, kernel, @@ -1616,6 +1784,7 @@ fn start_worker( RecordUsage, DrainSteering, ReportPin, + ReportSent, Finished, Collect, ), diff --git a/src/albedo/daemon/session_run.gleam b/src/albedo/daemon/session_run.gleam index 6df7aa90..0b14456b 100644 --- a/src/albedo/daemon/session_run.gleam +++ b/src/albedo/daemon/session_run.gleam @@ -5,6 +5,7 @@ import albedo/daemon/conversation import albedo/daemon/session_state import albedo/daemon/turn import albedo/daemon/usage +import albedo/daemon/warm import albedo/harness/extension import albedo/harness/loop import albedo/harness/runtime @@ -27,6 +28,7 @@ pub type Messages(message) { usage: fn(String, usage.Metadata, Subject(Result(Nil, String))) -> message, drain: fn(String, Subject(Result(List(types.Input), String))) -> message, pin: fn(String, Option(Int), Subject(Nil)) -> message, + sent: fn(String, warm.Sent, Subject(Nil)) -> message, finished: fn(String, Result(Nil, String)) -> message, collect: message, ) @@ -209,6 +211,21 @@ pub fn start( usage_fn(owner, run_id, messages, 10_000), drain_fn(owner, run_id, messages, 10_000), fn(head) { report(owner, messages.pin(run_id, head, _)) }, + fn(request, prefix, marks, usage, started, finished) { + report(owner, messages.sent( + run_id, + warm.Sent( + request, + prefix, + usage, + marks, + client.endpoint, + started, + finished, + ), + _, + )) + }, // The request ledger's identity for this session's provider calls. state.info.id, state.info.provider, @@ -221,6 +238,8 @@ pub fn start( messages.finished(run_id, case work { turn.Compaction -> loop.compact(worker, model_history) turn.Turn(_) -> loop.run(worker, run_id, model_history, 0) + // A ping never starts here; start_warm owns its worker. + turn.Warm -> Error("warm pings run through start_warm") }), ) }) @@ -233,8 +252,65 @@ pub fn start( history: None, activity: turn.Running(run), context: case work { - turn.Compaction -> state.context + turn.Compaction | turn.Warm -> state.context turn.Turn(_) -> session_state.unprepared() }, ) } + +/// One cache-warming ping: `sent`'s request re-sent with a tiny output +/// budget. It commits nothing and publishes nothing, so its worker carries +/// no history and answers only its own outcome and usage; the session queues +/// submissions behind it as it does for a compaction run. The run never +/// announces itself on the agents bus: an idle session stays idle. +pub fn start_warm( + state: session_state.State(message), + kernel: runtime.Session, + client: extension.Upstream, + sent: warm.Sent, + finished: fn(String, Result(Option(types.Usage), String)) -> message, +) -> session_state.State(message) { + let run_id = new_id() + let owner = state.self + // The worker's closures must capture these fields, never `state`: the + // session state carries the loaded transcript. + let worker = + loop.Loop( + state.info.model, + state.info.effort, + state.host, + kernel, + state.pin, + client, + // A ping must never show on the session's stream, and must never + // block on it either. + fn(_event) { True }, + fn(_inputs, _stage, _thought) { Error("a warm ping never commits") }, + fn(_request, _observation, _compacted) { Nil }, + fn(_metadata) { Ok(Nil) }, + fn() { Ok([]) }, + fn(_head) { Nil }, + fn(_request, _prefix, _marks, _usage, _started, _finished) { Nil }, + state.info.id, + state.info.provider, + ) + let pid = + process.spawn_unlinked(fn() { + label("albedo_warm", run_id) + process.send( + owner, + finished(run_id, loop.warm(worker, sent.request, sent.prefix)), + ) + }) + session_state.State( + ..state, + activity: turn.Running(turn.Run( + run_id, + pid, + process.monitor(pid), + False, + turn.latch(), + turn.Warm, + )), + ) +} diff --git a/src/albedo/daemon/session_state.gleam b/src/albedo/daemon/session_state.gleam index 1e99a301..3f7fe48c 100644 --- a/src/albedo/daemon/session_state.gleam +++ b/src/albedo/daemon/session_state.gleam @@ -7,6 +7,7 @@ import albedo/daemon/event_buffer import albedo/daemon/transcript import albedo/daemon/turn.{type Submission} import albedo/daemon/usage +import albedo/daemon/warm import albedo/harness/loop import albedo/harness/runtime import gleam/erlang/process.{type Subject} @@ -37,6 +38,8 @@ pub type State(message) { /// While the kernel boots: boot attempts so far, and the work waiting for /// it, re-sent to the actor once the kernel is ready. booting: Option(#(Int, List(message))), + /// The prompt-cache warmer's view: the last sent request, and its budget. + warm: warm.Warming, ) } diff --git a/src/albedo/daemon/turn.gleam b/src/albedo/daemon/turn.gleam index 6838f9c6..6d260e8a 100644 --- a/src/albedo/daemon/turn.gleam +++ b/src/albedo/daemon/turn.gleam @@ -42,6 +42,8 @@ pub type Submission { pub type Work { Compaction + /// A cache-warming ping: commits nothing and publishes nothing. + Warm /// `stage` is None until the turn's first commit. Turn(stage: Option(conversation.Stage)) } @@ -253,6 +255,7 @@ pub fn phase(activity: Activity) -> String { Resting -> "resting" Interrupted -> "interrupted" Running(Run(work: Compaction, ..)) -> "compacting" + Running(Run(work: Warm, ..)) -> "warming" Running(Run(work: Turn(None), ..)) -> "preparing" Running(Run(work: Turn(Some(stage)), ..)) -> conversation.stage_name(stage) } diff --git a/src/albedo/daemon/warm.gleam b/src/albedo/daemon/warm.gleam new file mode 100644 index 00000000..be9ee0c6 --- /dev/null +++ b/src/albedo/daemon/warm.gleam @@ -0,0 +1,404 @@ +//// The prompt-cache warmer. A session that sits idle waiting on work that +//// will wake it (its children still running) re-sends the request its last +//// turn actually sent, with only the output budget lowered, just before the +//// provider's cache would expire: the next real turn reads its context from +//// cache instead of rewriting it. The session keeps the last sent request in +//// memory only, so a daemon restart simply stops warming. +//// +//// Every decision here is a plain value, so the stop rules and their +//// arithmetic can hold without a kernel, a provider, or an actor. Each ping +//// also measures one TTL for free: its ledger row's `cachedInputTokens` says +//// whether the cache was still there. + +import albedo/daemon/bus +import albedo/daemon/configuration +import albedo/daemon/family +import albedo/daemon/ledger +import albedo/daemon/store +import albedo/harness/cache_ttl +import albedo/harness/settings +import albedo/openai_api/types +import gleam/dynamic/decode +import gleam/float +import gleam/int +import gleam/list +import gleam/option.{type Option, None, Some} +import gleam/result +import gleam/uri + +/// Settings from `extensions.json` under `"warm"`. +pub type Settings { + Settings(enabled: Bool, min_cached_tokens: Int) +} + +/// What the last successful turn call sent, as a ping must repeat it. +pub type Sent { + Sent( + request: types.Request, + prefix: ledger.Prefix, + /// The usage that call reported: its cached tokens are how big the warm + /// prefix is. + usage: Option(types.Usage), + /// Where the request asked the provider to cache, as the call's ledger + /// row recorded them: a ping repeats them. + marks: List(types.CacheMark), + /// The endpoint the call went to, for cache-table lookups. + endpoint: String, + started_ms: Int, + finished_ms: Int, + ) +} + +/// The live warming state of one session: the captured request, a generation +/// counter that any new activity raises (a tick scheduled under an older one +/// is dropped), how many consecutive pings this idle stretch has sent, and +/// the ping now in flight. +pub type Warming { + Warming( + sent: Option(Sent), + generation: Int, + pings: Int, + in_flight: Option(#(Int, Plan)), + ) +} + +pub fn fresh() -> Warming { + Warming(None, 0, 0, None) +} + +/// The last turn's call succeeded: what it sent is what a ping would repeat, +/// and a new idle stretch's ping budget begins. +pub fn captured(warming: Warming, sent: Sent) -> Warming { + Warming(..warming, sent: Some(sent), pings: 0) +} + +/// Any new activity: pending ticks scheduled under an older generation are +/// dropped when they arrive. +pub fn stirred(warming: Warming) -> Warming { + Warming(..warming, generation: warming.generation + 1) +} + +/// A compaction rewrote the prefix, so nothing the next turn sends is warm +/// yet; the capture is dropped and the generation moves on. +pub fn reset(warming: Warming) -> Warming { + Warming(None, warming.generation + 1, 0, None) +} + +/// The ping is going out now, under `plan`, started at `started_ms`. +pub fn pinging(warming: Warming, started_ms: Int, plan: Plan) -> Warming { + Warming( + ..warming, + pings: warming.pings + 1, + in_flight: Some(#(started_ms, plan)), + ) +} + +/// The ping came back: no ping is in flight any more. +pub fn settled(warming: Warming) -> Warming { + Warming(..warming, in_flight: None) +} + +/// Everything a scheduling decision needs, resolved once: settings re-read +/// from disk, and the plan the last request implies from the cache table. +pub type Decision { + Decision(settings: Settings, plan: Option(Plan)) +} + +/// How to keep a prefix warm: how often to ping, how many consecutive pings +/// still pay for themselves, and whether the TTL clock counts from the +/// send's start or its finish. +pub type Plan { + Plan(interval_ms: Int, cap: Int, from_start: Bool) +} + +/// The warm settings of `home`, or their defaults when unset or malformed. +pub fn read_settings(home: String) -> Settings { + settings.load_at(home, "warm", settings_decoder(), defaults) + |> result.unwrap(defaults) +} + +const defaults = Settings(True, 1024) + +fn settings_decoder() -> decode.Decoder(Settings) { + use enabled <- decode.optional_field("enabled", True, decode.bool) + use min_cached <- decode.optional_field("minCachedTokens", 1024, decode.int) + decode.success(Settings(enabled, min_cached)) +} + +/// Whether work that will wake this idle session is still running: one small +/// test, so persistent agents can add their own reason later. +pub fn wanted(db: store.Store, session: String) -> Bool { + case family.children(db, session) { + Ok(children) -> + list.any(children, fn(child) { + !child.closed && bus.is_running(child.session) + }) + Error(_) -> False + } +} + +/// One scheduling moment: settings re-read, the cache table looked up once +/// for the provider this session resolves to, and the plan it implies. +pub fn decision( + home: String, + provider: String, + model: String, + sent: Sent, +) -> Decision { + let settings = read_settings(home) + let extension = case configuration.named(home, provider) { + Ok(configured) -> configured.extension + Error(_) -> "" + } + Decision( + settings, + plan( + sent, + cache_ttl.lookup(extension, host(sent.endpoint), model), + settings.min_cached_tokens, + ), + ) +} + +/// The plan for one sent request, or `None` when no clock can be beat: the +/// shortest TTL its cache marks ask for, or with none (the OpenAI protocols, +/// which cache on their own) the cache table's first tier, and only while +/// the entry's policy is one a ping can extend. A prefix smaller than +/// `min_cached` tokens is not worth a round trip. +pub fn plan( + sent: Sent, + entry: Option(cache_ttl.Entry), + min_cached: Int, +) -> Option(Plan) { + case resolution(sent, entry) { + None -> None + Some(#(ttl_seconds, write, read, from_start)) -> + case cached_prefix(sent.usage) < min_cached { + True -> None + False -> + case cap(write, read) { + 0 -> None + pings -> Some(Plan(interval_ms(ttl_seconds), pings, from_start)) + } + } + } +} + +/// The clock to beat, with the prices of reading it and letting it go cold: +/// `#(ttl seconds, write multiplier, read multiplier, counts from send start)`. +fn resolution( + sent: Sent, + entry: Option(cache_ttl.Entry), +) -> Option(#(Int, Float, Float, Bool)) { + let read = case entry { + Some(entry) -> + entry.read + |> option.map(fn(price) { + case price >. 0.0 { + True -> price + False -> default_read + } + }) + |> option.unwrap(default_read) + None -> default_read + } + let from_start = case entry { + Some(entry) -> entry.clock == cache_ttl.Request + None -> False + } + case marks_ttl(sent) { + // The request asked for explicit cache entries: the shortest of them is + // the clock, and its matching tier prices the write. + Some(ttl_seconds) -> { + let write = tier_write(entry, ttl_seconds) + Some(#(ttl_seconds, write, read, from_start)) + } + // No marks: the provider caches on its own, so only the table knows, and + // only while its policy is one a ping can extend. + None -> + case entry { + Some(entry) -> + case extends(entry.policy), entry.tiers { + True, Some([tier, ..]) -> + Some(#( + tier.seconds, + option.unwrap(tier.write, default_write), + read, + from_start, + )) + _, _ -> None + } + None -> None + } + } +} + +/// Whether a ping can push the entry's lifetime back: a fixed clock or one +/// every hit restarts. Best-effort eviction has no clock to beat. +fn extends(policy: cache_ttl.Policy) -> Bool { + case policy { + cache_ttl.Refresh | cache_ttl.Fixed -> True + cache_ttl.Evict | cache_ttl.Unknown -> False + } +} + +const default_read = 0.1 + +const default_write = 1.0 + +/// The shortest TTL the request's cache marks ask for, when it asked for +/// any: the clock a ping has to beat. +fn marks_ttl(sent: Sent) -> Option(Int) { + list.fold(sent.marks, None, fn(shortest, mark) { + case shortest { + None -> Some(mark.ttl_seconds) + Some(seconds) -> Some(int.min(seconds, mark.ttl_seconds)) + } + }) +} + +/// The write multiplier of the tier matching `ttl_seconds`, when the entry +/// prices one; the default prices a rewrite at parity. +fn tier_write(entry: Option(cache_ttl.Entry), ttl_seconds: Int) -> Float { + case entry { + Some(entry) -> + case entry.tiers { + Some(tiers) -> + list.find(tiers, fn(tier) { tier.seconds == ttl_seconds }) + |> result.map(fn(tier) { option.unwrap(tier.write, default_write) }) + |> result.unwrap(default_write) + None -> default_write + } + None -> default_write + } +} + +/// Ping at the smaller of ten percent and ten seconds under the TTL, so the +/// entry is always refreshed with time to spare. +fn interval_ms(ttl_seconds: Int) -> Int { + let margin = { ttl_seconds - 10 } * 1000 + case ttl_seconds > 10 { + True -> int.min(ttl_seconds * 900, margin) + False -> ttl_seconds * 900 + } +} + +/// Ski rental: each ping costs about `read × cached` input, letting the cache +/// go cold costs one rewrite of about `write × cached`, so at most +/// `floor(write / read)` consecutive pings pay for themselves. Reading costs +/// at least as much as writing means warming never does. +fn cap(write: Float, read: Float) -> Int { + case write >=. read { + False -> 0 + True -> float.floor(write /. read) |> float.truncate + } +} + +/// Tokens the last turn found in the cache: what it read plus what it wrote. +/// An unreported count is zero, which keeps a first write-only turn warmable. +pub fn cached_prefix(usage: Option(types.Usage)) -> Int { + case usage { + Some(types.Usage( + cached_input_tokens: cached, + cache_creation_tokens: creation, + .., + )) -> option.unwrap(cached, 0) + option.unwrap(creation, 0) + None -> 0 + } +} + +/// Whether the next ping should go out: warming is enabled, work that will +/// wake the session still runs, a plan exists, and pings remain in budget. +pub fn next(decision: Decision, wanted: Bool, pings: Int) -> Option(Plan) { + case decision.settings.enabled, wanted, decision.plan { + False, _, _ | _, False, _ | _, _, None -> None + True, True, Some(plan) -> + case pings < plan.cap { + True -> Some(plan) + False -> None + } + } +} + +/// When the first ping of an idle stretch is due, if one is: the interval +/// after the turn's send, counted from its start when the entry's clock +/// does. Answers the delay and the generation the tick runs under. +pub fn first_ping( + home: String, + provider: String, + model: String, + warming: Warming, + wanted: Bool, + now_ms: Int, +) -> Option(#(Int, Int)) { + case warming.sent { + None -> None + Some(sent) -> + case next(decision(home, provider, model, sent), wanted, warming.pings) { + None -> None + Some(plan) -> Some(#(delay(sent, plan, now_ms), warming.generation)) + } + } +} + +fn delay(sent: Sent, plan: Plan, now_ms: Int) -> Int { + let anchor = case plan.from_start { + True -> sent.started_ms + False -> sent.finished_ms + } + int.max(0, anchor + plan.interval_ms - now_ms) +} + +/// Whether and when the next ping is due after one came back: the budget or +/// the cache itself can be gone, a cancelled or failed ping ends the stretch, +/// and a cold ping that should have been warm ends it because the TTL model +/// was wrong — the ledger keeps the evidence. +pub fn reschedule( + warming: Warming, + cancelled: Bool, + outcome: Result(Option(types.Usage), String), + finished_ms: Int, + now_ms: Int, +) -> Option(#(Int, Int)) { + case cancelled, outcome, warming.in_flight, warming.sent { + True, _, _, _ -> None + False, Error(_), _, _ -> None + False, _, None, _ -> None + False, Ok(usage), Some(#(started_ms, plan)), Some(sent) -> + case cache_lost(sent, usage), warming.pings < plan.cap { + True, _ | _, False -> None + False, True -> { + let anchor = case plan.from_start { + True -> started_ms + False -> finished_ms + } + Some(#( + int.max(0, anchor + plan.interval_ms - now_ms), + warming.generation, + )) + } + } + _, _, _, None -> None + } +} + +/// Whether a ping found the cache gone although the turn it repeats read or +/// wrote one. +pub fn cache_lost(sent: Sent, ping: Option(types.Usage)) -> Bool { + let hit = case ping { + Some(types.Usage(cached_input_tokens: Some(count), ..)) -> count + _ -> 0 + } + case hit > 0 { + True -> False + False -> cached_prefix(sent.usage) > 0 + } +} + +/// The host of an endpoint URL, as the cache table matches it. +fn host(endpoint: String) -> String { + case uri.parse(endpoint) { + Ok(uri.Uri(host: Some(host), ..)) -> host + _ -> "" + } +} diff --git a/src/albedo/harness/loop.gleam b/src/albedo/harness/loop.gleam index 75a8219f..44521b92 100644 --- a/src/albedo/harness/loop.gleam +++ b/src/albedo/harness/loop.gleam @@ -38,6 +38,17 @@ pub type Loop { /// Reports the pin's compaction baseline (`Some`) or that compaction /// replaced the history the pinned prompt was cached with (`None`). report_pin: fn(Option(Int)) -> Nil, + /// Reports a successful turn call's request, prefix identity, cache + /// marks, usage and timing, so the session can repeat it as a + /// cache-warming ping. + report_turn: fn( + types.Request, + ledger.Prefix, + List(types.CacheMark), + Option(types.Usage), + Int, + Int, + ) -> Nil, /// The session whose provider calls the request ledger records. session: String, /// The saved profile those calls went through. @@ -209,6 +220,45 @@ pub fn compact(state: Loop, inputs: List(types.Input)) -> Result(Nil, String) { Ok(Nil) } +/// One cache-warming ping: re-send a captured request with the smallest +/// output budget the protocol accepts, so the provider refreshes the cached +/// prefix. The request is exactly the one the turn sent — never rebuilt +/// through `prepare`, which could compact or call the summarizer — with only +/// its output budget lowered. A ping commits nothing and publishes nothing; +/// it answers the ping's usage, whose cached tokens say whether the cache was +/// still there. +pub fn warm( + state: Loop, + request: types.Request, + prefix: ledger.Prefix, +) -> Result(Option(types.Usage), String) { + let ping = + types.Request( + ..request, + max_output_tokens: Some(ping_budget(state.upstream.protocol)), + ) + call( + state, + ledger.Warm, + ping, + prefix, + // A retry is silent: a ping must never show on the session's stream. + fn(_event) { True }, + fn(_event) { types.Continue }, + ) + |> result.map(fn(attempt) { attempt.1.usage }) + |> result.map_error(describe(state.upstream, _)) +} + +/// The smallest output budget a ping may ask for: 1, or 16 on the Responses +/// protocol, whose minimum is higher. +fn ping_budget(protocol: types.Protocol) -> Int { + case protocol { + types.Responses -> 16 + types.ChatCompletions -> 1 + } +} + /// One tool call, committed before its transcript event; a client refusal /// between the progress event and the result cancels the turn. fn run_tool(state: Loop, call: types.ToolCall) -> Result(types.Input, String) { @@ -315,10 +365,12 @@ fn call( fn() { let started = ledger.now() let outcome = state.upstream.stream(request, on_event) + let finished = ledger.now() let usage = case outcome { Ok(turn) -> turn.usage Error(_) -> None } + let marks = state.upstream.cache_marks(request) // The account label is read after the attempt: rotation records which // account served while the request streams. let row = @@ -332,11 +384,11 @@ fn call( state.upstream.account(), request.model, started, - ledger.now(), + finished, ledger.outcome(outcome), usage, prefix, - state.upstream.cache_marks(request), + marks, ), ) let row = case row { @@ -351,6 +403,13 @@ fn call( None } } + // A turn call that succeeded is what a cache-warming ping would repeat, + // so the session keeps its request as sent, never a rebuilt one. + case kind, outcome { + ledger.Turn, Ok(_) -> + state.report_turn(request, prefix, marks, usage, started, finished) + _, _ -> Nil + } case outcome { Ok(turn) -> Ok(#(row, turn)) Error(error) -> Error(error) diff --git a/test/daemon/session_run_test.gleam b/test/daemon/session_run_test.gleam index ddb61bfa..385e0e6d 100644 --- a/test/daemon/session_run_test.gleam +++ b/test/daemon/session_run_test.gleam @@ -93,6 +93,7 @@ fn messages() -> session_run.Messages(Owner) { usage: fn(_, _, _) { Unused }, drain: fn(_, _) { Unused }, pin: fn(_, _, _) { Unused }, + sent: fn(_, _, _) { Unused }, finished: fn(_, _) { Unused }, collect: Unused, ) diff --git a/test/e2e/warm_test.py b/test/e2e/warm_test.py new file mode 100644 index 00000000..134c0da4 --- /dev/null +++ b/test/e2e/warm_test.py @@ -0,0 +1,217 @@ +"""The prompt-cache warmer: an idle session whose children still run re-sends +its last request with a tiny output budget before the provider cache would +expire. + +Catches: pings that rebuild the request instead of repeating the one the turn +sent (which would compact or summarize), ping ledger rows that do not carry +the turn's prefix identity, pings that commit or publish anything, warming +that never stops at its budget, warming that continues after the waking work +finished, and a submit that cannot get through while a ping holds the session. +None of this is observable from the transcript alone; the provider and the +ledger are the witnesses. +""" +import json +import threading +import time +import unittest + +from harness import Albedo, Provider, exclusive, text + +# The child's turn blocks here until the test releases it, so the parent sits +# idle on work that will wake it. +GATE = threading.Event() + +# A prefix big enough to matter: 4096 cached tokens against the 1024 floor. +PARENT_USAGE = {"input_tokens": 6000, "output_tokens": 4, + "input_tokens_details": {"cached_tokens": 4096}} +CHILD_USAGE = {"input_tokens": 300, "output_tokens": 4, + "input_tokens_details": {"cached_tokens": 100}} + +TASK = "hold the fort until released" + + +def reply(request): + if TASK in json.dumps(request): + GATE.wait(timeout=90) + return text("scout finished", usage=CHILD_USAGE) + return text("orchestrator reply", usage=PARENT_USAGE) + + +# A local cache-table override matching the fixture host: a 3s refresh TTL +# bought at 1.0x and read at 0.4x, so pings come ~2.7s apart and at most +# floor(1.0 / 0.4) = 2 of them pay for themselves. +CACHE_TTL = {"version": 1, "entries": [ + {"id": "fixture", "match": {"host": "127.0.0.1"}, "policy": "refresh", + "clock": "request", "tiers": [{"seconds": 3, "write": 1.0}], + "read": 0.4, "evidence": "measured"}]} + + +def wait_for(predicate, timeout=30): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + value = predicate() + if value: + return value + time.sleep(0.05) + raise AssertionError("warming never arrived") + + +class WarmTest(unittest.TestCase): + def setUp(self): + self.provider = Provider(reply) + providers = { + "orchestrator": {"baseUrl": self.provider.url + "/parent/v1", + "apiKey": "key", "model": "fixture-model", + "protocol": "responses"}, + "scout": {"baseUrl": self.provider.url + "/child/v1", + "apiKey": "key", "model": "scout-model", + "protocol": "responses"}, + } + self.app = Albedo(self.provider, protocol="responses", providers=providers) + self.app.__enter__() + self.addCleanup(self.provider.close) + self.addCleanup(self.app.__exit__, None, None, None) + # cache-ttl.json is daemon-home state the harness does not restore. + self.saved_ttl = self.read_ttl() + self.addCleanup(self.restore_ttl) + (self.app.home / "cache-ttl.json").write_text(json.dumps(CACHE_TTL)) + settings = json.loads((self.app.home / "extensions.json").read_text()) + settings["warm"] = {"enabled": True, "minCachedTokens": 1024} + (self.app.home / "extensions.json").write_text(json.dumps(settings)) + GATE.clear() + + def read_ttl(self): + path = self.app.home / "cache-ttl.json" + return path.read_bytes() if path.exists() else None + + def restore_ttl(self): + path = self.app.home / "cache-ttl.json" + if self.saved_ttl is None: + path.unlink(missing_ok=True) + else: + path.write_bytes(self.saved_ttl) + GATE.set() + + def api(self, path, body=None, method=None): + with self.app.api(path, body, method=method) as response: + return json.load(response) + + def requests(self, fragment): + return [record["request"] for record in self.provider.requests + if fragment in record["path"]] + + def pings(self, session_requests): + return [request for request in session_requests + if "max_output_tokens" in request] + + def rows(self, session): + with self.app.api(f"/sessions/{session}/ledger") as response: + return json.load(response)["rows"] + + def tree(self, session): + items, after = [], 0 + while True: + with self.app.api(f"/sessions/{session}/tree?after={after}&limit=100") \ + as response: + page = json.load(response) + items.extend(page["items"]) + if not page["hasMore"]: + return items + after = page["nextCursor"] + + def swarm(self): + """A parent whose turn has run while its scout is still working.""" + parent = self.app.session() + made = self.api(f"/sessions/{parent}/children", + {"name": "scout", "task": TASK, "model": "scout/scout-model"}) + child = made["member"]["session"] + wait_for(lambda: self.requests("/child/") or None) + self.app.prompt(parent, "run the swarm").close() + self.app.idle(parent) + return parent, child + + @exclusive + def test_pings_repeat_the_last_request_and_stop_at_the_budget(self): + parent, child = self.swarm() + before = self.tree(parent) + + def two_pings(): + found = self.pings(self.requests("/parent/")) + return found if len(found) == 2 else None + + pings = wait_for(two_pings) + turn = self.requests("/parent/")[0] + for ping in pings: + # A ping is the turn's request as sent, with only the output + # budget lowered — the Responses protocol's minimum. + self.assertEqual(ping, {**turn, "max_output_tokens": 16}) + + # The ledger rows carry the same prefix identity and cache marks as + # the turn they repeat, and never a transcript row of their own. + rows = self.rows(parent) + self.assertEqual([row["kind"] for row in rows], ["turn", "warm", "warm"]) + head = rows[0] + for row in rows[1:]: + self.assertEqual(row["headHash"], head["headHash"]) + self.assertEqual(row["inputs"], head["inputs"]) + self.assertEqual(row["replaced"], head["replaced"]) + self.assertEqual(row["projectionHash"], head["projectionHash"]) + self.assertEqual(row["cacheMarks"], head["cacheMarks"]) + self.assertEqual(row["cachedInputTokens"], 4096) + self.assertIsNone(row["seq"]) + self.assertEqual(row["outcome"], "ok") + self.assertEqual(row["profile"], "orchestrator") + self.assertEqual(row["model"], "fixture-model") + # A ping paid 4096 cached tokens: the hit the TTL model predicted. + self.assertGreaterEqual(rows[2]["startedMs"], rows[1]["startedMs"] + 2700) + + # The transcript is untouched by warming. + self.assertEqual(self.tree(parent), before) + + # The budget is spent: no third ping past another interval. + time.sleep(4) + self.assertEqual(len(self.pings(self.requests("/parent/"))), 2) + + # Once the scout finishes, nothing new is scheduled either. + GATE.set() + self.app.idle(child) + self.app.idle(parent) + time.sleep(4) + self.assertEqual(len(self.pings(self.requests("/parent/"))), 2) + # The scout's answer woke the parent with an ordinary turn. + turns = [request for request in self.requests("/parent/") + if "max_output_tokens" not in request] + self.assertEqual(len(turns), 2) + self.assertIn("scout finished", json.dumps(turns[1])) + + @exclusive + def test_a_submit_during_warming_answers_and_warming_ends_with_the_children(self): + parent, child = self.swarm() + + first = wait_for(lambda: self.pings(self.requests("/parent/")) or None) + turn = self.requests("/parent/")[0] + self.assertEqual(first[0], {**turn, "max_output_tokens": 16}) + + # A real submit while warming gets through and is answered normally. + self.app.prompt(parent, "real work now").close() + self.app.idle(parent) + self.assertIn("orchestrator reply", + [item["preview"] for item in self.tree(parent)]) + + # The work that kept the parent warm is done; warming ends with it. + GATE.set() + self.app.idle(child) + self.app.idle(parent) + settled = len(self.pings(self.requests("/parent/"))) + time.sleep(5) + self.assertEqual(len(self.pings(self.requests("/parent/"))), settled) + # Every ping that did go out repeated a turn's request, budget aside. + turns = [request for request in self.requests("/parent/") + if "max_output_tokens" not in request] + for ping in self.pings(self.requests("/parent/")): + self.assertIn(ping, [{**repeated, "max_output_tokens": 16} + for repeated in turns]) + + +if __name__ == "__main__": + unittest.main()