API Reference¶

def evaluate(
    project_root: Path,
    config: ClientConfig,
    *,
    agent_model: str | None = None,
    agent_version: str | None = None,
) -> AuditReport:
    """Evaluate every runtime dep in ``project_root/pyproject.toml``."""
    deps = get_runtime_deps(project_root / "pyproject.toml")
    verdicts: list[Verdict] = []
    if not deps:
        return AuditReport(project=config.project, deps_total=0, verdicts=[])
    with LexAlignClient(
        config, agent_model=agent_model, agent_version=agent_version
    ) as client:
        for name, spec in sorted(deps.items()):
            version = extract_pinned_version(spec)
            verdicts.append(client.check(name, version))
    return AuditReport(
        project=config.project,
        deps_total=len(deps),
        verdicts=verdicts,
    )

def format_report(report: AuditReport) -> str:
    """Render an :class:`AuditReport` as a human-readable summary."""
    lines: list[str] = [
        f"Audited {report.deps_total} runtime dep"
        f"{'s' if report.deps_total != 1 else ''} for project '{report.project}'.",
        "",
    ]
    if report.deps_total == 0:
        lines.append("No `[project].dependencies` found.")
        return "\n".join(lines)

    lines.append(
        f"  ALLOWED              : {len(report.allowed) - len(report.transport_errors)}"
    )
    lines.append(f"  PROVISIONALLY_ALLOWED: {len(report.provisional)}")
    lines.append(f"  DENIED               : {len(report.denied)}")
    if report.transport_errors:
        lines.append(f"  (server unreachable for {len(report.transport_errors)})")
    lines.append("")

    if report.denied:
        lines.append("DENIED:")
        for v in report.denied:
            lines.extend(_render_verdict(v, marker="✗"))
        lines.append("")
    if report.provisional:
        lines.append("PROVISIONALLY_ALLOWED:")
        for v in report.provisional:
            lines.extend(_render_verdict(v, marker="◎"))
        lines.append(
            "  → run `lex-align-client request-approval --package <name> "
            "--rationale \"<why>\"` to enqueue formal review."
        )
        lines.append("")
    return "\n".join(lines).rstrip() + "\n"

def run(
    project_root: Path,
    config: ClientConfig,
    *,
    as_json: bool,
    agent_model: str | None = None,
    agent_version: str | None = None,
) -> int:
    """CLI entry point. Returns the desired exit code (0/1/2)."""
    try:
        report = evaluate(
            project_root, config,
            agent_model=agent_model, agent_version=agent_version,
        )
    except ServerUnreachable as exc:
        print(f"[lex-align] server unreachable: {exc}", file=sys.stderr)
        return 1
    except ServerError as exc:
        print(f"[lex-align] server error: {exc}", file=sys.stderr)
        return 1
    if as_json:
        print(json.dumps(report.to_dict(), indent=2))
    else:
        print(format_report(report), end="")
    return 2 if report.denied else 0

def handle_pre_tool_use(
    event: dict, project_root: Path, config: ClientConfig
) -> Optional[tuple[str, str | None]]:
    """Return (decision, message) for pyproject.toml edits, else None."""
    tool_name = event.get("tool_name", "")
    tool_input = event.get("tool_input", {}) or {}
    path_str = tool_input.get("path", "") or tool_input.get("file_path", "")
    if "pyproject.toml" not in path_str:
        return None
    if tool_name not in ("Edit", "Write", "MultiEdit"):
        return None

    pyproject_path = (
        project_root / path_str if not Path(path_str).is_absolute() else Path(path_str)
    )
    if not pyproject_path.exists():
        return None

    current = pyproject_path.read_text()
    proposed = apply_edit(current, tool_name, tool_input)
    added, removed = diff_deps(current, proposed)
    if not added and not removed:
        return None

    header = ["You are modifying runtime dependencies:"]
    for name, spec in sorted(added.items()):
        header.append(f"  + {spec}")
    for name in sorted(removed):
        header.append(f"  - {name}")

    model, version = _detect_agent(event)
    blocks: list[str] = []
    notes: list[str] = []
    try:
        with LexAlignClient(
            config, agent_model=model, agent_version=version
        ) as client:
            for name, spec in sorted(added.items()):
                pinned = extract_pinned_version(spec)
                v = client.check(name, pinned)
                if v.denied:
                    blocks.append(_format_verdict(v, spec))
                elif v.verdict == "PROVISIONALLY_ALLOWED":
                    auto_note = ""
                    if config.auto_request_approval and v.is_requestable:
                        rationale = _auto_rationale(spec, v)
                        try:
                            client.request_approval(name, rationale)
                            auto_note = " (auto-enqueued for review)"
                        except (ServerUnreachable, ServerError) as exc:
                            auto_note = f" (auto-enqueue failed: {exc.__class__.__name__})"
                    elif v.is_requestable:
                        auto_note = " (run `lex-align-client request-approval` after this lands)"
                    notes.append(f"  ◎ {name} — provisional: {v.reason}{auto_note}")
                elif v.needs_rationale:
                    notes.append(
                        f"  • {name} — allowed, but registry-approved (neutral); "
                        "document the architectural need in your commit message or PR."
                    )
                else:
                    notes.append(f"  ✓ {name} — {v.reason}")
    except (ServerUnreachable, ServerError) as exc:
        if config.fail_open:
            return ("allow", "\n".join(header + ["", f"[lex-align] {exc} — fail_open=true; allowing edit."]))
        return ("block", f"[lex-align] cannot reach server: {exc}")

    if blocks:
        msg = "\n".join(header + ["", "ENFORCEMENT — blocked by registry:"] + blocks +
                        ["", "No dependencies were modified. Adjust the edit and retry."])
        return ("block", msg)
    if notes:
        return ("allow", "\n".join(header + [""] + notes))
    return ("allow", "\n".join(header))

