wip: preserve TZ-2 post-task and author-truth repairs

docs/DOMAIN_MAP.md

@@ -65,7 +65,7 @@ Rows may import columns (`[graph].allowed`). `·` = forbidden direction.
 
 ## Hidden coupling (classified out of the strict graph)
 
-- lazy-only cross-domain pairs: **104**
+- lazy-only cross-domain pairs: **105**
   - D01->D08
   - D01->D10
   - D01->D11
@@ -74,6 +74,7 @@ Rows may import columns (`[graph].allowed`). `·` = forbidden direction.
   - D03->D06
   - D03->D07
   - D03->D08
+  - D03->D09
   - D03->D10
   - D03->D11
   - D03->D14

docs/architecture/06-agent-core.md

@@ -537,7 +537,7 @@ Only roots synthesize; `root_phase_checkpoint` makes paid synthesis at-most-once
 
 Synthesis receives a sealed final package from the durable result — the submitted final text, its artifact manifest and completion_observations. Full redacted action observations live in the canonical artifact store (`task.budget_drive_root or drive_root`), in the write-once `source_handles/context_checkpoints` store with verified `task_source` refs, before compact publication and outside deliverables and inferred readiness; their native reader `get_task_result(include_completion_source=true)` returns complete length/hash first, then explicit `source_start_char`/`source_end_char` ranges (`artifacts.text_source_range_projection`, the shared work-order range contract), with bytes, kind, path containment and SHA checked before any excerpt. Packet-only reflection receives per-send-tool counts, each family's latest recorded return, and task-related skill readiness with coverage; full-source references are for later readers, not evidence the synthesizer has read. Positive observed facts correct error-trace impressions, while tool success does not prove owner receipt, empty material does not prove absence, and skill readiness does not attribute an owner's action to the task. Before context cleanup, `agent_task_pipeline.emit_task_results` also freezes `review_evidence.task_inputs` through `post_task_synthesis.capture_task_inputs`: `run_origin`, the existing task-local owner corpus, intact question/answer provenance and the canonical split-root verification-receipt union. Reflection receives the same complete redacted content through `reflection.task_inputs_prompt_section`, separate from bounded trace/review excerpts. A zero return code is positive evidence; an unrelated later pass cannot resolve another check's failure. Peer proposals stay attributed, and unavailable input is not evidence that approval or verification never existed. Recovery uses these stored observations and inputs, not a later conversation. A free `host_task_facts` row precedes paid stages: no model call or narrative; its metrics, routing and cost serve history.
 
-Pooled workers retain their slot until root post-task synthesis settles, for API-only and subscription tasks alike; early final-answer delivery keeps the response independent from that queue timing. Ordinary native post-work, including an inline Presence turn after its durable result is returned to the adapter, stays on its registered actor thread without a pooled worker slot: its `TaskModelWait` owner remains available through `POST_TASK_SYNTHESIS_INFLIGHT` after ordinary dialogue admission closes, detached server post-work binds its own live owner in the same registry, and the task mailbox stays available until the terminal post-task checkpoint. An open phase remains finalizing rather than appearing completed. Typed quota/auth waits resume only the unsettled call, stop or unknown outcomes degrade the phase without repeating finished stages, and restart recovery degrades an indeterminate `running` phase rather than replaying a possibly paid request.
+Pooled workers retain their slot until post-task work settles; early final-answer delivery is independent of that timing. Native work stays on its registered actor; `TaskModelWait` remains reachable through `POST_TASK_SYNTHESIS_INFLIGHT`, and detached work owns a separate live wait. Mailbox cleanup waits for the checkpoint. The solve-phase absolute ceiling does not cut a settled root's running post-work, but Stop, calendar deadline, monetary admission, per-call and idle rails still bind. A typed stop, budget refusal or unknown paid outcome skips later paid stages and degrades the checkpoint; ordinary stage failures are isolated and already-produced reflection actions still apply. A running checkpoint after restart degrades without replaying a paid request.
 
 #### Project registry and lease
 

docs/inventories/FACADE_INVENTORY.md

@@ -2,7 +2,7 @@
 
 AST-derived inventory of compatibility facades, regenerated by `python scripts/regenerate_inventories.py`. Do not edit. A facade row is any runtime module whose top-level `from <population module> import ...` statements carry the `noqa: F401` re-export marker — the codebase's declared "this binding exists for its binding, not for this module's own use" convention (reference FACADE_CONSUMERS method). Leaf domains come from `ouroboros/domains.toml`; a leaf outside the facade's domain is marked ✗ (that edge also appears in the manifest's pinned direction matrix). `tests/test_generated_inventories.py` pins byte-identity, so any re-export surface change must regenerate this file.
 
