Replace a provider and add context

Learn QitOS with complete, executable notes-project code.

Goal and prerequisites

A composition accepts a public model override and explicitly named extension factories. ObservedFakeProvider preserves the callable interface, counts actual requests and asserts the selected project context appears in each request. It then delegates to the visible deterministic provider.

The project contributor is a factory returning StaticContextContributor. Its registration name matches context.contributors. This avoids imports from arbitrary names embedded in YAML and keeps executable extension selection under application control.

This is a complete provider substitution and context integration example, not a new transport client. For a real provider, use the matching configuration and explicit credential resolver described in Configuration. Consult the extension index before replacing stores, sinks or sandboxes: preserving a Python method signature alone does not preserve their durability and security contracts.

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 provider_extension.py --root extension-run

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

provider replaced; context observed; requests=3

Behavior and support boundaries

No provider is universally compatible. Codec, continuation, tool schema and loss policy must agree. Inspect typed failures; do not disable loss checks to accept an incompatible response.

Exercise and answer

Rename the contributor key in both config and extensions. Leaving only one side changed must produce a configuration/extension failure.

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

provider_extension.py

"""Replace the provider and inject context, without changing the kernel."""
import argparse
from dataclasses import replace
from pathlib import Path

from notes import FakeProvider, configuration, summarize_note
from qitos.config import build_agent_composition
from qitos.core.context import StaticContextContributor


class ObservedFakeProvider(FakeProvider):
    """Keep the public provider call shape and validate selected context."""
    def __init__(self):
        super().__init__()
        self.requests = 0

    def call_raw(self, messages, **options):
        assert "notes-project-context" in str(messages)
        self.requests += 1
        return super().call_raw(messages, **options)


def run(root):
    root.mkdir(parents=True, exist_ok=False)
    config = replace(configuration(root), context={"contributors": ["project"]})
    provider = ObservedFakeProvider()
    with build_agent_composition(config, model_override=provider, extensions={
        "project": lambda: StaticContextContributor("notes.project", "project", "notes-project-context"),
    }) as composition:
        composition.tool_registry.register(summarize_note)
        result = composition.session("Index both notes").run()
        assert result.state.final_result == "Indexed 2 notes: Session, Artifact."
        assert provider.requests == 3
    print("provider replaced; context observed; requests=3")


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

consumer.py

"""Offline installed-wheel mechanism proof; no model capability claim."""
from __future__ import annotations

import json
import sys
from pathlib import Path
from tempfile import TemporaryDirectory
from threading import Event
from types import SimpleNamespace as NS
from unittest.mock import patch

from qitos.config import FakeCredentialResolver, load_agent_config, build_agent_composition
from qitos.core.conversation import ExchangeLog
from qitos.core.function_tool_decorator import function_tool
from qitos.tracing.journal_store import JournalTrajectoryStore