def install_claude_md(project_root: Path) -> tuple[Path, bool]:
    """Create CLAUDE.md or append the lex-align section if not already present.

    Returns (path, created_or_updated). Returns False for the second element
    when the section was already present and no write was performed.
    """
    path = project_root / "CLAUDE.md"
    if path.exists():
        existing = path.read_text(encoding="utf-8")
        if _SECTION_HEADER in existing:
            return path, False
        path.write_text(existing.rstrip("\n") + "\n\n" + _SECTION + "\n", encoding="utf-8")
    else:
        path.write_text(_SECTION + "\n", encoding="utf-8")
    return path, True

@click.group()
def main() -> None:
    """lex-align client — talks to the lex-align server."""

@main.command()
@click.option("--yes", "-y", is_flag=True, help="Accept defaults without prompting.")
@click.option("--server-url", default=None, help="Override server URL.")
@click.option("--project", "project_name", default=None, help="Override project name.")
@click.option("--mode", type=click.Choice(["single-user", "org"]), default=None)
@click.option("--no-claude-hooks", is_flag=True, help="Skip Claude Code hook install.")
@click.option("--no-precommit", is_flag=True, help="Skip git pre-commit install.")
@click.option("--no-claude-md", is_flag=True, help="Skip CLAUDE.md creation/update.")
def init(
    yes: bool,
    server_url: str | None,
    project_name: str | None,
    mode: str | None,
    no_claude_hooks: bool,
    no_precommit: bool,
    no_claude_md: bool,
) -> None:
    """One-time setup: write .lexalign.toml and install hooks."""
    project_root = Path.cwd()

    if config_path(project_root).exists() and not yes:
        if not click.confirm(
            f"{CONFIG_FILENAME} already exists. Overwrite?", default=False
        ):
            raise click.Abort()

    autodetected = detect_project_name(project_root / "pyproject.toml", project_root.name)
    if project_name is None:
        project_name = autodetected if yes else click.prompt(
            "Project name", default=autodetected
        )
    if server_url is None:
        server_url = "http://127.0.0.1:8765" if yes else click.prompt(
            "Server URL", default="http://127.0.0.1:8765"
        )
    if mode is None:
        mode = "single-user" if yes else click.prompt(
            "Mode", type=click.Choice(["single-user", "org"]), default="single-user"
        )

    config = ClientConfig(
        project=project_name,
        server_url=server_url,
        mode=mode,
        fail_open=(mode == "single-user"),
        auto_request_approval=(mode == "single-user"),
    )
    path = save_config(project_root, config)
    click.echo(f"Wrote {path}")

    if not no_claude_hooks:
        install_claude_hooks(project_root)
        click.echo("Installed Claude Code hooks in .claude/settings.json.")

    if not no_claude_md:
        md_existed = (project_root / "CLAUDE.md").exists()
        md_path, changed = install_claude_md(project_root)
        if changed:
            action = "Updated" if md_existed else "Created"
            click.echo(f"{action} {md_path.name} with lex-align usage instructions.")
        else:
            click.echo(f"{md_path.name} already contains lex-align section; skipped.")

    if not no_precommit:
        hook_path = install_precommit(project_root)
        if hook_path:
            click.echo(f"Installed git pre-commit hook at {hook_path}.")
        else:
            click.echo(
                "Skipped pre-commit install: this directory is not a git repo "
                "(run `git init` then `lex-align-client init` again)."
            )

    if mode == "org":
        click.echo("")
        click.echo(
            f"Organization mode: export your API key as ${config.api_key_env_var} "
            "before running checks."
        )

