Inspect and export with qita

Learn QitOS with complete, executable notes-project code.

Goal and prerequisites

Run the notes project, then inspect exactly the Session ID stored in control.json. default_reader selects the canonical journal while retaining explicit historical trace support. read_session reconstructs an observation view; it does not resume execution.

inspect_run.py reads snapshot pages and streams a public canonical export through owned staging. Its small re-import check compares every projected record. Re-import remains materializing. The lesson retains the empty selector directory required by the legacy replay/HTML CLI. The new canonical export takes a journal and run ID directly.

Use the CLI commands below to inspect, replay and export. Replay displays recorded execution; it does not call the model or repeat tools. Stop the read-only board with Ctrl-C.

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

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

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

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

When continuing Quickstart with an existing notes-run, skip the first notes.py creation command and inspect that run. Replay starts a local web server; stop it with Ctrl-C before running export/board.

Run and verify

python notes.py --root notes-run
python inspect_run.py --root notes-run

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

"records":

Inspect the saved Session

session_id=$(python -c 'import json; print(json.load(open("notes-run/control.json"))["session_id"])')
qit session inspect --config notes-run/agent.json --session-id "$session_id"
qita inspect session "$session_id" --logdir ./notes-run

run_id=$(python -c 'import json; print(json.load(open("notes-run/control.json"))["run_id"])')
qita replay --run "notes-run/$run_id"
qita export --run "notes-run/$run_id" --html notes-run/trajectory.html
qita export --run "$run_id" --journal notes-run/trajectory.journal --canonical notes-run/streamed.json
qita board --logdir ./notes-run

Behavior and support boundaries

Inspect needs neither Docker nor credentials. read_page retains a page, one bounded frame and fixed I/O/SQLite caches; its temporary disk index grows with the journal. Every page still hashes all snapshot bytes. Complete read_run/read_session, legacy HTML export and re-import remain materializing. Public export declares loss and cannot replace private recovery data.

A cursor binds the query/view, source, committed head and byte boundary, and expires when its reader closes. Concurrent appends are excluded from that traversal. After the last page, poll with a fresh query whose after_sequence is the last page’s watermark. Changed queries, replaced/truncated sources and snapshot damage reject with typed errors. Historical and third-party sources without this capability raise BoundedReadUnsupported; old complete calls retain their existing behavior.

Exercise and answer

Compare private and public records without exposing private payloads. Equal record counts do not establish lossless payload equivalence.

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.

Continuous iter_records and file export validate the full snapshot at the start and end, and each addressed frame during traversal. Iterator values are partial until exhaustion; early close releases resources but does not establish a completed traversal. File output is published only after final validation. Independent read_page calls retain full-history verification on every page.

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

inspect_run.py

"""Page an existing notes journal, stream a public export and verify re-import."""
import argparse
import json
from pathlib import Path

from qitos.qita.reader import default_reader
from qitos.tracing.exporter import CanonicalTrajectoryExporter, ExportArtifact
from qitos.tracing.paging import iter_records
from qitos.tracing.trajectory import LossReport, PrivacyView, TrajectoryQuery


def inspect(root):
    control = json.loads((root / "control.json").read_text())
    reader = default_reader(root)
    try:
        query = TrajectoryQuery(session_id=control["session_id"], limit=128)
        page = reader.read_page(query)
        assert page.records
        exporter = CanonicalTrajectoryExporter()
        target = root / "public-trajectory.json"
        receipt = exporter.export_file(reader, query, target)
        # Re-import remains a materializing API; this small lesson checks every record.
        data = target.read_bytes()
        value = json.loads(data)
        artifact = ExportArtifact(exporter.capabilities.exporter_id, exporter.capabilities.format_version,
                                  "application/json", data, receipt.digest, PrivacyView.REDACTED_PUBLIC,
                                  True, value["provenance"], LossReport.from_dict(value["loss"]))
        imported = exporter.reimport(artifact)
        expected = iter_records(reader, query)
        for actual in imported.records:
            assert actual.to_dict() == next(expected).to_dict()
        assert next(expected, None) is None
        # Legacy replay/HTML CLI still accepts an existing directory selector.
        (root / control["run_id"]).mkdir(exist_ok=True)
        print(json.dumps({"records": receipt.record_count, "lossless": imported.loss.is_lossless,
                          "run_selector": str(root / control["run_id"]), "bounded_export": True}))
    finally:
        reader.close()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, required=True)
    inspect(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).