Delegate, join and handoff

Learn QitOS with complete, executable notes-project code.

Each independent child has an explicit two-request allocation. Omitting it can reserve the parent’s entire remaining budget for the first child, leaving later workers with zero requests. Completion does not authorize an implicit refund or unknown-effect replay.

Goal and prerequisites

The parent first creates a durable paused Session. Its explicit resolver maps the known notes_agent descriptor to local subprocess execution. Each child restores a real Session using the installed package; there is no distributed worker service hidden behind this example.

delegate and spawn submit child work; fan_out submits a batch; join waits for the operation IDs under an explicit policy. Inspect four child completions and a closed join. The resolver handles join without rerunning its referenced children. Child final responses are scripted in this bounded scheduler lesson; successful orchestration is not proof of independent model reasoning.

Run handoff.py separately. The destination resumes the same work item in another process, ownership changes, and dispatch from the superseded source is rejected. In contrast, the lifecycle lesson’s fork creates an independent branch and preserves the source head.

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.

Run and verify

python multi_agent.py --root work-run
python handoff.py --root handoff-run

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

durable children=4; join=closed
handoff destination ran

Handoff supports both the original serial path and --concurrent, which immediately restores/runs the same Session inside the scheduler callable. The source commits transfer admission before dispatch and never writes the head from a late callback. The admission wait returns separately from the destination subprocess result: only the latter proves business completion. Destination restore uses the existing owner/generation CAS. Unknown invocation outcomes are not automatically replayed; the application owns waiting, retries and reconciliation.

Behavior and support boundaries

A timeout or outcome_unknown is terminal for the lesson wait, not a trigger to resubmit work. LocalWorkScheduler is process-local scheduling; durable receipts do not turn it into a distributed queue.

Exercise and answer

Inspect the four completion records and distinguish the two individual operations from the two fan-out children. Keep join references as operation IDs, not display names.

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

multi_agent.py

"""Durable spawn/delegate/fan-out/join with real child Session execution.

The local scheduler resolves only this tutorial's Agent; no distributed service.
"""
import argparse
from pathlib import Path
import subprocess
import sys
import time

from qitos.core.work_graph import WorkGraph
from qitos.engine.work_runtime import DurableWorkRuntime, LocalWorkScheduler
from dataclasses import replace
from qitos.config import build_agent_composition
from notes import FakeProvider, PauseAfterTool, summarize_note, configuration


def compose(root, *, pause=False, finish=False):
    config = configuration(root)
    config = replace(config, budgets=replace(config.budgets, max_requests=16),
                     lifecycle={"policy": "pause"})
    result = build_agent_composition(config, model_override=FakeProvider(start=2 if finish else 0),
                                     extensions={"pause": PauseAfterTool})
    result.tool_registry.register(summarize_note)
    return result


def wait(session, operation):
    deadline = time.monotonic() + 30
    while time.monotonic() < deadline:
        graph = WorkGraph.from_canonical_dict(session.inspect().work_graph)
        receipt = next(item for item in graph.operation_receipts if item.operation_id == operation.operation_id)
        if operation.operation == "handoff" and receipt.state in {
            "transfer_admitted", "ownership_committed", "running", "completed",
        }:
            # A bounded admission wait, distinct from waiting for task completion.
            assert graph.transfers
            return graph
        if receipt.state in {"completed", "failed", "outcome_unknown"}:
            assert receipt.state == "completed", receipt.state
            return graph
        time.sleep(0.02)
    raise AssertionError("child deadline exceeded; inspect before retrying")


