hylo-cli¶

the harness subcommands of hylo (run, evolve, learn, ...); hylo itself is in hylo-core

Repository hylo-harness, packages/hylo-cli. Licence: Apache-2.0. Depends on: hylo-core, hylo-eval, hylo-agents, hylo-memory, hylo-evolve, hylo-learn.

Registered entry points

Group	Name	Object
commands	`artifact`	`hylo.cli.commands:artifact`
commands	`calibrate`	`hylo.cli.commands:calibrate`
commands	`conformance`	`hylo.cli.commands:conformance`
commands	`contamination`	`hylo.cli.commands:contamination`
commands	`evolve`	`hylo.cli.commands:evolve`
commands	`ingest`	`hylo.cli.commands:ingest`
commands	`label`	`hylo.cli.commands:label`
commands	`learn`	`hylo.cli.commands:learn`
commands	`plugin`	`hylo.cli.commands:plugin`
commands	`plugins`	`hylo.cli.commands:plugins`
commands	`run`	`hylo.cli.commands:run`
commands	`schema`	`hylo.cli.commands:schema`
commands	`skills`	`hylo.cli.commands:skills`
commands	`trace`	`hylo.cli.commands:trace`

Module	Summary
`hylo.cli`	Hylo command line
`hylo.cli.commands`	The harness's subcommands of `hylo`, one entry point each in the "hylo.commands" group (core.cli)
`hylo.cli.main`	The harness's `hylo` subcommands (registered through commands.py; dispatched by core.cli)
`hylo.cli.scaffold`	`hylo plugin new GROUP NAME`: a working plugin package to start from, and `hylo plugin check GROUP NAME`

`hylo.cli`¶

Hylo command line.

`hylo.cli.commands`¶

The harness's subcommands of hylo, one entry point each in the "hylo.commands" group (core.cli).

Each forwards to main.py's parser, so hylo run ... and python -m hylo.cli.main run ... are the same.

HELP `module-attribute` ¶

HELP = {'run': 'run an experiment config', 'plugins': 'list plugins', 'schema': 'print a contract JSON Schema', 'trace': 'summarise a run or an episode trace', 'evolve': 'improvement campaigns: run, status, approve, rollback, shadow', 'skills': 'archive -> skill library', 'learn': 'run a learner job (TrainJob JSON) and seal its artifact', 'calibrate': 'calibrate a success judge against simulator outcomes', 'ingest': 'ingest an external data source as a run', 'label': 'label an ingested run with a calibrated success judge', 'contamination': 'check external runs against gate / held-out init states', 'artifact': 'verify a sealed artifact and print its manifest', 'plugin': 'scaffold a plugin package (new) or check one (check)', 'conformance': "check a config's environment (and agent) against the contracts"}

`hylo.cli.main`¶

The harness's hylo subcommands (registered through commands.py; dispatched by core.cli).

    hylo run CONFIG [--split SPLIT] [--output DIR] [--set key=value ...]
                                                     run an experiment config, print metrics
    hylo plugins [--all] [--json]                      list plugins: contract version, maturity, licence
    hylo schema [experiment|task|env-spec|candidate|memory-item]   print a contract's JSON Schema
    hylo trace RUN_DIR [EPISODE_ID]                    summarise a run or one episode trace
    hylo evolve run EVOLVE_CONFIG [--set key=value]    run (or resume) an improvement campaign
    hylo evolve status|approve|rollback CONFIG_OR_DIR  inspect, approve a pending promotion, roll back
    hylo skills ARCHIVE --out LIB [--tags ...] [--env E]   turn an improvement archive into a skill library
    hylo conformance CONFIG [--agent] [--set ...]      run the conformance kit on a config's env (and agent)

SCHEMAS `module-attribute` ¶

SCHEMAS = {'experiment': ('hylo.core.config', 'ExperimentConfig'), 'task': ('hylo.core.task', 'TaskSpec'), 'env-spec': ('hylo.core.types', 'EnvSpec'), 'agent-spec': ('hylo.core.agent', 'AgentSpec'), 'candidate': ('hylo.evolve.candidate', 'Candidate'), 'memory-item': ('hylo.core.memory', 'MemoryItem'), 'benchmark': ('hylo.eval.benchmark', 'Benchmark')}

PASSTHROUGH `module-attribute` ¶

PASSTHROUGH = {'skills': _skills, 'calibrate': _calibrate}

main ¶

main(argv: list[str] | None = None) -> int

`hylo.cli.scaffold`¶

hylo plugin new GROUP NAME: a working plugin package to start from, and hylo plugin check GROUP NAME.

