# Cli

- Object ID: `em:implementation:sha256:f9ff59ed09843a962929e421fcbd553563cb81f0a1d03f490cdfa314d284ae4b`
- Kind: `implementation`
- Repository path: [`src/epistemedia/cli.py`](https://github.com/yoheinakajima/epistemedia/blob/f92846570180dfa4511263f8ba98ecd18f7772c9/src/epistemedia/cli.py)
- Content digest: `24c4bbdface6cbd80f19e0a21924240459c3855885885dc3ba75dd8d353cdd15`

**Also filed under:** [Human and Agent Interfaces](https://epistemedia.org/topics/public-interfaces/)

## Source content

from __future__ import annotations

import argparse
import datetime as dt
import http.server
import json
import os
import socketserver
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

from .case_library import load_featured_library
from .core import (
    DEFAULT_API_URL,
    DEFAULT_BASE_URL,
    DEFAULT_MCP_URL,
    VERSION,
    PublicCatalog,
    audit_public,
    build_public,
    discover_root,
    envelope,
    validate_repository,
)
from .featured import FEATURE_VIEWS
from .mission import load_mission
from .open_dockets import prepare_submission
from .research_kit import (
    case_research_brief,
    proposal_template,
    protocol_document,
    validate_proposal,
)
from .server import Gateway, MCPRequestError


def utc_now() -> str:
    return (
        dt.datetime.now(dt.UTC)
        .replace(microsecond=0)
        .isoformat()
        .replace("+00:00", "Z")
    )


def parser() -> argparse.ArgumentParser:
    root = argparse.ArgumentParser(prog="epistemedia", description="Knowledge that can show its work.")
    root.add_argument("--version", action="version", version=f"epistemedia {VERSION}")
    root.add_argument("--root", type=Path, help="Repository root; inferred for local commands")
    sub = root.add_subparsers(dest="command", required=True)

    sub.add_parser("orient", help="Print the bounded agent orientation and current repository state")
    sub.add_parser("validate", help="Validate accepted repository inputs")

    build = sub.add_parser("build", help="Compile all public human and agent interfaces")
    build.add_argument("--output", type=Path, default=Path("generated/public"))
    build.add_argument("--base-url", default=DEFAULT_BASE_URL)
    build.add_argument("--api-url", default=DEFAULT_API_URL)
    build.add_argument("--mcp-url", default=DEFAULT_MCP_URL)

    audit = sub.add_parser("audit", help="Audit a compiled PublicProjection")
    audit.add_argument("--public", type=Path, default=Path("generated/public"))

    serve = sub.add_parser("serve", help="Serve compiled static output")
    serve.add_argument("--public", type=Path, default=Path("generated/public"))
    serve.add_argument("--host", default="127.0.0.1")
    serve.add_argument("--port", type=int, default=8000)

    api = sub.add_parser("api-serve", help="Serve API and MCP through an ASGI server")
    api.add_argument("--host", default="127.0.0.1")
    api.add_argument("--port", type=int, default=8080)

    search = sub.add_parser("search", help="Search a local realm or the public remote API")
    search.add_argument("query")
    search.add_argument("--limit", type=int, default=20)
    search.add_argument("--remote", action="store_true")
    search.add_argument("--api", default=DEFAULT_API_URL)

    get = sub.add_parser("get", help="Get an exact object")
    get.add_argument("id")
    get.add_argument("--remote", action="store_true")
    get.add_argument("--api", default=DEFAULT_API_URL)

    mission = sub.add_parser("mission", help="Get the versioned public mission")
    mission.add_argument("--remote", action="store_true")
    mission.add_argument("--api", default=DEFAULT_API_URL)

    project = sub.add_parser("project", help="Compile or retrieve a topic projection")
    project.add_argument("slug")
    project.add_argument("--lens", default="encyclopedia")
    project.add_argument("--remote", action="store_true")
    project.add_argument("--api", default=DEFAULT_API_URL)

    dossier = sub.add_parser("dossier", help="Get an accepted How We Know dossier")
    dossier.add_argument("slug")
    dossier.add_argument("--policy", choices=FEATURE_VIEWS, default="encyclopedia")
    dossier.add_argument("--remote", action="store_true")
    dossier.add_argument("--api", default=DEFAULT_API_URL)

    open_dockets = sub.add_parser("open-dockets", help="List or get reviewed open dockets")
    open_docket_sub = open_dockets.add_subparsers(dest="open_docket_command", required=True)
    open_docket_list = open_docket_sub.add_parser("list", help="List reviewed open dockets")
    open_docket_list.add_argument("--remote", action="store_true")
    open_docket_list.add_argument("--api", default=DEFAULT_API_URL)
    open_docket_get = open_docket_sub.add_parser("get", help="Get one reviewed open docket")
    open_docket_get.add_argument("slug")
    open_docket_get.add_argument("--remote", action="store_true")
    open_docket_get.add_argument("--api", default=DEFAULT_API_URL)

    research = sub.add_parser("research", help="Prepare and validate non-admitting research proposals")
    research_sub = research.add_subparsers(dest="research_command", required=True)
    research_sub.add_parser("protocol", help="Print the public agent research protocol")
    research_sub.add_parser("submission-guide", help="Print the autonomous GitHub submission guide")
    prepare = research_sub.add_parser("prepare", help="Create a deterministic draft proposal scaffold")
    prepare.add_argument("--question")
    prepare.add_argument("--case", dest="case_slug")
    prepare.add_argument("--cutoff", default="YYYY-MM-DD")
    prepare.add_argument("--output", type=Path)
    complete_research = research_sub.add_parser(
        "complete", help="Stamp completion time and fail-closed validate one proposal JSON file"
    )
    complete_research.add_argument("bundle", type=Path)
    validate_research = research_sub.add_parser("validate", help="Fail-closed validation of one proposal JSON file")
    validate_research.add_argument("bundle", type=Path)
    submit_research = research_sub.add_parser(
        "submit", help="Prepare a deterministic GitHub draft-PR submission directory"
    )
    submit_research.add_argument("bundle", type=Path)
    submit_research.add_argument("--trace", required=True, type=Path)
    submit_research.add_argument("--agent-id", required=True)
    submit_research.add_argument("--model-family", required=True)
    submit_research.add_argument("--run-id", required=True)
    submit_research.add_argument("--prompt-sha256", required=True)

    repo = sub.add_parser("repo", help="Repository-native agent operations")
    repo_sub = repo.add_subparsers(dest="repo_command", required=True)
    repo_sub.add_parser("next", help="Show the best currently discoverable task contracts")
    claim = repo_sub.add_parser("claim", help="Create an append-only local task-claim proposal")
    claim.add_argument("task_id")
    claim.add_argument("--agent", required=True)
    receipt = repo_sub.add_parser("receipt", help="Record a local append-only run receipt proposal")
    receipt.add_argument("task_id")
    receipt.add_argument("--run", required=True)
    receipt.add_argument("--command", dest="run_command", required=True)

    mcp = sub.add_parser("mcp", help="MCP utilities")
    mcp_sub = mcp.add_subparsers(dest="mcp_command", required=True)
    mcp_sub.add_parser("serve", help="Run a newline-delimited stdio MCP adapter")
    mcp_sub.add_parser("discover", help="Print server discovery metadata")

    return root


def resolve_root(args: argparse.Namespace, required: bool = True) -> Path | None:
    if args.root:
        return args.root.resolve()
    try:
        return discover_root()
    except FileNotFoundError:
        if required:
            raise
        return None


def print_json(value: Any) -> None:
    print(json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False))


def remote_get(api: str, path: str, query: dict[str, Any] | None = None) -> Any:
    url = api.rstrip("/") + "/" + path.lstrip("/")
    if query:
        url += "?" + urllib.parse.urlencode(query)
    request = urllib.request.Request(url, headers={"accept": "application/json", "user-agent": f"epistemedia-cli/{VERSION}"})
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return json.load(response)
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        raise SystemExit(f"remote API returned {exc.code}: {detail}") from exc
    except urllib.error.URLError as exc:
        raise SystemExit(f"remote API unavailable: {exc.reason}") from exc


def orient(root: Path) -> int:
    catalog = PublicCatalog.build(root)
    tasks = [obj for obj in catalog.objects if obj.kind == "task"]
    print("Epistemedia agent orientation")
    print("============================")
    print(f"Repository: {root}")
    print(f"Accepted commit: {catalog.commit}")
    print(f"Public catalog: {catalog.catalog_id}")
    print(f"Evidence frontier: {catalog.frontier}")
    print(f"Public objects: {len(catalog.objects)}")
    print(f"Topics: {len(catalog.topics)}")
    print(f"Task contracts visible: {len(tasks)}")
    print()
    print("Read order: AGENTS.md → selected task contract → execution plan → schemas/policies/tests.")
    print("Rule: propose changes through Git; do not become the source of truth or your own approver.")
    return 0


def next_tasks(root: Path) -> int:
    catalog = PublicCatalog.build(root)
    tasks = [obj for obj in catalog.objects if obj.kind == "task"]
    if not tasks:
        print("No public task contracts are currently discoverable.")
        return 0
    print_json({
        "catalog_id": catalog.catalog_id,
        "frontier": catalog.frontier,
        "tasks": [obj.as_dict(include_text=False) for obj in tasks[:20]],
        "instruction": "Read the immutable task contract and AGENTS.md before claiming work.",
    })
    return 0


def append_local_proposal(root: Path, kind: str, payload: dict[str, Any]) -> Path:
    from .core import digest, utc_now

    timestamp = utc_now()
    record = {
        "schema": f"https://epistemedia.com/schemas/{kind}-v1.json",
        "kind": kind,
        "recorded_at": timestamp,
        **payload,
    }
    record["id"] = f"em:{kind}:sha256:{digest(record)}"
    safe_time = timestamp.replace(":", "").replace("-", "")
    path = root / "runs" / "proposals" / f"{safe_time}-{record['id'].split(':')[-1][:12]}.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n")
    return path


