Skip to content

scheduler — top-level modules

scheduler.budget

Deterministic context budgeter (BUILD-SPEC §13).

A tokenizer + assembler — code, not a model. It composes each agent invocation to fit the active model's window with headroom for the reply, in the §13 priority order:

Constitution -> role -> retrieved RAG chunks -> history -> tool outputs -> task

When it won't fit, it trims in the §13 order: tighten retrieval depth first (the primary lever — retrieval is usually over-fetched), then compact history (sliding window, oldest first), then truncate tool outputs, and — if even the mandatory frame won't fit — flag escalate so the caller routes to a larger-window tier rather than silently dropping the Constitution. The Constitution, role, and task are never trimmed (Invariant 6); keeping the Constitution lean is therefore context-budget discipline, not just style.

Token counts are a deterministic ESTIMATE (no model in the loop); we bias slightly high and reserve reply headroom so the estimate is safe. A real tokenizer can be injected via Budgeter(estimator=...) without changing callers.

DEFAULT_REPLY_RESERVE = 1024 module-attribute

Estimator = Callable[[str], int] module-attribute

ContextParts dataclass

role instance-attribute

task instance-attribute

retrieved = () class-attribute instance-attribute

history = () class-attribute instance-attribute

tool_outputs = () class-attribute instance-attribute

constitution = None class-attribute instance-attribute

BudgetReport dataclass

window instance-attribute

reserve instance-attribute

used_tokens instance-attribute

retrieved_kept instance-attribute

retrieved_dropped instance-attribute

history_kept instance-attribute

history_dropped instance-attribute

tool_truncated instance-attribute

fits instance-attribute

escalate instance-attribute

BudgetedContext dataclass

messages instance-attribute

report instance-attribute

ConstitutionFrameError

Bases: RuntimeError

Refused: a caller tried to assemble a context whose outermost frame is not the canonical Constitution (Invariant 6/9) without a deliberate, visible override. The Constitution is a fixed point, not caller-substitutable content — closing the Threat-B assembly-logic seam.

Budgeter dataclass

window instance-attribute

reserve = DEFAULT_REPLY_RESERVE class-attribute instance-attribute

estimator = estimate_tokens class-attribute instance-attribute

assemble(parts, *, allow_constitution_override=False)

estimate_tokens(text)

Deterministic token estimate (~4 chars/token, rounded up). Stable across runs and biased slightly high so budgeting with reply headroom stays safe.

suggest_num_ctx(p95_tokens, *, headroom_frac=0.25, floor=2048, step=1024)

Right-size a role's load-time window from tracked usage (§13): p95 + headroom, rounded up to a step multiple, never below floor. The deterministic basis for the per-(model, role) window safe-lever (§14) — the OS agent may tune within bounds.

scheduler.cron

Cron / trough jobs — wiring the curator + dreaming agents into the supervisor (BUILD-SPEC §9, §13).

These are the §9 cognitive-tier jobs that "earn the big model": dreaming synthesis and curation. Both kinds already route to the synthesis tier (scheduler.router), and the supervisor's foreground gate keeps that tier out of foreground time (HEAVY_TIERS) — so these run trough-only, never concurrent with the owner's use (§13), and the two-slot loader's ceiling + swap discipline apply to them for free.

This module only builds the handlers and the enqueue helpers; the supervisor owns the loop. A full dreaming pass is run-to-completion here (the dreamer caps max_clusters); rewriting it as checkpointed per-cluster steps (the queue.checkpoint seam, roadmap §7) is a later refinement if a pass ever grows long enough to want to yield mid-way.

DREAM_KIND = 'dream' module-attribute

CURATE_KIND = 'curate' module-attribute

Handler = Callable[[Job], 'str | None'] module-attribute

dream_handler(dreamer)

curate_handler(curator)

cron_handlers(dreamer, curator)

The handler map for a supervisor that runs the trough jobs.

enqueue_dream(queue, router)

enqueue_curate(queue, router)

scheduler.interface

Wire the Ambassador / interface gateway into the scheduler (Track B / B1).