@main.command()
@click.option("--package", required=True)
@click.option("--version", default=None)
@click.option("--json", "as_json", is_flag=True, default=True, help="Emit JSON (default).")
@click.option(
    "--agent-model",
    default=None,
    help="Override agent model tag (default: $LEXALIGN_AGENT_MODEL).",
)
@click.option(
    "--agent-version",
    default=None,
    help="Override agent version tag (default: $LEXALIGN_AGENT_VERSION).",
)
def check(
    package: str,
    version: str | None,
    as_json: bool,
    agent_model: str | None,
    agent_version: str | None,
) -> None:
    """Check a package against the server policy."""
    project_root = find_project_root()
    config = _require_config(project_root)
    try:
        with LexAlignClient(
            config, agent_model=agent_model, agent_version=agent_version
        ) as client:
            verdict = client.check(package, version)
    except ServerUnreachable as exc:
        raise click.ClickException(f"Server unreachable: {exc}")
    except ServerError as exc:
        raise click.ClickException(f"Server error: {exc}")
    click.echo(json.dumps(verdict.to_dict(), indent=2))
    if verdict.denied:
        sys.exit(2)

@main.command("request-approval")
@click.option("--package", required=True)
@click.option("--rationale", required=True)
@click.option(
    "--agent-model",
    default=None,
    help="Override agent model tag (default: $LEXALIGN_AGENT_MODEL).",
)
@click.option(
    "--agent-version",
    default=None,
    help="Override agent version tag (default: $LEXALIGN_AGENT_VERSION).",
)
def request_approval(
    package: str,
    rationale: str,
    agent_model: str | None,
    agent_version: str | None,
) -> None:
    """Submit a non-blocking request to add `package` to the registry."""
    project_root = find_project_root()
    config = _require_config(project_root)
    try:
        with LexAlignClient(
            config, agent_model=agent_model, agent_version=agent_version
        ) as client:
            response = client.request_approval(package, rationale)
    except ServerUnreachable as exc:
        raise click.ClickException(f"Server unreachable: {exc}")
    except ServerError as exc:
        raise click.ClickException(f"Server error: {exc}")
    click.echo(json.dumps(response, indent=2))

@main.command()
@click.option("--json", "as_json", is_flag=True, help="Emit machine-readable JSON.")
@click.option(
    "--agent-model",
    default=None,
    help="Override agent model tag (default: $LEXALIGN_AGENT_MODEL).",
)
@click.option(
    "--agent-version",
    default=None,
    help="Override agent version tag (default: $LEXALIGN_AGENT_VERSION).",
)
def audit(
    as_json: bool, agent_model: str | None, agent_version: str | None
) -> None:
    """Re-evaluate every runtime dep in `pyproject.toml` against the server.

    Read-only sibling of the pre-commit hook: useful when adopting
    `lex-align` on an existing project, or to check policy without
    needing to stage and commit. Exits 2 if any dep is DENIED.
    """
    project_root = find_project_root()
    config = _require_config(project_root)
    sys.exit(
        audit_module.run(
            project_root, config, as_json=as_json,
            agent_model=agent_model, agent_version=agent_version,
        )
    )

@main.command()
@click.option("--json", "as_json", is_flag=True, help="Emit machine-readable JSON.")
def status(as_json: bool) -> None:
    """One-screen status: server reachability, hooks, pending approvals, denials."""
    project_root = find_project_root()
    config = _require_config(project_root)
    report = status_module.collect(project_root, config)
    if as_json:
        click.echo(json.dumps(report.to_dict(), indent=2))
    else:
        click.echo(status_module.format_report(report))

@main.command()
def precommit() -> None:
    """Pre-commit hook entry point. Exits non-zero on any DENIED dependency."""
    sys.exit(run_precommit())

