"""Deterministic JSON CLI for inspection and explicit derived-output integration.""" from __future__ import annotations import argparse import json import sys import webbrowser from pathlib import Path from .application import CanonicalApplicationService, GenericCanonicalApplier from .client_config import CLIENT_NAMES, generate_client_configuration from .context import compile_context from .doctor import run_doctor from .errors import DocForgeError from .graph_rendering import GraphRenderService from .index import ProjectIndex from .onboarding import assess_project, scaffold_project from .project import Project, project_root_fingerprint from .projection_policy import compose_projection_policy from .rendering import RenderService from .telemetry import request from .viewer_manager import ViewerManagerClient def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="docforge") parser.add_argument("--project-root", type=Path) parser.add_argument( "--diagnostics", action="store_true", help="Attach bounded request-local stage timings and counters", ) parser.add_argument( "--manual-render-policy", choices=("auto", "explicit", "disabled"), ) parser.add_argument( "--portable-graph-policy", choices=("explicit", "disabled"), ) parser.add_argument( "--live-viewer-policy", choices=("on-demand", "disabled"), ) commands = parser.add_subparsers(dest="command", required=True) configure = commands.add_parser("configure") configure.add_argument("client", choices=CLIENT_NAMES) configure.add_argument("--project", type=Path, required=True) configure.add_argument("--name") configure.add_argument( "--capability-mode", choices=("read", "proposal", "application"), default="read", ) configure.add_argument("--proposal-writer") configure.add_argument("--canonical-applier") configure.add_argument("--no-ast", action="store_true") configure.add_argument( "--manual-render-policy", choices=("auto", "explicit", "disabled"), default=argparse.SUPPRESS, ) configure.add_argument( "--portable-graph-policy", choices=("explicit", "disabled"), default=argparse.SUPPRESS, ) configure.add_argument( "--live-viewer-policy", choices=("on-demand", "disabled"), default=argparse.SUPPRESS, ) configure.add_argument("--startup-timeout", type=int, default=30) configure.add_argument("--tool-timeout", type=int, default=300) configure.add_argument("--output", type=Path) doctor = commands.add_parser("doctor") doctor.add_argument("--client", choices=CLIENT_NAMES, required=True) doctor.add_argument("--project", type=Path) doctor.add_argument("--config", type=Path) doctor.add_argument("--server-name") onboard = commands.add_parser("onboard") onboard.add_argument("--language", action="append", default=[]) onboard.add_argument("--scaffold", action="store_true") onboard.add_argument("--project-id") onboard.add_argument("--title") onboard.add_argument("--content-root", default="docs/docforge/content") commands.add_parser("info") commands.add_parser("validate") commands.add_parser("build") commands.add_parser("reindex") commands.add_parser("sync") commands.add_parser("check") commands.add_parser("validate-index") show = commands.add_parser("show") show.add_argument("node_id") search = commands.add_parser("search") search.add_argument("query") search.add_argument("--limit", type=int) filter_command = commands.add_parser("filter") filter_command.add_argument("--family") filter_command.add_argument("--authority") filter_command.add_argument("--status") filter_command.add_argument("--tag") filter_command.add_argument("--limit", type=int) for name in ("backlinks", "dependencies", "impact"): command = commands.add_parser(name) command.add_argument("node_id") if name == "backlinks": command.add_argument("--relation") else: command.add_argument("--depth", type=int, default=2) command.add_argument("--limit", type=int) context = commands.add_parser("context") context.add_argument("profile") context.add_argument("--budget", type=int) context.add_argument("--limit", type=int) context.add_argument("--cursor") generation_diff = commands.add_parser("generation-diff") generation_diff.add_argument("--limit", type=int) generation_diff.add_argument("--cursor") render = commands.add_parser("render") render.add_argument("view_id") render_status = commands.add_parser("render-status") render_status.add_argument("view_id", nargs="?") render_status.add_argument("--deep", action="store_true") graph_plan = commands.add_parser("graph-plan") graph_plan.add_argument("view_id") graph_render = commands.add_parser("graph-render") graph_render.add_argument("view_id") graph_render_status = commands.add_parser("graph-render-status") graph_render_status.add_argument("view_id", nargs="?") preview = commands.add_parser("preview") preview.add_argument("changeset_id") preview.add_argument("view_id") apply_command = commands.add_parser("apply") apply_command.add_argument("changeset_id") apply_command.add_argument("--changeset-hash", required=True) apply_command.add_argument("--applier", required=True) visualize = commands.add_parser("visualize") target = visualize.add_mutually_exclusive_group() target.add_argument("--node") target.add_argument("--query") visualize.add_argument("--depth", type=int, default=1) visualize.add_argument("--no-open", action="store_true") commands.add_parser("visualization-status") commands.add_parser("visualization-stop") return parser def _run(arguments: argparse.Namespace) -> dict[str, object]: if arguments.command == "configure": project = Project.open(arguments.project) return generate_client_configuration( project, arguments.client, server_name=arguments.name, capability_mode=arguments.capability_mode, proposal_writer=arguments.proposal_writer, canonical_applier=arguments.canonical_applier, no_ast=arguments.no_ast, manual_render_policy=arguments.manual_render_policy, portable_graph_policy=arguments.portable_graph_policy, live_viewer_policy=arguments.live_viewer_policy, startup_timeout=arguments.startup_timeout, tool_timeout=arguments.tool_timeout, output=arguments.output, ) if arguments.command == "doctor": root = arguments.project or arguments.project_root or Path.cwd() return run_doctor( Project.open(root), arguments.client, config_path=arguments.config, server_name=arguments.server_name, ) if arguments.project_root is None: raise DocForgeError( "missing_project_root", "This command requires --project-root", ) if arguments.command == "onboard": languages = tuple(arguments.language) if arguments.scaffold: scaffold = scaffold_project( arguments.project_root, requested_languages=languages, project_id=arguments.project_id, title=arguments.title, content_root=arguments.content_root, ) project = Project.open(arguments.project_root) projection_policy = compose_projection_policy( manual=arguments.manual_render_policy, portable_graph=arguments.portable_graph_policy, live_viewer=arguments.live_viewer_policy, manual_configured=project.descriptor.render is not None, portable_graph_configured=project.descriptor.graph_render is not None, application_enabled=False, ) build = ProjectIndex(project).build() render = ( { "status": "ok", "state": "skipped", "reason": "projection_policy_disabled", } if projection_policy.manual == "disabled" else RenderService( project, manual_policy=projection_policy.manual, ).render("manual") ) return {**scaffold, "build": build, "render": render} return assess_project(arguments.project_root, requested_languages=languages) project = Project.open(arguments.project_root) index = ProjectIndex(project) projection_policy = compose_projection_policy( manual=arguments.manual_render_policy, portable_graph=arguments.portable_graph_policy, live_viewer=arguments.live_viewer_policy, manual_configured=project.descriptor.render is not None, portable_graph_configured=project.descriptor.graph_render is not None, application_enabled=arguments.command == "apply", ) if arguments.command == "info": snapshot = project.load() return { "status": "ok", "project_id": snapshot.descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root), "title": snapshot.descriptor.title, "adapter": snapshot.descriptor.adapter, "revision": snapshot.revision, "source_hash": snapshot.source_hash, "node_count": len(snapshot.nodes), "edge_count": len(snapshot.edges), "index": str(snapshot.descriptor.index_path), } if arguments.command == "validate": snapshot = project.load() return { "status": "ok", "project_id": snapshot.descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(snapshot.descriptor.root), "revision": snapshot.revision, "source_hash": snapshot.source_hash, "node_count": len(snapshot.nodes), "edge_count": len(snapshot.edges), } if arguments.command == "build": return index.build() if arguments.command == "reindex": built = index.build() return { **built, "reindexed": True, "check": index.check(), } if arguments.command == "sync": return index.synchronize() if arguments.command == "check": return index.check() if arguments.command == "validate-index": return index.check() if arguments.command == "show": return index.get_node(arguments.node_id) if arguments.command == "search": return index.search(arguments.query, limit=arguments.limit) if arguments.command == "filter": return index.filter_nodes( family=arguments.family, authority=arguments.authority, status=arguments.status, tag=arguments.tag, limit=arguments.limit, ) if arguments.command == "backlinks": return index.backlinks( arguments.node_id, relation=arguments.relation, limit=arguments.limit, ) if arguments.command == "dependencies": return index.dependencies( arguments.node_id, depth=arguments.depth, limit=arguments.limit, ) if arguments.command == "impact": return index.impact( arguments.node_id, depth=arguments.depth, limit=arguments.limit, ) if arguments.command == "context": if arguments.limit is not None or arguments.cursor is not None: from .mcp_server import DocForgeService return DocForgeService(project).context( arguments.profile, arguments.budget, limit=arguments.limit, cursor=arguments.cursor, ) return compile_context(index, arguments.profile, arguments.budget) if arguments.command == "generation-diff": from .mcp_server import DocForgeService return DocForgeService( project, capability_mode_name="read", ).generation_diff( limit=arguments.limit, cursor=arguments.cursor, ) if arguments.command == "render": return RenderService( project, manual_policy=projection_policy.manual, ).render(arguments.view_id) if arguments.command == "render-status": rendering = RenderService( project, manual_policy=projection_policy.manual, ) return ( rendering.deep_status(arguments.view_id) if arguments.deep else rendering.status(arguments.view_id) ) if arguments.command == "graph-plan": return GraphRenderService( project, portable_graph_policy=projection_policy.portable_graph, ).plan(arguments.view_id) if arguments.command == "graph-render": return GraphRenderService( project, portable_graph_policy=projection_policy.portable_graph, ).render(arguments.view_id) if arguments.command == "graph-render-status": return GraphRenderService( project, portable_graph_policy=projection_policy.portable_graph, ).status(arguments.view_id) if arguments.command == "preview": return RenderService( project, manual_policy=projection_policy.manual, ).preview(arguments.changeset_id, arguments.view_id) if arguments.command == "apply": return CanonicalApplicationService( project, applier_id=arguments.applier, applier=GenericCanonicalApplier(project), manual_policy=projection_policy.manual, ).apply(arguments.changeset_id, arguments.changeset_hash) if arguments.command == "visualize": visualization = ViewerManagerClient( index, live_viewer_policy=projection_policy.live_viewer, ).start( node_id=arguments.node, query=arguments.query, depth=arguments.depth, ) opened = False if not arguments.no_open: opened = webbrowser.open(str(visualization["url"])) return { "status": "ok", "project_id": project.descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(project.descriptor.root), "adapter": project.descriptor.adapter, "opened_browser": opened, "visualization": visualization, } if arguments.command == "visualization-status": return ViewerManagerClient( index, live_viewer_policy=projection_policy.live_viewer, ).status() if arguments.command == "visualization-stop": return ViewerManagerClient( index, live_viewer_policy=projection_policy.live_viewer, ).stop() raise DocForgeError("invalid_command", "Unknown command") def main(argv: list[str] | None = None) -> int: parser = _parser() arguments = parser.parse_args(argv) with request( f"cli.{arguments.command}", enabled=arguments.diagnostics, ) as collector: try: result = _run(arguments) doctor_state = result.get("doctor_state") code = 2 if doctor_state == "unhealthy" else (1 if doctor_state == "degraded" else 0) except DocForgeError as error: result = {"status": "error", "error": error.as_dict()} code = 2 if collector is not None: result["diagnostics"] = collector.as_dict( outcome="ok" if result.get("status") == "ok" else "error", ) print(json.dumps(result, sort_keys=True, indent=2)) return code if __name__ == "__main__": sys.exit(main())