The interface substrate (edge gateway, core inbox, the Ambassador) was built but never scheduled — scheduler/cron.py had no reference to it. This module is that missing wiring, and it lives on the SCHEDULER side because it owns the queue: the Ambassador never imports the scheduler (it stays pure + testable), so the delegation seam (task → gate → queue) and the completed-result surfacing are injected from here as plain closures over the queue.

Two job kinds (scheduler/router.py): * ambassador — the inbox-drain tick: drive CoreInbox.process_once() (pinned tier, reactive) — for the scheduled/daemon path. * ambassador_task — the Ambassador's DELEGATED heavy work: a deep grounded answer on the synthesis tier, run trough-gated by the supervisor, result surfaced later.

ConversationRuntime is the in-process driver the CLI (scripts/talk.py) and the e2e test use: owner text → gateway → inbox → Ambassador → reply, with run_pending_tasks() standing in for the supervisor to complete delegated jobs between turns.

AMBASSADOR_KIND = 'ambassador' module-attribute

AMBASSADOR_TASK_KIND = 'ambassador_task' module-attribute

Handler = Callable[[Job], 'str | None'] module-attribute

ConversationRuntime dataclass

inbox instance-attribute

gateway instance-attribute

adapter instance-attribute

queue instance-attribute

task_handler instance-attribute

send(text, *, conversation='default')

Drive one full turn in-process: owner text → reply text (through the real gateway → filesystem handoff → core inbox → Ambassador → handoff → gateway path).

run_pending_tasks()

Complete any queued delegated tasks (the supervisor's job; the CLI calls this between turns so a delegated result is ready to surface on the owner's next message).

ambassador_inbox_handler(inbox)

Drain the core inbox each tick (the scheduled path). The per-message Ambassador reasoning happens inside process_once via the inbox's handler.

ambassador_task_handler(librarian)

Run one DELEGATED task: a deep grounded answer over the mirror. Returns the answer text, which the supervisor stores as the job result for the Ambassador to surface later.

enqueue_ambassador_inbox(queue, router)

Enqueue one inbox-drain tick (pinned tier, reactive).

build_task_delegation(queue, router, *, gate=None)

Return (delegate, pending_results) closures over the queue. delegate records the task in the gate (the routed-request ledger — visible, never auto-approved) and enqueues the delegated job; pending_results reads completed jobs back for the conversation. The Ambassador holds only these closures — never the queue itself.

build_conversation_runtime(config=None, *, server=None, embedder=None, store=None, drift=None)

Wire the full delegating Ambassador + inbox + gateway + queue for in-process use.

server/embedder/store are injectable (offline CLI + tests). The delegated-task librarian runs on the synthesis tier (heavy work) over the SAME store the Ambassador reads, so a delegated result lands where the next conversation can find it.

scheduler.presence

Foreground-presence detection (BUILD-SPEC §13 foreground check).

Heavy synthesis must not fire while the owner is actively using the machine. This reads the OS HID idle time (deterministic, no model) and reports whether the owner is present. The idle source is injectable so the scheduler is testable off-host and a non-macOS worker can supply its own probe.

Fail safe: if idle time is unknown, assume the owner IS present — never run heavy batch work on a blind guess.

DEFAULT_IDLE_THRESHOLD_S = 300.0 module-attribute

IdleProbe = Callable[[], 'float | None'] module-attribute

Presence dataclass

idle_probe = macos_idle_seconds class-attribute instance-attribute

threshold_s = DEFAULT_IDLE_THRESHOLD_S class-attribute instance-attribute

assume_present_when_unknown = True class-attribute instance-attribute

idle_seconds()

foreground_active()

True if the owner is actively using the machine (so heavy tiers are gated).

macos_idle_seconds()

Seconds since the last HID (keyboard/mouse) event, via ioreg. None if unavailable (e.g. not macOS, or the field is missing).

scheduler.queue

Durable job queue — the scheduler's heartbeat (BUILD-SPEC §8, §13; roadmap §7).

SQLite, WAL mode, single-writer by design: one supervisor owns this queue, so there is no write contention to reason about. The queue is the single safe serialization point — agents are config (re-composed per invocation from the stores), not OS processes, so "restoring" a job is cheap; the only heavyweight cost is a model load, which is what the scheduler is built to minimize.

