From e239408adb27c79fb4352c264c52b98fa4637021 Mon Sep 17 00:00:00 2001 From: wondercreatemaster <171356886+wondercreatemaster@users.noreply.github.com> Date: Fri, 31 Jul 2026 03:13:38 +0000 Subject: [PATCH] feat(cli): agent-native provisioning with init --agent and agents claim Closes #606. Lets an agent provision its own KB and identity, keep a local credential off stdout, and hand a human a claim token that binds the agent to a project through the adopt review gate. --- CHANGELOG.md | 13 + src/vouch/adopt.py | 117 ++++++--- src/vouch/agent_provision.py | 465 ++++++++++++++++++++++++++++++++++ src/vouch/cli.py | 146 ++++++++++- src/vouch/jsonl_server.py | 14 +- src/vouch/server.py | 17 +- tests/test_agent_provision.py | 255 +++++++++++++++++++ 7 files changed, 987 insertions(+), 40 deletions(-) create mode 100644 src/vouch/agent_provision.py create mode 100644 tests/test_agent_provision.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e5a72cb..a6d8fdda 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ All notable changes to vouch are documented here. Format follows ## [Unreleased] ### Added +- **agent-native provisioning — `vouch init --agent` + `vouch agents claim`** + (#606): an agent can provision its own identity and agent-scoped KB without + a human handing it one first. `--agent --agent-caller ` creates the KB + under `$XDG_DATA_HOME/vouch/agents//`, stamps `agent.caller` into + config as the persistent proposer identity (MCP/JSONL/CLI fall back to it + when `VOUCH_AGENT` is unset), and writes a local credential to + `~/.config/vouch/agent-credentials.yaml` (chmod 0600) that is **never** + echoed to stdout. It emits a claim token plus the exact + `vouch agents claim ` command (and `--json` for skills). Claiming + from a project KB transfers ownership and moves knowledge through the same + review gate as `vouch adopt`; the agent's credential and its own KB + artifacts stay untouched. Unclaimed agent proposals live only in the agent + KB until claim — they are not visible to the project beforehand. - **explicit pins — a working set that always enters the pack** (#615): `vouch pin ` / `vouch pins list` / `vouch unpin `. Pinned claims and pages lead every context pack instead of having to win the query each turn, diff --git a/src/vouch/adopt.py b/src/vouch/adopt.py index 6fa4b3ac..33cf61ba 100644 --- a/src/vouch/adopt.py +++ b/src/vouch/adopt.py @@ -37,9 +37,7 @@ # Claim statuses that never travel: superseded/archived knowledge was # retired on purpose, redacted knowledge must not propagate. -_DEAD_STATUSES = frozenset( - {ClaimStatus.SUPERSEDED, ClaimStatus.ARCHIVED, ClaimStatus.REDACTED} -) +_DEAD_STATUSES = frozenset({ClaimStatus.SUPERSEDED, ClaimStatus.ARCHIVED, ClaimStatus.REDACTED}) @dataclass @@ -91,16 +89,12 @@ def find_adoptable_sources(personal: KBStore, match_root: Path) -> list[Source]: out: list[Source] = [] for src in personal.list_sources(): origin_path = src.metadata.get("origin_path") - if isinstance(origin_path, str) and origin_path and _origin_matches( - origin_path, root - ): + if isinstance(origin_path, str) and origin_path and _origin_matches(origin_path, root): out.append(src) return out -def _claims_citing( - personal: KBStore, source_ids: set[str] -) -> list[tuple[Claim, Evidence | None]]: +def _claims_citing(personal: KBStore, source_ids: set[str]) -> list[tuple[Claim, Evidence | None]]: """Live personal claims citing any of ``source_ids``, with a receipt if any. A claim cites a source either directly (a bare source id in evidence) or @@ -148,25 +142,84 @@ def adopt( rejected by the receipt resolver. ``dry_run`` reports without writing. """ root = Path(match_root).resolve() - personal_identity = personal.identity() + sources = find_adoptable_sources(personal, root) + return _adopt_sources( + project, + personal, + sources, + origin=str(root), + rationale_prefix=(f"adopted from personal KB {{from_kb}} — captured in {root}"), + receipt_reason="adopted from personal KB (receipt re-verified)", + actor=actor, + retire=retire, + dry_run=dry_run, + pages_pending=_pending_pages_for_origin(personal, root), + ) + + +def adopt_kb( + project: KBStore, + source_kb: KBStore, + *, + actor: str = ADOPT_ACTOR, + retire: bool = False, + dry_run: bool = False, + origin_label: str | None = None, +) -> AdoptReport: + """Adopt every live source/claim from ``source_kb`` into ``project``. + + Same gate path as :func:`adopt`, but without an origin-folder filter — + used when the source KB *is* the agent's own store (agent-native + provisioning claim), not a personal catch-all that stamped + ``metadata.origin_path``. + """ + sources = list(source_kb.list_sources()) + label = origin_label or str(source_kb.root) + return _adopt_sources( + project, + source_kb, + sources, + origin=label, + rationale_prefix=f"adopted from agent KB {{from_kb}} ({label})", + receipt_reason="adopted from agent KB (receipt re-verified)", + actor=actor, + retire=retire, + dry_run=dry_run, + pages_pending=[], + ) + + +def _adopt_sources( + project: KBStore, + source_kb: KBStore, + sources: list[Source], + *, + origin: str, + rationale_prefix: str, + receipt_reason: str, + actor: str, + retire: bool, + dry_run: bool, + pages_pending: list[str], +) -> AdoptReport: + source_identity = source_kb.identity() project_identity = project.identity() + from_kb = source_identity[0] if source_identity else None report = AdoptReport( - origin=str(root), - from_kb=personal_identity[0] if personal_identity else None, + origin=origin, + from_kb=from_kb, to_kb=project_identity[0] if project_identity else None, dry_run=dry_run, ) - report.pages_pending_in_personal = _pending_pages_for_origin(personal, root) - sources = find_adoptable_sources(personal, root) + report.pages_pending_in_personal = list(pages_pending) if not sources: return report source_ids = {s.id for s in sources} - pairs = _claims_citing(personal, source_ids) + pairs = _claims_citing(source_kb, source_ids) + rationale = rationale_prefix.format(from_kb=from_kb or "(no id)") if dry_run: - report.sources = sorted( - sid for sid in source_ids if not _source_exists(project, sid) - ) + report.sources = sorted(sid for sid in source_ids if not _source_exists(project, sid)) queued = _pending_payload_ids(project) # Predict against the PROJECT's real gate — a dry run that promises # durable claims a closed gate will leave pending is worse than no @@ -184,7 +237,7 @@ def adopt( for src in sources: if _source_exists(project, src.id): continue # content-addressed: already here from a prior pass - content = personal.read_source_content(src.id) + content = source_kb.read_source_content(src.id) project.put_source( content, title=src.title, @@ -195,7 +248,7 @@ def adopt( **src.metadata, "adopted_from": report.from_kb, }, - # The project's own stamp, not the personal KB's: from here on + # The project's own stamp, not the source KB's: from here on # this knowledge belongs to this project. scope=proposals_mod.default_scope(project), ) @@ -206,7 +259,7 @@ def adopt( # another copy of the same claim into the review queue. queued = _pending_payload_ids(project) # Only claims that actually landed DURABLE in the project may be retired - # from the personal KB. Archiving one that is merely pending would strand + # from the source KB. Archiving one that is merely pending would strand # it: reject or expire the proposal and the knowledge is live in neither # KB, with no unarchive path and no second adopt pass (archived claims are # skipped as dead). @@ -215,10 +268,6 @@ def adopt( if _already_durable(project, claim) or claim.id in queued: report.claims_skipped.append(claim.id) continue - rationale = ( - f"adopted from personal KB {report.from_kb or '(no id)'} — " - f"captured in {root}" - ) if receipt is not None and receipt.quote: result = proposals_mod.propose_quoted_claim( project, @@ -241,7 +290,7 @@ def adopt( project, result.proposal, actor=actor, - reason="adopted from personal KB (receipt re-verified)", + reason=receipt_reason, ) if durable is not None: report.claims_durable.append(durable.id) @@ -277,9 +326,9 @@ def adopt( if retire: for claim_id in landed_durable: try: - lifecycle.archive(personal, claim_id=claim_id, actor=actor) + lifecycle.archive(source_kb, claim_id=claim_id, actor=actor) except Exception: - # Retiring is best-effort tidying of the personal KB; a claim + # Retiring is best-effort tidying of the source KB; a claim # that cannot be archived must not fail the adoption. continue report.retired.append(claim_id) @@ -300,7 +349,7 @@ def adopt( data={**data, "direction": "in", "from_kb": report.from_kb}, ) audit_mod.log_event( - personal.kb_dir, + source_kb.kb_dir, event="kb.adopt", actor=actor, data={**data, "direction": "out", "to_kb": report.to_kb}, @@ -324,8 +373,10 @@ def _pending_pages_for_origin(personal: KBStore, match_root: Path) -> list[str]: if not isinstance(meta, dict): continue origin_path = meta.get("origin_path") - if isinstance(origin_path, str) and origin_path and _origin_matches( - origin_path, match_root + if ( + isinstance(origin_path, str) + and origin_path + and _origin_matches(origin_path, match_root) ): out.append(proposal.id) return out @@ -334,9 +385,7 @@ def _pending_pages_for_origin(personal: KBStore, match_root: Path) -> list[str]: def _receipts_auto_approve(project: KBStore) -> bool: """Whether this KB's gate lets a verified receipt land durable by itself.""" cfg = proposals_mod._review_config(project) - return bool(cfg.get("auto_approve_on_receipt")) or ( - cfg.get("approver_role") == "trusted-agent" - ) + return bool(cfg.get("auto_approve_on_receipt")) or (cfg.get("approver_role") == "trusted-agent") def _pending_payload_ids(project: KBStore) -> set[str]: diff --git a/src/vouch/agent_provision.py b/src/vouch/agent_provision.py new file mode 100644 index 00000000..6218d135 --- /dev/null +++ b/src/vouch/agent_provision.py @@ -0,0 +1,465 @@ +"""Agent-native provisioning — ``vouch init --agent`` + ``vouch agents claim``. + +An agent that wants durable memory today has to be handed a KB by a human +first. This module inverts that (issue #606): the agent runs one command, +provisions its own identity and an agent-scoped KB, keeps a local credential +that is never printed, and emits a claim token the human runs to bind the +agent to a project KB. Claiming reuses the adopt gate path so nothing lands +past review; the agent's credential and its own KB artifacts stay untouched. + +Secrets live outside any ``.vouch/`` (same split as hub tokens): +``$XDG_CONFIG_HOME/vouch/agent-credentials.yaml`` (chmod 0600). Agent KB +content lives under ``$XDG_DATA_HOME/vouch/agents//``. +""" + +from __future__ import annotations + +import contextlib +import os +import re +import secrets +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import yaml + +from . import adopt as adopt_mod +from . import audit as audit_mod +from .models import utcnow_iso +from .storage import KBStore + +CALLER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") +CREDS_ENV = "VOUCH_AGENT_CREDS_PATH" +AGENTS_DATA_ENV = "VOUCH_AGENTS_DATA" +CLAIM_ACTOR = "vouch-agent-claim" + + +class AgentProvisionError(RuntimeError): + """A provision or claim operation could not be completed.""" + + +@dataclass(frozen=True) +class AgentRecord: + """One provisioned agent, secrets included — never log or echo ``credential``.""" + + caller: str + kb_root: str + credential: str + claim_token: str + created_at: str + claimed_at: str | None = None + claimed_project_kb_id: str | None = None + claimed_project_root: str | None = None + + @property + def kb_dir(self) -> Path: + return Path(self.kb_root) / ".vouch" + + def public_dict(self) -> dict[str, Any]: + """JSON-safe view with the credential omitted.""" + return { + "caller": self.caller, + "kb_root": self.kb_root, + "kb_dir": str(self.kb_dir), + "claim_token": self.claim_token, + "claim_command": f"vouch agents claim {self.claim_token}", + "credential_path": str(credentials_path()), + "created_at": self.created_at, + "claimed_at": self.claimed_at, + "claimed_project_kb_id": self.claimed_project_kb_id, + "claimed_project_root": self.claimed_project_root, + "unclaimed": self.claimed_at is None, + } + + +@dataclass(frozen=True) +class ProvisionResult: + record: AgentRecord + created_kb: bool + credential_path: Path + + def public_dict(self) -> dict[str, Any]: + out = self.record.public_dict() + out["created_kb"] = self.created_kb + out["credential_path"] = str(self.credential_path) + return out + + +@dataclass(frozen=True) +class ClaimResult: + record: AgentRecord + adopt: adopt_mod.AdoptReport + already_claimed: bool + + def public_dict(self) -> dict[str, Any]: + return { + "caller": self.record.caller, + "kb_root": self.record.kb_root, + "already_claimed": self.already_claimed, + "claimed_at": self.record.claimed_at, + "claimed_project_kb_id": self.record.claimed_project_kb_id, + "claimed_project_root": self.record.claimed_project_root, + "adopt": self.adopt.as_dict(), + } + + +def credentials_path() -> Path: + forced = os.environ.get(CREDS_ENV) + if forced: + return Path(forced).expanduser() + xdg = os.environ.get("XDG_CONFIG_HOME") + base = Path(xdg) if xdg else Path.home() / ".config" + return base / "vouch" / "agent-credentials.yaml" + + +def agents_data_root() -> Path: + forced = os.environ.get(AGENTS_DATA_ENV) + if forced: + return Path(forced).expanduser() + xdg = os.environ.get("XDG_DATA_HOME") + if xdg: + return Path(xdg) / "vouch" / "agents" + return Path.home() / ".local" / "share" / "vouch" / "agents" + + +def validate_caller(caller: str) -> str: + name = caller.strip() + if not CALLER_RE.fullmatch(name): + raise AgentProvisionError( + "agent-caller must be 1-64 chars: letters, digits, " + "`.`, `_`, `-`, starting with alphanumeric" + ) + return name + + +def _load_raw() -> dict[str, Any]: + path = credentials_path() + if not path.exists(): + return {"version": 1, "agents": {}} + try: + loaded = yaml.safe_load(path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError) as e: + raise AgentProvisionError(f"cannot read {path}: {e}") from e + if not isinstance(loaded, dict): + return {"version": 1, "agents": {}} + agents = loaded.get("agents") + if not isinstance(agents, dict): + loaded["agents"] = {} + loaded.setdefault("version", 1) + return loaded + + +def _save_raw(data: dict[str, Any]) -> Path: + path = credentials_path() + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=path.name + ".") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(yaml.safe_dump(data, sort_keys=False, allow_unicode=True)) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_name, path) + path.chmod(0o600) + except BaseException: + with contextlib.suppress(OSError): + os.unlink(tmp_name) + raise + return path + + +def _parse_record(caller: str, raw: object) -> AgentRecord | None: + if not isinstance(raw, dict): + return None + kb_root = raw.get("kb_root") + credential = raw.get("credential") + claim_token = raw.get("claim_token") + if not ( + isinstance(kb_root, str) + and kb_root + and isinstance(credential, str) + and credential + and isinstance(claim_token, str) + and claim_token + ): + return None + return AgentRecord( + caller=caller, + kb_root=kb_root, + credential=credential, + claim_token=claim_token, + created_at=str(raw.get("created_at") or ""), + claimed_at=raw.get("claimed_at") if isinstance(raw.get("claimed_at"), str) else None, + claimed_project_kb_id=( + raw.get("claimed_project_kb_id") + if isinstance(raw.get("claimed_project_kb_id"), str) + else None + ), + claimed_project_root=( + raw.get("claimed_project_root") + if isinstance(raw.get("claimed_project_root"), str) + else None + ), + ) + + +def _record_to_raw(record: AgentRecord) -> dict[str, Any]: + out: dict[str, Any] = { + "caller": record.caller, + "kb_root": record.kb_root, + "credential": record.credential, + "claim_token": record.claim_token, + "created_at": record.created_at, + } + if record.claimed_at is not None: + out["claimed_at"] = record.claimed_at + if record.claimed_project_kb_id is not None: + out["claimed_project_kb_id"] = record.claimed_project_kb_id + if record.claimed_project_root is not None: + out["claimed_project_root"] = record.claimed_project_root + return out + + +def list_records() -> list[AgentRecord]: + data = _load_raw() + agents = data.get("agents") or {} + out: list[AgentRecord] = [] + if not isinstance(agents, dict): + return out + for caller, raw in agents.items(): + if not isinstance(caller, str): + continue + rec = _parse_record(caller, raw) + if rec is not None: + out.append(rec) + return out + + +def find_by_caller(caller: str) -> AgentRecord | None: + name = validate_caller(caller) + for rec in list_records(): + if rec.caller == name: + return rec + return None + + +def find_by_claim_token(token: str) -> AgentRecord | None: + needle = token.strip() + if not needle: + raise AgentProvisionError("claim token is empty") + for rec in list_records(): + if secrets.compare_digest(rec.claim_token, needle): + return rec + return None + + +def agent_kb_root(caller: str, *, path: Path | None = None) -> Path: + if path is not None: + return path.resolve() + return (agents_data_root() / validate_caller(caller)).resolve() + + +def stamp_agent_config(store: KBStore, *, caller: str, unclaimed: bool) -> None: + """Persist the proposer identity in the agent KB's config.yaml.""" + text = store.config_path.read_text(encoding="utf-8") + loaded = yaml.safe_load(text) or {} + if not isinstance(loaded, dict): + loaded = {} + agent = loaded.get("agent") + block = dict(agent) if isinstance(agent, dict) else {} + block["caller"] = caller + block["unclaimed"] = unclaimed + if "provisioned_at" not in block: + block["provisioned_at"] = utcnow_iso() + loaded["agent"] = block + store.config_path.write_text( + yaml.safe_dump(loaded, sort_keys=False, allow_unicode=True), + encoding="utf-8", + ) + + +def caller_from_store(store: KBStore) -> str | None: + """Read ``agent.caller`` from a KB's config, if stamped.""" + try: + loaded = yaml.safe_load(store.config_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return None + if not isinstance(loaded, dict): + return None + agent = loaded.get("agent") + if not isinstance(agent, dict): + return None + caller = agent.get("caller") + return caller if isinstance(caller, str) and caller.strip() else None + + +def _new_secret() -> str: + return secrets.token_urlsafe(32) + + +def provision( + caller: str, + *, + bootstrap, + path: Path | None = None, + actor: str, +) -> ProvisionResult: + """Create (or re-bind) an agent-scoped KB and write local credentials. + + ``bootstrap`` is ``cli._bootstrap_kb`` — injected so this module stays + free of click and onboarding import cycles. Returns a result whose + ``public_dict`` never includes the credential value. + """ + name = validate_caller(caller) + existing = find_by_caller(name) + if existing is not None and existing.claimed_at is None: + # Idempotent re-run before claim: keep the same credential + token so + # a skill that retries init does not invalidate a token already shown. + kb_root = Path(existing.kb_root) + if not (kb_root / ".vouch").is_dir(): + raise AgentProvisionError( + f"agent {name!r} is registered at {kb_root} but the KB is " + "missing — remove its credentials row or restore the directory" + ) + return ProvisionResult( + record=existing, + created_kb=False, + credential_path=credentials_path(), + ) + if existing is not None and existing.claimed_at is not None: + raise AgentProvisionError( + f"agent {name!r} was already claimed " + f"(project {existing.claimed_project_root}); pick a new --agent-caller" + ) + + root = agent_kb_root(name, path=path) + root.mkdir(parents=True, exist_ok=True) + created = not (root / ".vouch" / "config.yaml").exists() + store, _seed, _template = bootstrap(root) + stamp_agent_config(store, caller=name, unclaimed=True) + identity = store.identity() + audit_mod.log_event( + store.kb_dir, + event="agent.provision", + actor=actor, + data={ + "caller": name, + "kb_id": identity[0] if identity else None, + "created_kb": created, + }, + ) + + record = AgentRecord( + caller=name, + kb_root=str(root), + credential=_new_secret(), + claim_token=_new_secret(), + created_at=utcnow_iso(), + ) + data = _load_raw() + agents = data.setdefault("agents", {}) + if not isinstance(agents, dict): + agents = {} + data["agents"] = agents + agents[name] = _record_to_raw(record) + cred_path = _save_raw(data) + return ProvisionResult(record=record, created_kb=created, credential_path=cred_path) + + +def claim( + token: str, + project: KBStore, + *, + actor: str, + dry_run: bool = False, + retire: bool = False, +) -> ClaimResult: + """Bind an unclaimed agent to ``project`` and adopt its knowledge through the gate.""" + record = find_by_claim_token(token) + if record is None: + raise AgentProvisionError("unknown claim token") + + identity = project.identity() + if identity is None: + identity = project.ensure_identity() + project_id = identity[0] + + if record.claimed_at is not None: + if record.claimed_project_kb_id == project_id: + # Idempotent: already bound here — still run adopt so late knowledge moves. + agent_store = KBStore(Path(record.kb_root)) + report = adopt_mod.adopt_kb( + project, + agent_store, + actor=CLAIM_ACTOR, + retire=retire, + dry_run=dry_run, + origin_label=f"agent:{record.caller}", + ) + return ClaimResult(record=record, adopt=report, already_claimed=True) + raise AgentProvisionError( + f"agent {record.caller!r} is already claimed by " + f"{record.claimed_project_root} ({record.claimed_project_kb_id})" + ) + + if project.kb_dir.resolve() == record.kb_dir.resolve(): + raise AgentProvisionError("claim must run inside a project KB, not the agent's own KB") + + agent_store = KBStore(Path(record.kb_root)) + if not agent_store.config_path.exists(): + raise AgentProvisionError(f"agent KB at {record.kb_root} is missing — cannot claim") + + report = adopt_mod.adopt_kb( + project, + agent_store, + actor=CLAIM_ACTOR, + retire=retire, + dry_run=dry_run, + origin_label=f"agent:{record.caller}", + ) + if dry_run: + return ClaimResult(record=record, adopt=report, already_claimed=False) + + claimed = AgentRecord( + caller=record.caller, + kb_root=record.kb_root, + credential=record.credential, + claim_token=record.claim_token, + created_at=record.created_at, + claimed_at=utcnow_iso(), + claimed_project_kb_id=project_id, + claimed_project_root=str(project.root), + ) + data = _load_raw() + agents = data.setdefault("agents", {}) + if not isinstance(agents, dict): + agents = {} + data["agents"] = agents + agents[claimed.caller] = _record_to_raw(claimed) + _save_raw(data) + stamp_agent_config(agent_store, caller=claimed.caller, unclaimed=False) + + audit_mod.log_event( + project.kb_dir, + event="agent.claim", + actor=actor, + data={ + "caller": claimed.caller, + "agent_kb_id": (agent_store.identity() or (None, None))[0], + "sources": len(report.sources), + "claims_durable": len(report.claims_durable), + "claims_pending": len(report.claims_pending), + }, + ) + audit_mod.log_event( + agent_store.kb_dir, + event="agent.claim", + actor=actor, + data={ + "caller": claimed.caller, + "project_kb_id": project_id, + "project_root": str(project.root), + }, + ) + return ClaimResult(record=claimed, adopt=report, already_claimed=False) diff --git a/src/vouch/cli.py b/src/vouch/cli.py index dfcaeb0f..4caf6154 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -27,6 +27,7 @@ from . import __version__, bundle, health, hub_client, volunteer_context from . import adopt as adopt_mod +from . import agent_provision as agent_provision_mod from . import audit as audit_mod from . import capture as capture_mod from . import chatgpt_import as chatgpt_import_mod @@ -116,6 +117,7 @@ def _cli_errors() -> Iterator[None]: migrations_mod.MigrationError, chatgpt_import_mod.ChatGPTImportError, codex_rollout_mod.CodexRolloutError, + agent_provision_mod.AgentProvisionError, pins_mod.PinError, ) as e: raise click.ClickException(str(e)) from e @@ -157,7 +159,17 @@ def _whoami() -> str: # agent invokes the CLI it sets VOUCH_AGENT; honour it as the actor so # multi-agent attribution stays consistent across transports. VOUCH_USER # remains an escape hatch; OS user is the friendly default for humans. - return os.environ.get("VOUCH_AGENT") or os.environ.get("VOUCH_USER") or getpass.getuser() + # Agent-provisioned KBs stamp `agent.caller` into config so the proposer + # identity survives even when the host forgot to export VOUCH_AGENT. + env = os.environ.get("VOUCH_AGENT") or os.environ.get("VOUCH_USER") + if env: + return env + with contextlib.suppress(Exception): + store = KBStore(discover_root()) + stamped = agent_provision_mod.caller_from_store(store) + if stamped: + return stamped + return getpass.getuser() def _emit_json(obj) -> None: @@ -262,14 +274,83 @@ def _bootstrap_kb( type=click.Choice(available_templates()), help="Seed preset applied on top of the starter KB.", ) -def init(path: str, template: str) -> None: - """Initialise a .vouch/ knowledge base at PATH.""" +@click.option( + "--agent/--no-agent", + default=False, + help="Provision an agent-scoped KB + local credential + claim token " + "(see --agent-caller). The credential is never printed.", +) +@click.option( + "--agent-caller", + default=None, + help="Persistent proposer identity for --agent (required with --agent).", +) +@click.option( + "--json", + "as_json", + is_flag=True, + help="Emit machine-readable JSON (agent mode: includes claim_command, " + "never the credential value).", +) +def init( + path: str, + template: str, + agent: bool, + agent_caller: str | None, + as_json: bool, +) -> None: + """Initialise a .vouch/ knowledge base at PATH. + + With ``--agent``, provisions an agent-scoped KB under the machine data + dir (or ``--path``), writes a local credential that is never echoed, and + prints a claim token the human runs via ``vouch agents claim``. + """ + if agent_caller and not agent: + raise click.UsageError("--agent-caller requires --agent") + if agent and not agent_caller: + raise click.UsageError("--agent requires --agent-caller") + + if agent: + assert agent_caller is not None + override = None if path == "." else Path(path).resolve() + with _cli_errors(): + result = agent_provision_mod.provision( + agent_caller, + bootstrap=lambda root: _bootstrap_kb(root, template=template), + path=override, + actor=_whoami(), + ) + if as_json: + _emit_json(result.public_dict()) + return + rec = result.record + verb = "Reused" if not result.created_kb else "Initialised" + click.echo(f"{verb} agent KB for {rec.caller!r} at {rec.kb_dir}") + click.echo(f"Credential written to {result.credential_path} (not shown)") + click.echo("Set this in the agent environment (never paste into chat):") + click.echo(f" export VOUCH_AGENT={rec.caller}") + click.echo(f" export VOUCH_KB_PATH={rec.kb_dir}") + click.echo("Claim this agent from a project KB with:") + click.echo(f" vouch agents claim {rec.claim_token}") + return + # _cli_errors so a refused identity mint (corrupt config.yaml) reads as # a one-line error, not a traceback. with _cli_errors(): store, seed, template_result = _bootstrap_kb( Path(path).resolve(), template=template ) + if as_json: + _emit_json({ + "kb_dir": str(store.kb_dir), + "root": str(store.root), + "starter_created": seed.created_anything, + "starter_claim_id": seed.claim_id, + "template": ( + None if template_result is None else template_result.template + ), + }) + return click.echo(f"Initialised KB at {store.kb_dir}") if seed.created_anything: click.echo(f"Seeded starter claim: {seed.claim_id}") @@ -615,6 +696,65 @@ def adopt( click.echo("review the pending ones with `vouch review`.") +@cli.group(name="agents") +def agents_group() -> None: + """Agent-native provisioning and claim handshake.""" + + +@agents_group.command("claim") +@click.argument("token") +@click.option( + "--retire", + is_flag=True, + help="Archive adopted claims in the agent KB after they land durable here.", +) +@click.option("--dry-run", is_flag=True, help="Report what would move; write nothing.") +@click.option("--json", "as_json", is_flag=True, help="Emit the claim report as JSON.") +def agents_claim( + token: str, retire: bool, dry_run: bool, as_json: bool +) -> None: + """Bind an agent (from ``vouch init --agent``) to this project KB. + + Ownership transfers to the project; the agent's local credential and its + own KB artifacts stay put. Knowledge moves through the same review gate + as ``vouch adopt``. + """ + store = _load_store() + with _cli_errors(): + result = agent_provision_mod.claim( + token, + store, + actor=_whoami(), + dry_run=dry_run, + retire=retire, + ) + if as_json: + _emit_json(result.public_dict()) + return + rec = result.record + report = result.adopt + if result.already_claimed: + click.echo( + f"agent {rec.caller!r} already claimed by this project " + f"({rec.claimed_project_root})" + ) + elif dry_run: + click.echo(f"would claim agent {rec.caller!r} into {store.root}:") + else: + click.echo(f"claimed agent {rec.caller!r} into {store.root}:") + click.echo(f" sources copied: {len(report.sources)}") + click.echo(f" claims durable: {len(report.claims_durable)}") + click.echo(f" claims pending: {len(report.claims_pending)}") + click.echo(f" claims skipped: {len(report.claims_skipped)} (already here)") + if retire: + click.echo( + f" retired (agent): {len(report.retired)} " + "(only claims that landed durable here)" + ) + if report.claims_pending and not dry_run: + click.echo("review the pending ones with `vouch review`.") + + @cli.command() def capabilities() -> None: """Emit the JSON capabilities descriptor (mirrors kb.capabilities).""" diff --git a/src/vouch/jsonl_server.py b/src/vouch/jsonl_server.py index 6e45e8a3..98400761 100644 --- a/src/vouch/jsonl_server.py +++ b/src/vouch/jsonl_server.py @@ -108,10 +108,22 @@ def _agent() -> str: # or one token could propose as one actor and approve as another to defeat # the self-approval gate. Only fall back to the header/env when the request # is unauthenticated (tokenless loopback/dev), which is trusted by design. + # Agent-provisioned KBs stamp `agent.caller` into config (#606). subject = trust_mod.current().auth_subject if subject is not None: return f"token:{subject}" - return _actor.get() or os.environ.get("VOUCH_AGENT", "unknown-agent") + env = _actor.get() or os.environ.get("VOUCH_AGENT") + if env: + return env + try: + from . import agent_provision as agent_provision_mod + + stamped = agent_provision_mod.caller_from_store(_store()) + if stamped: + return stamped + except Exception: + pass + return "unknown-agent" # --- per-method handlers --------------------------------------------------- diff --git a/src/vouch/server.py b/src/vouch/server.py index 000fa632..5969ef89 100644 --- a/src/vouch/server.py +++ b/src/vouch/server.py @@ -92,11 +92,24 @@ def _agent() -> str: # An authenticated bearer subject (set by the /mcp transport) is the # principal's real identity and must be what proposals/audit attribute to, # so a token cannot be spoofed and distinct tokens are distinct actors. - # VOUCH_AGENT is only the tokenless (stdio/dev) fallback. + # VOUCH_AGENT is only the tokenless (stdio/dev) fallback. Agent-provisioned + # KBs stamp `agent.caller` into config so the identity survives when the + # host forgot to export VOUCH_AGENT (#606). subject = trust_mod.current().auth_subject if subject is not None: return f"token:{subject}" - return os.environ.get("VOUCH_AGENT", "unknown-agent") + env = os.environ.get("VOUCH_AGENT") + if env: + return env + try: + from . import agent_provision as agent_provision_mod + + stamped = agent_provision_mod.caller_from_store(_store()) + if stamped: + return stamped + except Exception: + pass + return "unknown-agent" # === capabilities / status ================================================ diff --git a/tests/test_agent_provision.py b/tests/test_agent_provision.py new file mode 100644 index 00000000..4d770ee5 --- /dev/null +++ b/tests/test_agent_provision.py @@ -0,0 +1,255 @@ +"""Agent-native provisioning — issue #606.""" + +from __future__ import annotations + +import json +import stat +from pathlib import Path + +import pytest +import yaml +from click.testing import CliRunner + +from vouch import agent_provision as ap +from vouch import audit, proposals +from vouch.cli import cli +from vouch.storage import KBStore + + +@pytest.fixture(autouse=True) +def _isolated_machine(tmp_path_factory, monkeypatch): + """Fake HOME / XDG so tests never touch the real machine.""" + fake_home = tmp_path_factory.mktemp("home") + monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home)) + monkeypatch.setenv("XDG_CONFIG_HOME", str(fake_home / "config")) + monkeypatch.setenv("XDG_DATA_HOME", str(fake_home / "data")) + monkeypatch.delenv("VOUCH_KB_PATH", raising=False) + monkeypatch.delenv("VOUCH_PROJECT_DIR", raising=False) + monkeypatch.delenv("VOUCH_AGENT", raising=False) + monkeypatch.delenv(ap.CREDS_ENV, raising=False) + monkeypatch.delenv(ap.AGENTS_DATA_ENV, raising=False) + return fake_home + + +def _bootstrap(root: Path): + from vouch.cli import _bootstrap_kb + + return _bootstrap_kb(root) + + +def test_validate_caller_rejects_bad_names() -> None: + with pytest.raises(ap.AgentProvisionError): + ap.validate_caller("") + with pytest.raises(ap.AgentProvisionError): + ap.validate_caller("../evil") + with pytest.raises(ap.AgentProvisionError): + ap.validate_caller("has space") + assert ap.validate_caller("ci-bot_1.0") == "ci-bot_1.0" + + +def test_provision_writes_credential_never_in_public_dict() -> None: + result = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + assert result.created_kb is True + assert result.record.caller == "ci-bot" + assert (Path(result.record.kb_root) / ".vouch" / "config.yaml").is_file() + + public = result.public_dict() + assert "credential" not in public + assert public["claim_token"] == result.record.claim_token + assert public["claim_command"] == f"vouch agents claim {result.record.claim_token}" + assert public["credential_path"] == str(result.credential_path) + assert result.record.credential not in json.dumps(public) + + mode = result.credential_path.stat().st_mode + assert stat.S_IMODE(mode) == 0o600 + + raw = result.credential_path.read_text(encoding="utf-8") + assert result.record.credential in raw + assert result.record.claim_token in raw + + cfg = yaml.safe_load( + (Path(result.record.kb_root) / ".vouch" / "config.yaml").read_text(encoding="utf-8") + ) + assert cfg["agent"]["caller"] == "ci-bot" + assert cfg["agent"]["unclaimed"] is True + + +def test_provision_is_idempotent_before_claim() -> None: + first = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + second = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + assert second.created_kb is False + assert second.record.claim_token == first.record.claim_token + assert second.record.credential == first.record.credential + + +def test_cli_init_agent_json_omits_credential() -> None: + runner = CliRunner() + result = runner.invoke( + cli, + ["init", "--agent", "--agent-caller", "openclaw", "--json"], + ) + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["caller"] == "openclaw" + assert "credential" not in data + assert data["claim_command"].startswith("vouch agents claim ") + # Human-readable path must not leak the secret either. + assert "export VOUCH_" not in result.output or "credential" not in result.output.lower() + creds = Path(data["credential_path"]).read_text(encoding="utf-8") + # The secret lives only on disk. + stored = yaml.safe_load(creds)["agents"]["openclaw"]["credential"] + assert stored not in result.output + + +def test_cli_init_agent_requires_caller() -> None: + runner = CliRunner() + result = runner.invoke(cli, ["init", "--agent"]) + assert result.exit_code != 0 + assert "--agent-caller" in result.output + + +def test_cli_init_agent_caller_requires_agent_flag() -> None: + runner = CliRunner() + result = runner.invoke(cli, ["init", "--agent-caller", "x"]) + assert result.exit_code != 0 + assert "--agent" in result.output + + +def test_claim_adopts_through_the_gate_and_leaves_agent_kb( + tmp_path: Path, +) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + agent = KBStore(Path(provisioned.record.kb_root)) + + # Seed the agent KB with a receipt-backed claim the way capture would. + body = ( + b"The deploy cadence for this service is every second Tuesday.\n" + b"Rollbacks use the blue-green switch.\n" + ) + src = agent.put_source(body, title="ops-note", source_type="file") + quote = "The deploy cadence for this service is every second Tuesday." + filed = proposals.propose_quoted_claim( + agent, + text=quote, + source_id=src.id, + quote=quote, + proposed_by="ci-bot", + ) + assert filed is not None + durable = proposals.resolve_pending_receipt_claim( + agent, filed.proposal, actor="ci-bot", reason="self-approve under trusted-agent" + ) + assert durable is not None + + project_root = tmp_path / "proj" + project_root.mkdir() + project = KBStore.init(project_root) + + claimed = ap.claim(provisioned.record.claim_token, project, actor="human") + assert claimed.already_claimed is False + assert claimed.record.claimed_project_kb_id == project.identity()[0] + assert durable.id in claimed.adopt.claims_durable + # Agent KB still has the claim — artifacts untouched. + assert agent.get_claim(durable.id).text == quote + # Project received it through the gate. + assert project.get_claim(durable.id).text == quote + assert "adopted" in project.get_claim(durable.id).tags + + cfg = yaml.safe_load(agent.config_path.read_text(encoding="utf-8")) + assert cfg["agent"]["unclaimed"] is False + + events = [e for e in audit.read_events(project.kb_dir) if e.event == "agent.claim"] + assert events and events[0].data["caller"] == "ci-bot" + + +def test_claim_is_idempotent_on_same_project(tmp_path: Path) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + project = KBStore.init(tmp_path / "proj") + first = ap.claim(provisioned.record.claim_token, project, actor="human") + second = ap.claim(provisioned.record.claim_token, project, actor="human") + assert first.already_claimed is False + assert second.already_claimed is True + assert second.record.claimed_project_kb_id == project.identity()[0] + + +def test_claim_rejects_other_project(tmp_path: Path) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + a = KBStore.init(tmp_path / "a") + b = KBStore.init(tmp_path / "b") + ap.claim(provisioned.record.claim_token, a, actor="human") + with pytest.raises(ap.AgentProvisionError, match="already claimed"): + ap.claim(provisioned.record.claim_token, b, actor="human") + + +def test_claim_unknown_token(tmp_path: Path) -> None: + project = KBStore.init(tmp_path / "proj") + with pytest.raises(ap.AgentProvisionError, match="unknown claim token"): + ap.claim("not-a-real-token", project, actor="human") + + +def test_cli_agents_claim_end_to_end(tmp_path: Path, monkeypatch) -> None: + runner = CliRunner() + init = runner.invoke(cli, ["init", "--agent", "--agent-caller", "hermes", "--json"]) + assert init.exit_code == 0, init.output + data = json.loads(init.output) + token = data["claim_token"] + + # Put a claim in the agent KB via the library (CLI propose needs more setup). + agent = KBStore(Path(data["kb_root"])) + body = b"Hermes remembers that the staging refresh is nightly at 02:00 UTC.\n" + src = agent.put_source(body, title="note", source_type="file") + quote = "the staging refresh is nightly at 02:00 UTC" + assert quote.encode() in body + filed = proposals.propose_quoted_claim( + agent, + text=f"Remember: {quote}.", + source_id=src.id, + quote=quote, + proposed_by="hermes", + ) + assert filed is not None + proposals.resolve_pending_receipt_claim( + agent, filed.proposal, actor="hermes", reason="trusted-agent" + ) + + project = tmp_path / "app" + project.mkdir() + monkeypatch.chdir(project) + runner.invoke(cli, ["init"]) + claim = runner.invoke(cli, ["agents", "claim", token, "--json"]) + assert claim.exit_code == 0, claim.output + report = json.loads(claim.output) + assert report["caller"] == "hermes" + assert report["already_claimed"] is False + assert report["adopt"]["claims_durable"] or report["adopt"]["claims_pending"] + + +def test_caller_from_store_used_when_vouch_agent_unset( + monkeypatch, +) -> None: + result = ap.provision("stamped-bot", bootstrap=_bootstrap, actor="human") + monkeypatch.delenv("VOUCH_AGENT", raising=False) + monkeypatch.setenv("VOUCH_KB_PATH", str(result.record.kb_dir)) + # discover_root via VOUCH_KB_PATH + from vouch.cli import _whoami + + assert _whoami() == "stamped-bot" + + +def test_server_agent_falls_back_to_stamped_caller(monkeypatch) -> None: + result = ap.provision("mcp-bot", bootstrap=_bootstrap, actor="human") + monkeypatch.delenv("VOUCH_AGENT", raising=False) + monkeypatch.setenv("VOUCH_KB_PATH", str(result.record.kb_dir)) + from vouch import server + + # Reset any cached store + monkeypatch.setattr(server, "_store", lambda: KBStore(Path(result.record.kb_root))) + assert server._agent() == "mcp-bot" + + +def test_public_dict_never_contains_credential_after_claim(tmp_path: Path) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + project = KBStore.init(tmp_path / "proj") + claimed = ap.claim(provisioned.record.claim_token, project, actor="human") + blob = json.dumps(claimed.public_dict()) + assert provisioned.record.credential not in blob