Every template passes its conformance check as generated; replace the body with your method. The package is an ordinary Python distribution with an entry point, so Hylo finds it once it is installed (uv pip install -e <dir> or add it to a workspace). See docs/extending.md.

TEMPLATES `module-attribute` ¶

TEMPLATES: dict[str, tuple[str, str, str, str]] = {'guard': ('guards', '{cls}Guard', '"""Guard plugin: reviews every action between the shield\'s mandatory finite and bounds checks."""\n\nfrom __future__ import annotations\n\nimport numpy as np\n\nfrom hylo.core.safety import GuardVerdict, SafetyContext\n\n\nclass {cls}Guard:\n    name = "{name}"\n\n    def __init__(self, max_abs: float = 0.5) -> None:\n        self.max_abs = float(max_abs)\n\n    def review(self, data: np.ndarray, ctx: SafetyContext) -> GuardVerdict:\n        # ctx.state: latest observation state (may be None); ctx.last_action: previous approved action\n        limited = np.clip(data, -self.max_abs, self.max_abs)\n        changed = not np.array_equal(limited, data)\n        return GuardVerdict(limited, changed, "clamped to max_abs" if changed else "")\n', 'from hylo.testing.conformance import check_guard\nfrom {module}.plugin import {cls}Guard\n\n\ndef test_conformance():\n    rep = check_guard({cls}Guard())\n    assert rep.ok, rep.summary()\n'), 'gate_test': ('gate_tests', '{cls}Test', '"""Gate-test plugin: the paired statistic the improvement protocol decides with (H1: the candidate is better)."""\n\nfrom __future__ import annotations\n\nfrom hylo.core.scoring import PairedComparison\nfrom hylo.eval.scoring import one_sided_exact_mcnemar\n\n\nclass {cls}Test:\n    name = "{name}"\n\n    def __init__(self, metric: str = "success", direction: str = "higher") -> None:\n        self.metric, self.sign = metric, 1.0 if direction == "higher" else -1.0\n\n    def compare(self, candidate, parent) -> PairedComparison:\n        keys = [k for k in candidate if k in parent and candidate[k] is not None and parent[k] is not None]\n        d = [self.sign * (float(candidate[k]) - float(parent[k])) for k in keys]\n        wins, losses = sum(x > 0 for x in d), sum(x < 0 for x in d)\n        return PairedComparison(n_pairs=len(keys), candidate_wins=wins, parent_wins=losses,\n                                p_value=one_sided_exact_mcnemar(wins, losses))\n', 'from hylo.eval.conformance import check_gate_test\nfrom {module}.plugin import {cls}Test\n\n\ndef test_conformance():\n    rep = check_gate_test({cls}Test())\n    assert rep.ok, rep.summary()\n'), 'learner': ('learners', '{cls}Learner', '"""Learner plugin: development runs in, one sealed artifact out."""\n\nfrom __future__ import annotations\n\nimport json\nfrom pathlib import Path\n\nfrom hylo.core.artifact import ArtifactManifest, Provenance, development_runs, write_manifest\n\n\nclass {cls}Learner:\n    name = "{name}@0.1"\n\n    def fit(self, job) -> ArtifactManifest:\n        sources = development_runs(job.inputs, job.splits)           # gate / held-out are refused or skipped\n        episodes = [s for rd, _ in sources for s in json.loads((rd / "evallog.json").read_text())["samples"]]\n        out = Path(job.out)\n        out.mkdir(parents=True, exist_ok=True)\n        rate = sum(s["status"] == "success" for s in episodes) / max(1, len(episodes))\n        (out / "model.json").write_text(json.dumps({{"success_rate": rate}}))   # your model here\n        return write_manifest(out, kind="other", role="behaviour", metrics={{"episodes": len(episodes), "success_rate": rate}},\n                              provenance=Provenance(builder=self.name, sources=[r for _, r in sources], params=dict(job.params)))\n', 'from hylo.learn.conformance import check_learner\n\n\ndef test_conformance():\n    rep = check_learner("{module}.plugin:{cls}Learner")\n    assert rep.ok, rep.summary()\n'), 'online_learner': ('learners', '{cls}Online', '"""Online learner plugin: propose agent specs, see their rollouts on a development split, update; then seal."""\n\nfrom __future__ import annotations\n\nimport copy\nimport json\nfrom pathlib import Path\n\nimport numpy as np\n\nfrom hylo.core.artifact import ArtifactManifest, Provenance, write_manifest\n\n\nclass {cls}Online:\n    name = "{name}@0.1"\n\n    def start(self, job, agent) -> None:\n        self.job, self.base = job, agent\n        self.path = job.params.get("path", "params.kp")                # the agent parameter to search\n        self.low, self.high = job.params.get("low", 0.1), job.params.get("high", 4.0)\n        self.rng, self.best, self.tried = np.random.default_rng(0), None, []\n\n    def _spec(self, v: float):\n        spec = copy.deepcopy(self.base)\n        node = spec\n        *parents, leaf = self.path.split(".")\n        for k in parents:\n            node = node.setdefault(k, {{}})\n        node[leaf] = float(v)\n        return spec\n\n    def policies(self):\n        self.tried = list(self.rng.uniform(self.low, self.high, 3))  # random search; your method here\n        return [self._spec(v) for v in self.tried]\n\n    def update(self, results):\n        for v, r in zip(self.tried, results):\n            score = float(r["metrics"]["success_mean"])\n            if self.best is None or score > self.best[0]:\n                self.best = (score, v)\n        return {{"best_score": self.best[0], "best": self.best[1]}}\n\n    def finished(self) -> bool:\n        return False\n\n    def seal(self, out, sources) -> ArtifactManifest:\n        Path(out).mkdir(parents=True, exist_ok=True)\n        (Path(out) / "best.json").write_text(json.dumps({{"value": self.best[1], "score": self.best[0]}}))\n        return write_manifest(out, kind="policy_params", role="behaviour", metrics={{"best": self.best[1], "score": self.best[0]}},\n                              provenance=Provenance(builder=self.name, sources=sources,\n                                                    params={{k: v for k, v in self.job.params.items() if k != "experiment"}}))\n', 'from hylo.learn.conformance import check_learner\n\n\ndef test_conformance():\n    rep = check_learner("{module}.plugin:{cls}Online")\n    assert rep.ok, rep.summary()\n'), 'datasource': ('datasources', '{cls}Source', '"""Data-source plugin: external episodes (demonstrations, videos, other robots) with origin and licence."""\n\nfrom __future__ import annotations\n\nimport numpy as np\n\nfrom hylo.core.datasource import ExternalEpisode, ExternalFrame\n\n\nclass {cls}Source:\n    name = "{name}"\n\n    def __init__(self, n: int = 3, licence: str = "CC-BY-4.0") -> None:\n        self.n, self.licence, self.origin = n, licence, "synthetic://{name}"   # where the data comes from\n\n    def episodes(self):\n        for i in range(self.n):                                     # read your files here\n            frames = [ExternalFrame(state={{"pos": np.array([t / 10, 0.0])}}, images={{"cam": np.zeros((16, 16, 3), np.uint8)}})\n                      for t in range(10)]\n            yield ExternalEpisode(episode_id=f"{name}-{{i}}", task_id="{name}/task", instruction="do the task", frames=frames)\n', 'from hylo.learn.conformance import check_datasource\nfrom {module}.plugin import {cls}Source\n\n\ndef test_conformance():\n    rep = check_datasource({cls}Source())\n    assert rep.ok, rep.summary()\n'), 'judge': ('instruments', '{cls}Judge', '"""Instrument plugin: a success judge (labels episodes; trusted only after `hylo calibrate`)."""\n\nfrom __future__ import annotations\n\nfrom hylo.core.instrument import Judgement\n\n\nclass {cls}Judge:\n    name = "{name}@0.1"\n\n    def judge(self, episode) -> Judgement:\n        last = None\n        for ev in episode.of_type("observation"):                   # only what a real robot records\n            last = ev["payload"]\n        return Judgement(p_success=0.5, info={{"has_frame": bool(last and last.get("images"))}})   # your model here\n', 'from hylo.learn.conformance import check_judge\nfrom {module}.plugin import {cls}Judge\n\n\ndef test_conformance():\n    rep = check_judge({cls}Judge())\n    assert rep.ok, rep.summary()\n'), 'generator': ('generators', '{cls}Proposer', '"""Generator plugin: proposes config patches (candidates), each with a hypothesis; the gate decides."""\n\nfrom __future__ import annotations\n\nfrom hylo.core.improvement import Candidate\n\n\nclass {cls}Proposer:\n    name = "{name}@0.1"\n\n    def __init__(self, path: str = "agent.params.kp", factors: tuple[float, ...] = (1.5,)) -> None:\n        self.path, self.factors = path, factors\n\n    def propose(self, diagnosis, cfg, parent_id, archive):\n        node = cfg.model_dump(mode="json")\n        for k in self.path.split("."):\n            node = node[k]\n        return [(Candidate(patch={{self.path: float(node) * f}}, rationale=f"scale by {{f}}", parent_id=parent_id,\n                           generator=self.name), f"scaling {{self.path}} by {{f}} reduces the failures in the diagnosis")\n                for f in self.factors]\n', 'from hylo.evolve.conformance import check_generator\nfrom {module}.plugin import {cls}Proposer\n\n\ndef test_conformance():\n    rep = check_generator({cls}Proposer())\n    assert rep.ok, rep.summary()\n'), 'policy': ('policies', '{cls}Policy', '"""Policy plugin (core.policy.ChunkPolicy): draws action chunks, one per noise seed.\n\nRun it:      agent: {{type: policy, params: {{policy: {{type: {name}}}}}}}\nSteer it:    agent: {{type: steered, params: {{sampler: {{type: {name}}}, steering: {{n_samples: 4, selector: medoid}}}}}}\nACT-style:   agent: {{type: temporal_ensemble, params: {{policy: {{type: {name}}}, m: 0.1}}}}\nDraw all randomness from the seeds, so episodes replay and paired comparisons share noise. A learned policy loads\nits weights in __init__ (e.g. from an artifact path) and declares weights_license in its spec.\n\nAs generated it drives MuJoCo planar reach (hylo-bench) with analytic two-link IK. For your robot, change\nrequired_state, action_semantics / action_dim and sample(), and check it with\n`hylo plugin check policy {name} --experiment your_experiment.yaml`.\n"""\n\nfrom __future__ import annotations\n\nimport numpy as np\n\nfrom hylo.core.agent import AgentSpec, EpisodeContext\nfrom hylo.core.types import ActionSemantics, Observation\n\n\nclass {cls}Policy:\n    def __init__(self, gain: float = 0.3, horizon: int = 8, noise: float = 0.05, l1: float = 0.3, l2: float = 0.25,\n                 max_step: float = 0.05) -> None:\n        self.gain, self.horizon, self.noise, self.l1, self.l2, self.max_step = gain, horizon, noise, l1, l2, max_step\n\n    def spec(self) -> AgentSpec:\n        # what it reads and what its actions mean: checked against the environment before the first step\n        return AgentSpec(name="{name}", version="0.1", required_state=("qpos", "target_pos"),\n                         action_semantics=ActionSemantics(control_mode="joint_delta", frame="base"), action_dim=2,\n                         components={{"policy": "{name}@0.1"}})\n\n    def reset(self, ctx: EpisodeContext) -> None:\n        pass\n\n    def default_execution_steps(self) -> int:\n        return max(1, self.horizon // 2)\n\n    def _ik(self, x: float, y: float) -> np.ndarray:\n        c2 = np.clip((x * x + y * y - self.l1 ** 2 - self.l2 ** 2) / (2 * self.l1 * self.l2), -1.0, 1.0)\n        q2 = np.arccos(c2)\n        return np.array([np.arctan2(y, x) - np.arctan2(self.l2 * np.sin(q2), self.l1 + self.l2 * np.cos(q2)), q2])\n\n    def sample(self, observation: Observation, seeds: list[int] | None, noise_scale: float = 1.0) -> np.ndarray:\n        s = observation.state\n        err = (self._ik(*np.asarray(s["target_pos"], float)[:2]) - np.asarray(s["qpos"], float) + np.pi) % (2 * np.pi) - np.pi\n        a = np.clip(self.gain * err / self.max_step, -1.0, 1.0)\n        rngs = [np.random.default_rng(x) for x in seeds] if seeds else [np.random.default_rng()]\n        return np.stack([np.clip(a + self.noise * noise_scale * r.standard_normal((self.horizon, 2)), -1.0, 1.0)\n                         for r in rngs])                                    # (N, T, A)\n', 'import pytest\n\npytest.importorskip("hylo.envs.mujoco", reason="checked on MuJoCo planar reach (hylo-bench)")\n\nfrom hylo.agents.policy_agents import PolicyAgent  # noqa: E402\nfrom hylo.core.kernel import EpisodeRunner  # noqa: E402\nfrom hylo.testing.conformance import check_policy  # noqa: E402\nfrom hylo.testing.setup import check_experiment, env_and_task  # noqa: E402\nfrom {module}.plugin import {cls}Policy  # noqa: E402\n\n\ndef test_conformance():\n    env, task = env_and_task(check_experiment(prefer="mujoco"))\n    rep = check_policy({cls}Policy(), env, task)\n    assert rep.ok, rep.summary()\n\n\ndef test_it_reaches_the_target():\n    env, task = env_and_task(check_experiment(prefer="mujoco"))\n    assert EpisodeRunner(env, PolicyAgent({cls}Policy())).run(task).outcome.status == "success"\n'), 'agent': ('agents', '{cls}Agent', '"""Agent plugin: what the kernel talks to. spec / reset / act / on_step / on_episode_end.\n\nWrite an agent when the decision logic is more than drawing chunks from one model (planning, memory, switching\nbetween policies, a remote service); otherwise write a policy plugin and run it with the "policy" agent.\n\nAs generated it drives MuJoCo planar reach (hylo-bench) with analytic two-link IK. For your robot, change\nrequired_state, action_semantics / action_dim and act(), and check it with\n`hylo plugin check agent {name} --experiment your_experiment.yaml`.\n"""\n\nfrom __future__ import annotations\n\nimport numpy as np\n\nfrom hylo.core.agent import AgentBase, AgentSpec, EpisodeContext\nfrom hylo.core.types import ActionChunk, ActionSemantics, Observation\n\n\nclass {cls}Agent(AgentBase):\n    def __init__(self, gain: float = 0.3, horizon: int = 4, l1: float = 0.3, l2: float = 0.25, max_step: float = 0.05) -> None:\n        self.gain, self.horizon, self.l1, self.l2, self.max_step = gain, horizon, l1, l2, max_step\n\n    def spec(self) -> AgentSpec:\n        return AgentSpec(name="{name}", version="0.1", required_state=("qpos", "target_pos"),\n                         action_semantics=ActionSemantics(control_mode="joint_delta", frame="base"), action_dim=2,\n                         components={{"policy": "{name}@0.1"}})\n\n    def reset(self, ctx: EpisodeContext) -> None:\n        self.ctx = ctx                                  # ctx.emit(type, payload) writes into the trace\n\n    def act(self, observation: Observation) -> ActionChunk:\n        s = observation.state\n        x, y = np.asarray(s["target_pos"], float)[:2]\n        c2 = np.clip((x * x + y * y - self.l1 ** 2 - self.l2 ** 2) / (2 * self.l1 * self.l2), -1.0, 1.0)\n        q2 = np.arccos(c2)\n        q = np.array([np.arctan2(y, x) - np.arctan2(self.l2 * np.sin(q2), self.l1 + self.l2 * np.cos(q2)), q2])\n        err = (q - np.asarray(s["qpos"], float) + np.pi) % (2 * np.pi) - np.pi\n        a = np.clip(self.gain * err / self.max_step, -1.0, 1.0)\n        return ActionChunk(actions=np.repeat(a[None], self.horizon, axis=0), execution_steps=self.horizon,\n                           provenance={{"model": "{name}"}})\n', 'import pytest\n\npytest.importorskip("hylo.envs.mujoco", reason="checked on MuJoCo planar reach (hylo-bench)")\n\nfrom hylo.testing.conformance import check_agent  # noqa: E402\nfrom hylo.testing.setup import check_experiment, env_and_task  # noqa: E402\nfrom {module}.plugin import {cls}Agent  # noqa: E402\n\n\ndef test_conformance():\n    env, task = env_and_task(check_experiment(prefer="mujoco"))\n    rep = check_agent({cls}Agent(), env, task)\n    assert rep.ok, rep.summary()\n')}