Scheduling is cooperative and acts at job boundaries (roadmap §7): claim() selects the next job by priority, skipping tiers the caller says are currently blocked (the foreground gate), and — within the top-priority band — prefers a job that needs no model swap. A reactive escalation is just a high-priority job; it is dispatched next, never mid-generation. checkpoint/resume support long jobs (dreaming, curation) written as yielding steps.

PRIORITY_REACTIVE = 0 module-attribute

PRIORITY_INTERACTIVE = 10 module-attribute

PRIORITY_DEFAULT = 50 module-attribute

PRIORITY_BACKGROUND = 100 module-attribute

AgingPolicy dataclass

Anti-starvation aging (gap G6 — the liveness fix). A QUEUED job's EFFECTIVE priority improves (its number falls) the longer it waits, so background work (dreaming, curation) eventually outranks a perpetual stream of newer higher-priority jobs instead of starving under sustained foreground load — ◇ queued jobs eventually run.

Bounds, deliberately conservative: * a job that has waited < step_seconds ages zero steps, so NORMAL-load ordering is unchanged (jobs are usually claimed within seconds of enqueue); * aging never lifts a job above floor (default = INTERACTIVE), so an aged background job can come to tie with interactive work and win on FIFO, but can NEVER preempt a genuine REACTIVE escalation (a low-memory alarm must still go first — if those arrive perpetually the system is in crisis and background SHOULD wait).

step_seconds = 900.0 class-attribute instance-attribute

step = 10 class-attribute instance-attribute

floor = PRIORITY_INTERACTIVE class-attribute instance-attribute

Job dataclass

id instance-attribute

kind instance-attribute

tier instance-attribute

num_ctx instance-attribute

priority instance-attribute

state instance-attribute

payload instance-attribute

result instance-attribute

error instance-attribute

attempts instance-attribute

checkpoint instance-attribute

created_at instance-attribute

started_at instance-attribute

finished_at instance-attribute

load_key property

The (tier, window) that must be resident to run this job. Changing either forces a model reload (§13), so the supervisor batches jobs sharing a load_key.

JobQueue dataclass

path instance-attribute

aging = field(default_factory=AgingPolicy) class-attribute instance-attribute

enqueue(kind, tier, num_ctx, *, priority=PRIORITY_DEFAULT, payload=None)

claim(*, loaded_key=None, blocked_tiers=frozenset(), now=None)

Select + mark RUNNING the next eligible job (§13 policy): highest EFFECTIVE priority first (priority + anti-starvation aging, gap G6); within the top band prefer the job needing no model swap (matching loaded_key), then FIFO; skip tiers in blocked_tiers (the foreground gate) — they stay QUEUED and are revisited once the block clears. Returns None if nothing is runnable now.

complete(job_id, result=None)

fail(job_id, error)

defer(job_id, reason)

Park a job that cannot run under current conditions (e.g. ceiling breach). Not re-selected until revive_deferred() puts it back when conditions change.

revive_deferred()

Return deferred jobs to QUEUED (call when conditions change, e.g. RAM freed).

checkpoint(job_id, token)

Persist a resume token for a checkpointed-step job, then re-queue it so the next unit is dispatched at a job boundary (cooperative yielding, roadmap §7).

get(job_id)

list(state=None)

depth()

Number of jobs waiting to run (queue depth — a vital, §8).

counts()

close()

scheduler.router

Router + watchdog — the pinned tiny model's role, done in rules first (BUILD-SPEC §9, §13).

RULES FIRST (roadmap §8): the scheduler is rule-capable by design and the tiny router model is an enhancement, not a single point of failure — so Phase 3 ships a deterministic rule-based router with a seam for the model router and a fall-back-to-rules path. The router decides role/tier/window; deterministic code acts (loads, assembles, dispatches) — model advises, code acts (Invariant 3).

The watchdog reads system vitals (the reactive tier, §9) and raises flags only when a threshold is crossed; those become high-priority jobs the supervisor dispatches next, at the following job boundary (roadmap §7) — never a mid-generation interrupt.

