diff --git a/ouroboros/review_state.py b/ouroboros/review_state.py index f88679e7d..c392b8f92 100644 --- a/ouroboros/review_state.py +++ b/ouroboros/review_state.py @@ -439,13 +439,18 @@ def advisory_commit_ready( ) -> bool: """SSOT for every ``repo_commit_ready`` projection (H5, capinv-447). - Mirrors the real advisory gate: fresh/bypassed/skipped coverage, or a + Mirrors the real advisory gate: Cyber retains action authority; otherwise + fresh/bypassed/skipped coverage, or a typed technical failure permitted under owner-selected advisory enforcement. ``matching_run`` is supplied only after the caller matches current repo/hash; permission never changes its failure status or makes it fresh. Obligations and debt block only under blocking enforcement. Triad, scope, custody and every other commit requirement remain independent. """ + from ouroboros.tools.review_helpers import review_enforcement_blocks + + if not review_enforcement_blocks("blocking"): + return True if not effectively_fresh: from ouroboros.config import get_review_enforcement from ouroboros.tools.commit_gate import review_failure_is_technical diff --git a/ouroboros/skill_review_status.py b/ouroboros/skill_review_status.py index a5f447127..ff8088c70 100644 --- a/ouroboros/skill_review_status.py +++ b/ouroboros/skill_review_status.py @@ -200,14 +200,14 @@ def skill_review_gate( ) -> Dict[str, Any]: """Structured, agent-facing explanation of whether a review is executable. - A current author acceptance admits changed bytes only in Advisory; stale - still describes the original reviewer evidence, never the author hash. + Advisory author acceptance binds changed bytes; Cyber retains judgment even + without a prior verdict. Stale describes reviewer evidence, not author intent. Author fields are optional and appear only with a valid author disposition; callers without one retain the frozen gate key set. Deterministic hard-gate failures (e.g. skill_preflight) are persisted as STATUS_PENDING by `_run_deterministic_preflight`, so they are non-executable - here under every enforcement mode without needing per-caller findings — only + here outside Cyber without needing per-caller findings — only LLM blocker verdicts are overridable by advisory enforcement. ``findings`` is optional: a caller that has the persisted findings gets a @@ -224,8 +224,7 @@ def skill_review_gate( from ouroboros.review_records import validate_author_disposition author = validate_author_disposition(author_disposition) - author_current = bool(current_hash and author and author["subject_hash"] == current_hash - and author["enforcement"] == "advisory") + author_current = bool(current_hash and author and author["subject_hash"] == current_hash) raw_status = normalize_skill_review_status(status) if enforcement is None: try: @@ -234,11 +233,19 @@ def skill_review_gate( except Exception: enforcement = "blocking" enforcement = str(enforcement or "blocking").lower() - if raw_status == STATUS_PENDING: + from ouroboros.tools.review_helpers import review_enforcement_blocks + + cyber = not review_enforcement_blocks("blocking") + if cyber: + executable = True + reason = "cyber_authority" + summary = ("Cyber Pro permits acting on this payload by Ouroboros's judgment; " + "review status, staleness and failures remain independent evidence, not a PASS.") + elif raw_status == STATUS_PENDING: executable = False reason = "review_pending" summary = "Review is pending or did not produce an executable verdict." - elif author_current and enforcement == "advisory": + elif author_current and author["enforcement"] == "advisory" and enforcement == "advisory": executable = True reason = "author_accepted_advisory" summary = "The author accepted the current payload under Advisory; the original reviewer verdict and hash are unchanged." @@ -270,7 +277,7 @@ def skill_review_gate( return { "status": raw_status or STATUS_PENDING, "stale": bool(stale), - **({"author_accepted": reason == "author_accepted_advisory", + **({"author_accepted": reason == "author_accepted_advisory" or (cyber and author_current), "author_disposition": author} if author else {}), "executable_review": bool(executable), "blocking_reason": reason, diff --git a/ouroboros/task_results.py b/ouroboros/task_results.py index fa46fdcd7..4dbaad8bc 100644 --- a/ouroboros/task_results.py +++ b/ouroboros/task_results.py @@ -1134,19 +1134,15 @@ def plan_review_gate_projection( *, hard_rail: str = "", ) -> Dict[str, Any]: - """Project one plan-review finalization decision from existing authority. + """Project finalization permission without changing the durable review facts. - ``plan_review_state`` is the durable SSOT; the ``current_attempt`` pointer keeps a - newer fingerprint from falling back to an older closed wave. Statuses: ``closed`` - (allow) · ``rail_degraded`` (a task-wide rail released the hold — allow) · - ``advisory_open`` (advisory enforcement proceeds under loud disclosure) · - ``cycles_exhausted`` (the shared cap is spent on an OPEN wave: finalization is - released so the task can terminalize honestly as blocked — owner D27 — while - the wave itself stays open) · ``open`` / ``unavailable`` / ``pending`` / - ``legacy_open_requires_resubmission`` (blocking hold; EXCEPT an ``open`` wave - whose ``quorum_unreachable`` typed fact holds — B2b — which releases - finalization the same honest-blocked way while staying open) · ``absent``. Accepts a v2 - state, a loaded v1 wrapper, or a raw v1 record (read-only projection).""" + The current-attempt pointer prevents an older closed wave authorizing new + work. In ordinary Blocking, open/unavailable/pending/legacy-open reviews hold + finalization; spent cycles (D27), unreachable quorum (B2b) or a hard rail + release it for an honest blocked outcome. Advisory releases an open review. + Cyber retains judgment even with missing evidence: allow never implies closed + or PASS. Accepts v2 state, a v1 wrapper or raw v1 as a read-only projection. + """ policy = "blocking" if str(enforcement or "").lower() == "blocking" else "advisory" control: Dict[str, Any] = {} attempted = False @@ -1202,8 +1198,13 @@ def plan_review_gate_projection( status = str(control.get("status") or "unavailable") closed = bool(control.get("closed")) + from ouroboros.tools.review_helpers import review_enforcement_blocks + + cyber = not review_enforcement_blocks("blocking") if status == "closed" and closed: gate_status, allow = "closed", True + elif cyber: + gate_status, allow = "advisory_open", True elif hard_rail or status == "rail_degraded": gate_status, allow = "rail_degraded", True elif policy == "advisory" and status in { @@ -1225,6 +1226,7 @@ def plan_review_gate_projection( gate_status, allow = status, False return { "enforcement": policy, + **({"decision_authority": "cyber_pro", "review_status": status} if cyber else {}), "status": gate_status, "allow": allow, "attempted": attempted, diff --git a/ouroboros/tools/claude_advisory_review.py b/ouroboros/tools/claude_advisory_review.py index 1d19a04d3..29e76825a 100644 --- a/ouroboros/tools/claude_advisory_review.py +++ b/ouroboros/tools/claude_advisory_review.py @@ -769,6 +769,16 @@ def _next_step_guidance(latest: Optional["AdvisoryRunRecord"], state: "AdvisoryR one unbindable case stays as before: an uncomputable current hash cannot establish a mismatch either way. """ + from ouroboros.tools.review_helpers import review_enforcement_blocks + + if not review_enforcement_blocks("blocking"): + return ( + f"Cyber Pro: preflight status={getattr(latest, 'status', 'missing')}; " + f"stale={bool(stale_from_edit or not effective_is_fresh)}. " + "Ouroboros decides whether to continue or request more feedback. " + "Original findings, missing evidence and pending operations remain recorded; this is not a PASS." + ) + def _debt_hint() -> str: parts = [] if open_obs: diff --git a/ouroboros/tools/commit_gate.py b/ouroboros/tools/commit_gate.py index fde7edd2b..dbbe9f4d3 100644 --- a/ouroboros/tools/commit_gate.py +++ b/ouroboros/tools/commit_gate.py @@ -18,6 +18,7 @@ from ouroboros.review_state import ( infer_review_phase, ) from ouroboros.tools.registry import ToolContext +from ouroboros.tools.review_helpers import review_enforcement_blocks from ouroboros.utils import ( truncate_review_artifact as _truncate_review_reason, ) @@ -624,11 +625,12 @@ def _record_commit_attempt( if author_disposition is not None: subject = pre_review_fingerprint or str(getattr(existing, "pre_review_fingerprint", "") or "") author_record = validate_author_disposition(author_disposition, subject_hash=subject) or {} - if (not subject or not getattr(existing, "paid", False) + cyber = not review_enforcement_blocks("blocking") + if (not subject or (not cyber and not getattr(existing, "paid", False)) or subject != getattr(existing, "pre_review_fingerprint", "") or (post_review_fingerprint and post_review_fingerprint != subject) - or get_review_enforcement() != "advisory" - or author_record.get("enforcement") != "advisory"): + or review_enforcement_blocks(get_review_enforcement()) + or (not cyber and author_record.get("enforcement") != "advisory")): author_record = {} attempt = CommitAttemptRecord( ts=_utc_now(), @@ -708,6 +710,8 @@ def _record_commit_attempt( getattr(existing, "review_owner_pid", 0) or 0 ), ) + if status != "reviewing" and "late_result_pending" not in legacy_kwargs and not review_enforcement_blocks("blocking"): + attempt.late_result_pending = bool(getattr(existing, "late_result_pending", False)) or _attempt_has_active_review_custody(attempt) stamp_paid_review_owner(attempt, paid=bool(paid)) state.record_attempt(attempt, semantic_redirects=_obligation_redirects) @@ -809,6 +813,7 @@ def _check_overlapping_review_attempt(ctx: ToolContext) -> Optional[str]: expiration_window = _REVIEW_ATTEMPT_TTL_SEC + _REVIEW_ATTEMPT_GRACE_SEC ctx._review_resume_pending = False ctx._pending_review_attempt = None + ctx._review_cyber_pending = "" def _mutate(state): state.expire_stale_attempts(now_ts=_utc_now()) @@ -821,6 +826,9 @@ def _check_overlapping_review_attempt(ctx: ToolContext) -> Optional[str]: active_attempts = update_state(pathlib.Path(ctx.drive_root), _mutate) except Exception as e: log.warning("Failed to check overlapping review attempts: %s", e) + if not review_enforcement_blocks("blocking"): + ctx._review_cyber_pending = f"Review custody is unreadable: {e}. No new reviewer will be dispatched." + return None return ( "⚠️ REVIEW_STATE_UNAVAILABLE: active paid-review custody could not " "be verified, so no reviewer dispatch was started. Retry after the " @@ -828,6 +836,13 @@ def _check_overlapping_review_attempt(ctx: ToolContext) -> Optional[str]: ) if not active_attempts: return None + if not review_enforcement_blocks("blocking"): + ctx._review_cyber_pending = ( + "Existing review custody remains active: " + + ", ".join(f"{item.tool_name}#{item.attempt}" for item in active_attempts) + + ". No new reviewer will be dispatched; the original attempts remain collectible." + ) + return None task_id = str(getattr(ctx, "task_id", "") or "") tool_name = _current_review_tool_name(ctx) @@ -930,6 +945,20 @@ def _check_advisory_freshness(ctx: ToolContext, commit_message: str, _record_advisory_override(ctx, warning) ctx._review_advisory = list(getattr(ctx, "_review_advisory", []) or []) + [warning, *matching_run.items] + if not review_enforcement_blocks("blocking"): + from ouroboros.tools.review import _record_advisory_override + + if not fresh or open_obs or open_debts: + warning = ("Cyber Pro: preflight status=" + str(getattr(matching_run, "status", "missing")) + + ("; current" if fresh else "; stale or unavailable") + + ". Ouroboros may continue; this does not create review evidence.\n" + + str(getattr(matching_run, "raw_result", "") or "") + + "\n" + "\n".join([*_render_obligations(), *_render_debts()])) + ctx._last_review_block_reason = "advisory_cyber_authority" + _record_advisory_override(ctx, warning) + ctx._review_advisory = list(getattr(ctx, "_review_advisory", []) or []) + [warning] + return None + if (fresh or technical_failure) and not open_obs and not open_debts: return None diff --git a/ouroboros/tools/git.py b/ouroboros/tools/git.py index 1c9699cf6..b7be7f9be 100644 --- a/ouroboros/tools/git.py +++ b/ouroboros/tools/git.py @@ -16,6 +16,7 @@ import time from typing import Any, Dict, List, Optional, Tuple from ouroboros.config import get_runtime_mode # noqa: F401 +from ouroboros.tools.review_helpers import review_enforcement_blocks from ouroboros.runtime_mode_policy import ( core_patch_notice, # noqa: F401 format_protected_paths, # noqa: F401 @@ -117,6 +118,8 @@ def _free_cycle_gate( disclosure and WITHOUT buying another review.""" from ouroboros.config import get_review_enforcement + if getattr(ctx, "_review_cyber_pending", "") and not review_enforcement_blocks("blocking"): + return {"advisory_replay": ctx._review_cyber_pending, "replay_reason": "review_pending"} fp = pre_fingerprint.get("fingerprint", "") rebuttal_sha = compute_rebuttal_sha256(review_rebuttal) contract_fp = commit_review_contract_fingerprint() @@ -169,7 +172,7 @@ def _free_cycle_gate( cycles_paid=int(ceiling["cycles_paid"]), cap=int(ceiling["cap"]), enforcement=enforcement, root_task_id=root_task_id, fingerprint=str(fp), ) - if enforcement != "blocking": + if not review_enforcement_blocks(enforcement): # ADVISORY: neither state hard-blocks a commit — disclose loudly (typed # event + result message) and reuse the recorded outcome for free. # The identical-replay half of this branch is structurally near-dead @@ -573,6 +576,12 @@ def _advisory_and_tests_gate( ctx, runner=lambda c, **kw: _run_review_preflight_tests(c, **kw)) if test_err: msg = _tests_preflight_block_message(_managed_needs_proof, test_err) + if not review_enforcement_blocks("blocking"): + from ouroboros.tools.review import _handle_review_block_or_warning + + ctx._last_review_block_reason = "tests_preflight_blocked" + _handle_review_block_or_warning(ctx, True, msg, "") + return None try: run_cmd(["git", "reset", "HEAD"], cwd=ctx.repo_dir) except Exception: diff --git a/ouroboros/tools/git_review_cycle.py b/ouroboros/tools/git_review_cycle.py index 5dee0caaf..ccd7cee14 100644 --- a/ouroboros/tools/git_review_cycle.py +++ b/ouroboros/tools/git_review_cycle.py @@ -1,14 +1,6 @@ -"""Staging, advisory/triad/scope review and reviewed-material binding for the -commit gate, split out of ``ouroboros/tools/git.py`` (v7 module-size -discipline). Every span is extracted VERBATIM from the parent's tip bytes by -scripts/v7next_transplant.py; the parent re-exports every moved name. -Parent-scope helpers the monolith read as module globals — including the -post-cutoff paid-cycle gate family — are read through the call-time handle -``_git()`` — never a from-import — so the facade binding stays the one tests -monkeypatch. ``_sanitize_git_error`` is the one f-string-read exception (the -byte gate cannot rewrite f-string internals): it binds the plumbing owner at -import time. -""" +"""Commit staging, review, continuation and binding. ``tools.git`` re-exports +these functions; ``_git()`` preserves its patchable facade bindings, while +neutral plumbing imports bind their own owner.""" from __future__ import annotations @@ -23,22 +15,15 @@ import time from typing import Any, Dict, List, Optional from ouroboros.tools.registry import ToolContext -from ouroboros.tools.git_plumbing import _sanitize_git_error -from ouroboros.tools.git_plumbing import _publish_git_error, _publish_review_blocked +from ouroboros.tools.git_plumbing import _sanitize_git_error, _publish_git_error, _publish_review_blocked +from ouroboros.tools.review_helpers import review_enforcement_blocks -# The parent's logger name is pinned so moved log records keep their %(name)s -# in server.log/stdout — the same logger object the parent binds. +# Keep the public facade's logger name in server/stdout records. log = logging.getLogger("ouroboros.tools.git") def _git(): - """The parent module, read at call time. - - The parent owns the rebindable module state and the members tests - monkeypatch there; reading them through the module at each call keeps - one binding, where a from-import would freeze the value this leaf saw - at import time (the owner-approved D18/D33 mechanical exception). - """ + """Read the public facade at call time so monkeypatches keep one binding.""" from ouroboros.tools import git return git @@ -276,8 +261,11 @@ def _finalize_pending_review( *, pre_fingerprint: Dict[str, Any], post_fingerprint: Dict[str, Any], -) -> str: - """Persist the non-terminal wave and leave its exact retry fail-closed.""" +) -> Optional[str]: + """Retain non-terminal custody; Cyber may continue without closing the wave.""" + if getattr(ctx, "_review_cyber_pending", "") and not review_enforcement_blocks("blocking"): + # The pending row belongs to an earlier invocation, not this free continuation. + return None custody_lost = bool(getattr(ctx, "_review_custody_lost", False)) message = ( "⚠️ REVIEW_CUSTODY_LOST: the paid review wave is still unresolved, but " @@ -312,6 +300,13 @@ def _finalize_pending_review( degraded_reasons=list(getattr(ctx, "_review_degraded_reasons", []) or []), review_retry_key=str(getattr(ctx, "_current_review_retry_key", "") or ""), ) + if not review_enforcement_blocks("blocking"): + from ouroboros.tools.review import _handle_review_block_or_warning + + ctx._last_review_block_reason = "review_late_result_pending" + _handle_review_block_or_warning(ctx, True, + "Physical review remains pending; its source and invocation are retained for collection.", "") + return None # The index is part of this live wave's identity. Retain it for exact # reconciliation; rebuilding it from the worktree could review new bytes. return message @@ -571,6 +566,9 @@ def _reconcile_advisory_before_preparation(ctx, commit_message, *, goal, scope, from ouroboros.tools.preflight_review_run import pending_advisory_execution ctx._advisory_reconciled = False + if not review_enforcement_blocks("blocking"): + # Commit continuation leaves each old critic's source/custody with its invocation. + return "" try: execution, _ = pending_advisory_execution( ctx, commit_message, goal=goal, scope=scope, paths=paths, review_rebuttal=review_rebuttal, @@ -774,7 +772,9 @@ def _run_reviewed_stage_cycle( # ships without a fresh review and the disclosure must say so. review_err, scope_result, triad_block_reason, triad_advisory = None, None, "", [] replay_reason = str(advisory_replay.get("replay_reason") or "") - if replay_reason == _git().IDENTICAL_DIFF_BLOCK_REASON: + if not review_enforcement_blocks("blocking"): + progress_note = "Cyber Pro: continuing without a new reviewer dispatch; original review facts are retained." + elif replay_reason == _git().IDENTICAL_DIFF_BLOCK_REASON: progress_note = ( "Max Review Cycles: identical staged diff — reusing the recorded " "review verdict, no paid triad+scope dispatch." @@ -790,6 +790,8 @@ def _run_reviewed_stage_cycle( f"this commit ({replay_reason}); no fresh automatic preflight was bought. " + str(advisory_replay.get("advisory_replay") or "") ) + if not review_enforcement_blocks("blocking"): + disclosure = "Cyber Pro: proceeding without a new review. " + str(advisory_replay.get("advisory_replay") or "") advisory_list = getattr(ctx, "_review_advisory", None) if isinstance(advisory_list, list): advisory_list.append(disclosure) @@ -826,16 +828,12 @@ def _run_reviewed_stage_cycle( if isinstance(advisory_list, list): advisory_list.extend(scope_advisory) post_fingerprint = _git()._fingerprint_staged_diff(pathlib.Path(ctx.repo_dir)) - if _git()._review_custody_pending(ctx): + if _git()._review_custody_pending(ctx) and (pending_message := _git()._finalize_pending_review( + ctx, commit_message, commit_start, + pre_fingerprint=pre_fingerprint, post_fingerprint=post_fingerprint)): return { "status": "blocked", - "message": _git()._finalize_pending_review( - ctx, - commit_message, - commit_start, - pre_fingerprint=pre_fingerprint, - post_fingerprint=post_fingerprint, - ), + "message": pending_message, "block_reason": ( "review_custody_lost" if bool(getattr(ctx, "_review_custody_lost", False)) @@ -980,10 +978,13 @@ def _run_non_committing_review_cycle( ) ctx._scope_review_history = {} outcome["message"] = ( + "Cyber Pro: review-only operation completed; independent failures and pending work remain recorded. " + if not review_enforcement_blocks("blocking") else "Review-only cycle completed under advisory enforcement; failed or missing review remains recorded. " if "review_technical_failure_advisory" in (getattr(ctx, "_review_degraded_reasons", []) or []) else "Review-only cycle passed. " - ) + "Commit was not created and the index was unstaged." + ) + ("Commit was not created; the index is retained while review custody is pending." + if _git()._review_custody_pending(ctx) else "Commit was not created and the index was unstaged.") return outcome finally: try: diff --git a/ouroboros/tools/parallel_review.py b/ouroboros/tools/parallel_review.py index 6f1794287..b176250a1 100644 --- a/ouroboros/tools/parallel_review.py +++ b/ouroboros/tools/parallel_review.py @@ -11,7 +11,7 @@ import time from ouroboros.utils import run_cmd from ouroboros.review_substrate import scope_reviewer_slots -from ouroboros.tools.review_helpers import build_scope_actor_record, format_review_history_entry +from ouroboros.tools.review_helpers import build_scope_actor_record, format_review_history_entry, review_enforcement_blocks from ouroboros.tools.scope_review import ( run_scope_review, ScopeReviewResult, @@ -160,7 +160,7 @@ def _format_scope_advisory_msg(scope_result) -> str: """Format advisory scope findings as a readable message (advisory enforcement path).""" parts = [] if scope_result.critical_findings: - parts.append("Scope advisory findings (enforcement=advisory):\n" + + parts.append("Scope review findings:\n" + "\n".join(f" • {f['item']}: {f.get('reason', '')}" for f in scope_result.critical_findings)) if scope_result.advisory_findings: @@ -493,7 +493,7 @@ def _run_scope(ctx, commit_message, scope_rows, dispatch, *, goal, scope, ) if partial_quorum_shortfall: from ouroboros.config import get_review_enforcement - if get_review_enforcement() == "blocking": + if review_enforcement_blocks(get_review_enforcement()): blocked = True block_messages.append(_qmsg) # Surface any non-blocking shortfall LOUDLY (advisory, never a silent @@ -742,7 +742,7 @@ def run_parallel_review( from ouroboros.config import get_review_enforcement blocking_review = bool((triad_prepared or {}).get( - "blocking_review", get_review_enforcement() == "blocking")) + "blocking_review", review_enforcement_blocks(get_review_enforcement()))) and review_enforcement_blocks("blocking") if not hasattr(ctx, "_review_degraded_reasons"): ctx._review_degraded_reasons = [] ctx._review_degraded_reasons.append( @@ -969,20 +969,22 @@ def aggregate_review_verdict(review_err, scope_result, triad_block_reason, triad "responded", "skipped_low_context_mode", "not_dispatched", }] technical_scope = bool(failed_scope) and all(review_failure_is_technical(row) for row in failed_scope) - if (get_review_enforcement() == "advisory" + cyber = not review_enforcement_blocks("blocking") + if cyber or (get_review_enforcement() == "advisory" and (not review_err or triad_block_reason == "fixed_overflow") and (scope_result is None or not scope_result.blocked or technical_scope)): from ouroboros.tools.review import _record_advisory_override disclosure = ( - "Review enforcement=advisory: technical review failure permits continuing " - "on the independently bound candidate; failed or missing review is not a PASS.\n" + ("Cyber Pro: independent review does not prohibit action " if cyber else + "Review enforcement=advisory: technical review failure permits continuing ") + + "on the independently bound candidate; failed or missing review is not a PASS.\n" + combined_msg ) ctx._last_review_block_reason = block_reason _record_advisory_override(ctx, disclosure) ctx._review_advisory.append(disclosure) - ctx._review_degraded_reasons = list(getattr(ctx, "_review_degraded_reasons", []) or []) + ["review_technical_failure_advisory"] + ctx._review_degraded_reasons = list(getattr(ctx, "_review_degraded_reasons", []) or []) + ["review_cyber_authority" if cyber else "review_technical_failure_advisory"] return False, combined_msg, block_reason, _combined_findings, _scope_advisory_items return True, combined_msg, block_reason, _combined_findings, _scope_advisory_items diff --git a/ouroboros/tools/plan_render.py b/ouroboros/tools/plan_render.py index dab96fac4..cf144099d 100644 --- a/ouroboros/tools/plan_render.py +++ b/ouroboros/tools/plan_render.py @@ -10,6 +10,7 @@ from typing import Any, Dict, List, Optional from ouroboros.task_results import plan_review_notes_are_annotatable from ouroboros.tools.review_synthesis import PLAN_REVIEW_CONTROL_PREFIX from ouroboros.tools.plan_spec import MAX_FINDINGS_PER_SLOT +from ouroboros.tools.review_helpers import review_enforcement_blocks # B2 (honest DEGRADED): every aggregate reaches the control line as itself — the @@ -128,10 +129,20 @@ def _next_step(wave: dict, *, enforcement: str, cap: Optional[int], cycles_paid: f"Author finish recorded as {author.get('disposition')} against this exact " "review fingerprint; raw reviewer findings remain evidence. " ) - if enforcement == "blocking": + if review_enforcement_blocks(enforcement): author_note += "Blocking enforcement still holds the open plan gate. " + elif not review_enforcement_blocks("blocking"): + author_note += "Cyber Pro preserves final judgment with Ouroboros. " else: author_note += "Advisory enforcement permits proceeding with the review open. " + if not review_enforcement_blocks("blocking"): + return ( + author_note + "Cyber Pro: Ouroboros decides whether and how to continue. " + "The recorded verdict, open findings and any unresolved physical reviewers remain " + "independent facts; continuation does not close the wave or create a PASS. " + f"The existing $0 plan_task(review_disposition={{review_fingerprint: '{fp}', items: [...]}}) " + "can collect results or record a disposition without a new panel." + ) if bool(wave.get("closed")): if plan_review_notes_are_annotatable(wave): return ( diff --git a/ouroboros/tools/plan_review.py b/ouroboros/tools/plan_review.py index af793f39a..85aa4f6a5 100644 --- a/ouroboros/tools/plan_review.py +++ b/ouroboros/tools/plan_review.py @@ -96,6 +96,7 @@ from ouroboros.tools.plan_review_references import ( from ouroboros.tools.registry import ToolContext, ToolEntry from ouroboros.tools.review_helpers import review_wave_binding_fence, review_wave_budget_gate from ouroboros.review_records import build_author_disposition_from_mapping +from ouroboros.tools.review_helpers import review_enforcement_blocks from ouroboros.tools.review_synthesis import ( PLAN_REVIEW_CONTROL_PREFIX, ) @@ -556,7 +557,7 @@ async def _run_plan_review_async(ctx: ToolContext, request: _PlanRequest, *, col elif not resume_in_flight: # stale ⇒ identical envelope re-dispatches fresh stale, replay_snapshot = _plan_wave_replay_decision(slots_fn, existing) if not stale: - if enforcement == "advisory": + if not review_enforcement_blocks(enforcement): # Still-OPEN wave: re-invoke the emitter so a durable append that FAILED # at record time retries on replay (memo only on success ⇒ landed dedups). _emit_plan_review_advisory_open(ctx, state_root, task_id=task_id, @@ -730,7 +731,7 @@ async def _run_plan_review_async(ctx: ToolContext, request: _PlanRequest, *, col except (OSError, TimeoutError, ValueError) as exc: return _typed_refusal(ctx, "TOOL_ERROR", f"ERROR: PLAN_REVIEW_STATE_INVALID: {exc}") _emit_plan_review_reference(ctx, task_id, state_root=state_root) - if enforcement == "advisory" and not stored.get("closed"): + if not review_enforcement_blocks(enforcement) and not stored.get("closed"): # B2: loud at the moment — ONE typed owner-visible event per recorded open wave. _emit_plan_review_advisory_open(ctx, state_root, task_id=task_id, wave=stored, cycles_paid=paid_now, cap=cap) @@ -835,7 +836,7 @@ def _cycles_exhausted( f"⚠️ PLAN_REVIEW_CYCLES_EXHAUSTED: {cycles_paid} of {cap} paid plan-review cycles are spent " "for this task; no reviewer was called and no cycle was consumed. " ) - if enforcement == "blocking": + if review_enforcement_blocks(enforcement): head += ( "Blocking enforcement: the plan review stays OPEN, so implementation stays held — but " "finalization is RELEASED so the task can end honestly instead of waiting for a panel it " @@ -843,6 +844,8 @@ def _cycles_exhausted( "revised spec once the owner raises OUROBOROS_REVIEW_MAX_CYCLES, or finalizing now with " "outcome_tier=blocked_with_evidence. Do not start the work under an open blocking review." ) + elif not review_enforcement_blocks("blocking"): + head += "Cyber Pro permits proceeding by Ouroboros's judgment; the open review and spent cycles remain recorded facts." else: head += ( "Advisory enforcement: you may proceed with the review open; the host records and " @@ -907,7 +910,7 @@ def _apply_disposition(ctx: ToolContext, disposition: dict) -> str: text, state, wave = _collect.collect_wave_sync(ctx, state_root=root, task_id=task_id, wave=wave) except PlanReviewSourceUnavailable as exc: return _plan_unavailable(ctx, str(exc), "plan_review_exact_artifact_unavailable") - if not disposition.get("items"): # a pure $0 peek; items are applied even while slots run + if not disposition.get("items") and not disposition.get("author_disposition"): return text cycles_paid = int(state.get("cycles_paid") or 0) if wave.get("closed") and not plan_review_notes_are_annotatable(wave): @@ -934,9 +937,9 @@ def _apply_disposition(ctx: ToolContext, disposition: dict) -> str: if review_retry_cancelled(ctx): return _bad("ERROR: PLAN_REVIEW_DISPOSITION_INVALID: cancellation prevents author finish") - if not wave.get("paid"): + if not wave.get("paid") and review_enforcement_blocks("blocking"): return _bad("ERROR: PLAN_REVIEW_DISPOSITION_INVALID: author finish requires an actual first review dispatch") - if enforcement != "advisory": + if review_enforcement_blocks(enforcement): return _bad( "ERROR: PLAN_REVIEW_DISPOSITION_INVALID: author_disposition is advisory-only; " "the selected blocking enforcement remains authoritative" diff --git a/ouroboros/tools/review.py b/ouroboros/tools/review.py index dd667c0f6..8883912e0 100644 --- a/ouroboros/tools/review.py +++ b/ouroboros/tools/review.py @@ -55,6 +55,7 @@ from ouroboros.tools.review_helpers import ( format_name_status_for_preflight, format_review_history_entry as _format_review_entry, REVIEW_PROMPT_TOKEN_BUDGET, # noqa: F401 — patchable seam (see note above) + review_enforcement_blocks, single_line as _single_line, ) @@ -702,9 +703,12 @@ def _handle_review_block_or_warning( blocked_msg: str, advisory_prefix: str, ) -> Optional[str]: - """Either block immediately or downgrade to advisory warning.""" - if blocking_review: + """Apply action authority while preserving the independent review signal.""" + cyber = not review_enforcement_blocks("blocking") + if blocking_review and not cyber: return blocked_msg + if cyber: + advisory_prefix = "Cyber Pro: review does not prohibit action; original signal follows. " _record_advisory_override(ctx, blocked_msg) _append_review_warning(ctx, advisory_prefix + blocked_msg) ctx._review_iteration_count = 0 @@ -725,6 +729,8 @@ def _record_advisory_override(ctx: ToolContext, blocked_msg: str) -> None: append_jsonl(ctx.drive_logs() / "events.jsonl", { "ts": utc_now_iso(), "type": "review_advisory_override", + "review_enforcement": _cfg.get_review_enforcement(), + "decision_authority": "cyber_pro" if not review_enforcement_blocks("blocking") else "advisory", "block_reason": reason, "message_head": str(blocked_msg or "")[:600], "task_id": str(getattr(ctx, "task_id", "") or ""), @@ -955,7 +961,7 @@ def _prepare_unified_review(ctx: ToolContext, commit_message: str, ctx._triad_withheld_seat_records = [] # reset Q28-dropped seat records ctx._review_degraded_reasons = [] # reset degraded participation markers review_enforcement = _cfg.get_review_enforcement() - blocking_review = review_enforcement == "blocking" + blocking_review = review_enforcement_blocks(review_enforcement) diff_text, subject, capture_block = _capture_triad_staged_diff(ctx, target_repo, blocking_review) if diff_text is None: # capture failed: block (blocking) or advisory-skip (None) @@ -1207,7 +1213,7 @@ def _review_actor_label(row: dict) -> str: def _dispatch_unified_review(ctx: ToolContext, commit_message: str, prepared: dict) -> Optional[str]: """Dispatch an assembled triad packet and post-process the panel verdict.""" - blocking_review = prepared["blocking_review"] + blocking_review = prepared["blocking_review"] and review_enforcement_blocks("blocking") try: result_json = _handle_multi_model_review( ctx, @@ -1331,7 +1337,9 @@ def _dispatch_unified_review(ctx: ToolContext, commit_message: str, prepared: di _record_advisory_override(ctx, "; ".join(critical_fails[:5])) _append_review_warning( ctx, - "Review enforcement=Advisory: critical review findings did not block commit.", + ("Cyber Pro: critical review findings do not prohibit action." + if not review_enforcement_blocks("blocking") else + "Review enforcement=Advisory: critical review findings did not block commit."), ) for finding in getattr(ctx, "_last_review_critical_findings", []) or []: _append_review_warning(ctx, finding) diff --git a/ouroboros/tools/review_helpers.py b/ouroboros/tools/review_helpers.py index 4bc5ee72f..dc738f52d 100644 --- a/ouroboros/tools/review_helpers.py +++ b/ouroboros/tools/review_helpers.py @@ -36,6 +36,15 @@ REPO_ROOT = Path(__file__).resolve().parent.parent.parent # non-blocking skip gate leaves headroom for default 1M-context reviewer models. REVIEW_PROMPT_TOKEN_BUDGET = 920_000 + +def review_enforcement_blocks(enforcement: str | None = None) -> bool: + """Project action authority without changing configured policy or review facts.""" + from ouroboros.config import get_review_enforcement, get_runtime_mode + from ouroboros.runtime_mode_policy import runtime_mode_at_least + + selected = get_review_enforcement() if enforcement is None else enforcement + return selected == "blocking" and not runtime_mode_at_least(get_runtime_mode(), "cyber_pro") + # Tokenizer-density calibration shared by every review surface (triad, scope, plan, # deep self-review). estimate_tokens (chars/4) tracks GPT-style tokenizers, but a # real Claude scope pack estimated at 739,508 tokens measured 1,166,914 REAL tokens diff --git a/ouroboros/tools/scope_review.py b/ouroboros/tools/scope_review.py index 7a542acdf..e03a10e73 100644 --- a/ouroboros/tools/scope_review.py +++ b/ouroboros/tools/scope_review.py @@ -1,9 +1,8 @@ """Enforcement-aware Atlas-backed scope reviewer for the commit pipeline. Runs beside triad review and sees touched context plus a generated repo atlas. Critical findings follow -``OUROBOROS_REVIEW_ENFORCEMENT``: blocking enforcement blocks, advisory -enforcement reports them without blocking. Failed rows retain their original -status and typed origin. The commit aggregate applies advisory permission to +the selected enforcement outside Cyber; Cyber findings are advisory to action. +Failed rows retain their original status and typed origin. The commit aggregate applies permission to technical failures independently of candidate, custody and owner admission. In owner-selected ``low`` context mode no reviewer runs and a typed skip is recorded. """ @@ -48,6 +47,7 @@ from ouroboros.tools.review_helpers import ( build_touched_file_pack, # noqa: F401 -- facade import surface; leaves read it through the call-time handle load_checklist_section, # noqa: F401 -- facade import surface; leaves read it through the call-time handle review_drive_root, + review_enforcement_blocks, CRITICAL_FINDING_CALIBRATION, # noqa: F401 -- facade import surface; leaves read it through the call-time handle BINARY_EXTENSIONS, # noqa: F401 -- facade import surface; leaves read it through the call-time handle _SENSITIVE_EXTENSIONS, # noqa: F401 -- facade import surface; leaves read it through the call-time handle @@ -868,7 +868,7 @@ def run_scope_review( if critical_findings: from ouroboros import config as _cfg - if _cfg.get_review_enforcement() == "blocking": + if review_enforcement_blocks(_cfg.get_review_enforcement()): return ScopeReviewResult( blocked=True, block_message=_build_block_message(critical_findings, advisory_findings), diff --git a/ouroboros/tools/skill_exec.py b/ouroboros/tools/skill_exec.py index 2a2a4eee6..ab71fe047 100644 --- a/ouroboros/tools/skill_exec.py +++ b/ouroboros/tools/skill_exec.py @@ -575,19 +575,21 @@ def _author_finish_existing_skill_review( disposition: str, rationale: str, ) -> Optional[Dict[str, Any]]: - """Apply an explicit advisory author finish without buying a new panel. + """Record author finish in Advisory or Cyber without buying a new panel. - The first reviewer panel remains the source of findings. A later finish - call accepts the current payload after deterministic preflight, including - after a local fix. Reviewer hash, findings and status stay intact; - only the author record binds the newly accepted bytes. + Advisory needs prior feedback and a passing current preflight; Cyber may + continue with either missing or failed. Reviewer hash, findings and status + stay intact; only the author record binds the newly accepted bytes. """ from ouroboros.config import get_review_enforcement from ouroboros.review_records import build_author_disposition from ouroboros.skill_loader import compute_content_hash, load_review_state, save_review_state from ouroboros.skill_review import _run_deterministic_preflight + from ouroboros.tools.review_helpers import review_enforcement_blocks - if str(get_review_enforcement() or "").strip().lower() != "advisory": + enforcement = str(get_review_enforcement() or "").strip().lower() + cyber = not review_enforcement_blocks("blocking") + if review_enforcement_blocks(enforcement): return {"error": "SKILL_REVIEW_ERROR: explicit author finish requires advisory enforcement."} loaded = load_bound_skill(binding) if loaded is None: @@ -599,9 +601,9 @@ def _author_finish_existing_skill_review( ) drive_root = binding.state_drive_root review_state = load_review_state(drive_root, skill_name, skill_type=loaded.manifest.type, skill_dir=loaded.skill_dir) - if review_state.status == "pending": + if not cyber and review_state.status == "pending": return {"error": "SKILL_REVIEW_ERROR: existing review is pending or has no reviewer verdict."} - if not (review_state.findings or review_state.raw_actor_records or review_state.raw_result): + if not cyber and not (review_state.findings or review_state.raw_actor_records or review_state.raw_result): return {"error": "SKILL_REVIEW_ERROR: no prior reviewer evidence is available for author finish."} try: author_record = build_author_disposition( @@ -609,20 +611,29 @@ def _author_finish_existing_skill_review( rationale=rationale, subject_hash=current_hash, reviewer_signal=review_state.status, - enforcement="advisory", + enforcement=enforcement, ) except ValueError as exc: return {"error": f"SKILL_REVIEW_ERROR: {exc}"} previous_hash = str(review_state.reviewed_content_hash or review_state.content_hash or "") + preflight_facts = None if previous_hash != current_hash: - # A changed payload is accepted only after the existing deterministic - # gate checks the complete current payload. This is not a reviewer - # PASS: the prior findings remain attached as historical evidence. + # Current preflight is independent evidence; Cyber may continue with its + # failure, while ordinary Advisory still requires it to pass. preflight = _run_deterministic_preflight( ctx, drive_root, loaded, current_hash, persist=False, binding=binding, ) - if preflight is not None: + if preflight is not None and not cyber: return {"error": "SKILL_REVIEW_ERROR: deterministic preflight did not pass for the current payload."} + if preflight is not None: + from ouroboros.utils import append_jsonl, utc_now_iso + + preflight_facts = {"content_hash": current_hash, "status": preflight.status, + "findings": list(preflight.findings or []), "error": preflight.error} + append_jsonl(ctx.drive_logs() / "events.jsonl", { + "ts": utc_now_iso(), "type": "skill_review_author_preflight", + "skill_name": skill_name, "decision_authority": "cyber_pro", **preflight_facts, + }) review_state.author_disposition = author_record save_review_state(drive_root, skill_name, review_state) from ouroboros.skill_loader import auto_grant_if_enabled @@ -648,6 +659,7 @@ def _author_finish_existing_skill_review( "deps_status": deps_status, "deps_error": deps_error, "extension": extension, "review_stale": review_state.is_stale_for(current_hash), "review_gate": review_state.gate_for(current_hash), + **({"author_preflight": preflight_facts} if preflight_facts is not None else {}), } diff --git a/tests/test_review_cyber_authority.py b/tests/test_review_cyber_authority.py new file mode 100644 index 000000000..19109018c --- /dev/null +++ b/tests/test_review_cyber_authority.py @@ -0,0 +1,226 @@ +"""Cyber action authority is separate from reviewer and physical-operation facts.""" + +import copy +import json + +import pytest + +from ouroboros import config +from ouroboros.tools import git, plan_review +from ouroboros.tools.parallel_review import aggregate_review_verdict +from ouroboros.tools.review_helpers import build_scope_actor_record, review_enforcement_blocks +from ouroboros.tools.scope_review import ScopeReviewResult +from tests.test_advisory_inline_freshness import candidate # noqa: F401 +from tests.test_plan_review_engine import harness, _call, _state # noqa: F401 + + +@pytest.fixture(params=["pro", "cyber_pro"]) +def access(request, monkeypatch): + config.reset_runtime_mode_baseline_for_tests() + config.initialize_runtime_mode_baseline(request.param) + monkeypatch.setenv("OUROBOROS_REVIEW_ENFORCEMENT", "blocking") + yield request.param + config.reset_runtime_mode_baseline_for_tests() + + +def test_effective_authority_keeps_configured_enforcement(access): + assert config.get_review_enforcement() == "blocking" + assert review_enforcement_blocks() == (access == "pro") + assert not review_enforcement_blocks("advisory") + + +@pytest.mark.parametrize("status,phase", [ + ("responded", ""), ("error", "context"), ("error", "delivery"), + ("parse_failure", "format"), ("sub_floor", "window_authority"), + ("not_dispatched", "admission"), ("pending", "delivery"), +]) +def test_scope_review_facts_survive_action_authority(candidate, access, status, phase): # noqa: F811 + finding = {"item": "contract", "severity": "critical", "verdict": "FAIL", "reason": "Original criticism"} + result = ScopeReviewResult( + blocked=True, status=status, failure_phase=phase, raw_text="Exact original review", + block_message="Original failure", critical_findings=[finding], + operation_state="in_flight" if status == "pending" else "settled", + ) + candidate._last_scope_raw_results = [build_scope_actor_record(result)] + before = copy.deepcopy(result.__dict__) + blocked, _, _, findings, _ = aggregate_review_verdict( + None, result, "", [], candidate, "candidate", 0, candidate.repo_dir, + ) + assert blocked == (access == "pro") + assert result.__dict__ == before + assert findings[0]["verdict"] == "FAIL" + if access == "cyber_pro": + event = json.loads((candidate.drive_logs() / "events.jsonl").read_text().splitlines()[-1]) + assert event["review_enforcement"] == "blocking" + assert event["decision_authority"] == "cyber_pro" + + +def test_missing_preflight_does_not_become_a_review(candidate, access): # noqa: F811 + from ouroboros.review_state import load_state + + outcome = git._check_advisory_freshness(candidate, "candidate", paths=["change.py"]) + assert (outcome is None) == (access == "cyber_pro") + assert load_state(candidate.drive_root).advisory_runs == [] + + +def test_review_status_readiness_matches_actual_cyber_gate(candidate, access): # noqa: F811 + from ouroboros.tools.claude_advisory_review import _handle_review_status + from ouroboros.review_state import load_state + + projection = json.loads(_handle_review_status(candidate)) + assert projection["repo_commit_ready"] == (access == "cyber_pro") + assert not projection["advisory_runs"] + assert projection["latest_advisory_status"] != "fresh" + assert not load_state(candidate.drive_root).advisory_runs + + +def test_actual_staged_candidate_can_continue_after_failed_review(candidate, access, monkeypatch): # noqa: F811 + from ouroboros.tools import git_review_cycle + + result = ScopeReviewResult(blocked=True, status="error", failure_phase="context", block_message="Missing required source") + monkeypatch.setattr(git, "_advisory_and_tests_gate", lambda *a, **k: None) + monkeypatch.setattr(git, "_install_paid_dispatch_stamp", lambda *a, **k: None) + monkeypatch.setattr(git, "_reconcile_and_clear_review_roster", lambda *a, **k: None) + monkeypatch.setattr(git, "_run_parallel_review", lambda *a, **k: (None, result, "", [])) + git._reset_commit_review_state(candidate) + outcome = git_review_cycle._run_reviewed_stage_cycle( + candidate, "candidate", 0, paths=["change.py"], require_release_tag=False, + ) + assert outcome["status"] == ("passed" if access == "cyber_pro" else "blocked") + assert result.status == "error" and result.blocked + if access == "cyber_pro": + assert outcome["pre_fingerprint"]["fingerprint"] == outcome["post_fingerprint"]["fingerprint"] + assert "value = 2" in git.run_cmd(["git", "show", ":change.py"], cwd=candidate.repo_dir) + + +def test_pending_review_retains_custody_when_author_continues(candidate, access): # noqa: F811 + from ouroboros.review_state import load_state + + candidate._last_triad_raw_results = [{"slot_id": "s1", "status": "error", "operation_state": "in_flight", "operation_id": "op-1"}] + candidate._current_review_retry_key = "same-paid-work" + result = git._finalize_pending_review(candidate, "candidate", 0, + pre_fingerprint={"fingerprint": "fp"}, post_fingerprint={"fingerprint": "fp"}) + assert (result is None) == (access == "cyber_pro") + saved = load_state(candidate.drive_root).attempts[-1] + assert saved.status == "reviewing" and saved.late_result_pending + assert saved.triad_raw_results[0]["operation_id"] == "op-1" + if access == "cyber_pro": + git._record_commit_attempt(candidate, "candidate", "succeeded") + saved = load_state(candidate.drive_root).attempts[-1] + assert saved.late_result_pending + assert saved.triad_raw_results[0]["operation_state"] == "in_flight" + + +def test_pending_cyber_commit_uses_no_second_dispatch(candidate, access, monkeypatch): # noqa: F811 + from ouroboros.review_state import load_state + + candidate._current_review_retry_key = "old-review" + candidate._last_triad_raw_results = [{"slot_id": "critic", "operation_id": "original-op", "operation_state": "in_flight"}] + git._finalize_pending_review(candidate, "old candidate", 0, + pre_fingerprint={"fingerprint": "old"}, post_fingerprint={"fingerprint": "old"}) + before = load_state(candidate.drive_root).attempts[-1].triad_raw_results + git._reset_commit_review_state(candidate) + outcome = git._check_overlapping_review_attempt(candidate) + if access == "cyber_pro": + assert outcome is None + monkeypatch.setattr(git, "check_review_cycles_ceiling", lambda *a, **k: pytest.fail("new review admission")) + free = git._free_cycle_gate(candidate, "new candidate", 0, + pre_fingerprint={"fingerprint": "new"}, review_rebuttal="") + assert free["replay_reason"] == "review_pending" + assert load_state(candidate.drive_root).attempts[-1].triad_raw_results == before + monkeypatch.setattr(git, "_advisory_and_tests_gate", lambda *a, **k: None) + monkeypatch.setattr(git, "_run_parallel_review", lambda *a, **k: pytest.fail("duplicate paid panel")) + cycle = git._run_reviewed_stage_cycle(candidate, "new candidate", 0, + paths=["change.py"], require_release_tag=False) + assert cycle["status"] == "passed" + git._record_commit_attempt(candidate, "new candidate", "succeeded") + attempts = load_state(candidate.drive_root).attempts + assert attempts[-2].late_result_pending and attempts[-2].triad_raw_results == before + assert not attempts[-1].late_result_pending and not attempts[-1].triad_raw_results + else: + assert candidate._review_resume_pending or outcome is not None + + +@pytest.mark.parametrize("status,stale", [("blockers", False), ("pending", False), ("clean", True)]) +def test_skill_gate_does_not_relabel_the_verdict(access, status, stale): + from ouroboros.skill_review_status import skill_review_gate + + result = skill_review_gate(status, stale=stale, findings=[{"item": "skill_preflight", "verdict": "FAIL"}]) + assert result["executable_review"] == (access == "cyber_pro") + assert result["status"] == status and result["stale"] == stale + assert result["review_enforcement"] == "blocking" + assert result["preflight_failed"] == (not stale) + + +def test_skill_author_can_finish_without_fabricating_first_feedback(tmp_path, access, monkeypatch): + from ouroboros.skill_loader import load_review_state, save_enabled + from ouroboros.tool_access_types import ResolvedResourceBinding + from ouroboros.tools import skill_exec + from tests.test_skill_exec import _build_skill, _make_ctx + + ctx = _make_ctx(tmp_path) + directory = _build_skill(ctx.drive_root / "skills" / "external", "demo") + binding = ResolvedResourceBinding(profile="self_modification", root="skill_payload", operation="review", + base_path=directory, target_path=directory, source="test", skill_name="demo", state_drive_root=ctx.drive_root) + monkeypatch.setattr(skill_exec, "run_skill_review_lifecycle_blocking", lambda *a, **k: pytest.fail("Unexpected panel"), raising=False) + result = skill_exec._author_finish_existing_skill_review(ctx, binding, "demo", + disposition="accepted", rationale="Run this local greeter with the available evidence.") + if access == "pro": + assert "requires advisory" in result["error"] + return + assert "error" not in result, result + saved = load_review_state(ctx.drive_root, "demo") + assert saved.status == "pending" and not saved.raw_actor_records and not saved.raw_result + assert saved.author_disposition["reviewer_signal"] == "pending" + assert saved.author_disposition["enforcement"] == "blocking" + save_enabled(ctx.drive_root, "demo", True) + actual = json.loads(skill_exec._handle_skill_exec(ctx, skill="demo", script="hello.py")) + assert actual["exit_code"] == 0 and "hello from skill" in actual["stdout"] + + +def test_plan_author_finish_preserves_degraded_wave(harness, access, monkeypatch): # noqa: F811 + from ouroboros.tools.plan_review_artifacts import read_wave + + sub = harness.install({"s1": "", "s2": "", "s3": ""}) + ctx = harness.make_ctx() + _call(ctx) + before = _state(harness)["waves"][-1] + fingerprint = before["request_fingerprint"] + result = plan_review._apply_disposition(ctx, { + "review_fingerprint": fingerprint, "items": [], + "author_disposition": {"disposition": "deferred", "rationale": "Proceed with available evidence."}, + }) + after = _state(harness)["waves"][-1] + assert len(sub.calls) == 1 + assert after["aggregate"] == before["aggregate"] == "DEGRADED" + assert after["closed"] is False + if access == "cyber_pro": + exact = read_wave(harness.drive, ctx.task_id, after["wave_artifact"]) + assert exact["author_disposition"]["enforcement"] == "blocking" + assert "Cyber Pro" in result + else: + assert "DISPOSITION_INVALID" in result + + +@pytest.mark.parametrize("state", [None, {}, {"schema_version": 2, "current_attempt": {"fingerprint": "fp", "status": "open"}}]) +def test_plan_projection_preserves_unknown_evidence(access, state): + from ouroboros.task_results import plan_review_gate_projection + + before = copy.deepcopy(state) + result = plan_review_gate_projection(state, "blocking") + assert result["allow"] == (access == "cyber_pro") + assert result["enforcement"] == "blocking" + assert result["outcome"] == "" and not result["closed"] + assert state == before + if access == "cyber_pro": + assert result["decision_authority"] == "cyber_pro" + assert result["review_status"] in {"invalid", "absent", "open"} + + +def test_real_force_plan_decision_uses_cyber_authority(harness, access): # noqa: F811 + from ouroboros.owner_hurry import force_plan_decision + + ctx = harness.make_ctx(force_plan=True) + result = force_plan_decision(ctx, {}, enforcement="blocking") + assert result["allow"] == (access == "cyber_pro") + assert result["enforcement"] == "blocking" and not result["closed"]