> ## 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. # Voyager-inspired executable skills > Professional, replaceable agent designs; course qualification in progress. Status: in development, not course-qualified. The working source below is not a claim of a passed 3×3 live matrix. See the [execution ledger](/tutorials/agent-design-lab-results) for qualified facts and open gaps. ## Design and adaptation boundary Learn through execution feedback and retain reusable programs. The adaptation uses Docker and data programs, not Minecraft; open-ended exploration and embedding retrieval are not reproduced. [Primary source](https://voyager.minedojo.org/). ## Framework versus application CurriculumState tracks mastery and actual loading. publish\_skill runs controller checks and binds verified source/artifact digests. load\_skill writes selected code only through the configured Env, never exec on the host. 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 curriculum ordering, requiring the same verification before mastery and retaining unsuccessful attempts. Composition: Compose normalize and weighted skills in the summarize objective; compare with no-skills in an isolated library. 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. ```bash theme={null} python -m pip install . python -m qitos_lab_voyager validate --config agent.yaml --root /tmp/lab-validation python -m qitos_lab_voyager run --config agent.yaml --model-config /private-config/model.yaml --credentials /private-config/credentials.yaml --root /private-runs/voyager-attempt --task 0 --live --phase learn --shared-root /private-runs/shared-notebook ``` Complete implementation: `examples/projects/voyager_skills/src/qitos_lab_voyager/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. ```python theme={null} def prepare(self, state): return json.dumps( { "objective": state.task, "mastered": state.mastered, "actually_loaded": state.reused, "feedback": state.feedback[-3:], } ) def reduce(self, state, observation, decision): for item in observation.action_results: value = item.output if isinstance(value, dict): if value.get("published"): state.mastered = sorted(set(state.mastered + [value["published"]])) if value.get("loaded"): state.reused.append(value["loaded"]) state.feedback.append(item.to_model_dict(max_chars=6000)) state.feedback = state.feedback[-6:] return state ``` ## Complete files: save in the project root ```python title="src/qitos_lab_voyager/__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() verdict = evaluate(result, task) checks = verdict["checks"] if args.variant != "no-skills": stored = factory.library.get(task["skill"]) checks["verified_skill_persisted"] = ( stored is not None and stored.metadata.get("verified") is True ) if args.phase == "recall": checks["actual_skill_reuse"] = task["skill"] in result.state.reused 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 ``` ```json title="src/qitos_lab_voyager/tasks.json" theme={null} [ { "id": "V31", "task": "Learn a reusable normalize_rows(rows) function in skill.py: trim all string fields, parse value as float, count as nonnegative int, reject missing/invalid data. Verify and publish it as normalize. On reuse, load and execute the existing skill; do not regenerate it.", "inputs": { "requirements.md": "Use only stdlib. normalize_rows returns fresh dictionaries, preserves other fields and rejects negative count. Never mutate input.\n" }, "outputs": [ "skill.py" ], "checks": "from skill import normalize_rows\nr=[{'name':' x ','value':'2.5','count':'3','tag':'keep'}]\no=normalize_rows(r)\nassert o==[{'name':'x','value':2.5,'count':3,'tag':'keep'}]\nassert r[0]['value']=='2.5'\ntry: normalize_rows([{'value':'x','count':'1'}])\nexcept ValueError: pass\nelse: raise AssertionError('invalid row accepted')\ntry: normalize_rows([{'value':'2','count':'-1'}])\nexcept ValueError: pass\nelse: raise AssertionError('negative count accepted')\n", "skill": "normalize" }, { "id": "V32", "task": "Learn a reusable weighted_mean(rows) function in skill.py for numeric value/count dicts: exact weighted result, empty or zero total rejects, negative weights reject. Verify/publish as weighted. On reuse, compose with previously learned normalize if available.", "inputs": { "requirements.md": "weighted_mean accepts numeric dictionaries. Use stdlib only; do not round intermediate values.\n" }, "outputs": [ "skill.py" ], "checks": "from skill import weighted_mean\nassert weighted_mean([{'value':10,'count':8},{'value':30,'count':2}])==14\nfor rows in ([],[{'value':10,'count':0}],[{'value':2,'count':-1}]):\n try: weighted_mean(rows)\n except ValueError: pass\n else: raise AssertionError('invalid weights accepted')\n", "skill": "weighted" }, { "id": "V33", "task": "Compose reusable summarize(rows) in skill.py. Use previously learned normalize and weighted skills if available. Return per-group weighted means, preserving group isolation. Verify/publish as summarize. On reuse, actually load and execute the published code.", "inputs": { "requirements.md": "summarize returns dict group -> weighted mean. Strip group whitespace, parse numeric strings, do not mutate input. Empty input returns {}. Use stdlib.\n" }, "outputs": [ "skill.py" ], "checks": "from skill import summarize\nr=[{'group':' a ','value':'10','count':'8'},{'group':'a','value':'30','count':'2'},{'group':'b','value':'5','count':'1'}]\nassert summarize(r)=={'a':14,'b':5}\nassert r[0]['group']==' a '\nassert summarize([])=={}\n", "skill": "summarize" } ] ``` ```python title="src/qitos_lab_voyager/evaluate.py" theme={null} """Read successful controller-owned verification, not the model's final claim.""" def evaluate(result, task, *, prior_records=()): verified_step = -1 edited_step = -1 source_digests = {} for record in [*prior_records, *result.records]: for item in record.action_results: if item.tool_name in { "write_file", "edit_file", "run_command", "load_skill", }: edited_step = record.step_id if item.tool_name == "verify_project" and item.status == "success": output = item.output if isinstance(output, dict) and output.get("verified") is True: verified_step = record.step_id source_digests = output.get("source_digests", {}) else: verified_step = -1 checks = { "independent_checks": verified_step >= 0, "no_edits_after_verification": verified_step >= edited_step, "all_outputs_captured": set(task["outputs"]) == set(source_digests), "read_before_edit": result.tool_calls_by_name.get("read_file", 0) > 0, "final": str(result.state.stop_reason) == "final", } return { "passed": all(checks.values()), "checks": checks, "source_digests": source_digests, } ``` ```python title="src/qitos_lab_voyager/agent.py" theme={null} """Automatic curriculum state and verified executable skill accumulation.""" from dataclasses import dataclass, field import hashlib from pathlib import Path import re from typing import Any, Dict, Optional import json from qitos.core.agent_module import AgentModule from qitos.core.function_tool_decorator import function_tool from qitos.core.state import StateSchema from qitos.core.tool_result import ToolResult from qitos.core.tool import ToolPermission from qitos.kit.tool.library.base import ToolArtifact from qitos.kit.tool.library.sqlite_store import SqliteToolLibrary from .extension import verification_tools @dataclass class CurriculumState(StateSchema): mastered: list[str] = field(default_factory=list) reused: list[str] = field(default_factory=list) feedback: list[dict] = field(default_factory=list) class SkillAgent(AgentModule): def init_state(self, task, **kwargs): return CurriculumState(task=task, max_steps=self.config.get("max_steps", 80)) def base_persona_prompt(self, state): return ( "Learn reusable Python skills through actual execution feedback. " "This is a Docker data-programming adaptation of Voyager, not Minecraft. " "Inspect requirements, search the skill catalog, load useful complete programs, " "compose them when appropriate, test, and only publish verified programs." ) def task_policy_prompt(self, state): return ( "Write the current solution to skill.py. catalog_skills returns only descriptions; " "load_skill retrieves one full version into the sandbox, using skill.py for the " "current objective and name.py for dependencies. Use verify_project for feedback, " "then publish_skill. Never claim a failed program has been mastered. " "The current objective is the next curriculum item; previous skills persist " "across processes but the environment is disposable." ) # docs:start design def prepare(self, state): return json.dumps( { "objective": state.task, "mastered": state.mastered, "actually_loaded": state.reused, "feedback": state.feedback[-3:], } ) def reduce(self, state, observation, decision): for item in observation.action_results: value = item.output if isinstance(value, dict): if value.get("published"): state.mastered = sorted(set(state.mastered + [value["published"]])) if value.get("loaded"): state.reused.append(value["loaded"]) state.feedback.append(item.to_model_dict(max_chars=6000)) state.feedback = state.feedback[-6:] return state # docs:end design class SkillFactory: def __init__(self, task, *, root, shared_root=None, variant="default", **kwargs): self.task, self.variant = task, variant location = Path(shared_root or root / "library") location.mkdir(parents=True, exist_ok=True) self.library = SqliteToolLibrary( location / "skills.sqlite3", namespace="data-programming" ) def __enter__(self): return self def __exit__(self, *args): self.library.close() def __call__(self, *, config, model, tool_registry, protocol, parser): verify = verification_tools(self.task)[0] @function_tool(read_only=True) def catalog_skills(query: str): """Search descriptions without loading executable bodies.""" return ( [] if self.variant == "no-skills" else self.library.catalog(query, limit=8) ) @function_tool( required_ops=["file"], permissions=ToolPermission(filesystem_write=True), concurrency_safe=False, ) def load_skill(name: str, runtime_context: Optional[Dict[str, Any]] = None): """Load an exact verified skill version into the configured sandbox.""" if not re.fullmatch(r"[a-z][a-z0-9_]{0,48}", name): raise ValueError("invalid_skill_name") item = None if self.variant == "no-skills" else self.library.get(name) if ( item is None or not item.active or item.metadata.get("verified") is not True ): raise ValueError("verified_skill_unavailable") target = "skill.py" if name == self.task["skill"] else name + ".py" (runtime_context or {})["ops"]["file"].atomic_write_text( target, item.source ) return { "loaded": name, "version": item.version, "path": target, "source": item.source, "sha256": hashlib.sha256(item.source.encode()).hexdigest(), "provenance": item.metadata, } @function_tool( required_ops=["file", "process"], permissions=ToolPermission(filesystem_read=True, command=True), concurrency_safe=False, ) def publish_skill( description: str, runtime_context: Optional[Dict[str, Any]] = None ): """Re-test exact source; publish only on a controller-owned passing receipt.""" context = runtime_context or {} tested = verify.execute({}, context) if not tested.output.get("verified"): return tested source = context["ops"]["file"].read_text("skill.py") if ( hashlib.sha256(source.encode()).hexdigest() != tested.output["source_digests"]["skill.py"] ): raise ValueError("skill_changed_after_validation") name = self.task["skill"] if self.variant == "no-skills": return ToolResult( output={ "verified": True, "published": None, "persistence": "disabled", }, tool_name="publish_skill", ) item = self.library.add_or_update( ToolArtifact( name, description, source, metadata={ "verified": True, "objective": self.task["id"], "checks_digest": tested.output["checks_digest"], "artifacts": [ref.to_dict() for ref in tested.artifact_refs], }, ) ) return ToolResult( output={"published": name, "version": item.version, "verified": True}, tool_name="publish_skill", artifact_refs=tested.artifact_refs, ) for tool in (verify, catalog_skills, load_skill, publish_skill): tool_registry.register(tool) return SkillAgent( llm=model, tool_registry=tool_registry, model_protocol=protocol.id, max_steps=config.max_steps, model_parser=parser, ) def build_factory(task, **resources): return SkillFactory(task, **resources) ``` ```python title="src/qitos_lab_voyager/extension.py" theme={null} """Installable project extension. All generated-code execution stays in Env.""" import hashlib import shlex from typing import Optional, Dict, Any from qitos.core.artifact import ArtifactRef from qitos.core.function_tool_decorator import function_tool from qitos.core.tool import ToolPermission from qitos.core.tool_result import ToolResult def verification_tools(task): @function_tool( required_ops=["file", "process"], permissions=ToolPermission(filesystem_read=True, command=True), concurrency_safe=False, ) def verify_project(runtime_context: Optional[Dict[str, Any]] = None): """Run controller-owned checks, capture exact tested source artifacts.""" context = runtime_context or {} fs, process = context["ops"]["file"], context["ops"]["process"] bodies = {path: fs.read_text(path).encode() for path in task["outputs"]} outcome = dict( process.run("python -c " + shlex.quote(task["checks"]), timeout=30) ) unchanged = all( fs.read_text(path).encode() == body for path, body in bodies.items() ) passed = ( outcome.get("returncode") == 0 and not outcome.get("outcome_unknown") and unchanged ) refs = [] for path, body in bodies.items(): digest = hashlib.sha256(body).hexdigest() ref = ArtifactRef( artifact_id="sha256:" + digest, resolver_key="tool-result-output", sha256=digest, byte_length=len(body), media_type="text/x-python", ) context["artifact_resolver"].put(ref, body) refs.append(ref) report = { "verified": passed, "source_digests": { path: hashlib.sha256(body).hexdigest() for path, body in bodies.items() }, "checks_digest": hashlib.sha256(task["checks"].encode()).hexdigest(), "returncode": outcome.get("returncode"), "source_unchanged": unchanged, "feedback": str(outcome.get("stderr", ""))[-3000:], } return ToolResult( output=report, model_output=report, artifact_refs=tuple(refs), tool_name="verify_project", ) return (verify_project,) ``` ```python title="src/qitos_lab_voyager/__init__.py" theme={null} """Persistent learning application over QitOS.""" ``` ```toml title="pyproject.toml" theme={null} [build-system] requires = ["setuptools>=68"] build-backend = "setuptools.build_meta" [project] name = "qitos-lab-voyager" version = "0.1.0" requires-python = ">=3.10" dependencies = ["qitos[openai]"] [tool.setuptools.packages.find] where = ["src"] [tool.setuptools.package-data] qitos_lab_voyager = ["tasks.json"] ``` ```yaml title="agent.yaml" theme={null} schema: qitos.agent agent: name: design-lab-voyager-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 ``` ```python title="src/qitos_lab_voyager/curriculum.py" theme={null} """A bounded mastery-driven curriculum; ordering is replaceable application policy.""" import argparse from importlib.resources import files import json from pathlib import Path import subprocess import sys from qitos.kit.tool.library.sqlite_store import SqliteToolLibrary def next_objective(tasks, library): """Advance only past controller-verified persisted skills, not model claims.""" for index, task in enumerate(tasks): skill = library.get(task["skill"]) if ( skill is None or not skill.active or skill.metadata.get("verified") is not True ): return index return None def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, required=True) parser.add_argument("--config", type=Path, required=True) parser.add_argument("--model-config", type=Path, required=True) parser.add_argument("--credentials", type=Path, required=True) parser.add_argument("--live", action="store_true") args = parser.parse_args() if not args.live: raise ValueError("explicit_live_required") root = args.root.resolve() if any((parent / ".git").exists() for parent in (root, *root.parents)): raise ValueError("curriculum_root_must_be_outside_git") root.mkdir(parents=True, exist_ok=False, mode=0o700) shared = root / "skills" shared.mkdir() tasks = json.loads(files(__package__).joinpath("tasks.json").read_text()) with SqliteToolLibrary( shared / "skills.sqlite3", namespace="data-programming" ) as library: for attempt in range(len(tasks)): index = next_objective(tasks, library) if index is None: break result = subprocess.run( [ sys.executable, "-m", "qitos_lab_voyager", "run", "--config", str(args.config.resolve()), "--model-config", str(args.model_config.resolve()), "--credentials", str(args.credentials.resolve()), "--root", str(root / f"objective-{attempt}"), "--shared-root", str(shared), "--task", str(index), "--phase", "learn", "--live", ], timeout=3900, ) if result.returncode: print( json.dumps( { "status": "objective_unmastered", "objective": tasks[index]["id"], } ) ) return 2 complete = next_objective(tasks, library) is None print( json.dumps( { "status": ( "curriculum_complete" if complete else "curriculum_incomplete" ), "skills": library.catalog(), } ) ) return 0 if complete else 2 if __name__ == "__main__": raise SystemExit(main()) ```