engine.py
python
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1
docs: MuseHub-first before ISR #74 — staging solidify NEXT
Human
1 day ago
| 1 | """Thin HTTP handlers that call existing CLI engine functions (§Q0.7.1).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | from argparse import Namespace |
| 7 | from pathlib import Path |
| 8 | from typing import Any |
| 9 | |
| 10 | from cli.commands.governance_sync import run_governance_sync_command |
| 11 | from cli.commands.review import run_review |
| 12 | from cli.commands.status import run_status |
| 13 | from cli.context import CliContext |
| 14 | from tools.app.capture import CapturingOutputContext |
| 15 | from tools.app.docs import load_repo_config, read_handover, read_roadmap, resolve_app_repo |
| 16 | from tools.app.envelope import ApiEnvelope, bad_request, engine_failure, engine_success |
| 17 | from tools.honesty.ledger import append_entry, show_entries, verify_ledger_file |
| 18 | from tools.honesty.status import HonestyStatusOptions, run_honesty_status |
| 19 | from tools.honesty.types import ENTRY_KINDS, LedgerAppendOptions |
| 20 | |
| 21 | |
| 22 | def _unknown_keys(body: dict[str, Any], allowed: frozenset[str]) -> list[str]: |
| 23 | return sorted(key for key in body if key not in allowed) |
| 24 | |
| 25 | |
| 26 | def handle_health(*, port: int, bind: str, repo_root: Path | str) -> ApiEnvelope: |
| 27 | """Process liveness endpoint with bound checkout path (§LAC.6.3). |
| 28 | |
| 29 | ``repo_root`` is the absolute filesystem path this process mutates — not a |
| 30 | credential and not living-doc contents. Auth remains Bearer-gated (Q0 §Q0.6). |
| 31 | """ |
| 32 | root = Path(repo_root).resolve() |
| 33 | return engine_success( |
| 34 | { |
| 35 | "status": "ok", |
| 36 | "port": port, |
| 37 | "bind": bind, |
| 38 | "repo_root": str(root), |
| 39 | } |
| 40 | ) |
| 41 | |
| 42 | |
| 43 | def handle_status(ctx: CliContext, *, repo_arg: str | None = None) -> ApiEnvelope: |
| 44 | """Mirror ``overseer status --json`` and always compute exit-code conditions.""" |
| 45 | repo_root = resolve_app_repo(ctx.cwd, repo_arg) |
| 46 | |
| 47 | capture = CapturingOutputContext() |
| 48 | app_ctx = CliContext( |
| 49 | kit=ctx.kit, |
| 50 | runner=ctx.runner, |
| 51 | output=capture, |
| 52 | cwd=ctx.cwd, |
| 53 | review_provider_factory=ctx.review_provider_factory, |
| 54 | script_executor=ctx.script_executor, |
| 55 | ) |
| 56 | args = Namespace( |
| 57 | repo=repo_arg, |
| 58 | config=None, |
| 59 | exit_code=True, |
| 60 | check_footprint=False, |
| 61 | ) |
| 62 | exit_code = run_status(args, app_ctx) |
| 63 | payload = capture.json_payload or {} |
| 64 | return ApiEnvelope( |
| 65 | ok=exit_code == 0, |
| 66 | exit_code=exit_code, |
| 67 | error=None if exit_code == 0 else _status_error_token(payload), |
| 68 | result=payload, |
| 69 | http_status=200, |
| 70 | ) |
| 71 | |
| 72 | |
| 73 | def _status_error_token(payload: dict[str, Any]) -> str | None: |
| 74 | if payload.get("error"): |
| 75 | return "config" |
| 76 | substrate = payload.get("substrate") or {} |
| 77 | if substrate.get("ok") is False: |
| 78 | return "substrate" |
| 79 | muse_sync = payload.get("muse_sync") or {} |
| 80 | if muse_sync.get("ok") is False: |
| 81 | return "muse_sync" |
| 82 | footprint = payload.get("footprint_self_integrity") or {} |
| 83 | if footprint.get("ok") is False: |
| 84 | return "footprint_self_integrity" |
| 85 | if payload.get("footprint_integrity") == "mismatch": |
| 86 | return "footprint_integrity" |
| 87 | drift = payload.get("drift") or {} |
| 88 | if drift.get("status") in {"behind", "ahead"}: |
| 89 | return "drift" |
| 90 | return "status" |
| 91 | |
| 92 | |
| 93 | def handle_gates(ctx: CliContext, *, repo_arg: str | None = None) -> ApiEnvelope: |
| 94 | """Return the pending-gates slice already embedded in status JSON.""" |
| 95 | status = handle_status(ctx, repo_arg=repo_arg) |
| 96 | if status.result is None or not isinstance(status.result, dict): |
| 97 | return status |
| 98 | gates = status.result.get("governance_gates") |
| 99 | return ApiEnvelope( |
| 100 | ok=status.ok, |
| 101 | exit_code=status.exit_code, |
| 102 | error=status.error, |
| 103 | result=gates, |
| 104 | http_status=status.http_status, |
| 105 | ) |
| 106 | |
| 107 | |
| 108 | def handle_docs_roadmap(ctx: CliContext, *, repo_arg: str | None = None) -> ApiEnvelope: |
| 109 | repo_root = resolve_app_repo(ctx.cwd, repo_arg) |
| 110 | if not (repo_root / ".overseer").is_dir(): |
| 111 | return engine_failure(exit_code=2, error="config", result=None) |
| 112 | config, config_refusal = load_repo_config(repo_root) |
| 113 | if config_refusal is not None or config is None: |
| 114 | return config_refusal or engine_failure(exit_code=2, error="config", result=None) |
| 115 | return read_roadmap(repo_root=repo_root, config=config) |
| 116 | |
| 117 | |
| 118 | def handle_docs_handover(ctx: CliContext, *, repo_arg: str | None = None) -> ApiEnvelope: |
| 119 | repo_root = resolve_app_repo(ctx.cwd, repo_arg) |
| 120 | if not (repo_root / ".overseer").is_dir(): |
| 121 | return engine_failure(exit_code=2, error="config", result=None) |
| 122 | config, config_refusal = load_repo_config(repo_root) |
| 123 | if config_refusal is not None or config is None: |
| 124 | return config_refusal or engine_failure(exit_code=2, error="config", result=None) |
| 125 | return read_handover(repo_root=repo_root, config=config) |
| 126 | |
| 127 | |
| 128 | def handle_review_freeze( |
| 129 | ctx: CliContext, |
| 130 | body: dict[str, Any], |
| 131 | *, |
| 132 | repo_arg: str | None = None, |
| 133 | ) -> ApiEnvelope: |
| 134 | allowed = frozenset({"path", "dry_run", "no_stamp"}) |
| 135 | unknown = _unknown_keys(body, allowed) |
| 136 | if unknown: |
| 137 | return bad_request("unknown_fields") |
| 138 | |
| 139 | path = body.get("path") |
| 140 | if not isinstance(path, str) or not path.strip(): |
| 141 | return bad_request("path_required") |
| 142 | |
| 143 | dry_run = body.get("dry_run", True) |
| 144 | no_stamp = body.get("no_stamp", False) |
| 145 | if not isinstance(dry_run, bool) or not isinstance(no_stamp, bool): |
| 146 | return bad_request("invalid_boolean") |
| 147 | |
| 148 | capture = CapturingOutputContext() |
| 149 | app_ctx = CliContext( |
| 150 | kit=ctx.kit, |
| 151 | runner=ctx.runner, |
| 152 | output=capture, |
| 153 | cwd=ctx.cwd, |
| 154 | review_provider_factory=ctx.review_provider_factory, |
| 155 | script_executor=ctx.script_executor, |
| 156 | ) |
| 157 | args = Namespace( |
| 158 | repo=repo_arg, |
| 159 | config=None, |
| 160 | freeze_path=path, |
| 161 | dry_run=dry_run, |
| 162 | no_stamp=no_stamp, |
| 163 | mode=None, |
| 164 | provider=None, |
| 165 | model=None, |
| 166 | checklist=None, |
| 167 | ) |
| 168 | exit_code = run_review(args, app_ctx) |
| 169 | result = capture.json_payload |
| 170 | return ApiEnvelope( |
| 171 | ok=exit_code == 0, |
| 172 | exit_code=exit_code, |
| 173 | error=_review_error_token(exit_code), |
| 174 | result=result, |
| 175 | http_status=200, |
| 176 | ) |
| 177 | |
| 178 | |
| 179 | def _review_error_token(exit_code: int) -> str | None: |
| 180 | if exit_code == 0: |
| 181 | return None |
| 182 | if exit_code in {7, 8}: |
| 183 | return "review" |
| 184 | if exit_code == 4: |
| 185 | return "path" |
| 186 | if exit_code == 2: |
| 187 | return "gate" |
| 188 | return "review" |
| 189 | |
| 190 | |
| 191 | def handle_governance_sync( |
| 192 | ctx: CliContext, |
| 193 | body: dict[str, Any], |
| 194 | *, |
| 195 | repo_arg: str | None = None, |
| 196 | ) -> ApiEnvelope: |
| 197 | allowed = frozenset({"write"}) |
| 198 | unknown = _unknown_keys(body, allowed) |
| 199 | if unknown: |
| 200 | return bad_request("unknown_fields") |
| 201 | |
| 202 | write = body.get("write", False) |
| 203 | if not isinstance(write, bool): |
| 204 | return bad_request("invalid_boolean") |
| 205 | |
| 206 | capture = CapturingOutputContext() |
| 207 | app_ctx = CliContext( |
| 208 | kit=ctx.kit, |
| 209 | runner=ctx.runner, |
| 210 | output=capture, |
| 211 | cwd=ctx.cwd, |
| 212 | review_provider_factory=ctx.review_provider_factory, |
| 213 | script_executor=ctx.script_executor, |
| 214 | ) |
| 215 | args = Namespace( |
| 216 | repo=repo_arg, |
| 217 | config=None, |
| 218 | write=write, |
| 219 | dry_run=not write, |
| 220 | lane=None, |
| 221 | all_lanes=False, |
| 222 | ) |
| 223 | exit_code = run_governance_sync_command(args, app_ctx) |
| 224 | result = capture.json_payload |
| 225 | return ApiEnvelope( |
| 226 | ok=exit_code == 0, |
| 227 | exit_code=exit_code, |
| 228 | error=_governance_sync_error_token(exit_code), |
| 229 | result=result, |
| 230 | http_status=200, |
| 231 | ) |
| 232 | |
| 233 | |
| 234 | def _governance_sync_error_token(exit_code: int) -> str | None: |
| 235 | if exit_code == 0: |
| 236 | return None |
| 237 | if exit_code == 2: |
| 238 | return "gate" |
| 239 | if exit_code == 4: |
| 240 | return "path" |
| 241 | return "governance_sync" |
| 242 | |
| 243 | |
| 244 | def handle_ledger_show( |
| 245 | ctx: CliContext, |
| 246 | *, |
| 247 | last: int | None = None, |
| 248 | repo_arg: str | None = None, |
| 249 | ) -> ApiEnvelope: |
| 250 | repo_root = resolve_app_repo(ctx.cwd, repo_arg) |
| 251 | if not (repo_root / ".overseer").is_dir(): |
| 252 | return engine_failure(exit_code=2, error="config", result=None) |
| 253 | |
| 254 | config, config_refusal = load_repo_config(repo_root) |
| 255 | if config_refusal is not None or config is None: |
| 256 | return config_refusal or engine_failure(exit_code=2, error="config", result=None) |
| 257 | |
| 258 | last_n = 20 if last is None else last |
| 259 | result = show_entries(config=config, repo_root=repo_root, last_n=last_n) |
| 260 | entries = [json.loads(line) for line in result.stdout_lines] |
| 261 | return ApiEnvelope( |
| 262 | ok=result.exit_code == 0, |
| 263 | exit_code=result.exit_code, |
| 264 | error=_ledger_error_token(result.exit_code), |
| 265 | result={"entries": entries, "lines": result.stdout_lines}, |
| 266 | http_status=200, |
| 267 | ) |
| 268 | |
| 269 | |
| 270 | def handle_ledger_verify(ctx: CliContext, body: dict[str, Any], *, repo_arg: str | None = None) -> ApiEnvelope: |
| 271 | if body and _unknown_keys(body, frozenset()): |
| 272 | return bad_request("unknown_fields") |
| 273 | |
| 274 | repo_root = resolve_app_repo(ctx.cwd, repo_arg) |
| 275 | if not (repo_root / ".overseer").is_dir(): |
| 276 | return engine_failure(exit_code=2, error="config", result=None) |
| 277 | |
| 278 | config, config_refusal = load_repo_config(repo_root) |
| 279 | if config_refusal is not None or config is None: |
| 280 | return config_refusal or engine_failure(exit_code=2, error="config", result=None) |
| 281 | |
| 282 | result = verify_ledger_file(config=config, repo_root=repo_root) |
| 283 | return ApiEnvelope( |
| 284 | ok=result.exit_code == 0, |
| 285 | exit_code=result.exit_code, |
| 286 | error=_ledger_error_token(result.exit_code), |
| 287 | result={"verified": result.exit_code == 0}, |
| 288 | http_status=200, |
| 289 | ) |
| 290 | |
| 291 | |
| 292 | def handle_ledger_append( |
| 293 | ctx: CliContext, |
| 294 | body: dict[str, Any], |
| 295 | *, |
| 296 | repo_arg: str | None = None, |
| 297 | ) -> ApiEnvelope: |
| 298 | allowed = frozenset({"kind", "entry"}) |
| 299 | unknown = _unknown_keys(body, allowed) |
| 300 | if unknown: |
| 301 | return bad_request("unknown_fields") |
| 302 | |
| 303 | kind = body.get("kind") |
| 304 | entry = body.get("entry") |
| 305 | if not isinstance(kind, str) or kind not in ENTRY_KINDS: |
| 306 | return bad_request("kind_required") |
| 307 | if not isinstance(entry, dict): |
| 308 | return bad_request("entry_required") |
| 309 | |
| 310 | repo_root = resolve_app_repo(ctx.cwd, repo_arg) |
| 311 | if not (repo_root / ".overseer").is_dir(): |
| 312 | return engine_failure(exit_code=2, error="config", result=None) |
| 313 | |
| 314 | config, config_refusal = load_repo_config(repo_root) |
| 315 | if config_refusal is not None or config is None: |
| 316 | return config_refusal or engine_failure(exit_code=2, error="config", result=None) |
| 317 | |
| 318 | result = append_entry( |
| 319 | config=config, |
| 320 | repo_root=repo_root, |
| 321 | options=LedgerAppendOptions(kind=kind, body=entry), |
| 322 | ) |
| 323 | return ApiEnvelope( |
| 324 | ok=result.exit_code == 0, |
| 325 | exit_code=result.exit_code, |
| 326 | error=_ledger_error_token(result.exit_code), |
| 327 | result={"appended": result.exit_code == 0}, |
| 328 | http_status=200, |
| 329 | ) |
| 330 | |
| 331 | |
| 332 | def _ledger_error_token(exit_code: int) -> str | None: |
| 333 | mapping = { |
| 334 | 0: None, |
| 335 | 1: "usage", |
| 336 | 2: "invalid", |
| 337 | 4: "refused", |
| 338 | 5: "write_failed", |
| 339 | 21: "approval_integrity", |
| 340 | 22: "chain_broken", |
| 341 | 23: "role_violation", |
| 342 | 24: "evidence_free", |
| 343 | 25: "provenance", |
| 344 | 26: "signature_required", |
| 345 | } |
| 346 | return mapping.get(exit_code, "ledger") |
| 347 | |
| 348 | |
| 349 | def handle_honesty_status( |
| 350 | ctx: CliContext, |
| 351 | body: dict[str, Any], |
| 352 | *, |
| 353 | repo_arg: str | None = None, |
| 354 | ) -> ApiEnvelope: |
| 355 | allowed = frozenset( |
| 356 | { |
| 357 | "hook", |
| 358 | "artifact", |
| 359 | "producer_session", |
| 360 | "verification_evidence", |
| 361 | "frozen_spec", |
| 362 | } |
| 363 | ) |
| 364 | unknown = _unknown_keys(body, allowed) |
| 365 | if unknown: |
| 366 | return bad_request("unknown_fields") |
| 367 | |
| 368 | repo_root = resolve_app_repo(ctx.cwd, repo_arg) |
| 369 | if not (repo_root / ".overseer").is_dir(): |
| 370 | return engine_failure(exit_code=2, error="config", result=None) |
| 371 | |
| 372 | config, config_refusal = load_repo_config(repo_root) |
| 373 | if config_refusal is not None or config is None: |
| 374 | return config_refusal or engine_failure(exit_code=2, error="config", result=None) |
| 375 | |
| 376 | options = HonestyStatusOptions( |
| 377 | hook=body.get("hook"), |
| 378 | artifact=body.get("artifact"), |
| 379 | producer_session=body.get("producer_session"), |
| 380 | verification_evidence=body.get("verification_evidence"), |
| 381 | frozen_spec=body.get("frozen_spec"), |
| 382 | emit_json=True, |
| 383 | ) |
| 384 | result = run_honesty_status(config=config, repo_root=repo_root, options=options) |
| 385 | return ApiEnvelope( |
| 386 | ok=result.exit_code == 0, |
| 387 | exit_code=result.exit_code, |
| 388 | error=_honesty_error_token(result.exit_code), |
| 389 | result=result.json_payload.to_dict(), |
| 390 | http_status=200, |
| 391 | ) |
| 392 | |
| 393 | |
| 394 | def _honesty_error_token(exit_code: int) -> str | None: |
| 395 | mapping = { |
| 396 | 0: None, |
| 397 | 1: "usage", |
| 398 | 4: "refused", |
| 399 | 20: "missing_verdict", |
| 400 | 25: "provenance", |
| 401 | 26: "signature_required", |
| 402 | 33: "missing_verification_evidence", |
| 403 | } |
| 404 | return mapping.get(exit_code, "honesty") |
File History
1 commit
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1
docs: MuseHub-first before ISR #74 — staging solidify NEXT
Human
1 day ago