From 4d042dca15e6aedd19ba8acf7329f283fd69be02 Mon Sep 17 00:00:00 2001 From: "John M. P. Knox" Date: Wed, 29 Jul 2026 12:20:59 -0500 Subject: [PATCH] feat: show full escalation context before the operator decides MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the host (Claude Code) prompted the operator to decide an escalated deadlock, it showed only the bare operator_question — not the finding — so the operator decided without the context (claim, evidence, both positions) that resolved items DO get in `report show`. - New `impasse_report.py escalations ` renders ONLY the still-open deadlocks in full, reusing `_render_finding` for parity with `show`: claim, anchored evidence, both positions, deadlock kind, and the operator_question. - SKILL.md step 5 now requires the host to render + paste that full context BEFORE invoking AskUserQuestion. The command GUARANTEES full context or refuses: `_escalation_problems` is a total validator, and the CLI exits 2 (diagnostic on stderr, nothing on stdout) unless it can show, for every deadlock, the finding's claim + real anchored evidence (so the reviewer-response must be recorded under this review_id), both positions (item- or escalation-level), and a non-blank operator_question. It also rejects unrecognized/typo'd states (which would silently hide an escalation), duplicate finding_ids, and a reviewer-response whose own review_id doesn't match. Hardening surfaced by three cross-provider Impasse review rounds (18 findings, all verified + fixed): the shared render helpers now degrade rather than crash on untrusted reviewer data (non-dict evidence/anchor/ external_source, unhashable severity/state/result), required text is judged AFTER _clean (a control-char-only value counts as blank), and the whole load+validate+render runs inside one exception boundary. Tests: 33 escalations assertions (full-context render, only-deadlocks filter incl. resolved-but-escalated exclusion, positions-in-escalation, every refusal branch, totality on malformed input, CLI exit codes + stderr/no-stdout). 330 tests pass; ruff + schema gates green. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01X8XaW2bqgYnp7oRVgfHXid --- SKILL.md | 20 +++- scripts/impasse_report.py | 206 ++++++++++++++++++++++++++++++++++++-- tests/test_helpers.py | 138 +++++++++++++++++++++++++ 3 files changed, 354 insertions(+), 10 deletions(-) diff --git a/SKILL.md b/SKILL.md index f0021a8..73f7de4 100644 --- a/SKILL.md +++ b/SKILL.md @@ -224,8 +224,24 @@ backend is the cross-provider reviewer). The host is auto-detected (`IMPASSE_HOS 4. **Treat `response` as partially validated.** The runner confirms it's JSON with the required top-level fields; full schema validation runs in CI (`tests/validate_schemas.py`), not at runtime. Don't rely on fields the runner didn't check without validating them yourself. -5. **Verify, reconcile, and escalate** per the protocol. In Claude Code, put each deadlock's - `operator_question` to the operator with `AskUserQuestion`; batch multiple deadlocks. +5. **Verify, reconcile, and escalate** per the protocol. **Before you prompt the operator to + decide anything, show them the full escalated issue(s) — not just the question.** Build the + reconciliation with each deadlock's item fully populated (both positions and + `escalation.operator_question`, `state: deadlocked`), write it to a file, and render the pending + decisions in full: + ```bash + python3 "$IMPASSE_ROOT/scripts/impasse_report.py" escalations + ``` + This prints each escalated finding with the **same detail `show` gives resolved items** — the + reviewer's claim, its anchored evidence, both positions, and the question — so the operator + decides with full context, never a bare question stripped of what it's about. **It refuses + (non-zero exit) unless it can show full context for every deadlock** — the finding's claim + + anchored evidence (so the reviewer-response must be recorded under this `review_id`), both + positions, and the `operator_question` — so you cannot accidentally prompt with a partial view; + fix the reconciliation and retry. **Paste that rendered output** to + the operator. THEN, in Claude Code, put each deadlock's `operator_question` to the operator with + `AskUserQuestion` (batch multiple deadlocks). After they answer, move each decided item to + `resolved` (their ruling as the `resolution`) and save it (step 6). **Operator rulings count as escalations regardless of channel.** If an operator ruling decides an item's disposition — whether the question traveled through a formal diff --git a/scripts/impasse_report.py b/scripts/impasse_report.py index 4a4c89c..1dde22e 100644 --- a/scripts/impasse_report.py +++ b/scripts/impasse_report.py @@ -51,7 +51,15 @@ def _wrap(label: str, text: str, cont: str = " ") -> str: return textwrap.fill(_clean(text), width=96, initial_indent=label, subsequent_indent=cont) +def _safe_get(mapping: dict, key, default): + """`mapping.get` that never raises on an unhashable (malformed, non-string) key from untrusted + reviewer output — a JSON array/object where a string was expected returns `default`, not TypeError.""" + return mapping.get(key, default) if isinstance(key, str) else default + + def _anchor_desc(anchor: dict) -> str: + if not isinstance(anchor, dict): # untrusted reviewer output — a non-dict anchor must not crash the render + return "?" t = anchor.get("type") if t == "file_range": loc = anchor.get("path", "") @@ -73,30 +81,60 @@ def _anchor_desc(anchor: dict) -> str: return t or "?" +def _item_position(item: dict, key: str): + """A reconciliation item's position, from the item or (schema-permitted) its escalation object.""" + v = item.get(key) + if isinstance(v, str) and v: + return v + esc = item.get("escalation") + return esc.get(key) if isinstance(esc, dict) else None + + +def _renders_nonblank(v) -> bool: + """True iff `v` is a string that still has content AFTER `_clean` strips control chars — so a + required field that is only terminal-escape bytes (blank once cleaned) does not pass as present.""" + return isinstance(v, str) and _clean(v).strip() != "" + + +def _has_anchored_evidence(evidence) -> bool: + """True iff at least one evidence entry RENDERS as real anchored evidence: a dict whose anchor is a + dict that yields a genuine locator (not blank/'?') AND a non-blank observation. A dict-shaped but + empty anchor (`{}`) or a blank observation is hollow context — it must not pass as full context.""" + if not isinstance(evidence, list): + return False + return any(isinstance(e, dict) and isinstance(e.get("anchor"), dict) + and _anchor_desc(e["anchor"]).strip() not in ("", "?") + and _renders_nonblank(e.get("observation")) + for e in evidence) + + def _render_finding(f: dict, item: dict | None) -> list[str]: lines = [] - sev = SEVERITY.get(f.get("severity"), _clean(f.get("severity", "?"))) - state = STATE.get((item or {}).get("state"), "🔎 raised (not yet reconciled)") + sev = _safe_get(SEVERITY, f.get("severity"), _clean(f.get("severity", "?"))) + state = _safe_get(STATE, (item or {}).get("state"), "🔎 raised (not yet reconciled)") cat = _clean(f.get("category", "")) lines.append(f"{_clean(f.get('id', '?'))} {sev} {state}" + (f" · {cat}" if cat else "")) lines.append(_wrap(" 🔎 Reviewer: ", f.get("claim", ""))) for ev in f.get("evidence", []): + if not isinstance(ev, dict): # untrusted reviewer output — a non-dict entry must not crash the render + continue desc = _anchor_desc(ev.get("anchor", {})) obs = ev.get("observation", "") grounding = ev.get("grounding", "") lines.append(_wrap(" 📌 Evidence: ", f"{desc} — {obs} [{grounding}]")) - if ev.get("external_source"): - src = ev["external_source"] + src = ev.get("external_source") + if isinstance(src, dict): lines.append(_wrap(" ↗ source: ", src.get("uri") or src.get("title") or "external source")) if item: - vers = item.get("verification") or [] + vers = [v for v in (item.get("verification") or []) if isinstance(v, dict)] if vers: - checks = " · ".join(f"{_clean(v.get('method'))} {VRESULT.get(v.get('result'), _clean(v.get('result')))}" for v in vers) + checks = " · ".join(f"{_clean(v.get('method'))} {_safe_get(VRESULT, v.get('result'), _clean(v.get('result')))}" for v in vers) lines.append(f" 🧪 Verified: {checks}") for v in vers: if v.get("detail"): lines.append(_wrap(" ", v["detail"])) - rp, hp = item.get("reviewer_position"), item.get("host_position") + # positions may sit on the item or (schema-permitted) inside the escalation — show either + rp, hp = _item_position(item, "reviewer_position"), _item_position(item, "host_position") if rp or hp: lines.append(" 🗣️ Back-and-forth:") if rp: @@ -200,6 +238,121 @@ def render_findings(response: dict) -> str: return "\n".join(out) +_RECOGNIZED_STATES = frozenset({"accepted", "rejected", "resolved", "deadlocked", "withdrawn"}) + + +def _escalation_problems(rec: dict, rev: dict | None) -> list: + """Return the reasons the escalations view CANNOT show full context for every pending decision — + empty means safe to render. This is the guarantee behind the feature: the operator must never be + prompted with a partial view (a bare question, or a deadlock whose claim/evidence isn't on disk), + which is exactly the defect it fixes. So a deadlock missing its finding context (claim + anchored + evidence), positions, or question — an item with an UNRECOGNIZED state (a typo that would silently + hide an escalation), a duplicate finding_id, or a reviewer-response that isn't the one for this + review — is a hard error, not something a hollow render papers over. TOTAL: never raises, even on + malformed input; reviewer findings are UNTRUSTED. Required text is judged AFTER `_clean`, so a + control-char-only value (blank once rendered) counts as missing.""" + problems = [] + if not isinstance(rec, dict): + return ["reconciliation is not an object"] + items = rec.get("items") + if not isinstance(items, list): + return ["reconciliation 'items' is not a list"] + for i, it in enumerate(items): + if not isinstance(it, dict): + problems.append(f"item[{i}] is not an object") + elif not (isinstance(it.get("state"), str) and it["state"] in _RECOGNIZED_STATES): + # isinstance guard first: a non-string (e.g. a JSON list) is unhashable and would raise + # on the set membership test — this function must stay total. + problems.append(f"item[{i}] (finding {it.get('finding_id')!r}) has an unrecognized state " + f"{it.get('state')!r} — a real escalation could be silently hidden") + # finding_ids must be unique (JSON Schema can't enforce it — the runner/CI must): a duplicate + # would render the same decision twice and inflate the count. + seen = {} + for it in items: + fid = it.get("finding_id") if isinstance(it, dict) else None + if isinstance(fid, str): + seen[fid] = seen.get(fid, 0) + 1 + problems += [f"duplicate finding_id {k!r} across items" for k, n in sorted(seen.items()) if n > 1] + opens = _open_escalations(rec) + if not opens: + return problems # nothing to decide (or only the structural problems above) + rid = rec.get("review_id") + if not (isinstance(rid, str) and rid): + problems.append("review_id is missing/not a string, so the reviewer-response (finding claims + " + "evidence) can't be located") + findings = {} + if isinstance(rev, dict): + if rev.get("review_id") != rid: # the loaded record must be THIS review's, not a crossed one + problems.append(f"reviewer-response review_id {rev.get('review_id')!r} does not match the " + f"reconciliation's {rid!r} — refusing to show another review's findings") + revf = rev.get("findings") + if isinstance(revf, list): + dup_fids = set() + for f in revf: + if isinstance(f, dict) and isinstance(f.get("id"), str): + if f["id"] in findings: # duplicate reviewer finding id: can't tell which the deadlock means + dup_fids.add(f["id"]) + findings[f["id"]] = f + problems += [f"reviewer-response has a duplicate finding id {k!r} — ambiguous which the " + "deadlock refers to" for k in sorted(dup_fids)] + else: + problems.append("reviewer-response 'findings' is not a list") + elif rev is None: + problems.append(f"reviewer-response not found for review_id {rid!r} — run the FULL protocol so " + "the findings are recorded, or point at the correct review_id") + else: + problems.append("reviewer-response is malformed (not an object)") + for it in opens: + fid = it.get("finding_id") + label = f"deadlock {fid!r}" + if not (isinstance(fid, str) and fid): + problems.append(f"{label}: finding_id is missing or not a string") + elif isinstance(rev, dict): + f = findings.get(fid) + if f is None: + problems.append(f"{label}: no matching finding in the reviewer-response — its claim/" + "evidence can't be shown") + else: + if not _renders_nonblank(f.get("claim")): + problems.append(f"{label}: the matched finding has no claim text") + if not _has_anchored_evidence(f.get("evidence")): + problems.append(f"{label}: the matched finding has no anchored evidence to show") + if not _renders_nonblank(_item_position(it, "reviewer_position")): + problems.append(f"{label}: missing reviewer_position") + if not _renders_nonblank(_item_position(it, "host_position")): + problems.append(f"{label}: missing host_position") + esc = it.get("escalation") + if not (isinstance(esc, dict) and _renders_nonblank(esc.get("operator_question"))): + problems.append(f"{label}: missing escalation.operator_question (the footer promises one)") + return problems + + +def render_escalations(rec: dict, rev: dict | None) -> str: + """Render, IN FULL, only the items still awaiting the operator — the deadlocks — so they see each + escalated issue's evidence and both positions BEFORE being asked to decide, symmetric with how + resolved items appear in `show`. `rec` is the (draft or saved) reconciliation the deadlock items + live in; `rev` is the reviewer-response holding the finding claims/evidence (its findings are + UNTRUSTED — `_render_finding` cleans them). Presentation only, and the sanctioned path is the CLI + subcommand, which runs `_escalation_problems` FIRST and refuses a partial view — so this ASSUMES + validated input. It does not itself re-validate (a direct importer must run the check), but the + shared render helpers are hardened so malformed sub-structures degrade rather than crash.""" + opens = _open_escalations(rec) + review_id = _clean(rec.get("review_id") or (rev or {}).get("review_id") or "?") + if not opens: + return f"✅ No escalated decisions for '{review_id}' — nothing needs you." + findings = {f["id"]: f for f in ((rev or {}).get("findings") or []) + if isinstance(f, dict) and isinstance(f.get("id"), str)} + out = [f"⚖️ {len(opens)} decision(s) need you — full context before you choose", + f" review: {review_id}", + "─" * 78] + for it in opens: + fid = it.get("finding_id") + out += _render_finding(findings.get(fid) or {"id": fid}, it) + out.append("─" * 78) + out.append("Answer each ❓ question; your ruling becomes that item's resolution.") + return "\n".join(out) + + def lifetime_recap() -> str: """A short, honest value recap across every reconciled run on disk — printed at the end of a `show` so the operator sees what independent review has surfaced for them. Facts only: @@ -243,7 +396,7 @@ def _open_escalations(rec: dict) -> list: """Items still deadlocked — an escalation the operator hasn't resolved yet. Once the operator decides, the host re-saves the reconciliation with that item moved to 'resolved', so it stops showing as open.""" - return [it for it in (rec.get("items") or []) if it.get("state") == "deadlocked"] + return [it for it in (rec.get("items") or []) if isinstance(it, dict) and it.get("state") == "deadlocked"] def open_runs() -> list: @@ -286,6 +439,8 @@ def _main(argv=None) -> int: s.add_argument("run_id") fnd = sub.add_parser("findings", help="render a reviewer-response's raw findings (a review's --raw output)") fnd.add_argument("path", help="a reviewer-response JSON file, or a review result JSON (uses its .response)") + esc = sub.add_parser("escalations", help="render, IN FULL, the deadlocks awaiting the operator in a (draft) reconciliation — show BEFORE prompting for decisions") + esc.add_argument("path", help="a reconciliation-result JSON (draft or saved); its review_id locates the reviewer-response for finding context") sr = sub.add_parser("save-reconciliation") sr.add_argument("path") fg = sub.add_parser("forget") @@ -357,6 +512,41 @@ def _main(argv=None) -> int: return 2 print(render_findings(resp)) return 0 + if args.cmd == "escalations": + try: + with open(args.path, encoding="utf-8") as f: + rec = json.load(f) + except (OSError, ValueError) as e: + print(f"cannot read reconciliation file: {e}", file=sys.stderr) + return 2 + if not (isinstance(rec, dict) and isinstance(rec.get("items"), list)): + print("not a reconciliation-result (no 'items' list) — expected a reconciliation JSON " + "(draft or saved).", file=sys.stderr) + return 2 + rid = rec.get("review_id") + # One boundary around load + validate + render: untrusted/malformed data must yield a + # controlled exit 2, never a traceback (the validator is total, but load_run/render can still + # raise on storage or degenerate input). + try: + rev = lib.load_run(rid).get("reviewer_response") if isinstance(rid, str) and rid else None + problems = _escalation_problems(rec, rev) + if problems: + out = None + else: + out = render_escalations(rec, rev) + except Exception as e: + print(f"escalations: could not prepare the view: {e}", file=sys.stderr) + return 2 + if out is None: # refuse to prompt with a partial view — the whole point is full context + print("escalations: cannot show full context for every pending decision — refusing to " + "present a partial view:", file=sys.stderr) + for p in problems: + print(f" - {p}", file=sys.stderr) + print("Populate each deadlock's positions + operator_question and ensure the reviewer-" + "response is recorded under this review_id, then retry.", file=sys.stderr) + return 2 + print(out) + return 0 if args.cmd == "save-reconciliation": try: with open(args.path, encoding="utf-8") as f: diff --git a/tests/test_helpers.py b/tests/test_helpers.py index 2059a97..a75ff40 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -1462,6 +1462,144 @@ def _spy_cwd(argv, **kw): deleted3, _k = report.prune(1, include_open=True) check("old-open" in deleted3, "housekeeping: prune --include-open removes even open runs") + # --- escalations view: GUARANTEE full deadlock context BEFORE the operator decides (symmetric with `show`) --- + esc_rev = {"schema_version": "1.0", "review_id": "esc-run", "findings": [ + {"id": "F001", "severity": "high", "category": "correctness", "claim": "CLAIM_TEXT_UNIQUE", + "evidence": [{"anchor": {"type": "file_range", "path": "anchor_a.py", "line_start": 1, "line_end": 2}, + "observation": "EVIDENCE_OBS_UNIQUE", "grounding": "artifact_observed"}]}, + {"id": "F002", "severity": "low", "category": "style", "claim": "RESOLVED_CLAIM", "evidence": []}, + {"id": "F003", "severity": "medium", "category": "x", "claim": "HISTORIC_ESC_CLAIM", "evidence": []}]} + + def _esc_item(fid, state, **kw): + it = {"finding_id": fid, "state": state} + it.update(kw) + return it + + def _deadlock(fid="F001", q="OPERATOR_Q_UNIQUE?"): + esc = {"dispute_kind": "value_or_priority_tradeoff", "stop_reason": "operator_authority_required"} + if q is not None: + esc["operator_question"] = q + return _esc_item(fid, "deadlocked", reviewer_position="REVIEWER_POS_UNIQUE", + host_position="HOST_POS_UNIQUE", escalation=esc) + + esc_rec = {"schema_version": "1.0", "reconciliation_id": "y", "review_id": "esc-run", + "outcome": "deadlocked", "items": [ + _deadlock(), + _esc_item("F002", "resolved", resolution="already decided"), + # a RESOLVED item that still carries historical escalation data must NOT re-appear: + _esc_item("F003", "resolved", resolution="ruled", + escalation={"dispute_kind": "x", "stop_reason": "y", "operator_question": "OLD_Q?"})]} + _e = report.render_escalations(esc_rec, esc_rev) + check(all(t in _e for t in ("CLAIM_TEXT_UNIQUE", "anchor_a.py:1-2", "EVIDENCE_OBS_UNIQUE", + "REVIEWER_POS_UNIQUE", "HOST_POS_UNIQUE", "OPERATOR_Q_UNIQUE?", "ESCALATED")), + "escalations: renders full deadlock context (claim, evidence anchor, both positions, question)") + check(not any(t in _e for t in ("already decided", "RESOLVED_CLAIM", "HISTORIC_ESC_CLAIM", "OLD_Q?")), + "escalations: shows ONLY pending deadlocks — resolved items (even ones with old escalation data) are excluded") + check("nothing needs you" in report.render_escalations( + {"review_id": "r", "items": [{"finding_id": "F001", "state": "resolved"}]}, esc_rev), + "escalations: no deadlocks -> 'nothing needs you'") + # sanitize UNTRUSTED text across every field a finding/item contributes (claim, position, question) + _e3 = report.render_escalations( + {"review_id": "r", "items": [_deadlock(q="q\x1b[0m?") | {"reviewer_position": "p\x1b[31mos"}]}, + {"findings": [{"id": "F001", "claim": "c\x1b[1mlaim", "severity": "high"}]}) + check("\x1b" not in _e3, "escalations: sanitizes terminal escapes in untrusted finding/item text") + + # _escalation_problems is the guarantee: it must flag EVERY way full context could be missing. + check(report._escalation_problems(esc_rec, esc_rev) == [], "escalations: a fully-populated deadlock has no problems") + + def _one(rec, rev): # at least one problem flagged + return len(report._escalation_problems(rec, rev)) >= 1 + + def _mk(items): + return {"review_id": "esc-run", "items": items} + check(_one(_mk([_deadlock()]), None), "escalations problem: missing reviewer-response (e.g. --no-record run)") + check(_one(_mk([_deadlock(fid="NOPE")]), esc_rev), "escalations problem: deadlock finding_id has no matching finding") + check(_one(_mk([_deadlock(fid="F002")]), esc_rev), "escalations problem: matched finding has no anchored evidence") + check(_one(_mk([_deadlock(q=None)]), esc_rev), "escalations problem: deadlock missing operator_question") + check(_one(_mk([_deadlock() | {"reviewer_position": ""}]), esc_rev), "escalations problem: deadlock missing a position") + check(_one(_mk([_deadlock() | {"reviewer_position": "\x1b"}]), esc_rev), + "escalations problem: a control-char-only position is blank once rendered") + check(_one({"review_id": None, "items": [_deadlock()]}, esc_rev), "escalations problem: missing/non-string review_id") + check(_one(_mk([_deadlock()]), {"review_id": "OTHER", "findings": esc_rev["findings"]}), + "escalations problem: the loaded reviewer-response is for a different review_id") + check(_one(_mk([_deadlock(), _deadlock()]), esc_rev), "escalations problem: duplicate finding_id across items") + check(_one(_mk([{"finding_id": "F001", "state": "deadlock"}]), esc_rev), # typo'd state + "escalations problem: an unrecognized state (could silently hide an escalation)") + check(_one(_mk([_deadlock()]), {"review_id": "esc-run", "findings": 5}), + "escalations problem: reviewer-response findings is not a list (total, no crash)") + check(report._escalation_problems(_mk([_esc_item("F002", "resolved", resolution="x")]), esc_rev) == [], + "escalations: an all-resolved reconciliation is problem-free (nothing to decide)") + # evidence must render to REAL anchored content, not merely be dict-shaped (an empty/blank anchor is hollow) + check(_one(_mk([_deadlock(fid="F9")]), + {"review_id": "esc-run", "findings": [{"id": "F9", "claim": "c", "evidence": [{"anchor": {}, "observation": "obs"}]}]}), + "escalations problem: evidence anchor is empty/unrenderable") + check(_one(_mk([_deadlock(fid="F9")]), + {"review_id": "esc-run", "findings": [{"id": "F9", "claim": "c", "evidence": [{"anchor": {"type": "file_range", "path": "p"}, "observation": ""}]}]}), + "escalations problem: evidence observation is blank") + # totality: malformed shapes yield problems, never a raise + check(_one(_mk([{"finding_id": "F1", "state": ["not", "a", "string"]}]), esc_rev), + "escalations problem: an unhashable item state is flagged, not crashed (validator stays total)") + check(report._escalation_problems("not a dict", esc_rev) == ["reconciliation is not an object"], + "escalations: a non-dict reconciliation is total (no crash)") + # duplicate finding id in the reviewer-response itself -> ambiguous which the deadlock means + _rev_dup = {"review_id": "esc-run", "findings": [ + {"id": "F1", "claim": "a", "evidence": [{"anchor": {"type": "file_range", "path": "p", "line_start": 1}, "observation": "o"}]}, + {"id": "F1", "claim": "b", "evidence": [{"anchor": {"type": "file_range", "path": "q", "line_start": 2}, "observation": "o2"}]}]} + check(_one(_mk([_deadlock(fid="F1")]), _rev_dup), "escalations problem: duplicate finding id in the reviewer-response") + + def _good_rev_for(rid): # a complete, single, matching finding — everything valid except review_id + return {"review_id": rid, "findings": [{"id": "F1", "claim": "c", + "evidence": [{"anchor": {"type": "file_range", "path": "p", "line_start": 1}, "observation": "o"}]}]} + # review_id association is DECISIVE on its own: identical rec/finding, only the response's own review_id differs + check(report._escalation_problems(_mk([_deadlock(fid="F1")]), _good_rev_for("esc-run")) == [], + "escalations: matching review_id + complete finding -> no problems (de-confounds the mismatch test)") + check(_one(_mk([_deadlock(fid="F1")]), _good_rev_for("OTHER")), + "escalations problem: same complete finding but the response's own review_id differs -> rejected") + # a non-string (unhashable) severity must not crash the renderer + check("F1" in report.render_escalations(_mk([_deadlock(fid="F1")]), + {"review_id": "esc-run", "findings": [{"id": "F1", "claim": "c", "severity": ["oops"], + "evidence": [{"anchor": {"type": "file_range", "path": "p", "line_start": 1}, "observation": "o"}]}]}), + "escalations: a non-string severity renders without crashing") + # positions stored INSIDE the escalation object (schema-permitted) are accepted AND rendered + _pos_esc = {"finding_id": "F001", "state": "deadlocked", + "escalation": {"dispute_kind": "x", "stop_reason": "y", "operator_question": "q?", + "reviewer_position": "RP_IN_ESC", "host_position": "HP_IN_ESC"}} + check(report._escalation_problems(_mk([_pos_esc]), esc_rev) == [], + "escalations: positions inside the escalation object are accepted (schema-valid placement)") + check(all(t in report.render_escalations(_mk([_pos_esc]), esc_rev) for t in ("RP_IN_ESC", "HP_IN_ESC")), + "escalations: renders positions that live in the escalation object") + + # CLI dispatch: a fully-populated draft renders + exit 0; every incomplete/bad input -> exit 2 (never a partial view) + lib.save_run_doc("esc-run", "reviewer-response", esc_rev) + def _esc_cli(rec_obj): + p = os.path.join(tmp, "esc-cli.json") + with open(p, "w") as f: + _json.dump(rec_obj, f) + ob, eb = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(ob), contextlib.redirect_stderr(eb): + rc = report._main(["escalations", p]) + return rc, ob.getvalue(), eb.getvalue() + _rc, _outp, _errp = _esc_cli(esc_rec) + check(_rc == 0 and "CLAIM_TEXT_UNIQUE" in _outp and "OPERATOR_Q_UNIQUE?" in _outp, + "escalations CLI: fully-populated draft -> exit 0 + full context (reviewer-response loaded by review_id)") + # a refusal writes the diagnostic to STDERR and prints NOTHING to stdout (no partial view can leak) + _rc2, _out2, _err2 = _esc_cli({"review_id": "esc-run", "items": [_deadlock(q=None)]}) + check(_rc2 == 2 and _out2 == "" and "refusing to present a partial view" in _err2, + "escalations CLI: a deadlock missing its question -> exit 2, diagnostic on stderr, empty stdout") + check(_esc_cli({"review_id": "no-such-run", "items": [_deadlock()]})[0] == 2, + "escalations CLI: no recorded reviewer-response for the review_id -> exit 2") + check(_esc_cli({"review_id": "esc-run", "items": [123, "nope"]})[0] == 2, + "escalations CLI: non-dict items -> exit 2, not a traceback") + # untrusted malformed reviewer data (unhashable finding id) must fail controlled, not traceback + lib.save_run_doc("bad-rev", "reviewer-response", {"schema_version": "1.0", "review_id": "bad-rev", + "findings": [{"id": ["not", "hashable"], "claim": "x"}]}) + check(_esc_cli({"review_id": "bad-rev", "items": [_deadlock()]})[0] == 2, + "escalations CLI: malformed reviewer-response (unhashable id) -> exit 2, not a crash") + _badf = os.path.join(tmp, "not-rec.json") + with open(_badf, "w") as f: + f.write('{"hello": "world"}') + check(report._main(["escalations", _badf]) == 2, "escalations CLI: a non-reconciliation file -> exit 2") + # --- hardening fixes surfaced by the cross-provider code audit --- check(lib._safe_id("..") == "unknown" and lib._safe_id(".") == "unknown", "safe_id: '.'/'..' collapse to 'unknown' (no traversal)") check("/" not in lib._safe_id("a/b/../../etc"), "safe_id: path separators collapsed")