def stdio_mcp(root: Path) -> int:
    gateway = Gateway(root)
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        request: Any = None
        try:
            request = json.loads(line)
            if isinstance(request, dict) and "id" not in request:
                if request.get("method") == "notifications/cancelled":
                    gateway.mcp_method(
                        "notifications/cancelled", request.get("params") or {}
                    )
                continue
            request_id, method, params = gateway.validate_mcp_request(request)
            result = gateway.mcp_method(method, params)
            if isinstance(result, dict):
                result = gateway.decorate_mcp_result(result)
            print(json.dumps({"jsonrpc": "2.0", "id": request_id, "result": result}, sort_keys=True), flush=True)
        except json.JSONDecodeError:
            print(json.dumps(gateway.rpc_error(None, -32700, "Parse error"), sort_keys=True), flush=True)
        except MCPRequestError as exc:
            request_id = request.get("id") if isinstance(request, dict) else None
            print(
                json.dumps(
                    gateway.rpc_error(request_id, exc.code, str(exc), exc.data),
                    sort_keys=True,
                ),
                flush=True,
            )
        except KeyError as exc:
            request_id = request.get("id") if isinstance(request, dict) else None
            print(
                json.dumps(
                    gateway.rpc_error(
                        request_id,
                        -32004,
                        "Not found",
                        {"id": exc.args[0] if exc.args else ""},
                    ),
                    sort_keys=True,
                ),
                flush=True,
            )
        except ValueError as exc:
            request_id = request.get("id") if isinstance(request, dict) else None
            print(
                json.dumps(
                    gateway.rpc_error(request_id, -32602, str(exc)),
                    sort_keys=True,
                ),
                flush=True,
            )
        except Exception:
            request_id = request.get("id") if isinstance(request, dict) else None
            print(
                json.dumps(
                    gateway.rpc_error(request_id, -32603, "Internal error"),
                    sort_keys=True,
                ),
                flush=True,
            )
    return 0