@main.command()
@click.option("--yes", "-y", is_flag=True)
def uninstall(yes: bool) -> None:
    """Remove .claude hooks and the git pre-commit shim. Leaves .lexalign.toml."""
    project_root = find_project_root()
    if not yes and not click.confirm(
        "Remove lex-align hooks (Claude + git pre-commit)?", default=False
    ):
        raise click.Abort()
    remove_claude_hooks(project_root)
    remove_precommit(project_root)
    click.echo("Removed Claude hooks and git pre-commit shim.")
    click.echo(f"{CONFIG_FILENAME} preserved.")

def find_project_root(start: Optional[Path] = None) -> Path:
    """Walk up looking for `.lexalign.toml`. Falls back to cwd."""
    cwd = (start or Path.cwd()).resolve()
    for parent in [cwd] + list(cwd.parents):
        if (parent / CONFIG_FILENAME).exists():
            return parent
    return cwd

def get_runtime_deps(pyproject_path: Path) -> dict[str, str]:
    """Return {normalized_name: raw_spec} from [project].dependencies."""
    if not pyproject_path.exists():
        return {}
    with open(pyproject_path, "rb") as f:
        data = tomllib.load(f)
    deps = data.get("project", {}).get("dependencies", []) or []
    return {normalize_name(d): d.strip() for d in deps}

def diff_deps(old_content: str, new_content: str) -> tuple[dict[str, str], set[str]]:
    """Return ({added_name: raw_spec}, {removed_names})."""
    old = parse_deps_from_content(old_content)
    new = parse_deps_from_content(new_content)
    added = {name: spec for name, spec in new.items() if name not in old}
    removed = set(old) - set(new)
    return added, removed

def extract_pinned_version(spec: str) -> Optional[str]:
    """Pull a concrete version out of a spec like 'redis>=5.0' or 'httpx==0.28.1'."""
    m = re.search(r"(?:>=|<=|!=|~=|==|>|<)\s*([0-9][0-9a-zA-Z\.\-\+\_]*)", spec)
    return m.group(1) if m else None

def apply_edit(current_content: str, tool_name: str, tool_input: dict) -> str:
    """Simulate how a Claude Code Edit/Write/MultiEdit affects file content."""
    if tool_name == "Write":
        return tool_input.get("content", "")
    if tool_name == "Edit":
        old_str = tool_input.get("old_string", "")
        new_str = tool_input.get("new_string", "")
        return current_content.replace(old_str, new_str, 1)
    if tool_name == "MultiEdit":
        content = current_content
        for edit in tool_input.get("edits", []):
            content = content.replace(edit.get("old_string", ""), edit.get("new_string", ""), 1)
        return content
    return current_content

def detect_project_name(pyproject_path: Path, fallback: str) -> str:
    """Best-effort autodetect for `lex-align-client init`.

    Prefer [project].name from pyproject.toml; fall back to the directory name.
    """
    if pyproject_path.exists():
        try:
            with open(pyproject_path, "rb") as f:
                data = tomllib.load(f)
            name = data.get("project", {}).get("name")
            if isinstance(name, str) and name.strip():
                return name.strip()
        except Exception:
            pass
    return fallback

def precommit_installed(project_root: Path) -> bool:
    """True iff the git pre-commit hook contains the lex-align marker."""
    hook_path = _precommit_path(project_root)
    if not hook_path.exists():
        return False
    try:
        return _PRECOMMIT_MARKER in hook_path.read_text()
    except OSError:
        return False

def install_precommit(project_root: Path) -> Path | None:
    """Install or augment `.git/hooks/pre-commit`. Returns the path if written."""
    git_dir = project_root / ".git"
    if not git_dir.exists():
        return None
    hook_path = _precommit_path(project_root)
    hook_path.parent.mkdir(parents=True, exist_ok=True)

    snippet = (
        f"\n{_PRECOMMIT_MARKER}\n"
        "if command -v lex-align-client >/dev/null 2>&1; then\n"
        "  lex-align-client precommit || exit $?\n"
        "elif command -v uv >/dev/null 2>&1; then\n"
        "  uv run lex-align-client precommit || exit $?\n"
        "fi\n"
    )

    if hook_path.exists():
        existing = hook_path.read_text()
        if _PRECOMMIT_MARKER in existing:
            return hook_path
        new = existing.rstrip() + "\n" + snippet
        hook_path.write_text(new)
    else:
        hook_path.write_text("#!/usr/bin/env bash\nset -e\n" + snippet)
    try:
        hook_path.chmod(hook_path.stat().st_mode | 0o111)
    except OSError:
        pass
    return hook_path