Harden Living UI workflow engine and fix build-loop feedback bugs

.gitignore

@@ -56,4 +56,7 @@ agent_file_system/TASK_HISTORY.md
 !build_template.py
 docs/LIVING_UI_DEVELOPER_GUIDE.md
 agent_file_system/ACTIONS.md
-agent_bundle/
+agent_bundle/
+.playwright-mcp/
+.ruff_cache/
+.pytest_cache/

agent_core/core/impl/context/engine.py

@@ -495,10 +495,39 @@ def get_task_state(self, session_id: Optional[str] = None) -> str:
                 "Mode: Complex task - use todos in event stream to track progress",
             ]
 
-            skill_instructions = self.get_skill_instructions(session_id=session_id)
-            if skill_instructions:
-                lines.append("")
-                lines.append(skill_instructions)
+            # Sub-workflow tasks carry their own purpose-built system prompt;
+            # re-rendering full skill bodies here every turn is exactly the
+            # bloat they exist to avoid (skill files stay readable on disk).
+            include_skills = True
+            workflow = None
+            try:
+                from agent_core.core.registry.task_workflows import (
+                    resolve_task_workflow,
+                )
+
+                workflow = resolve_task_workflow(current_task)
+                if workflow is not None and not workflow.include_skills:
+                    include_skills = False
+            except Exception:
+                include_skills = True
+
+            if include_skills:
+                skill_instructions = self.get_skill_instructions(session_id=session_id)
+                if skill_instructions:
+                    lines.append("")
+                    lines.append(skill_instructions)
+
+            # Workflow-computed per-turn state (e.g. the Living UI build
+            # directive: code decides the next step, the LLM executes it).
+            # Fail-open: a workflow hook failure must never break a turn.
+            if workflow is not None:
+                try:
+                    provided = workflow.task_state(current_task)
+                    if provided:
+                        lines.append("")
+                        lines.append(str(provided))
+                except Exception as e:
+                    logger.debug(f"[CONTEXT_ENGINE] workflow.task_state failed: {e}")
 
             lines.append("</current_task>")
             return "\n".join(lines)

agent_core/core/impl/llm/interface.py

