diff --git a/habitat/__init__.py b/habitat/__init__.py
new file mode 100644
index 0000000..29739b3
--- /dev/null
+++ b/habitat/__init__.py
@@ -0,0 +1 @@
+__version__='1.3.0'
diff --git a/habitat/__main__.py b/habitat/__main__.py
new file mode 100644
index 0000000..9d94473
--- /dev/null
+++ b/habitat/__main__.py
@@ -0,0 +1,543 @@
+from __future__ import annotations
+import argparse, json, os, sys, secrets
+from pathlib import Path
+from .schema import Habitat, Job, Action, utcnow
+from .store import Store
+from .config import load_config, validate_config
+from .runtime import run_once, run_loop
+from .claims import Claim
+from .verify import verify_claim
+from .verify_api import proof_status, verify_and_prove
+from .adapters import JSONFileEvidenceAdapter, HTTPJSONEvidenceAdapter, GitHubIssueEvidenceAdapter
+from .events import signature_for, ingest_event
+from .server import serve
+from .generate import render_dashboard, write_dashboard
+from .doctor import diagnose
+from .backup import backup, restore
+from .proof_sign import sign_proof
+from .proof_card import render_proof_card
+from .verdicts import explain_reason, boundary_sentence
+
+DEFAULT_DB = Path(".habitat/habitat.db")
+
+
+def get_store(a):
+ return Store(a.db)
+
+
+def cmd_init(a):
+ cfg = validate_config(load_config(a.config))
+ s = Store(a.db)
+ now = utcnow()
+ s.save_habitat(Habitat(cfg["id"], cfg["name"], cfg.get("model", "unknown"), now, now))
+ for j in cfg.get("jobs", []):
+ s.save_job(
+ Job(
+ j["id"],
+ cfg["id"],
+ j["name"],
+ j.get("schedule"),
+ j.get("enabled", True),
+ j.get("command"),
+ )
+ )
+ s.close()
+ print(f"Initialized Habitat {cfg['id']} at {a.db}")
+ return 0
+
+
+DEMO_RUN_ID = "run-42"
+DEMO_ACTION = "run_job"
+DEMO_CLAIM = "deployed v1.2.3 to production"
+DEMO_PROVE_COMMAND = (
+ f'habitat prove --run-id {DEMO_RUN_ID} --claim "{DEMO_CLAIM}" '
+ "-o proof.json --card"
+)
+
+
+def cmd_demo(a):
+ """Record one clearly-labelled sample action so the README prove flow works."""
+ s = get_store(a)
+ try:
+ h = s.habitat()
+ existing = s.matching_actions(h.id, action=DEMO_ACTION, run_id=DEMO_RUN_ID)
+ if existing:
+ print(
+ json.dumps(
+ {
+ "run_id": DEMO_RUN_ID,
+ "action": DEMO_ACTION,
+ "status": existing[0].status,
+ "sample": True,
+ "seeded": False,
+ "next": DEMO_PROVE_COMMAND,
+ },
+ indent=2,
+ )
+ )
+ return 0
+ jobs = s.jobs()
+ job_id = jobs[0].id if jobs else None
+ s.save_action(
+ Action(
+ id="demo-run-42-action",
+ habitat_id=h.id,
+ timestamp=utcnow(),
+ actor="agent",
+ action=DEMO_ACTION,
+ status="ok",
+ job_id=job_id,
+ details={"sample": True, "target": "production", "version": "1.2.3"},
+ run_id=DEMO_RUN_ID,
+ )
+ )
+ print(
+ json.dumps(
+ {
+ "run_id": DEMO_RUN_ID,
+ "action": DEMO_ACTION,
+ "job_id": job_id,
+ "status": "ok",
+ "sample": True,
+ "seeded": True,
+ "boundary": boundary_sentence("demo"),
+ "next": DEMO_PROVE_COMMAND,
+ },
+ indent=2,
+ )
+ )
+ print()
+ print("Sample demo data recorded (not a live verification).")
+ print(f"Next: {DEMO_PROVE_COMMAND}")
+ return 0
+ except RuntimeError as exc:
+ print(
+ json.dumps(
+ {
+ "ok": False,
+ "status": "rejected",
+ "error": "not_initialized",
+ "hint": "habitat init --config examples/simple.yaml",
+ "detail": str(exc),
+ **(explain_reason("not_initialized") or {}),
+ }
+ )
+ )
+ return 1
+ finally:
+ s.close()
+
+
+def cmd_status(a):
+ s = get_store(a)
+ h = s.habitat()
+ print(
+ json.dumps(
+ {
+ "id": h.id,
+ "name": h.name,
+ "status": h.status,
+ "jobs": len(s.jobs()),
+ "active_signals": len(s.active_signals()),
+ "agents": len(s.agents()),
+ },
+ indent=2,
+ )
+ )
+ s.close()
+ return 0
+
+
+def cmd_run(a):
+ s = get_store(a)
+ try:
+ if a.watch:
+ run_loop(s, interval=a.interval, stop_after=a.duration)
+ else:
+ run_once(s)
+ finally:
+ s.close()
+ return 0
+
+
+def cmd_inspect(a):
+ s = get_store(a)
+ print(
+ json.dumps(
+ {
+ "habitat": s.habitat().__dict__,
+ "jobs": [j.__dict__ for j in s.jobs()],
+ "signals": [x.__dict__ for x in s.signals()],
+ },
+ default=str,
+ indent=2,
+ )
+ )
+ s.close()
+ return 0
+
+
+def cmd_claim(a):
+ s = get_store(a)
+ c = Claim.new(s.habitat().id, a.claim, a.job, a.action, a.expected_status)
+ c.run_id = a.run_id
+ s.save_claim(c)
+ print(json.dumps({"id": c.id, "status": c.status, "run_id": c.run_id, "claim": c.claim}, default=str))
+ s.close()
+ return 0
+
+
+def _evidence(a):
+ if a.evidence_file:
+ return JSONFileEvidenceAdapter(a.evidence_file)
+ if a.evidence_url:
+ return HTTPJSONEvidenceAdapter(a.evidence_url, a.token_env)
+ if a.github_issue:
+ return GitHubIssueEvidenceAdapter(a.token_env)
+ return None
+
+
+def cmd_verify(a):
+ s = get_store(a)
+ c = s.get_claim(a.id) if a.id else (s.claims(1)[0] if s.claims(1) else None)
+ if not c:
+ print("No claim found")
+ s.close()
+ return 1
+ expected = json.loads(a.evidence_expected) if a.evidence_expected else None
+ c = verify_claim(s, c, _evidence(a), a.query or a.github_issue, expected)
+ result = {"id": c.id, "status": c.status, "evidence": c.evidence}
+ explanation = explain_reason(c.evidence.get("reason") if isinstance(c.evidence, dict) else None)
+ if explanation:
+ result["explanation"] = explanation
+ boundary = boundary_sentence(c.status)
+ if boundary:
+ result["boundary"] = boundary
+ print(json.dumps(result, default=str))
+ s.close()
+ return 0
+
+
+def cmd_proof(a):
+ s = get_store(a)
+ try:
+ result = verify_and_prove(s, a.id) if a.reverify else proof_status(s, a.id)
+ proof = result["proof"]
+ if a.signing_key_env:
+ private_key = os.getenv(a.signing_key_env)
+ if not private_key:
+ raise ValueError(f"Signing key environment variable is not set: {a.signing_key_env}")
+ proof = sign_proof(proof, private_key, a.key_id, a.agent_id)
+ result["proof"] = proof
+ result["signed"] = True
+ else:
+ result["signed"] = False
+ if a.output:
+ Path(a.output).write_text(
+ json.dumps(proof, default=str, indent=2, sort_keys=True) + "\n",
+ encoding="utf-8",
+ )
+ print(f"Wrote proof to {a.output}")
+ else:
+ print(json.dumps(result, default=str, indent=2))
+ return 0
+ except KeyError:
+ print(json.dumps({"error": "claim_not_found"}))
+ return 1
+ except (RuntimeError, ValueError) as exc:
+ print(json.dumps({"error": str(exc)}))
+ return 1
+ finally:
+ s.close()
+
+
+def cmd_prove(a):
+ """Create a claim for a run, verify it, and export a portable proof (+ optional card)."""
+ s = get_store(a)
+ try:
+ if not a.run_id and not a.claim:
+ print(json.dumps({"error": "run_id_or_claim_required"}))
+ return 2
+
+ claim_text = a.claim or f"run {a.run_id} completed successfully"
+ c = Claim.new(
+ s.habitat().id,
+ claim_text,
+ job_id=a.job,
+ action=a.action or "run_job",
+ expected_status=a.expected_status or "ok",
+ )
+ c.run_id = a.run_id
+ s.save_claim(c)
+
+ c = verify_claim(s, c)
+ if c.status != "verified":
+ result = {
+ "claim_id": c.id,
+ "status": c.status,
+ "evidence": c.evidence,
+ "error": "claim_not_verified",
+ }
+ reason = c.evidence.get("reason") if isinstance(c.evidence, dict) else None
+ explanation = explain_reason(reason)
+ if explanation:
+ result["reason"] = reason
+ result["explanation"] = explanation
+ result["hint"] = explanation["next_action"]
+ boundary = boundary_sentence(c.status)
+ if boundary:
+ result["boundary"] = boundary
+ print(json.dumps(result, default=str, indent=2))
+ if explanation:
+ print()
+ print(f"Verdict: {c.status} — {explanation['meaning']}")
+ print(f"What would change this verdict: {explanation['what_would_change']}")
+ print(f"Next action: {explanation['next_action']}")
+ return 1
+
+ result = verify_and_prove(s, c.id) if a.reverify else proof_status(s, c.id)
+ proof = result["proof"]
+
+ if a.signing_key_env:
+ private_key = os.getenv(a.signing_key_env)
+ if not private_key:
+ raise ValueError(f"Signing key environment variable is not set: {a.signing_key_env}")
+ proof = sign_proof(proof, private_key, a.key_id, a.agent_id)
+ result["proof"] = proof
+ result["signed"] = True
+ else:
+ result["signed"] = False
+
+ # --card must never reference a proof file that was not written:
+ # without an explicit -o, default to proof.json so the card's
+ # verify command names a file that actually exists (UX-06).
+ if a.card and not a.output:
+ a.output = "proof.json"
+
+ if a.output:
+ Path(a.output).write_text(
+ json.dumps(proof, default=str, indent=2, sort_keys=True) + "\n",
+ encoding="utf-8",
+ )
+ print(f"Wrote proof → {a.output}")
+
+ if a.card:
+ card = render_proof_card(
+ proof, {"valid": True, "verdict": c.status}, proof_filename=a.output or "proof.json"
+ )
+ card_path = Path(a.output).with_suffix(".md") if a.output else Path("proof.md")
+ card_path.write_text(card, encoding="utf-8")
+ print(f"Wrote proof card → {card_path}")
+ print()
+ print(card)
+
+ print(
+ json.dumps(
+ {
+ "claim_id": c.id,
+ "status": c.status,
+ "run_id": c.run_id,
+ "proof_file": a.output,
+ "signed": result.get("signed", False),
+ "boundary": boundary_sentence(c.status),
+ },
+ indent=2,
+ default=str,
+ )
+ )
+ return 0
+ except Exception as exc:
+ print(json.dumps({"error": str(exc)}))
+ return 1
+ finally:
+ s.close()
+
+
+def cmd_event(a):
+ s = get_store(a)
+ body = Path(a.file).read_bytes() if a.file else a.json.encode()
+ secret = os.getenv(a.secret_env) if a.secret_env else None
+ agent_secret = os.getenv(a.agent_secret_env) if a.agent_secret_env else None
+ sig = a.signature or (signature_for(secret, body) if secret else None)
+ try:
+ print(json.dumps(ingest_event(s, body, sig, secret, not a.no_signature, a.agent_id, agent_secret), default=str))
+ return 0
+ except (ValueError, PermissionError) as e:
+ print(json.dumps({"accepted": False, "error": str(e)}))
+ return 1
+ finally:
+ s.close()
+
+
+def cmd_agent_add(a):
+ s = get_store(a)
+ h = s.habitat()
+ secret = os.getenv(a.secret_env) if a.secret_env else None
+ secret = secret or secrets.token_urlsafe(32)
+ s.save_agent(a.id, h.id, a.name, True, a.permission, secret)
+ s.close()
+ print(json.dumps({"agent_id": a.id, "secret": secret, "permissions": a.permission}))
+ return 0
+
+
+def cmd_agents(a):
+ s = get_store(a)
+ print(json.dumps(s.agents(), indent=2, default=str))
+ s.close()
+ return 0
+
+
+def cmd_doctor(a):
+ checks = diagnose(a.db, a.config)
+ bad = False
+ for n, ok, msg in checks:
+ print(f"{'✓' if ok else '✗'} {n}: {msg}")
+ bad |= not ok
+ print("Result:", "NEEDS ATTENTION" if bad else "HEALTHY")
+ return 1 if bad else 0
+
+
+def cmd_backup(a):
+ print(f"Backup written to {backup(a.db, a.output)}")
+ return 0
+
+
+def cmd_restore(a):
+ print(f"Restored database to {restore(a.input, a.db)}")
+ return 0
+
+
+def cmd_generate(a):
+ s = get_store(a)
+ try:
+ h = s.habitat()
+ p = write_dashboard(
+ render_dashboard(h, s.jobs(), s.signals(), s.actions(), h.name, s.claims()),
+ a.output,
+ )
+ print(f"Wrote {p.resolve()}")
+ return 0
+ finally:
+ s.close()
+
+
+def main(argv=None):
+ p = argparse.ArgumentParser(
+ prog="habitat",
+ description="Local-first supervision and accountability runtime for autonomous agents.",
+ )
+ p.add_argument("--db", default=str(DEFAULT_DB))
+ sub = p.add_subparsers(dest="command", required=True)
+
+ x = sub.add_parser("init")
+ x.add_argument("--config", required=True)
+ x.set_defaults(func=cmd_init)
+
+ x = sub.add_parser(
+ "demo",
+ aliases=["seed"],
+ help="Record one sample action (run-42) so the README prove flow works on a fresh DB",
+ )
+ x.set_defaults(func=cmd_demo)
+
+ sub.add_parser("status").set_defaults(func=cmd_status)
+
+ x = sub.add_parser("run")
+ x.add_argument("--watch", action="store_true")
+ x.add_argument("--interval", type=float, default=10)
+ x.add_argument("--duration", type=float)
+ x.set_defaults(func=cmd_run)
+
+ sub.add_parser("inspect").set_defaults(func=cmd_inspect)
+
+ x = sub.add_parser("generate")
+ x.add_argument("-o", "--output", default="dashboard.html")
+ x.set_defaults(func=cmd_generate)
+
+ x = sub.add_parser("claim")
+ x.add_argument("claim")
+ x.add_argument("--job")
+ x.add_argument("--action")
+ x.add_argument("--expected-status", default="ok")
+ x.add_argument("--run-id")
+ x.set_defaults(func=cmd_claim)
+
+ x = sub.add_parser("verify")
+ x.add_argument("--id")
+ x.add_argument("--query")
+ x.add_argument("--evidence-file")
+ x.add_argument("--evidence-url")
+ x.add_argument("--github-issue")
+ x.add_argument("--token-env", default="GITHUB_TOKEN")
+ x.add_argument("--evidence-expected")
+ x.set_defaults(func=cmd_verify)
+
+ x = sub.add_parser("proof")
+ x.add_argument("id")
+ x.add_argument("--reverify", action="store_true")
+ x.add_argument("-o", "--output", help="write only the portable proof bundle to this JSON file")
+ x.add_argument("--signing-key-env", help="environment variable containing a base64 Ed25519 private key")
+ x.add_argument("--key-id", help="trusted-key registry identifier to embed in the signature")
+ x.add_argument("--agent-id", help="publisher agent identifier to bind into the signature")
+ x.set_defaults(func=cmd_proof)
+
+ x = sub.add_parser("prove", help="Verify a claim for a run and export a portable proof")
+ x.add_argument("--claim", help="Human-readable claim text")
+ x.add_argument("--run-id", help="Bind the claim to this Habitat run ID")
+ x.add_argument("--job", help="Optional job ID")
+ x.add_argument("--action", help="Expected action name (default: run_job)")
+ x.add_argument("--expected-status", default="ok")
+ x.add_argument("-o", "--output", help="Write portable proof.json to this path")
+ x.add_argument("--card", action="store_true", help="Also write a short markdown proof card (defaults -o to proof.json if unset)")
+ x.add_argument("--reverify", action="store_true")
+ x.add_argument("--signing-key-env")
+ x.add_argument("--key-id")
+ x.add_argument("--agent-id")
+ x.set_defaults(func=cmd_prove)
+
+ x = sub.add_parser("event")
+ x.add_argument("--file")
+ x.add_argument("--json", default="")
+ x.add_argument("--signature")
+ x.add_argument("--secret-env", default="HABITAT_WEBHOOK_SECRET")
+ x.add_argument("--agent-id")
+ x.add_argument("--agent-secret-env", default="HABITAT_AGENT_SECRET")
+ x.add_argument("--no-signature", action="store_true")
+ x.set_defaults(func=cmd_event)
+
+ sub.add_parser("agents").set_defaults(func=cmd_agents)
+
+ x = sub.add_parser("agent-add")
+ x.add_argument("id")
+ x.add_argument("name")
+ x.add_argument("--permission", action="append", default=["submit_events"])
+ x.add_argument("--secret-env", default="HABITAT_AGENT_SECRET")
+ x.set_defaults(func=cmd_agent_add)
+
+ x = sub.add_parser("doctor")
+ x.add_argument("--config")
+ x.set_defaults(func=cmd_doctor)
+
+ x = sub.add_parser("backup")
+ x.add_argument("output")
+ x.set_defaults(func=cmd_backup)
+
+ x = sub.add_parser("restore")
+ x.add_argument("input")
+ x.set_defaults(func=cmd_restore)
+
+ x = sub.add_parser("serve")
+ x.add_argument("--host", default="127.0.0.1")
+ x.add_argument("--port", type=int, default=8787)
+ x.add_argument("--secret-env", default="HABITAT_WEBHOOK_SECRET")
+ x.add_argument("--allow-unsigned", action="store_true")
+ x.set_defaults(
+ func=lambda a: serve(a.db, a.host, a.port, os.getenv(a.secret_env), not a.allow_unsigned)
+ )
+
+ args = p.parse_args(argv)
+ return args.func(args)
+
+
+if __name__ == "__main__":
+ sys.exit(main(argv=None))
diff --git a/habitat/adapters.py b/habitat/adapters.py
new file mode 100644
index 0000000..7ae4520
--- /dev/null
+++ b/habitat/adapters.py
@@ -0,0 +1,209 @@
+"""Integration boundaries for Habitat runtimes and evidence sources.
+
+Adapters are intentionally small and stdlib-only. They let Habitat verify claims
+against systems outside its own process without making those systems part of the
+core domain model.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import ipaddress
+import json
+import os
+from pathlib import Path
+import re
+import shlex
+import socket
+import subprocess
+from typing import Any, Protocol
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlparse
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+MAX_EVIDENCE_RESPONSE_BYTES = 1_048_576
+_SAFE_GITHUB_NAME = re.compile(r"^[A-Za-z0-9_.-]+$")
+
+
+class _NoRedirectHandler(HTTPRedirectHandler):
+ """Do not silently turn an evidence URL into an attacker-chosen URL."""
+
+ def redirect_request(self, req, fp, code, msg, headers, newurl):
+ raise URLError("redirect_not_allowed")
+
+
+_NO_REDIRECT_OPENER = build_opener(_NoRedirectHandler)
+
+
+def _validate_http_url(url: str) -> None:
+ parsed = urlparse(url)
+ if parsed.scheme not in {"http", "https"}:
+ raise ValueError("unsupported_evidence_url_scheme")
+ if not parsed.hostname:
+ raise ValueError("evidence_url_missing_host")
+ if parsed.username is not None or parsed.password is not None:
+ raise ValueError("evidence_url_credentials_not_allowed")
+
+ # Evidence URLs are configuration, but a compromised/untrusted configuration
+ # must not turn the adapter into an easy SSRF primitive. Resolve every address
+ # before opening the URL and reject any hostname that can resolve to a local,
+ # private, link-local, loopback, multicast, reserved, or unspecified address.
+ try:
+ addresses = {item[4][0] for item in socket.getaddrinfo(parsed.hostname, parsed.port or (443 if parsed.scheme == "https" else 80), type=socket.SOCK_STREAM)}
+ except socket.gaierror as exc:
+ raise ValueError("evidence_url_host_unresolvable") from exc
+ if not addresses:
+ raise ValueError("evidence_url_host_unresolvable")
+ for address in addresses:
+ try:
+ ip = ipaddress.ip_address(address)
+ except ValueError as exc:
+ raise ValueError("evidence_url_invalid_address") from exc
+ if not ip.is_global:
+ raise ValueError("evidence_url_private_address")
+
+
+def _read_limited(response) -> bytes:
+ content_length = response.headers.get("Content-Length")
+ if content_length:
+ try:
+ if int(content_length) > MAX_EVIDENCE_RESPONSE_BYTES:
+ raise ValueError("evidence_response_too_large")
+ except ValueError:
+ if content_length != "0":
+ raise ValueError("invalid_content_length")
+ data = response.read(MAX_EVIDENCE_RESPONSE_BYTES + 1)
+ if len(data) > MAX_EVIDENCE_RESPONSE_BYTES:
+ raise ValueError("evidence_response_too_large")
+ return data
+
+
+@dataclass
+class ExecutionResult:
+ ok: bool
+ stdout: str = ""
+ stderr: str = ""
+ exit_code: int | None = None
+
+
+class JobAdapter(Protocol):
+ def execute(self, command: str, timeout: int = 300) -> ExecutionResult: ...
+
+
+class ShellAdapter:
+ """Local command adapter. Safe argv execution is the default; shell mode is explicit."""
+
+ def __init__(self, allow_shell: bool = False):
+ self.allow_shell = allow_shell
+
+ def execute(self, command: str, timeout: int = 300) -> ExecutionResult:
+ try:
+ argv = command if self.allow_shell else shlex.split(command)
+ p = subprocess.run(argv, shell=self.allow_shell, capture_output=True, text=True, timeout=timeout)
+ return ExecutionResult(p.returncode == 0, p.stdout[-4000:], p.stderr[-4000:], p.returncode)
+ except subprocess.TimeoutExpired as exc:
+ return ExecutionResult(False, str(exc.stdout or "")[-4000:], "command timed out", None)
+ except Exception as exc:
+ return ExecutionResult(False, "", str(exc)[-4000:], None)
+
+
+@dataclass
+class Evidence:
+ """Normalized evidence returned by an external system."""
+
+ source: str
+ observed: bool
+ status: str | None = None
+ data: dict[str, Any] = field(default_factory=dict)
+ error: str | None = None
+
+
+class EvidenceAdapter(Protocol):
+ def observe(self, query: str) -> Evidence: ...
+
+
+class JSONFileEvidenceAdapter:
+ """Read deterministic evidence from a local JSON document."""
+
+ def __init__(self, path: str | Path):
+ self.path = Path(path)
+
+ def observe(self, query: str) -> Evidence:
+ try:
+ payload = json.loads(self.path.read_text(encoding="utf-8"))
+ value: Any = payload
+ for part in query.split(".") if query else []:
+ if not isinstance(value, dict) or part not in value:
+ return Evidence(str(self.path), False, error=f"path_not_found:{query}")
+ value = value[part]
+ if isinstance(value, dict):
+ return Evidence(str(self.path), True, value.get("status"), value)
+ return Evidence(str(self.path), True, str(value) if value is not None else None, {"value": value})
+ except Exception as exc:
+ return Evidence(str(self.path), False, error=str(exc))
+
+
+class HTTPJSONEvidenceAdapter:
+ """Fetch bounded JSON evidence from a fixed HTTP endpoint without redirects."""
+
+ def __init__(self, base_url: str, token_env: str | None = None, timeout: int = 15):
+ _validate_http_url(base_url)
+ if timeout <= 0 or timeout > 300:
+ raise ValueError("invalid_evidence_timeout")
+ self.base_url = base_url.rstrip("/")
+ self.token_env = token_env
+ self.timeout = timeout
+
+ def observe(self, query: str) -> Evidence:
+ req = Request(self.base_url, headers={"Accept": "application/json", "User-Agent": "Habitat/1.0"})
+ token = os.getenv(self.token_env) if self.token_env else None
+ if token:
+ req.add_header("Authorization", f"Bearer {token}")
+ try:
+ with _NO_REDIRECT_OPENER.open(req, timeout=self.timeout) as response:
+ payload = json.loads(_read_limited(response).decode("utf-8"), parse_constant=lambda value: (_ for _ in ()).throw(ValueError("invalid_json_constant")))
+ value: Any = payload
+ for part in query.split(".") if query else []:
+ if not isinstance(value, dict) or part not in value:
+ return Evidence(self.base_url, False, error=f"path_not_found:{query}")
+ value = value[part]
+ if isinstance(value, dict):
+ return Evidence(self.base_url, True, value.get("status"), value)
+ return Evidence(self.base_url, True, str(value) if value is not None else None, {"value": value})
+ except HTTPError as exc:
+ return Evidence(self.base_url, False, error=f"http_{exc.code}")
+ except (URLError, TimeoutError, ValueError) as exc:
+ return Evidence(self.base_url, False, error=str(exc))
+
+
+class GitHubIssueEvidenceAdapter:
+ """Verify that a GitHub issue exists. Query format: owner/repo#number."""
+
+ def __init__(self, token_env: str = "GITHUB_TOKEN", timeout: int = 15):
+ if timeout <= 0 or timeout > 300:
+ raise ValueError("invalid_evidence_timeout")
+ self.token_env = token_env
+ self.timeout = timeout
+
+ def observe(self, query: str) -> Evidence:
+ try:
+ repo, number = query.rsplit("#", 1)
+ owner, name = repo.split("/", 1)
+ if not _SAFE_GITHUB_NAME.fullmatch(owner) or not _SAFE_GITHUB_NAME.fullmatch(name):
+ raise ValueError("invalid_github_repository")
+ issue_number = int(number)
+ if issue_number <= 0:
+ raise ValueError("invalid_github_issue_number")
+ url = f"https://api.github.com/repos/{owner}/{name}/issues/{issue_number}"
+ req = Request(url, headers={"Accept": "application/vnd.github+json", "User-Agent": "Habitat/1.0"})
+ token = os.getenv(self.token_env)
+ if token:
+ req.add_header("Authorization", f"Bearer {token}")
+ with _NO_REDIRECT_OPENER.open(req, timeout=self.timeout) as response:
+ payload = json.loads(_read_limited(response).decode("utf-8"), parse_constant=lambda value: (_ for _ in ()).throw(ValueError("invalid_json_constant")))
+ return Evidence("github", True, "ok", {"number": payload.get("number"), "state": payload.get("state"), "title": payload.get("title"), "url": payload.get("html_url")})
+ except HTTPError as exc:
+ if exc.code == 404:
+ return Evidence("github", False, error="issue_not_found")
+ return Evidence("github", False, error=f"http_{exc.code}")
+ except (ValueError, URLError, TimeoutError) as exc:
+ return Evidence("github", False, error=str(exc))
diff --git a/habitat/anchor.py b/habitat/anchor.py
new file mode 100644
index 0000000..9d97f11
--- /dev/null
+++ b/habitat/anchor.py
@@ -0,0 +1,75 @@
+"""Optional external tip anchoring for Habitat proofs.
+
+Local SHA-256 chains are tamper-evident within the producer's database.
+An external tip raises the bar against whole-ledger rewrite by recording the
+current tip digest outside the Habitat database (git notes, a tip file, or a
+simple public location). This module stays local-first and optional.
+
+Habitat still does not claim external-world truth or remote attestation.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+def tip_digest(proof_or_ledger: dict[str, Any]) -> str:
+ """Compute a stable tip digest from a proof bundle or a ledger object."""
+ if "content_sha256" in proof_or_ledger and isinstance(proof_or_ledger["content_sha256"], str):
+ return proof_or_ledger["content_sha256"]
+ actions = []
+ if isinstance(proof_or_ledger.get("ledger"), dict):
+ actions = proof_or_ledger["ledger"].get("actions") or []
+ elif isinstance(proof_or_ledger.get("actions"), list):
+ actions = proof_or_ledger["actions"]
+ payload = json.dumps(actions, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+ return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def write_tip_file(
+ tip: str,
+ path: str | Path,
+ *,
+ claim_id: str | None = None,
+ run_id: str | None = None,
+ proof_path: str | None = None,
+) -> Path:
+ """Write a small tip record that a verifier or human can inspect later."""
+ out = Path(path)
+ out.parent.mkdir(parents=True, exist_ok=True)
+ record = {
+ "habitat_tip_version": 1,
+ "recorded_at": datetime.now(timezone.utc).isoformat(),
+ "tip": tip,
+ "claim_id": claim_id,
+ "run_id": run_id,
+ "proof_path": proof_path,
+ }
+ out.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+ return out
+
+
+def read_tip_file(path: str | Path) -> dict[str, Any]:
+ data = json.loads(Path(path).read_text(encoding="utf-8"))
+ if not isinstance(data, dict) or data.get("habitat_tip_version") != 1:
+ raise ValueError("not a Habitat tip file")
+ if not isinstance(data.get("tip"), str) or len(data["tip"]) != 64:
+ raise ValueError("tip digest missing or malformed")
+ return data
+
+
+def check_tip(proof: dict[str, Any], tip_path: str | Path) -> dict[str, Any]:
+ """Compare a proof's content digest against an external tip file."""
+ expected = read_tip_file(tip_path)
+ actual = proof.get("content_sha256")
+ match = isinstance(actual, str) and actual == expected["tip"]
+ return {
+ "match": match,
+ "expected_tip": expected["tip"],
+ "actual_tip": actual,
+ "tip_file": str(tip_path),
+ "recorded_at": expected.get("recorded_at"),
+ }
diff --git a/habitat/backup.py b/habitat/backup.py
new file mode 100644
index 0000000..870ea16
--- /dev/null
+++ b/habitat/backup.py
@@ -0,0 +1,14 @@
+from pathlib import Path
+import sqlite3
+
+def backup(db_path,destination):
+ src=sqlite3.connect(db_path); dst_path=Path(destination); dst_path.parent.mkdir(parents=True,exist_ok=True); dst=sqlite3.connect(dst_path)
+ try:src.backup(dst)
+ finally:dst.close();src.close()
+ return dst_path
+
+def restore(source,db_path):
+ src=sqlite3.connect(source); dst=sqlite3.connect(db_path)
+ try:src.backup(dst)
+ finally:dst.close();src.close()
+ return Path(db_path)
diff --git a/habitat/claims.py b/habitat/claims.py
new file mode 100644
index 0000000..c31c37c
--- /dev/null
+++ b/habitat/claims.py
@@ -0,0 +1,28 @@
+"""Agent claims: assertions that Habitat can verify against trusted state."""
+from __future__ import annotations
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any, Literal
+import uuid
+from .schema import utcnow
+
+ClaimStatus = Literal["pending", "verified", "rejected", "inconclusive"]
+
+@dataclass
+class Claim:
+ id: str
+ habitat_id: str
+ job_id: str | None
+ claim: str
+ action: str | None = None
+ expected_status: str = "ok"
+ created_at: datetime = field(default_factory=utcnow)
+ verified_at: datetime | None = None
+ status: ClaimStatus = "pending"
+ evidence: dict[str, Any] = field(default_factory=dict)
+ run_id: str | None = None
+
+ @classmethod
+ def new(cls, habitat_id: str, claim: str, job_id: str | None = None,
+ action: str | None = None, expected_status: str = "ok") -> "Claim":
+ return cls(str(uuid.uuid4()), habitat_id, job_id, claim, action, expected_status)
diff --git a/habitat/config.py b/habitat/config.py
new file mode 100644
index 0000000..a689443
--- /dev/null
+++ b/habitat/config.py
@@ -0,0 +1,51 @@
+"""Dependency-free configuration loader and validator."""
+from __future__ import annotations
+import json
+from pathlib import Path
+
+def _scalar(value):
+ value=value.strip()
+ if not value:return None
+ if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):return value[1:-1]
+ if value.lower() in {'true','false'}:return value.lower()=='true'
+ if value.lower() in {'null','none'}:return None
+ return value
+
+def load_yaml(path):
+ lines=Path(path).read_text(encoding='utf-8').splitlines(); result={}; jobs=[]; in_jobs=False; current=None
+ for raw in lines:
+ line=raw.split('#',1)[0].rstrip(); stripped=line.strip()
+ if not stripped:continue
+ if stripped=='jobs:':in_jobs=True;result['jobs']=jobs;continue
+ if in_jobs and stripped.startswith('- '):
+ current={};jobs.append(current); item=stripped[2:].strip()
+ if item:key,value=item.split(':',1);current[key.strip()]=_scalar(value)
+ continue
+ if in_jobs and line.startswith(' ') and current is not None and ':' in stripped:
+ key,value=stripped.split(':',1);current[key.strip()]=_scalar(value);continue
+ in_jobs=False
+ if ':' in stripped:key,value=stripped.split(':',1);result[key.strip()]=_scalar(value)
+ return result
+
+def load_config(path):
+ p=Path(path)
+ if p.suffix.lower()=='.json':return json.loads(p.read_text(encoding='utf-8'))
+ if p.suffix.lower() in {'.yaml','.yml'}:return load_yaml(p)
+ raise ValueError('Config must be .yaml, .yml, or .json')
+
+def validate_config(cfg):
+ if not isinstance(cfg,dict):raise ValueError('configuration must be an object')
+ if not isinstance(cfg.get('id'),str) or not cfg['id'].strip():raise ValueError('id must be a non-empty string')
+ if not isinstance(cfg.get('name'),str) or not cfg['name'].strip():raise ValueError('name must be a non-empty string')
+ jobs=cfg.get('jobs',[])
+ if not isinstance(jobs,list):raise ValueError('jobs must be a list')
+ seen=set()
+ for i,j in enumerate(jobs):
+ if not isinstance(j,dict):raise ValueError(f'jobs[{i}] must be an object')
+ for key in ('id','name'):
+ if not isinstance(j.get(key),str) or not j[key].strip():raise ValueError(f'jobs[{i}].{key} must be a non-empty string')
+ if j['id'] in seen:raise ValueError(f'duplicate job id: {j["id"]}')
+ seen.add(j['id'])
+ if 'enabled' in j and not isinstance(j['enabled'],bool):raise ValueError(f'jobs[{i}].enabled must be boolean')
+ if j.get('command') is not None and not isinstance(j['command'],str):raise ValueError(f'jobs[{i}].command must be string')
+ return cfg
diff --git a/habitat/doctor.py b/habitat/doctor.py
new file mode 100644
index 0000000..ae44b9b
--- /dev/null
+++ b/habitat/doctor.py
@@ -0,0 +1,25 @@
+from __future__ import annotations
+import os, sys
+from .store import Store
+
+def diagnose(db_path, config_path=None):
+ checks = []
+ try:
+ s = Store(db_path)
+ s.habitat()
+ version = s.conn.execute("SELECT value FROM schema_meta WHERE key='version'").fetchone()[0]
+ integrity_ok = s.verify_action_integrity()
+ checks.append(('database', True, 'SQLite readable'))
+ checks.append(('schema', True, f'version {version}'))
+ checks.append(('action_integrity', integrity_ok, 'hash chain intact' if integrity_ok else 'action hash chain failed'))
+ config_ok = config_path is None or os.path.exists(config_path)
+ checks.append(('configuration', config_ok, 'present' if config_ok else 'missing'))
+ checks.append(('jobs', True, f'{len(s.jobs())} configured'))
+ checks.append(('agents', True, f'{len(s.agents())} registered'))
+ s.close()
+ except Exception as exc:
+ checks.append(('database', False, str(exc)))
+ checks.append(('python', sys.version_info >= (3, 10), sys.version.split()[0]))
+ secret_set = bool(os.getenv('HABITAT_WEBHOOK_SECRET'))
+ checks.append(('signing_secret', secret_set, 'set' if secret_set else 'not set; configure HABITAT_WEBHOOK_SECRET before starting signed event ingestion'))
+ return checks
diff --git a/habitat/events.py b/habitat/events.py
new file mode 100644
index 0000000..ba3b473
--- /dev/null
+++ b/habitat/events.py
@@ -0,0 +1,132 @@
+"""Secure structured event ingestion."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import uuid
+
+from .claims import Claim
+from .schema import Action, utcnow
+from .store import Store
+from .verify import verify_claim
+
+MAX_EVENT_BYTES = 256 * 1024
+ALLOWED_TYPES = {"agent.heartbeat", "job.started", "job.completed", "job.failed", "claim.submitted"}
+PERMISSIONS = {"agent.heartbeat": "submit_events", "job.started": "submit_events", "job.completed": "submit_events", "job.failed": "submit_events", "claim.submitted": "submit_claims"}
+
+
+def signature_for(secret, body):
+ return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
+
+
+def valid_signature(secret, body, supplied):
+ if not secret or not supplied:
+ return False
+ normalized = supplied.removeprefix("sha256=").strip()
+ if len(normalized) != 64:
+ return False
+ try:
+ int(normalized, 16)
+ except ValueError:
+ return False
+ return hmac.compare_digest(signature_for(secret, body), normalized)
+
+
+def _reject_constant(value):
+ raise ValueError(f"non_standard_json_number:{value}")
+
+
+def _validate(p, store):
+ eid, typ, hid = p.get("id"), p.get("type"), p.get("habitat_id")
+ if not isinstance(eid, str) or not eid.strip() or len(eid) > 200:
+ raise ValueError("invalid_id")
+ if typ not in ALLOWED_TYPES:
+ raise ValueError("unsupported_event_type")
+ if not isinstance(hid, str) or not hid:
+ raise ValueError("missing_habitat_id")
+ if store.habitat().id != hid:
+ raise ValueError("habitat_id_mismatch")
+ if typ == "agent.heartbeat":
+ return eid, typ, hid
+ if typ == "claim.submitted":
+ if not isinstance(p.get("claim"), str) or not p["claim"].strip():
+ raise ValueError("missing_claim")
+ if p.get("job_id") is not None and not isinstance(p["job_id"], str):
+ raise ValueError("invalid_job_id")
+ else:
+ if not isinstance(p.get("job_id"), str):
+ raise ValueError("missing_job_id")
+ if not any(j.id == p["job_id"] for j in store.jobs()):
+ raise ValueError("unknown_job_id")
+ return eid, typ, hid
+
+
+def ingest_event(store, body, signature, secret, require_signature=True, agent_id=None, agent_secret=None):
+ if len(body) > MAX_EVENT_BYTES:
+ raise ValueError("event_too_large")
+ try:
+ p = json.loads(body.decode("utf-8"), parse_constant=_reject_constant)
+ except Exception as exc:
+ raise ValueError("invalid_json") from exc
+ if not isinstance(p, dict):
+ raise ValueError("event_must_be_object")
+ eid, typ, hid = _validate(p, store)
+ ok = valid_signature(secret or "", body, signature)
+ if require_signature and not ok:
+ raise PermissionError("invalid_signature")
+ payload_agent = p.get("agent_id")
+ if payload_agent is not None and not isinstance(payload_agent, str):
+ raise ValueError("invalid_agent_id")
+ if agent_id is not None and payload_agent is not None and payload_agent != agent_id:
+ raise PermissionError("agent_identity_mismatch")
+ declared_agent = payload_agent or agent_id
+
+ # Event acceptance, agent last-seen state, derived actions/jobs, and claims
+ # are one transaction. A downstream failure must not leave a duplicate event
+ # that causes the retry to be discarded while its side effects are missing.
+ with store.transaction(immediate=True):
+ if declared_agent:
+ if not agent_secret:
+ raise PermissionError("agent_secret_required")
+ if not store.authenticate_agent(declared_agent, agent_secret, PERMISSIONS[typ]):
+ raise PermissionError("agent_not_authorized")
+ inserted = store.save_event(eid, hid, typ, utcnow().isoformat(), ok, p, declared_agent)
+ result = {"accepted": True, "duplicate": not inserted, "id": eid, "type": typ}
+ if not inserted:
+ return result
+ correlation = p.get("correlation_id") or p.get("run_id") or eid
+ if typ == "agent.heartbeat":
+ result["agent_id"] = declared_agent
+ result["heartbeat"] = True
+ return result
+ if typ in {"job.completed", "job.failed"}:
+ job = next(j for j in store.jobs() if j.id == p["job_id"])
+ status = "ok" if typ == "job.completed" else "failed"
+ now = utcnow()
+ store.save_action(Action(str(uuid.uuid4()), hid, now, declared_agent or "agent", "run_job", status, job.id, {"source": "event", "event_id": eid, "payload": p}, correlation))
+ job.last_run_at = now
+ job.last_status = status
+ job.last_error = p.get("error") if status == "failed" else None
+ job.failure_streak = job.failure_streak + 1 if status == "failed" else 0
+ store.save_job(job)
+ result.update(job_status=status, run_id=correlation)
+ elif typ == "job.started":
+ job = next(j for j in store.jobs() if j.id == p["job_id"])
+ job.last_status = "running"
+ job.last_run_at = utcnow()
+ store.save_job(job)
+ result.update(job_status="running", run_id=correlation)
+ else:
+ c = Claim.new(hid, p["claim"], p.get("job_id"), p.get("action"), p.get("expected_status", "ok"))
+ c.run_id = p.get("run_id") or p.get("correlation_id")
+ if c.job_id and not c.run_id:
+ c.status = "inconclusive"
+ c.evidence = {"source": "habitat", "reason": "run_id_required_for_network_claim"}
+ c.verified_at = utcnow()
+ store.save_claim(c)
+ else:
+ store.save_claim(c)
+ verify_claim(store, c, None, None, p.get("evidence_expected"))
+ result.update(claim_id=c.id, claim_status=c.status)
+ return result
diff --git a/habitat/generate.py b/habitat/generate.py
new file mode 100644
index 0000000..51f7c29
--- /dev/null
+++ b/habitat/generate.py
@@ -0,0 +1,115 @@
+"""Render Habitat's public landing page and native supervision surface."""
+from __future__ import annotations
+
+from html import escape
+from pathlib import Path
+
+from .claims import Claim
+from .schema import Action, Habitat, Job, Signal
+
+
+def _page(title: str, body: str) -> str:
+ return f"""
+
+
+
+
+
+{escape(title)}
+{body}"""
+
+
+def render_landing(title: str = "Habitat") -> str:
+ body = f"""Habitat
+
+
Local-first accountability for agents
+
Don't just log what an agent did. Check the evidence.
+
Habitat is a small supervision and evidence runtime for autonomous agents. It turns authenticated events into a verifiable local record, checks claims against that record, and produces portable proof another machine can inspect.
A layer for accountability, not another observability suite.
+
Keep your agent framework and existing telemetry. Habitat focuses on the narrower question that matters when an agent makes a claim: what evidence does the system actually have?
+
01 / RECORD
Capture trusted actions
Accept structured, authenticated events and correlate work with run IDs instead of inventing another tracing system.
+
02 / VERIFY
Test claims against evidence
Check exact action/run relationships and explicit external evidence without treating a valid record as proof of real-world truth.
+
03 / EXCHANGE
Export portable proof
Produce a proof artifact that can be verified away from the producer database, including in CI.
+
Habitat is deliberately small. It does not replace tracing, evaluation, orchestration, or human judgment. It gives those systems an accountability surface they can inspect.
+"""
+ return _page(f"{title} — Agent accountability", body)
+
+
+def _claim_evidence(claim: Claim) -> str:
+ evidence = claim.evidence if isinstance(claim.evidence, dict) else {}
+ if not evidence:
+ return "No recorded evidence"
+ parts = []
+ for key in ("source", "status", "action_id", "run_id", "adapter"):
+ value = evidence.get(key)
+ if value not in (None, ""):
+ parts.append(f"{escape(str(key))}: {escape(str(value))}")
+ return " · ".join(parts) if parts else "Recorded evidence"
+
+
+def render_dashboard(habitat: Habitat, jobs: list[Job], signals: list[Signal], actions: list[Action], title: str = "Habitat", claims: list[Claim] | None = None) -> str:
+ claims = claims or []
+ unresolved = [s for s in signals if s.resolved_at is None and s.severity in ("critical", "warning")]
+ unresolved.sort(key=lambda s: (0 if s.severity == "critical" else 1, s.detected_at))
+ enabled = sum(j.enabled for j in jobs)
+ ok = sum(j.last_status == "ok" for j in jobs)
+ failed = sum(j.last_status == "failed" for j in jobs)
+ attention = "".join(f'