From fe3df888f9c7c95436c95916de26b111e5cb40e4 Mon Sep 17 00:00:00 2001 From: ohdearquant <122793010+ohdearquant@users.noreply.github.com> Date: Sun, 27 Sep 2026 18:18:46 -0400 Subject: [PATCH 1/5] apps/cli: the lion command, the areas page and its tests - apps/cli (lion-cli, a workspace member): `lion agent` builds a long-running agent from its chores.toml (trusted senders, the bounded khive client, the gate, the chores, the desk when configured), claims its directory before any branch, and serves, runs once or reruns; `lion chat`, `lion context` and `lion areas` sit beside it. - The verdict triage is built only with [desk.verdicts]; a section that names no verdict_dir is refused at build. - hub/areas.py and hub/app.py: the residents by area and the page and API that show them. - Tests for the command, the areas page, chat, the box tools and the resident's checkpoint; ADR-0019 points at where those tests live. - CI installs the CLI in the lowest-dependency and built-wheel jobs. --- .github/workflows/ci.yml | 10 +- apps/cli/lion_cli/__init__.py | 17 + apps/cli/lion_cli/agent.py | 383 +++++++ apps/cli/lion_cli/areas.py | 659 +++++++++++ apps/cli/lion_cli/chat.py | 428 +++++++ apps/cli/lion_cli/context.py | 257 +++++ apps/cli/lion_cli/main.py | 163 +++ apps/cli/lion_cli/view.py | 264 +++++ apps/cli/pyproject.toml | 22 + apps/cli/tests/conftest.py | 5 + apps/cli/tests/test_chat_command.py | 335 ++++++ .../019-checkpoint/ADR-0019-the-checkpoint.md | 2 +- hub/app.py | 427 +++++++ hub/areas.py | 602 ++++++++++ pyproject.toml | 9 +- tests/conftest.py | 13 + tests/test_agent.py | 339 ++++++ tests/test_areas.py | 1014 +++++++++++++++++ tests/test_chat.py | 221 +++- tests/test_context.py | 317 +++++- tests/test_daytona.py | 400 ++++++- tests/test_desk.py | 182 +++ tests/test_instruments.py | 81 ++ tests/test_turn_boundaries.py | 128 +++ uv.lock | 57 + 25 files changed, 6325 insertions(+), 10 deletions(-) create mode 100644 apps/cli/lion_cli/__init__.py create mode 100644 apps/cli/lion_cli/agent.py create mode 100644 apps/cli/lion_cli/areas.py create mode 100644 apps/cli/lion_cli/chat.py create mode 100644 apps/cli/lion_cli/context.py create mode 100644 apps/cli/lion_cli/main.py create mode 100644 apps/cli/lion_cli/view.py create mode 100644 apps/cli/pyproject.toml create mode 100644 apps/cli/tests/conftest.py create mode 100644 apps/cli/tests/test_chat_command.py create mode 100644 hub/app.py create mode 100644 hub/areas.py create mode 100644 tests/test_areas.py create mode 100644 tests/test_turn_boundaries.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 31e7d0c..5f2bee3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -63,7 +63,7 @@ jobs: - name: install at the floor run: | uv venv --python 3.11 "$RUNNER_TEMP/lowest" - uv pip install --python "$RUNNER_TEMP/lowest" --resolution lowest-direct -e . "pytest>=8.3" "hypothesis>=6.100" + uv pip install --python "$RUNNER_TEMP/lowest" --resolution lowest-direct -e . -e ./apps/cli "pytest>=8.3" "hypothesis>=6.100" - name: pydantic is the declared floor run: | "$RUNNER_TEMP/lowest/bin/python" - <<'PY' @@ -88,7 +88,9 @@ jobs: with: python-version: "3.11" - name: build - run: uv build + run: | + uv build + uv build apps/cli --out-dir dist-cli - name: the wheel holds exactly the tracked package files run: | uv run --no-project python - <<'PY' @@ -104,11 +106,11 @@ jobs: - name: test the installed wheel, away from the source tree run: | uv venv --python 3.11 "$RUNNER_TEMP/wheel" - uv pip install --python "$RUNNER_TEMP/wheel" dist/*.whl "pytest>=8.3" "hypothesis>=6.100" + uv pip install --python "$RUNNER_TEMP/wheel" dist/*.whl dist-cli/*.whl "pytest>=8.3" "hypothesis>=6.100" mkdir "$RUNNER_TEMP/suite" cp -R tests "$RUNNER_TEMP/suite/" cd "$RUNNER_TEMP/suite" - "$RUNNER_TEMP/wheel/bin/python" -c "import lionagi; assert 'site-packages' in lionagi.__file__, lionagi.__file__; print(lionagi.__file__)" + "$RUNNER_TEMP/wheel/bin/python" -c "import lionagi, lion_cli; assert all('site-packages' in m.__file__ for m in (lionagi, lion_cli)); print(lionagi.__file__, lion_cli.__file__)" "$RUNNER_TEMP/wheel/bin/python" -m pytest -W error --ignore=tests/test_exports.py --ignore=tests/test_hygiene.py --ignore=tests/register tests ci-ok: diff --git a/apps/cli/lion_cli/__init__.py b/apps/cli/lion_cli/__init__.py new file mode 100644 index 0000000..9246095 --- /dev/null +++ b/apps/cli/lion_cli/__init__.py @@ -0,0 +1,17 @@ +"""`lion`: the command line of lionagi v1, one entry with subcommands. + + lion chat [-c | -r ] [-v] # talk with an actor here; the record is the memory + lion agent --dir # serve an actor's long-running agent until interrupted + lion agent --dir --once [--wait 30] # one wake, the record printed + lion agent --dir --check disk # one instrument, printed and landed, no mail + lion agent --dir --ticks # create the missing reminders only + lion agent --dir --arm # landing arm: identity + boundary, landed + lion agent --dir --stats # the day's tokens, wakes, mail and idle polls + +The runtime is `lionagi`, the tools and guidance are `hub`; this package only composes them and +talks to a terminal. +""" + +from .main import main, parser + +__all__ = ("main", "parser") diff --git a/apps/cli/lion_cli/agent.py b/apps/cli/lion_cli/agent.py new file mode 100644 index 0000000..0a25eb3 --- /dev/null +++ b/apps/cli/lion_cli/agent.py @@ -0,0 +1,383 @@ +"""`lion agent`: an actor's long-running agent built from its directory, served, or run one step at a +time with its evidence landed there (ADR-0013, ADR-0016).""" + +from __future__ import annotations + +import argparse +import json +import os +import time +import tomllib +from pathlib import Path +from typing import Any + +from hub.agent.agent import Agent, AgentError, Budget, Gtd, Wake +from hub.agent.chores import Bounded, Chores, Outcome +from hub.agent.desk import Desk, Reader +from hub.agent.instruments import GH, GIT, LAUNCHCTL, instruments +from hub.agent.verdicts import Verdicts +from hub.guidance import guidance +from hub.harness import ClaudeCodeSession, OpenRouter +from hub.hooks import install, load +from hub.khive import Khive +from hub.tools import LocalTree, files, khive_ops +from lionagi import Actor, Profile +from lionagi.actor import DEFAULT_HANDLERS + +__all__ = ( + "build", + "chair", + "land", + "arm", + "render", + "stats", + "backend_for", + "agent_main", +) + + +def build(directory: Path, backend) -> tuple[Chores, Agent]: + """The agent from its directory: the trusted senders are the floor (owner, steward, itself) plus + the config's own `trusted` list; the khive client is bounded to read verbs and to those + recipients and the desks `[desk.routes]` names; mail from anyone else is dropped at the agent's + gate before any run.""" + cfg = tomllib.loads((directory / "chores.toml").read_text()) + actor_id = cfg["actor"] + trusted = frozenset({cfg["owner"], cfg.get("steward"), actor_id, *cfg.get("trusted", [])} - {None}) + routes = dict((cfg.get("desk") or {}).get("routes", {})) # the desk forwards to these; `Desk` checks them + if routes and not cfg["desk"].get("front", True): # no Desk would check them, yet they widen the sends + raise ValueError("[desk.routes] needs the front desk: front = false names no routes") + khive = Bounded(Khive(directory, actor_id), recipients=trusted | frozenset(routes.values())) + actor = Actor("chores", privileges={"comm.send"}, notes_dir=directory / "notes") + if (hooks := directory / "hooks.toml").exists(): + install(actor, load(hooks), directory) + profiles: dict[str, Profile] = {} + budget = Budget.load(cfg.get("budget")) # `[budget]`: the caps, each defaulting to Budget's own + agent = Agent( + actor, + khive, + backend, + lambda mail: profiles["wake"], + # an actor as the chair takes every sender's mail (`[chair] admit = "all"`): the gate would silently + # shelve a new peer; the guidance still says whose mail is work. Anything else keeps the trusted gate. + admit=None + if isinstance(cfg.get("chair"), dict) and cfg["chair"].get("admit") == "all" + else (lambda m: m is not None and chores.admit(m)), + owner=cfg["owner"], + # sweep, route, OUT is three clean rounds; a slip (measured: an unquoted dict key) costs one each + max_rounds=6, + batch=int(cfg.get("batch", 5)), + budget=budget, + # the chair is one conversation for the life of the process: a burst of mail, a plain reply, + # the next burst in the same run; no wake of its own for what arrives meanwhile + continuous=isinstance(cfg.get("chair"), dict), + # `[wake]`: which mail wakes now, which waits for a batch, which is dropped (none = every mail now); + # the file is re-read while serving, so a cap or a rule edited from outside takes effect + wake_rules=Wake.load(cfg.get("wake")), + gtd=Gtd.load(cfg.get("gtd")), # `[gtd]`: the queue as a wake source and a notification line + config=directory / "chores.toml", + ) + reader = None + if cfg.get("desk"): # the owner's box, read through a reader that only reads: the front desk's + reader = Reader(Khive(Path(cfg["desk"]["actor_dir"]), cfg["owner"])) + chores = Chores( + agent, + cfg["owner"], + cfg.get("steward"), + instruments(cfg, khive, agent.state), + directory / "ledger.jsonl", + trusted=trusted, + aged_after=int(cfg.get("aged_after", 2)), + aged_every=cfg.get("aged_every", "every:1h"), + dead_after=int(cfg.get("dead_after", 3)), + ) + # the front desk reads the owner's box through that reader; `front = false` builds no front desk + if cfg.get("desk") and cfg["desk"].get("front", True): + d = cfg["desk"] + chores.digest_every = str(d.get("brief", "every:12h")) # the digest is the front desk's brief + v = d.get("verdicts") # `[desk.verdicts]` absent: a verdict notice is listed like any mail + if v is not None and not v.get("verdict_dir"): + raise ValueError("[desk.verdicts] names the verdict_dir the review daemon writes its verdicts to") + agent.desk = Desk( + agent, + chores, + reader, + directory / "desk.jsonl", + answers_per_hour=int(d.get("answers_per_hour", 10)), + clarify_hours=int(d.get("clarify_hours", 24)), + escalate_from=tuple(d.get("escalate_from", ())), + escalations_per_day=int(d.get("escalations_per_day", 40)), + routes=routes, + tasks_per_day=int(d.get("tasks_per_day", 20)), + verdicts=None + if v is None + else Verdicts( + verdict_dir=str(v["verdict_dir"]), + actor_map=dict(v.get("actor_map", {})), + actor_prefix=dict(v.get("actor_prefix", {})), + author_actor=dict(v.get("author_actor", {})), + stub_decisions=frozenset(v.get("stub_decisions", ["ENGINE-UNAVAILABLE"])), + ), + verdict_from=tuple((v or {}).get("from", ())), + ) + profiles["chores"], profiles["reader"] = chores.profiles() + # `[chair]`: the chair itself is the resident; admitted mail runs under its own hat, not the chores' + profiles["wake"] = chair(directory, cfg, actor, khive) if "chair" in cfg else profiles["chores"] + return chores, agent + + +def chair(directory: Path, cfg: dict, actor: Actor, khive: Bounded) -> Profile: + """The chair as the resident: what the trusted senders mail is its own work, answered on the + thread with `send`, with the directory's files, the bounded khive reads and its notes as the + runtime's tools; `[chair].system` adds the chair's own role text. No chore runs under it. The + runtime's tools cannot act irreversibly; the CLI's own tools under `[claude_code]` can (Bash + reaches git and gh), and there the chair's red lines are guidance, as for a Claude Code session: no + merge, no approve, no push, no process signal, no secret; a click stays with the steward.""" + if not isinstance( + cfg["chair"], dict + ): # `chair = true` is not a section: name the shape, not an AttributeError + raise ValueError("[chair] is a table: `[chair]` with an optional `system`") + specs = files(actor, LocalTree(directory)) | khive_ops(actor, khive) | {"send"} + specs |= {h.name for h in DEFAULT_HANDLERS} + tools = ( + f"read(path), list_dir(path), khive_ops(ops) signing as {khive.actor} ({len(khive.allowed)} verbs: " + "reads, comm.read, schedule.remind, memory.remember at spawn salience), send(to, thread, text), " + "note.get(key), note.list()" + ) + senders = ", ".join(sorted(khive.recipients or ())) + if own := (cfg.get("claude_code") or {}).get("tools"): + dirs = ", ".join([str(directory), *map(str, cfg["claude_code"].get("add_dirs", ()))]) + tools += ( + f"; and your own Claude Code tools ({', '.join(own)}) run inside each round in {dirs}, " + "with the khive MCP signing as you when it is configured" + ) + text = guidance("chair", actor=cfg["actor"], owner=cfg["owner"], senders=senders, tools=tools) + if role := str(cfg["chair"].get("system", "")).strip(): + text += "\n\n" + role + return Profile(name="chair", specs=frozenset(specs), privileges=frozenset({"comm.send"}), system=text) + + +def land(directory: Path, name: str, text: str) -> Path: + """Landing evidence lives in the agent directory where a reader can open it, not on a console: + `landing/-.txt`, overwritten by the day's latest run of that name.""" + path = directory / "landing" / f"{time.strftime('%Y%m%d')}-{name}.txt" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + return path + + +async def arm(chores: Chores, agent: Agent) -> tuple[int, list[str]]: + """The landing arm, run once from the agent directory: (1) identity, the actor's own inbox read + through the bounded client returns rows addressed to it; (2) boundary, write verbs, a call the + boundary cannot name, mail as an op, a memory above spawn salience and a send to an outsider are + refused by the client, the client's own denial text quoted. 0 only when both hold.""" + rc, lines = 0, [] + (page,) = await agent.khive.exec('comm.inbox(limit=5, status="all")') + rows = page.get("messages", []) if isinstance(page, dict) else (page or []) + mine = [m for m in rows if (m.get("properties") or {}).get("to_actor") == agent.khive.actor] + lines.append( + f"identity: {agent.khive.actor} from {agent.khive.cwd}: {len(rows)} row(s), " + f"{len(mine)} addressed to it" + ) + if not mine: + lines.append("identity: FAILED, no row addressed to the actor (seed one with comm.send first)") + rc = 1 + for probe in ( + 'gtd.complete(id="00000000-0000-0000-0000-000000000000")', + 'update(id="x", content="y")', + 'x.gtd.complete(id="a")', + 'comm.send(to="email:outsider@example.com", content="arm")', + 'memory.remember(content="arm", salience=0.9)', + ): + try: + await agent.khive.exec(probe) + lines.append(f"boundary: FAILED, {probe} was not refused") + rc = 1 + except AgentError as e: + lines.append(f"boundary: refused {probe}: {e}") + try: + await agent.khive.send("email:outsider@example.com", "arm") + lines.append("boundary: FAILED, a send to an outsider was not refused") + rc = 1 + except AgentError as e: + lines.append(f"boundary: refused send to an outsider: {e}") + lines.append(f"trusted senders: {sorted(chores.trusted)}") + lines.append(f"programs: gh={GH} git={GIT} launchctl={LAUNCHCTL}") + # the claude CLI child inherits this process's env and prefers ANTHROPIC_API_KEY when it is set: a + # resident on the subscription shows both absent (the CLI's own login) or the OAuth token alone + seen = { + k: "set" if os.environ.get(k) else "absent" for k in ("ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN") + } + account = ( + os.environ.get("CC_ACCOUNT") or "unnamed (the CLI's own login)" + ) # CC_ACCOUNT: the account's name, as whatever launched the agent set it + lines.append(f"auth env the claude child would inherit: {seen}; account: {account}") + if seen["ANTHROPIC_API_KEY"] == "set": + lines.append("auth: FAILED, ANTHROPIC_API_KEY is set; a resident bills the key, not the subscription") + rc = 1 + return rc, lines + + +def render(name: str, inst: Any, out: Outcome) -> str: + """One instrument's outcome as the landing file and the console show it.""" + control = "ok" if out.control else "FAILED" + lines = [ + f"{name}: control {control}, {len(out.findings)} finding(s), escalate={out.escalate}, " + f"count={out.count}, direction {inst.direction}", + f"answers: {inst.answers}", + f"known-positive: {out.positive or '(none)'}", + f"population: {out.population}\npredicate: {out.predicate}", + ] + lines += [f"- {f}" for f in out.findings] + return "\n".join(lines + [out.evidence]) + + +def stats(directory: Path) -> str: + """The numbers behind "does a long-running agent save tokens": the day's spend row (model calls, + tokens in and out, dollars, wakes that ran, mail handled, idle polls at zero) beside the last + wake's cursor, read from the agent's own notes; nothing is computed that the agent did not write.""" + notes = directory / "notes" / "agent.json" + if not notes.exists(): + return f"no notes at {notes}: the agent has not run from this directory" + store = json.loads(notes.read_text()).get("notes", {}) + spend = (store.get("spend") or {}).get("value") or {} + cursor = (store.get("cursor") or {}).get("value") or {} + tokens = cursor.get("tokens") or {} + mail = spend.get("mail", 0) + lines = [ + f"day {spend.get('day')}: {spend.get('calls', 0)} model call(s), {spend.get('in', 0)} in, " + f"{spend.get('out', 0)} out, ${spend.get('cost', 0.0):.4f}; {spend.get('runs', 0)} wake(s) with " + f"{mail} mail; {spend.get('idle', 0)} idle poll(s) at 0 tokens", + ( + f"per handled message: {(spend.get('in', 0) + spend.get('out', 0)) / mail:.0f} tokens" + if mail + else "per handled message: no mail handled today" + ), + f"last wake {cursor.get('wake')} at {cursor.get('at')}: {cursor.get('outcome')} after " + f"{cursor.get('rounds')} round(s), {len(cursor.get('mail') or [])} mail, {cursor.get('sent')} sent; " + f"{tokens.get('calls', 0)} call(s), {tokens.get('in', 0)} in, {tokens.get('out', 0)} out, " + f"${tokens.get('cost', 0.0):.4f}; {cursor.get('idle_polls', 0)} idle poll(s) before it", + ] + ledger = directory / "ledger.jsonl" + if ledger.exists(): + rows = [json.loads(ln) for ln in ledger.read_text().splitlines() if ln.strip()] + answered = sum(1 for r in rows if r.get("answered", r.get("control"))) + lines.append(f"ledger: {len(rows)} check(s), {answered} answered, {len(rows) - answered} not") + desk = directory / "desk.jsonl" + if desk.exists() and spend.get("day"): + rows = [json.loads(ln) for ln in desk.read_text().splitlines() if ln.strip()] + mine = [r for r in rows if r["at"].startswith(spend["day"])] + answered = sum(1 for r in mine if r["class"] == "answered") + left = len(mine) - answered + lines.append( + f"desk: {len(mine)} of the owner's message(s) read today, {answered} answered, {left} left" + ) + return "\n".join(lines) + + +def backend_for( + model: str, options: dict | None = None, *, cwd: Path | None = None +) -> OpenRouter | ClaudeCodeSession: + """The console and agent backend: DeepSeek models route to CoreWeave, then DeepInfra or Fireworks, + never the first-party endpoint; other models take OpenRouter's own routing. `claude_code` or + `claude_code/` is the claude CLI on the subscription, never an API key, on one resumed + session per run: each round sends only what the view gained, so the prefix caches (the chat shape + resent the whole transcript every round). `options` + is the config's `[claude_code]` table: `tools` gives the CLI its own tools inside each round, in + `cwd` (the agent directory: khive signs as its actor) with `add_dirs` reachable, `allowed` for + permission rules beyond the tools, `mcp_config` for the one MCP it loads, `max_turns` per round, + `timeout` seconds a round may take (a round of real work can run past the default, 300), + `effort` the CLI's effort level.""" + if model == "claude_code" or model.startswith("claude_code/"): + o = options or {} + return ClaudeCodeSession( + model=model.partition("/")[2] or None, + tools=tuple(o.get("tools", ())), + allowed=tuple(o.get("allowed", ())), + max_turns=int(o.get("max_turns", 1)), + timeout=float(o.get("timeout", 300.0)), + effort=o.get("effort"), # the CLI's --effort; absent = the CLI's default + cwd=cwd, + add_dirs=tuple(str(d) for d in o.get("add_dirs", ())), + mcp_config=o.get("mcp_config"), + cli=o.get("cli"), # under launchd the PATH holds no claude: the config names the binary + ) + deepseek = model.startswith("deepseek") + return OpenRouter( + model=model, + timeout=120, + provider=( + { + "order": ["coreweave", "deepinfra", "fireworks"], + "allow_fallbacks": False, + "data_collection": "deny", + } + if deepseek + else None + ), + reasoning={"max_tokens": 2000}, + stop=["3} {e.name:>6} r{e.round}]", + str(e.content)[:600].replace("\n", "\n" + " " * 18), + ) + return + # a chair runs no chore: no digest, no aged pass, no tick in its box (each would be a model call) + created = [] if "chair" in cfg else await chores.ensure_ticks() + settings = ( # the values this process holds; the file may say otherwise until a restart + f"; claude_code timeout={backend.timeout:.0f}s max_turns={backend.max_turns}" + if isinstance(backend, ClaudeCodeSession) + else "" + ) + rules = agent.wake_rules # [budget] and [wake] follow the file while serving; the start values here + settings += ( + f"; budget cost_per_day={agent.budget.cost_per_day} wakes_per_hour={agent.budget.wakes_per_hour}" + f"; wake {rules.describe() if rules is not None else 'every mail now'}" + f"; gtd {agent.gtd.describe() if agent.gtd is not None else 'off'}" + ) + print( + f"serving {agent.khive.actor} from {directory} as pid {os.getpid()}; ticks created: {created}; " + f"programs gh={GH} git={GIT}{settings}", + flush=True, + ) + await agent.serve(log=lambda line: print(time.strftime("%Y-%m-%d %H:%M:%S"), line, flush=True)) + finally: + agent._release(pid) diff --git a/apps/cli/lion_cli/areas.py b/apps/cli/lion_cli/areas.py new file mode 100644 index 0000000..386ac27 --- /dev/null +++ b/apps/cli/lion_cli/areas.py @@ -0,0 +1,659 @@ +"""`lion areas`: the residents by area, at the terminal, as a page that reloads itself, or as the API +the phone app reads (live over Server-Sent Events, with controls back onto each resident).""" + +from __future__ import annotations + +import argparse +import asyncio +import ipaddress +import json +import math +import os +import plistlib +import re +import subprocess +import threading +import time +import tomllib +from collections import OrderedDict +from datetime import datetime +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, urlencode, urlsplit +from xml.parsers.expat import ExpatError + +from hub.app import app_shell, icon_svg, manifest_webmanifest, service_worker_js +from hub.areas import ( + Area, + area_rows, + controls_info, + identity, + load_areas, + log_tail, + page, + resident, + resident_json, + resident_page, + rows_json, + status, + table, + talk, + wakes, +) +from hub.khive import Khive, KhiveError, lit +from lionagi.checkpoint import CheckpointError + +from .context import checkpoint as context_checkpoint +from .context import restore as context_restore +from .context import set_var as context_set +from .context import status as context_status + +__all__ = ("context_api", "serve", "areas_main") + +# Tailscale's CGNAT range: the page answers only over a tailnet or loopback. +_TAILSCALE_CGNAT = ipaddress.ip_network("100.64.0.0/10") +_LAUNCHCTL = "/bin/launchctl" # absolute: under launchd PATH is /usr/bin:/bin:/usr/sbin:/sbin +_HISTORY = 20 # the serve.log lines per desk an `/events` connection opens with +_POLL_S = 1.0 # how often `/events` re-reads the registry, the residents and their logs +_PING_S = 15.0 +_WRITE_S = 30.0 # a client that takes longer than this to accept a write is dropped +_ONCE_KEEP = 100 # the `/control` answers kept for a replayed `once` +_MAX_BODY = 1 << 20 + +# a table or array-of-tables header ends the table before it; `[budget]` may carry a comment +_TABLE_RE = re.compile(r"^[ \t]*\[") +_BUDGET_RE = re.compile(r"^[ \t]*\[[ \t]*budget[ \t]*\][ \t]*(?:#.*)?$") +_KEY_RE = re.compile( + r"^(?P[ \t]*)(?P[A-Za-z_][\w-]*)(?P[ \t]*=[ \t]*)" + r"(?P[^#\r\n]*?)(?P[ \t]*(?:#.*)?\r?\n?)$" +) + + +def _peer_allowed(addr: str) -> bool: + """Loopback or inside the Tailscale CGNAT range; anything else is refused before the request + is read any further.""" + try: + ip = ipaddress.ip_address(addr) + except ValueError: + return False + if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None: + ip = ip.ipv4_mapped + return ip.is_loopback or ip in _TAILSCALE_CGNAT + + +def _rewrite_budget(text: str, updates: dict[str, str]) -> str: + """Rewrite only the named keys inside `[budget]`, keeping every other byte (comments + included); a missing key is added after the table's last key, a missing table appended at the + end.""" + lines = text.splitlines(keepends=True) + start = next((i for i, line in enumerate(lines) if _BUDGET_RE.match(line.rstrip("\r\n"))), None) + if start is None: + head = text if not text or text.endswith("\n") else text + "\n" + block = "[budget]\n" + "".join(f"{k} = {v}\n" for k, v in updates.items()) + return head + ("\n" if head else "") + block + end = next((j for j in range(start + 1, len(lines)) if _TABLE_RE.match(lines[j])), len(lines)) + remaining = dict(updates) + last = start # the table's last line that is neither blank nor a comment + for i in range(start + 1, end): + body = lines[i].strip() + if body and not body.startswith("#"): + last = i + m = _KEY_RE.match(lines[i]) + if m and m["key"] in remaining: + lines[i] = f"{m['indent']}{m['key']}{m['eq']}{remaining.pop(m['key'])}{m['rest']}" + if remaining: + if not lines[last].endswith("\n"): + lines[last] += "\n" + lines.insert(last + 1, "".join(f"{k} = {v}\n" for k, v in remaining.items())) + return "".join(lines) + + +def _text(body: dict, key: str) -> str: + v = body.get(key) + return "" if v is None else str(v) + + +def _one_line(text: str) -> str: + return " ".join(str(text).split()) + + +def _tail( + log: Path, state: tuple[int, bytes, int] | None +) -> tuple[list[tuple[str, bool]], tuple[int, bytes, int]]: + """The whole lines a serve.log gained since `state` (the offset read to, a partial last line and + the inode read), each with whether it is history. First sight is the last `_HISTORY` lines as + history; a partial last line waits for its newline; a log that shrank (truncated) or is another + file at the name (rotated, replaced) is read from its start; no log yet is no lines, and its + first lines are new.""" + try: + f = open(log, "rb") + except FileNotFoundError: + return [], state or (0, b"", 0) + with f: + st = os.fstat(f.fileno()) + size, ino = st.st_size, st.st_ino + if state is None: + f.seek(max(size - 65536, 0)) + data = f.read() + whole, _, partial = data.rpartition(b"\n") + lines = whole.split(b"\n") if whole else [] + if size > 65536 and lines: + lines = lines[1:] # the seek landed inside a line + said = [s for s in (b.decode(errors="replace").rstrip("\r") for b in lines) if s] + return [(s, True) for s in said[-_HISTORY:]], (f.tell(), partial, ino) + pos, partial, seen = state + if size < pos or ino != seen: + pos, partial = 0, b"" + f.seek(pos) + data = partial + f.read() + end = f.tell() + whole, _, partial = data.rpartition(b"\n") + said = [s for s in (b.decode(errors="replace").rstrip("\r") for b in whole.split(b"\n")) if s] + return [(s, False) for s in said], (end, partial, ino) + + +def _desk(registry: Path, want: str) -> tuple[Area, str] | None: + # as written in the registry, or as a Resident's `dir` prints it (the link the page carries) + for a in load_areas(registry): + if any(want in (d, str(Path(d))) for d in a.desks): + return a, want + return None + + +_CONTEXT_ROUTES = { + ("GET", "/api/context/status"), + ("GET", "/api/context/restore"), + ("POST", "/api/context/checkpoint"), + ("POST", "/api/context/set"), +} + + +def context_api(registry: Path, method: str, path: str, params: dict) -> tuple[int, dict]: + """`GET /api/context/status?dir=`, `GET /api/context/restore?dir=&id=`, `POST + /api/context/checkpoint` `{dir, why}` and `POST /api/context/set` `{dir, name, kind, value}` onto a + registered desk: (status, the JSON answer). The work is `lion context`'s own; a dir the registry + does not name is 404, a refused take, restore or declaration 409.""" + if (method, path) not in _CONTEXT_ROUTES: + return 404, {"ok": False, "detail": f"no route {method} {path}"} + d = str(params.get("dir") or "") + if _desk(registry, d) is None: + return 404, { + "ok": False, + "detail": f"no resident at {d!r}: not a desk in {registry}; nothing was read", + } + try: + if path == "/api/context/status": + return 200, context_status(d) + if path == "/api/context/restore": + return 200, context_restore(d, str(params.get("id") or "") or None) + if path == "/api/context/set": + name, kind = str(params.get("name") or ""), str(params.get("kind") or "") + return 200, context_set(d, name, kind, params.get("value")) + return 200, context_checkpoint(d, str(params.get("why") or "manual")) + except CheckpointError as e: + return 409, {"ok": False, "detail": str(e)} + + +def serve( + registry: Path, + host: str, + port: int, + refresh: int, + identity: tuple[str, Khive] | None = None, + allow: list[str] | None = None, +) -> None: + """`/` is the page, `/api/areas` the rows as JSON; every request re-reads the registry and the + residents' own files, so the page is never staler than its refresh. `/r?dir=` is one + resident's page (`/api/resident?dir=` its JSON) and `POST /send` mails that resident as + `identity`. `/events` streams the rows and the logs live, `POST /control` is the write path back + onto a resident. A dir the registry does not name is 404 everywhere, so nothing is read, + written or mailed outside it.""" + me, khive = identity or (None, None) + hosts: set[str] = set() # the names this server answers to, known once the port is + origins: set[str] = set() # the Origins a browser POST may carry: http(s) + an allowed host + once_cache: OrderedDict[str, tuple[int, dict]] = OrderedDict() # /control's idempotency, last 100 + # one control at a time: a replayed `once` racing its first request must find the first answer + control_lock = threading.Lock() + # `/events`: the registry and the residents are read once per `_POLL_S` for every connection, + # not once per connection (each read runs a pgrep per serving resident) + snapshot: dict[str, Any] = {"at": float("-inf"), "areas": [], "rows": None} + snapshot_lock = threading.Lock() + + def poll_rows() -> tuple[list[Area], list[dict] | None]: + with snapshot_lock: + if time.monotonic() - snapshot["at"] >= _POLL_S: + try: + areas = load_areas(registry) + rows: list[dict] | None = area_rows(status(areas)) + except (OSError, ValueError): + areas, rows = [], None # a registry mid-write: the next poll reads it whole + snapshot.update(at=time.monotonic(), areas=areas, rows=rows) + return snapshot["areas"], snapshot["rows"] + + def desk(want: str) -> tuple[Area, str] | None: + return _desk(registry, want) + + class Handler(BaseHTTPRequestHandler): + def answer( + self, code: int, body: str, ctype: str = "text/plain; charset=utf-8", location: str = "" + ) -> None: + data = body.encode() + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(data))) + if location: + self.send_header("Location", location) + self.end_headers() + self.wfile.write(data) + + def answer_json(self, code: int, doc: dict) -> None: + self.answer(code, json.dumps(doc), "application/json") + + def wants_json(self) -> bool: + # `/control` always answers JSON; `/send` does under `Accept: application/json` + path = urlsplit(self.path).path + return path == "/control" or "application/json" in self.headers.get("Accept", "") + + def guarded(self) -> bool: + # off the tailnet or loopback: refused before anything else is read. A loopback page is + # reachable by any site the browser visits: a Host that is not this server (DNS + # rebinding) or a POST a browser sent from another origin (a cross-site form) is refused + # too. A native client sends no Origin: the peer and Host guards are its admission + peer = self.client_address[0] + refusal = "" + if not _peer_allowed(peer): + refusal = f"{peer} is outside loopback/Tailscale; nothing was done" + elif (host := self.headers.get("Host", "")) not in hosts: + refusal = f"host {host!r} is not this server; nothing was done" + elif self.command == "POST": + origin = self.headers.get("Origin") + if origin is not None and origin not in origins: + refusal = f"origin {origin!r} is not this server; nothing was done" + if not refusal: + return True + if self.command == "POST" and self.wants_json(): + self.answer_json(403, {"ok": False, "action": "", "detail": refusal}) + else: + self.answer(403, refusal + "\n") + return False + + def run(self, coro): + # each request runs in its own thread with no loop running: one loop per call; the + # envelopes Khive keeps are for a console's log, and a server would hold them forever + try: + return asyncio.run(coro) + finally: + khive.calls.clear() + + def body(self) -> dict | str: + """The POST body's fields: JSON under `Content-Type: application/json` (or a body that + opens with `{`, a native client that left the header out), else form-encoded. A str + says what was wrong with it.""" + try: + length = int(self.headers.get("Content-Length") or 0) + except ValueError: + return "Content-Length is not a number" + if not 0 <= length <= _MAX_BODY: + return f"the body is {length} bytes; at most {_MAX_BODY}" + text = (self.rfile.read(length) if length else b"").decode(errors="replace") + ctype = self.headers.get("Content-Type", "").lower() + if "application/json" in ctype or text.lstrip().startswith("{"): + try: + doc = json.loads(text or "{}") + except ValueError: + return "the body is not JSON" + return doc if isinstance(doc, dict) else "the body is not a JSON object" + return {k: (v[0] if v else "") for k, v in parse_qs(text, keep_blank_values=True).items()} + + # ---- GET ---- + + def do_GET(self) -> None: # noqa: N802 (the stdlib's name) + if not self.guarded(): + return + url = urlsplit(self.path) + if url.path == "/favicon.ico": + self.answer(204, "") + return + if url.path == "/app": + self.answer(200, app_shell(), "text/html; charset=utf-8") + return + if url.path == "/app/manifest.webmanifest": + self.answer(200, manifest_webmanifest(), "application/manifest+json") + return + if url.path == "/app/sw.js": + self.answer(200, service_worker_js(), "text/javascript; charset=utf-8") + return + if url.path == "/app/icon.svg": + self.answer(200, icon_svg(), "image/svg+xml") + return + if url.path == "/events": + self._events() + return + if url.path == "/api/areas": + self.answer(200, rows_json(status(load_areas(registry))), "application/json") + return + if url.path == "/": + areas = load_areas(registry) + self.answer(200, page(areas, status(areas), refresh, me), "text/html; charset=utf-8") + return + if url.path.startswith("/api/context/"): + params = {k: v[0] for k, v in parse_qs(url.query).items() if v} + self.answer_json(*context_api(registry, "GET", url.path, params)) + return + if url.path not in ("/r", "/api/resident"): + self.send_error(404) + return + query = parse_qs(url.query) + want = (query.get("dir") or [""])[0] + found = desk(want) + if found is None: + self.answer(404, f"no resident at {want!r}: not a desk in {registry}\n") + return + area, d = found + row = resident(area, d) + wake_rows, tail = wakes(d), log_tail(d) + talk_rows: list[dict] | str = [] + if khive is not None and row.actor in ("?", ""): + talk_rows = f"the resident's actor could not be read: {row.problem}" + elif khive is not None: + try: + talk_rows = self.run(talk(khive, row.actor)) + except KhiveError as e: + talk_rows = str(e) + if url.path == "/r": + sent = (query.get("sent") or [""])[0] + body = resident_page(row, wake_rows, tail, talk_rows, me, refresh, d, sent=sent) + self.answer(200, body, "text/html; charset=utf-8") + else: + self.answer(200, resident_json(row, wake_rows, tail, talk_rows, me), "application/json") + + def _events(self) -> None: + """SSE, one thread per connection (ThreadingHTTPServer's). Every `_POLL_S` re-read the + registry and the residents: `areas` on connect and whenever a row changed; per desk the + last `_HISTORY` serve.log lines as history on first sight, then each new line; `ping` + every `_PING_S`. A write the client is gone for, or stalled on, ends the thread.""" + try: + self.connection.settimeout(_WRITE_S) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + except OSError: + return + + def emit(event: str, data: Any) -> bool: + try: + self.wfile.write(f"event: {event}\ndata: {json.dumps(data)}\n\n".encode()) + self.wfile.flush() + return True + except OSError: # BrokenPipe, ConnectionReset: the client went away + return False + + tails: dict[str, tuple[int, bytes, int]] = {} + said: str | None = None + next_ping = time.monotonic() + _PING_S + while True: + areas, rows = poll_rows() + if rows is not None and (sig := json.dumps(rows)) != said: + if not emit("areas", rows): + return + said = sig + for a in areas: + for listed in a.desks: + d = str(Path(listed)) # as the row's `dir` prints it, so the app can join them + lines, tails[d] = _tail(Path(d) / "serve.log", tails.get(d)) + for line, history in lines: + if not emit("log", {"dir": d, "line": line, "history": history}): + return + if time.monotonic() >= next_ping: + if not emit("ping", {}): + return + next_ping = time.monotonic() + _PING_S + time.sleep(_POLL_S) + + # ---- POST ---- + + def do_POST(self) -> None: # noqa: N802 (the stdlib's name) + if not self.guarded(): + return + path = urlsplit(self.path).path + if path == "/control": + self._control() + elif path == "/send": + self._send() + elif path.startswith("/api/context/"): + body = self.body() + if isinstance(body, str): + self.answer_json(400, {"ok": False, "detail": f"{body}; nothing was taken"}) + return + self.answer_json(*context_api(registry, "POST", path, body)) + else: + self.send_error(404) + + def _send(self) -> None: + wants_json = self.wants_json() + + def fail(code: int, detail: str) -> None: + if wants_json: + self.answer_json(code, {"ok": False, "detail": detail}) + else: + self.answer(code, detail + "\n") + + body = self.body() + if isinstance(body, str): + fail(400, f"{body}; nothing was sent") + return + d, subject, content, once = (_text(body, k) for k in ("dir", "subject", "content", "once")) + found = desk(d) + if found is None: + fail(404, f"no resident at {d!r}: not a desk in {registry}; nothing was sent") + return + if not content.strip(): + fail(400, "content is empty: nothing was sent") + return + if khive is None: + fail(400, "no identity: start lion areas with --as DIR; nothing was sent") + return + # the recipient is the resident's own actor, read from its chores.toml, never the form + row = resident(*found) + if row.actor in ("?", ""): + fail(500, f"the resident's actor could not be read: {row.problem}; nothing was sent") + return + args = [f"to={lit(row.actor)}"] + if subject.strip(): + args.append(f"subject={lit(subject)}") + args.append(f"content={lit(content)}") + once = once.replace("-", "").lower() # a UUID with its hyphens is the same token + if re.fullmatch(r"[0-9a-f]{32}", once): # the page's own token: a repeat replays, never resends + args.append(f"idempotency_key={lit('ui:' + once)}") + try: + (res,) = self.run(khive.exec(f"comm.send({', '.join(args)})")) + except KhiveError as e: + fail(502, str(e)) + return + sent_id = str((res or {}).get("id") or "") + if wants_json: + self.answer_json(200, {"ok": True, "id": sent_id}) + return + where = {"dir": d, "sent": sent_id} + self.answer(303, "", location="/r?" + urlencode({k: v for k, v in where.items() if v})) + + def _control(self) -> None: + body = self.body() + if isinstance(body, str): + self.answer_json(400, {"ok": False, "action": "", "detail": body}) + return + d, action = _text(body, "dir"), _text(body, "action") + once = _text(body, "once").replace("-", "").lower() # a UUID with its hyphens is the same token + with control_lock: + if once and once in once_cache: # the first answer again, nothing done twice + code, resp = once_cache[once] + self.answer_json(code, resp) + return + if desk(d) is None: + detail = f"no resident at {d!r}: not a desk in {registry}; nothing was done" + self.answer_json(404, {"ok": False, "action": action, "detail": detail}) + return + code, resp = self._act(Path(d), action, body) + if once: + once_cache[once] = (code, resp) + while len(once_cache) > _ONCE_KEEP: + once_cache.popitem(last=False) + self.answer_json(code, resp) + + def _act(self, d: Path, action: str, body: dict) -> tuple[int, dict]: + """One control onto a registered resident: (status, the JSON answer). The resident reads + `hold`, `wake-now` and chores.toml at its next poll (hub/agent/agent.py); this process + only writes them, and notes each control in `/landing/controls.log`.""" + who = me or "unknown" + now = datetime.now().astimezone().isoformat(timespec="seconds") + extra: dict[str, Any] = {} + try: + if action == "hold": + reason = _one_line(_text(body, "reason")) # the first line is the resident's reason + (d / "hold").write_text(f"held by {who} from the app at {now}: {reason}\n") + code, detail = 200, f"held: {reason}" if reason else "held" + elif action == "resume": + try: + (d / "hold").unlink() + code, detail = 200, "resumed" + except FileNotFoundError: + code, detail = 200, "was not held" + elif action == "wake": + (d / "wake-now").write_text(f"{who} at {now}\n") + code, detail = 200, "wake requested" + elif action == "restart": + code, detail, extra = self._restart(d, body) + elif action == "cap": + code, detail, extra = self._cap(d, body) + else: + code, detail = 400, f"unknown action {action!r}: hold, resume, wake, restart or cap" + except OSError as e: + code, detail = 500, f"{action}: {type(e).__name__}: {e}" + try: + (d / "landing").mkdir(parents=True, exist_ok=True) + with open(d / "landing" / "controls.log", "a") as f: + f.write(f"{now} {who} {_one_line(action) or '-'} {code} {_one_line(detail)}\n") + except OSError: + pass # the note is for people; the control's own answer says what happened + return code, {"ok": code == 200, "action": action, "detail": detail, **extra} + + def _restart(self, d: Path, body: dict) -> tuple[int, str, dict]: + plists = sorted(d.glob("*.plist")) + if len(plists) != 1: + if not plists: + return 409, f"no launchd plist in {d}", {} + return 409, f"{len(plists)} launchd plists in {d}: which one is the job is not known", {} + try: + with open(plists[0], "rb") as f: + doc = plistlib.load(f) + except (OSError, plistlib.InvalidFileException, ValueError, ExpatError) as e: + return 409, f"could not read {plists[0].name}: {e}", {} + label = doc.get("Label") if isinstance(doc, dict) else None + if not isinstance(label, str) or not label: + return 409, f"{plists[0].name} has no Label", {} + force = body.get("force") in ("1", 1, True, "true") + if not force and not controls_info(str(d))["idle"]: + return 409, "in a wake", {"label": label} + argv = [_LAUNCHCTL, "kickstart", "-k", f"gui/{os.getuid()}/{label}"] + try: + proc = subprocess.run(argv, capture_output=True, text=True, timeout=15) + except (OSError, subprocess.TimeoutExpired) as e: + return 502, f"launchctl: {type(e).__name__}: {e}", {"label": label} + said = _one_line(proc.stderr or proc.stdout) + detail = f"kickstart {label}: rc {proc.returncode}" + (f": {said}" if said else "") + return (200 if proc.returncode == 0 else 502), detail, {"label": label, "rc": proc.returncode} + + def _cap(self, d: Path, body: dict) -> tuple[int, str, dict]: + values: dict[str, float | int] = {} + raw = body.get("cost_per_day") + if raw is not None and str(raw) != "": + try: + if isinstance(raw, bool): + raise ValueError + v = float(raw) + except (TypeError, ValueError): + return 400, f"cost_per_day {raw!r} is not a number", {} + if not math.isfinite(v) or v < 0: + return 400, f"cost_per_day {raw!r}: a finite number >= 0", {} + values["cost_per_day"] = v + raw = body.get("wakes_per_hour") + if raw is not None and str(raw) != "": + try: + if isinstance(raw, bool): + raise ValueError + n = float(raw) + except (TypeError, ValueError): + return 400, f"wakes_per_hour {raw!r} is not a number", {} + if not n.is_integer() or n < 0: + return 400, f"wakes_per_hour {raw!r}: a whole number >= 0", {} + values["wakes_per_hour"] = int(n) + if not values: + return 400, "cap needs cost_per_day and/or wakes_per_hour", {} + path = d / "chores.toml" + try: + text = path.read_text() + except FileNotFoundError: + return 409, f"no chores.toml in {d}", {} + try: + before = tomllib.loads(text) + except tomllib.TOMLDecodeError as e: + return 409, f"chores.toml does not parse, nothing was written: {e}", {} + new_text = _rewrite_budget(text, {k: repr(v) for k, v in values.items()}) + # the rewrite is textual: read it back and require that the two keys, and nothing else, + # changed, so a `[budget]` written some other way (dotted keys, an inline table) is + # refused rather than broken for the resident's next reload + try: + after = tomllib.loads(new_text) + except tomllib.TOMLDecodeError: + after = None + budget = before.get("budget", {}) + if not isinstance(budget, dict) or after != {**before, "budget": {**budget, **values}}: + return 409, "chores.toml's [budget] could not be rewritten in place; nothing was written", {} + tmp = path.with_name(f".{path.name}.{os.getpid()}.{threading.get_ident()}.tmp") + tmp.write_text(new_text) + os.chmod(tmp, path.stat().st_mode & 0o7777) + os.replace(tmp, path) + detail = " ".join(f"{k}={v}" for k, v in values.items()) + return 200, detail, {**values, "budget": after["budget"]} + + def log_message(self, fmt: str, *args) -> None: # one line per request is noise here + pass + + server = ThreadingHTTPServer((host, port), Handler) + bound = server.server_address[1] + hosts.update({f"{host}:{bound}", f"127.0.0.1:{bound}", f"localhost:{bound}"}) + for a in allow or []: + hosts.add(f"{a}:{bound}") + origins.update(f"{scheme}://{h}" for scheme in ("http", "https") for h in hosts) + who = f"you are {me}" if me else "no identity" + # flushed: under nohup or launchd stdout is a file, and the line would wait for the buffer + print( + f"lion areas on http://{host}:{bound}/ (registry {registry}); {who}; " + f"answers to {', '.join(sorted(hosts))}; Ctrl-C ends", + flush=True, + ) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + server.server_close() + + +def areas_main(a: argparse.Namespace) -> None: + registry = Path(a.file).expanduser() + if a.serve is not None: + serve(registry, a.host, a.serve, a.refresh, identity(a.as_dir), a.allow) + return + areas = load_areas(registry) + rows = status(areas) + if a.json: + print( + json.dumps( + {"areas": [vars(x) for x in areas], "residents": json.loads(rows_json(rows))}, indent=1 + ) + ) + else: + print(table(areas, rows)) diff --git a/apps/cli/lion_cli/chat.py b/apps/cli/lion_cli/chat.py new file mode 100644 index 0000000..3b831ba --- /dev/null +++ b/apps/cli/lion_cli/chat.py @@ -0,0 +1,428 @@ +"""`lion chat`: the actor a person talks to, built from the hub, and the command around the loop.""" + +from __future__ import annotations + +import argparse +import asyncio +import os +import re +import shlex +import signal +import subprocess +import sys +import time +import tomllib +from collections.abc import Callable +from pathlib import Path + +from hub import Box, BoxTree, DaytonaSandbox, LocalTree, Sandbox, code, files, guidance, khive_ops, shell +from hub.agent.chores import Bounded +from hub.chat import chat, load_log +from hub.hooks import install, load +from hub.khive import Khive, KhiveError +from hub.tools.code import Tree +from lionagi import Actor, Kind, Profile, lndl +from lionagi.actor import DEFAULT_HANDLERS, Run +from lionagi.record import Entry + +from .agent import backend_for +from .view import Console + +__all__ = ( + "console", + "prepare_box", + "take_patch", + "spent", + "resolve_chat", + "fresh_log", + "unsettled", + "Interrupt", + "chat_main", +) + + +def console( + directory: Path, + *, + name: str = "console", + system: str = "", + model: str = "an unnamed model", + rounds: int = 200, + sandbox: Box | None = None, + coding: bool = False, + notify: Callable[[str], None] | None = None, + tree: Tree | None = None, + about_box: str | None = None, +) -> tuple[Actor, Profile]: + """The actor a person talks to: the file and directory readers, khive when the directory is an + actor's own (its `.khive/config.toml` names the actor the reads sign as), a shell when a sandbox + is given, and with `coding` the tools a fix is found and written with (read_lines, search, edit) + plus the tree's own rules (AGENTS.md, CLAUDE.md at the root) in the guidance. `tree` is where the + file and code tools read and write: the directory itself by default, the copy of it in a box + with `about_box` saying so (the VM's own guidance otherwise).""" + directory = Path(directory).resolve() + tree = tree if tree is not None else LocalTree(directory) + actor = Actor(name, notes_dir=directory / ".lion" / "notes") + if (hooks := directory / ".lion" / "hooks.toml").exists(): + install(actor, load(hooks), directory) + specs = files(actor, tree) + tools = "read(path), list_dir(path)" + about_sandbox = "" + if sandbox is not None: + specs |= shell(actor, sandbox) + tools += ", bash(command), python(code)" + if about_box is not None: + about_sandbox = about_box + else: + network = " The VM has no network." if getattr(sandbox, "offline", False) else "" + about_sandbox = guidance("sandbox", network=network) + if coding: + specs |= code(actor, tree) + tools += ", read_lines(path, start, end), search(pattern, path, include), edit(path, old, new)" + rules = "\n\n".join( + f"{name} at the root says:\n{(directory / name).read_text(errors='replace').strip()}" + for name in ("AGENTS.md", "CLAUDE.md") + if (directory / name).exists() + ) + about_sandbox += "\n\n" + guidance("code", root=directory, branch=_branch(directory), rules=rules) + config = directory / ".khive" / "config.toml" + if config.exists(): + actor_id = tomllib.loads(config.read_text()).get("actor", {}).get("id") + if actor_id is None: + if notify is not None: + notify(f"{config}: no [actor].id; khive tool unavailable") + else: + try: + khive = Bounded(Khive(directory, actor_id), recipients=frozenset()) + except KhiveError as e: + if notify is not None: + notify(f"khive unavailable: {e}") + else: + specs |= khive_ops(actor, khive) + tools += ( + f", khive_ops(ops) signing as {actor_id} ({len(khive.allowed)} verbs: reads, comm.read, " + "schedule.remind, memory.remember at spawn salience)" + ) + profile = Profile( + name=name, + specs=frozenset(specs) | {h.name for h in DEFAULT_HANDLERS}, + system=system or guidance("console", model=model, tools=tools, rounds=rounds, sandbox=about_sandbox), + ) + return actor, profile + + +async def prepare_box(box: Box, directory: Path, say: Callable[[str], None]) -> str: + """Start `box`, push `directory` into it at the same path, and commit what the push archived as + the copy's baseline in a repository of its own (`git ls-files` leaves `.git` out); a tracked file + that .gitignore also matches is in the push, so it is added by name. Returns the baseline's SHA: + the patch at the end reads against it, whatever the model commits in the box.""" + await box.start() + pushed = await box.push(directory) # type: ignore[attr-defined] + say(f"box {getattr(box, 'image', '?')}: {len(pushed)} file(s) pushed to {box.workdir}") + git = "git -c user.email=lion@localhost -c user.name=lion" + line = f"cd {shlex.quote(box.workdir)} && {{ [ -e .git ] || git init -q; }} && {git} add -A && " + if pushed: + line += f"{git} --literal-pathspecs add -f --pathspec-from-file=- --pathspec-file-nul && " + line += f"{git} commit -qm baseline --allow-empty && git rev-parse HEAD" + r = await box.exec( + ["bash", "-o", "pipefail", "-c", line], + stdin=b"".join(os.fsencode(n) + b"\0" for n in pushed), + timeout=300, + ) + out = r["stdout"] + r["stderr"] + sha = r["stdout"].strip().rsplit("\n", 1)[-1] + if r["rc"] != 0 or re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", sha) is None: + raise RuntimeError(f"the box's git baseline failed (rc {r['rc']}): {out[-400:]}") + return sha + + +async def take_patch( + box: Box, + directory: Path, + chats: Path, + chat_id: str, + *, + baseline: str, + apply: bool, + say: Callable[[str], None], +) -> Path | None: + """The diff of the copy in `box` since `baseline` (prepare_box's commit), printed and saved beside + the chat log as `.patch`; with `apply` it is also applied to `directory` by `git apply`, + from the top of the repository `directory` sits in (a patch from a subdirectory gets its prefix), + which lands whole or not at all. None when nothing changed; a diff that fails raises. A patch that + will not apply is still saved, the failure is said, and the directory is left as it was.""" + patch = await box.diff(baseline) + if not patch.strip(): + say("box: no change in the copy") + return None + chats.mkdir(parents=True, exist_ok=True) + path = chats / f"{chat_id}.patch" + path.write_text(patch) + say(f"box: the copy's diff, {patch.count(chr(10))} line(s), saved as {path}") + print(patch, end="" if patch.endswith("\n") else "\n") + top = subprocess.run( + ["git", "-C", str(directory), "rev-parse", "--show-toplevel", "--show-prefix"], + capture_output=True, + text=True, + ) + if top.returncode == 0: # from a subdirectory `git apply` skips every path and exits 0 + root, prefix = top.stdout.split("\n")[:2] + argv = ["git", "-C", root, "apply", *([f"--directory={prefix}"] if prefix else []), str(path)] + else: # not a repository: git apply works the directory as plain files + argv = ["git", "-C", str(directory), "apply", str(path)] + if apply: + done = subprocess.run(argv, capture_output=True, text=True) + if done.returncode == 0: + say(f"box: applied to {directory} (git apply)") + else: + say(f"box: not applied, git apply exited {done.returncode}: {done.stderr.strip()[-300:]}") + else: + say(f"box: not applied; `{shlex.join(argv)}` does, or --apply next time") + return path + + +def spent(calls: list[dict]) -> float: + """What the session was billed: each call's `total_cost_usd` (every attempt, the failed ones + too), falling back to the last attempt's `usage.cost` for a backend that reports only that.""" + return sum(float(c.get("total_cost_usd") or (c.get("usage") or {}).get("cost") or 0) for c in calls) + + +def _branch(directory: Path) -> str: + """The checked-out branch, from git itself (so a linked worktree's `.git` file and a + subdirectory both resolve), or `(no git)`.""" + try: + result = subprocess.run( + ["git", "-C", str(directory), "rev-parse", "--abbrev-ref", "HEAD"], + capture_output=True, + text=True, + ) + except FileNotFoundError: + return "(no git)" + if result.returncode != 0: + return "(no git)" + return result.stdout.strip() + + +def resolve_chat(chats: Path, which: str) -> Path: + """`last`, a chat id or a unique prefix of one (the log's stem), or a path to a log.""" + logs = sorted(chats.glob("*.jsonl"), key=lambda p: p.stem) # by id: `X-02` after `X`, before `X+1s` + if which == "last": + if not logs: + raise SystemExit(f"nothing to continue: no chat logged under {chats} yet") + return logs[-1] + hits = [p for p in logs if p.stem.startswith(which)] + if exact := [p for p in hits if p.stem == which]: # `X` beside `X-02`: the id as given wins + return exact[0] + if len(hits) == 1: + return hits[0] + if len(hits) > 1: + raise SystemExit(f"chat {which!r} is ambiguous: {', '.join(p.stem for p in hits)}") + path = Path(which) + if path.exists(): + return path + raise SystemExit( + f"no chat {which!r}: none of {len(logs)} under {chats} starts with it, and it is not a path" + ) + + +def fresh_log(chats: Path, chat_id: str) -> tuple[str, Path]: + """A log file created here and now, exclusively: the id as given, or with `-02`, `-03`… when a + session already took it this second (two consoles started together must not share a log).""" + chats.mkdir(parents=True, exist_ok=True) + for n in range(1, 100): + name = chat_id if n == 1 else f"{chat_id}-{n:02d}" + try: + with (chats / f"{name}.jsonl").open("x"): + return name, chats / f"{name}.jsonl" + except FileExistsError: + continue + raise SystemExit(f"{chats}: 99 logs already start with {chat_id}") + + +def unsettled(history: tuple[Entry, ...]) -> list[str]: + """The commands a logged session asked for and never saw settle: their aliases, in order. A + session that ended mid-round leaves them; the model is told, and nothing runs again on its own.""" + asked: list[str] = [] + settled = {e.alias for e in history if e.kind is Kind.RESULT} + for e in history: + if e.kind is not Kind.TEXT: + continue + try: + asked += [la.alias for la in lndl.parse(str(e.content)).lacts] + except lndl.LNDLError: + continue + return [a for a in asked if a not in settled and not a.startswith("_")] + + +class Interrupt: + """Ctrl-C. Mid-run, the first one hands the turn back at the next round boundary (the round in + flight finishes, its results land, then `you ▸`); the second one, or one at the prompt, ends the + conversation now. The counter resets each time the person is asked.""" + + def __init__(self, runs: list[Run], task: asyncio.Task, say: Console): + self.runs, self.task, self.say = runs, task, say + self.presses = 0 + + def __call__(self) -> None: + self.presses += 1 + if self.presses == 1 and self.runs and not self.say.asking: + self.runs[0].pause = True + self.say.notice("stopping after this round; Ctrl-C again to quit now") + return + self.task.cancel() + + +async def chat_main(a: argparse.Namespace) -> None: + directory = Path(a.dir).resolve() + chats = directory / ".lion" / "chats" + prior = resolve_chat(chats, a.resume) if a.resume else None # before this session's own log exists + if a.log: + log = Path(a.log) + if log.exists(): + raise SystemExit(f"{log} exists; `lion chat -r {log}` continues it, another --log starts fresh") + chat_id = log.stem + else: + chat_id, log = fresh_log(chats, time.strftime("%Y%m%d-%H%M%S")) + try: + await _chat_main(a, directory, chats, prior, chat_id, log) + except BaseException: + # a start that failed before anything was logged (a bad hooks.toml, no image) must not leave + # an empty log as `-c`'s next target: the previous history would be lost to the next session + if log.exists() and log.stat().st_size == 0: # --log included: an existing path was refused above + log.unlink() + raise + + +async def _chat_main( + a: argparse.Namespace, directory: Path, chats: Path, prior: Path | None, chat_id: str, log: Path +) -> None: + boxed = getattr(a, "box", "vm") == "daytona" + sandbox: Box | None = None + tree = about_box = None + if boxed: + sandbox = DaytonaSandbox(a.image, workdir=str(directory)) + tree = BoxTree(sandbox) + about_box = guidance("box", kind="Daytona", image=a.image, root=directory) + elif a.shell or a.code: + sandbox = Sandbox(directory, f"lion-{chat_id}", offline=a.offline) + degraded: list[str] = [] + actor, profile = console( + directory, + model=a.model, + rounds=a.rounds, + sandbox=sandbox, + coding=a.code, + notify=degraded.append, + tree=tree, + about_box=about_box, + ) + backend = backend_for(a.model) + history: tuple = () + resumed = notice = None + if prior is not None: + torn: list[str] = [] + history = load_log(prior, torn) + resumed = (prior.stem, len(history)) + notes = [f"the previous session's log ended mid-write: {t}" for t in torn] + if left := unsettled(history): + notes.append( + f"the previous session ended before these commands settled: {', '.join(left)}. " + "They did not run again; ask again for any you still need." + ) + notice = " ".join(notes) or None + say = Console( + a.model, + log, + raw=a.raw, + echo=a.say is not None or not sys.stdin.isatty(), + verbose=a.verbose, + calls=getattr(backend, "calls", None), + ) + say.quiet_until = len(history) - 1 + say.open(a.model, directory, chat_id, sorted(profile.specs), resumed) + for msg in degraded: + say.notice(msg) + if notice: + say.notice(notice) + if hooked := len(actor.befores) + len(actor.afters) + len(actor.sections): + say.notice(f"{hooked} hook(s) from .lion/hooks.toml") + baseline = "" # the box copy's baseline commit, this conversation's own + if boxed: + try: + baseline = await prepare_box(sandbox, directory, say.notice) + except BaseException: # before any model work: nothing in the box is worth keeping + await asyncio.shield(sandbox.stop()) + raise + elif sandbox is not None: + if (build := await sandbox.ensure_image()) is not None: + raise SystemExit(f"the sandbox image {sandbox.image} did not build:\n{build}") + say.notice(f"sandbox {sandbox.name}: a VM over {directory}" + (", no network" if a.offline else "")) + await sandbox.start() + runs: list[Run] = [] + off = actor.bus.subscribe("run.started", lambda ev: runs.append(actor.runs[ev.payload["run"]])) + interrupt = Interrupt(runs, asyncio.current_task(), say) + asyncio.get_running_loop().add_signal_handler(signal.SIGINT, interrupt) + + async def ask() -> str | None: + interrupt.presses = 0 + return await say.ask() + + interrupted = False + try: + run = await chat( + actor, + profile, + backend, + ask, + say=say, + first=a.say, + max_rounds=a.rounds, + history=history, + notice=notice, + ) + ended = f"ended {type(run.outcome).__name__} after {run.round} round(s)" + except (KeyboardInterrupt, asyncio.CancelledError): + ended, interrupted = "interrupted", True + finally: + off() + if sandbox is not None: + try: + if boxed and sandbox.dead is not None: + say.notice( + f"box: it ended before its patch was read ({sandbox.dead}); recovery unavailable" + ) + elif boxed: # the patch first: stop deletes the box + try: + await asyncio.shield( + take_patch( + sandbox, + directory, + chats, + chat_id, + baseline=baseline, + apply=a.apply, + say=say.notice, + ) + ) + except Exception as e: # the patch was not taken whole: the box may hold the only copy + ident = getattr(sandbox, "box_id", None) + why = f"taking its patch failed ({type(e).__name__}: {e})" + if sandbox.dead is None and ident: + say.notice( + f"box {ident}: kept, not stopped: {why}; the work is there, at " + f"{sandbox.workdir}, until the box's idle stop deletes it" + ) + else: + reason = sandbox.dead or "the provider named no box" + say.notice(f"box {ident or '?'}: {why}; recovery unavailable: {reason}") + raise + await asyncio.shield(sandbox.stop()) + except asyncio.CancelledError: # a third Ctrl-C while the VM stops: still close, still leave + ended, interrupted = "interrupted", True + calls = getattr(backend, "calls", []) + say.close(ended, len(calls), spent(calls), log) + if interrupted: + # a backend call may still be on its thread; the loop's shutdown would wait for it + # (measured: a Ctrl-C that seemed to do nothing). The log is written; leave now. + sys.stdout.flush() + os._exit(130) diff --git a/apps/cli/lion_cli/context.py b/apps/cli/lion_cli/context.py new file mode 100644 index 0000000..cf54d9a --- /dev/null +++ b/apps/cli/lion_cli/context.py @@ -0,0 +1,257 @@ +"""`lion context {set,checkpoint,restore,status,show}`: a Claude Code session's checkpoint and restore, +from the session's directory (ADR-0019/D3). One implementation: the areas server's `/api/context/*` +routes call these functions; a refused take exits non-zero with the reason on stderr.""" + +from __future__ import annotations + +import argparse +import asyncio +import contextlib +import fcntl +import json +import os +import sys +from dataclasses import asdict +from datetime import datetime +from pathlib import Path +from typing import Any + +from hub.context import ( + PROJECTS, + TYPED, + actor_id, + context_figure, + cursor, + directives, + lanes, + newest_transcript, + posture, + slug, + spawns, + topic, + transcript_source, + transcript_turns, + watches, +) +from hub.khive import Khive, KhiveError +from lionagi.checkpoint import ( + KINDS, + WHYS, + CheckpointError, + Var, + assert_ignored, + load, + render_prompt, + save, + take, +) + +__all__ = ("BUDGET", "set_var", "checkpoint", "restore", "status", "show", "context_main") + +# tokens, estimated: the kept view is half of it, so a fresh session starts from a few thousand +BUDGET = 8_000 + + +def _now() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +@contextlib.contextmanager +def _pending_lock(directory: Path): + ctx = directory / ".khive" / "context" + ctx.mkdir(parents=True, exist_ok=True) + with (ctx / ".pending.lock").open("w") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + try: + yield ctx / "pending.jsonl" + finally: + fcntl.flock(lock, fcntl.LOCK_UN) + + +def _pending(directory: Path) -> tuple[list[Var], str]: + """The declared vars waiting for a checkpoint, and the text they were read from.""" + path = directory / ".khive" / "context" / "pending.jsonl" + try: + text = path.read_text() + except FileNotFoundError: + return [], "" + out = [] + for i, line in enumerate(text.splitlines(), 1): + if not line.strip(): + continue + try: + out.append(Var(**json.loads(line))) + except (ValueError, TypeError) as e: + raise CheckpointError(f"{path} line {i} is not a declared var ({e}); nothing was taken") from None + return out, text + + +def _drain(directory: Path, taken: str) -> None: + """The declared vars a checkpoint took leave `pending.jsonl`; any declared since stay.""" + if not taken: + return + with _pending_lock(directory) as path: + text = path.read_text() if path.exists() else "" + rest = text[len(taken) :] if text.startswith(taken) else text + tmp = path.with_name(f".{path.name}.{os.getpid()}.tmp") + tmp.write_text(rest) + os.replace(tmp, path) + + +def set_var(directory: str | Path, name: str, kind: str, value: Any) -> dict: + """A var declared mid-session, waiting in `.khive/context/pending.jsonl` for the next + checkpoint; the Var stored.""" + d = Path(directory).resolve() + if kind not in KINDS: + raise CheckpointError(f"kind {kind!r}: one of {', '.join(KINDS)}") + if not name.strip(): + raise CheckpointError("a declared var needs a name") + assert_ignored(d) + var = Var(name=name, kind=kind, value=value, source="declared", at=_now()) + with _pending_lock(d) as path, path.open("a") as f: + f.write(json.dumps(asdict(var), ensure_ascii=False, default=str) + "\n") + return asdict(var) + + +def checkpoint( + directory: str | Path, + why: str, + *, + session: str | None = None, + transcript: str | Path | None = None, + budget: int = BUDGET, + khive: Khive | None = None, + now: datetime | None = None, +) -> dict: + """Take and save a Claude Code session's checkpoint: `{id, path, context, vars, kept, hidden}`. The + gitignore assertion runs before anything is read; lanes khive cannot give are a value, not a refusal.""" + d = Path(directory).resolve() + if why not in WHYS: + raise CheckpointError(f"why {why!r}: one of {', '.join(WHYS)}") + assert_ignored(d) + actor = actor_id(d) + path = Path(transcript).expanduser() if transcript else newest_transcript(d) + if not path.is_file(): + raise CheckpointError(f"no transcript at {path}") + sid = session or path.stem + lane_vars: list[Var] = [] + if khive is None: + try: + khive = Khive(d, actor) + except KhiveError as e: + lane_note = f"unavailable: {e}" + if khive is not None: + lane_vars, lane_note = asyncio.run(lanes(khive)) + declared, taken = _pending(d) + harvested = [ + *topic(PROJECTS / slug(d) / "memory" / "sessions_registry.json"), + *directives(sid, path, TYPED), + *watches(path, actor, now), + *spawns(path, now), + *lane_vars, + *posture(path), + *cursor(d), + ] + source = { + "kind": "claude-code", + "session_id": sid, + **transcript_source(path), + "cwd": str(d), + "context": context_figure(sid), + "lanes": lane_note, + } + view = transcript_turns(path, budget) + ckpt = take(source, [*harvested, *declared], actor=actor, why=why, budget=budget, view=view) + written = save(ckpt, d) + _drain(d, taken) + return { + "id": written.stem, + "path": str(written), + "context": source["context"], + "vars": [v.name for v in ckpt.vars], + "kept": len(view["kept"]), + "hidden": len(view["hidden"]), + } + + +def restore(directory: str | Path, id: str | None = None) -> dict: + """`{id, first_prompt, watches, spawns}` of the checkpoint `id`, the latest by default.""" + c = load(Path(directory).resolve(), id) + return { + "id": c.id, + "first_prompt": render_prompt(c), + "watches": [v.value for v in c.vars if v.kind == "watch"], + "spawns": [v.value for v in c.vars if v.kind == "spawn"], + } + + +def status(directory: str | Path, *, session: str | None = None) -> dict: + """`{latest, taken_at, context_now, stale, pending_vars}`. Stale: no checkpoint, or the transcript or + record it points at changed after the take.""" + d = Path(directory).resolve() + try: + c = load(d) + except CheckpointError: + c = None + if session is None: + try: + session = newest_transcript(d).stem + except CheckpointError: + session = None + stale = True + if c is not None: + pointed = c.source.get("transcript") or c.source.get("record") + try: + changed = datetime.fromtimestamp(Path(str(pointed)).stat().st_mtime).astimezone() + stale = changed > datetime.fromisoformat(c.taken_at) + except (OSError, ValueError, TypeError): + stale = True + return { + "latest": c.id if c else None, + "taken_at": c.taken_at if c else None, + "context_now": context_figure(session), + "stale": stale, + "pending_vars": len(_pending(d)[0]), + } + + +def show(directory: str | Path, name: str, id: str | None = None) -> Any: + """The value of the var `name` in the checkpoint `id`, the latest by default: what a lazy var's line + in the restore prompt points at. A var whose value lives elsewhere (its ref) is refused, saying where.""" + c = load(Path(directory).resolve(), id) + v = next((v for v in c.vars if v.name == name), None) + if v is None: + raise CheckpointError(f"{c.id} has no var {name!r}") + if v.ref is not None and v.value is None: + raise CheckpointError(f"{c.id} does not carry {name!r} ({v.size} chars): {v.ref} loads it") + return v.value + + +def _value(text: str) -> Any: + """A declared value: JSON when it reads as an object or a list, else the text as given.""" + if text.lstrip().startswith(("{", "[")): + try: + return json.loads(text) + except ValueError: + pass + return text + + +def context_main(a: argparse.Namespace) -> None: + try: + if a.verb == "set": + print(json.dumps(set_var(a.dir, a.name, a.kind, _value(a.value)), ensure_ascii=False)) + elif a.verb == "checkpoint": + done = checkpoint(a.dir, a.why, session=a.session, transcript=a.transcript, budget=a.budget) + print(json.dumps(done, ensure_ascii=False, indent=1)) + elif a.verb == "restore": + got = restore(a.dir, a.id) + print(json.dumps(got, ensure_ascii=False, indent=1) if a.json else got["first_prompt"]) + elif a.verb == "status": + print(json.dumps(status(a.dir, session=a.session), ensure_ascii=False, indent=1)) + elif a.verb == "show": + value = show(a.dir, a.var, a.id) + print(value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=1)) + except CheckpointError as e: + print(f"lion context {a.verb}: {e}", file=sys.stderr) + raise SystemExit(1) from None diff --git a/apps/cli/lion_cli/main.py b/apps/cli/lion_cli/main.py new file mode 100644 index 0000000..fbb7cdc --- /dev/null +++ b/apps/cli/lion_cli/main.py @@ -0,0 +1,163 @@ +"""The parser and the entry: `lion `.""" + +from __future__ import annotations + +import argparse +import asyncio + +from lionagi.checkpoint import KINDS, WHYS + +from .agent import agent_main +from .areas import areas_main +from .chat import chat_main +from .context import BUDGET, context_main + +__all__ = ("parser", "main") + + +def parser() -> argparse.ArgumentParser: + ap = argparse.ArgumentParser(prog="lion", description="lionagi v1") + sub = ap.add_subparsers(dest="command", required=True) + ag = sub.add_parser("agent", help="an actor's long-running agent, from its directory") + ag.add_argument( + "--dir", required=True, help="the agent directory: .khive/config.toml, chores.toml, notes, ledger" + ) + ag.add_argument("--model", default=None, help="the model; overrides chores.toml") + ag.add_argument("--once", action="store_true", help="one wake, the record printed, then exit") + ag.add_argument( + "--rerun", + nargs="+", + default=None, + metavar="ID", + help="read mail back through one wake, by id; then exit", + ) + ag.add_argument("--wait", type=int, default=30, help="seconds --once waits for mail") + ag.add_argument("--check", default=None, help="run one instrument, print and land it; nothing is sent") + ag.add_argument("--ticks", action="store_true", help="create the missing tick reminders and exit") + ag.add_argument("--arm", action="store_true", help="the landing arm: identity and boundary, then exit") + ag.add_argument("--stats", action="store_true", help="the day's tokens, wakes, mail and idle polls") + ch = sub.add_parser("chat", help="talk with an actor at the terminal; the record prints as it grows") + ch.add_argument( + "--dir", default=".", help="the console's directory: files, and khive if it holds .khive/config.toml" + ) + ch.add_argument("--model", default="deepseek/deepseek-v4.1-flash") + ch.add_argument("--say", default=None, help="the first message; otherwise asked for") + ch.add_argument("--rounds", type=int, default=200, help="the run's round budget; the model is told") + ch.add_argument( + "--shell", + action="store_true", + help="give the model bash and python in a Linux VM over this directory", + ) + ch.add_argument("--offline", action="store_true", help="the VM gets no network (with --shell)") + ch.add_argument( + "--box", + choices=("vm", "daytona"), + default="vm", + help="where bash, python and the file tools work: the local VM over this directory (vm), or a " + "Daytona box the directory is pushed into at the same path; its diff comes back as a patch", + ) + ch.add_argument("--image", default="python:3.12", help="the Daytona box's image (with --box daytona)") + ch.add_argument( + "--apply", + action="store_true", + help="apply the box's patch to this directory at the end (git apply --3way); default: print + save", + ) + ch.add_argument( + "--code", + action="store_true", + help="the coding preset: read_lines, search and edit on this directory, the shell VM (implies " + "--shell), and the root AGENTS.md/CLAUDE.md in the guidance", + ) + ch.add_argument( + "--log", default=None, help="the record, one JSON line per entry; default .lion/chats/