@@ -163,9 +163,18 @@ def __init__(
         self._log_to_db = log_to_db
         self._record_llm_call = record_llm_call
 
-        # Consecutive failure tracking to prevent infinite retry loops
+        # Consecutive failure tracking to prevent infinite retry loops.
+        # HALF-OPEN breaker: after the threshold, calls abort instantly only
+        # while the cooldown since the LAST failure is running; once it
+        # elapses, ONE probe call is allowed through — success resets the
+        # counter (provider recovered, e.g. credits topped up), failure
+        # re-latches for another cooldown. Without this the breaker
+        # hard-latched until app restart: no call could ever run, so the
+        # success-reset was unreachable.
         self._consecutive_failures = 0
         self._max_consecutive_failures = 5
+        self._last_failure_at = 0.0
+        self._breaker_cooldown_s = 60.0
 
         # Defer imports to avoid circular dependency
         from app.models.factory import ModelFactory
@@ -505,12 +514,17 @@ def _generate_response_sync(
 
         # Check if we've hit the consecutive failure threshold
         if self._consecutive_failures >= self._max_consecutive_failures:
-            logger.critical(
-                f"[LLM ABORT] Consecutive failure threshold reached "
-                f"({self._consecutive_failures}/{self._max_consecutive_failures}). "
-                f"Aborting to prevent infinite retries."
+            if time.time() - self._last_failure_at < self._breaker_cooldown_s:
+                logger.critical(
+                    f"[LLM ABORT] Consecutive failure threshold reached "
+                    f"({self._consecutive_failures}/{self._max_consecutive_failures}). "
+                    f"Aborting to prevent infinite retries."
+                )
+                raise LLMConsecutiveFailureError(self._consecutive_failures)
+            logger.warning(
+                f"[LLM BREAKER] Cooldown elapsed after "
+                f"{self._consecutive_failures} failures — allowing a probe call"
             )
-            raise LLMConsecutiveFailureError(self._consecutive_failures)
 
         if log_response:
             logger.info(f"[LLM SEND] system={system_prompt} | user={user_prompt}")
@@ -564,6 +578,7 @@ def _generate_response_sync(
                 logger.error(f"[LLM ERROR] {error_detail}")
                 # Track consecutive failure
                 self._consecutive_failures += 1
+                self._last_failure_at = time.time()
                 logger.warning(
                     f"[LLM CONSECUTIVE FAILURE] Count: {self._consecutive_failures}/{self._max_consecutive_failures}"
                 )
@@ -602,6 +617,7 @@ def _generate_response_sync(
         except Exception as e:
             # Track consecutive failure for any other exception
             self._consecutive_failures += 1
+            self._last_failure_at = time.time()
             logger.warning(
                 f"[LLM CONSECUTIVE FAILURE] Count: {self._consecutive_failures}/{self._max_consecutive_failures} | Error: {e}"
             )
@@ -916,6 +932,7 @@ def _finalize_session_response(
                 )
             logger.error(f"[LLM ERROR] {error_detail}")
             self._consecutive_failures += 1
+            self._last_failure_at = time.time()
             logger.warning(
                 f"[LLM CONSECUTIVE FAILURE] Count: "
                 f"{self._consecutive_failures}/{self._max_consecutive_failures}"
@@ -970,12 +987,17 @@ def _generate_response_with_session_sync(
         # session path previously had none, so a persistent provider error
         # (e.g. out-of-credits) retried forever instead of aborting.
         if self._consecutive_failures >= self._max_consecutive_failures:
-            logger.critical(
-                f"[LLM ABORT] Consecutive failure threshold reached "
-                f"({self._consecutive_failures}/{self._max_consecutive_failures}). "
-                f"Aborting to prevent infinite retries."
+            if time.time() - self._last_failure_at < self._breaker_cooldown_s:
+                logger.critical(
+                    f"[LLM ABORT] Consecutive failure threshold reached "
+                    f"({self._consecutive_failures}/{self._max_consecutive_failures}). "
+                    f"Aborting to prevent infinite retries."
+                )
+                raise LLMConsecutiveFailureError(self._consecutive_failures)
+            logger.warning(
+                f"[LLM BREAKER] Cooldown elapsed after "
+                f"{self._consecutive_failures} failures — allowing a probe call"
             )
-            raise LLMConsecutiveFailureError(self._consecutive_failures)
 
         if log_response:
             logger.info(

agent_core/core/impl/mcp/client.py

@@ -58,6 +58,11 @@ def __init__(self):
         self._servers: Dict[str, MCPServerConnection] = {}
         self._config: Optional[MCPConfig] = None
         self._event_loop: Optional[asyncio.AbstractEventLoop] = None
+        # Auto-revive bookkeeping: one lock per server so concurrent callers
+        # respawn it once, and a cooldown so a server that genuinely cannot
+        # start is not re-spawned on every single tool call.
+        self._revive_locks: Dict[str, asyncio.Lock] = {}
+        self._revive_failed_at: Dict[str, float] = {}
         self._initialized = True
 
     @property
@@ -234,6 +239,61 @@ def get_server_tools(self, server_name: str) -> List[MCPTool]:
             return []
         return self._servers[server_name].tools
 
+    # A stdio server whose child has exited is not a permanent condition — it
+    # is a dead subprocess we can spawn again. Without this, the first crash
+    # of an MCP server would kill its tools for the rest of the session:
+    # every later call would fail instantly with "connection lost".
+    _REVIVE_COOLDOWN_S = 30.0
+
+    async def _revive_server(
+        self, server_name: str, server: MCPServerConnection
+    ) -> bool:
+        """Respawn a server whose process has died. True if it is usable now.
+
+        Serialized per server (concurrent callers revive it once) and rate
+        limited, so a server that cannot start costs one attempt per cooldown
+        rather than one per tool call. Never raises: a failed revive is
+        reported to the caller as an unavailable tool.
+        """
+        lock = self._revive_locks.setdefault(server_name, asyncio.Lock())
+        async with lock:
+            if server.is_connected:
+                return True  # another caller already brought it back
+
+            import time
+
+            now = time.monotonic()
+            last_failure = self._revive_failed_at.get(server_name)
+            if (
+                last_failure is not None
+                and now - last_failure < self._REVIVE_COOLDOWN_S
+            ):
+                return False
+
+            logger.warning(
+                f"[MCP] Server '{server_name}' has died — restarting it "
+                f"(its tools were unavailable)"
+            )
+            try:
+                revived = await server.reconnect()
+            except Exception as e:
+                logger.error(f"[MCP] Restart of '{server_name}' raised: {e}")
+                revived = False
+
+            if revived:
+                self._revive_failed_at.pop(server_name, None)
+                logger.info(
+                    f"[MCP] Server '{server_name}' restarted with "
+                    f"{len(server.tools)} tools — its actions work again"
+                )
+            else:
+                self._revive_failed_at[server_name] = now
+                logger.error(
+                    f"[MCP] Server '{server_name}' could not be restarted; "
+                    f"retrying in at most {self._REVIVE_COOLDOWN_S:.0f}s"
+                )
+            return revived
+
     async def call_tool(
         self, server_name: str, tool_name: str, arguments: Dict[str, Any]
     ) -> Dict[str, Any]:
@@ -255,10 +315,16 @@ async def call_tool(
             }
 
         server = self._servers[server_name]
-        if not server.is_connected:
+        if not server.is_connected and not await self._revive_server(
+            server_name, server
+        ):
             return {
                 "status": "error",
-                "message": f"MCP server '{server_name}' connection lost",
+                "message": (
+                    f"MCP server '{server_name}' connection lost and could not "
+                    f"be restarted. Its tools are unavailable — use another way "
+                    f"to do this, or continue without it."
+                ),
             }
 
         result = await server.call_tool(tool_name, arguments)

agent_core/core/impl/task/manager.py

@@ -382,9 +382,14 @@ def create_task(
         # the LLM's expansion of the user message; for proactive / scheduled
         # tasks it's the trigger description. inject_memory_event no-ops if
         # nothing passes min_relevance, so noise is filtered automatically.
-        from agent_core.core.impl.memory.injector import inject_memory_event
+        # Sub-workflow tasks opt out: their focused prompt is the context.
+        from agent_core.core.registry.task_workflows import get_workflow
 
-        inject_memory_event(query=task_instruction, session_id=task_id)
+        task_workflow = get_workflow(workflow_id)
+        if task_workflow is None or task_workflow.inject_memory:
+            from agent_core.core.impl.memory.injector import inject_memory_event
+
+            inject_memory_event(query=task_instruction, session_id=task_id)
 
         self._set_agent_property("current_task_id", task_id)
 
@@ -400,12 +405,25 @@ def create_task(
         return task_id
 
     def _create_session_caches(self, task_id: str) -> None:
-        """Create session caches for a task."""
+        """Create session caches for a task. Sub-workflow tasks seed their
+        purpose-built system prompt instead of the general agent's."""
         try:
-            system_prompt, _ = self.context_engine.make_prompt(
-                user_flags={"query": False, "expected_output": False},
-                system_flags={},
-            )
+            task = self.tasks.get(task_id)
+            try:
+                from agent_core.core.registry.task_workflows import (
+                    resolve_task_workflow,
+                )
+
+                workflow = resolve_task_workflow(task)
+            except Exception:
+                workflow = None
+            if workflow is not None:
+                system_prompt = workflow.system_prompt
+            else:
+                system_prompt, _ = self.context_engine.make_prompt(
+                    user_flags={"query": False, "expected_output": False},
+                    system_flags={},
+                )
             for call_type in [
                 LLMCallType.REASONING,
                 LLMCallType.ACTION_SELECTION,

agent_core/core/prompts/__init__.py

@@ -105,9 +105,9 @@
     ACTION_SET_SELECTION_PROMPT,
 )
 
-# Sub-agent prompts now live alongside the sub-agent runtime, in
-# ``app.subagent.definitions`` (per-type system prompts) and
-# ``app.subagent.context_engine`` (shared output-format contract).
+# Sub-agent prompts live with their workflows — per-type system prompts
+# in ``app/workflows/<domain>/subagents/`` — and the shared output-format
+# contract in ``app.subagent.context_engine``.
 
 __all__ = [
     # Registry