diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ed8832f..050ff1fa 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` plus a claim handshake** + (#606): an agent can provision its own scoped KB and persistent proposer + identity (`agent.caller`) without a human first creating a project KB. + `vouch init --agent --agent-caller ` writes a local credential + (chmod 0600, never echoed to stdout) under + `$XDG_CONFIG_HOME/vouch/agent-credentials.yaml`, stamps the agent KB under + `$XDG_DATA_HOME/vouch/agents//`, and emits a claim token. The human + binds that agent to a project with `vouch agents claim `, which + transfers ownership and adopts knowledge through the same review gate as + `vouch adopt` (via additive `adopt_kb`, leaving the personal-fallback + `adopt` path byte-stable). MCP/JSONL/CLI fall back to the stamped + `agent.caller` when `VOUCH_AGENT` is unset. Claiming also registers the + agent in the project's `#607` registry when possible. - **correction capture — the pushback becomes a proposal** (#430): the adapter captured tool *outcomes* passively but never the single highest-signal event in a session, the user correcting the agent ("no, we deploy from `main` not diff --git a/src/vouch/adopt.py b/src/vouch/adopt.py index 6fa4b3ac..b32309ba 100644 --- a/src/vouch/adopt.py +++ b/src/vouch/adopt.py @@ -308,6 +308,166 @@ def adopt( return report +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 (issue #606 claim), + not a personal catch-all that stamped ``metadata.origin_path``. + + Kept as a sibling of ``adopt`` rather than a rewrite of it: the personal + fallback path stays byte-stable; this path is the ownership-transfer + surface and carries its own tests in ``tests/test_adopt.py``. + """ + label = origin_label or str(source_kb.root) + source_identity = source_kb.identity() + project_identity = project.identity() + report = AdoptReport( + origin=label, + from_kb=source_identity[0] if source_identity else None, + to_kb=project_identity[0] if project_identity else None, + dry_run=dry_run, + ) + sources = list(source_kb.list_sources()) + if not sources: + return report + source_ids = {s.id for s in sources} + pairs = _claims_citing(source_kb, source_ids) + rationale = ( + f"adopted from agent KB {report.from_kb or '(no id)'} ({label})" + ) + + if dry_run: + report.sources = sorted( + sid for sid in source_ids if not _source_exists(project, sid) + ) + queued = _pending_payload_ids(project) + gate_open = _receipts_auto_approve(project) + for claim, receipt in pairs: + if _already_durable(project, claim) or claim.id in queued: + report.claims_skipped.append(claim.id) + elif receipt is not None and gate_open: + report.claims_durable.append(claim.id) + else: + report.claims_pending.append(claim.id) + return report + + for src in sources: + if _source_exists(project, src.id): + continue + content = source_kb.read_source_content(src.id) + project.put_source( + content, + title=src.title, + source_type=str(src.type), + media_type=src.media_type, + tags=_with_tag(src.tags, "adopted"), + metadata={ + **src.metadata, + "adopted_from": report.from_kb, + }, + scope=proposals_mod.default_scope(project), + ) + report.sources.append(src.id) + + queued = _pending_payload_ids(project) + landed_durable: list[str] = [] + for claim, receipt in pairs: + if _already_durable(project, claim) or claim.id in queued: + report.claims_skipped.append(claim.id) + continue + if receipt is not None and receipt.quote: + result = proposals_mod.propose_quoted_claim( + project, + text=claim.text, + source_id=receipt.source_id, + quote=receipt.quote, + proposed_by=actor, + claim_type=str(claim.type), + confidence=claim.confidence, + tags=_with_tag(claim.tags, "adopted"), + rationale=rationale, + slug_hint=claim.id, + ) + if result is None: + report.claims_skipped.append(claim.id) + continue + durable = proposals_mod.resolve_pending_receipt_claim( + project, + result.proposal, + actor=actor, + reason="adopted from agent KB (receipt re-verified)", + ) + if durable is not None: + report.claims_durable.append(durable.id) + landed_durable.append(claim.id) + else: + try: + filed = project.get_proposal(result.proposal.id) + except ArtifactNotFoundError: + filed = None + if filed is not None and filed.status == ProposalStatus.PENDING: + report.claims_pending.append(claim.id) + else: + report.claims_skipped.append(claim.id) + else: + evidence = [eid for eid in claim.evidence if eid in source_ids] + if not evidence: + report.claims_skipped.append(claim.id) + continue + proposals_mod.propose_claim( + project, + text=claim.text, + evidence=evidence, + proposed_by=actor, + claim_type=str(claim.type), + confidence=claim.confidence, + tags=_with_tag(claim.tags, "adopted"), + rationale=rationale, + slug_hint=claim.id, + ) + report.claims_pending.append(claim.id) + + if retire: + for claim_id in landed_durable: + try: + lifecycle.archive(source_kb, claim_id=claim_id, actor=actor) + except Exception: + continue + report.retired.append(claim_id) + + moved = bool(report.sources or report.claims_durable or report.claims_pending) + if moved: + data = { + "origin": report.origin, + "sources": len(report.sources), + "claims_durable": len(report.claims_durable), + "claims_pending": len(report.claims_pending), + "retired": len(report.retired), + } + audit_mod.log_event( + project.kb_dir, + event="kb.adopt", + actor=actor, + data={**data, "direction": "in", "from_kb": report.from_kb}, + ) + audit_mod.log_event( + source_kb.kb_dir, + event="kb.adopt", + actor=actor, + data={**data, "direction": "out", "to_kb": report.to_kb}, + ) + return report + + def _already_durable(project: KBStore, claim: Claim) -> bool: try: project.get_claim(claim.id) 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/capture.py b/src/vouch/capture.py index 9aeb98b8..ec99b654 100644 --- a/src/vouch/capture.py +++ b/src/vouch/capture.py @@ -75,9 +75,15 @@ def load_config(store: KBStore) -> CaptureConfig: return CaptureConfig( enabled=coerce_bool(raw.get("enabled", DEFAULT_ENABLED), DEFAULT_ENABLED), realtime=coerce_bool(raw.get("realtime", DEFAULT_REALTIME), DEFAULT_REALTIME), - min_observations=int(raw.get("min_observations", DEFAULT_MIN_OBSERVATIONS)), - dedup_window_seconds=float( - raw.get("dedup_window_seconds", DEFAULT_DEDUP_WINDOW_SECONDS) + min_observations=coerce_numeric( + raw.get("min_observations", DEFAULT_MIN_OBSERVATIONS), + DEFAULT_MIN_OBSERVATIONS, + int, + ), + dedup_window_seconds=coerce_numeric( + raw.get("dedup_window_seconds", DEFAULT_DEDUP_WINDOW_SECONDS), + DEFAULT_DEDUP_WINDOW_SECONDS, + float, ), answer_mode=answer_mode, ) diff --git a/src/vouch/cli.py b/src/vouch/cli.py index 9dba52bd..d69ba6c9 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 agents as agents_mod from . import audit as audit_mod from . import capture as capture_mod @@ -121,6 +122,7 @@ def _cli_errors() -> Iterator[None]: chatgpt_import_mod.ChatGPTImportError, codex_rollout_mod.CodexRolloutError, agents_mod.AgentError, + agent_provision_mod.AgentProvisionError, pins_mod.PinError, ) as e: raise click.ClickException(str(e)) from e @@ -162,7 +164,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: @@ -267,14 +279,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}") @@ -3996,6 +4077,62 @@ def agents_revoke(name: str) -> None: _transition(name, agents_mod.AgentStatus.REVOKED, "revoked") +@agents_group.command("claim") +@click.argument("token") +@click.option( + "--dry-run", + is_flag=True, + help="Preview what would move; write nothing and leave the claim token usable.", +) +@click.option( + "--retire", + is_flag=True, + help="Archive agent-KB copies of claims that landed durable in the project.", +) +@click.option("--json", "as_json", is_flag=True, help="Emit the claim result as JSON.") +def agents_claim(token: str, dry_run: bool, retire: bool, as_json: bool) -> None: + """Bind a provisioned agent to this project KB and adopt its knowledge. + + Run from the project after ``vouch init --agent`` printed a claim token. + Ownership transfer goes through the same review gate as ``vouch adopt``; + the agent's local credential and agent-KB artifacts stay untouched. + """ + store = _load_store() + with _cli_errors(): + result = agent_provision_mod.claim( + token, + store, + actor=_whoami(), + dry_run=dry_run, + retire=retire, + ) + # Mirror into the project agent registry (#607) so claim shows up in + # `vouch agents list`. Best-effort: a prior manual register must not + # block the ownership transfer that already succeeded. + if not dry_run and not result.already_claimed: + with contextlib.suppress(agents_mod.AgentError): + agents_mod.register( + store, + subject=trust_mod.auth_subject_for_token(result.record.credential), + name=result.record.caller, + actor=_whoami(), + note="claimed via vouch agents claim", + ) + if as_json: + _emit_json(result.public_dict()) + return + verb = "Already bound" if result.already_claimed else "Claimed" + if dry_run: + verb = "Would claim" + click.echo( + f"{verb} agent {result.record.caller!r} → {store.root} " + f"(sources={len(result.adopt.sources)}, " + f"durable={len(result.adopt.claims_durable)}, " + f"pending={len(result.adopt.claims_pending)}, " + f"skipped={len(result.adopt.claims_skipped)})" + ) + + @cli.command("pin") @click.argument("artifact_id") @click.option("--local", is_flag=True, diff --git a/src/vouch/jsonl_server.py b/src/vouch/jsonl_server.py index 38a34cb8..e40f90f7 100644 --- a/src/vouch/jsonl_server.py +++ b/src/vouch/jsonl_server.py @@ -17,6 +17,7 @@ from __future__ import annotations +import contextlib import json import logging import os @@ -114,7 +115,21 @@ def _agent() -> str: 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") + header = _actor.get() + if header: + return header + env = os.environ.get("VOUCH_AGENT") + if env: + return env + # Agent-provisioned KBs stamp `agent.caller` so attribution survives when + # the host forgot to export VOUCH_AGENT (issue #606). + from . import agent_provision as agent_provision_mod + + with contextlib.suppress(Exception): + stamped = agent_provision_mod.caller_from_store(_store()) + if stamped: + return stamped + return "unknown-agent" # --- per-method handlers --------------------------------------------------- diff --git a/src/vouch/server.py b/src/vouch/server.py index 5bcf4932..c545c800 100644 --- a/src/vouch/server.py +++ b/src/vouch/server.py @@ -12,6 +12,7 @@ from __future__ import annotations import asyncio +import contextlib import os from pathlib import Path from typing import Any @@ -100,7 +101,18 @@ def _agent() -> str: 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 + # Agent-provisioned KBs stamp `agent.caller` so attribution survives when + # the host forgot to export VOUCH_AGENT (issue #606). + from . import agent_provision as agent_provision_mod + + with contextlib.suppress(Exception): + stamped = agent_provision_mod.caller_from_store(_store()) + if stamped: + return stamped + return "unknown-agent" # === capabilities / status ================================================ diff --git a/tests/test_adopt.py b/tests/test_adopt.py index 8a03daa9..587643ee 100644 --- a/tests/test_adopt.py +++ b/tests/test_adopt.py @@ -517,3 +517,210 @@ def test_global_install_survives_a_failing_personal_kb( assert r.exit_code == 0, r.output assert "could not set up the personal KB" in r.output assert (fake_home / ".claude" / "settings.json").is_file() + + +# --- adopt_kb (agent claim ownership transfer, issue #606) ----------------- + + +def _seed_agent_receipt_claim(agent: KBStore) -> tuple[str, str]: + """Return (source_id, claim_id) for a receipt-backed durable claim.""" + from vouch import proposals + + 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="agent", + ) + assert filed is not None + durable = proposals.resolve_pending_receipt_claim( + agent, filed.proposal, actor="agent", reason="trusted-agent" + ) + assert durable is not None + return src.id, durable.id + + +def test_adopt_kb_empty_source_returns_quietly(tmp_path: Path) -> None: + source = KBStore.init(tmp_path / "agent") + project = KBStore.init(tmp_path / "proj") + # Drop starter sources so the agent KB is empty of adoptable content. + for src in list(source.list_sources()): + (source.kb_dir / "sources" / f"{src.id}.yaml").unlink(missing_ok=True) + blob = source.kb_dir / "blobs" / src.id + if blob.exists(): + blob.unlink() + # list_sources may still see starters via index — use a brand-new empty dir + empty = KBStore.init(tmp_path / "empty-agent") + # Wipe everything under sources/ + sources_dir = empty.kb_dir / "sources" + if sources_dir.is_dir(): + for p in sources_dir.iterdir(): + p.unlink() + report = adopt_mod.adopt_kb(project, empty, origin_label="agent:empty") + assert report.sources == [] + assert report.claims_durable == [] + assert report.origin == "agent:empty" + + +def test_adopt_kb_moves_receipt_claims_through_the_gate(tmp_path: Path) -> None: + agent = KBStore.init(tmp_path / "agent") + _src_id, claim_id = _seed_agent_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + + report = adopt_mod.adopt_kb(project, agent, origin_label="agent:ci") + assert claim_id in report.claims_durable + assert report.sources + assert project.get_claim(claim_id).text + assert "adopted" in project.get_claim(claim_id).tags + proj_events = [e for e in audit.read_events(project.kb_dir) if e.event == "kb.adopt"] + assert proj_events and proj_events[0].data["from_kb"] == agent.identity()[0] + + +def test_adopt_kb_dry_run_and_closed_gate(tmp_path: Path) -> None: + agent = KBStore.init(tmp_path / "agent") + _src_id, claim_id = _seed_agent_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + before = {c.id for c in project.list_claims()} + + preview = adopt_mod.adopt_kb(project, agent, dry_run=True) + assert claim_id in preview.claims_durable + assert {c.id for c in project.list_claims()} == before + + cfg = yaml.safe_load(project.config_path.read_text(encoding="utf-8")) + cfg.setdefault("review", {})["auto_approve_on_receipt"] = False + cfg["review"].pop("approver_role", None) + project.config_path.write_text(yaml.safe_dump(cfg), encoding="utf-8") + + closed_preview = adopt_mod.adopt_kb(project, agent, dry_run=True) + assert claim_id in closed_preview.claims_pending + + report = adopt_mod.adopt_kb(project, agent) + assert claim_id in report.claims_pending + assert claim_id not in report.claims_durable + + +def test_adopt_kb_is_idempotent_and_skips_queued(tmp_path: Path) -> None: + agent = KBStore.init(tmp_path / "agent") + _src_id, claim_id = _seed_agent_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + first = adopt_mod.adopt_kb(project, agent) + assert claim_id in first.claims_durable + again = adopt_mod.adopt_kb(project, agent) + assert claim_id in again.claims_skipped + assert again.sources == [] + + +def test_adopt_kb_retire_archives_source_copies(tmp_path: Path) -> None: + agent = KBStore.init(tmp_path / "agent") + _src_id, claim_id = _seed_agent_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + report = adopt_mod.adopt_kb(project, agent, retire=True) + assert claim_id in report.retired + assert agent.get_claim(claim_id).status == ClaimStatus.ARCHIVED + assert project.get_claim(claim_id).status == ClaimStatus.WORKING + + +def test_adopt_kb_evidence_only_lands_pending(tmp_path: Path) -> None: + """Bare source-id evidence (no receipt) always files PENDING.""" + from vouch.models import Claim + + agent = KBStore.init(tmp_path / "agent") + body = b"Bare evidence claim about nightly refresh at 02:00 UTC.\n" + src = agent.put_source(body, title="note", source_type="file") + claim = agent.put_claim( + Claim(id="bare-evidence", text="nightly refresh at 02:00 UTC", evidence=[src.id]) + ) + project = KBStore.init(tmp_path / "proj") + report = adopt_mod.adopt_kb(project, agent) + assert claim.id in report.claims_pending + pending = project.list_proposals(ProposalStatus.PENDING) + assert any(p.payload.get("id") == claim.id for p in pending) + + +def test_adopt_kb_skips_receiptless_evidence_object_citations(tmp_path: Path) -> None: + """Evidence rows without a quote cite the source but cannot re-propose.""" + from vouch.models import Claim, Evidence + + agent = KBStore.init(tmp_path / "agent") + body = b"A source that is only cited via an evidence object, no quote.\n" + src = agent.put_source(body, title="note", source_type="file") + ev = agent.put_evidence( + Evidence(id="ev-no-quote", source_id=src.id, locator="L1", quote=None) + ) + claim = agent.put_claim( + Claim( + id="via-ev", + text="cited only through evidence object", + evidence=[ev.id], + ) + ) + project = KBStore.init(tmp_path / "proj") + report = adopt_mod.adopt_kb(project, agent) + assert claim.id in report.claims_skipped + + +def test_adopt_kb_retire_continues_when_archive_fails( + tmp_path: Path, monkeypatch +) -> None: + from vouch import lifecycle + + agent = KBStore.init(tmp_path / "agent") + _src_id, claim_id = _seed_agent_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + + def boom(*_a, **_k): + raise RuntimeError("archive denied") + + monkeypatch.setattr(lifecycle, "archive", boom) + report = adopt_mod.adopt_kb(project, agent, retire=True) + assert claim_id in report.claims_durable + assert report.retired == [] + + +def test_adopt_kb_skips_when_propose_returns_none(tmp_path: Path, monkeypatch) -> None: + from vouch import proposals + + agent = KBStore.init(tmp_path / "agent") + _src_id, claim_id = _seed_agent_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + monkeypatch.setattr(proposals, "propose_quoted_claim", lambda *a, **k: None) + report = adopt_mod.adopt_kb(project, agent) + assert claim_id in report.claims_skipped + + +def test_adopt_kb_skips_when_proposal_vanishes_after_resolve( + tmp_path: Path, monkeypatch +) -> None: + from vouch import proposals + from vouch.storage import ArtifactNotFoundError + + agent = KBStore.init(tmp_path / "agent") + _src_id, claim_id = _seed_agent_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + + class _Fake: + id = "vanished-proposal" + + class _Result: + proposal = _Fake() + + monkeypatch.setattr( + proposals, "propose_quoted_claim", lambda *a, **k: _Result() + ) + monkeypatch.setattr( + proposals, "resolve_pending_receipt_claim", lambda *a, **k: None + ) + + def _missing(_self, _pid): + raise ArtifactNotFoundError("gone") + + monkeypatch.setattr(KBStore, "get_proposal", _missing) + report = adopt_mod.adopt_kb(project, agent) + assert claim_id in report.claims_skipped diff --git a/tests/test_agent_provision.py b/tests/test_agent_provision.py new file mode 100644 index 00000000..59d82a5f --- /dev/null +++ b/tests/test_agent_provision.py @@ -0,0 +1,549 @@ +"""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 agents, audit, proposals, trust +from vouch.cli import cli +from vouch.models import Claim +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("VOUCH_USER", 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 _seed_receipt_claim(store: KBStore, *, caller: str = "ci-bot") -> Claim: + body = ( + b"The deploy cadence for this service is every second Tuesday.\n" + b"Rollbacks use the blue-green switch.\n" + ) + src = store.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( + store, + text=quote, + source_id=src.id, + quote=quote, + proposed_by=caller, + ) + assert filed is not None + durable = proposals.resolve_pending_receipt_claim( + store, filed.proposal, actor=caller, reason="trusted-agent" + ) + assert durable is not None + return durable + + +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_credentials_and_data_path_overrides(tmp_path: Path, monkeypatch) -> None: + creds = tmp_path / "creds.yaml" + data = tmp_path / "agents-data" + monkeypatch.setenv(ap.CREDS_ENV, str(creds)) + monkeypatch.setenv(ap.AGENTS_DATA_ENV, str(data)) + assert ap.credentials_path() == creds + assert ap.agents_data_root() == data + + +def test_agents_data_root_falls_back_to_home(monkeypatch) -> None: + monkeypatch.delenv("XDG_DATA_HOME", raising=False) + monkeypatch.delenv(ap.AGENTS_DATA_ENV, raising=False) + root = ap.agents_data_root() + assert root.as_posix().endswith(".local/share/vouch/agents") + + +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_provision_rejects_missing_kb_on_rerun() -> None: + first = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + import shutil + + shutil.rmtree(Path(first.record.kb_root) / ".vouch") + with pytest.raises(ap.AgentProvisionError, match="KB is missing"): + ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + + +def test_provision_rejects_already_claimed(tmp_path: Path) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + project = KBStore.init(tmp_path / "proj") + ap.claim(provisioned.record.claim_token, project, actor="human") + with pytest.raises(ap.AgentProvisionError, match="already claimed"): + ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + + +def test_provision_with_explicit_path(tmp_path: Path) -> None: + root = tmp_path / "custom-agent" + result = ap.provision( + "path-bot", bootstrap=_bootstrap, path=root, actor="human" + ) + assert Path(result.record.kb_root) == root.resolve() + assert ap.agent_kb_root("path-bot", path=root) == root.resolve() + + +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 ") + assert "export VOUCH_" not in result.output or "credential" not in result.output.lower() + creds = Path(data["credential_path"]).read_text(encoding="utf-8") + stored = yaml.safe_load(creds)["agents"]["openclaw"]["credential"] + assert stored not in result.output + + +def test_cli_init_agent_human_text() -> None: + runner = CliRunner() + result = runner.invoke(cli, ["init", "--agent", "--agent-caller", "text-bot"]) + assert result.exit_code == 0, result.output + assert "Initialised agent KB" in result.output or "Reused agent KB" in result.output + assert "Credential written" in result.output + assert "vouch agents claim " in result.output + assert "export VOUCH_AGENT=text-bot" in result.output + + +def test_cli_init_human_json(tmp_path: Path, monkeypatch) -> None: + project = tmp_path / "proj" + project.mkdir() + monkeypatch.chdir(project) + runner = CliRunner() + result = runner.invoke(cli, ["init", "--json"]) + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert "kb_dir" in data + assert data["starter_created"] is True + + +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)) + durable = _seed_receipt_claim(agent) + + 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 + assert agent.get_claim(durable.id).text == durable.text + assert project.get_claim(durable.id).text == durable.text + 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_dry_run_writes_nothing(tmp_path: Path) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + agent = KBStore(Path(provisioned.record.kb_root)) + _seed_receipt_claim(agent) + project = KBStore.init(tmp_path / "proj") + before = ap.find_by_caller("ci-bot") + assert before is not None and before.claimed_at is None + result = ap.claim( + provisioned.record.claim_token, project, actor="human", dry_run=True + ) + assert result.already_claimed is False + assert result.adopt.dry_run is True + assert result.adopt.claims_durable + after = ap.find_by_caller("ci-bot") + assert after is not None and after.claimed_at is None + + +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_claim_empty_token(tmp_path: Path) -> None: + project = KBStore.init(tmp_path / "proj") + with pytest.raises(ap.AgentProvisionError, match="empty"): + ap.claim(" ", project, actor="human") + + +def test_claim_refuses_agent_own_kb() -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + agent = KBStore(Path(provisioned.record.kb_root)) + with pytest.raises(ap.AgentProvisionError, match="project KB"): + ap.claim(provisioned.record.claim_token, agent, actor="human") + + +def test_claim_refuses_missing_agent_kb(tmp_path: Path) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + import shutil + + shutil.rmtree(Path(provisioned.record.kb_root) / ".vouch") + project = KBStore.init(tmp_path / "proj") + with pytest.raises(ap.AgentProvisionError, match="missing"): + ap.claim(provisioned.record.claim_token, project, actor="human") + + +def test_claim_mints_project_identity_when_absent(tmp_path: Path) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + project = KBStore.init(tmp_path / "proj") + cfg = yaml.safe_load(project.config_path.read_text(encoding="utf-8")) + cfg.pop("kb", None) + project.config_path.write_text(yaml.safe_dump(cfg), encoding="utf-8") + assert project.identity() is None + claimed = ap.claim(provisioned.record.claim_token, project, actor="human") + assert claimed.record.claimed_project_kb_id == project.identity()[0] + + +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"] + + agent = KBStore(Path(data["kb_root"])) + _seed_receipt_claim(agent, caller="hermes") + + 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"] + + store = KBStore(project) + registered = agents.find(store, "hermes") + assert registered is not None + assert registered.subject == trust.auth_subject_for_token( + yaml.safe_load(Path(data["credential_path"]).read_text(encoding="utf-8")) + ["agents"]["hermes"]["credential"] + ) + + +def test_cli_agents_claim_human_text(tmp_path: Path, monkeypatch) -> None: + runner = CliRunner() + init = runner.invoke(cli, ["init", "--agent", "--agent-caller", "text-claim", "--json"]) + data = json.loads(init.output) + project = tmp_path / "app" + project.mkdir() + monkeypatch.chdir(project) + runner.invoke(cli, ["init"]) + claim = runner.invoke(cli, ["agents", "claim", data["claim_token"]]) + assert claim.exit_code == 0, claim.output + assert "Claimed agent 'text-claim'" in claim.output + + +def test_cli_agents_claim_dry_run_text(tmp_path: Path, monkeypatch) -> None: + runner = CliRunner() + init = runner.invoke(cli, ["init", "--agent", "--agent-caller", "dry", "--json"]) + data = json.loads(init.output) + project = tmp_path / "app" + project.mkdir() + monkeypatch.chdir(project) + runner.invoke(cli, ["init"]) + claim = runner.invoke(cli, ["agents", "claim", data["claim_token"], "--dry-run"]) + assert claim.exit_code == 0, claim.output + assert "Would claim" in claim.output + + +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)) + from vouch.cli import _whoami + + assert _whoami() == "stamped-bot" + + +def test_whoami_prefers_vouch_agent_env(monkeypatch) -> None: + monkeypatch.setenv("VOUCH_AGENT", "env-bot") + from vouch.cli import _whoami + + assert _whoami() == "env-bot" + + +def test_caller_from_store_error_paths(tmp_path: Path) -> None: + store = KBStore.init(tmp_path / "plain") + store.config_path.write_text("[]\n", encoding="utf-8") + assert ap.caller_from_store(store) is None + store.config_path.unlink() + assert ap.caller_from_store(store) is None + + +def test_stamp_agent_config_recovers_non_mapping(tmp_path: Path) -> None: + store = KBStore.init(tmp_path / "plain") + # Truthy non-mapping so `yaml.safe_load(...) or {}` does not short-circuit. + store.config_path.write_text("[1]\n", encoding="utf-8") + ap.stamp_agent_config(store, caller="bot", unclaimed=True) + loaded = yaml.safe_load(store.config_path.read_text(encoding="utf-8")) + assert loaded["agent"]["caller"] == "bot" + + +def test_list_records_rejects_non_dict_agents(monkeypatch) -> None: + monkeypatch.setattr( + ap, "_load_raw", lambda: {"version": 1, "agents": ["not", "a", "map"]} + ) + assert ap.list_records() == [] + + +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) + from vouch import server + + monkeypatch.setattr(server, "_store", lambda: KBStore(Path(result.record.kb_root))) + with trust.trust_context(trust.MCP_STDIO): + assert server._agent() == "mcp-bot" + + +def test_server_agent_env_and_unknown(monkeypatch, tmp_path: Path) -> None: + from vouch import server + + monkeypatch.setenv("VOUCH_AGENT", "env-mcp") + with trust.trust_context(trust.MCP_STDIO): + assert server._agent() == "env-mcp" + + monkeypatch.delenv("VOUCH_AGENT", raising=False) + plain = KBStore.init(tmp_path / "plain-mcp") + monkeypatch.setattr(server, "_store", lambda: plain) + with trust.trust_context(trust.MCP_STDIO): + assert server._agent() == "unknown-agent" + + +def test_jsonl_agent_falls_back_to_stamped_caller(monkeypatch) -> None: + result = ap.provision("jsonl-bot", bootstrap=_bootstrap, actor="human") + monkeypatch.delenv("VOUCH_AGENT", raising=False) + from vouch import jsonl_server + + monkeypatch.setattr( + jsonl_server, "_store", lambda: KBStore(Path(result.record.kb_root)) + ) + token = jsonl_server._actor.set(None) + try: + with trust.trust_context(trust.JSONL_HTTP): + assert jsonl_server._agent() == "jsonl-bot" + finally: + jsonl_server._actor.reset(token) + + +def test_jsonl_agent_header_env_and_unknown(monkeypatch, tmp_path: Path) -> None: + from vouch import jsonl_server + + token = jsonl_server._actor.set("header-bot") + try: + with trust.trust_context(trust.JSONL_HTTP): + assert jsonl_server._agent() == "header-bot" + finally: + jsonl_server._actor.reset(token) + + token = jsonl_server._actor.set(None) + try: + monkeypatch.setenv("VOUCH_AGENT", "env-jsonl") + with trust.trust_context(trust.JSONL_HTTP): + assert jsonl_server._agent() == "env-jsonl" + monkeypatch.delenv("VOUCH_AGENT", raising=False) + plain = KBStore.init(tmp_path / "plain-jsonl") + monkeypatch.setattr(jsonl_server, "_store", lambda: plain) + with trust.trust_context(trust.JSONL_HTTP): + assert jsonl_server._agent() == "unknown-agent" + finally: + jsonl_server._actor.reset(token) + + +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 + + +def test_load_raw_handles_corrupt_and_non_mapping(tmp_path: Path, monkeypatch) -> None: + creds = tmp_path / "creds.yaml" + monkeypatch.setenv(ap.CREDS_ENV, str(creds)) + creds.write_text("{not yaml", encoding="utf-8") + with pytest.raises(ap.AgentProvisionError, match="cannot read"): + ap.list_records() + + creds.write_text("[]\n", encoding="utf-8") + assert ap.list_records() == [] + + creds.write_text("version: 1\nagents: []\n", encoding="utf-8") + assert ap.list_records() == [] + + creds.write_text( + "version: 1\nagents:\n 1: {kb_root: x, credential: y, claim_token: z}\n" + " bad: not-a-map\n" + " incomplete: {kb_root: x}\n", + encoding="utf-8", + ) + # non-string keys skipped; bad/incomplete parse to None + assert ap.list_records() == [] + + +def test_save_raw_cleans_up_temp_on_failure(tmp_path: Path, monkeypatch) -> None: + creds = tmp_path / "creds.yaml" + monkeypatch.setenv(ap.CREDS_ENV, str(creds)) + # Seed a valid provision first so _save_raw is exercised, then break replace. + ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + + def boom(*_a, **_k): + raise OSError("disk full") + + monkeypatch.setattr(ap.os, "replace", boom) + with pytest.raises(OSError, match="disk full"): + ap._save_raw({"version": 1, "agents": {}}) + + +def test_provision_repairs_non_dict_agents_bucket(tmp_path: Path, monkeypatch) -> None: + creds = tmp_path / "creds.yaml" + monkeypatch.setenv(ap.CREDS_ENV, str(creds)) + creds.parent.mkdir(parents=True, exist_ok=True) + # Non-empty list so setdefault returns it and the isinstance repair runs. + creds.write_text("version: 1\nagents: [x]\n", encoding="utf-8") + # _load_raw normalises list → {}; force a truthy non-dict through setdefault. + real_load = ap._load_raw + + def load_with_list_agents(): + data = real_load() + data["agents"] = ["x"] + return data + + monkeypatch.setattr(ap, "_load_raw", load_with_list_agents) + result = ap.provision("repair-bot", bootstrap=_bootstrap, actor="human") + assert result.record.caller == "repair-bot" + + +def test_claim_repairs_non_dict_agents_bucket(tmp_path: Path, monkeypatch) -> None: + provisioned = ap.provision("ci-bot", bootstrap=_bootstrap, actor="human") + # Corrupt agents to a list but keep the record reachable via find_by_claim_token + # by writing the record under a dict first, then... actually find uses list_records + # which returns [] for list agents. So claim unknown. Instead corrupt AFTER find + # by patching _load_raw mid-claim — simpler: rewrite file between find and save + # by making agents a list that still... won't work with find. + + # Corrupt after locating: mutate file before claim's final _save_raw by + # writing agents as list containing nothing useful, then force find via + # monkeypatch. + project = KBStore.init(tmp_path / "proj") + record = provisioned.record + + def load_then_corrupt(): + data = {"version": 1, "agents": []} + return data + + monkeypatch.setattr(ap, "_load_raw", load_then_corrupt) + monkeypatch.setattr(ap, "find_by_claim_token", lambda _t: record) + claimed = ap.claim(record.claim_token, project, actor="human") + assert claimed.record.claimed_at is not None