DEPS `module-attribute` ¶

DEPS = {'guard': ['hylo-core'], 'gate_test': ['hylo-core', 'hylo-eval'], 'learner': ['hylo-core', 'hylo-learn'], 'online_learner': ['hylo-core', 'hylo-learn'], 'datasource': ['hylo-core', 'hylo-learn'], 'judge': ['hylo-core', 'hylo-learn'], 'generator': ['hylo-core', 'hylo-evolve'], 'policy': ['hylo-core', 'hylo-agents'], 'agent': ['hylo-core']}

new_plugin ¶

new_plugin(group: str, name: str, directory: str | Path = 'plugins') -> Path

declaration_warnings ¶

declaration_warnings(group: str, name: str) -> list[str]

What is missing from a plugin's declaration (contract version, maturity, licence).

check_plugin ¶

check_plugin(group: str, name: str, params: dict[str, Any] | None = None, experiment: str | None = None)

Build a registered plugin (or module:attr) and run its group's conformance check. Environment-dependent checks run on experiment (a config path), else MuJoCo planar reach when installed, else the test double.

hylo-cli¶

hylo.cli¶

hylo.cli.commands¶

HELP module-attribute ¶

hylo.cli.main¶

SCHEMAS module-attribute ¶

PASSTHROUGH module-attribute ¶