Plan dataclass

kind instance-attribute

tier instance-attribute

num_ctx instance-attribute

priority instance-attribute

Flag dataclass

metric instance-attribute

value instance-attribute

threshold instance-attribute

note instance-attribute

Router dataclass

config instance-attribute

tier_for(kind)

plan(kind, *, priority=None)

Resolve a job kind to (tier, window, priority) from the rules + the model lineup. The window is the model's configured load-time num_ctx (§13); same-window jobs batch together to avoid reloads.

Watchdog dataclass

reader instance-attribute

min_available_gb = 2.0 class-attribute instance-attribute

check()

Read the latest vitals and return any crossed thresholds. Deterministic; escalates to a model only by enqueuing a job, which the supervisor dispatches by priority.

scheduler.supervisor

The supervisor — one loop owns the queue and the worker slot (BUILD-SPEC §13; roadmap §7).

Cooperative, job-boundary scheduling: 1. claim the next eligible job (priority; swap-avoidance within a priority band; heavy tiers gated while the owner is present — the foreground check); 2. make its (tier, window) resident via the two-slot loader, which refuses any load that would breach the RAM ceiling (Invariant 8) — such a job is deferred, not crashed; 3. run its handler to completion (or one checkpointed step), counting worker swaps (the pinned router doing interstitial work never evicts the worker, roadmap §7); 4. record vitals (queue depth, model-load time) and repeat.

A reactive escalation is simply a high-priority job; it is dispatched at the next boundary, never as a mid-generation interrupt. A handler that raises must not take down the loop.

HEAVY_TIERS = frozenset({'synthesis', 'stretch'}) module-attribute

Handler = Callable[[Job], 'str | None'] module-attribute

Supervisor dataclass

queue instance-attribute

loader instance-attribute

handlers instance-attribute

presence = field(default_factory=Presence) class-attribute instance-attribute

telemetry = None class-attribute instance-attribute

secrets = None class-attribute instance-attribute

warm = True class-attribute instance-attribute

swaps = 0 class-attribute instance-attribute

mint_token(role, ttl='10m')

Mint an ephemeral token scoped to role's policy (vault-runtime-auth.md §2). Returns the whole MintedToken: the supervisor passes .token to the agent (Phase 5) and records .accessor in the action's attestation (the Step-5 join) — it holds minting authority only, never reading the secret it mints a token for; that happens later when the agent itself calls get_secret(name, token=...).

blocked_tiers()

tick()

Dispatch at most one job. Returns False when nothing is runnable right now.

run(*, max_ticks=None)

Drain the queue cooperatively. Returns the number of jobs dispatched. Stops when nothing is runnable (e.g. only heavy jobs remain while the owner is present).

scheduler.vault_sync

Wire the vault watcher + incremental re-ingest into the scheduler (vault-sync task).

The watcher (core) only signals; this scheduler-side module turns that signal into a durable background vault_sync job and provides the handler that runs the idempotent re-ingest. So all store mutation happens on the single supervisor writer (the queue's discipline), and the core watcher stays free of any scheduler import (clean layering: scheduler depends on core, never the reverse).

vault_sync is routed to the pinned tier (the router): it needs no chat model — it calls the embedder directly — so making it "resident" is a no-op and the worker slot is never evicted. It runs at BACKGROUND priority (yields to interactive/reactive work) but is NOT in HEAVY_TIERS, so a note saved mid-session is re-ingested promptly rather than waiting for a trough (the owner may want to query what they just wrote).

VAULT_SYNC_KIND = 'vault_sync' module-attribute

Handler = Callable[[Job], 'str | None'] module-attribute

vault_sync_handler(sync)

enqueue_vault_sync(queue, router)

Enqueue one background re-ingest. Coalescing happens upstream (the watcher debounce); duplicate jobs are harmless because rescan() is idempotent.

build_vault_watcher(queue, router, config=None)

A watcher whose on_change enqueues a background vault_sync job. Call .start() to run.

The supervisor must have the vault_sync handler registered (see vault_sync_handler) to actually process the enqueued jobs.