> ## Documentation Index
> Fetch the complete documentation index at: https://qitor.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Hermes-like 持久研究笔记本

> 使用专业项目学习可替换的 Agent 设计；当前为开发中课程。

状态：开发中，尚未完成课程资格验证。下列源码是工作实现，不是已通过 3×3 真实矩阵的发布承诺。基础机制与已发生的结果见[实施记录](/zh/tutorials/agent-design-lab-results)。

## 原始思想与改编边界

区分事实记忆、历史片段和流程技能。目录只公开描述，选定技能后才加载完整正文。 [原始来源](https://hermes-agent.nousresearch.com/docs/guides/work-with-skills)。

## 框架与用户的分工

Memdir 事实与历史分别绑定根目录；SqliteToolLibrary 持久化技能版本和来源。学习与回忆两个进程仅共享显式绑定的存储。

QitOS 已提供：模型事务与 usage、工具权限/校验、Env 执行、Session、ArtifactRef 和 Trajectory。用户编写：任务、策略、完成检查及记忆/技能选择。本轮补齐：自定义 agent\_factory 接入、持久技能版本、完整正文加载、显式 Memdir 删除与产物引用的数据权限修正。

核心增量位于完整源码中的 agent.py；CLI 负责显式配置与资源生命周期，evaluate.py 是控制端检查器。任务由项目自有合成材料组成，不是论文基准或真实客户数据。

## 安装、配置与运行

先按仓库构建说明安装本轮 QitOS wheel（当前 PyPI 版本不能替代未发布实现），再安装本课程。真实地址与凭据仅放在仓库外。模型配置采用完整 qitos.agent 文件，课程只读取其中 model；不是通过环境变量注入 key。

先执行 --phase learn，再用新的 --root 执行 --phase recall；两次必须使用同一仓库外 --shared-root。不要复用上一轮 input 目录。

默认输出上限 10,240，可通过私有模型配置提高；单任务步数、请求数和运行时间读取公开配置。validate 不调用模型，真实执行必须显式 --live。Docker 不可用时不能降级到宿主。模型失败、未完成任务和人工干预均保留，不能自动改为通过。

## 验证、练习与组合

独立检查器核验资料/数值或实际代码结果，不接受模型自报成功。计划修订、技能实际加载、子 Session 身份与失败也须单独验收。Session 恢复不是文件回滚；生成代码只在受限 Env 内执行。

练习：把词法检索替换为另一 MemorySource，保留命名空间隔离和显式遗忘。

组合：向 PlanAct 提供批准的协议事实；不可把尚未核验的历史摘要当成已验证证据。

完整矩阵：三种任务各三轮。ReAct/PlanAct 使用同样任务；PlanAct 的 static、Hermes 的 no-memory、Voyager 的 no-skills 是明确对照，不以一次成功作性能结论。私有原始轨迹不能直接公开；仅在检查许可与脱敏后发布衍生摘要。

```bash theme={null}
python -m pip install .
python -m qitos_lab_hermes validate --config agent.yaml --root /tmp/lab-validation
python -m qitos_lab_hermes run --config agent.yaml --model-config /private-config/model.yaml --credentials /private-config/credentials.yaml --root /private-runs/hermes-attempt --task 0 --live --phase learn --shared-root /private-runs/shared-notebook
```

完整实现路径: `examples/projects/hermes_notebook/src/qitos_lab_hermes/agent.py`.

## 从完整源码提取：设计增量

下列片段由完整项目同步生成，不是另一份手工维护的实现。完整安装文件在后面。

```python theme={null}
    def prepare(self, state):
        selected = []
        for name, version in state.selected_skills.items():
            item = self.skills.get_version(name, version)
            if item is None:
                raise ValueError("required_skill_version_missing")
            selected.append(
                {"name": name, "version": version, "instructions": item.source}
            )
        # Full selected documents are part of the current task input. A request
        # budget failure must be explicit; slicing instructions is not recall.
        return json.dumps(
            {
                "task": state.task,
                "selected_skills": selected,
                "recent_results": state.recent[-4:],
            }
        )

    def reduce(self, state, observation, decision):
        for item in observation.action_results:
            output = item.output
            if (
                isinstance(output, dict)
                and "instructions" in output
                and "sha256" in output
            ):
                state.selected_skills[output["name"]] = output["version"]
        state.recent.extend(
            item.to_model_dict(max_chars=6000) for item in observation.action_results
        )
        state.recent = state.recent[-8:]
        return state
```

## 完整文件：复制到项目根目录

```yaml title="agent.yaml" theme={null}
schema: qitos.agent
agent:
  name: design-lab-hermes-research
  protocol: json_decision_multi_v1
model:
  provider: openai_compatible
  model: example-model
  base_url: https://provider.example/v1
  credential:
    ref: research-model
  request:
    max_tokens: 10240
    timeout_seconds: 180
    retries: 0
tools:
  preset: env_coding
runtime:
  environment:
    type: docker
    workspace: .
    image: python:3.12-slim
  session:
    mode: durable
    store: sqlite
    path: ./sessions.sqlite3
  trajectory:
    enabled: true
    output: ./trajectory.journal
budgets:
  max_steps: 80
  max_requests: 80
  max_runtime_seconds: 3600
failure_policy:
  tool: continue
```

```toml title="pyproject.toml" theme={null}
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[project]
name = "qitos-lab-hermes"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = ["qitos[openai]"]

[tool.setuptools.packages.find]
where = ["src"]

[tool.setuptools.package-data]
qitos_lab_hermes = ["tasks.json"]
```

```python title="src/qitos_lab_hermes/__init__.py" theme={null}
"""Persistent learning application over QitOS."""
```

```python title="src/qitos_lab_hermes/agent.py" theme={null}
"""Facts, episodic recall and selectively loaded procedures are distinct."""

from dataclasses import asdict, dataclass, field
import hashlib
import json
from pathlib import Path

from qitos.core.agent_module import AgentModule
from qitos.core.function_tool_decorator import function_tool
from qitos.core.memory import MemoryRecord
from qitos.core.state import StateSchema
from qitos.kit.memory.memdir_memory import MemdirMemory
from qitos.kit.tool.library.base import ToolArtifact
from qitos.kit.tool.library.sqlite_store import SqliteToolLibrary


@dataclass
class ResearchState(StateSchema):
    recent: list[dict] = field(default_factory=list)
    selected_skills: dict[str, int] = field(default_factory=dict)


class ResearchAgent(AgentModule):
    def __init__(self, *, skills, **kwargs):
        super().__init__(**kwargs)
        self.skills = skills  # Borrowed resolver, never serialized into State.

    def init_state(self, task, **kwargs):
        return ResearchState(task=task, max_steps=self.config.get("max_steps", 80))

    def base_persona_prompt(self, state):
        return (
            "You maintain a research notebook across independent Sessions. "
            "Facts, past episode summaries and reusable procedures are different resources. "
            "Recall first, inspect fresh evidence, reconcile changes, and cite sources. "
            "Only remember explicit useful facts; do not store guesses as facts."
        )

    def task_policy_prompt(self, state):
        return (
            "Use search_memory for facts, search_history for past outcomes, catalog_skills "
            "for descriptions and load_skill only for a relevant complete procedure. "
            "Save a reusable verification procedure when justified. Forget superseded facts "
            "with forget_fact. Use submit_report with JSON conclusion, metrics, citations, "
            "limitations; finish afterwards. Conclusion is a string, metrics is an object "
            "of numbers, citations is a list of exact source filenames (including the "
            "original source of recalled facts), and limitations is a list of strings. "
            "Memory/history search uses a literal substring, not a semantic query: "
            "try a short identifier; an empty query lists recent records. "
            "State missing evidence honestly."
        )

    # docs:start design
    def prepare(self, state):
        selected = []
        for name, version in state.selected_skills.items():
            item = self.skills.get_version(name, version)
            if item is None:
                raise ValueError("required_skill_version_missing")
            selected.append(
                {"name": name, "version": version, "instructions": item.source}
            )
        # Full selected documents are part of the current task input. A request
        # budget failure must be explicit; slicing instructions is not recall.
        return json.dumps(
            {
                "task": state.task,
                "selected_skills": selected,
                "recent_results": state.recent[-4:],
            }
        )

    def reduce(self, state, observation, decision):
        for item in observation.action_results:
            output = item.output
            if (
                isinstance(output, dict)
                and "instructions" in output
                and "sha256" in output
            ):
                state.selected_skills[output["name"]] = output["version"]
        state.recent.extend(
            item.to_model_dict(max_chars=6000) for item in observation.action_results
        )
        state.recent = state.recent[-8:]
        return state

    # docs:end design


class NotebookFactory:
    """The application owns its bound resources; composition borrows them."""

    def __init__(self, task, *, root, shared_root=None, variant="default", **kwargs):
        self.task, self.variant = task, variant
        location = Path(shared_root or root / "notebook")
        location.mkdir(parents=True, exist_ok=True)
        self.memory = MemdirMemory(str(location / "facts"), create=True)
        self.history = MemdirMemory(str(location / "episodes"), create=True)
        self.skills = SqliteToolLibrary(
            location / "skills.sqlite3", namespace="research"
        )

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.skills.close()

    def __call__(self, *, config, model, tool_registry, protocol, parser):
        @function_tool(read_only=True)
        def search_memory(query: str):
            """Search durable factual memory, not the current conversation."""
            return (
                []
                if self.variant == "no-memory"
                else [
                    asdict(row)
                    for row in self.memory.retrieve(
                        {"contains": query, "max_items": 12}
                    )
                ]
            )

        @function_tool(read_only=True)
        def search_history(query: str):
            """Recall source-labelled prior episode summaries."""
            return (
                []
                if self.variant == "no-memory"
                else [
                    asdict(row)
                    for row in self.history.retrieve(
                        {"contains": query, "max_items": 8}
                    )
                ]
            )

        @function_tool(concurrency_safe=False)
        def remember_fact(identity: str, content: str, source: str):
            """Create/update an explicitly sourced fact; identity stays stable."""
            if self.variant == "no-memory":
                return {"status": "disabled"}
            self.memory.append(
                MemoryRecord(
                    "reference", content + "\nSource: " + source, 0, record_id=identity
                )
            )
            return {"stored": identity}

        @function_tool(concurrency_safe=False)
        def forget_fact(identity: str):
            """Delete a fact only from this notebook namespace."""
            return {"removed": self.memory.delete(identity)}

        @function_tool(read_only=True)
        def catalog_skills(query: str):
            """List matching procedure names/descriptions, without loading bodies."""
            return self.skills.catalog(query, limit=8)

        @function_tool(read_only=True)
        def load_skill(name: str):
            """Load one complete procedure with exact version and source digest."""
            item = self.skills.get(name)
            if item is None or not item.active:
                raise ValueError("skill_unavailable")
            return {
                "name": name,
                "version": item.version,
                "instructions": item.source,
                "sha256": hashlib.sha256(item.source.encode()).hexdigest(),
                "provenance": item.metadata,
            }

        @function_tool(concurrency_safe=False)
        def save_procedure(name: str, description: str, instructions: str, source: str):
            """Store a procedural document, not verified executable code."""
            item = self.skills.add_or_update(
                ToolArtifact(
                    name,
                    description,
                    instructions,
                    metadata={"source": source, "validation": "document_only"},
                )
            )
            return {"name": item.name, "version": item.version}

        @function_tool(concurrency_safe=False)
        def submit_report(report_json: str):
            """Submit structured findings and persist a labelled episode summary."""
            report = json.loads(report_json)
            if not isinstance(report, dict):
                raise ValueError("report_must_be_object")
            if self.variant != "no-memory":
                self.history.append(
                    MemoryRecord(
                        "runtime",
                        json.dumps(
                            {
                                "task": self.task.get("id", self.task["task"]),
                                "report": report,
                                "verification": "not_yet_independently_checked",
                            }
                        ),
                        0,
                    )
                )
            return {"submitted_report": report}

        for tool in (
            search_memory,
            search_history,
            remember_fact,
            forget_fact,
            catalog_skills,
            load_skill,
            save_procedure,
            submit_report,
        ):
            tool_registry.register(tool)
        return ResearchAgent(
            llm=model,
            tool_registry=tool_registry,
            skills=self.skills,
            model_protocol=protocol.id,
            max_steps=config.max_steps,
            model_parser=parser,
        )


def build_factory(task, **resources):
    return NotebookFactory(task, **resources)
```

```python title="src/qitos_lab_hermes/evaluate.py" theme={null}
"""Controller-side evaluator, never staged into the agent's workspace."""

import math


def evaluate(result, task):
    submitted = []
    read_sources = set()
    for record in result.records:
        for item in record.action_results:
            output = item.output
            if (
                item.tool_name == "read_file"
                and item.status == "success"
                and isinstance(output, dict)
            ):
                if isinstance(output.get("path"), str):
                    read_sources.add(output["path"])
            if (
                item.tool_name == "submit_report"
                and item.status == "success"
                and isinstance(output, dict)
            ):
                if "submitted_report" in output:
                    submitted.append(output["submitted_report"])
    report = submitted[-1] if submitted and isinstance(submitted[-1], dict) else {}
    citations = report.get("citations", [])
    valid_citations = isinstance(citations, list) and all(
        isinstance(item, str) for item in citations
    )
    limitations = report.get("limitations", [])
    checks = {
        "report_submitted": bool(submitted),
        "conclusion": isinstance(report.get("conclusion"), str)
        and bool(report.get("conclusion")),
        "limitations": isinstance(limitations, list)
        and bool(limitations)
        and all(isinstance(item, str) and item.strip() for item in limitations),
        "citations": valid_citations
        and set(task["required_sources"]) <= set(citations),
        "actual_reads": (set(task["required_sources"]) - {"protocol.md"})
        <= read_sources,
        "final": str(result.state.stop_reason) == "final",
    }
    metrics = report.get("metrics", {})
    for key, expected in task["expected_metrics"].items():
        value = metrics.get(key) if isinstance(metrics, dict) else None
        checks[key] = type(value) in (int, float) and math.isclose(
            value, expected, rel_tol=0.005, abs_tol=0.005
        )
    return {"passed": all(checks.values()), "checks": checks, "submitted": report}
```

```json title="src/qitos_lab_hermes/tasks.json" theme={null}
[
  {
    "task": "Recall the previously learned protocol and procedure from the durable notebook, then Audit whether candidate B reduces latency without reducing reliability. Reconcile the preliminary memo with raw data and the registered weighting rule. Compute weighted_latency_a, weighted_latency_b, latency_reduction_percent and failure_rate_b_percent. Produce a qualified recommendation, not a blanket claim.",
    "inputs": {
      "protocol.md": "Protocol R17. Compare weighted latency using production mix 80% small, 20% large. Report failure rate over all attempted requests; do not remove failures. Latency measurements are milliseconds for successful requests. This is a synthetic teaching dataset, not a real deployment benchmark.\n",
      "results.csv": "group,latency_a,latency_b,attempts_b,failures_b\nsmall,100,80,800,8\nlarge,500,450,200,12\n",
      "preliminary.md": "PRELIMINARY, not validated: candidate B halves latency and has no errors. Based only on a smoke test. Superseded by results.csv.\n",
      "operations.md": "Reliability policy: failure rate must be <=1%. A performance improvement alone is insufficient for adoption. Error recovery must be investigated separately.\n",
      "limitations.md": "Two workload groups, one run. No confidence interval or long-duration measurement is available.\n"
    },
    "required_sources": [
      "protocol.md",
      "results.csv",
      "operations.md"
    ],
    "expected_metrics": {
      "weighted_latency_a": 180,
      "weighted_latency_b": 154,
      "latency_reduction_percent": 14.444444,
      "failure_rate_b_percent": 2
    },
    "learning_task": "Read all inputs. Remember this project's protocol with explicit source: The registered mixture for R17 is 80% small and 20% large; reliability threshold is at most 1%. Preserve that protocol when evaluating a later run. Save a reusable checking procedure under r17-audit. Submit a concise structured report of what was learned. A later independent Session will receive fresh data without the original protocol file.",
    "recall_inputs": {
      "results.csv": "group,latency_a,latency_b,attempts_b,failures_b\nsmall,100,80,800,8\nlarge,500,450,200,12\n",
      "preliminary.md": "PRELIMINARY, not validated: candidate B halves latency and has no errors. Based only on a smoke test. Superseded by results.csv.\n",
      "operations.md": "Reliability policy: failure rate must be <=1%. A performance improvement alone is insufficient for adoption. Error recovery must be investigated separately.\n",
      "limitations.md": "Two workload groups, one run. No confidence interval or long-duration measurement is available.\n",
      "continuation.md": "This is a new Session. The original protocol remains authoritative and is in the previously approved notebook. Do not invent a missing protocol."
    }
  },
  {
    "task": "Recall the previously learned protocol and procedure from the durable notebook, then Audit experiment R18 for sample leakage and subgroup regressions. Compute valid_samples, excluded_samples, baseline_accuracy_percent, candidate_accuracy_percent, minority_change_points using only valid rows. Explain why aggregate results are insufficient for adoption.",
    "inputs": {
      "protocol.md": "R18: exclude rows with split=leaked. Accuracy = total correct / total count. Report minority change as candidate accuracy minus baseline accuracy, in percentage points. Real deployment qualification is out of scope.\n",
      "results.csv": "group,split,count,correct_a,correct_b\nmajority,valid,80,64,72\nminority,valid,20,16,12\nmajority,leaked,10,10,10\n",
      "preliminary.md": "An early memo claims all subgroups improved. It included leaked samples and did not break down minority outcomes.\n",
      "operations.md": "Adoption requires no subgroup regression. Minority group is small; collect more evidence before a final deployment decision.\n",
      "limitations.md": "Synthetic teaching experiment, no repeat runs or variance estimates.\n"
    },
    "required_sources": [
      "protocol.md",
      "results.csv",
      "operations.md"
    ],
    "expected_metrics": {
      "valid_samples": 100,
      "excluded_samples": 10,
      "baseline_accuracy_percent": 80,
      "candidate_accuracy_percent": 84,
      "minority_change_points": -20
    },
    "learning_task": "Read all inputs. Remember this project's protocol with explicit source: R18 excludes the leaked split. Report minority accuracy separately; never let majority weighting hide subgroup harm. Save a reusable checking procedure under r18-audit. Submit a concise structured report of what was learned. A later independent Session will receive fresh data without the original protocol file.",
    "recall_inputs": {
      "results.csv": "group,split,count,correct_a,correct_b\nmajority,valid,80,64,72\nminority,valid,20,16,12\nmajority,leaked,10,10,10\n",
      "preliminary.md": "An early memo claims all subgroups improved. It included leaked samples and did not break down minority outcomes.\n",
      "operations.md": "Adoption requires no subgroup regression. Minority group is small; collect more evidence before a final deployment decision.\n",
      "limitations.md": "Synthetic teaching experiment, no repeat runs or variance estimates.\n",
      "continuation.md": "This is a new Session. The original protocol remains authoritative and is in the previously approved notebook. Do not invent a missing protocol."
    }
  },
  {
    "task": "Recall the previously learned protocol and procedure from the durable notebook, then Reconcile R19 throughput and recovery claims. Compute steady_throughput, recovery_seconds, availability_percent from the authoritative log under the registered formula. Explain the contradiction with the sales memo and distinguish process recovery from external exactly-once effects.",
    "inputs": {
      "protocol.md": "R19 window is 100 seconds. Availability excludes seconds explicitly marked unavailable. Steady throughput excludes recovery and equals steady completed tasks / steady seconds. Recovery duration is restored_at minus failure_at.\n",
      "results.csv": "steady_tasks,steady_seconds,failure_at,restored_at,unavailable_seconds\n720,90,30,40,10\n",
      "preliminary.md": "Sales draft: zero downtime, throughput 10/s, exactly-once external effects. These statements were never independently tested.\n",
      "operations.md": "Logs prove process restoration only. External write acknowledgement was lost; effect reconciliation remains unresolved. Do not equate retry success with exactly-once delivery.\n",
      "limitations.md": "One induced failure; no cross-host replication or high-availability evidence.\n"
    },
    "required_sources": [
      "protocol.md",
      "results.csv",
      "operations.md"
    ],
    "expected_metrics": {
      "steady_throughput": 8,
      "recovery_seconds": 10,
      "availability_percent": 90
    },
    "learning_task": "Read all inputs. Remember this project's protocol with explicit source: R19 recovery time is the unavailable interval, not whole observation duration. Throughput uses only steady-state seconds. Do not claim external exactly-once from local receipts. Save a reusable checking procedure under r19-audit. Submit a concise structured report of what was learned. A later independent Session will receive fresh data without the original protocol file.",
    "recall_inputs": {
      "results.csv": "steady_tasks,steady_seconds,failure_at,restored_at,unavailable_seconds\n720,90,30,40,10\n",
      "preliminary.md": "Sales draft: zero downtime, throughput 10/s, exactly-once external effects. These statements were never independently tested.\n",
      "operations.md": "Logs prove process restoration only. External write acknowledgement was lost; effect reconciliation remains unresolved. Do not equate retry success with exactly-once delivery.\n",
      "limitations.md": "One induced failure; no cross-host replication or high-availability evidence.\n",
      "continuation.md": "This is a new Session. The original protocol remains authoritative and is in the previously approved notebook. Do not invent a missing protocol."
    }
  }
]
```

```python title="src/qitos_lab_hermes/__main__.py" theme={null}
"""Explicit launch configuration; no credentials or model calls during validate."""

import argparse
from dataclasses import asdict, replace
from contextlib import ExitStack
from importlib.resources import files
import json
from pathlib import Path
import sys

from qitos.config import (
    BudgetConfig,
    LocalCredentialFileResolver,
    SessionConfig,
    TrajectoryConfig,
    build_agent_composition,
    load_agent_config,
)

from .agent import build_factory
from .evaluate import evaluate


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=["validate", "run", "resume", "inspect"])
    parser.add_argument("--config", type=Path, required=True)
    parser.add_argument("--model-config", type=Path)
    parser.add_argument("--credentials", type=Path)
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--task", type=int, choices=[0, 1, 2], default=0)
    parser.add_argument("--session")
    parser.add_argument("--shared-root", type=Path)
    parser.add_argument("--phase", choices=["learn", "recall"], default="recall")
    parser.add_argument("--live", action="store_true")
    parser.add_argument(
        "--variant",
        choices=["default", "static", "no-memory", "no-skills"],
        default="default",
    )
    args = parser.parse_args()
    config = load_agent_config(args.config)
    tasks = json.loads(files(__package__).joinpath("tasks.json").read_text())
    task = tasks[args.task]
    if args.phase == "learn":
        task = dict(task, task=task.get("learning_task", task["task"]))
    elif "recall_inputs" in task:
        task = dict(task, inputs=task["recall_inputs"])
    if args.command == "validate":
        print(
            json.dumps(
                {"status": "configuration_valid", "tasks": len(tasks), "live": False}
            )
        )
        return 0
    root = args.root.resolve()
    if any((parent / ".git").exists() for parent in (root, *root.parents)):
        raise ValueError("run_root_must_be_outside_git")
    root.mkdir(parents=True, exist_ok=True, mode=0o700)
    if args.shared_root is not None:
        shared = args.shared_root.resolve()
        if any((parent / ".git").exists() for parent in (shared, *shared.parents)):
            raise ValueError("shared_root_must_be_outside_git")
    if args.command == "inspect":
        from qitos.qita.reader import candidate_file_reader

        reader = candidate_file_reader(root / "trajectory.journal")
        print(json.dumps({"runs": [asdict(run) for run in reader.discover_runs()]}))
        return 0
    if not args.live or args.credentials is None:
        raise ValueError("explicit_live_and_credentials_required")
    if args.model_config is not None:
        private = args.model_config.resolve()
        if any((parent / ".git").exists() for parent in private.parents):
            raise ValueError("model_config_must_be_outside_git")
        config = replace(config, model=load_agent_config(private).model)
    output_limit = max(10240, config.model.request.max_tokens or 0)
    request = replace(config.model.request, max_tokens=output_limit)
    config = replace(
        config, model=replace(config.model, request=request, max_tokens=output_limit)
    )
    workspace = root / "input"
    if args.command == "run":
        workspace.mkdir(exist_ok=False)
        for name, content in task["inputs"].items():
            target = workspace / name
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(content)
    config = replace(
        config,
        budgets=config.budgets
        or BudgetConfig(max_steps=80, max_requests=80, max_runtime_seconds=3600),
        runtime=replace(
            config.runtime,
            environment=replace(config.runtime.environment, workspace=str(workspace)),
            session=SessionConfig(store="sqlite", path=str(root / "sessions.sqlite3")),
            trajectory=TrajectoryConfig(output=str(root / "trajectory.journal")),
        ),
    )
    resolver = LocalCredentialFileResolver(args.credentials, repository_root=Path.cwd())
    with ExitStack() as stack:
        factory = stack.enter_context(
            build_factory(
                task, root=root, shared_root=args.shared_root, variant=args.variant
            )
        )
        composition = stack.enter_context(
            build_agent_composition(
                config, credential_resolver=resolver, agent_factory=factory
            )
        )
        session = (
            composition.restore(args.session)
            if args.command == "resume"
            else composition.session(task["task"])
        )
        (root / "session.json").write_text(
            json.dumps({"session_id": session.session_id.value})
        )
        result = session.run()

        def successful(name):
            return [
                outcome.output
                for record in result.records
                for outcome in record.action_results
                if outcome.tool_name == name and outcome.status == "success"
            ]

        verdict = evaluate(result, task)
        if args.phase == "learn":
            checks = {
                "final": str(result.state.stop_reason) == "final",
                "facts_persisted": bool(factory.memory.retrieve()),
                "procedure_persisted": bool(factory.skills.list_active()),
                "facts_written_this_session": bool(successful("remember_fact")),
                "procedure_written_this_session": bool(successful("save_procedure")),
                "sources_read": result.tool_calls_by_name.get("read_file", 0) >= 3,
            }
            verdict = {
                "passed": all(checks.values()),
                "checks": checks,
                "phase": "learn",
            }
        elif args.variant != "no-memory":
            checks = verdict["checks"]
            checks["memory_recalled"] = any(
                bool(value) for value in successful("search_memory")
            )
            checks["skill_loaded"] = bool(result.state.selected_skills)
            verdict["passed"] = all(checks.values())
        report = {
            "session_id": session.session_id.value,
            "run_id": result.run_id,
            "stop_reason": str(result.state.stop_reason),
            "evaluation": verdict,
            "tool_calls": result.tool_calls_by_name,
            "variant": args.variant,
        }
        (root / "report.json").write_text(
            json.dumps(report, ensure_ascii=False, indent=2)
        )
        print(json.dumps(report, ensure_ascii=False))
        return 0 if verdict["passed"] else 2


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as error:
        # Raw transport/factory errors can contain private endpoint details.
        print(
            json.dumps({"status": "failed", "error_type": type(error).__name__}),
            file=sys.stderr,
        )
        raise SystemExit(2) from None
```