def main(argv: list[str] | None = None) -> int:
    args = parser().parse_args(argv)
    command = args.command

    if command == "search" and args.remote:
        print_json(remote_get(args.api, "search", {"q": args.query, "limit": args.limit}))
        return 0
    if command == "get" and args.remote:
        print_json(remote_get(args.api, "objects/" + urllib.parse.quote(args.id, safe="")))
        return 0
    if command == "mission" and args.remote:
        print_json(remote_get(args.api, "mission"))
        return 0
    if command == "project" and args.remote:
        print_json(remote_get(args.api, "topics/" + urllib.parse.quote(args.slug, safe=""), {"lens": args.lens}))
        return 0
    if command == "dossier" and args.remote:
        print_json(
            remote_get(
                args.api,
                "dossiers/" + urllib.parse.quote(args.slug, safe=""),
                {"policy": args.policy},
            )
        )
        return 0
    if command == "open-dockets" and args.remote:
        path = "open-dockets"
        if args.open_docket_command == "get":
            path += "/" + urllib.parse.quote(args.slug, safe="")
        print_json(remote_get(args.api, path))
        return 0

    root = resolve_root(args)
    assert root is not None

    if command == "orient":
        return orient(root)
    if command == "validate":
        errors = validate_repository(root)
        if errors:
            print_json({"valid": False, "errors": errors})
            return 1
        catalog = PublicCatalog.build(root)
        print_json({"valid": True, "catalog_id": catalog.catalog_id, "frontier": catalog.frontier, "objects": len(catalog.objects), "topics": len(catalog.topics)})
        return 0
    if command == "build":
        output = args.output if args.output.is_absolute() else root / args.output
        manifest = build_public(root, output, base_url=args.base_url, api_url=args.api_url, mcp_url=args.mcp_url)
        print_json(manifest)
        return 0
    if command == "audit":
        public = args.public if args.public.is_absolute() else root / args.public
        findings = audit_public(root, public)
        print_json({"ok": not findings, "findings": findings})
        return 1 if findings else 0
    if command == "serve":
        public = args.public if args.public.is_absolute() else root / args.public
        if not (public / "index.html").exists():
            build_public(root, public)
        os.chdir(public)
        handler = http.server.SimpleHTTPRequestHandler
        with socketserver.ThreadingTCPServer((args.host, args.port), handler) as server:
            print(f"Serving {public} at http://{args.host}:{args.port}")
            server.serve_forever()
        return 0
    if command == "api-serve":
        try:
            import uvicorn  # type: ignore
        except ImportError as exc:
            raise SystemExit("Install the server extra: pip install 'epistemedia[server]'") from exc
        os.environ["EPISTEMEDIA_ROOT"] = str(root)
        uvicorn.run("epistemedia.server:app", host=args.host, port=args.port, reload=False)
        return 0
    if command == "search":
        catalog = PublicCatalog.build(root)
        print_json({"catalog_id": catalog.catalog_id, "frontier": catalog.frontier, "query": args.query, "results": catalog.search(args.query, args.limit)})
        return 0
    if command == "get":
        catalog = PublicCatalog.build(root)
        obj = catalog.object_map().get(args.id)
        if not obj:
            raise SystemExit(f"unknown object: {args.id}")
        print_json({"catalog_id": catalog.catalog_id, "frontier": catalog.frontier, "data": obj.as_dict()})
        return 0
    if command == "mission":
        catalog = PublicCatalog.build(root)
        print_json(envelope(catalog, load_mission(root)))
        return 0
    if command == "project":
        from .core import topic_projection

        catalog = PublicCatalog.build(root)
        topic = catalog.topic_map().get(args.slug)
        if not topic:
            raise SystemExit(f"unknown topic: {args.slug}")
        print_json(topic_projection(catalog, topic, args.lens, DEFAULT_BASE_URL))
        return 0
    if command == "dossier":
        catalog = PublicCatalog.build(root)
        library = load_featured_library(root)
        try:
            dossier = library.get(args.slug) if library is not None else None
        except KeyError:
            dossier = None
        if dossier is None:
            raise SystemExit(f"unknown dossier: {args.slug}")
        print_json(envelope(catalog, dossier.projection(args.policy)))
        return 0
    if command == "open-dockets":
        from .open_dockets import load_open_dockets

        catalog = PublicCatalog.build(root)
        dockets, errors = load_open_dockets(root)
        if errors:
            raise SystemExit("; ".join(errors))
        if args.open_docket_command == "list":
            print_json(
                envelope(
                    catalog,
                    [docket.projection(DEFAULT_BASE_URL) for docket in dockets],
                )
            )
            return 0
        docket = next((item for item in dockets if item.slug == args.slug), None)
        if docket is None:
            raise SystemExit(f"unknown open docket: {args.slug}")
        print_json(envelope(catalog, docket.projection(DEFAULT_BASE_URL)))
        return 0
    if command == "research":
        if args.research_command == "protocol":
            print_json(protocol_document(DEFAULT_BASE_URL))
            return 0
        if args.research_command == "submission-guide":
            from .open_dockets import submission_guide

            print_json(submission_guide(DEFAULT_BASE_URL))
            return 0
        if args.research_command == "prepare":
            question = args.question
            brief = None
            if args.case_slug:
                library = load_featured_library(root)
                try:
                    selected = library.get(args.case_slug) if library is not None else None
                except KeyError:
                    selected = None
                if selected is None:
                    raise SystemExit(f"unknown dossier: {args.case_slug}")
                projection = selected.projection(selected.default_view)
                brief = case_research_brief(projection, DEFAULT_BASE_URL)
                question = question or projection["question"]
            if not isinstance(question, str) or not question.strip():
                raise SystemExit("research prepare requires --question or --case")
            proposal = proposal_template(
                question.strip(), cutoff=args.cutoff, case_slug=args.case_slug
            )
            proposal["runtime"]["started_at"] = utc_now()
            output = {"proposal": proposal, "case_brief": brief}
            if args.output:
                destination = args.output if args.output.is_absolute() else root / args.output
                destination.parent.mkdir(parents=True, exist_ok=True)
                destination.write_text(
                    json.dumps(proposal, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
                    encoding="utf-8",
                )
                print_json({"output": str(destination), **validate_proposal(proposal)})
            else:
                print_json(output)
            return 0
        if args.research_command == "complete":
            bundle_path = args.bundle if args.bundle.is_absolute() else root / args.bundle
            try:
                bundle = json.loads(bundle_path.read_text(encoding="utf-8"))
                runtime = bundle.get("runtime")
                if not isinstance(runtime, dict):
                    raise ValueError("proposal runtime must be an object")
                runtime["completed_at"] = utc_now()
                bundle_path.write_text(
                    json.dumps(bundle, indent=2, sort_keys=True, ensure_ascii=False) + "\n",
                    encoding="utf-8",
                )
            except (OSError, json.JSONDecodeError, ValueError) as exc:
                print_json(
                    {
                        "valid": False,
                        "errors": [str(exc)],
                        "submitted": False,
                        "admitted": False,
                    }
                )
                return 1
            result = validate_proposal(bundle)
            print_json({**result, "output": str(bundle_path)})
            return 0 if result["valid"] else 1
        if args.research_command == "validate":
            bundle_path = args.bundle if args.bundle.is_absolute() else root / args.bundle
            try:
                bundle = json.loads(bundle_path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError) as exc:
                print_json({"valid": False, "errors": [str(exc)], "submitted": False, "admitted": False})
                return 1
            result = validate_proposal(bundle)
            print_json(result)
            return 0 if result["valid"] else 1
        if args.research_command == "submit":
            bundle_path = args.bundle if args.bundle.is_absolute() else root / args.bundle
            trace_path = args.trace if args.trace.is_absolute() else root / args.trace
            try:
                bundle = json.loads(bundle_path.read_text(encoding="utf-8"))
                trace = json.loads(trace_path.read_text(encoding="utf-8"))
                runtime = bundle.get("runtime")
                if not isinstance(runtime, dict):
                    raise ValueError("proposal runtime must be an object")
                runtime["completed_at"] = utc_now()
                prepared = prepare_submission(
                    root,
                    bundle,
                    trace,
                    agent_id=args.agent_id,
                    model_family=args.model_family,
                    run_id=args.run_id,
                    prompt_sha256=args.prompt_sha256,
                )
            except (OSError, json.JSONDecodeError, ValueError) as exc:
                print_json(
                    {
                        "valid": False,
                        "errors": [str(exc)],
                        "submitted": False,
                        "admitted": False,
                    }
                )
                return 1
            directory = prepared["directory"]
            print_json(
                {
                    "valid": True,
                    "submitted": False,
                    "admitted": False,
                    "slug": prepared["slug"],
                    "proposal_id": prepared["proposal_id"],
                    "proposal_sha256": prepared["proposal_sha256"],
                    "directory": str(directory.relative_to(root)),
                    "next_steps": [
                        f"git switch -c submission/{prepared['slug']}",
                        f"git add {directory.relative_to(root)}",
                        f"git commit -m 'research: submit open docket {prepared['slug']}'",
                        f"git push -u origin submission/{prepared['slug']}",
                        (
                            "gh pr create --draft "
                            f"--title {json.dumps(prepared['pull_request_title'])} "
                            f"--body-file {prepared['pull_request_body'].relative_to(root)}"
                        ),
                    ],
                    "note": "Stop after opening the draft PR; do not review, merge, or publish it.",
                }
            )
            return 0
    if command == "repo":
        if args.repo_command == "next":
            return next_tasks(root)
        if args.repo_command == "claim":
            path = append_local_proposal(root, "task-claim", {"task_id": args.task_id, "agent_id": args.agent, "status": "proposed"})
            print(path.relative_to(root))
            return 0
        if args.repo_command == "receipt":
            path = append_local_proposal(root, "run-receipt", {"task_id": args.task_id, "run_id": args.run, "command": args.run_command, "status": "reported"})
            print(path.relative_to(root))
            return 0
    if command == "mcp":
        if args.mcp_command == "serve":
            return stdio_mcp(root)
        if args.mcp_command == "discover":
            print_json(Gateway(root).mcp_method("server/discover", {}))
            return 0
    raise SystemExit(f"unhandled command: {command}")


if __name__ == "__main__":
    raise SystemExit(main())