def run(root):
    root.mkdir(parents=True, exist_ok=False)

    class Resolver:
        resolver_id = "tutorial.notes_agent.worker"

        def resolve(self, descriptor):
            def execute():
                for identity in (() if descriptor.operation == "join" else descriptor.child_session_ids):
                    subprocess.run([sys.executable, __file__, "--root", str(root), "--child", identity],
                                   check=True, capture_output=True, text=True, timeout=20)
                return {"children": list(descriptor.child_session_ids)}
            return execute

    with compose(root, pause=True) as composition:
        composition.runtime.work_runtime = DurableWorkRuntime(LocalWorkScheduler(Resolver(), max_workers=2))
        parent = composition.session("Index, then ask independent note workers")
        parent.run()
        assert parent.lifecycle.value == "paused"
        delegated = parent.submit_work("delegate", {"agent": "notes_agent", "task": "Describe the Session note",
                                                    "budget": {"model_requests": 2}})
        wait(parent, delegated)
        spawned = parent.submit_work("spawn", {"agent": "notes_agent", "task": "Describe the Artifact note",
                                               "budget": {"model_requests": 2}})
        wait(parent, spawned)
        batch = parent.fan_out([{"agent": "notes_agent", "task": "Review Session", "budget": {"model_requests": 2}},
                                {"agent": "notes_agent", "task": "Review Artifact", "budget": {"model_requests": 2}}])
        wait(parent, batch)
        joined = parent.join([delegated.operation_id, spawned.operation_id, batch.operation_id], policy="all")
        graph = wait(parent, joined)
        assert graph.joins[-1].state == "closed"
        assert len(graph.completions) == 4
        print("durable children=4; join=closed; parent retains ownership")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--child")
    args = parser.parse_args()
    root = args.root.resolve()
    if args.child:
        with compose(root, finish=True, pause=True) as composition:
            result = composition.restore(args.child).run()
            assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
    else:
        run(root)

handoff.py

"""Transfer one work item's owner; demonstrate the superseded source fence."""
import argparse
from pathlib import Path
import subprocess
import sys
import threading

from notes import compose
from multi_agent import wait
from qitos.engine.work_runtime import DurableWorkRuntime, LocalWorkScheduler, WorkRuntimeError


def run(root, *, concurrent=False):
    root.mkdir(parents=True, exist_ok=False)

    destination_done = threading.Event()
    destination_errors = []

    def destination(identity):
        result = subprocess.run([
            sys.executable, __file__, "--root", str(root), "--destination", identity,
        ], capture_output=True, text=True, timeout=20)
        if result.returncode:
            raise RuntimeError(result.stderr)

    class Resolver:
        resolver_id = "notes.handoff.worker"

        def resolve(self, descriptor):
            def execute():
                # Runtime admission is already durable. This worker may restore
                # and run the same Session before the source callback returns.
                if concurrent:
                    try:
                        destination(descriptor.parent_session_id)
                    except Exception as error:
                        destination_errors.append(error)
                    finally:
                        destination_done.set()
                return None
            return execute

    with compose(root, pause=True) as composition:
        composition.runtime.work_runtime = DurableWorkRuntime(LocalWorkScheduler(Resolver()))
        source = composition.session("Index notes, then transfer ownership")
        source.run()
        identity = source.work_item_id
        operation = source.handoff("notes_agent", rationale="Finish with the destination worker")
        graph = wait(source, operation)
        transfer = graph.transfers[-1]
        assert transfer.from_agent_id != transfer.to_agent_id
        assert graph.work_items[identity].owner.agent_id == transfer.to_agent_id
        try:
            source.spawn("notes_agent", task="A stale owner must not dispatch")
        except WorkRuntimeError as error:
            assert error.code == "superseded_owner"
        else:
            raise AssertionError("Superseded source unexpectedly dispatched")
        identity = source.session_id.value
    if concurrent:
        assert destination_done.wait(20), "destination deadline exceeded"
        assert not destination_errors, destination_errors
    else:
        # The original serial path remains supported as an application choice.
        destination(identity)
    print("handoff destination ran; owner changed; source fenced")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--destination")
    parser.add_argument("--concurrent", action="store_true")
    args = parser.parse_args()
    if args.destination:
        with compose(args.root.resolve(), start=1, pause=True) as composition:
            result = composition.restore(args.destination).run()
            assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
    else:
        run(args.root.resolve(), concurrent=args.concurrent)

Next step and API

API Reference · Configuration · Learning path · Next

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