Pause, restore and fork

Learn QitOS with complete, executable notes-project code.

Goal and prerequisites

The first process pauses at a lifecycle boundary after indexing the Session note, writes the Session identity and matching configuration, and exits. The second process loads SQLite, forks the paused snapshot, runs the child, then restores and runs the parent.

The fake provider cursor is explicitly restarted at 1 because this controlled fixture always pauses after response 0. This is not provider continuation persistence. The Session state, ownership and journal are restored through QitOS. A separate composition is used for parent execution so it cannot accidentally reuse the child’s advanced fake cursor.

Compare the parent head before and after child execution, then inspect a steering record in the parent’s Trajectory. The parent must execute the remaining Artifact tool, not merely receive a scripted final answer.

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

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

First process: pause

def create(root):
    root.mkdir(parents=True, exist_ok=False)
    with compose(root, pause=True) as composition:
        session = composition.session("Index the two notes")
        result = session.run()
        assert session.lifecycle.value == "paused"
        assert result.records[0].action_results[0].output["title"] == "Session"
        document = composition.config.to_dict()
        document["runtime"]["environment"] = {"type": "unsafe_host", "workspace": str(root)}
        (root / "agent.json").write_text(json.dumps(document), encoding="utf-8")
        (root / "control.json").write_text(json.dumps({"session_id": session.session_id.value}), encoding="utf-8")
        print("paused after first note; exit this process before restore")

Second process: restore

def restore(root):
    identity = json.loads((root / "control.json").read_text())["session_id"]
    # This fixture always pauses after its first response. Only the fake cursor
    # is supplied here; QitOS restores the real Session state from SQLite.
    with compose(root, start=1, pause=True) as composition:
        before = composition.runtime.checkpoint_store.get_session_head(identity)
        child = composition.fork(identity)
        child.run(steering="Finish an independent index.")
        assert child.session_id.value != identity
        assert composition.runtime.checkpoint_store.get_session_head(identity) == before
    with compose(root, start=1, pause=True) as composition:
        session = composition.restore(identity)
        result = session.run(steering="Finish the index concisely.")
        assert any(a.tool_name == "summarize_note" and a.output["title"] == "Artifact"
                   for record in result.records for a in record.action_results)
        assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
        trajectory = default_reader(root).read_session(identity, view=PrivacyView.RAW_PRIVATE)
        assert any(record.kind.value == "steering" for record in trajectory.records)
        (root / "control.json").write_text(json.dumps({"session_id": identity, "run_id": result.run_id}), encoding="utf-8")
        print("restored; steering recorded; fork left parent head unchanged")

Run and verify

python lifecycle.py create --root session-run
python lifecycle.py restore --root session-run

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

paused after first note
fork left parent head unchanged

Inspect the saved Session

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

Behavior and support boundaries

Fork a paused persisted head before restore claims ownership. Ephemeral and process-local Memory stores do not promise cross-process restore. CLI live pause/steer and unresolved-approval restore are unsupported. Steering changes instructions, not permissions.

Exercise and answer

Change the child steering text and verify the parent head assertion still passes; then inspect the child and parent as distinct Session identities.

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

lifecycle.py

"""Two processes, a durable checkpoint, steering and independent fork."""
import argparse
import json
from pathlib import Path

from notes import compose
from qitos.qita.reader import default_reader
from qitos.tracing.trajectory import PrivacyView


# docs:start create
def create(root):
    root.mkdir(parents=True, exist_ok=False)
    with compose(root, pause=True) as composition:
        session = composition.session("Index the two notes")
        result = session.run()
        assert session.lifecycle.value == "paused"
        assert result.records[0].action_results[0].output["title"] == "Session"
        document = composition.config.to_dict()
        document["runtime"]["environment"] = {"type": "unsafe_host", "workspace": str(root)}
        (root / "agent.json").write_text(json.dumps(document), encoding="utf-8")
        (root / "control.json").write_text(json.dumps({"session_id": session.session_id.value}), encoding="utf-8")
        print("paused after first note; exit this process before restore")
# docs:end create


# docs:start restore
def restore(root):
    identity = json.loads((root / "control.json").read_text())["session_id"]
    # This fixture always pauses after its first response. Only the fake cursor
    # is supplied here; QitOS restores the real Session state from SQLite.
    with compose(root, start=1, pause=True) as composition:
        before = composition.runtime.checkpoint_store.get_session_head(identity)
        child = composition.fork(identity)
        child.run(steering="Finish an independent index.")
        assert child.session_id.value != identity
        assert composition.runtime.checkpoint_store.get_session_head(identity) == before
    with compose(root, start=1, pause=True) as composition:
        session = composition.restore(identity)
        result = session.run(steering="Finish the index concisely.")
        assert any(a.tool_name == "summarize_note" and a.output["title"] == "Artifact"
                   for record in result.records for a in record.action_results)
        assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
        trajectory = default_reader(root).read_session(identity, view=PrivacyView.RAW_PRIVATE)
        assert any(record.kind.value == "steering" for record in trajectory.records)
        (root / "control.json").write_text(json.dumps({"session_id": identity, "run_id": result.run_id}), encoding="utf-8")
        print("restored; steering recorded; fork left parent head unchanged")
# docs:end restore


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("phase", choices=("create", "restore"))
    parser.add_argument("--root", type=Path, required=True)
    args = parser.parse_args()
    {"create": create, "restore": restore}[args.phase](args.root.resolve())

Next step and API

API Reference · Configuration · Learning path · Next

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