#!/usr/bin/env python3 """CASAN commercial-readiness discovery and assessment. This module is intentionally stdlib-only. It assesses the CASAN-owned installation contract without changing application source or inventing domain traceability. Core, Domain Pipeline, and Provider Telemetry are independent dimensions because each has a different owner and activation lifecycle. """ from __future__ import annotations from datetime import datetime, timezone import json import os from pathlib import Path from typing import Any SCHEMA_VERSION = 1 SKIP_DIRECTORIES = { ".casan", ".git", ".hg", ".idea", ".specify", ".svn", ".venv", "coverage", "dist", "node_modules", "target", "vendor", } PROJECT_MARKERS = ( "package.json", "pyproject.toml", "pom.xml", "build.gradle", "build.gradle.kts", "go.mod", "Cargo.toml", "composer.json", ) SOURCE_ROOT_NAMES = ("src", "app", "apps", "packages", "lib", "services") REQUIREMENT_NAMES = ( "requirements.md", "requirement.md", "srs.md", "spec.md", "product-requirements.md", ) def utc_now() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def read_json(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) return value if isinstance(value, dict) else {} except (OSError, TypeError, ValueError): return {} def write_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text( json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) os.replace(temporary, path) def configure_domain(root: str | Path, manifest: str | Path) -> dict[str, Any]: """Select an existing project manifest without copying or editing app code.""" project_root = Path(root).expanduser().resolve() selected = Path(manifest).expanduser() if not selected.is_absolute(): selected = project_root / selected selected = selected.resolve() try: relative = selected.relative_to(project_root).as_posix() except ValueError as error: raise ValueError("domain manifest must stay inside the project root") from error if not selected.is_file(): raise ValueError("domain manifest does not exist") try: document = json.loads(selected.read_text(encoding="utf-8")) except (OSError, TypeError, ValueError) as error: raise ValueError("domain manifest must be valid JSON") from error if not isinstance(document, dict) or not ( document.get("project_id") or document.get("schema_version") ): raise ValueError( "domain manifest must declare project_id or schema_version") value = { "schema_version": SCHEMA_VERSION, "configured_at": utc_now(), "manifest": relative, "ownership": "project_source_referenced_by_casan", } write_json(project_root / ".casan" / "domain.json", value) return value def _relative(root: Path, path: Path) -> str: return path.relative_to(root).as_posix() def _bounded_files(root: Path, maximum_depth: int = 5) -> list[Path]: files: list[Path] = [] for current, directories, names in os.walk(root): current_path = Path(current) try: depth = len(current_path.relative_to(root).parts) except ValueError: continue directories[:] = sorted( name for name in directories if name not in SKIP_DIRECTORIES and not name.startswith(".cache") ) if depth >= maximum_depth: directories[:] = [] for name in sorted(names): files.append(current_path / name) if len(files) >= 10_000: return files return files def discover(root: str | Path) -> dict[str, Any]: project_root = Path(root).expanduser().resolve() priority_files: list[Path] = [] for relative in ("apps", "packages", "docs"): candidate = project_root / relative if candidate.is_dir(): priority_files.extend(_bounded_files(candidate, maximum_depth=6)) files = list(dict.fromkeys( priority_files + _bounded_files(project_root) )) markers = [ name for name in PROJECT_MARKERS if (project_root / name).is_file() ] source_roots = [ name for name in SOURCE_ROOT_NAMES if (project_root / name).is_dir() ] domain_packs: list[str] = [] manifests: list[str] = [] requirements: list[str] = [] for path in files: relative = _relative(project_root, path) lowered = path.name.lower() if lowered == "domain-pack.yaml": domain_packs.append(relative) elif lowered == "project.manifest.json": manifests.append(relative) elif ( lowered in REQUIREMENT_NAMES or ("requirement" in lowered and lowered.endswith(".md")) ): requirements.append(relative) return { "schema_version": SCHEMA_VERSION, "generated_at": utc_now(), "project_root": str(project_root), "project_markers": markers[:32], "source_roots": source_roots[:32], "domain_pack_candidates": domain_packs[:64], "project_manifest_candidates": manifests[:64], "requirements_candidates": requirements[:64], "scan": { "maximum_depth": 5, "maximum_files_per_root": 10_000, "priority_roots": [ value for value in ("apps", "packages", "docs") if (project_root / value).is_dir() ], "files_observed": len(files), "truncated": len(files) >= 10_000, }, } def _contains(path: Path, *needles: str) -> bool: try: if not path.is_file() or path.stat().st_size > 2 * 1024 * 1024: return False content = path.read_text(encoding="utf-8", errors="replace") return all(needle in content for needle in needles) except OSError: return False def _runtime_harness(root: Path, config: dict[str, Any]) -> Path | None: runtime_path = config.get("runtime_path") if not isinstance(runtime_path, str) or not runtime_path.strip(): return None base = Path(runtime_path).expanduser() if not base.is_absolute(): base = root / base candidate = base.resolve() nested = candidate / "packages" / "casan-harness" if nested.is_dir(): return nested return candidate if candidate.name == "casan-harness" and candidate.is_dir() else None def _check( gate: str, title: str, status: str, summary: str, evidence: list[str] | None = None, ) -> dict[str, Any]: return { "gate": gate, "title": title, "status": status, "summary": summary, "evidence": evidence or [], } def _provider_available(root: Path) -> bool: candidates = ( root / ".specify" / "logs" / "level5" / "provider-usage.jsonl", root / ".specify" / "logs" / "cost" / "metrics.jsonl", ) for path in candidates: try: if path.is_file() and path.stat().st_size > 2: text = path.read_text(encoding="utf-8", errors="replace")[-262_144:] for line in reversed(text.splitlines()): try: row = json.loads(line) except (TypeError, ValueError): continue if not isinstance(row, dict): continue if ( isinstance(row.get("provider"), str) and row["provider"].strip() or isinstance(row.get("model"), str) and row["model"].strip() or isinstance(row.get("total_tokens"), (int, float)) or isinstance(row.get("cost_estimate"), (int, float)) ): return True except OSError: continue return False def assess( root: str | Path, *, persist: bool = False, client_state: dict[str, Any] | None = None, ) -> dict[str, Any]: project_root = Path(root).expanduser().resolve() config_path = project_root / ".casan" / "config.json" lock_path = project_root / ".casan" / "version.lock" bootstrap_path = project_root / ".casan" / "casan-hook.py" config = read_json(config_path) lock = read_json(lock_path) domain_config = read_json(project_root / ".casan" / "domain.json") previous_readiness = read_json(project_root / ".casan" / "readiness.json") if client_state is None: previous_core = ( (previous_readiness.get("dimensions") or {}).get("core") or {} ) previous_client_state = previous_core.get("client_activation") client_state = ( previous_client_state if isinstance(previous_client_state, dict) else {} ) discovery = discover(project_root) clients = [ str(value) for value in config.get("clients", []) if isinstance(value, str) ] runtime = _runtime_harness(project_root, config) h1_ok = bool(config.get("project_id") and config.get("schema_version")) h2_evidence: list[str] = [] client_integrations: list[dict[str, Any]] = [] for client in clients: declared_state = (client_state or {}).get(client) or {} if client == "claude": relative = ".claude/settings.json" present = _contains( project_root / relative, "casan-hook.py", "--client claude") operational = bool( present and declared_state.get("operational", True)) elif client == "codex": relative = ".codex/hooks.json" present = _contains( project_root / relative, "casan-hook.py", "--client codex") trust_state = declared_state operational = bool( present and trust_state.get("verified") and trust_state.get("operational") ) elif client == "vscode-copilot": relative = ".casan/casan-hook.py" present = (project_root / relative).is_file() operational = bool( present and declared_state.get("operational", True)) else: client_integrations.append({ "client": client, "status": "unsupported", "evidence": None, }) continue if present: h2_evidence.append(relative) client_integrations.append({ "client": client, "status": ( "operational" if operational else "action_required" if present else "missing" ), "evidence": relative if present else None, }) operational_clients = sum( item["status"] == "operational" for item in client_integrations) configured_clients = sum( item["status"] in {"operational", "action_required"} for item in client_integrations) h2_status = ( "warning" if operational_clients and operational_clients < len(client_integrations) else "pass" if operational_clients else "action_required" if configured_clients else "fail" ) report_engine = ( runtime / "scripts" / "python" / "local_report.py" if runtime else Path() ) h3_ok = bool(runtime and report_engine.is_file()) enforcement = str(config.get("enforcement_mode") or "unknown") h4_status = "pass" if enforcement == "enforce" else "warning" state_dirs = ( project_root / ".specify" / "logs", project_root / ".specify" / "state", ) h5_ok = all( path.is_dir() and os.access(path, os.W_OK | os.X_OK) for path in state_dirs ) provider_available = _provider_available(project_root) h7_ok = bool( lock.get("harness_hash") and lock.get("harness_version") and bootstrap_path.is_file() ) trust = (client_state or {}).get("codex") or {} codex_action_required = ( "codex" in clients and not (trust.get("verified") and trust.get("operational")) ) checks = [ _check("H1", "Project contract", "pass" if h1_ok else "fail", "CASAN project identity and schema are valid." if h1_ok else "CASAN project configuration is missing or invalid.", [".casan/config.json"] if h1_ok else []), _check("H2", "Client activation", h2_status, f"{operational_clients} configured client route(s) are operational." if h2_status in {"pass", "warning"} else "Configured client hooks require an activation step." if h2_status == "action_required" else "No complete supported client integration is installed.", h2_evidence), _check("H3", "Local report capability", "pass" if h3_ok else "fail", "Core Local Assurance Viewer and on-demand export are available." if h3_ok else "The resolved Core report runtime is unavailable.", [str(report_engine)] if h3_ok else []), _check("H4", "Enforcement boundary", h4_status, "Policy enforcement is active." if enforcement == "enforce" else "Observe mode records evidence but does not enforce decisions.", [".casan/config.json"]), _check("H5", "Evidence persistence", "pass" if h5_ok else "fail", "Local logs and state directories are writable." if h5_ok else "CASAN evidence directories are missing or not writable.", [".specify/logs", ".specify/state"] if h5_ok else []), _check("H6", "Provider telemetry", "pass" if provider_available else "optional_unavailable", "Provider token/cost telemetry is available." if provider_available else "Core runtime evidence is available; provider token/cost telemetry is optional and not observed yet."), _check("H7", "Pinned runtime", "pass" if h7_ok else "fail", "The project bootstrap and pinned runtime contract are present." if h7_ok else "The bootstrap or version pin is incomplete.", [".casan/casan-hook.py", ".casan/version.lock"] if h7_ok else []), ] blocking = any(item["status"] == "fail" for item in checks) action_required = any( item["status"] == "action_required" for item in checks ) attention = any(item["status"] == "warning" for item in checks) core_status = ( "blocked" if blocking else "action_required" if action_required else "ready_with_attention" if attention else "ready" ) manifests = discovery["project_manifest_candidates"] packs = discovery["domain_pack_candidates"] selected_manifest = domain_config.get("manifest") selected_path = ( project_root / selected_manifest if isinstance(selected_manifest, str) else None ) if selected_path and selected_path.is_file(): domain_status = "configured" domain_summary = ( "A project manifest is selected. Run project validation or the " "project-owned pipeline to produce domain acceptance evidence." ) elif manifests: domain_status = "discovered" domain_summary = ( "Project manifest candidates were discovered but none is selected. " "Use `casan domain configure ` when domain acceptance is needed." ) elif packs: domain_status = "discovered" domain_summary = ( "A Domain Pack was discovered, but no project manifest proves a " "runnable domain pipeline." ) else: domain_status = "not_configured" domain_summary = ( "No Domain Pack is configured. Core prompt assurance and reports " "remain fully available without modifying application source." ) telemetry_status = "available" if provider_available else "optional_unavailable" next_actions: list[dict[str, str]] = [] if blocking: next_actions.append({ "code": "repair_core", "message": "Run `casan init` again, then `casan doctor`.", }) if codex_action_required: reason = str(trust.get("reason") or trust.get("status") or "not_verified") next_actions.append({ "code": "verify_codex_trust", "message": ( "Open `/hooks` in Codex and review the project hooks, then run " f"`casan doctor` (current trust state: {reason})." ), }) for integration in client_integrations: if ( integration["status"] != "operational" and integration["client"] != "codex" ): next_actions.append({ "code": "activate_%s" % integration["client"].replace("-", "_"), "message": ( f"Complete activation for the selected " f"{integration['client']} client route, then run `casan doctor`." ), }) optional_actions: list[dict[str, str]] = [] if domain_status != "configured": optional_actions.append({ "code": "configure_domain_pipeline", "message": ( "Optional: install/configure a CASAN Domain Pack when the team " "needs project-specific SRS→test acceptance." ), }) readiness = { "schema_version": SCHEMA_VERSION, "generated_at": utc_now(), "project_id": str(config.get("project_id") or project_root.name), "overall_status": core_status, "dimensions": { "core": { "status": core_status, "summary": ( "Core is ready for governed prompts and visual local reports." if core_status == "ready" else "Core is operational through at least one client route; another selected route needs attention." if core_status == "ready_with_attention" else "Core requires an explicit activation step." if core_status == "action_required" else "Core installation is incomplete." ), "acceptance": checks, "client_activation": client_state or {}, "client_integrations": client_integrations, }, "domain_pipeline": { "status": domain_status, "summary": domain_summary, "domain_packs": packs, "project_manifests": manifests, "selected_manifest": selected_manifest if selected_path and selected_path.is_file() else None, }, "provider_telemetry": { "status": telemetry_status, "summary": ( "Provider token and cost telemetry has been observed." if provider_available else "Provider token and cost telemetry is optional and has not " "been observed; this does not block Core." ), }, }, "next_actions": next_actions, "optional_actions": optional_actions, "discovery_file": ".casan/discovery.json", } if persist: write_json(project_root / ".casan" / "discovery.json", discovery) write_json(project_root / ".casan" / "readiness.json", readiness) return readiness if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", default=os.getcwd()) parser.add_argument("--write", action="store_true") arguments = parser.parse_args() print(json.dumps( assess(arguments.root, persist=arguments.write), ensure_ascii=False, indent=2, ))