-- facade modules: **63**; marked re-export bindings: **2342**; cross-domain facade→leaf pairs: **133**
+- facade modules: **63**; marked re-export bindings: **2343**; cross-domain facade→leaf pairs: **133**
 
 | facade | domain | bindings | leaves |
 |---|---|---:|---|
@@ -10,7 +10,7 @@ AST-derived inventory of compatibility facades, regenerated by `python scripts/r
 | `ouroboros/agent.py` | D01 | 32 | `ouroboros/agent_dispatch.py` (15)<br>`ouroboros/agent_startup_checks.py` (4)<br>`ouroboros/config.py` (2 ✗D12)<br>`ouroboros/subagent_dispatch_notes.py` (4 ✗D07)<br>`ouroboros/subagents.py` (7 ✗D07) |
 | `ouroboros/agent_task_pipeline.py` | D01 | 27 | `ouroboros/dialogue_provenance.py` (2 ✗D15)<br>`ouroboros/post_task_synthesis.py` (10)<br>`ouroboros/synthesis_cost_text.py` (5)<br>`ouroboros/task_finalization.py` (10) |
 | `ouroboros/config.py` | D12 | 142 | `ouroboros/model_slots.py` (17)<br>`ouroboros/provider_models.py` (6 ✗D02)<br>`ouroboros/review_model_routes.py` (10)<br>`ouroboros/runtime_limits.py` (67)<br>`ouroboros/settings_defaults.py` (19)<br>`ouroboros/settings_integrity.py` (4)<br>`ouroboros/settings_scales.py` (17)<br>`ouroboros/update_channels.py` (2) |
-| `ouroboros/context.py` | D03 | 4 | `ouroboros/context_runtime_facts.py` (4) |
+| `ouroboros/context.py` | D03 | 5 | `ouroboros/context_runtime_facts.py` (5) |
 | `ouroboros/delegate_custody.py` | D07 | 10 | `ouroboros/delegate_custody_reconcile.py` (9)<br>`ouroboros/delegate_evidence.py` (1) |
 | `ouroboros/extension_loader.py` | D14 | 101 | `ouroboros/contracts/plugin_api.py` (7 ✗D19)<br>`ouroboros/extension_child_catalog.py` (8)<br>`ouroboros/extension_companion.py` (3)<br>`ouroboros/extension_import_staging.py` (6)<br>`ouroboros/extension_isolated_deps.py` (4)<br>`ouroboros/extension_liveness.py` (8)<br>`ouroboros/extension_plugin_api.py` (6)<br>`ouroboros/extension_registry_state.py` (20)<br>`ouroboros/extension_surface_names.py` (12)<br>`ouroboros/extension_ui_validation.py` (5)<br>`ouroboros/gateway/host_service.py` (1 ✗D11)<br>`ouroboros/provider_models.py` (1 ✗D02)<br>`ouroboros/skill_loader.py` (15)<br>`ouroboros/skill_token.py` (1)<br>`ouroboros/tools/skill_exec.py` (1)<br>`ouroboros/utils.py` (3 ✗D18) |
 | `ouroboros/gateway/_helpers.py` | D11 | 2 | `ouroboros/jsonl_tail.py` (2 ✗D18) |

ouroboros/agent_task_pipeline.py

@@ -170,6 +170,7 @@ def _run_post_task_processing_async(
     def _run_scoped() -> None:
         checkpoint_status = "degraded"
         skipped: list[str] = []
+        interrupted = ""
         try:
             # The free facts row precedes every paid stage, so neither Stop nor a
             # failed paid stage costs the card its facts; it is not a stage.
@@ -181,11 +182,7 @@ def _run_post_task_processing_async(
             task_memory = Memory(drive_root=env.drive_root, repo_dir=env.repo_dir)
 
             def _promotion() -> None:
-                from ouroboros.project_facts import resolve_project_id
-
                 reflection_entry = result.get("reflection_entry")
-                _pid = resolve_project_id(task_snapshot)
-                _apply_reflection_memory_actions(env, reflection_entry, project_id=_pid)
                 if is_presence_task(task_snapshot):
                     return
                 # Project facts stay scoped; generic process lessons remain global.
@@ -213,21 +210,50 @@ def _run_post_task_processing_async(
                     review_evidence_snapshot, sealed_final=sealed_snapshot))),
                 ("promotion", _promotion),
             ]
-            for index, (_name, run_stage) in enumerate(stages):
+            from ouroboros.usage_accounting import BudgetExceeded
+
+            stage_errors = False
+            for index, (name, run_stage) in enumerate(stages):
                 if _owner_stop_requested():
-                    # Stop-now: the remaining paid stages are skipped and NAMED
-                    # in the typed disclosure below; what already ran stays.
-                    skipped = [name for name, _run in stages[index:]]
+                    interrupted = "owner_stopped"
+                    skipped = [stage for stage, _run in stages[index:]]
                     break
-                run_stage()
-            if not skipped:
+                try:
+                    run_stage()
+                except Exception as error:
+                    if isinstance(error, BudgetExceeded):
+                        interrupted = "budget_exhausted"
+                    else:
+                        try:
+                            propagate_model_error(error)
+                        except Exception as control:
+                            interrupted = str(getattr(control, "control_reason", "") or
+                                              getattr(control, "code", "") or "provider_outcome_unknown")
+                    if interrupted:
+                        skipped = [stage for stage, _run in stages[index + 1:]]
+                        log.warning("Post-task paid stage %s interrupted for %s: %s",
+                                    name, stage_task_id, interrupted)
+                        break
+                    stage_errors = True
+                    log.warning("Post-task stage %s failed for %s", name, stage_task_id, exc_info=True)
+            if not interrupted and not stage_errors:
                 checkpoint_status = "completed"
         except Exception:
-            log.warning("Async post-task processing failed", exc_info=True)
+            log.warning("Post-task setup failed for %s", stage_task_id, exc_info=True)
         finally:
+            # Applying actions already produced by reflection is free and must
+            # survive a later paid-stage refusal; never run the paid promotion here.
+            if (result.get("reflection_entry") is not None
+                    and interrupted not in {"owner_stopped", "cancelled", "finalize_requested"}):
+                try:
+                    from ouroboros.project_facts import resolve_project_id
+                    _apply_reflection_memory_actions(
+                        env, result["reflection_entry"], project_id=resolve_project_id(task_snapshot))
+                except Exception:
+                    log.warning("Completed reflection actions could not be applied for %s", stage_task_id, exc_info=True)
             _set_root_post_task_checkpoint(
                 env, task_snapshot, checkpoint_status,
-                stop_reason=f"owner_stopped:skipped={','.join(skipped)}" if skipped else "",
+                stop_reason=(f"{interrupted}:skipped={','.join(skipped)}" if interrupted else ""),
             )
             if post_task_key is not None:
                 with _POST_TASK_SYNTHESIS_LOCK:
@@ -250,7 +276,17 @@ def _run_post_task_processing_async(
                 if post_task_key is not None and parent_wait is not None and not parent_wait.worker_slot_held:
                     with _POST_TASK_SYNTHESIS_LOCK:
                         _POST_TASK_SYNTHESIS_INFLIGHT[post_task_key] = parent_wait
-                _run_scoped()
+                # The task's optional absolute execution ceiling ends the solve
+                # phase, not already-started post-work. Calendar deadlines and
+                # logical call bounds remain checked before owner_control.
+                prior_control = parent_wait.owner_control if parent_wait is not None else None
+                if parent_wait is not None:
+                    parent_wait.owner_control = lambda: "owner_stopped" if _owner_stop_requested() else None
+                try:
+                    _run_scoped()
+                finally:
+                    if parent_wait is not None:
+                        parent_wait.owner_control = prior_control
             else:
                 # A detached thread must not inherit its parent's closing scope.
                 with task_model_wait_scope(task=task_snapshot, drive_root=env.drive_root,

ouroboros/context.py

@@ -313,6 +313,7 @@ from ouroboros.context_runtime_facts import (  # noqa: E402,F401 — re-exported
     _project_room_fact,
     _queue_context_fact,
     _runtime_budget_info,
+    task_execution_clock_fact,
 )
 
 
@@ -395,6 +396,7 @@ def build_runtime_section(env: Any, task: Dict[str, Any], *, ctx: Any = None, sc
             "child_drive_root": task.get("child_drive_root"),
             "budget_drive_root": task.get("budget_drive_root"),
             "deadline_at": task.get("deadline_at"),
+            **task_execution_clock_fact(task, ctx),
             "allowed_resources": task.get("allowed_resources"),
             "context": task.get("context"),
         },

ouroboros/context_runtime_facts.py

@@ -21,6 +21,39 @@ from ouroboros.config import runtime_setting
 log = logging.getLogger(__name__)
 
 
+def task_execution_clock_fact(task: Dict[str, Any], ctx: Any) -> Dict[str, Any]:
+    """Current finite execution-ceiling estimate, not a calendar deadline.
+
+    Quota/budget pauses can move the estimate after this context is assembled;
+    an unknown start or unlimited ceiling yields null instead of a false date.
+    """
+    import datetime
+    import math
+    import time
+    from ouroboros.config import get_task_abs_ceiling_sec
+    from ouroboros.deadline_utils import parse_deadline_ts
+    from ouroboros.model_wait import current_model_wait, execution_elapsed_seconds
+
+    raw = getattr(ctx, "task_started_at", None) or task.get("started_at")
+    try:
+        start = float(raw)
+    except (TypeError, ValueError):
+        parsed = parse_deadline_ts(raw)
+        start = parsed.timestamp() if parsed is not None else 0.0
+    ceiling = get_task_abs_ceiling_sec()
+    started = datetime.datetime.fromtimestamp(start, datetime.timezone.utc).isoformat() if start > 0 and math.isfinite(start) else None
+    projected = None
+    if started and ceiling is not None:
+        now = time.time()
+        owner = current_model_wait()
+        elapsed = (owner.executed_seconds() if owner is not None and owner.task_id == str(task.get("id") or "")
+                   else execution_elapsed_seconds({**task, "started_at": start}, now))
+        projected = datetime.datetime.fromtimestamp(now + max(0.0, ceiling - elapsed),
+                                                    datetime.timezone.utc).isoformat()
+    return {"started_at": started, "absolute_ceiling_at": projected,
+            "absolute_ceiling_at_basis": "current estimate; quota or budget pauses may move it" if projected else "not_set"}
+
+
 def _queue_context_fact(task: Dict[str, Any]) -> Dict[str, Any]:
     """One dated canonical-queue view, frozen with the task's ContextCore."""
     from ouroboros.config import DATA_DIR, get_max_active_subagents_per_root, get_max_workers

ouroboros/domains.toml

@@ -844,6 +844,7 @@ lazy_only = [
   "D03->D06",
   "D03->D07",
   "D03->D08",
+  "D03->D09",
   "D03->D10",
   "D03->D11",
   "D03->D14",

ouroboros/loop_acceptance_review.py

@@ -640,7 +640,7 @@ def _finish_advisory_author(ctx: _TaskAcceptanceContext) -> bool:
     disposition = str(stance.get("agent_disposition") or "")
     from ouroboros.loop_delivery import delivery_evidence_fingerprint
 
-    if (not intent or disposition not in {"accepted", "rejected", "partial", "deferred"}
+    if (not intent or (disposition and disposition not in {"accepted", "rejected", "partial", "deferred"})
             or (action != "stop" and (not feedback or intent.get("review_binding_hash") != feedback.get("binding_hash")))
             or intent.get("tool_count") != len(ctx.llm_trace.get("tool_calls") or [])
             or intent.get("owner_directives") != len(getattr(ctx.tools._ctx, "_owner_directives", []) or [])
@@ -653,8 +653,8 @@ def _finish_advisory_author(ctx: _TaskAcceptanceContext) -> bool:
         subject_hash=ctx.review_binding["binding_hash"],
         reviewer_signal=str((feedback or {}).get("aggregate_signal") or ""),
         enforcement="blocking" if review_enforcement_blocks(_loop().get_review_enforcement()) else "advisory",
+        action=action,
     )
-    author["action"] = action
     from ouroboros.task_results import project_task_acceptance_review_capacity
 
     capacity = project_task_acceptance_review_capacity(ctx.tools._ctx, task_id=ctx.task_id) if action == "stop" else {}

ouroboros/size_ratchet_manifest.py

@@ -153,6 +153,7 @@ BAND_PATHS = {
     "ouroboros/task_status.py": None,
     "ouroboros/tools/browser.py": None,
     "ouroboros/tools/commit_gate.py": "Grew INTO the band by the review-wave fix binding the actor reference (delivery class) into the commit review contract fingerprint \u2014 same-module contract identity, splitting it would separate the fingerprint from its gate.",
+    "ouroboros/tools/control_routing.py": "One owner for promote, route, and steer receipts; this change shares objective provenance instead of duplicating per-route logic.",
     "ouroboros/tools/control_task_results.py": "serial addressed turns: await_messages (the mailbox wait, its window bounds and catalog entry) lives beside wait_task/wait_tasks, whose transport-wait peek and cache-horizon note it shares; splitting the three waits would separate one reader from its consumers",
     "ouroboros/tools/core.py": "D05 ledger split (rows 311-349): read/list and owner-chat delivery spans moved to core_file_tools/core_artifacts; facade re-enters the band from above (2283 -> 1373) and shrinks further when the residual catalog split lands",
     "ouroboros/tools/plan_review.py": "Entered the band from 999 lines: the required-affected_paths form (owner 9=A) added the schema field and the PLAN_RESOURCE_FORM_REQUIRED refusal, which must name the task's open wave and the $0 disposition exit \u2014 it belongs beside the one preamble both the paid and dry-run paths share, not in the pure plan_spec companion that owns no task state.",

ouroboros/tools/control_routing.py

@@ -61,6 +61,17 @@ def _attach_origin_from_metadata(ctx: ToolContext, evt: Dict[str, Any]) -> None:
         evt["origin_suppressed"] = True
 
 
+def _attach_drafted_objective(ctx: ToolContext, evt: Dict[str, Any]) -> None:
+    """Keep the routed objective's author separate from the ingress owner corpus."""
+    evt["objective_author"] = {"kind": "task", "task_id": str(getattr(ctx, "task_id", "") or "")}
+    owner_rows = [dict(row) for row in (getattr(ctx, "_owner_directives", None) or [])
+                  if isinstance(row, dict) and row.get("source") in {
+                      "owner_mailbox", "owner_quiz_answer", "origin_message", "owner_corpus", "direct_incoming"}]
+    if evt.get("source_text") and not any(row.get("source") == "origin_message" for row in owner_rows):
+        owner_rows.insert(0, {"source": "origin_message", "content": evt["source_text"]})
+    evt["owner_corpus"] = owner_rows
+
+
 def _durable_project_of_request(ctx: ToolContext) -> str:
     """The project this request's work ALREADY has, durably: the promoting task's
     own binding — the one truth about a task's project, which a "Turn into
@@ -461,14 +472,7 @@ def _promote_chat_to_task(
 
     evt.update(consciousness_origin_metadata(metadata))
     _attach_origin_from_metadata(ctx, evt)
-    # A model-drafted objective has a different author from its owner source.
-    evt["objective_author"] = {"kind": "task", "task_id": str(getattr(ctx, "task_id", "") or "")}
-    owner_rows = [dict(row) for row in (getattr(ctx, "_owner_directives", None) or [])
-                  if isinstance(row, dict) and row.get("source") in {
-                      "owner_mailbox", "owner_quiz_answer", "origin_message", "owner_corpus", "direct_incoming"}]
-    if evt.get("source_text") and not any(row.get("source") == "origin_message" for row in owner_rows):
-        owner_rows.insert(0, {"source": "origin_message", "content": evt["source_text"]})
-    evt["owner_corpus"] = owner_rows
+    _attach_drafted_objective(ctx, evt)
     predecessor_error = _attach_predecessor_authority_from_metadata(
         ctx, evt, predecessor_task_id,
     )
@@ -758,6 +762,7 @@ def _route_to_project(
 
     evt.update(consciousness_origin_metadata(metadata))
     _attach_origin_from_metadata(ctx, evt)
+    _attach_drafted_objective(ctx, evt)
     evt.update(predecessor_event)
     _attach_client_surface(ctx, evt)
     # Owner 3=A holds on this verb too: a route starts a NEW root exactly like a

ouroboros/tools/review.py

@@ -259,7 +259,7 @@ def _handle_task_acceptance_review(
     agent_decision = {}
     if disposition or agent_rationale or normalized_ob or author_action:
         agent_decision = {
-            "disposition": disposition or "partial",
+            "disposition": disposition,
             "explicit_finish": bool(disposition or author_action),
             "author_action": author_action or "finish",
             "rationale": agent_rationale[:1000],

supervisor/queue_timeouts.py

@@ -257,6 +257,22 @@ def _enforce_task_timeouts_locked(
                        or model_waiting(meta) or waiting_on_owner
                        or _active_operation_progressing(meta, now))
         ceiling_reached = abs_ceiling is not None and runtime_sec >= float(abs_ceiling)
+        if (ceiling_reached and not task.get("parent_task_id")
+                and task_id == str(task.get("root_task_id") or task_id)):
+            # A settled answer may still own post-task memory work in this
+            # RUNNING worker. The solve ceiling cannot turn that work into a
+            # failed answer; idle, per-call, deadline and cancellation remain.
+            from ouroboros.task_results import load_task_result
+            from ouroboros.task_status import SETTLED_STATUSES
+
+            try:
+                stored = load_task_result(_queue().DRIVE_ROOT, str(task_id)) or {}
+            except Exception:
+                stored = {}  # unreadable terminal proof never widens the ceiling
+            checkpoint = stored.get("root_phase_checkpoint") or {}
+            if (stored.get("status") in SETTLED_STATUSES and isinstance(checkpoint, dict)
+                    and checkpoint.get("post_task_synthesis") == "running"):
+                ceiling_reached = False
 
         if (
             str(task_id) in owner_stop_held

tests/test_acceptance_preparation_incident.py

@@ -425,6 +425,20 @@ def test_the_root_nomination_never_runs_the_builder_and_records_the_stance(tmp_p
     assert len(payload["evidence_revision"]) == 64
 
 
+def test_action_only_nomination_does_not_invent_partial_stance(tmp_path, monkeypatch):
+    from ouroboros.tools.review import _handle_task_acceptance_review
+
+    monkeypatch.setenv("OUROBOROS_TASK_REVIEW_MODE", "auto")
+    payload = json.loads(_handle_task_acceptance_review(
+        _tool_ctx(tmp_path), claim="saved result", goal="deliver result",
+        rationale="Informed advisory finish with open critic notes", author_action="finish",
+    ))
+    assert payload["status"] == "deferred_to_host_acceptance"
+    assert payload["agent_decision"]["disposition"] == ""
+    assert payload["agent_decision"]["author_action"] == "finish"
+    assert payload["agent_decision"]["explicit_finish"] is True
+
+
 def test_the_child_path_still_builds_its_packet_and_a_broken_builder_still_raises(tmp_path, monkeypatch):
     import ouroboros.review_evidence as re_mod
     from ouroboros.tools.review import _handle_task_acceptance_review

tests/test_owner_stop_s3.py

@@ -430,6 +430,33 @@ def test_tool_call_spanning_expiry_keeps_the_episode_whole(tmp_path, monkeypatch
     assert ostop.sweep_owner_stop_hold(q_isolated, "t-span", updated, now=expiry + 121.0) is False
 
 
+def test_solve_ceiling_spares_settled_root_post_work_but_not_deadline_or_closed_phase(tmp_path, monkeypatch):
+    from supervisor import queue as q
+    from ouroboros.task_results import write_task_result
+
+    now = time.time()
+    task_id = "post-ceiling"
+    task = {"id": task_id, "root_task_id": task_id, "chat_id": 0, "type": "task"}
+    meta = {"task": task, "started_at": now - 100.0, "last_heartbeat_at": now,
+            "last_progress_at": now, "attempt": 1}
+    q_isolated, workers_mod, jobs = _expiry_enforcement_queue(
+        monkeypatch, tmp_path, task_id, meta)
+    monkeypatch.setattr(q, "get_task_abs_ceiling_sec", lambda: 50.0)
+    write_task_result(tmp_path, task_id, "completed",
+                      root_phase_checkpoint={"post_task_synthesis": "running"})
+    q_isolated._enforce_task_timeouts_locked(workers_mod, now, 0, {})
+    assert not meta.get("finalization_requested_at") and not jobs
+    monkeypatch.setattr(q, "_task_deadline_ts", lambda _task: now - 1)
+    q_isolated._enforce_task_timeouts_locked(workers_mod, now, 0, {})
+    assert meta.get("finalization_requested_at") and meta.get("finalization_reason") == "deadline"
+    meta.pop("finalization_requested_at")
+    monkeypatch.setattr(q, "_task_deadline_ts", lambda _task: 0)
+    write_task_result(tmp_path, task_id, "completed",
+                      root_phase_checkpoint={"post_task_synthesis": "completed"})
+    q_isolated._enforce_task_timeouts_locked(workers_mod, now, 0, {})
+    assert meta.get("finalization_reason") == "absolute_ceiling"
+
+
 def test_non_progressing_task_at_expiry_is_not_reaped_or_cloned(tmp_path, monkeypatch):
     """MAJOR-B trace (b): a NON-progressing task at episode expiry must not be
     idle_timeout-reaped by the generic rail nor cloned into a new-id retry that

tests/test_post_task_model_wait.py

@@ -31,6 +31,9 @@ def phase(tmp_path, monkeypatch):
     monkeypatch.setenv("TOTAL_BUDGET", "100")
     monkeypatch.setenv("OUROBOROS_DATA_DIR", str(root))
     monkeypatch.setenv("OUROBOROS_SETTINGS_PATH", str(root / "settings.json"))
+    # Foreground pytest may inherit the calling agent worker's environment;
+    # these tests model detached post-work except where a test opts into a pool worker.
+    monkeypatch.delenv("OUROBOROS_IN_WORKER", raising=False)
     monkeypatch.setattr(config, "CLAUDEXOR_MODEL_POLL_INTERVAL_SEC", 0.005)
     monkeypatch.setattr(config, "NETWORK_WAIT_BACKOFF_START_SEC", 0.005)
     monkeypatch.setattr(config, "NETWORK_WAIT_BACKOFF_MAX_SEC", 0.01)
@@ -165,6 +168,53 @@ def test_detached_decision_mailbox_and_activity_remain_live_after_task_done(phas
     assert len(forwarded) == 1  # An ended post owner cannot be resurrected.
 
 
+@pytest.mark.parametrize("cause", ["budget", "deadline", "unknown", "ordinary"])
+def test_paid_stage_interruption_closes_checkpoint_without_buying_following_stages(phase, monkeypatch, cause):
+    from ouroboros.usage_accounting import BudgetExceeded
+
+    f = phase
+    failures = {
+        "budget": BudgetExceeded("root wallet spent"),
+        "deadline": model_wait.ModelWaitInterrupted("deadline"),
+        "unknown": transport.ClaudexorModelError({"code": "model_outcome_unknown", "message": "unknown"}, unknown=True),
+        "ordinary": RuntimeError("one stage failed"),
+    }
+    def chat(*_args):
+        f.stages.append("chat")
+        raise failures[cause]
+    monkeypatch.setattr(pipeline, "_run_chat_consolidation", chat)
+    monkeypatch.setattr(pipeline, "_run_reflection", lambda *a, **k: f.stages.append("reflection") or None)
+    launch(f)
+    assert f.done.wait(5)
+    stored = load_task_result(f.root, f.task["id"])
+    checkpoint = stored["root_phase_checkpoint"]
+    assert checkpoint["post_task_synthesis"] == "degraded"
+    if cause == "ordinary":
+        assert f.stages == ["facts", "chat", "scratch", "reflection", "backlog"]
+    else:
+        assert f.stages == ["facts", "chat"]
+        assert checkpoint["post_task_stop_reason"].startswith({
+            "budget": "budget_exhausted", "deadline": "deadline", "unknown": "model_outcome_unknown"}[cause])
+        assert "scratchpad_consolidation,reflection,promotion" in checkpoint["post_task_stop_reason"]
+    assert not f.engine.creates  # no provider send after the first interrupted stage
+
+
+def test_completed_reflection_actions_survive_a_later_paid_stage_interruption(phase, monkeypatch):
+    f = phase
+    applied = []
+    monkeypatch.setattr(pipeline, "_run_reflection", lambda *a, **k: {"memory_actions": [{"type": "knowledge_write"}]})
+    monkeypatch.setattr(pipeline, "_apply_reflection_memory_actions", lambda *a, **k: applied.append(1))
+    def stop_promotion(*_args):
+        raise model_wait.ModelWaitInterrupted("deadline")
+    monkeypatch.setattr(pipeline, "_update_improvement_backlog", stop_promotion)
+    launch(f)
+    assert f.done.wait(5)
+    assert applied == [1]
+    checkpoint = load_task_result(f.root, f.task["id"])["root_phase_checkpoint"]
+    assert checkpoint["post_task_synthesis"] == "degraded"
+    assert checkpoint["post_task_stop_reason"].startswith("deadline:")
+
+
 @pytest.mark.parametrize("unknown", [False, True])
 def test_stop_or_unknown_never_marks_post_work_completed(phase, unknown):
     f = phase
@@ -181,6 +231,24 @@ def test_stop_or_unknown_never_marks_post_work_completed(phase, unknown):
     assert ledger(f.root)[-1]["state"] == ("unresolved" if unknown else "released")
 
 
+def test_blocking_post_work_exempts_solve_ceiling_only_within_its_scope(phase, monkeypatch):
+    f = phase
+    monkeypatch.setattr(config, "get_task_abs_ceiling_sec", lambda: 1)
+    monkeypatch.setattr(pipeline, "_run_reflection", lambda *a, **k: None)
+    observed = []
+    def check(*_args):
+        observed.append(model_wait.current_model_wait().control_reason())
+    monkeypatch.setattr(pipeline, "_run_chat_consolidation", check)
+    with model_wait.task_model_wait_scope(task=f.task, drive_root=f.root, event_queue=f.events,
+                                          worker_slot_held=True) as owner:
+        monkeypatch.setattr(owner, "executed_seconds", lambda **_kwargs: 100)
+        assert owner.control_reason() == "absolute_ceiling"
+        pipeline._run_post_task_processing_async(f.env, f.task, {}, {}, {}, f.root / "logs", blocking=True)
+        assert owner.control_reason() == "absolute_ceiling"  # restored for the solve owner
+    assert observed == [None]
+    assert load_task_result(f.root, f.task["id"])["root_phase_checkpoint"]["post_task_synthesis"] == "completed"
+
+
 def test_pooled_post_work_holds_return_but_delivers_answer_early_once(phase, monkeypatch):
     from ouroboros.utils import in_worker_process
 

tests/test_route_to_project.py

@@ -87,6 +87,7 @@ def test_route_to_existing_project_emits_event_and_receipt(tmp_path):
         "origin_message_ref": origin_ref,
         "origin_message_text": "continue the engine tuning",
     })
+    ctx.task_id = "drafter"
     out = _route_to_project(ctx, "racer", "paraphrased: keep tuning the engine", reason="follow-up", predecessor_task_id="")
     assert out.startswith("⚠️ ROUTE_UNCONFIRMED:")
     assert "do not retry automatically" in out.lower()
@@ -102,6 +103,8 @@ def test_route_to_existing_project_emits_event_and_receipt(tmp_path):
     assert evt["routing_token"]
     assert evt["source_ref"] == origin_ref
     assert evt["source_text"] == "continue the engine tuning"
+    assert evt["objective_author"] == {"kind": "task", "task_id": ctx.task_id}
+    assert evt["owner_corpus"] == [{"source": "origin_message", "content": "continue the engine tuning"}]
     assert ctx._typed_routing_action_emitted == "route_to_project"
 
 

tests/test_tz2_clock_projection.py

@@ -0,0 +1,27 @@
+"""TZ-2 B3: time facts are projections, not invented calendar limits."""
+
+import datetime
+import time
+
+from ouroboros import config, context, model_wait
+
+
+def test_started_and_optional_ceiling_are_explicitly_sourced(tmp_path, monkeypatch):
+    task = {"id": "clock", "started_at": time.time() - 10}
+    monkeypatch.setattr(config, "get_task_abs_ceiling_sec", lambda: None)
+    unlimited = context.task_execution_clock_fact(task, None)
+    assert unlimited["started_at"] and unlimited["absolute_ceiling_at"] is None
+    monkeypatch.setattr(config, "get_task_abs_ceiling_sec", lambda: 100)
+    estimated = context.task_execution_clock_fact(task, None)
+    remaining = (datetime.datetime.fromisoformat(estimated["absolute_ceiling_at"])
+                 - datetime.datetime.now(datetime.timezone.utc)).total_seconds()
+    assert 87 <= remaining <= 93
+    assert "may move" in estimated["absolute_ceiling_at_basis"]
+    assert context.task_execution_clock_fact({"id": "clock"}, None)["absolute_ceiling_at"] is None
+    with model_wait.task_model_wait_scope(task=task, drive_root=tmp_path, event_queue=None,
+                                          worker_slot_held=True) as owner:
+        monkeypatch.setattr(owner, "executed_seconds", lambda **_kwargs: 7)
+        live = context.task_execution_clock_fact(task, None)
+    live_remaining = (datetime.datetime.fromisoformat(live["absolute_ceiling_at"])
+                      - datetime.datetime.now(datetime.timezone.utc)).total_seconds()
+    assert 91 <= live_remaining <= 95

web/modules/review_presentation.js

@@ -829,10 +829,10 @@ export function formatReviewProjection(projection) {
 }
 
 function authorDispositionText(author, label = '') {
-    if (!author || typeof author !== 'object' || !text(author.disposition)) return '';
+    if (!author || typeof author !== 'object' || (!text(author.disposition) && !text(author.action))) return '';
     const actionLabel = label || `Author ${author.action === 'stop' ? 'stop' : 'finish'}`;
     return [
-        `${actionLabel}: ${text(author.disposition)}`,
+        `${actionLabel}${text(author.disposition) ? `: ${text(author.disposition)}` : ''}`,
         text(author.reviewer_signal) ? `reviewer signal=${text(author.reviewer_signal)}` : '',
         text(author.rationale),
         text(author.subject_hash) ? `subject_hash=${text(author.subject_hash)}` : '',

web/tests/review_action_only.test.js

@@ -0,0 +1,19 @@
+import test from 'node:test';
+import assert from 'node:assert/strict';
+import { taskAcceptanceGroupFromTaskDetail } from '../modules/review_presentation.js';
+
+test('action-only task finish remains visible without an invented stance', () => {
+    const group = taskAcceptanceGroupFromTaskDetail({
+        task_id: 'root',
+        review_projection: { panels: [{ panel_id: 'p1', surface: 'task_acceptance', aggregate_signal: 'FAIL' }] },
+        review_status: { acceptance_decision: {
+            status: 'finalized_unaccepted', reason: 'author_finish',
+            author_disposition: { action: 'finish', disposition: '', rationale: 'Result retained despite criticism.',
+                subject_hash: 'current', reviewer_signal: 'FAIL', source: 'author' },
+        } },
+    });
+    assert.match(group.authorDecisionText, /Author finish/);
+    assert.match(group.authorDecisionText, /Result retained despite criticism/);
+    assert.doesNotMatch(group.authorDecisionText, /partial|accepted|: FAIL/);
+    assert.equal(group.attempts[0].verdict, 'FAIL');
+});