Cli

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())