Design and adaptation boundary
Separate factual memory, episode retrieval and procedural skills. A catalog advertises descriptions; only a selected skill loads its complete body. Primary source.Framework versus application
Memdir facts and episodes have separate roots. SqliteToolLibrary persists procedures with versions and source metadata. A learning process and a recall process share only explicitly bound storage. QitOS supplies model transactions/usage, tool permission/validation, Env execution, Session, ArtifactRef and Trajectory. The application supplies tasks, policy, independent acceptance checks and memory/skill selection. This iteration adds custom agent_factory composition, persistent skill revisions, full-body selection, explicit Memdir deletion and correct artifact data authority. The core design increment is in agent.py below. The CLI shows configuration and resource ownership explicitly; evaluate.py is the controller checker. Tasks use repository-owned synthetic professional scenarios, not paper benchmarks or customer data.Install, configure and run
First install the QitOS wheel built from this iteration; the current PyPI release cannot stand in for unpublished APIs. Then install this project. Keep actual addresses and credentials outside Git. The private model file is a full qitos.agent configuration; this launcher selects only its model section, never an environment-variable key. Run —phase learn first, then —phase recall with a new —root and the same external —shared-root. Do not reuse the old input directory. The output default is 10,240 and may be raised in private model configuration. Task request/step/time guards come from configuration. validate does not call a model; —live is mandatory for execution. Docker failure must not silently fall back to the host. Retain unsuccessful results and human interventions.Verification, exercise and composition
Independent checks examine sources/numbers or executed code, not a model’s success claim. Plan revisions, actual skill loading and child identities require separate mechanism evidence. Session restore is not filesystem rollback. Generated code executes only in the restricted Env. Exercise: Replace lexical recall with another MemorySource, retaining namespace isolation and explicit forgetting. Composition: Supply approved protocol facts to PlanAct; never treat an unverified episode summary as checked evidence. The required matrix is three tasks, three repetitions each. ReAct/PlanAct share tasks; static planning, no-memory and no-skills are explicit controls. A single pass is not a performance result. Raw traces stay private until redistribution and sanitization checks authorize a derived publication.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.
Extracted from complete source: the design increment
This excerpt is generated from the complete project, not a separately maintained implementation. Complete installable files follow. 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
Complete files: save in the project root
agent.yaml
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
pyproject.toml
[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"]
src/qitos_lab_hermes/__init__.py
"""Persistent learning application over QitOS."""
src/qitos_lab_hermes/agent.py
"""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)
src/qitos_lab_hermes/evaluate.py
"""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}
src/qitos_lab_hermes/tasks.json
[
{
"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."
}
}
]
src/qitos_lab_hermes/__main__.py
"""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