Context and memory

Learn QitOS with complete, executable notes-project code.

Goal and prerequisites

The notes Agent needs project instructions and remembered facts. Bind concrete implementations in extensions, then select memory.sources: [project_memory] and compaction.provider: closed_window. MemorySourceAdapter works with both MemdirMemory and MarkdownFileMemory; users do not construct contributions, provider messages or compaction receipts.

This fixture makes four scripted tool calls (two passes over the two notes), then finishes. ClosedExchangeWindowCompactor omits the oldest eligible closed exchanges without a summary only when the request exceeds its configured budget. ContextBudget.protected_recent_exchanges=2 protects the recent window. Without an explicitly selected compactor, history overflow fails instead of silently dropping exchanges.

The public YAML loader accepts context.budget_policy: budget and context.allow_codec_loss: true. Bind budget explicitly through extensions to a context budget policy factory. Loss must be a real boolean; unknown fields and unbound factories fail before the first model request. The complete tutorial below also demonstrates Python composition of other context services; the installed B consumer uses these options directly in YAML.

Basic Python is required. The package declares Python ≥3.10; local qualification uses Python 3.12.7. Each chapter runs independently; reuse the environment and matching files when continuing your project. Commands below use a macOS/Linux shell.

Prepare the project

python3 -m venv .venv
source .venv/bin/activate
python -m pip install "qitos @ git+https://github.com/WhitzardAgent/WhitzardOS.git@f7d4b2d666a156d361da496a41868278f84ffabf"
mkdir notes_lesson
cd notes_lesson

This lesson uses the integrated R1 implementation pinned below; a source install does not imply a PyPI release.

Save the complete files below in this directory. No repository clone, editable install, or copied tests are needed.

Run and verify

python context.py --root context-run

Expected output fragments (generated IDs vary). Each run verifies tool results or persistence assertions and exits 0 on success.

compaction loss recorded

Behavior and support boundaries

Memory is request context data at user placement, not a system instruction, permission or checkpoint store. The namespace is a logical label bound to a memory resource by the application factory; it is not permission to read arbitrary directories. No global memory is read by default.

Memdir text records retain an independent record_id across processes. Equal text in two independent records remains two records; content edits change the contribution revision/digest. Recall reads disk afresh rather than merging stale append-time records. External edits/deletions are reflected in retrieval and summary. Host paths are not added to contribution IDs, content or metadata. Use create=True only for explicit initialization; normal construction restores an existing root and raises MemoryResourceError when it is missing. This changes the old implicit-create default, including memdir_memory(...).

The adapter borrows the memory. Adapter reset/close do not reset, close or delete it. Memdir reset/evict affect only its in-memory cache, not persistent files. Text records are the durable reference; arbitrary Python/JSON records or historical metadata are not a lossless storage promise. MarkdownFileMemory works within its current instance, but a fresh instance does not reload its append-only log. No metadata evaluation or guessed reconstruction occurs.

Selection never mutates ExchangeLog bytes or tool completion order. Tool declarations/results are retained or omitted together. Incomplete exchanges, pending steering and required context/artifacts stay protected; opaque continuation state conservatively protects the complete history because no smaller dependency closure is declared. If protected content cannot fit, selection raises a typed failure. Recalling an unchanged revision after an earlier omission or restore injects it again.

The installed two-process consumer seeds remembered-value=17, exits, restores the configured memory and verifies actual encoded requests and repeated budget compaction. It uses a deterministic provider, not a live model. Legacy CompactHistory remains compatible. Markdown durable reading, model summaries, complete legacy-history retirement, cross-Agent memory policy and original-Agent migration experiments remain out of scope.

Exercise and answer

Edit the persisted text without changing its record ID, then verify its revision changes. Restore the same resource in a second process; a different factory-bound namespace should stay empty.

Common errors and cleanup

ModuleNotFoundError: activate the environment with the specified installation and save every file on this page. Existing run root: choose a new --root instead of overwriting evidence. Assertion failure: inspect the first failed tool or typed error, not only the final text. After all processes and the board stop, remove only this lesson’s newly created run directories if no longer needed; retain SQLite, journals and reports you want to debug.

Complete files: save in the project root

notes.py

"""Synthetic notes; fake model, real tools, composition, Session and journal."""
import argparse
from dataclasses import replace
import json
from pathlib import Path

from qitos.config import build_agent_composition, load_agent_config
from qitos.core.function_tool_decorator import function_tool
from qitos.engine.runtime import LifecyclePolicy

# docs:start fixture
NOTES = (
    "Session: A durable session can resume after a process exits.",
    "Artifact: Large tool outputs can be retained outside model context.",
)


@function_tool(read_only=True, concurrency_safe=True)
def summarize_note(index: int) -> dict:
    """Extract a title and word count from a synthetic in-memory note."""
    text = NOTES[index]
    return {"title": text.split(":", 1)[0], "words": len(text.split())}
# docs:end fixture


# docs:start provider
class FakeProvider:
    """Scripted responses; this does not summarize or reason like a real model."""
    model = "notes-fake"
    qitos_protocol = "react_text_v1"

    def __init__(self, start=0):
        self.stage = start

    def call_raw(self, messages, **options):
        if self.stage < len(NOTES):
            content = f"Thought: inspect a note\nAction: summarize_note(index={self.stage})"
        else:
            content = "Final Answer: Indexed 2 notes: Session, Artifact."
        self.stage += 1
        return {"choices": [{"message": {"content": content}}]}
# docs:end provider


class PauseAfterTool(LifecyclePolicy):
    policy_id = "notes.pause_after_tool"

    def should_pause(self, context):
        return context.step_id == 0


