227 lines
9.8 KiB
Python
227 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
|
|
from claridoc import __version__
|
|
from claridoc.corpus import (
|
|
DEFAULT_INCLUDES,
|
|
build_query_from_brief,
|
|
collect_sources,
|
|
merge_source_packs,
|
|
)
|
|
from claridoc.lint import lint_document, render_lint_markdown
|
|
from claridoc.models import Brief, PipelineConfig, SourcePack, ValidationError
|
|
from claridoc.pipeline import PipelineExecutionError, run_pipeline
|
|
from claridoc.providers import ProviderError, create_provider
|
|
from claridoc.structures import create_outline
|
|
from claridoc.templates import mock_pipeline_config, starter_brief, starter_sources
|
|
from claridoc.utils import read_json, write_json
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
prog="claridoc",
|
|
description="Evidence-aware, multi-agent harness for reader-facing technical documentation.",
|
|
)
|
|
parser.add_argument("--version", action="version", version=f"claridoc {__version__}")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
init = sub.add_parser("init", help="Create starter brief, source pack, and pipeline configs.")
|
|
init.add_argument("directory", nargs="?", default="claridoc-workspace")
|
|
init.add_argument("--force", action="store_true")
|
|
|
|
validate = sub.add_parser("validate", help="Validate a brief and its evidence inputs.")
|
|
validate.add_argument("--brief", required=True)
|
|
_add_source_options(validate)
|
|
|
|
outline = sub.add_parser("outline", help="Generate the deterministic document-type outline contract.")
|
|
outline.add_argument("--brief", required=True)
|
|
_add_source_options(outline)
|
|
outline.add_argument("--output")
|
|
|
|
lint = sub.add_parser("lint", help="Lint an existing Markdown document against a brief.")
|
|
lint.add_argument("document")
|
|
lint.add_argument("--brief", required=True)
|
|
_add_source_options(lint)
|
|
lint.add_argument("--output")
|
|
lint.add_argument("--json", action="store_true", dest="as_json")
|
|
|
|
run = sub.add_parser("run", help="Run plan, draft, review, revise, and quality-gate stages.")
|
|
run.add_argument("--brief", required=True)
|
|
_add_source_options(run)
|
|
run.add_argument("--config", help="Pipeline JSON. Defaults to an offline mock pipeline.")
|
|
run.add_argument("--output", required=True)
|
|
|
|
collect = sub.add_parser(
|
|
"collect",
|
|
help="Search a local documentation repository and build an internal evidence pack.",
|
|
)
|
|
collect.add_argument("--root", required=True)
|
|
collect.add_argument("--query", action="append", required=True, help="Retrieval query; may be repeated.")
|
|
collect.add_argument("--include", action="append", dest="includes")
|
|
collect.add_argument("--top-k", type=int, default=24)
|
|
collect.add_argument("--max-per-file", type=int, default=3)
|
|
collect.add_argument("--output", required=True)
|
|
|
|
doctor = sub.add_parser("doctor", help="Check provider binaries or SDKs referenced by a pipeline config.")
|
|
doctor.add_argument("--config", required=True)
|
|
doctor.add_argument("--json", action="store_true", dest="as_json")
|
|
return parser
|
|
|
|
|
|
def _add_source_options(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--sources", help="Existing source-pack JSON.")
|
|
parser.add_argument(
|
|
"--source-root",
|
|
help="Local documentation repository to search before planning and drafting.",
|
|
)
|
|
parser.add_argument(
|
|
"--source-include",
|
|
action="append",
|
|
dest="source_includes",
|
|
help=(
|
|
"Repository-relative directory to scan; may be repeated. Defaults to "
|
|
+ ", ".join(DEFAULT_INCLUDES)
|
|
),
|
|
)
|
|
parser.add_argument("--source-top-k", type=int, default=24)
|
|
parser.add_argument("--source-max-per-file", type=int, default=3)
|
|
|
|
|
|
def main(argv: Sequence[str] | None = None) -> int:
|
|
parser = build_parser()
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
if args.command == "init":
|
|
return _cmd_init(Path(args.directory), args.force)
|
|
if args.command == "collect":
|
|
sources = collect_sources(
|
|
args.root,
|
|
"\n".join(args.query),
|
|
includes=args.includes,
|
|
top_k=args.top_k,
|
|
max_per_file=args.max_per_file,
|
|
)
|
|
write_json(args.output, sources.to_dict())
|
|
print(f"WROTE: {Path(args.output).resolve()} ({len(sources.sources)} evidence chunks)")
|
|
return 0
|
|
if args.command == "validate":
|
|
brief, sources = _load_contracts_from_args(args)
|
|
print(f"VALID: {brief.title} ({brief.document_type.value}), {len(sources.sources)} sources")
|
|
return 0
|
|
if args.command == "outline":
|
|
brief, sources = _load_contracts_from_args(args)
|
|
data = create_outline(brief, sources).to_dict()
|
|
if args.output:
|
|
write_json(args.output, data)
|
|
print(f"WROTE: {Path(args.output).resolve()}")
|
|
else:
|
|
print(json.dumps(data, ensure_ascii=False, indent=2))
|
|
return 0
|
|
if args.command == "lint":
|
|
brief, sources = _load_contracts_from_args(args)
|
|
text = Path(args.document).read_text(encoding="utf-8")
|
|
report = lint_document(text, brief, create_outline(brief, sources), sources)
|
|
rendered = (
|
|
json.dumps(report.to_dict(), ensure_ascii=False, indent=2)
|
|
if args.as_json
|
|
else render_lint_markdown(report)
|
|
)
|
|
if args.output:
|
|
Path(args.output).parent.mkdir(parents=True, exist_ok=True)
|
|
Path(args.output).write_text(
|
|
rendered + ("\n" if not rendered.endswith("\n") else ""),
|
|
encoding="utf-8",
|
|
)
|
|
print(f"WROTE: {Path(args.output).resolve()}")
|
|
else:
|
|
print(rendered)
|
|
return 0 if not any(issue.severity.value in {"blocker", "error"} for issue in report.issues) else 4
|
|
if args.command == "run":
|
|
brief, sources = _load_contracts_from_args(args)
|
|
config_data = read_json(args.config) if args.config else mock_pipeline_config()
|
|
config = PipelineConfig.from_dict(config_data)
|
|
result = run_pipeline(brief, sources, config, args.output)
|
|
print(f"GATE: {'PASS' if result.passed else 'FAIL'}")
|
|
print(f"SCORE: {result.final_score:.1f}/100")
|
|
print(f"DOCUMENT: {result.final_path}")
|
|
print(f"REPORT: {result.report_path}")
|
|
print(f"PROVENANCE: {result.output_dir / 'final' / 'provenance.md'}")
|
|
return 0 if result.passed else 4
|
|
if args.command == "doctor":
|
|
config = PipelineConfig.from_dict(read_json(args.config))
|
|
checks = _provider_checks(config)
|
|
if args.as_json:
|
|
print(json.dumps(checks, ensure_ascii=False, indent=2))
|
|
else:
|
|
for check in checks:
|
|
status = "OK" if check.get("available") else "MISSING"
|
|
print(
|
|
f"[{status}] {check.get('provider')}: {check.get('mode')} — "
|
|
f"{check.get('executable', check.get('note', ''))}"
|
|
)
|
|
return 0 if all(item.get("available") for item in checks) else 3
|
|
except (ValidationError, json.JSONDecodeError) as exc:
|
|
print(f"CONTRACT ERROR: {exc}", file=sys.stderr)
|
|
return 2
|
|
except (ProviderError, PipelineExecutionError, OSError) as exc:
|
|
print(f"EXECUTION ERROR: {exc}", file=sys.stderr)
|
|
return 3
|
|
parser.error("unknown command")
|
|
return 2
|
|
|
|
|
|
def _load_contracts_from_args(args: argparse.Namespace) -> tuple[Brief, SourcePack]:
|
|
brief = Brief.from_dict(read_json(args.brief))
|
|
manual = SourcePack.from_dict(read_json(args.sources) if args.sources else {"sources": []})
|
|
if not args.source_root:
|
|
return brief, manual
|
|
collected = collect_sources(
|
|
args.source_root,
|
|
build_query_from_brief(brief),
|
|
includes=args.source_includes,
|
|
top_k=args.source_top_k,
|
|
max_per_file=args.source_max_per_file,
|
|
)
|
|
return brief, merge_source_packs(manual, collected)
|
|
|
|
|
|
def _load_contracts(brief_path: str, sources_path: str | None) -> tuple[Brief, SourcePack]:
|
|
"""Backward-compatible helper retained for programmatic callers."""
|
|
brief = Brief.from_dict(read_json(brief_path))
|
|
sources = SourcePack.from_dict(read_json(sources_path) if sources_path else {"sources": []})
|
|
return brief, sources
|
|
|
|
|
|
def _cmd_init(directory: Path, force: bool) -> int:
|
|
if directory.exists() and any(directory.iterdir()) and not force:
|
|
raise ValidationError(f"directory is not empty: {directory}; use --force to overwrite starter files")
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
write_json(directory / "brief.json", starter_brief())
|
|
write_json(directory / "sources.json", starter_sources())
|
|
write_json(directory / "pipeline.mock.json", mock_pipeline_config())
|
|
project_root = Path(__file__).resolve().parents[2]
|
|
multi = project_root / "config" / "pipeline.multi-agent.example.json"
|
|
if multi.exists():
|
|
shutil.copy2(multi, directory / multi.name)
|
|
print(f"INITIALIZED: {directory.resolve()}")
|
|
return 0
|
|
|
|
|
|
def _provider_checks(config: PipelineConfig) -> list[dict[str, object]]:
|
|
specs = [config.planner, config.writer, config.reviser, *[item.provider for item in config.reviewers]]
|
|
unique: dict[tuple[str, str, str], object] = {}
|
|
for spec in specs:
|
|
key = (spec.provider, spec.model, json.dumps(spec.options, sort_keys=True, ensure_ascii=False))
|
|
unique.setdefault(key, spec)
|
|
return [create_provider(spec).check() for spec in unique.values()] # type: ignore[arg-type]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|