def run(interrupted=False):
    second_finished = Event()
    completions = []
    executions = []
    requests = []
    resources = []

    @function_tool(read_only=True, concurrency_safe=True)
    def add(a: int, b: int) -> int:
        """Add two integers."""
        executions.append(('add', a, b))
        if (a, b) == (2, 3):
            assert second_finished.wait(5)
        return a + b

    @function_tool(read_only=True, concurrency_safe=True)
    def multiply(a: int, b: int) -> int:
        """Multiply two integers."""
        executions.append(('multiply', a, b))
        return a * b

    class CompletionHook:
        def on_event(self, event, state, record, engine):
            if event.payload.get('stage') == 'tool_slot_terminal':
                name = event.payload['tool_result']['tool_name']
                completions.append(name)
                if name == 'multiply':
                    second_finished.set()

    def chunk(text='', calls=None, finish=None):
        return NS(choices=[NS(delta=NS(content=text, tool_calls=calls),
                              finish_reason=finish)], usage=None)

    def call(index, identifier, name, arguments):
        return NS(index=index, id=identifier, type='function',
                  function=NS(name=name, arguments=arguments))

    class Stream:
        def __init__(self, values):
            self.values = iter(values)
            self.closed = 0
            resources.append(self)

        def __iter__(self):
            return self

        def __next__(self):
            return next(self.values)

        def close(self):
            self.closed += 1

    class Client:
        def __init__(self, **kwargs):
            assert kwargs['max_retries'] == 0
            self.closed = 0
            self.chat = NS(completions=self)
            resources.append(self)

        def create(self, **kwargs):
            assert kwargs['stream'] is True
            assert kwargs['max_tokens'] == 10240
            requests.append(kwargs)
            results = {m['tool_call_id']: json.loads(m['content'])
                       for m in kwargs['messages'] if m['role'] == 'tool'}
            if len(requests) == 1:
                return Stream([chunk(calls=[call(0, 'one', 'add', '{"a":2,"b":3}'),
                                             call(1, 'two', 'multiply', '{"a":2,"b":3}')]),
                               chunk(finish='tool_calls')])
            if len(requests) == 2:
                assert results == {'one': 5, 'two': 6}, results
                values = [chunk(calls=[call(0, 'three', 'add', '{"a":5,"b":6}' if not interrupted else '{"a":5')])]
                if not interrupted:
                    values.append(chunk(finish='tool_calls'))
                return Stream(values)
            assert len(requests) == 3
            assert results['three'] == 11
            return Stream([chunk('{"final_answer":"11"}'), chunk(finish='stop')])

        def close(self):
            self.closed += 1

    with TemporaryDirectory(prefix='qitos-r1-consumer-') as directory:
        root = Path(directory)
        path = root / 'agent.yaml'
        journal = root / 'trajectory.journal'
        path.write_text(f'''schema: qitos.agent
agent:
  name: offline-multi-round
  protocol: json_decision_multi_v1
  parser: auto
model:
  provider: openai_compatible
  model: offline
  base_url: https://offline.invalid/v1
  credential:
    ref: offline
  request:
    max_tokens: 10240
    retries: 0
tools:
  preset: none
  policy: auto
runtime:
  environment:
    type: unsafe_host
    workspace: {root}
  session:
    enabled: true
    store: memory
  trajectory:
    enabled: true
    output: {journal}
    privacy: private
budgets:
  max_steps: 5
  max_requests: 3
''')
        with patch.dict(sys.modules, {'openai': NS(OpenAI=Client, APIConnectionError=ConnectionError, APITimeoutError=TimeoutError)}):
            with build_agent_composition(load_agent_config(path),
                                         credential_resolver=FakeCredentialResolver()) as composition:
                composition.tool_registry.register(add)
                composition.tool_registry.register(multiply)
                composition.engine.hooks.append(CompletionHook())
                composition.engine.stream_callback = lambda text: None
                session = composition.session('Compute add(2,3) and multiply(2,3), then add their results.')
                result = session.run()
                inspection = session.inspect()
                assert inspection.budget['model_requests_consumed'] == len(requests)
                snapshot = composition.runtime.checkpoint_store.get_session_snapshot(
                    session.current_head.snapshot_id.value)
                conversation = next(c for c in snapshot.payload['components'] if c['slot'] == 'conversation')
                log = ExchangeLog.from_dict(conversation['payload']['exchange_log'])
                assert log.open_batch_id() is None
                batches = [item.batch_id for item in log.items if getattr(item, 'batch_id', None)
                           and getattr(item, 'tool_calls', None)]
                assert len(batches) == (1 if interrupted else 2)
                for batch in batches:
                    calls = log.declared_calls(batch)
                    terminals = log.results_for_batch(batch)
                    assert len(calls) == len(terminals)
                    assert len({item.identity.key() for item in terminals}) == len(calls)
                    ordered = log.results_for_batch_in_declaration_order(batch)
                    assert [item.identity for item in ordered] == [item.identity for item in calls]
                assert [item.identity.call_id for item in log.results_for_batch(batches[0])] == ['two', 'one']
                assert completions[:2] == ['multiply', 'add']
                assert len(completions) == len(executions)
                assert len(executions) == (2 if interrupted else 3)
                assert len(requests) == (2 if interrupted else 3)
                if interrupted:
                    assert ('add', 5, 6) not in executions
                    assert result.error_code == 'provider_stream_protocol_error'
                    assert result.state.final_result is None
                else:
                    assert str(result.state.final_result) == '11', result
                session_id = session.session_id.value
                run_id = session.run_id.value
        store = JournalTrajectoryStore(journal, read_only=True)
        trajectory = store.read_session(session_id)
        assert trajectory.records
        assert all(record.session_id == session_id for record in trajectory.records)
        assert all(record.run_id == run_id for record in trajectory.records)
        assert all(resource.closed == 1 for resource in resources)
        return {'interrupted': interrupted, 'requests': len(requests),
                'executions': len(executions), 'completion_order': completions,
                'final': result.state.final_result}


if __name__ == '__main__':
    print(json.dumps([run(), run(interrupted=True)], ensure_ascii=False))

Next step and API

API Reference · Configuration · Learning path · Next

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

Native model I/O mechanism proof

The additional consumer.py uses the built-in OpenAI-compatible adapter with an offline SDK transport, public composition and Session, and real parallel tools. It asserts three requests, three executions, reversed first-batch completion, declaration-order results, final 11 and journal identities. The truncated second batch executes no tool and produces no successful final. This proves framework mechanisms, not live model capability. Install qitos[openai] before running it outside the source checkout.

Stream construction failures raise sanitized typed failures without becoming answer text. Native streams require an explicit provider terminal, collect trailing usage once and close owned resources on completion, failure and consumer close. Unknown usage stays unknown. Cancellation remains cooperative for Python workers.