# docs:start composition
def configuration(root):
    config = load_agent_config(Path(__file__).with_name("agent.yaml"))
    return replace(config, runtime=replace(
        config.runtime, data_root=str(root / "data"),
        environment=replace(config.runtime.environment, workspace=str(root)),
        session=replace(config.runtime.session, path=str(root / "sessions.sqlite3")),
        trajectory=replace(config.runtime.trajectory, output=str(root / "trajectory.journal")),
    ))


def compose(root, *, start=0, pause=False):
    config = configuration(root)
    if pause:
        config = replace(config, lifecycle={"policy": "pause"})
    composition = build_agent_composition(
        config, model_override=FakeProvider(start), extensions={"pause": PauseAfterTool},
    )
    composition.tool_registry.register(summarize_note)
    return composition
# docs:end composition


# docs:start run
def run(root):
    root.mkdir(parents=True, exist_ok=False)
    with compose(root) as composition:
        session = composition.session("Index both synthetic notes")
        result = session.run()
        outputs = [action.output for record in result.records for action in record.action_results
                   if action.tool_name == "summarize_note"]
        assert [output["title"] for output in outputs] == ["Session", "Artifact"]
        assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
        config = composition.config.to_dict()
        config["runtime"]["environment"] = {"type": "unsafe_host", "workspace": str(root)}
        (root / "agent.json").write_text(json.dumps(config), encoding="utf-8")
        control = {"session_id": session.session_id.value, "run_id": result.run_id}
        (root / "control.json").write_text(json.dumps(control), encoding="utf-8")
        print(json.dumps({**control, "result": result.state.final_result, "outputs": outputs}))
# docs:end run


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=Path("notes-run"))
    run(parser.parse_args().root.resolve())

agent.yaml

schema: qitos.agent
agent:
  name: notes_agent
  protocol: react_text_v1
model:
  provider: openai_compatible
  model: notes-fake
  credential:
    ref: notes-provider
  request:
    max_tokens: 512
    timeout_seconds: 30
    retries: 0
tools:
  preset: none
runtime:
  environment:
    type: unsafe_host
    workspace: .
  session:
    mode: durable
    store: sqlite
    path: ./notes-run/sessions.sqlite3
  trajectory:
    enabled: true
    output: ./notes-run/trajectory.journal
budgets:
  max_steps: 6
  max_requests: 6
  max_runtime_seconds: 30
failure_policy:
  tool: fail_closed

context.py

"""Explicit contributor, memory, selector and compactor factories; no network."""
import argparse
from dataclasses import replace
from pathlib import Path

from qitos.config import build_agent_composition
from qitos.core.context import (
    DeclaredContextBudgetPolicy, PriorityContextSelectionPolicy, StaticContextContributor,
)
from qitos.core.memory import MemoryRecord
from qitos.kit.context.compaction import ClosedExchangeWindowCompactor
from qitos.kit.memory.adapter import MemorySourceAdapter
from qitos.kit.memory.memdir_memory import MemdirMemory
from qitos.qita.reader import default_reader
from qitos.tracing.trajectory import PrivacyView

from notes import FakeProvider, summarize_note, configuration


class ContextProvider(FakeProvider):
    """Four scripted tool turns make closed-exchange omission observable."""

    def __init__(self):
        self.stage = 0

    def call_raw(self, messages, **options):
        content = (f"Thought: inspect\nAction: summarize_note(index={self.stage % 2})"
                   if self.stage < 4 else "Final Answer: Indexed 2 notes: Session, Artifact.")
        self.stage += 1
        return {"choices": [{"message": {"content": content}}]}


class AuditedSelector(PriorityContextSelectionPolicy):
    def __init__(self):
        self.seen = set()

    def select(self, contributions, **options):
        contributions = tuple(contributions)
        self.seen.update(item.contribution_id for item in contributions)
        return super().select(contributions, **options)


def run(root):
    root.mkdir(parents=True, exist_ok=False)
    config = replace(
        configuration(root),
        context={"contributors": ["project"], "selector": "selector", "budget_policy": "budget", "allow_codec_loss": True},
        memory={"sources": ["project_memory"]}, compaction={"provider": "closed_window"})
    config = replace(config, budgets=replace(config.budgets, max_steps=6, max_requests=6))
    selector = AuditedSelector()
    memory = MemdirMemory(str(root / "memory"), create=True)
    memory.append(MemoryRecord("user", 'Session and Artifact are the two note titles.', 0))
    with build_agent_composition(config, model_override=ContextProvider(), extensions={
        "project": lambda: StaticContextContributor("lesson.project", "project", "Use only the supplied notes."),
        "project_memory": lambda: MemorySourceAdapter(memory, namespace="notes", required=True),
        "selector": selector, "closed_window": ClosedExchangeWindowCompactor,
        "budget": lambda: DeclaredContextBudgetPolicy(default_max_input_units=30000, protected_recent_exchanges=2),
    }) as composition:
        composition.tool_registry.register(summarize_note)
        session = composition.session("Index both notes")
        assert session.run().state.final_result == "Indexed 2 notes: Session, Artifact."
        assert "lesson.project" in selector.seen
        assert any(identity.startswith("memory:notes:") for identity in selector.seen)
        trajectory = default_reader(root).read_session(session.session_id.value, view=PrivacyView.RAW_PRIVATE)
        assert any(record.kind.value == "compaction" and not record.loss.is_lossless for record in trajectory.records)
    print("context selected; memory selected; compaction loss recorded")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, required=True)
    run(parser.parse_args().root.resolve())

Next step and API

API Reference · Configuration · Learning path · Next

Source file (optional; all required code is already on this page).