"""Reference solution — The Agent Kernel. A runtime that hosts agents the way an OS hosts processes: bounded execution, explicit lifecycle, externalized state, tiered memory, and a recorded execution chain. Everything is deterministic. The four ambient things a kernel touches are all injected: the model (a ``Policy`` callable), the clock (``now``), the tool registry, and the summarizer used for compaction. There is no randomness and no wall clock. Run ``python solution.py`` for the worked example. """ from __future__ import annotations import hashlib from dataclasses import dataclass, field, replace from enum import Enum from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple # ====================================================================================== # 1. Lifecycle # ====================================================================================== class RunState(str, Enum): CREATED = "created" PLANNING = "planning" ACTING = "acting" WAITING_INPUT = "waiting_input" SUSPENDED = "suspended" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" class Event(str, Enum): START = "start" PROPOSE = "propose" # planning -> acting (the model proposed a tool call) OBSERVE = "observe" # acting -> planning (the tool returned) NEED_INPUT = "need_input" # -> waiting_input (human-in-the-loop) RESUME_INPUT = "resume_input" SUSPEND = "suspend" # -> suspended (checkpointed, evictable) RESUME = "resume" FINISH = "finish" FAIL = "fail" CANCEL = "cancel" TERMINAL_STATES: frozenset = frozenset( {RunState.COMPLETED, RunState.FAILED, RunState.CANCELLED} ) #: The complete transition table. Anything not listed is illegal — a kernel that allows #: an undeclared transition cannot be reasoned about, and "it worked in testing" is not #: an argument you can make to an auditor. TRANSITIONS: Mapping[Tuple[RunState, Event], RunState] = { (RunState.CREATED, Event.START): RunState.PLANNING, (RunState.CREATED, Event.CANCEL): RunState.CANCELLED, (RunState.PLANNING, Event.PROPOSE): RunState.ACTING, (RunState.PLANNING, Event.NEED_INPUT): RunState.WAITING_INPUT, (RunState.PLANNING, Event.SUSPEND): RunState.SUSPENDED, (RunState.PLANNING, Event.FINISH): RunState.COMPLETED, (RunState.PLANNING, Event.FAIL): RunState.FAILED, (RunState.PLANNING, Event.CANCEL): RunState.CANCELLED, (RunState.ACTING, Event.OBSERVE): RunState.PLANNING, (RunState.ACTING, Event.NEED_INPUT): RunState.WAITING_INPUT, (RunState.ACTING, Event.SUSPEND): RunState.SUSPENDED, (RunState.ACTING, Event.FAIL): RunState.FAILED, (RunState.ACTING, Event.CANCEL): RunState.CANCELLED, (RunState.WAITING_INPUT, Event.RESUME_INPUT): RunState.PLANNING, (RunState.WAITING_INPUT, Event.SUSPEND): RunState.SUSPENDED, (RunState.WAITING_INPUT, Event.CANCEL): RunState.CANCELLED, (RunState.WAITING_INPUT, Event.FAIL): RunState.FAILED, (RunState.SUSPENDED, Event.RESUME): RunState.PLANNING, (RunState.SUSPENDED, Event.CANCEL): RunState.CANCELLED, (RunState.SUSPENDED, Event.FAIL): RunState.FAILED, } class IllegalTransition(RuntimeError): def __init__(self, state: RunState, event: Event) -> None: super().__init__(f"illegal transition: {state.value} --{event.value}-->") self.state = state self.event = event def transition(state: RunState, event: Event) -> RunState: """Apply one lifecycle event. Unknown pair -> :class:`IllegalTransition`.""" if state in TERMINAL_STATES: raise IllegalTransition(state, event) try: return TRANSITIONS[(state, event)] except KeyError: raise IllegalTransition(state, event) from None # ====================================================================================== # 2. Memory # ====================================================================================== def estimate_tokens(text: str) -> int: """A deterministic stand-in for a tokenizer: ~4 characters per token, rounded up. Real kernels call the model's tokenizer. What matters here is that the budget is computed from a *function of the text*, not guessed — and that the same text always costs the same, so tests are exact. """ if not text: return 0 return (len(text) + 3) // 4 @dataclass(frozen=True) class Step: """One turn of the loop, as recorded in the scratchpad and the execution chain.""" index: int thought: str tool: Optional[str] = None arguments: Tuple[Tuple[str, str], ...] = () observation: Optional[str] = None error: Optional[str] = None tokens_in: int = 0 tokens_out: int = 0 started_at: float = 0.0 ended_at: float = 0.0 def render(self) -> str: parts = [f"[{self.index}] thought: {self.thought}"] if self.tool is not None: args = ", ".join(f"{k}={v}" for k, v in self.arguments) parts.append(f" action: {self.tool}({args})") if self.observation is not None: parts.append(f" observation: {self.observation}") if self.error is not None: parts.append(f" error: {self.error}") return "\n".join(parts) @dataclass class Scratchpad: """Short-term working memory: the accumulating thought/action/observation record. This is the object whose growth is quadratic in step count, so it owns compaction. """ max_tokens: int summarize: Callable[[Sequence[Step]], str] keep_recent: int = 2 steps: List[Step] = field(default_factory=list) summary: str = "" compactions: int = 0 def append(self, step: Step) -> None: self.steps.append(step) def token_count(self) -> int: total = estimate_tokens(self.summary) if self.summary else 0 for step in self.steps: total += estimate_tokens(step.render()) return total def render(self) -> str: blocks = [] if self.summary: blocks.append(f"[summary of earlier steps]\n{self.summary}") blocks.extend(step.render() for step in self.steps) return "\n".join(blocks) def compact_if_needed(self) -> bool: """Fold everything but the last ``keep_recent`` steps into a summary. Returns True if a compaction happened. Compaction is *lossy on purpose*: the alternative is a context window that grows until the run dies, and a bill that grows quadratically. The full detail survives in the execution chain, which is the audit artifact — the scratchpad is only the model's working set. """ if self.token_count() <= self.max_tokens: return False if len(self.steps) <= self.keep_recent: return False cutoff = len(self.steps) - self.keep_recent folded, self.steps = self.steps[:cutoff], self.steps[cutoff:] new_summary = self.summarize(folded) self.summary = (self.summary + " " + new_summary).strip() if self.summary else new_summary self.compactions += 1 return True @dataclass(frozen=True) class Fact: """One item of long-term semantic memory.""" key: str value: str scope: str # "user" | "tenant" | "app" owner: str # user id / tenant id / app id, matching the scope tags: Tuple[str, ...] = () class SemanticMemory: """Durable facts, partitioned by scope and owner. The partition key is not optional. A memory store that can return another tenant's fact is the same class of defect as a shared vector index with no namespace. """ def __init__(self) -> None: self._facts: Dict[Tuple[str, str, str], Fact] = {} def put(self, fact: Fact) -> None: if fact.scope not in ("user", "tenant", "app"): raise ValueError(f"unknown scope: {fact.scope!r}") self._facts[(fact.scope, fact.owner, fact.key)] = fact def get(self, scope: str, owner: str, key: str) -> Optional[Fact]: return self._facts.get((scope, owner, key)) def search(self, *, scopes: Mapping[str, str], tags: Sequence[str], limit: int = 5) -> List[Fact]: """Return facts visible to the caller, ranked by tag overlap then key. ``scopes`` maps a scope name to the caller's owner id, e.g. ``{"user": "u-1", "tenant": "wholesale"}``. Facts in scopes the caller does not hold are invisible — not filtered after ranking, *never retrieved*. """ wanted = set(tags) visible = [ f for f in self._facts.values() if scopes.get(f.scope) == f.owner ] scored = [(len(wanted & set(f.tags)), f) for f in visible] scored = [(score, f) for score, f in scored if score > 0 or not wanted] scored.sort(key=lambda pair: (-pair[0], pair[1].key)) return [f for _, f in scored[:limit]] @dataclass(frozen=True) class Episode: """One completed task, remembered for next time.""" episode_id: str session_id: str goal: str outcome: str # "completed" | "failed" | "cancelled" steps: int tags: Tuple[str, ...] = () lesson: str = "" class EpisodicMemory: """Append-only memory of *what happened*, indexed by tag overlap and recency. Distinguished from semantic memory by what it stores: episodes, not facts. It is what lets an agent say "last time this vendor's invoice failed validation it was the date format" instead of rediscovering it. """ def __init__(self) -> None: self._episodes: List[Episode] = [] def record(self, episode: Episode) -> None: self._episodes.append(episode) def recall(self, tags: Sequence[str], limit: int = 3) -> List[Episode]: wanted = set(tags) scored = [ (len(wanted & set(e.tags)), i, e) for i, e in enumerate(self._episodes) ] scored = [t for t in scored if t[0] > 0] # Highest tag overlap first, then most recent (higher index) first. scored.sort(key=lambda t: (-t[0], -t[1])) return [e for _, _, e in scored[:limit]] def __len__(self) -> int: return len(self._episodes) # ====================================================================================== # 3. Session state, checkpoints and optimistic concurrency # ====================================================================================== class ConcurrentModification(RuntimeError): """Two workers tried to advance the same session. Exactly one may win.""" @dataclass(frozen=True) class SessionSnapshot: """An immutable checkpoint of everything needed to resume a run.""" session_id: str tenant: str user_id: str goal: str state: RunState version: int steps: Tuple[Step, ...] summary: str tokens_used: int cost_micros: int pending_question: Optional[str] = None created_at: float = 0.0 updated_at: float = 0.0 class SessionStore: """Externalized session state with optimistic concurrency control. Session state lives *here*, not in a worker's memory. That is what makes affinity an optimization rather than a requirement, and it is what makes a pod restart a non-event instead of a lost conversation. """ def __init__(self) -> None: self._sessions: Dict[str, SessionSnapshot] = {} def create(self, snapshot: SessionSnapshot) -> SessionSnapshot: if snapshot.session_id in self._sessions: raise KeyError(f"session already exists: {snapshot.session_id}") stored = replace(snapshot, version=1) self._sessions[stored.session_id] = stored return stored def load(self, session_id: str) -> SessionSnapshot: try: return self._sessions[session_id] except KeyError: raise KeyError(f"no such session: {session_id}") from None def save(self, snapshot: SessionSnapshot, *, expected_version: int) -> SessionSnapshot: """Compare-and-swap on ``version``. Stale write -> :class:`ConcurrentModification`.""" current = self.load(snapshot.session_id) if current.version != expected_version: raise ConcurrentModification( f"session {snapshot.session_id}: expected version {expected_version}, " f"store has {current.version}" ) stored = replace(snapshot, version=expected_version + 1) self._sessions[stored.session_id] = stored return stored def exists(self, session_id: str) -> bool: return session_id in self._sessions # ====================================================================================== # 4. Session affinity # ====================================================================================== def _hash_to_int(value: str) -> int: """Deterministic 64-bit hash. NOT Python's ``hash()`` — that is salted per process, so a ring built with it reshuffles on every restart.""" digest = hashlib.blake2b(value.encode("utf-8"), digest_size=8).digest() return int.from_bytes(digest, "big") class AffinityRouter: """Consistent-hash routing of sessions to replicas, with draining. Affinity is an *optimization*: it keeps a session's warm caches and open connections on one replica. Because state is externalized, losing affinity costs a cache miss, not a conversation. """ def __init__(self, replicas: Iterable[str], *, virtual_nodes: int = 64) -> None: if virtual_nodes <= 0: raise ValueError("virtual_nodes must be > 0") self.virtual_nodes = virtual_nodes self._replicas: List[str] = [] self._draining: set = set() self._ring: List[Tuple[int, str]] = [] for replica in replicas: self.add(replica) def _rebuild(self) -> None: ring: List[Tuple[int, str]] = [] for replica in self._replicas: for i in range(self.virtual_nodes): ring.append((_hash_to_int(f"{replica}#{i}"), replica)) ring.sort() self._ring = ring def add(self, replica: str) -> None: if replica in self._replicas: return self._replicas.append(replica) self._replicas.sort() self._draining.discard(replica) self._rebuild() def remove(self, replica: str) -> None: if replica not in self._replicas: raise KeyError(f"unknown replica: {replica}") self._replicas.remove(replica) self._draining.discard(replica) self._rebuild() def drain(self, replica: str) -> None: """Stop routing NEW sessions here; existing ones keep their placement until they move. This is what a rolling deploy needs and what naive removal breaks.""" if replica not in self._replicas: raise KeyError(f"unknown replica: {replica}") self._draining.add(replica) def route(self, session_id: str) -> str: if not self._ring: raise RuntimeError("no replicas available") available = [r for r in self._replicas if r not in self._draining] if not available: raise RuntimeError("all replicas are draining") point = _hash_to_int(session_id) # Walk the ring clockwise from the session's point; skip draining replicas. candidates = [entry for entry in self._ring if entry[0] >= point] + self._ring for _, replica in candidates: if replica not in self._draining: return replica raise RuntimeError("no replicas available") # pragma: no cover - unreachable def distribution(self, session_ids: Sequence[str]) -> Dict[str, int]: counts: Dict[str, int] = {r: 0 for r in self._replicas if r not in self._draining} for sid in session_ids: counts[self.route(sid)] += 1 return counts # ====================================================================================== # 5. Tools and budgets # ====================================================================================== @dataclass(frozen=True) class ToolResult: ok: bool output: str tokens: int = 0 cost_micros: int = 0 Tool = Callable[[Mapping[str, str]], ToolResult] @dataclass(frozen=True) class Budgets: """The kernel's quotas. An agent that exceeds one is stopped BY THE KERNEL — this is not the agent author's responsibility, exactly as a process does not get to decide its own memory limit.""" max_steps: int = 12 max_tokens: int = 20_000 max_cost_micros: int = 500_000 deadline_seconds: float = 30.0 def __post_init__(self) -> None: for name in ("max_steps", "max_tokens", "max_cost_micros"): if getattr(self, name) <= 0: raise ValueError(f"{name} must be > 0") if self.deadline_seconds <= 0: raise ValueError("deadline_seconds must be > 0") @dataclass(frozen=True) class BudgetBreach: kind: str # "steps" | "tokens" | "cost" | "deadline" detail: str # ====================================================================================== # 6. The model policy and its decisions # ====================================================================================== @dataclass(frozen=True) class Decision: """What the injected 'model' returned. A pure function of the rendered scratchpad.""" kind: str # "act" | "finish" | "ask" thought: str tool: Optional[str] = None arguments: Tuple[Tuple[str, str], ...] = () answer: Optional[str] = None question: Optional[str] = None tokens_in: int = 0 tokens_out: int = 0 def __post_init__(self) -> None: if self.kind not in ("act", "finish", "ask"): raise ValueError(f"unknown decision kind: {self.kind!r}") if self.kind == "act" and not self.tool: raise ValueError("an 'act' decision must name a tool") if self.kind == "ask" and not self.question: raise ValueError("an 'ask' decision must carry a question") Policy = Callable[[str], Decision] # ====================================================================================== # 7. The kernel # ====================================================================================== @dataclass class RunResult: session_id: str state: RunState answer: Optional[str] steps: Tuple[Step, ...] breach: Optional[BudgetBreach] tokens_used: int cost_micros: int compactions: int question: Optional[str] = None @property def succeeded(self) -> bool: return self.state is RunState.COMPLETED class AgentKernel: """Bounded, checkpointed, resumable execution of a model-driven loop. The kernel owns five things the agent author must not: the lifecycle, the budgets, the state, the memory tiers, and the execution chain. """ def __init__( self, *, store: SessionStore, tools: Mapping[str, Tool], policy: Policy, now: Callable[[], float], budgets: Budgets = Budgets(), summarize: Optional[Callable[[Sequence[Step]], str]] = None, scratchpad_max_tokens: int = 400, episodic: Optional[EpisodicMemory] = None, semantic: Optional[SemanticMemory] = None, ) -> None: self.store = store self.tools = dict(tools) self.policy = policy self.now = now self.budgets = budgets self.summarize = summarize or default_summarizer self.scratchpad_max_tokens = scratchpad_max_tokens self.episodic = episodic if episodic is not None else EpisodicMemory() self.semantic = semantic if semantic is not None else SemanticMemory() # -- session management ------------------------------------------------------- def create_session(self, *, session_id: str, tenant: str, user_id: str, goal: str) -> SessionSnapshot: return self.store.create(SessionSnapshot( session_id=session_id, tenant=tenant, user_id=user_id, goal=goal, state=RunState.CREATED, version=0, steps=(), summary="", tokens_used=0, cost_micros=0, created_at=self.now(), updated_at=self.now(), )) # -- the loop ----------------------------------------------------------------- def run(self, session_id: str, *, resume_answer: Optional[str] = None) -> RunResult: snapshot = self.store.load(session_id) started = self.now() deadline = started + self.budgets.deadline_seconds state = snapshot.state if state is RunState.CREATED: state = transition(state, Event.START) elif state is RunState.SUSPENDED: state = transition(state, Event.RESUME) elif state is RunState.WAITING_INPUT: if resume_answer is None: raise ValueError("session is waiting for input; pass resume_answer") state = transition(state, Event.RESUME_INPUT) elif state in TERMINAL_STATES: raise IllegalTransition(state, Event.START) pad = Scratchpad( max_tokens=self.scratchpad_max_tokens, summarize=self.summarize, steps=list(snapshot.steps), summary=snapshot.summary, ) tokens_used = snapshot.tokens_used cost_micros = snapshot.cost_micros version = snapshot.version breach: Optional[BudgetBreach] = None answer: Optional[str] = None question: Optional[str] = None if resume_answer is not None: pad.append(Step(index=_next_index(snapshot), thought="human provided input", observation=resume_answer, started_at=self.now(), ended_at=self.now())) while True: # --- budget checks happen BEFORE the expensive call, every iteration --- step_no = _next_index(snapshot) if not pad.steps else pad.steps[-1].index + 1 if step_no > self.budgets.max_steps: breach = BudgetBreach("steps", f"step {step_no} > max_steps={self.budgets.max_steps}") state = transition(state, Event.FAIL) break if tokens_used > self.budgets.max_tokens: breach = BudgetBreach("tokens", f"{tokens_used} > max_tokens={self.budgets.max_tokens}") state = transition(state, Event.FAIL) break if cost_micros > self.budgets.max_cost_micros: breach = BudgetBreach("cost", f"{cost_micros} > max_cost_micros={self.budgets.max_cost_micros}") state = transition(state, Event.FAIL) break if self.now() >= deadline: breach = BudgetBreach("deadline", f"exceeded {self.budgets.deadline_seconds}s") state = transition(state, Event.FAIL) break decision = self.policy(pad.render()) tokens_used += decision.tokens_in + decision.tokens_out started_at = self.now() if decision.kind == "finish": pad.append(Step(index=step_no, thought=decision.thought, observation=decision.answer, tokens_in=decision.tokens_in, tokens_out=decision.tokens_out, started_at=started_at, ended_at=self.now())) answer = decision.answer state = transition(state, Event.FINISH) break if decision.kind == "ask": pad.append(Step(index=step_no, thought=decision.thought, observation=f"awaiting human input: {decision.question}", tokens_in=decision.tokens_in, tokens_out=decision.tokens_out, started_at=started_at, ended_at=self.now())) question = decision.question state = transition(state, Event.NEED_INPUT) break # --- act --- state = transition(state, Event.PROPOSE) tool = self.tools.get(decision.tool or "") if tool is None: step = Step(index=step_no, thought=decision.thought, tool=decision.tool, arguments=decision.arguments, error=f"unknown tool: {decision.tool!r}", tokens_in=decision.tokens_in, tokens_out=decision.tokens_out, started_at=started_at, ended_at=self.now()) pad.append(step) # An unknown tool is a *recoverable* error: the observation goes back to # the model so it can correct itself. It is not a kernel failure. state = transition(state, Event.OBSERVE) else: result = tool(dict(decision.arguments)) tokens_used += result.tokens cost_micros += result.cost_micros step = Step(index=step_no, thought=decision.thought, tool=decision.tool, arguments=decision.arguments, observation=result.output if result.ok else None, error=None if result.ok else result.output, tokens_in=decision.tokens_in, tokens_out=decision.tokens_out, started_at=started_at, ended_at=self.now()) pad.append(step) state = transition(state, Event.OBSERVE) pad.compact_if_needed() snapshot, version = self._checkpoint(snapshot, state, pad, tokens_used, cost_micros, version, None) snapshot, version = self._checkpoint(snapshot, state, pad, tokens_used, cost_micros, version, question) if state in TERMINAL_STATES: self.episodic.record(Episode( episode_id=f"{snapshot.session_id}#{len(self.episodic) + 1}", session_id=snapshot.session_id, goal=snapshot.goal, outcome=state.value, steps=len(pad.steps), tags=tuple(sorted({s.tool for s in pad.steps if s.tool})), lesson=breach.detail if breach else "", )) return RunResult( session_id=snapshot.session_id, state=state, answer=answer, steps=tuple(pad.steps), breach=breach, tokens_used=tokens_used, cost_micros=cost_micros, compactions=pad.compactions, question=question, ) def _checkpoint(self, snapshot, state, pad, tokens_used, cost_micros, version, question): updated = SessionSnapshot( session_id=snapshot.session_id, tenant=snapshot.tenant, user_id=snapshot.user_id, goal=snapshot.goal, state=state, version=version, steps=tuple(pad.steps), summary=pad.summary, tokens_used=tokens_used, cost_micros=cost_micros, pending_question=question, created_at=snapshot.created_at, updated_at=self.now(), ) stored = self.store.save(updated, expected_version=version) return stored, stored.version # -- introspection ------------------------------------------------------------- def execution_chain(self, session_id: str) -> List[Mapping[str, object]]: """The audit artifact: every step with identity, timing and outcome. Note this reads from the STORE, not from the scratchpad — compaction is lossy for the model but never for the record. """ snapshot = self.store.load(session_id) return [ { "session_id": snapshot.session_id, "tenant": snapshot.tenant, "user_id": snapshot.user_id, "step": s.index, "tool": s.tool, "arguments": dict(s.arguments), "ok": s.error is None, "error": s.error, "tokens": s.tokens_in + s.tokens_out, "duration": round(s.ended_at - s.started_at, 6), } for s in snapshot.steps ] def _next_index(snapshot: SessionSnapshot) -> int: return (snapshot.steps[-1].index + 1) if snapshot.steps else 1 def default_summarizer(steps: Sequence[Step]) -> str: """Deterministic compaction: name the tools used and the last observation. A real kernel calls a model here. The lesson survives the substitution: compaction is a *policy* the kernel owns, and its quality is a tunable that trades cost against the risk of forgetting something load-bearing. """ tools = [s.tool for s in steps if s.tool] last = next((s.observation for s in reversed(steps) if s.observation), "") return f"ran {len(steps)} step(s) using {sorted(set(tools)) or 'no tools'}; last result: {last}" # ====================================================================================== # Worked example # ====================================================================================== def _demo_policy_factory() -> Policy: """A scripted 'model': a pure function of the rendered scratchpad.""" def policy(rendered: str) -> Decision: if "lookup_payment" not in rendered: return Decision("act", "I need the payment record first.", tool="lookup_payment", arguments=(("reference", "PMT-771"),), tokens_in=estimate_tokens(rendered), tokens_out=20) if "check_sanctions" not in rendered: return Decision("act", "Now screen the beneficiary.", tool="check_sanctions", arguments=(("name", "Acme Trading FZE"),), tokens_in=estimate_tokens(rendered), tokens_out=18) if "unknown tool" not in rendered: return Decision("act", "Try a tool that does not exist.", tool="teleport_funds", arguments=(), tokens_in=estimate_tokens(rendered), tokens_out=12) return Decision("finish", "I have enough to answer.", answer="PMT-771 is held pending sanctions review of Acme Trading FZE.", tokens_in=estimate_tokens(rendered), tokens_out=30) return policy def main() -> None: # pragma: no cover - narrative output ticks = iter(range(0, 10_000)) clock = lambda: float(next(ticks)) / 10.0 # 100 ms per call, deterministic tools: Dict[str, Tool] = { "lookup_payment": lambda a: ToolResult(True, f"payment {a.get('reference')} status=HELD amount=250000 AED", tokens=40, cost_micros=200), "check_sanctions": lambda a: ToolResult(True, f"{a.get('name')}: 1 possible match, score 0.83", tokens=35, cost_micros=180), } store = SessionStore() kernel = AgentKernel(store=store, tools=tools, policy=_demo_policy_factory(), now=clock, budgets=Budgets(max_steps=8, deadline_seconds=1000.0), scratchpad_max_tokens=120) kernel.create_session(session_id="s-1", tenant="wholesale", user_id="u-42", goal="Why is payment PMT-771 held?") print("=" * 78) print("1. A RUN") print("=" * 78) result = kernel.run("s-1") print(f" state={result.state.value} steps={len(result.steps)} " f"tokens={result.tokens_used} cost={result.cost_micros}µ$ " f"compactions={result.compactions}") print(f" answer: {result.answer}") print() for step in result.steps: print(" " + step.render().replace("\n", "\n ")) print() print("=" * 78) print("2. THE EXECUTION CHAIN (the audit artifact)") print("=" * 78) for row in kernel.execution_chain("s-1"): print(f" step={row['step']} tool={row['tool']!r} ok={row['ok']} " f"tokens={row['tokens']} dur={row['duration']}s error={row['error']!r}") print() print("=" * 78) print("3. LIFECYCLE: THE ILLEGAL TRANSITION") print("=" * 78) try: transition(RunState.COMPLETED, Event.PROPOSE) except IllegalTransition as exc: print(f" {exc}") print(f" legal from PLANNING: " f"{sorted(e.value for (s, e) in TRANSITIONS if s is RunState.PLANNING)}") print() print("=" * 78) print("4. HUMAN-IN-THE-LOOP: ASK, SUSPEND, RESUME") print("=" * 78) asked = {"done": False} def hitl_policy(rendered: str) -> Decision: if not asked["done"] and "awaiting human input" not in rendered: asked["done"] = True return Decision("ask", "This releases funds; I need approval.", question="Approve release of AED 250,000 to Acme Trading FZE?", tokens_in=10, tokens_out=10) return Decision("finish", "Approved.", answer="Released.", tokens_in=10, tokens_out=5) k2 = AgentKernel(store=store, tools=tools, policy=hitl_policy, now=clock) k2.create_session(session_id="s-2", tenant="wholesale", user_id="u-42", goal="Release PMT-771") r1 = k2.run("s-2") print(f" after first run: state={r1.state.value} question={r1.question!r}") r2 = k2.run("s-2", resume_answer="approved by u-99 and u-100") print(f" after resume : state={r2.state.value} answer={r2.answer!r} " f"steps={len(r2.steps)}") print() print("=" * 78) print("5. BUDGETS: THE KERNEL STOPS A LOOPING AGENT") print("=" * 78) looper = lambda rendered: Decision("act", "again", tool="lookup_payment", arguments=(("reference", "X"),), tokens_in=5, tokens_out=5) k3 = AgentKernel(store=store, tools=tools, policy=looper, now=clock, budgets=Budgets(max_steps=4, deadline_seconds=1000.0)) k3.create_session(session_id="s-3", tenant="retail", user_id="u-7", goal="loop forever") r3 = k3.run("s-3") print(f" state={r3.state.value} breach={r3.breach}") print(f" steps recorded: {len(r3.steps)} (the kernel stopped it, not the agent)") print() print("=" * 78) print("6. SESSION AFFINITY") print("=" * 78) router = AffinityRouter(["pod-a", "pod-b", "pod-c"]) sessions = [f"s-{i}" for i in range(300)] before = {s: router.route(s) for s in sessions} print(f" distribution over 3 pods: {router.distribution(sessions)}") router.add("pod-d") after = {s: router.route(s) for s in sessions} moved = sum(1 for s in sessions if before[s] != after[s]) print(f" after adding pod-d : {router.distribution(sessions)}") print(f" sessions that moved : {moved}/300 ({moved / 3:.1f}%) " f"— naive modulo would move ~75%") router.drain("pod-a") print(f" after draining pod-a : {router.distribution(sessions)}") print() print("=" * 78) print("7. OPTIMISTIC CONCURRENCY: TWO WORKERS, ONE SESSION") print("=" * 78) snap = store.load("s-1") print(f" current version: {snap.version}") store.save(replace(snap, tokens_used=snap.tokens_used + 1), expected_version=snap.version) try: store.save(replace(snap, tokens_used=snap.tokens_used + 2), expected_version=snap.version) except ConcurrentModification as exc: print(f" second writer rejected: {exc}") print() print("=" * 78) print("8. MEMORY TIERS") print("=" * 78) sem = SemanticMemory() sem.put(Fact("rm_owner", "Layla Al Mansouri", "tenant", "wholesale", ("acme", "relationship"))) sem.put(Fact("risk_appetite", "conservative", "tenant", "retail", ("policy",))) visible = sem.search(scopes={"tenant": "wholesale", "user": "u-42"}, tags=["acme"]) print(f" wholesale caller sees : {[f.key for f in visible]}") invisible = sem.search(scopes={"tenant": "wholesale"}, tags=["policy"]) print(f" ... and cannot see retail's fact: {[f.key for f in invisible]}") print(f" episodic recall for ['lookup_payment']: " f"{[e.outcome for e in kernel.episodic.recall(['lookup_payment'])]}") if __name__ == "__main__": # pragma: no cover main()