fix(autonomy): preserve deferred work across runtime admission

Author: codex

core/agency/autonomous_task_engine.py

@@ -40,6 +40,14 @@ class StepStatus(Enum):
     ROLLED_BACK = "rolled_back"
 
 
+class TaskExecutionDeferred(RuntimeError):
+    """Execution admission was postponed without consuming a task attempt."""
+
+    def __init__(self, reason: str):
+        self.reason = str(reason or "execution_admission_deferred")[:240]
+        super().__init__(self.reason)
+
+
 def _persistable_context(context: Any) -> dict[str, Any]:
     """Plan context reduced to what can survive a restart.
 
@@ -468,6 +476,19 @@ def _repair_plan_argument_contracts(self, plan: TaskPlan) -> list[str]:
                 errors.append(f"{step.step_id or step.tool}:{step.tool}:{','.join(missing)}")
         return errors
 
+    @staticmethod
+    def _tool_result_deferral_reason(result: Any) -> str:
+        if not isinstance(result, dict):
+            return ""
+        if str(result.get("status", "") or "").strip().lower() != "deferred":
+            return ""
+        return str(
+            result.get("reason")
+            or result.get("message")
+            or result.get("error")
+            or "execution_admission_deferred"
+        )[:240]
+
     def _persist_active_plans(self) -> None:
         try:
             payload = {
@@ -1036,6 +1057,34 @@ async def execute_goal(
         # 2. Execute steps
         await self._execute_plan(plan, on_progress)
 
+        if plan.status == "deferred":
+            deferred_reason = str(
+                plan.context.get("execution_deferred_reason", "")
+                or "execution_admission_deferred"
+            )[:240]
+            await self._report_progress_event(
+                on_progress,
+                {
+                    "event": "execution_deferred",
+                    "plan_id": plan.plan_id,
+                    "status": "deferred",
+                    "reason": deferred_reason,
+                    "steps_completed": len(plan.succeeded_steps),
+                    "steps_total": len(plan.steps),
+                },
+            )
+            self._persist_plan_state(plan)
+            return TaskResult(
+                plan_id=plan.plan_id,
+                goal=goal,
+                succeeded=False,
+                summary="Execution remains queued until runtime admission clears.",
+                steps_completed=len(plan.succeeded_steps),
+                steps_total=len(plan.steps),
+                trace_id=trace_id,
+                deferred_reason=deferred_reason,
+            )
+
         # 3. Synthesize result
         result = await self._synthesize_result(plan, time.time() - plan.created_at)
         self._record_coding_execution(
@@ -2587,6 +2636,8 @@ async def _execute_plan(
                         completed_ids.add(step.step_id)
                     await self._emit_step_progress(step, on_progress, plan)
                 self._persist_plan_state(plan)
+                if plan.status == "deferred":
+                    return
                 failed_step = next(
                     (step for step in parallel_wave if step.status == StepStatus.FAILED), None
                 )
@@ -2612,6 +2663,9 @@ async def _execute_plan(
             await self._emit_step_progress(step, on_progress, plan)
             self._persist_plan_state(plan)
 
+            if plan.status == "deferred":
+                return
+
             if step.status == StepStatus.FAILED:
                 await self._fail_plan(
                     plan,
@@ -2735,6 +2789,10 @@ async def _execute_step_with_retry(self, step: TaskStep, plan: TaskPlan) -> None
                     timeout=step_timeout,
                 )
 
+                deferred_reason = self._tool_result_deferral_reason(raw_result)
+                if deferred_reason:
+                    raise TaskExecutionDeferred(deferred_reason)
+
                 step.raw_result = self._compact_tool_result(raw_result)
                 step.result_summary = step.raw_result
 
@@ -2846,6 +2904,24 @@ async def _execute_step_with_retry(self, step: TaskStep, plan: TaskPlan) -> None
                         )
                     self._persist_plan_state(plan)
 
+            except TaskExecutionDeferred as exc:
+                # Admission pressure and boot grace are scheduling facts, not
+                # failed actions. Preserve the exact step and its attempt
+                # budget so the plan can resume when the runtime is ready.
+                step.attempts = max(0, step.attempts - 1)
+                step.status = StepStatus.PENDING
+                step.error = f"execution deferred: {exc.reason}"
+                step.completed_at = None
+                plan.status = "deferred"
+                plan.context["execution_deferred_reason"] = exc.reason
+                plan.context["execution_deferred_at"] = time.time()
+                self._persist_plan_state(plan)
+                logger.info(
+                    "TaskEngine: step '%s' deferred without consuming an attempt: %s",
+                    step.description[:40],
+                    exc.reason,
+                )
+                return
             except TimeoutError:
                 step.error = f"timeout after {step_timeout}s"
                 self._record_coding_execution(
@@ -3006,6 +3082,7 @@ async def _verify_step(self, step: TaskStep, result: Any) -> bool:
                 f"Result: {result_str}\n\n"
                 "Answer with ONLY 'YES' or 'NO' followed by one sentence of evidence."
             )
+            verification_started_at = time.time()
             raw = await asyncio.wait_for(
                 llm.think(
                     prompt,
@@ -3017,6 +3094,12 @@ async def _verify_step(self, step: TaskStep, result: Any) -> bool:
                 timeout=15.0,
             )
             if not str(raw or "").strip():
+                deferral = take_deferral(
+                    origin="autonomous_task_engine",
+                    not_before=verification_started_at,
+                )
+                if deferral is not None:
+                    raise TaskExecutionDeferred(str(deferral.reason or "verifier_deferred"))
                 # Blank verifier output is NOT a pass: an empty verdict is a
                 # verifier outage, and treating "nothing came back" as success
                 # let unrelated or failure output satisfy arbitrary criteria.
@@ -3033,6 +3116,8 @@ async def _verify_step(self, step: TaskStep, result: Any) -> bool:
             # Cancellation is shutdown/caller intent, never a verification
             # outcome — propagate it rather than converting it into a pass.
             raise
+        except TaskExecutionDeferred:
+            raise
         except (RuntimeError, TimeoutError, AttributeError) as e:
             # Verifier exception or timeout: FAIL CLOSED. Assuming pass here
             # let every verifier outage certify arbitrary output as correct.
@@ -3106,7 +3191,13 @@ async def _get_alternative_approach(self, step: TaskStep) -> dict | None:
                     "Alternative approach returned identical args for '%s'", step.description[:40]
                 )
         except (json.JSONDecodeError, TypeError, ValueError) as e:
-            record_degradation("autonomous_task_engine", e)
+            record_degradation(
+                "autonomous_task_engine_retry",
+                e,
+                severity="warning",
+                action="failed the step after retry synthesis produced no executable alternative",
+                enforce_failure_policy=False,
+            )
             logger.debug("Alternative approach generation failed: %s", e)
 
         # Signal exhaustion rather than silently looping identically

core/runtime/overt_action_loop.py

@@ -207,15 +207,18 @@ async def run_once(self, *, force: bool = False) -> dict[str, Any]:
             return action.to_dict()
 
     def _background_reason(self) -> str:
+        allow_boot_anchor = os.getenv(
+            "AURA_OVERT_ACTION_ALLOW_BOOT_ANCHOR",
+            "1",
+        ).strip().lower() not in {"0", "false", "off", "no"}
         reason = background_activity_reason(
             self._orchestrator(),
             min_idle_seconds=float(os.getenv("AURA_OVERT_ACTION_IDLE_S", "30")),
             max_memory_percent=float(os.getenv("AURA_OVERT_ACTION_MAX_MEMORY_PERCENT", "88")),
             max_failure_pressure=float(os.getenv("AURA_OVERT_ACTION_MAX_FAILURE_PRESSURE", "0.35")),
             require_conversation_ready=False,
+            allow_no_user_anchor=allow_boot_anchor,
         )
-        if reason == "no_user_anchor" and os.getenv("AURA_OVERT_ACTION_ALLOW_BOOT_ANCHOR", "1").strip().lower() not in {"0", "false", "off", "no"}:
-            return ""
         return str(reason or "")
 
     def _record_skip(self, reason: str) -> OvertActionResult:

tests/test_autonomous_task_engine_runtime.py

@@ -7,6 +7,7 @@
 from core.agency.autonomous_task_engine import (
     AutonomousTaskEngine,
     StepStatus,
+    TaskExecutionDeferred,
     TaskPlan,
     TaskStep,
 )
@@ -672,6 +673,37 @@ async def test_task_engine_verifier_fails_closed_on_blank_llm_verdict():
     llm.think.assert_awaited()
 
 
+@pytest.mark.asyncio
+async def test_task_engine_verifier_propagates_inference_admission_deferral():
+    from core.brain.llm import deferral_record
+
+    async def deferred_think(*_args, **_kwargs):
+        deferral_record.record_deferral(
+            origin="autonomous_task_engine",
+            reason="foreground_quiet_window",
+        )
+        return ""
+
+    engine = AutonomousTaskEngine.__new__(AutonomousTaskEngine)
+    engine.kernel = SimpleNamespace(
+        organs={"llm": SimpleNamespace(get_instance=lambda: SimpleNamespace(think=deferred_think))}
+    )
+    step = TaskStep(
+        step_id="s1",
+        description="Verify grounded research",
+        tool="web_search",
+        args={"query": "consensus history"},
+        success_criterion="sources support the claim",
+    )
+
+    deferral_record.reset_for_test()
+    try:
+        with pytest.raises(TaskExecutionDeferred, match="foreground_quiet_window"):
+            await engine._verify_step(step, {"ok": True, "results": ["source"]})
+    finally:
+        deferral_record.reset_for_test()
+
+
 @pytest.mark.asyncio
 async def test_task_engine_records_execution_repair_pressure(monkeypatch):
     events: list[tuple[str, dict]] = []
@@ -740,6 +772,74 @@ async def test_task_engine_fails_fast_when_no_alternative_args_exist():
     assert step.attempts == 1
 
 
+@pytest.mark.asyncio
+async def test_task_engine_preserves_tool_deferral_without_spending_retry_budget():
+    engine = AutonomousTaskEngine.__new__(AutonomousTaskEngine)
+    engine._invoke_tool = AsyncCallRecorder(
+        return_value={"ok": False, "status": "deferred", "reason": "boot_grace_17s"}
+    )
+    engine._verify_step = AsyncCallRecorder(return_value=False)
+    engine._persist_plan_state = lambda plan: None
+    engine._record_coding_execution = lambda *_args, **_kwargs: None
+
+    step = TaskStep(
+        step_id="plan-deferred_s0",
+        description="Search for consensus history",
+        tool="web_search",
+        args={"query": "distributed systems consensus history"},
+        success_criterion="recent sources returned",
+    )
+    plan = TaskPlan(
+        plan_id="plan-deferred",
+        goal="Research consensus history",
+        steps=[step],
+        trace_id="trace",
+    )
+
+    await AutonomousTaskEngine._execute_step_with_retry(engine, step, plan)
+
+    assert plan.status == "deferred"
+    assert plan.context["execution_deferred_reason"] == "boot_grace_17s"
+    assert step.status == StepStatus.PENDING
+    assert step.attempts == 0
+    assert "boot_grace_17s" in step.error
+    engine._verify_step.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_task_engine_scheduler_retains_deferred_plan_for_resume():
+    engine = AutonomousTaskEngine.__new__(AutonomousTaskEngine)
+    engine._safety_registry = SimpleNamespace(is_allowed=AsyncCallRecorder(return_value=True))
+    engine._invoke_tool = AsyncCallRecorder(
+        return_value={"ok": False, "status": "deferred", "reason": "boot_grace_17s"}
+    )
+    engine._verify_step = AsyncCallRecorder(return_value=False)
+    engine._persist_plan_state = lambda plan: None
+    engine._record_coding_execution = lambda *_args, **_kwargs: None
+    engine._can_run_in_parallel = lambda _step: False
+
+    step = TaskStep(
+        step_id="plan-deferred_s0",
+        description="Search for consensus history",
+        tool="web_search",
+        args={"query": "distributed systems consensus history"},
+        success_criterion="recent sources returned",
+    )
+    plan = TaskPlan(
+        plan_id="plan-deferred",
+        goal="Research consensus history",
+        steps=[step],
+        trace_id="trace",
+    )
+
+    await AutonomousTaskEngine._execute_plan(engine, plan, on_progress=None)
+
+    assert plan.status == "deferred"
+    assert step.status == StepStatus.PENDING
+    assert step.attempts == 0
+    assert plan.any_failed is False
+
+
 @pytest.mark.asyncio
 async def test_task_engine_verify_step_uses_deterministic_result_checks_before_llm():
     llm = SimpleNamespace(think=AsyncCallRecorder(return_value="NO"))

tests/test_substrate_primary_architecture.py

@@ -374,6 +374,42 @@ def test_overt_action_selection_requires_explicit_web_intent_or_structured_skill
     assert incidental.reason == "retained_memory_is_evidence_not_an_action"
 
 
+def test_overt_action_boot_anchor_relaxes_only_user_anchor_policy(monkeypatch):
+    import core.runtime.overt_action_loop as overt_action_module
+
+    calls = []
+
+    def policy_reason(*args, **kwargs):
+        calls.append((args, kwargs))
+        return "boot_grace_17s"
+
+    monkeypatch.setenv("AURA_OVERT_ACTION_ALLOW_BOOT_ANCHOR", "1")
+    monkeypatch.setattr(overt_action_module, "background_activity_reason", policy_reason)
+
+    reason = overt_action_module.OvertActionLoop()._background_reason()
+
+    assert reason == "boot_grace_17s"
+    assert calls[0][1]["allow_no_user_anchor"] is True
+
+
+def test_overt_action_can_require_a_real_user_anchor(monkeypatch):
+    import core.runtime.overt_action_loop as overt_action_module
+
+    calls = []
+
+    def policy_reason(*args, **kwargs):
+        calls.append((args, kwargs))
+        return "no_user_anchor"
+
+    monkeypatch.setenv("AURA_OVERT_ACTION_ALLOW_BOOT_ANCHOR", "0")
+    monkeypatch.setattr(overt_action_module, "background_activity_reason", policy_reason)
+
+    reason = overt_action_module.OvertActionLoop()._background_reason()
+
+    assert reason == "no_user_anchor"
+    assert calls[0][1]["allow_no_user_anchor"] is False
+
+
 def test_overt_action_selection_semantically_plans_paraphrased_objectives(tmp_path):
     from core.runtime.overt_action_loop import OvertActionLoop
     from core.runtime.receipts import ReceiptStore