diff --git a/CHANGELOG.md b/CHANGELOG.md index 61781716..503c9a39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,35 @@ All notable changes to ForgeLM are documented here. > Per-PR CHANGELOG entries below collapse into the v0.5.5 release > notes at tag time. +### Added — Wave 2a / Phase 37 — `forgelm approvals` listing subcommand + +- **`forgelm approvals --pending [--output-dir DIR]`** lists every run whose + audit log carries a `human_approval.required` event without a matching + terminal decision. Tabular text output or a JSON envelope + (`{"success": true, "pending": [...], "count": N}`) when + `--output-format json` is set, so CI can filter the queue programmatically. +- **`forgelm approvals --show RUN_ID --output-dir DIR`** prints the full + approval-gate audit chain (request → terminal decision) plus the on-disk + staging directory layout. Useful for forensic review of granted / + rejected runs and confirming the staging contents match what the + operator approved. +- Closes the Phase 9 follow-up gap from `ghost-features-analysis-20260502` + (GH-007 — listing was missing; an operator could not discover pending + runs without `ls`-ing the filesystem by hand). +- New module: `forgelm/cli/subcommands/_approvals.py`. Reuses the audit- + log parsing pattern from `_approve.py` (skip malformed lines, emit one + summary warning at end). +- 15 new tests in `tests/test_approvals_listing.py` covering: empty + audit log, three-runs-one-decided, rejected runs excluded from + pending, JSON envelope structure, table rendering, `--show` pending / + granted / unknown run_id, `--show` with no audit log, monkeypatch + facade resolution, malformed-line robustness, dispatcher defensive + double-check. +- Operator docs updated: `docs/usermanuals/{en,tr}/compliance/human- + oversight.md` "Inspecting pending runs" section now matches the + shipped output (table format, `--output-dir` flag) instead of the + pre-implementation sketch. + ### Added — Wave 1 closure (Faz 9, 11, 12, 13, 25, 31, 32 — see PR description) - **Article 14 staging directory + `forgelm approve` / `forgelm reject` (Faz 9)** — diff --git a/docs/usermanuals/en/compliance/human-oversight.md b/docs/usermanuals/en/compliance/human-oversight.md index 071e03ef..2d56ee11 100644 --- a/docs/usermanuals/en/compliance/human-oversight.md +++ b/docs/usermanuals/en/compliance/human-oversight.md @@ -127,18 +127,37 @@ Default is 48 hours. Set to 0 for "no timeout — wait forever" (not recommended ## Inspecting pending runs +`forgelm approvals` is the discovery counterpart to `approve` / `reject`. It scans the audit log under `--output-dir` and reports every run whose `human_approval.required` event has no matching terminal decision. + ```shell -$ forgelm approvals --pending -RUN_ID REQUESTED_AT ARTIFACTS EXPIRES -abc123 2026-04-29T14:33Z checkpoints/run/artifacts/ in 47h -def456 2026-04-29T09:12Z checkpoints/sft-only/artifacts/ in 42h +$ forgelm approvals --pending --output-dir checkpoints/ +Pending approvals (2): + +RUN_ID AGE REQUESTED_AT STAGING +---------------- ---- ------------------------- ------- +fg-abc123def456 3h 2026-04-30T11:33:10+00:00 present +fg-def456abc789 1d 2026-04-29T14:12:55+00:00 present ``` +`--output-format json` returns a structured envelope (`{"success": true, "pending": [...], "count": 2}`) so CI can filter the queue programmatically. + ```shell -$ forgelm approvals --show abc123 -... full artifact summary including audit, benchmarks, safety, model card ... +$ forgelm approvals --show fg-abc123def456 --output-dir checkpoints/ +Run: fg-abc123def456 +Status: pending + +Audit chain (oldest first): + [2026-04-30T11:33:10+00:00] human_approval.required — require_human_approval=true + +Staging contents (4 entries): + - adapter_config.json + - adapter_model.safetensors + - tokenizer.json + - tokenizer_config.json ``` +A `--show` against a granted / rejected run prints the full timeline (request → decision) plus the final approver and comment. `--show` against an unknown `run_id` exits 1 with a clear error. + ## Common pitfalls :::warn diff --git a/docs/usermanuals/tr/compliance/human-oversight.md b/docs/usermanuals/tr/compliance/human-oversight.md index d951916d..a67b2ff4 100644 --- a/docs/usermanuals/tr/compliance/human-oversight.md +++ b/docs/usermanuals/tr/compliance/human-oversight.md @@ -127,18 +127,37 @@ Varsayılan 48 saat. "Timeout yok — sonsuza kadar bekle" için 0 (CI'da öneri ## Bekleyen koşuları inceleme +`forgelm approvals`, `approve` / `reject` komutlarının tamamlayıcısıdır: `--output-dir` altındaki audit log'u tarar ve `human_approval.required` event'i için terminal kararı (granted/rejected) bulunmayan tüm koşuları raporlar. + ```shell -$ forgelm approvals --pending -RUN_ID REQUESTED_AT ARTIFACTS EXPIRES -abc123 2026-04-29T14:33Z checkpoints/run/artifacts/ 47s sonra -def456 2026-04-29T09:12Z checkpoints/sft-only/artifacts/ 42s sonra +$ forgelm approvals --pending --output-dir checkpoints/ +Pending approvals (2): + +RUN_ID AGE REQUESTED_AT STAGING +---------------- ---- ------------------------- ------- +fg-abc123def456 3h 2026-04-30T11:33:10+00:00 present +fg-def456abc789 1d 2026-04-29T14:12:55+00:00 present ``` +`--output-format json` yapısal bir zarf döner (`{"success": true, "pending": [...], "count": 2}`); CI bu çıktıyla kuyruğu programatik olarak filtreleyebilir. + ```shell -$ forgelm approvals --show abc123 -... audit, benchmark'lar, güvenlik, model card dahil tam artifact özeti ... +$ forgelm approvals --show fg-abc123def456 --output-dir checkpoints/ +Run: fg-abc123def456 +Status: pending + +Audit chain (oldest first): + [2026-04-30T11:33:10+00:00] human_approval.required — require_human_approval=true + +Staging contents (4 entries): + - adapter_config.json + - adapter_model.safetensors + - tokenizer.json + - tokenizer_config.json ``` +Granted / rejected bir koşu üzerinde `--show` çalıştırıldığında tam zaman çizelgesi (talep → karar) ve son onaylayan + yorum yazdırılır. Bilinmeyen bir `run_id` üzerinde `--show` net bir hata mesajıyla 1 koduyla çıkar. + ## Sık hatalar :::warn diff --git a/forgelm/cli/__init__.py b/forgelm/cli/__init__.py index 08007312..10531af5 100644 --- a/forgelm/cli/__init__.py +++ b/forgelm/cli/__init__.py @@ -80,6 +80,7 @@ # Parser (registrars + parse_args). from ._parser import ( + _add_approvals_subcommand, # noqa: F401 — re-export for tests _add_approve_subcommand, # noqa: F401 — re-export for tests _add_audit_subcommand, # noqa: F401 — re-export for tests _add_chat_subcommand, # noqa: F401 — re-export for tests @@ -113,6 +114,13 @@ # Wizard mode. from ._wizard import _maybe_run_wizard # noqa: F401 — re-export for tests +# Approvals listing subcommand (Phase 9 follow-up). +from .subcommands._approvals import ( + _collect_pending_runs, # noqa: F401 — re-export for tests + _collect_run_audit_chain, # noqa: F401 — re-export for tests + _run_approvals_cmd, # noqa: F401 — re-export for tests +) + # Approve / reject subcommands (Article 14 human-approval gate). from .subcommands._approve import ( _atomic_rename_or_move, # noqa: F401 — re-export for tests @@ -123,6 +131,7 @@ _resolve_approver_identity, # noqa: F401 — re-export for tests _run_approve_cmd, # noqa: F401 — re-export for tests _run_reject_cmd, # noqa: F401 — re-export for tests + _staging_path_inside_output_dir, # noqa: F401 — re-export for tests + _approvals reuse ) # Audit subcommand (+ legacy --data-audit worker). diff --git a/forgelm/cli/_dispatch.py b/forgelm/cli/_dispatch.py index 4a7bad3f..6b185a96 100644 --- a/forgelm/cli/_dispatch.py +++ b/forgelm/cli/_dispatch.py @@ -70,6 +70,9 @@ def _dispatch_subcommand(command: str, args) -> None: elif command == "reject": _cli_facade._run_reject_cmd(args, getattr(args, "output_format", "text")) sys.exit(EXIT_SUCCESS) + elif command == "approvals": + _cli_facade._run_approvals_cmd(args, getattr(args, "output_format", "text")) + sys.exit(EXIT_SUCCESS) else: logger.error("Unrecognized subcommand: %r. This is a bug — please report it.", command) sys.exit(EXIT_TRAINING_ERROR) diff --git a/forgelm/cli/_parser.py b/forgelm/cli/_parser.py index 55bc2098..b5b3eb0b 100644 --- a/forgelm/cli/_parser.py +++ b/forgelm/cli/_parser.py @@ -499,6 +499,49 @@ def _add_approve_subcommand(subparsers) -> None: _add_common_subparser_flags(p, include_output_format=True) +def _add_approvals_subcommand(subparsers) -> None: + """Article 14 follow-up: list / inspect pending approval requests. + + Exactly one of ``--pending`` and ``--show RUN_ID`` must be set; argparse + enforces this with a mutually-exclusive group so the dispatcher only + sees a validated args namespace. ``--output-dir`` is required in both + modes — there is no useful default; an operator running on a freshly- + cloned workstation must point at the training output explicitly. + """ + p = subparsers.add_parser( + "approvals", + help="List pending Article 14 approval requests or inspect a single run.", + description=( + "Discovery counterpart to `forgelm approve` / `forgelm reject`. " + "`--pending` lists every run whose audit log carries a " + "`human_approval.required` event without a matching terminal " + "decision. `--show RUN_ID` prints the full approval-gate audit " + "chain plus the on-disk staging directory layout for one run." + ), + ) + mode = p.add_mutually_exclusive_group(required=True) + mode.add_argument( + "--pending", + action="store_true", + help="List every run awaiting an approval decision.", + ) + mode.add_argument( + "--show", + type=str, + default=None, + metavar="RUN_ID", + help="Show the full audit chain + staging contents for a single run.", + ) + p.add_argument( + "--output-dir", + type=str, + required=True, + metavar="DIR", + help="Training output directory containing audit_log.jsonl and final_model.staging/.", + ) + _add_common_subparser_flags(p, include_output_format=True) + + def _add_reject_subcommand(subparsers) -> None: """Article 14: discard a staged model (preserves staging dir for forensics).""" p = subparsers.add_parser( @@ -563,6 +606,7 @@ def parse_args(): _add_verify_audit_subcommand(subparsers) _add_approve_subcommand(subparsers) _add_reject_subcommand(subparsers) + _add_approvals_subcommand(subparsers) # --- Top-level flags (training / config-driven mode) --- parser.add_argument("--config", type=str, help="Path to the YAML configuration file.") diff --git a/forgelm/cli/subcommands/_approvals.py b/forgelm/cli/subcommands/_approvals.py new file mode 100644 index 00000000..78588a10 --- /dev/null +++ b/forgelm/cli/subcommands/_approvals.py @@ -0,0 +1,427 @@ +"""``forgelm approvals`` listing subcommand (Article 14 follow-up). + +The Phase 9 closure shipped ``forgelm approve`` and ``forgelm reject`` so an +operator can act on a single staged run, but it left the *discovery* step +out: an operator who walks up to a workstation cold has no way to ask +"which runs are awaiting my review?". This module fills the gap. + +Two query modes: + +- ``forgelm approvals --pending [--output-dir DIR]`` lists every run whose + audit log carries a ``human_approval.required`` event without a matching + terminal decision (``human_approval.granted`` / ``human_approval.rejected``). +- ``forgelm approvals --show RUN_ID --output-dir DIR`` prints the full + approval-related audit chain for a single run plus the on-disk staging + directory layout. + +The audit-log JSONL parser (skip malformed, emit one summary warning at +end) lives in :mod:`._audit_log_reader` and is shared with +:mod:`._approve` (and the upcoming :mod:`._purge`). The +``output_dir/audit_log.jsonl`` convention + Article 14 event vocabulary +live here so the dispatcher stays cohesive; only the *parser* is shared. +""" + +from __future__ import annotations + +import json +import os +import sys +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +from .._exit_codes import EXIT_CONFIG_ERROR, EXIT_SUCCESS, EXIT_TRAINING_ERROR +from .._logging import logger + +# Audit event names. Mirror :mod:`._approve` so a future rename of one set +# can't drift across the listing dispatcher and the decision dispatcher. +_EVT_HUMAN_APPROVAL_REQUIRED = "human_approval.required" +_EVT_HUMAN_APPROVAL_GRANTED = "human_approval.granted" +_EVT_HUMAN_APPROVAL_REJECTED = "human_approval.rejected" +_TERMINAL_DECISION_EVENTS = frozenset({_EVT_HUMAN_APPROVAL_GRANTED, _EVT_HUMAN_APPROVAL_REJECTED}) + +# Default audit-log filename. Centralised so a future rename does not have to +# touch every dispatcher. +_AUDIT_LOG_FILENAME = "audit_log.jsonl" + + +def _output_error_and_exit(output_format: str, msg: str, exit_code: int) -> None: + """Emit ``msg`` as a structured JSON error or a log record, then exit. + + Mirrors :func:`forgelm.cli.subcommands._approve._output_error_and_exit` + so the JSON envelope contract is identical across the approval family + of subcommands. + """ + if output_format == "json": + print(json.dumps({"success": False, "error": msg})) + else: + logger.error(msg) + sys.exit(exit_code) + + +# Wave 2a Round-1 review (XPR-01): the JSONL parser was extracted into +# the shared :mod:`._audit_log_reader` module so a future malformed-line +# policy fix lands in one place across the approve / approvals / purge +# family. The local re-export below preserves the import name tests +# already use (``from forgelm.cli import _iter_audit_events``). +from ._audit_log_reader import iter_audit_events as _iter_audit_events # noqa: F401,E402 + + +def _collect_pending_runs(audit_log_path: str) -> List[Dict[str, Any]]: + """Return the list of pending approval requests, newest-first. + + A "pending" run is one whose audit log carries a + ``human_approval.required`` event whose ``run_id`` does **not** appear + in any *later* ``human_approval.granted`` / ``human_approval.rejected`` + event after that requirement was issued. + + **Latest-wins semantics** (Wave 2a Round-1 review F-25-03 fix): when a + run is restarted with the same ``run_id`` after a prior decision — + e.g. a rejected run is re-staged for a second review — the latest + ``human_approval.required`` event reflects the current pending state + and earlier terminal decisions for that ``run_id`` are no longer the + operative status. Mirrors :func:`._approve._find_human_approval_decision_event`'s + "most-recent matching event wins" semantic so the family stays + consistent. + + Implementation: walk the log in append order, recording the latest + ``required`` and the latest terminal decision per ``run_id``. A run + is pending iff its last ``required`` event came strictly after its + last terminal decision (or no terminal decision exists). Line-number + fallback is used when timestamp is missing so re-imported logs without + timestamps still have a deterministic ordering. + """ + latest_required: Dict[str, tuple[int, Dict[str, Any]]] = {} + latest_decision_line: Dict[str, int] = {} + + for line_no, event in _iter_audit_events(audit_log_path): + run_id = event.get("run_id") + if not isinstance(run_id, str): + # Required field on every approval-gate event; if missing the + # event is malformed and we cannot reason about it. Skip. + continue + event_name = event.get("event") + if event_name == _EVT_HUMAN_APPROVAL_REQUIRED: + latest_required[run_id] = (line_no, event) + elif event_name in _TERMINAL_DECISION_EVENTS: + latest_decision_line[run_id] = line_no + + pending = [ + event for run_id, (req_line, event) in latest_required.items() if req_line > latest_decision_line.get(run_id, 0) + ] + # Newest-pending first so an operator opening the list sees the most + # recent request at the top. ``timestamp`` may be missing on + # synthetic / hand-edited entries — sort missing values to the bottom. + pending.sort(key=lambda e: e.get("timestamp") or "", reverse=True) + return pending + + +def _collect_run_audit_chain(audit_log_path: str, run_id: str) -> List[Dict[str, Any]]: + """Return every approval-gate event for ``run_id`` in append order. + + Used by ``--show`` to render the full request → decision chain. We + return the events in their on-disk order (which is also wall-clock + order since the audit log is append-only) so an operator scanning the + output can read top-to-bottom as the timeline. + """ + chain: List[Dict[str, Any]] = [] + approval_events = {_EVT_HUMAN_APPROVAL_REQUIRED, *_TERMINAL_DECISION_EVENTS} + for _line_no, event in _iter_audit_events(audit_log_path): + if event.get("event") in approval_events and event.get("run_id") == run_id: + chain.append(event) + return chain + + +def _staging_dir_for_event(output_dir: str, event: Dict[str, Any]) -> Optional[str]: + """Best-effort: return the on-disk staging path the event refers to. + + The trainer records ``staging_path`` on every ``human_approval.required`` + event (Wave 1 round-2 audit-event vocabulary). Older audit logs may + pre-date that addition; for those we synthesise the canonical path + ``/final_model.staging`` and let the caller decide whether + to fall back when ``os.path.isdir`` is False. + + **Defence in depth (Wave 2a Round-1 review fix):** when an audit log + lives on shared / unsigned storage, an attacker who can append events + could plant a ``staging_path`` value pointing outside ``output_dir`` + (e.g. ``/etc``). ``_staging_contents`` would then leak directory + listings as a tampered-audit-log oracle. We reuse the + ``_staging_path_inside_output_dir`` guard Phase 9's :mod:`._approve` + module already ships and refuse declared paths that escape the + operator-supplied ``output_dir`` boundary. + """ + # Late import via the package facade so the helper resolves through + # the same monkeypatch surface tests already use for _approve.py. + from forgelm import cli as _cli_facade + + declared = event.get("staging_path") + if isinstance(declared, str) and declared: + if not _cli_facade._staging_path_inside_output_dir(declared, output_dir): + logger.warning( + "Refusing staging_path %r from audit event for output_dir %r: " + "resolved path escapes the output_dir boundary (audit-log tampering " + "guard). Falling back to canonical final_model.staging if present.", + declared, + output_dir, + ) + declared = None + if isinstance(declared, str) and declared: + return declared + fallback = os.path.join(output_dir, "final_model.staging") + return fallback if os.path.isdir(fallback) else None + + +def _age_seconds(event_timestamp: Optional[str]) -> Optional[float]: + """Compute ``now - timestamp`` in seconds; ``None`` when unparseable. + + The audit log emits ISO-8601 with a UTC suffix (``+00:00``). We accept + a missing or malformed timestamp gracefully because hand-edited test + fixtures sometimes drop it, and a missing age is far less actionable + than a stack trace. + """ + if not isinstance(event_timestamp, str) or not event_timestamp: + return None + try: + # Python 3.10 ``datetime.fromisoformat`` does not accept the ``Z`` + # suffix. Normalise to ``+00:00`` for older interpreters. + normalised = event_timestamp.replace("Z", "+00:00") + when = datetime.fromisoformat(normalised) + except ValueError: + return None + if when.tzinfo is None: + # Treat naive timestamps as UTC; mirrors how AuditLogger emits them + # but is defensive against external producers. + when = when.replace(tzinfo=timezone.utc) + delta = datetime.now(timezone.utc) - when + return delta.total_seconds() + + +def _staging_contents(staging_path: Optional[str]) -> List[str]: + """List files in ``staging_path`` (sorted) or ``[]`` when missing/unreadable.""" + if not staging_path or not os.path.isdir(staging_path): + return [] + try: + return sorted(os.listdir(staging_path)) + except OSError as exc: + logger.warning("Cannot list staging directory %s: %s", staging_path, exc) + return [] + + +def _summarise_pending(event: Dict[str, Any], output_dir: str) -> Dict[str, Any]: + """Build the per-run dict used in ``--pending`` output (text + JSON).""" + staging_path = _staging_dir_for_event(output_dir, event) + timestamp = event.get("timestamp") + return { + "run_id": event.get("run_id"), + "staging_path": staging_path, + "staging_exists": bool(staging_path and os.path.isdir(staging_path)), + "requested_at": timestamp, + "age_seconds": _age_seconds(timestamp), + "metrics": event.get("metrics") or {}, + "config_hash": event.get("config_hash") or event.get("config_fingerprint"), + "reason": event.get("reason"), + } + + +def _format_age(age_seconds: Optional[float]) -> str: + """Render ``age_seconds`` as a human-friendly ``Nh Mm`` / ``Nd`` string.""" + if age_seconds is None: + return "unknown" + if age_seconds < 60: + return f"{int(age_seconds)}s" + if age_seconds < 3600: + return f"{int(age_seconds // 60)}m" + if age_seconds < 86400: + hours = int(age_seconds // 3600) + minutes = int((age_seconds % 3600) // 60) + return f"{hours}h {minutes}m" if minutes else f"{hours}h" + days = int(age_seconds // 86400) + return f"{days}d" + + +def _emit_pending_text(pending_summaries: List[Dict[str, Any]]) -> None: + """Print a tabular pending list (or a friendly empty notice).""" + if not pending_summaries: + print("No pending approvals.") + return + print(f"Pending approvals ({len(pending_summaries)}):") + print() + # Tabular layout. Plain text (no rich) so the output stays usable in + # CI logs, redirected files, and SSH sessions without colour support. + headers = ("RUN_ID", "AGE", "REQUESTED_AT", "STAGING") + rows = [ + ( + summary["run_id"] or "(missing run_id)", + _format_age(summary["age_seconds"]), + summary["requested_at"] or "(unknown)", + "present" if summary["staging_exists"] else "MISSING", + ) + for summary in pending_summaries + ] + widths = [max(len(h), max(len(r[i]) for r in rows)) for i, h in enumerate(headers)] + fmt = " ".join(f"{{:<{w}}}" for w in widths) + print(fmt.format(*headers)) + print(fmt.format(*("-" * w for w in widths))) + for row in rows: + print(fmt.format(*row)) + + +def _emit_pending_json(pending_summaries: List[Dict[str, Any]]) -> None: + """Print the pending list as a JSON array of summaries.""" + print( + json.dumps( + {"success": True, "pending": pending_summaries, "count": len(pending_summaries)}, + indent=2, + ) + ) + + +def _run_approvals_list_pending(args, output_format: str) -> None: + """Handle ``forgelm approvals --pending`` end-to-end.""" + output_dir = args.output_dir + audit_log_path = os.path.join(output_dir, _AUDIT_LOG_FILENAME) + if not os.path.isfile(audit_log_path): + # Treat missing audit log the same as "no pending approvals" rather + # than as an error — a freshly-bootstrapped output dir legitimately + # has neither yet. Operators get an informative empty summary. + if output_format == "json": + _emit_pending_json([]) + else: + print(f"No audit log at {audit_log_path}; nothing to list.") + sys.exit(EXIT_SUCCESS) + + try: + pending_events = _collect_pending_runs(audit_log_path) + except OSError as exc: + _output_error_and_exit( + output_format, + f"Failed to scan audit log {audit_log_path!r}: {exc}", + EXIT_TRAINING_ERROR, + ) + + summaries = [_summarise_pending(event, output_dir) for event in pending_events] + if output_format == "json": + _emit_pending_json(summaries) + else: + _emit_pending_text(summaries) + sys.exit(EXIT_SUCCESS) + + +def _classify_chain(chain: List[Dict[str, Any]]) -> str: + """Return ``pending`` / ``granted`` / ``rejected`` / ``unknown``.""" + decisions = [e.get("event") for e in chain if e.get("event") in _TERMINAL_DECISION_EVENTS] + if not decisions: + # If we have a `required` but no decision, it is pending. If we + # have neither, the run is unknown to the audit log. + if any(e.get("event") == _EVT_HUMAN_APPROVAL_REQUIRED for e in chain): + return "pending" + return "unknown" + # Most-recent decision wins (the chain is already in append order). + last = decisions[-1] + if last == _EVT_HUMAN_APPROVAL_GRANTED: + return "granted" + return "rejected" + + +def _emit_show_text(run_id: str, chain: List[Dict[str, Any]], status: str, staging_listing: List[str]) -> None: + """Render ``--show RUN_ID`` output as a human-readable timeline.""" + print(f"Run: {run_id}") + print(f"Status: {status}") + print() + print("Audit chain (oldest first):") + if not chain: + print(" (no approval-gate events found)") + for event in chain: + ts = event.get("timestamp", "(no timestamp)") + name = event.get("event", "(no event name)") + # Render the most operator-relevant fields; full payload is in JSON + # output below. + approver = event.get("approver") + comment = event.get("comment") + reason = event.get("reason") + line = f" [{ts}] {name}" + if approver: + line += f" by {approver}" + if reason: + line += f" — {reason}" + if comment: + line += f" ({comment})" + print(line) + print() + if staging_listing: + print(f"Staging contents ({len(staging_listing)} entries):") + for entry in staging_listing: + print(f" - {entry}") + else: + print("Staging directory: missing or empty.") + + +def _emit_show_json(run_id: str, chain: List[Dict[str, Any]], status: str, staging_listing: List[str]) -> None: + """Render ``--show RUN_ID`` output as a structured JSON object.""" + print( + json.dumps( + { + "success": True, + "run_id": run_id, + "status": status, + "chain": chain, + "staging_contents": staging_listing, + }, + indent=2, + ) + ) + + +def _run_approvals_show(args, output_format: str) -> None: + """Handle ``forgelm approvals --show `` end-to-end.""" + run_id: str = args.show + output_dir = args.output_dir + audit_log_path = os.path.join(output_dir, _AUDIT_LOG_FILENAME) + if not os.path.isfile(audit_log_path): + _output_error_and_exit( + output_format, + f"No audit log at {audit_log_path!r}; cannot show run {run_id!r}.", + EXIT_CONFIG_ERROR, + ) + + chain = _collect_run_audit_chain(audit_log_path, run_id) + if not chain: + _output_error_and_exit( + output_format, + f"No approval-gate events found for run_id={run_id!r} in {audit_log_path!r}.", + EXIT_CONFIG_ERROR, + ) + + status = _classify_chain(chain) + # Staging path is read from the *first* required event so a + # post-rename layout (final/ exists, staging/ deleted) still surfaces + # the originally-staged directory name. + required_event = next((e for e in chain if e.get("event") == _EVT_HUMAN_APPROVAL_REQUIRED), {}) + staging_path = _staging_dir_for_event(output_dir, required_event) if required_event else None + staging_listing = _staging_contents(staging_path) + + if output_format == "json": + _emit_show_json(run_id, chain, status, staging_listing) + else: + _emit_show_text(run_id, chain, status, staging_listing) + sys.exit(EXIT_SUCCESS) + + +def _run_approvals_cmd(args, output_format: str) -> None: + """Top-level dispatcher for ``forgelm approvals``. + + Exactly one of ``--pending`` / ``--show RUN_ID`` must be set; argparse + enforces the mutual exclusion via the parser registrar so this + function only sees a valid args namespace. + """ + show_run_id = getattr(args, "show", None) + if show_run_id: + _run_approvals_show(args, output_format) + elif getattr(args, "pending", False): + _run_approvals_list_pending(args, output_format) + else: + # argparse should have prevented this; defensive double-check. + _output_error_and_exit( + output_format, + "forgelm approvals: one of --pending / --show RUN_ID is required.", + EXIT_CONFIG_ERROR, + ) diff --git a/forgelm/cli/subcommands/_approve.py b/forgelm/cli/subcommands/_approve.py index 22062d54..cc2e15dd 100644 --- a/forgelm/cli/subcommands/_approve.py +++ b/forgelm/cli/subcommands/_approve.py @@ -81,40 +81,19 @@ def _resolve_approver_identity() -> str: def _find_human_approval_required_event(audit_log_path: str, run_id: str) -> Optional[dict]: """Return the most-recent ``human_approval.required`` event for *run_id*. - Reads ``audit_log.jsonl`` line-by-line to keep memory usage flat for - long-lived training directories. Returns ``None`` when no matching event - exists. Malformed lines are skipped and counted; the operator gets a - warning if any lines were skipped so a corrupt entry can't silently mask - a genuine "no event" result. + Wave 2a Round-1 review consolidated the audit-log JSONL parser into + :mod:`._audit_log_reader` so a future malformed-line policy fix lands + in one place. This helper now delegates; the original line-by-line + streaming + skipped-line warning behaviour is preserved exactly + (verified by the existing approve / reject test suites). """ - if not os.path.isfile(audit_log_path): - return None + from ._audit_log_reader import find_latest_event_for_run - latest_match = None - skipped_lines = 0 - try: - fh = open(audit_log_path, "r", encoding="utf-8") - except OSError as exc: - logger.error("Cannot open audit log %s: %s", audit_log_path, exc) - return None - with fh: - for raw in fh: - line = raw.strip() - if not line: - continue - try: - event = json.loads(line) - except json.JSONDecodeError: - skipped_lines += 1 - continue - if not isinstance(event, dict): - skipped_lines += 1 - continue - if event.get("event") == "human_approval.required" and event.get("run_id") == run_id: - latest_match = event - if skipped_lines: - logger.warning("Skipped %d malformed line(s) while parsing %s.", skipped_lines, audit_log_path) - return latest_match + return find_latest_event_for_run( + audit_log_path, + run_id=run_id, + matches=lambda e: e.get("event") == "human_approval.required", + ) _TERMINAL_DECISION_EVENTS = frozenset({_EVT_HUMAN_APPROVAL_GRANTED, _EVT_HUMAN_APPROVAL_REJECTED}) @@ -127,38 +106,18 @@ def _find_human_approval_decision_event(audit_log_path: str, run_id: str) -> Opt ``human_approval.rejected``. Finding one before attempting promotion prevents double-approve and approve-after-reject races. - Mirrors :func:`_find_human_approval_required_event`'s malformed-line - accounting so a corrupt entry cannot silently mask a real terminal - decision and let the caller re-promote. + Same Wave 2a Round-1 consolidation as + :func:`_find_human_approval_required_event` — delegates to the shared + :mod:`._audit_log_reader` so the malformed-line policy lives in + one place. """ - if not os.path.isfile(audit_log_path): - return None + from ._audit_log_reader import find_latest_event_for_run - latest_decision = None - skipped_lines = 0 - try: - fh = open(audit_log_path, "r", encoding="utf-8") - except OSError as exc: - logger.error("Cannot open audit log %s: %s", audit_log_path, exc) - return None - with fh: - for raw in fh: - line = raw.strip() - if not line: - continue - try: - event = json.loads(line) - except json.JSONDecodeError: - skipped_lines += 1 - continue - if not isinstance(event, dict): - skipped_lines += 1 - continue - if event.get("event") in _TERMINAL_DECISION_EVENTS and event.get("run_id") == run_id: - latest_decision = event - if skipped_lines: - logger.warning("Skipped %d malformed line(s) while parsing %s.", skipped_lines, audit_log_path) - return latest_decision + return find_latest_event_for_run( + audit_log_path, + run_id=run_id, + matches=lambda e: e.get("event") in _TERMINAL_DECISION_EVENTS, + ) def _atomic_rename_or_move(src: str, dst: str) -> str: diff --git a/forgelm/cli/subcommands/_audit_log_reader.py b/forgelm/cli/subcommands/_audit_log_reader.py new file mode 100644 index 00000000..39b06df9 --- /dev/null +++ b/forgelm/cli/subcommands/_audit_log_reader.py @@ -0,0 +1,118 @@ +"""Single-source audit-log JSONL reader for the approval / approvals / purge family. + +Background — Wave 2a Round-1 review (XPR-01) flagged that ForgeLM had three +near-identical JSONL parsers across :mod:`._approve` and :mod:`._approvals`, +with a fourth coming in Phase 21 (``forgelm purge``). Each copy duplicated +the malformed-line accounting, the OSError handling, and the +``isinstance(event, dict)`` guard — so a future fix to any one of those +policies (e.g. switching to streaming reads, adding a maximum line length, +emitting structured error events) would have to be applied in ``N`` +places at once. + +This module is the single place that policy lives. Two helpers cover the +two read patterns the family uses: + +- :func:`iter_audit_events` — generator yielding ``(line_number, event)`` + pairs. Skips blank lines + malformed JSON + non-dict roots silently; + emits one summary warning at the end. Used by anything that needs to + walk the whole log (e.g. ``forgelm approvals --pending``). +- :func:`find_latest_event_for_run` — convenience wrapper that returns the + most-recent event matching a predicate, or ``None``. Used by the + approve / reject decision-guard checks. + +Both helpers route OSError on file-open through the module logger and +return an empty result so the caller sees a missing log as "no matching +event" rather than as a crash. +""" + +from __future__ import annotations + +import json +import logging +import os +from typing import Any, Callable, Dict, Iterator, Optional, Tuple + +logger = logging.getLogger("forgelm.cli.audit_log_reader") + + +def iter_audit_events(audit_log_path: str) -> Iterator[Tuple[int, Dict[str, Any]]]: + """Yield ``(line_number, event_dict)`` from an append-only JSONL audit log. + + Skips blank lines and malformed entries (non-JSON lines, JSON whose + root is not a ``dict``) silently — emitting a per-skip warning here + would be very noisy on a long-lived audit log. The number of skipped + lines is summarised in a single ``logger.warning`` call when iteration + finishes (so callers learn about corruption without paying per-line + log cost). + + Returns nothing (no events yielded, no warning) when: + - the file does not exist (a freshly-bootstrapped output dir + legitimately has no audit log yet); or + - the file cannot be opened (OSError logged at ERROR level so + operators see why nothing came through). + + The caller is responsible for closing nothing — this generator owns + the file handle via ``with`` and releases it on iterator exhaustion. + """ + if not os.path.isfile(audit_log_path): + return + try: + fh = open(audit_log_path, "r", encoding="utf-8") + except OSError as exc: + logger.error("Cannot open audit log %s: %s", audit_log_path, exc) + return + skipped_lines = 0 + with fh: + for line_number, raw in enumerate(fh, start=1): + line = raw.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + skipped_lines += 1 + continue + if not isinstance(event, dict): + skipped_lines += 1 + continue + yield line_number, event + if skipped_lines: + logger.warning( + "Skipped %d malformed line(s) while parsing %s.", + skipped_lines, + audit_log_path, + ) + + +def find_latest_event_for_run( + audit_log_path: str, + *, + run_id: str, + matches: Callable[[Dict[str, Any]], bool], +) -> Optional[Dict[str, Any]]: + """Return the most-recent event matching ``matches`` for ``run_id``. + + Walks the entire log (audit logs are append-only so "most recent" + means the last matching entry encountered). Returns ``None`` when + no event matches. + + ``matches`` is the predicate the caller cares about — typically + ``lambda e: e.get("event") == "human_approval.required"`` or a + membership check against a frozenset of decision-event names. + Keeping the predicate as a callable lets every caller in the + approve / approvals / purge family share the parser without + couping it to a specific event vocabulary. + """ + latest: Optional[Dict[str, Any]] = None + for _line_no, event in iter_audit_events(audit_log_path): + if event.get("run_id") != run_id: + continue + if matches(event): + latest = event + return latest + + +__all__ = [ + "iter_audit_events", + "find_latest_event_for_run", +] diff --git a/tests/test_approvals_listing.py b/tests/test_approvals_listing.py new file mode 100644 index 00000000..2459be5d --- /dev/null +++ b/tests/test_approvals_listing.py @@ -0,0 +1,526 @@ +"""Phase 37: ``forgelm approvals`` listing + show subcommand. + +Mirrors the ``test_human_approval_gate.py`` style — synthetic JSONL audit +log fixtures so the tests stay torch-free and run on every CI matrix combo. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# --------------------------------------------------------------------------- +# Helpers (mirror tests/test_human_approval_gate.py) +# --------------------------------------------------------------------------- + + +def _write_event(audit_path: Path, event: str, run_id: str, **fields: object) -> None: + """Append a synthetic event to ``audit_path``. + + Produces the same JSONL shape AuditLogger writes: one event per line, + a ``timestamp`` / ``operator`` / ``event`` / ``run_id`` core, and + whatever extra fields the caller passes (``staging_path``, ``metrics``, + ``approver``, ``comment`` etc.). + """ + entry = { + "timestamp": "2026-04-30T12:00:00+00:00", + "operator": "tester", + "event": event, + "run_id": run_id, + "prev_hash": "genesis", + } + entry.update(fields) + audit_path.parent.mkdir(parents=True, exist_ok=True) + with open(audit_path, "a", encoding="utf-8") as fh: + fh.write(json.dumps(entry) + "\n") + + +def _seed_run( + output_dir: Path, + run_id: str, + *, + decision: str | None = None, + create_staging: bool = True, +) -> Path: + """Create a run layout: staging dir + ``human_approval.required`` event. + + When ``decision`` is given (``"granted"`` / ``"rejected"``), append the + matching terminal event so the run is no longer pending. + """ + output_dir.mkdir(parents=True, exist_ok=True) + staging_dir = output_dir / "final_model.staging" + if create_staging: + staging_dir.mkdir(exist_ok=True) + (staging_dir / "adapter_config.json").write_text('{"r": 8}', encoding="utf-8") + audit_path = output_dir / "audit_log.jsonl" + _write_event( + audit_path, + "human_approval.required", + run_id, + staging_path=str(staging_dir), + gate="final_model", + reason="require_human_approval=true", + metrics={"eval_loss": 0.42}, + ) + if decision == "granted": + _write_event( + audit_path, + "human_approval.granted", + run_id, + approver="alice", + comment="LGTM", + promote_strategy="rename", + ) + elif decision == "rejected": + _write_event( + audit_path, + "human_approval.rejected", + run_id, + approver="bob", + comment="regression", + staging_path=str(staging_dir), + ) + return staging_dir + + +def _build_args(output_dir: Path, *, pending: bool = False, show: str | None = None) -> MagicMock: + """Build a MagicMock ``args`` namespace mimicking argparse output.""" + args = MagicMock() + args.pending = pending + args.show = show + args.output_dir = str(output_dir) + return args + + +# --------------------------------------------------------------------------- +# --pending mode +# --------------------------------------------------------------------------- + + +class TestApprovalsPending: + def test_empty_audit_log_returns_empty_list(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + output_dir.mkdir() + # No audit_log.jsonl on disk → friendly empty path, exit 0. + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="text") + assert ei.value.code == 0 + captured = capsys.readouterr().out + assert "nothing to list" in captured.lower() or "no pending" in captured.lower() + + def test_three_runs_one_decided_two_pending(self, tmp_path: Path, capsys) -> None: + # Three runs, one already granted → listing returns the other two. + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-aaaa00000000") + _seed_run(output_dir, "fg-bbbb00000000", decision="granted") + _seed_run(output_dir, "fg-cccc00000000") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json") + assert ei.value.code == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is True + assert payload["count"] == 2 + run_ids = {summary["run_id"] for summary in payload["pending"]} + assert run_ids == {"fg-aaaa00000000", "fg-cccc00000000"} + + def test_rejected_run_also_excluded_from_pending(self, tmp_path: Path, capsys) -> None: + # Reject (like granted) is a terminal decision; rejected runs must + # not appear in --pending even though their staging dir lingers. + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-rejected0000", decision="rejected") + _seed_run(output_dir, "fg-pending00000") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json") + assert ei.value.code == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["count"] == 1 + assert payload["pending"][0]["run_id"] == "fg-pending00000" + + def test_pending_summary_carries_metrics_and_staging_path(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + staging_dir = _seed_run(output_dir, "fg-zzzz00000000") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json") + payload = json.loads(capsys.readouterr().out) + summary = payload["pending"][0] + assert summary["staging_path"] == str(staging_dir) + assert summary["staging_exists"] is True + assert summary["metrics"] == {"eval_loss": 0.42} + assert summary["reason"] == "require_human_approval=true" + + def test_text_output_renders_table(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-xyz000000000") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="text") + out = capsys.readouterr().out + # Table headers + run id present. + assert "RUN_ID" in out + assert "fg-xyz000000000" in out + # The "present" / "MISSING" staging column is rendered. + assert "present" in out + + +# --------------------------------------------------------------------------- +# --show mode +# --------------------------------------------------------------------------- + + +class TestApprovalsShow: + def test_show_pending_run_emits_full_chain(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-pp0000000000") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-pp0000000000"), output_format="json") + assert ei.value.code == 0 + + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is True + assert payload["status"] == "pending" + assert len(payload["chain"]) == 1 + assert payload["chain"][0]["event"] == "human_approval.required" + # Staging dir was seeded, so contents include adapter_config.json. + assert "adapter_config.json" in payload["staging_contents"] + + def test_show_granted_run_status_granted(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-gr0000000000", decision="granted") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-gr0000000000"), output_format="json") + assert ei.value.code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "granted" + # Two events: required + granted. + assert len(payload["chain"]) == 2 + assert payload["chain"][-1]["event"] == "human_approval.granted" + + def test_show_unknown_run_id_exits_1(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-real00000000") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-fake00000000"), output_format="text") + assert ei.value.code == 1 + # Error message names the missing run id. + # (Captured in stderr via the logger; capsys captures stdout for the + # JSON path but the text path goes through ``logger.error``; assert + # via caplog instead in a dedicated test below.) + + def test_show_no_audit_log_exits_1(self, tmp_path: Path) -> None: + output_dir = tmp_path / "run" + output_dir.mkdir() + # No audit_log.jsonl on disk; --show cannot operate. + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-anything0000"), output_format="text") + assert ei.value.code == 1 + + def test_show_text_renders_timeline(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-tt0000000000", decision="granted") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, show="fg-tt0000000000"), output_format="text") + out = capsys.readouterr().out + assert "fg-tt0000000000" in out + assert "human_approval.required" in out + assert "human_approval.granted" in out + assert "by alice" in out # approver rendered + + +# --------------------------------------------------------------------------- +# Subcommand registration smoke +# --------------------------------------------------------------------------- + + +class TestApprovalsRegistration: + def test_approvals_subcommand_registered(self) -> None: + """``forgelm approvals --help`` must succeed (subparser exists).""" + import subprocess + import sys + + result = subprocess.run( + [sys.executable, "-m", "forgelm.cli", "approvals", "--help"], + capture_output=True, + text=True, + timeout=30, + ) + assert result.returncode == 0, result.stderr + # Both modes are advertised in the help text. + assert "--pending" in result.stdout + assert "--show" in result.stdout + assert "--output-dir" in result.stdout + + +# --------------------------------------------------------------------------- +# Robustness against malformed audit log lines +# --------------------------------------------------------------------------- + + +class TestApprovalsMalformedLines: + def test_skipped_lines_do_not_hide_pending_runs(self, tmp_path: Path, capsys) -> None: + # Simulate a corrupt audit log: a real `required` event followed by + # a malformed line and a JSON list (non-dict root). The pending + # listing must still surface the legitimate run. + output_dir = tmp_path / "run" + output_dir.mkdir() + audit_path = output_dir / "audit_log.jsonl" + _write_event( + audit_path, + "human_approval.required", + "fg-real00000000", + staging_path=str(output_dir / "final_model.staging"), + ) + with open(audit_path, "a", encoding="utf-8") as fh: + fh.write("not even json\n") + fh.write('["array", "root"]\n') + # Create the staging dir so staging_exists is True (defensive — not + # required for the test, just realistic). + (output_dir / "final_model.staging").mkdir() + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json") + payload = json.loads(capsys.readouterr().out) + assert payload["count"] == 1 + assert payload["pending"][0]["run_id"] == "fg-real00000000" + + +# --------------------------------------------------------------------------- +# Defensive: dispatcher rejects neither-mode args +# --------------------------------------------------------------------------- + + +class TestApprovalsDispatcherDefensive: + def test_neither_pending_nor_show_exits_1(self, tmp_path: Path) -> None: + """argparse normally enforces the mutex; this exercises the + defensive double-check inside the dispatcher in case a future + refactor drops the parser-level guard.""" + output_dir = tmp_path / "run" + output_dir.mkdir() + args = _build_args(output_dir, pending=False, show=None) + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(args, output_format="text") + assert ei.value.code == 1 + + +# --------------------------------------------------------------------------- +# Patch-based test for monkeypatch resolution discipline +# --------------------------------------------------------------------------- + + +class TestApprovalsMonkeyPatchSurface: + def test_helper_reachable_via_facade(self) -> None: + """``forgelm.cli._collect_pending_runs`` must resolve so tests can + ``patch("forgelm.cli._collect_pending_runs", ...)`` if they want.""" + from forgelm import cli as _cli_facade + + assert callable(_cli_facade._collect_pending_runs) + assert callable(_cli_facade._collect_run_audit_chain) + assert callable(_cli_facade._run_approvals_cmd) + + +@pytest.mark.usefixtures("caplog") +class TestApprovalsErrorMessages: + def test_unknown_run_logs_actionable_error(self, tmp_path: Path, caplog) -> None: + import logging + + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-real00000000") + + from forgelm.cli import _run_approvals_cmd + + with caplog.at_level(logging.ERROR, logger="forgelm.cli"): + with patch("forgelm.cli._collect_pending_runs"): # ensure not called by --show + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-fake00000000"), output_format="text") + assert ei.value.code == 1 + assert "fg-fake00000000" in caplog.text + + +# --------------------------------------------------------------------------- +# Wave 2a Round-1 review fixes — security + bot-suggested coverage +# --------------------------------------------------------------------------- + + +class TestApprovalsPathTraversalGuard: + """F-25-01: tampered audit log with staging_path outside output_dir + must not produce a directory-listing oracle via _staging_contents.""" + + def test_show_refuses_external_staging_path(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + output_dir.mkdir() + # Malicious audit event: staging_path points outside output_dir. + _write_event( + output_dir / "audit_log.jsonl", + "human_approval.required", + "fg-tampered00000", + staging_path="/etc", # would expose /etc listing if guard absent + ) + # Provide a benign fallback so the test can verify the listing + # is empty (not /etc/...). + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, show="fg-tampered00000"), output_format="json") + payload = json.loads(capsys.readouterr().out) + # The dispatcher must NOT have leaked /etc entries into staging_contents. + assert all("etc" not in entry for entry in payload["staging_contents"]), ( + f"path-traversal guard failed: leaked /etc entries: {payload['staging_contents']}" + ) + + def test_pending_refuses_external_staging_path(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + output_dir.mkdir() + _write_event( + output_dir / "audit_log.jsonl", + "human_approval.required", + "fg-tampered00000", + staging_path="/etc", + ) + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json") + payload = json.loads(capsys.readouterr().out) + # staging_path field falls back to None / canonical when the + # declared one is rejected; staging_exists must be False. + summary = payload["pending"][0] + assert summary["staging_exists"] is False + # staging_path is either None (no fallback found) or the + # canonical final_model.staging — never the rejected /etc. + assert summary["staging_path"] != "/etc" + + +class TestApprovalsLatestWinsRunIdReuse: + """F-25-03: latest-wins semantics for re-staged runs. + + A run that was rejected and then re-staged with the same run_id + (a second human_approval.required event) must show as PENDING + again; the older terminal decision is no longer operative.""" + + def test_re_staged_run_shows_as_pending(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-restaged00000", decision="rejected") + # Now re-stage: append another human_approval.required AFTER the + # rejection. Latest-wins → pending again. + _write_event( + output_dir / "audit_log.jsonl", + "human_approval.required", + "fg-restaged00000", + staging_path=str(output_dir / "final_model.staging"), + metrics={"eval_loss": 0.30}, + ) + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json") + payload = json.loads(capsys.readouterr().out) + run_ids = {summary["run_id"] for summary in payload["pending"]} + assert "fg-restaged00000" in run_ids + + +class TestApprovalsShowRejectedAndMissingStaging: + """Bot-suggested coverage: rejected --show + MISSING staging table.""" + + def test_show_rejected_run_status_rejected(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-rj0000000000", decision="rejected") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-rj0000000000"), output_format="json") + assert ei.value.code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "rejected" + # Two events: required + rejected. + assert len(payload["chain"]) == 2 + assert payload["chain"][-1]["event"] == "human_approval.rejected" + + def test_pending_table_renders_missing_when_staging_absent(self, tmp_path: Path, capsys) -> None: + # Seed without creating the staging directory. + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-nostaging000", create_staging=False) + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit): + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="text") + out = capsys.readouterr().out + assert "MISSING" in out + + def test_pending_empty_audit_log_json_envelope(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + output_dir.mkdir() + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json") + assert ei.value.code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload == {"success": True, "pending": [], "count": 0} + + def test_show_unknown_run_id_json_envelope(self, tmp_path: Path, capsys) -> None: + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-real00000000") + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-fake00000000"), output_format="json") + assert ei.value.code == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["success"] is False + assert "fg-fake00000000" in payload["error"] + + def test_show_robust_to_malformed_lines(self, tmp_path: Path, capsys) -> None: + # --show must skip malformed lines the same way --pending does. + output_dir = tmp_path / "run" + _seed_run(output_dir, "fg-real00000000") + with open(output_dir / "audit_log.jsonl", "a", encoding="utf-8") as fh: + fh.write("not even json\n") + fh.write('["array", "root"]\n') + + from forgelm.cli import _run_approvals_cmd + + with pytest.raises(SystemExit) as ei: + _run_approvals_cmd(_build_args(output_dir, show="fg-real00000000"), output_format="json") + assert ei.value.code == 0 + payload = json.loads(capsys.readouterr().out) + # Real human_approval.required event is still surfaced. + assert any(e.get("event") == "human_approval.required" for e in payload["chain"])