Repository navigation
Conversation
Closes the Phase 9 discovery gap. forgelm approve / reject ship with
Wave 1 but an operator who walks up to a workstation cold has no way
to ask "which runs are awaiting review?". forgelm approvals fills
the gap.
Two query modes (mutually exclusive, argparse-enforced):
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 or {"success": true, "pending": [...],
"count": N} JSON envelope.
forgelm approvals --show RUN_ID --output-dir DIR
Prints the full approval-gate audit chain (request -> terminal
decision) plus the on-disk staging directory listing. Status
classifies as pending / granted / rejected / unknown.
Implementation:
- forgelm/cli/subcommands/_approvals.py (new) — dispatcher + helpers.
Reuses the audit-log parsing pattern from _approve.py (iter events,
skip malformed, emit one summary warning at end).
- forgelm/cli/_parser.py — _add_approvals_subcommand registrar with
a mutually-exclusive --pending / --show RUN_ID group; --output-dir
required in both modes (no useful default).
- forgelm/cli/_dispatch.py — approvals branch.
- forgelm/cli/__init__.py — facade re-exports for tests / monkeypatch.
Operator docs:
- docs/usermanuals/en/compliance/human-oversight.md "Inspecting
pending runs" section rewritten to match the shipped output (table
format with RUN_ID / AGE / REQUESTED_AT / STAGING columns, the
--output-dir flag the dispatcher actually requires).
- docs/usermanuals/tr/compliance/human-oversight.md mirror updated
in the same shape.
Tests (tests/test_approvals_listing.py — 15 cases, all torch-free):
- empty audit log -> empty list, friendly notice
- three runs, one granted -> two pending in JSON envelope
- rejected run also excluded from pending
- pending summary carries metrics + staging_path + reason
- text output renders the table headers + run id + staging column
- --show pending run emits the chain + staging_contents
- --show granted run status == "granted", chain length == 2
- --show unknown run_id exits 1 with actionable error
- --show with no audit log exits 1
- text show renders the timeline + approver
- subcommand registered (forgelm approvals --help)
- malformed audit lines (non-json + non-dict-root) do not hide a
legitimate pending run
- dispatcher defensive: neither --pending nor --show -> exit 1
- monkeypatch facade resolution: forgelm.cli._collect_pending_runs
is callable + reachable
- caplog assertion: unknown run_id error names the missing id
Coverage 67.95% -> 68.42% (+0.47 from the new module + tests).
Total tests: 1013 -> 1028.
Closes: closure-plan-202604300906.md §9.5 Phase 37 (GH-7)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Reviewer's GuideImplements a new Sequence diagram for the forgelm approvals --pending flowsequenceDiagram
actor Operator
participant CLI as forgelm_cli_entrypoint
participant Parser as parse_args
participant Dispatcher as _dispatch_subcommand
participant Facade as _cli_facade
participant Approvals as _run_approvals_cmd
participant ListPending as _run_approvals_list_pending
participant Collector as _collect_pending_runs
participant Iter as _iter_audit_events
participant FS as filesystem
Operator->>CLI: run forgelm approvals --pending --output-dir DIR
CLI->>Parser: parse_args()
Parser-->>CLI: args(command=approvals, pending=True)
CLI->>Dispatcher: _dispatch_subcommand("approvals", args)
Dispatcher->>Facade: _run_approvals_cmd(args, output_format)
Facade->>Approvals: _run_approvals_cmd(args, output_format)
Approvals->>ListPending: _run_approvals_list_pending(args, output_format)
ListPending->>FS: check DIR/audit_log.jsonl exists
alt audit_log.jsonl missing
ListPending-->>Operator: friendly empty summary (text or JSON)
else audit_log.jsonl present
ListPending->>Collector: _collect_pending_runs(audit_log_path)
Collector->>Iter: _iter_audit_events(audit_log_path)
Iter->>FS: read JSONL lines
FS-->>Iter: line contents
Iter-->>Collector: (line_number, event_dict)
Collector-->>ListPending: pending_events
ListPending-->>Operator: pending approvals table or JSON envelope
end
Sequence diagram for the forgelm approvals --show RUN_ID flowsequenceDiagram
actor Operator
participant CLI as forgelm_cli_entrypoint
participant Parser as parse_args
participant Dispatcher as _dispatch_subcommand
participant Facade as _cli_facade
participant Approvals as _run_approvals_cmd
participant ShowRun as _run_approvals_show
participant ChainCollector as _collect_run_audit_chain
participant Iter as _iter_audit_events
participant Classifier as _classify_chain
participant StagingDir as _staging_dir_for_event
participant StagingList as _staging_contents
participant FS as filesystem
Operator->>CLI: run forgelm approvals --show RUN_ID --output-dir DIR
CLI->>Parser: parse_args()
Parser-->>CLI: args(command=approvals, show=RUN_ID)
CLI->>Dispatcher: _dispatch_subcommand("approvals", args)
Dispatcher->>Facade: _run_approvals_cmd(args, output_format)
Facade->>Approvals: _run_approvals_cmd(args, output_format)
Approvals->>ShowRun: _run_approvals_show(args, output_format)
ShowRun->>FS: check DIR/audit_log.jsonl exists
alt audit_log.jsonl missing
ShowRun-->>Operator: structured error (text or JSON), exit EXIT_CONFIG_ERROR
else audit_log.jsonl present
ShowRun->>ChainCollector: _collect_run_audit_chain(audit_log_path, RUN_ID)
ChainCollector->>Iter: _iter_audit_events(audit_log_path)
Iter->>FS: read JSONL lines
FS-->>Iter: line contents
Iter-->>ChainCollector: (line_number, event_dict)
ChainCollector-->>ShowRun: chain
alt chain empty
ShowRun-->>Operator: structured error for unknown run_id, exit EXIT_CONFIG_ERROR
else chain nonempty
ShowRun->>Classifier: _classify_chain(chain)
Classifier-->>ShowRun: status (pending/granted/rejected)
ShowRun->>StagingDir: _staging_dir_for_event(DIR, first_required_event)
StagingDir->>FS: resolve staging path
FS-->>StagingDir: path or None
StagingDir-->>ShowRun: staging_path
ShowRun->>StagingList: _staging_contents(staging_path)
StagingList->>FS: list directory entries
FS-->>StagingList: filenames
StagingList-->>ShowRun: staging_listing
ShowRun-->>Operator: human-readable timeline + staging contents or JSON object
end
end
Flow diagram for pending approvals audit_log.jsonl processingflowchart TD
start([Start _collect_pending_runs])
read_events[_iter_audit_events reads audit_log.jsonl]
classify{event.event}
required[Store first human_approval.required per run_id in required_by_run]
decision[Add run_id to decided_run_ids]
other[Ignore non approval-gate events]
pending_build[Build pending list where run_id not in decided_run_ids]
sort[Sort pending by timestamp descending]
return_pending([Return pending events])
start --> read_events
read_events --> classify
classify -->|human_approval.required| required
classify -->|human_approval.granted or human_approval.rejected| decision
classify -->|other| other
required --> read_events
decision --> read_events
other --> read_events
read_events --> pending_build
pending_build --> sort
sort --> return_pending
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Contributor
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Review Summary by QodoAdd
WalkthroughsDescription• Add forgelm approvals subcommand for discovering pending approval requests - --pending mode lists all runs awaiting human review with tabular/JSON output - --show RUN_ID mode displays full audit chain and staging directory contents • Implement audit log parsing with malformed-line resilience (reuses _approve.py pattern) • Support both text and JSON output formats for CI integration • Add 15 comprehensive tests covering empty logs, multiple runs, rejected exclusion, and error cases • Update operator documentation in English and Turkish with actual shipped output examples Diagramflowchart LR
Parser["Parser registrar<br/>_add_approvals_subcommand"]
Dispatch["Dispatcher<br/>_dispatch_subcommand"]
Approvals["Approvals module<br/>_approvals.py"]
AuditLog["Audit log<br/>audit_log.jsonl"]
Output["Output<br/>text/JSON"]
Parser -- "register --pending<br/>--show RUN_ID" --> Dispatch
Dispatch -- "route approvals<br/>command" --> Approvals
Approvals -- "parse events<br/>collect runs" --> AuditLog
Approvals -- "emit pending list<br/>or show details" --> Output
File Changes1. forgelm/cli/subcommands/_approvals.py
|
Code Review by Qodo
1.
|
Not up to standards ⛔🔴 Issues
|
| Category | Results |
|---|---|
| ErrorProne | 1 high |
🟢 Metrics 111 complexity · 11 duplication
Metric Results Complexity 111 Duplication 11
NEW Get contextual insights on your PRs based on Codacy's metrics, along with PR and Jira context, without leaving GitHub. Enable AI reviewer
TIP This summary will be updated as you push new changes.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the forgelm approvals subcommand, which allows operators to discover pending approval requests and inspect specific runs using the --pending and --show flags. The implementation includes a new CLI module, updated documentation in English and Turkish, and a comprehensive test suite. Review feedback identifies a logic error in how pending runs are tracked if run IDs are reused, a potential path traversal vulnerability when processing staging paths from the audit log, and opportunities to reduce code duplication between the new module and the existing _approve.py logic.
Comment on lines
+111
to
+130
| required_by_run: Dict[str, Dict[str, Any]] = {} | ||
| decided_run_ids: set[str] = set() | ||
|
|
||
| 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: | ||
| # Keep the first (oldest) requirement event for the run. A | ||
| # second `required` for the same run_id would be unusual but | ||
| # not malformed — we keep the first because that is the | ||
| # original request the operator is reviewing. | ||
| required_by_run.setdefault(run_id, event) | ||
| elif event_name in _TERMINAL_DECISION_EVENTS: | ||
| decided_run_ids.add(run_id) | ||
|
|
||
| pending = [event for run_id, event in required_by_run.items() if run_id not in decided_run_ids] |
There was a problem hiding this comment.
The current logic for identifying pending runs is flawed if a run_id is reused after a terminal decision (e.g., if a training run is restarted with the same ID after a rejection). Because it uses a static set of decided_run_ids collected from the entire log, any previous decision for a run_id will permanently exclude it from the pending list, even if a new approval request has been issued later in the log.
A more robust approach is to use a state-machine pattern while iterating through the log: a human_approval.required event adds the run to the pending set, and a terminal decision removes it.
References
- Ensure logic handles state transitions correctly over time in append-only logs.
Comment on lines
+163
to
+165
| declared = event.get("staging_path") | ||
| if isinstance(declared, str) and declared: | ||
| return declared |
There was a problem hiding this comment.
The staging_path retrieved from the audit log is used directly without validation. This presents a path traversal risk: a tampered or malicious audit log could specify an absolute path (e.g., /etc) or a path using .. to escape the output_dir.
Since this tool is designed for compliance and forensic review, it should maintain the defense-in-depth posture established in _approve.py by validating that the staging path remains within the expected output directory boundary.
Comment on lines
+17
to
+21
| The dispatcher reuses the helpers shipped with Phase 9 in | ||
| :mod:`forgelm.cli.subcommands._approve` (`_find_human_approval_required_event`, | ||
| `_find_human_approval_decision_event`) so the JSONL parser, the malformed-line | ||
| accounting, and the ``output_dir/audit_log.jsonl`` convention live in exactly | ||
| one place. |
There was a problem hiding this comment.
The module docstring claims to reuse helpers from _approve.py, but the implementation actually duplicates much of the logic (e.g., audit log parsing, event name constants, and error handling). This increases the maintenance burden and the risk of the two subcommands drifting apart in behavior.
Consider importing shared constants and helpers from _approve.py or moving them to a common utility module within the cli package to ensure consistency across the approval family of commands.
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 6 issues, and left some high level feedback:
- _staging_dir_for_event
's implementation doesn't match its docstring: for older logs withoutstaging_pathyou currently returnNonewhen<output_dir>/final_model.stagingis missing, whereas the docstring says you always synthesise the canonical path and let the caller decide how to handle a non-existent directory; consider always returning the fallback path and lettingstaging_existsbe derived solely fromos.path.isdir`. - The approval-related constants/helpers (event names, terminal set, audit-log filename, error-envelope helper, JSONL parsing) are redefined here instead of being shared with
_approve; extracting these into a small shared module would prevent them drifting if the audit schema or filenames change in the future.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- _staging_dir_for_event`'s implementation doesn't match its docstring: for older logs without `staging_path` you currently return `None` when `<output_dir>/final_model.staging` is missing, whereas the docstring says you always synthesise the canonical path and let the caller decide how to handle a non-existent directory; consider always returning the fallback path and letting `staging_exists` be derived solely from `os.path.isdir`.
- The approval-related constants/helpers (event names, terminal set, audit-log filename, error-envelope helper, JSONL parsing) are redefined here instead of being shared with `_approve`; extracting these into a small shared module would prevent them drifting if the audit schema or filenames change in the future.
## Individual Comments
### Comment 1
<location path="tests/test_approvals_listing.py" line_range="135-144" />
<code_context>
+ def test_rejected_run_also_excluded_from_pending(self, tmp_path: Path, capsys) -> None:
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test for `--show` on a rejected run to exercise the `rejected` status classification and chain rendering.
You already test that rejected runs are excluded from `--pending`, but there’s no test that `forgelm approvals --show` reports `status == "rejected"` and includes the final `human_approval.rejected` event in the chain. Please add a JSON-path test like `test_show_granted_run_status_granted`, but for `decision="rejected"`, to cover `_classify_chain`’s rejected branch and verify the rejected-run UX.
Suggested implementation:
```python
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["success"] is True
# Only the still-pending run should be reported as pending
assert payload["count"] == 1
run_ids = {summary["run_id"] for summary in payload["pending"]}
assert run_ids == {"fg-pending00000"}
def test_show_rejected_run_status_rejected(self, tmp_path: Path, capsys) -> None:
# `forgelm approvals --show` for a rejected run should:
# * classify the run status as "rejected"
# * include the final human_approval.rejected event in the chain
output_dir = tmp_path / "run"
run_id = "fg-rejected0000"
_seed_run(output_dir, run_id, decision="rejected")
from forgelm.cli import _run_approvals_cmd
with pytest.raises(SystemExit) as ei:
_run_approvals_cmd(
_build_args(output_dir, show=True, run_ids=[run_id]),
output_format="json",
)
assert ei.value.code == 0
payload = json.loads(capsys.readouterr().out)
# Basic success + status classification
assert payload["success"] is True
# Mirror the JSON-path used in `test_show_granted_run_status_granted`
# (typically something like payload["run"]["status"] or payload["status"])
# Here we assume the top-level "status" key is used, as in the granted test.
assert payload["status"] == "rejected"
# Verify the rendered chain includes the final human_approval.rejected event.
# This mirrors the chain JSON-path used in the granted-status test.
chain = payload["chain"]
assert isinstance(chain, list)
assert chain, "expected at least one event in the rendered chain"
last_event = chain[-1]
# Assuming the event type is exposed as "type" and matches "human_approval.rejected"
assert last_event["type"] == "human_approval.rejected"
```
To align this with your existing tests and JSON schema, you should:
1. Adjust the JSON paths to match what `test_show_granted_run_status_granted` already asserts:
- If that test uses `payload["run"]["status"]` (or `payload["show"]["status"]`, etc.), update `assert payload["status"] == "rejected"` to use the same path.
- If the chain is nested (e.g. `payload["run"]["chain"]` or `payload["events"]`), update `chain = payload["chain"]` accordingly.
2. Align the event-type assertion with your existing event schema:
- If events use a different key (e.g. `"event_type"` or `"name"`), change `last_event["type"]` to that key.
- If the value for a rejected approval is different (e.g. `"human_approval.rejected.v1"`), update the expected string to match.
3. If `test_rejected_run_also_excluded_from_pending` already had additional assertions after the snippet you shared, re-apply or merge them into the updated function body as needed.
</issue_to_address>
### Comment 2
<location path="tests/test_approvals_listing.py" line_range="152-161" />
<code_context>
+ 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:
</code_context>
<issue_to_address>
**suggestion (testing):** Consider a test that covers the fallback `staging_exists=False` / `"MISSING"` path when the staging directory is absent.
Currently the tests only cover the case where the staging directory exists and `staging_path` is present. Since `_summarise_pending` and `_emit_pending_text` also handle missing staging (setting `staging_exists` to `False` and rendering `"MISSING"`), please add a test that seeds a `human_approval.required` event without `final_model.staging` (or with `create_staging=False`) and asserts that `staging_exists is False` and `"MISSING"` appears in the text output.
Suggested implementation:
```python
def test_pending_summary_missing_staging_sets_flag_and_text_fallback(
self, tmp_path: Path, capsys
) -> None:
output_dir = tmp_path / "run"
_seed_run(output_dir, "fg-missing0000000", create_staging=False)
from forgelm.cli import _run_approvals_cmd
# JSON output: staging_exists should be False when staging is missing.
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_exists"] is False
# Text output: the table should render the "MISSING" fallback.
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_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
```
This test assumes that `_seed_run` accepts a `create_staging` keyword argument which, when set to `False`, will omit the `final_model.staging` directory and associated metadata while still emitting a `human_approval.required` event. If the helper uses a different parameter name or mechanism to skip staging, update the `_seed_run` call accordingly (e.g. `with_staging=False` or omitting a `staging_dir` argument).
</issue_to_address>
### Comment 3
<location path="tests/test_approvals_listing.py" line_range="104-113" />
<code_context>
+
+
+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:
</code_context>
<issue_to_address>
**suggestion (testing):** Add a JSON-output variant for the missing `audit_log.jsonl` case to verify the `pending` envelope shape.
This covers the text output when `audit_log.jsonl` is missing, but `_run_approvals_list_pending` has distinct logic for `output_format="json"` (empty `pending` array with `count: 0`). Please add a similar test using `_run_approvals_cmd` with `output_format="json"` in the same setup and assert the JSON payload (e.g., `{"success": True, "pending": [], "count": 0}`) to lock in that contract.
Suggested implementation:
```python
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_empty_audit_log_returns_empty_json_envelope(self, tmp_path: Path, capsys) -> None:
output_dir = tmp_path / "run"
output_dir.mkdir()
# No audit_log.jsonl on disk → JSON envelope with empty pending list, count 0, exit 0.
from forgelm.cli import _run_approvals_cmd
import json
with pytest.raises(SystemExit) as ei:
_run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json")
assert ei.value.code == 0
captured = capsys.readouterr().out
payload = json.loads(captured)
assert payload.get("success") is True
assert payload.get("pending") == []
assert payload.get("count") == 0
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")
```
If `json` is already imported at the top of the file, you can remove the inline `import json` inside the new test and rely on the module-level import instead to avoid duplication.
</issue_to_address>
### Comment 4
<location path="tests/test_approvals_listing.py" line_range="290-299" />
<code_context>
+class TestApprovalsMalformedLines:
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding a malformed-lines test for the `--show` path to exercise `_collect_run_audit_chain`'s robustness.
You already cover malformed JSONL for `--pending` via `_iter_audit_events`, but there’s no equivalent coverage for `--show`. Please add a test where a run has one valid `human_approval.required` event, followed by a malformed line and a non-dict JSON value, then invoke `--show` (JSON output) for that run and assert the `chain` and `status` remain correct. That will exercise this robustness on the `--show` path as well.
Suggested implementation:
```python
# ---------------------------------------------------------------------------
# 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.
#
# This exercises robustness of the `--pending` path (via
# `_iter_audit_events`), ensuring that bad lines do not hide a
# legitimately pending run.
run_id = "run_pending_malformed"
output_dir = tmp_path / run_id
output_dir.mkdir()
audit_path = output_dir / "audit_log.jsonl"
# 1) Valid human_approval.required event
_write_event(
audit_path,
"human_approval.required",
run_id=run_id,
data={"prompt": "Please approve this run"},
)
# 2) Malformed JSON line that should be skipped
audit_path.write_text(
audit_path.read_text()
+ "{ this is not valid json\n",
encoding="utf-8",
)
# 3) A syntactically valid JSON value with a non-dict root that
# should also be ignored by the iterator.
audit_path.write_text(
audit_path.read_text()
+ json.dumps(["not-a-dict-root"]) # type: ignore[list-item]
+ "\n",
encoding="utf-8",
)
# Run the approvals listing in "pending" mode.
#
# NOTE: The exact helper/CLI invocation may differ in your codebase
# (e.g. click runner, subprocess, or a direct function call). See
# <additional_changes> for what may need to be adjusted.
result = _run_approvals_cli(
[
"approvals",
"--pending",
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
pending = json.loads(stdout)
run_ids = {item["run_id"] for item in pending}
assert run_id in run_ids
def test_malformed_lines_do_not_break_show_chain_json(self, tmp_path: Path, capsys) -> None:
# Similar corruption scenario as above, but exercise the `--show`
# path (which uses `_collect_run_audit_chain`) instead of
# `--pending`. We ensure that:
#
# * The valid `human_approval.required` event remains in the chain
# * Malformed and non-dict JSONL lines do not break the command
# * The reported status for the run remains correct
run_id = "run_show_malformed"
output_dir = tmp_path / run_id
output_dir.mkdir()
audit_path = output_dir / "audit_log.jsonl"
# 1) Valid required event that should be present in the chain
_write_event(
audit_path,
"human_approval.required",
run_id=run_id,
data={"prompt": "Approve me via --show"},
)
# 2) Malformed JSON line
audit_path.write_text(
audit_path.read_text()
+ "{ malformed json line\n",
encoding="utf-8",
)
# 3) Valid JSON but with non-dict root (e.g. a simple string)
audit_path.write_text(
audit_path.read_text()
+ json.dumps("not-a-dict-root")
+ "\n",
encoding="utf-8",
)
# Invoke the approvals listing in "show" mode with JSON output for
# this specific run. The CLI is expected to use
# `_collect_run_audit_chain` internally and gracefully skip the bad
# lines while still providing a correct chain and status.
result = _run_approvals_cli(
[
"approvals",
"--show",
run_id,
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
show_output = json.loads(stdout)
# The exact schema may differ, but typically we expect something like:
# {
# "run_id": "...",
# "status": "pending" | "...",
# "chain": [ ... events ... ],
# ...
# }
#
# We assert that:
# * status is still a non-terminal status (e.g. "pending") since
# only `human_approval.required` exists in the log.
# * the chain contains exactly one event corresponding to the
# required-approval event we wrote.
assert show_output["run_id"] == run_id
# Adjust the expected status string here if your implementation uses
# a different vocabulary (for example "waiting_for_approval").
assert show_output["status"] in ("pending", "waiting_for_approval")
chain = show_output["chain"]
assert isinstance(chain, list)
assert len(chain) == 1
event = chain[0]
assert event["event_type"] == "human_approval.required"
assert event["run_id"] == run_id
assert event["data"]["prompt"] == "Approve me via --show"
```
for what may need to be adjusted.
result = _run_approvals_cli(
[
"approvals",
"--pending",
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
pending = json.loads(stdout)
run_ids = {item["run_id"] for item in pending}
assert run_id in run_ids
def test_malformed_lines_do_not_break_show_chain_json(self, tmp_path: Path, capsys) -> None:
# Similar corruption scenario as above, but exercise the `--show`
# path (which uses `_collect_run_audit_chain`) instead of
# `--pending`. We ensure that:
#
# * The valid `human_approval.required` event remains in the chain
# * Malformed and non-dict JSONL lines do not break the command
# * The reported status for the run remains correct
run_id = "run_show_malformed"
output_dir = tmp_path / run_id
output_dir.mkdir()
audit_path = output_dir / "audit_log.jsonl"
# 1) Valid required event that should be present in the chain
_write_event(
audit_path,
"human_approval.required",
run_id=run_id,
data={"prompt": "Approve me via --show"},
)
# 2) Malformed JSON line
audit_path.write_text(
audit_path.read_text()
+ "{ malformed json line\n",
encoding="utf-8",
)
# 3) Valid JSON but with non-dict root (e.g. a simple string)
audit_path.write_text(
audit_path.read_text()
+ json.dumps("not-a-dict-root")
+ "\n",
encoding="utf-8",
)
# Invoke the approvals listing in "show" mode with JSON output for
# this specific run. The CLI is expected to use
# `_collect_run_audit_chain` internally and gracefully skip the bad
# lines while still providing a correct chain and status.
result = _run_approvals_cli(
[
"approvals",
"--show",
run_id,
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
show_output = json.loads(stdout)
# The exact schema may differ, but typically we expect something like:
# {
# "run_id": "...",
# "status": "pending" | "...",
# "chain": [ ... events ... ],
# ...
# }
#
# We assert that:
# * status is still a non-terminal status (e.g. "pending") since
# only `human_approval.required` exists in the log.
# * the chain contains exactly one event corresponding to the
# required-approval event we wrote.
assert show_output["run_id"] == run_id
# Adjust the expected status string here if your implementation uses
# a different vocabulary (for example "waiting_for_approval").
assert show_output["status"] in ("pending", "waiting_for_approval")
chain = show_output["chain"]
assert isinstance(chain, list)
assert len(chain) == 1
event = chain[0]
assert event["event_type"] == "human_approval.required"
assert event["run_id"] == run_id
assert event["data"]["prompt"] == "Approve me via --show"
>>>>>>> REPLACE
</file_operation>
</file_operations>
<additional_changes>
1. The above tests assume:
- A helper `_run_approvals_cli(argv: list[str])` exists and returns an object with a `stdout` attribute, or that `capsys` can capture its printed output. If your file currently invokes the CLI differently (e.g. using a pytest `cli_runner`, `subprocess.run`, or a direct function), adapt the calls to match that pattern.
- The JSON schema for `--pending --format json` is a list of entries containing a `"run_id"` field.
- The JSON schema for `--show --format json` is a single-object response with `"run_id"`, `"status"`, and `"chain"` keys, where `"chain"` is a list of events that each include `"event_type"`, `"run_id"`, and `"data"`.
2. If your CLI uses a different option name than `"--format json"` (e.g. `"--json"` or `"--output-format json"`), update the argument list accordingly in both tests.
3. If the status terminology for a run that is awaiting human approval is not `"pending"` or `"waiting_for_approval"`, update the `assert show_output["status"] in (...)` line to match the actual status value used in your implementation.
4. Ensure `json` and any helper imports used here (`_run_approvals_cli`, `_write_event`) are present at the top of `tests/test_approvals_listing.py`. If `_write_event` is defined in a different module, import it from there instead of relying on it being in the same file.
</issue_to_address>
### Comment 5
<location path="tests/test_approvals_listing.py" line_range="358-367" />
<code_context>
+
+
+@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
</code_context>
<issue_to_address>
**suggestion (testing):** You might also want a JSON-output error test for `--show` on an unknown run to lock in the structured error envelope.
Since `_output_error_and_exit` also formats errors for JSON as `{ "success": False, "error": ... }`, please add a similar test that calls `_run_approvals_cmd` with `output_format="json"` and a non-existent run ID, then asserts `success is False` and that the `error` message includes the run ID. This will lock in the behavior relied on by CI/automation for JSON mode.
</issue_to_address>
### Comment 6
<location path="docs/usermanuals/tr/compliance/human-oversight.md" line_range="159" />
<code_context>
+ - 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 kodu ile çıkar.
+
## Sık hatalar
</code_context>
<issue_to_address>
**nitpick (typo):** Küçük Türkçe dilbilgisi düzeltmesi: "1 kodu ile" yerine "1 koduyla" daha doğal.
Anlaşılır olsa da Türkçe’de bu tip durumlarda “1 koduyla çıkar” daha doğal. Cümleyi buna göre güncelleyebilirsiniz.
```suggestion
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.
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+135
to
+144
| 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: |
There was a problem hiding this comment.
suggestion (testing): Add a test for --show on a rejected run to exercise the rejected status classification and chain rendering.
You already test that rejected runs are excluded from --pending, but there’s no test that forgelm approvals --show reports status == "rejected" and includes the final human_approval.rejected event in the chain. Please add a JSON-path test like test_show_granted_run_status_granted, but for decision="rejected", to cover _classify_chain’s rejected branch and verify the rejected-run UX.
Suggested implementation:
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["success"] is True
# Only the still-pending run should be reported as pending
assert payload["count"] == 1
run_ids = {summary["run_id"] for summary in payload["pending"]}
assert run_ids == {"fg-pending00000"}
def test_show_rejected_run_status_rejected(self, tmp_path: Path, capsys) -> None:
# `forgelm approvals --show` for a rejected run should:
# * classify the run status as "rejected"
# * include the final human_approval.rejected event in the chain
output_dir = tmp_path / "run"
run_id = "fg-rejected0000"
_seed_run(output_dir, run_id, decision="rejected")
from forgelm.cli import _run_approvals_cmd
with pytest.raises(SystemExit) as ei:
_run_approvals_cmd(
_build_args(output_dir, show=True, run_ids=[run_id]),
output_format="json",
)
assert ei.value.code == 0
payload = json.loads(capsys.readouterr().out)
# Basic success + status classification
assert payload["success"] is True
# Mirror the JSON-path used in `test_show_granted_run_status_granted`
# (typically something like payload["run"]["status"] or payload["status"])
# Here we assume the top-level "status" key is used, as in the granted test.
assert payload["status"] == "rejected"
# Verify the rendered chain includes the final human_approval.rejected event.
# This mirrors the chain JSON-path used in the granted-status test.
chain = payload["chain"]
assert isinstance(chain, list)
assert chain, "expected at least one event in the rendered chain"
last_event = chain[-1]
# Assuming the event type is exposed as "type" and matches "human_approval.rejected"
assert last_event["type"] == "human_approval.rejected"To align this with your existing tests and JSON schema, you should:
- Adjust the JSON paths to match what
test_show_granted_run_status_grantedalready asserts:- If that test uses
payload["run"]["status"](orpayload["show"]["status"], etc.), updateassert payload["status"] == "rejected"to use the same path. - If the chain is nested (e.g.
payload["run"]["chain"]orpayload["events"]), updatechain = payload["chain"]accordingly.
- If that test uses
- Align the event-type assertion with your existing event schema:
- If events use a different key (e.g.
"event_type"or"name"), changelast_event["type"]to that key. - If the value for a rejected approval is different (e.g.
"human_approval.rejected.v1"), update the expected string to match.
- If events use a different key (e.g.
- If
test_rejected_run_also_excluded_from_pendingalready had additional assertions after the snippet you shared, re-apply or merge them into the updated function body as needed.
Comment on lines
+152
to
+161
| 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] |
There was a problem hiding this comment.
suggestion (testing): Consider a test that covers the fallback staging_exists=False / "MISSING" path when the staging directory is absent.
Currently the tests only cover the case where the staging directory exists and staging_path is present. Since _summarise_pending and _emit_pending_text also handle missing staging (setting staging_exists to False and rendering "MISSING"), please add a test that seeds a human_approval.required event without final_model.staging (or with create_staging=False) and asserts that staging_exists is False and "MISSING" appears in the text output.
Suggested implementation:
def test_pending_summary_missing_staging_sets_flag_and_text_fallback(
self, tmp_path: Path, capsys
) -> None:
output_dir = tmp_path / "run"
_seed_run(output_dir, "fg-missing0000000", create_staging=False)
from forgelm.cli import _run_approvals_cmd
# JSON output: staging_exists should be False when staging is missing.
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_exists"] is False
# Text output: the table should render the "MISSING" fallback.
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_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 outThis test assumes that _seed_run accepts a create_staging keyword argument which, when set to False, will omit the final_model.staging directory and associated metadata while still emitting a human_approval.required event. If the helper uses a different parameter name or mechanism to skip staging, update the _seed_run call accordingly (e.g. with_staging=False or omitting a staging_dir argument).
Comment on lines
+104
to
+113
| 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 |
There was a problem hiding this comment.
suggestion (testing): Add a JSON-output variant for the missing audit_log.jsonl case to verify the pending envelope shape.
This covers the text output when audit_log.jsonl is missing, but _run_approvals_list_pending has distinct logic for output_format="json" (empty pending array with count: 0). Please add a similar test using _run_approvals_cmd with output_format="json" in the same setup and assert the JSON payload (e.g., {"success": True, "pending": [], "count": 0}) to lock in that contract.
Suggested implementation:
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_empty_audit_log_returns_empty_json_envelope(self, tmp_path: Path, capsys) -> None:
output_dir = tmp_path / "run"
output_dir.mkdir()
# No audit_log.jsonl on disk → JSON envelope with empty pending list, count 0, exit 0.
from forgelm.cli import _run_approvals_cmd
import json
with pytest.raises(SystemExit) as ei:
_run_approvals_cmd(_build_args(output_dir, pending=True), output_format="json")
assert ei.value.code == 0
captured = capsys.readouterr().out
payload = json.loads(captured)
assert payload.get("success") is True
assert payload.get("pending") == []
assert payload.get("count") == 0
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")If json is already imported at the top of the file, you can remove the inline import json inside the new test and rely on the module-level import instead to avoid duplication.
Comment on lines
+290
to
+299
| 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, |
There was a problem hiding this comment.
suggestion (testing): Consider adding a malformed-lines test for the --show path to exercise _collect_run_audit_chain's robustness.
You already cover malformed JSONL for --pending via _iter_audit_events, but there’s no equivalent coverage for --show. Please add a test where a run has one valid human_approval.required event, followed by a malformed line and a non-dict JSON value, then invoke --show (JSON output) for that run and assert the chain and status remain correct. That will exercise this robustness on the --show path as well.
Suggested implementation:
# ---------------------------------------------------------------------------
# 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.
#
# This exercises robustness of the `--pending` path (via
# `_iter_audit_events`), ensuring that bad lines do not hide a
# legitimately pending run.
run_id = "run_pending_malformed"
output_dir = tmp_path / run_id
output_dir.mkdir()
audit_path = output_dir / "audit_log.jsonl"
# 1) Valid human_approval.required event
_write_event(
audit_path,
"human_approval.required",
run_id=run_id,
data={"prompt": "Please approve this run"},
)
# 2) Malformed JSON line that should be skipped
audit_path.write_text(
audit_path.read_text()
+ "{ this is not valid json\n",
encoding="utf-8",
)
# 3) A syntactically valid JSON value with a non-dict root that
# should also be ignored by the iterator.
audit_path.write_text(
audit_path.read_text()
+ json.dumps(["not-a-dict-root"]) # type: ignore[list-item]
+ "\n",
encoding="utf-8",
)
# Run the approvals listing in "pending" mode.
#
# NOTE: The exact helper/CLI invocation may differ in your codebase
# (e.g. click runner, subprocess, or a direct function call). See
# <additional_changes> for what may need to be adjusted.
result = _run_approvals_cli(
[
"approvals",
"--pending",
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
pending = json.loads(stdout)
run_ids = {item["run_id"] for item in pending}
assert run_id in run_ids
def test_malformed_lines_do_not_break_show_chain_json(self, tmp_path: Path, capsys) -> None:
# Similar corruption scenario as above, but exercise the `--show`
# path (which uses `_collect_run_audit_chain`) instead of
# `--pending`. We ensure that:
#
# * The valid `human_approval.required` event remains in the chain
# * Malformed and non-dict JSONL lines do not break the command
# * The reported status for the run remains correct
run_id = "run_show_malformed"
output_dir = tmp_path / run_id
output_dir.mkdir()
audit_path = output_dir / "audit_log.jsonl"
# 1) Valid required event that should be present in the chain
_write_event(
audit_path,
"human_approval.required",
run_id=run_id,
data={"prompt": "Approve me via --show"},
)
# 2) Malformed JSON line
audit_path.write_text(
audit_path.read_text()
+ "{ malformed json line\n",
encoding="utf-8",
)
# 3) Valid JSON but with non-dict root (e.g. a simple string)
audit_path.write_text(
audit_path.read_text()
+ json.dumps("not-a-dict-root")
+ "\n",
encoding="utf-8",
)
# Invoke the approvals listing in "show" mode with JSON output for
# this specific run. The CLI is expected to use
# `_collect_run_audit_chain` internally and gracefully skip the bad
# lines while still providing a correct chain and status.
result = _run_approvals_cli(
[
"approvals",
"--show",
run_id,
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
show_output = json.loads(stdout)
# The exact schema may differ, but typically we expect something like:
# {
# "run_id": "...",
# "status": "pending" | "...",
# "chain": [ ... events ... ],
# ...
# }
#
# We assert that:
# * status is still a non-terminal status (e.g. "pending") since
# only `human_approval.required` exists in the log.
# * the chain contains exactly one event corresponding to the
# required-approval event we wrote.
assert show_output["run_id"] == run_id
# Adjust the expected status string here if your implementation uses
# a different vocabulary (for example "waiting_for_approval").
assert show_output["status"] in ("pending", "waiting_for_approval")
chain = show_output["chain"]
assert isinstance(chain, list)
assert len(chain) == 1
event = chain[0]
assert event["event_type"] == "human_approval.required"
assert event["run_id"] == run_id
assert event["data"]["prompt"] == "Approve me via --show"for what may need to be adjusted.
result = _run_approvals_cli(
[
"approvals",
"--pending",
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
pending = json.loads(stdout)
run_ids = {item["run_id"] for item in pending}
assert run_id in run_ids
def test_malformed_lines_do_not_break_show_chain_json(self, tmp_path: Path, capsys) -> None:
# Similar corruption scenario as above, but exercise the `--show`
# path (which uses `_collect_run_audit_chain`) instead of
# `--pending`. We ensure that:
#
# * The valid `human_approval.required` event remains in the chain
# * Malformed and non-dict JSONL lines do not break the command
# * The reported status for the run remains correct
run_id = "run_show_malformed"
output_dir = tmp_path / run_id
output_dir.mkdir()
audit_path = output_dir / "audit_log.jsonl"
# 1) Valid required event that should be present in the chain
_write_event(
audit_path,
"human_approval.required",
run_id=run_id,
data={"prompt": "Approve me via --show"},
)
# 2) Malformed JSON line
audit_path.write_text(
audit_path.read_text()
+ "{ malformed json line\n",
encoding="utf-8",
)
# 3) Valid JSON but with non-dict root (e.g. a simple string)
audit_path.write_text(
audit_path.read_text()
+ json.dumps("not-a-dict-root")
+ "\n",
encoding="utf-8",
)
# Invoke the approvals listing in "show" mode with JSON output for
# this specific run. The CLI is expected to use
# `_collect_run_audit_chain` internally and gracefully skip the bad
# lines while still providing a correct chain and status.
result = _run_approvals_cli(
[
"approvals",
"--show",
run_id,
"--output-dir",
str(tmp_path),
"--format",
"json",
]
)
captured = capsys.readouterr()
stdout = captured.out or result.stdout
show_output = json.loads(stdout)
# The exact schema may differ, but typically we expect something like:
# {
# "run_id": "...",
# "status": "pending" | "...",
# "chain": [ ... events ... ],
# ...
# }
#
# We assert that:
# * status is still a non-terminal status (e.g. "pending") since
# only `human_approval.required` exists in the log.
# * the chain contains exactly one event corresponding to the
# required-approval event we wrote.
assert show_output["run_id"] == run_id
# Adjust the expected status string here if your implementation uses
# a different vocabulary (for example "waiting_for_approval").
assert show_output["status"] in ("pending", "waiting_for_approval")
chain = show_output["chain"]
assert isinstance(chain, list)
assert len(chain) == 1
event = chain[0]
assert event["event_type"] == "human_approval.required"
assert event["run_id"] == run_id
assert event["data"]["prompt"] == "Approve me via --show"
REPLACE
</file_operation>
</file_operations>
<additional_changes>
-
The above tests assume:
- A helper
_run_approvals_cli(argv: list[str])exists and returns an object with astdoutattribute, or thatcapsyscan capture its printed output. If your file currently invokes the CLI differently (e.g. using a pytestcli_runner,subprocess.run, or a direct function), adapt the calls to match that pattern. - The JSON schema for
--pending --format jsonis a list of entries containing a"run_id"field. - The JSON schema for
--show --format jsonis a single-object response with"run_id","status", and"chain"keys, where"chain"is a list of events that each include"event_type","run_id", and"data".
- A helper
-
If your CLI uses a different option name than
"--format json"(e.g."--json"or"--output-format json"), update the argument list accordingly in both tests. -
If the status terminology for a run that is awaiting human approval is not
"pending"or"waiting_for_approval", update theassert show_output["status"] in (...)line to match the actual status value used in your implementation. -
Ensure
jsonand any helper imports used here (_run_approvals_cli,_write_event) are present at the top oftests/test_approvals_listing.py. If_write_eventis defined in a different module, import it from there instead of relying on it being in the same file.
Comment on lines
+358
to
+367
| 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"): |
There was a problem hiding this comment.
suggestion (testing): You might also want a JSON-output error test for --show on an unknown run to lock in the structured error envelope.
Since _output_error_and_exit also formats errors for JSON as { "success": False, "error": ... }, please add a similar test that calls _run_approvals_cmd with output_format="json" and a non-existent run ID, then asserts success is False and that the error message includes the run ID. This will lock in the behavior relied on by CI/automation for JSON mode.
| - 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 kodu ile çıkar. |
There was a problem hiding this comment.
nitpick (typo): Küçük Türkçe dilbilgisi düzeltmesi: "1 kodu ile" yerine "1 koduyla" daha doğal.
Anlaşılır olsa da Türkçe’de bu tip durumlarda “1 koduyla çıkar” daha doğal. Cümleyi buna göre güncelleyebilirsiniz.
Suggested change
| 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 kodu ile çıkar. | |
| 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. |
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Closes the two HIGH findings + four bot suggestions. HIGH fixes: - F-25-01 (HIGH security): _staging_dir_for_event now reuses _staging_path_inside_output_dir from _approve.py to refuse staging_path values that escape output_dir. A tampered audit log pointing staging_path at /etc no longer leaks a directory listing through _staging_contents. Two new regression tests (TestApprovalsPathTraversalGuard, --show + --pending modes). - F-25-02 (HIGH refactor): consolidated the audit-log JSONL parser into a new shared module forgelm/cli/subcommands/_audit_log_reader.py. _approve.py's _find_human_approval_required_event + _find_human_approval_decision_event and _approvals.py's _iter_audit_events all now delegate to iter_audit_events / find_latest_event_for_run there. Future malformed-line policy fixes land in one place. XPR-01 cross-PR observation closed (Phase 21 forgelm purge will use the same module). MEDIUM fixes: - F-25-03 (MEDIUM correctness): latest-wins semantics for re-staged runs. _collect_pending_runs walked the log with setdefault (first required wins) — inconsistent with _approve.py's latest-wins decision-event pattern. Now both use latest-wins (track the latest required line + latest terminal-decision line per run_id; a run is pending iff its latest required > latest decision). New regression: TestApprovalsLatestWinsRunIdReuse exercises the re-staged-after-rejection scenario. Bot-suggested coverage (4 tests + TR typo): - test_show_rejected_run_status_rejected: covers the rejected --show path classifier branch. - test_pending_table_renders_missing_when_staging_absent: covers the MISSING staging cell rendering when create_staging=False. - test_pending_empty_audit_log_json_envelope: pins the JSON envelope shape for the no-audit-log-yet case. - test_show_unknown_run_id_json_envelope: pins the structured error envelope from _output_error_and_exit on unknown run_id under JSON output. - test_show_robust_to_malformed_lines: --show inherits the same malformed-line tolerance as --pending (auto via shared _audit_log_reader). - TR typo: "1 kodu ile" → "1 koduyla" (sourcery nit). DEFERRED (out of this PR's scope): - F-25-04 (MEDIUM doc): JSON envelope schema page — docs/reference/cli_json_schemas.md is XPR-03 cross-PR concern; same page should also document Phase 34 doctor envelope. Better as one consolidated docs PR after this batch. - F-25-05 (MEDIUM test): hardcoded timestamp parametrisation — noise-to-signal too low for this PR's blast radius; absorbed into Wave 2b factory growth (XPR-03 recommendation). - F-25-06 (MEDIUM contract): defensive "neither flag set" branch — kept as belt-and-suspenders; argparse exit code 2 vs our 1 divergence is intentional (operator misuse vs internal contract violation). - LOW / NIT findings (07-14): not regressions; kept on the backlog. Tests: 1028 → 1036 (+8). ruff format/check clean. Coverage 68.42% → 68.68% (+0.26). Closes (review IDs): F-25-01, F-25-02, F-25-03 + 4 bot tests + TR typo Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
|
This was referenced May 2, 2026
Collaborator
Author
Wave 2a Round-1 multi-agent review — fixes pushedAll CRITICAL + HIGH findings from the multi-agent review (and the relevant inline bot comments) have been addressed in the latest commit on this branch. See the commit message for the per-finding mapping (review IDs F-25-*, gemini bot IDs, qodo bot IDs, sourcery bot IDs). Per-PR test counts:
ruff format/check clean across all five PRs. Cross-PR XPR-01 concern (audit-log parser proliferation) closed in PR #25 via the new Items deferred to follow-up PRs (with rationale) are listed under DEFERRED in each commit message. @coderabbitai review |
This was referenced May 2, 2026
Contributor
|
Tip For best results, initiate chat on the files or code changes.
I'll kick off a fresh review now to verify the fixes against the previous findings. [review] |
8 tasks
Collaborator
Author
|
Superseded by #28 (consolidated Wave 2a integration). All commits from this branch are merged into |
cemililik
added a commit
that referenced
this pull request
Jun 15, 2026
… secret (PR#63-review) #24: export_compliance_artifacts rollback swallowed os.remove/os.replace failures with except OSError: pass; collect and logger.error them before re-raising the original promotion error. #25: strict-HMAC guard rejected only None, letting an empty-string secret through; gate on 'not hmac_secret'. CodeRabbit #24,#25. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Closes the Phase 9 discovery gap (GH-7 from `ghost-features-analysis-20260502`). `forgelm approve` / `reject` shipped in Wave 1 but an operator coming to the workstation cold had no way to ask "which runs are awaiting my review?".
`forgelm approvals` fills the gap with two query modes:
```shell
$ 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
$ 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):
```
`--output-format json` returns a structured envelope so CI can filter the queue programmatically.
Files
Test plan
Risk
Low. Pure additive feature: no changes to `approve` / `reject` behaviour, no schema migrations, no new heavy dependencies. The audit-log read path is shared with the existing `_approve` module so any regression there would surface in both subcommands at once.
Closes
🤖 Generated with Claude Code
Summary by Sourcery
Add a new
forgelm approvalsCLI subcommand to list and inspect pending human approval runs, wire it into the CLI dispatcher, and document the feature and its behavior.New Features:
forgelm approvals --pending --output-dir DIRto list runs awaiting human approval, with tabular text or JSON envelope output for CI consumption.forgelm approvals --show RUN_ID --output-dir DIRto display the full approval audit chain and staging directory contents for a single run.Enhancements:
forgelm.clifacade for easier monkeypatching and testing.Documentation:
forgelm approvalssubcommand, its modes, JSON output, and error behavior.Tests: