Tools and parallel calls

Learn QitOS with complete, executable notes-project code.

Observation retains its dict compatibility but stores each field once. obs.task = value and obs["task"] = value update the same authority; mapping step and step_id name the step_id attribute. Contradictory aliases, invalid field types and deletion of schema fields reject atomically. update and |= validate the whole mutation first; extension keys support pop, del, setdefault and popitem. clear is rejected because it would remove required identity. State/metadata nested edits are shared between attribute and mapping access. Constructor inputs are copied; to_dict and to_legacy_dict are independent snapshots. Attribute action results are canonical ToolResult objects and to_dict serializes their canonical shape. Mapping action results remain a fresh legacy projection for existing reducers; assign obs["action_results"] to validate and replace them. Unbound dict mutators are outside this contract.

Legacy host Read uses a zero-based offset and a positive line limit: offset 2, limit 2 selects only lines 3/4. Empty files and EOF return empty output; negative or non-integer windows fail. Selection metadata reports truncation and more lines. Edit requires a unique match unless replace_all=True; no match, ambiguous match, permission denial and explicit expected_sha256 conflicts leave the file unchanged. These two aliases now return canonical ToolResult (read numbered text in .output, status/error in their named fields), rather than successful strings containing errors. Direct callers expecting strings should use .output after checking .status. The existing validation/permission pipeline remains authoritative. Host compatibility remains host execution; Env read_file/edit_file keep their Env-only operations and one-based line_offset interface. No fallback to host or new isolation guarantee is implied.

Goal and prerequisites

Register a decorated function with ToolRegistry before an Agent can call it. The name and typed parameters form the callable interface; read_only and concurrency_safe describe this tool’s behavior. These are commitments by the tool author, not an automatic permission grant.

First run the sequential and parallel notes example. Then run parallel.py: the first declared tool waits for an explicit in-memory event from the second. Two workers therefore finish in reverse order, while the reducer still receives results in declaration order. The event and list are visible, process-local teaching instrumentation.

The five-second wait is a fixture deadline, not a performance test. A single worker cannot satisfy it. Tool errors, timeout, partial outputs and unknown external effects must be inspected through ToolResult rather than interpreted as successful text.

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.

Notes and tools

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())}

Run and verify

python custom_agent.py
python parallel.py

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

completion=[1, 0]; declaration=[0, 1]

Behavior and support boundaries

Timeout is not hard cancellation. worker_still_running and outcome_unknown require reconciliation; do not automatically replay an operation with unknown effects.

Exercise and answer

Set max_concurrency=1 in the reverse-order fixture and explain the failure. Restore 2 to pass; do not increase the timeout to hide the dependency.

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

custom_agent.py

"""A custom notes AgentModule through the public Engine/Session path."""
from dataclasses import dataclass, field

from qitos import Action, AgentModule, Decision, Engine, StateSchema, ToolRegistry
from qitos.engine.action_executor import ActionExecutionPolicy
from qitos.engine.runtime import RuntimeComposition
from notes import summarize_note


# docs:start agent
@dataclass
class NotesState(StateSchema):
    titles: list[str] = field(default_factory=list)


class NotesAgent(AgentModule):
    def __init__(self):
        registry = ToolRegistry()
        registry.register(summarize_note)
        super().__init__(tool_registry=registry)

    def init_state(self, task, **kwargs):
        return NotesState(task=task, max_steps=3)

    def decide(self, state, observation):
        if state.titles:
            return Decision.final(", ".join(state.titles))
        return Decision.act([
            Action(name="summarize_note", args={"index": index}) for index in (0, 1)
        ])

    def reduce(self, state, observation, decision):
        state.titles.extend(item["output"]["title"] for item in observation.get("action_results", []))
        return state
# docs:end agent


# docs:start run
def main():
    for mode in ("sequential", "parallel"):
        engine = Engine(NotesAgent(), runtime=RuntimeComposition(),
                        action_execution_policy=ActionExecutionPolicy(mode=mode, max_concurrency=2))
        result = engine.session("Index the two notes").run()
        assert result.state.titles == ["Session", "Artifact"]
        assert result.state.final_result == "Session, Artifact"
        print(f"{mode}: Session, Artifact; process_local=true")
# docs:end run


if __name__ == "__main__":
    main()

parallel.py

"""Force reverse completion while retaining declaration order in reduce."""
from threading import Event

from qitos import Action, Decision, Engine, ToolRegistry
from qitos.core.function_tool_decorator import function_tool
from qitos.engine.action_executor import ActionExecutionPolicy
from qitos.engine.runtime import RuntimeComposition
from custom_agent import NotesAgent
from notes import NOTES

# Explicit test instrumentation for one local invocation, not persistent state.
second_finished = Event()
completion_order = []


@function_tool(read_only=True, concurrency_safe=True)
def analyze_note(index: int) -> dict:
    """An in-memory fixture that controls completion order without file/network I/O."""
    if index == 0:
        if not second_finished.wait(5):
            raise RuntimeError("This fixture requires two parallel workers")
    completion_order.append(index)
    if index == 1:
        second_finished.set()
    return {"title": NOTES[index].split(":", 1)[0]}


class ParallelNotesAgent(NotesAgent):
    def __init__(self):
        super().__init__()
        self.tool_registry = ToolRegistry()
        self.tool_registry.register(analyze_note)

    def decide(self, state, observation):
        if state.titles:
            return Decision.final(", ".join(state.titles))
        return Decision.act([Action(name="analyze_note", args={"index": i}) for i in (0, 1)])


def run():
    second_finished.clear()
    completion_order.clear()
    engine = Engine(ParallelNotesAgent(), runtime=RuntimeComposition(),
                    action_execution_policy=ActionExecutionPolicy(mode="parallel", max_concurrency=2))
    result = engine.session("Compare completion with declaration order").run()
    assert completion_order == [1, 0]
    assert result.state.titles == ["Session", "Artifact"]
    print("completion=[1, 0]; declaration=[0, 1]; titles=Session, Artifact")


if __name__ == "__main__":
    run()

Next step and API

API Reference · Configuration · Learning path · Next

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