feat(harness): continuous conversation execution across providers

python/main.py

@@ -1,4 +1,4 @@
-# 329:314 0:0 10:20
+# 329:315 0:0 10:20
 import os
 import time
 from contextlib import asynccontextmanager
@@ -350,9 +350,10 @@ async def lifespan(app: FastAPI):
         for _ddl in (
             "ALTER TABLE conversations ADD COLUMN IF NOT EXISTS max_tool_rounds INTEGER",
             "ALTER TABLE conversations ADD COLUMN IF NOT EXISTS inference_mode VARCHAR(20)",
+            "ALTER TABLE conversations ADD COLUMN IF NOT EXISTS harness_state JSONB",
         ):
             await _sess.execute(_sa_text(_ddl))
-    print("[conversations] enabled_tools/max_tool_rounds/inference_mode columns ensured")
+    print("[conversations] enabled_tools/max_tool_rounds/inference_mode/harness_state columns ensured")
     async with get_session() as _sess:
         await _sess.execute(_sa_text("""
             CREATE TABLE IF NOT EXISTS transcript_uploads (
@@ -675,4 +676,4 @@ async def ui_structure():
     @app.get("/{full_path:path}", include_in_schema=False)
     async def serve_spa(full_path: str):
         return FileResponse(os.path.join(STATIC_DIR, "index.html"))
-# 329:314 0:0 10:20
+# 329:315 0:0 10:20

python/models.py

@@ -1,4 +1,4 @@
-# 358:22 0:0 5:0
+# 361:22 0:0 5:0
 from sqlalchemy import (
     Column, Integer, String, Text, Boolean, Float, DateTime, JSON,
     ARRAY, ForeignKey, UniqueConstraint, Index, text
@@ -55,6 +55,9 @@ class Conversation(Base):
     archived = Column(Boolean, nullable=False, server_default=text("false"))
     agent_id = Column(Integer)
     enabled_tools = Column(JSONB)
+    max_tool_rounds = Column(Integer)
+    inference_mode = Column(String(20), server_default="agentic")
+    harness_state = Column(JSONB)
     created_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"))
     updated_at = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"))
 
@@ -447,4 +450,4 @@ class GeneratedImage(Base):
     featured = Column(Boolean, nullable=False, server_default=text("false"))
     tags = Column(JSONB, server_default=text("'[]'::jsonb"))
     skill_origin = Column(Text)
-# 358:22 0:0 5:0
+# 361:22 0:0 5:0

python/routes/chat.py

@@ -1,4 +1,4 @@
-# 685:192 2:7 2:16
+# 677:185 2:7 2:16
 import time
 import traceback
 from fastapi import APIRouter, HTTPException, Request
@@ -710,25 +710,14 @@ async def _grant_scope_if_valid(scope: str) -> tuple[bool, str]:
         prior_msgs = await storage.get_messages(conv_id)
         msg_ids = [m["id"] for m in prior_msgs if m["role"] in ("user", "assistant")]
         att_map = await storage.get_attachments_for_messages(msg_ids) if msg_ids else {}
-        history: list[dict] = []
-        for m in prior_msgs:
-            if m["role"] not in ("user", "assistant"):
-                continue
-            entry: dict = {"role": m["role"], "content": m["content"]}
-            atts = att_map.get(m["id"], [])
-            if atts:
-                entry["attachments"] = [
-                    {"storage_url": a.get("storage_url"), "mime_type": a.get("mime_type")}
-                    for a in atts
-                ]
-            history.append(entry)
-
-        # Cap context window: keep first message (task anchor) + last N-1 messages.
-        # Prevents runaway input costs on long write-heavy sessions (e.g. website
-        # construction) where assistant messages accumulate large file content.
-        _MAX_HISTORY = 40
-        if len(history) > _MAX_HISTORY:
-            history = history[:1] + history[-(_MAX_HISTORY - 1):]
+        from ..services.work_harness import prepare_continuity
+        history, harness_state = prepare_continuity(
+            prior_msgs,
+            att_map,
+            conv.get("harness_state"),
+        )
+        if harness_state is not None:
+            await storage.update_conversation_harness_state(conv_id, harness_state)
 
         from ..services.tool_executor import (
             set_approval_scope_user_id,
@@ -802,28 +791,19 @@ async def _grant_scope_if_valid(scope: str) -> tuple[bool, str]:
                 # execution would use agent_inst.model_id. agent_inst's
                 # role is purely persona/metadata loading (already folded
                 # into system_prompt via build_system_prompt above).
-                inst = AgentInstance.from_model(
+                from ..services.work_harness import run_single_turn
+                _turn_tools = _inf_mode != "direct"
+                if agent_inst is not None and not agent_inst.use_tools:
+                    _turn_tools = False
+                content, usage, provider_id = await run_single_turn(
                     model_id=model_id,
+                    messages=history,
                     user_id=uid or None,
-                    enforce_tier=False,
-                    enforce_enabled=False,
-                )
-                if _inf_mode == "direct":
-                    inst.use_tools = False
-                elif agent_inst is not None and not agent_inst.use_tools:
-                    # Forge agent has no tools configured — propagate that
-                    # to the executor so the provider never sees tool schemas.
-                    inst.use_tools = False
-                content, usage = await inst.run(
-                    history,
-                    system_prompt_override=system_prompt or None,
+                    system_prompt=system_prompt or None,
                     pin_requested_provider=provider_pin_requested,
-                    enforce_routed_tier=True,
+                    use_tools=_turn_tools,
                 )
-                print(f"[chat-dbg] provider={inst.provider_id!r} hist_len={len(history)} content_len={len(content or '')} content_preview={repr((content or '')[:80])}")
-                # Use the resolved provider_id from the instance — for forge
-                # agents whose model_id is "gpt-5-mini" this is "openai".
-                provider_id = inst.provider_id or provider_id
+                print(f"[chat-dbg] provider={provider_id!r} hist_len={len(history)} content_len={len(content or '')} content_preview={repr((content or '')[:80])}")
                 usage = dict(usage or {})
                 usage.setdefault("orchestration_mode", "single")
                 usage.setdefault("providers", [provider_id])
@@ -865,6 +845,11 @@ async def _grant_scope_if_valid(scope: str) -> tuple[bool, str]:
             current_max_tool_rounds.reset(_t_mr)
             current_client_run_id.reset(_t_cri)
 
+        if harness_state is not None:
+            from ..services.work_harness import continuity_state_summary
+            usage = dict(usage or {})
+            usage["harness_continuity"] = continuity_state_summary(harness_state)
+
         if usage.get("approval_state") == "pending":
             _store_pending_gate(conv_id, approval_gate_service.pending_gate_entry(
                 usage, history=history, system_prompt=system_prompt or None,
@@ -962,4 +947,4 @@ async def _grant_scope_if_valid(scope: str) -> tuple[bool, str]:
 #   then: the route removes its staged message and returns HTTP 403 instead of leaving a dangling turn or returning 500
 #   class: security
 # === END CONTRACTS ===
-# 685:192 2:7 2:16
+# 677:185 2:7 2:16

python/services/gemini_native.py

@@ -1,4 +1,4 @@
-# 190:66 0:0 1:1
+# 191:66 0:0 1:1
 """Native google-genai SDK adapter for Gemini 2.5 and Gemini 3.
 
 Replaces the OpenAI-compat HTTP path for Gemini providers. Unlocks:
@@ -38,6 +38,8 @@
 import os
 from typing import Any, Optional
 
+from .provider_failure import mark_provider_failure
+
 from google import genai
 from google.genai import types as gtypes
 
@@ -242,7 +244,7 @@ async def call_gemini_native(
                 config=config,
             )
         except Exception as exc:
-            return f"[gemini error: {type(exc).__name__}]", accumulated
+            return mark_provider_failure(f"[gemini error: {type(exc).__name__}]", "gemini", exc), accumulated
 
         _accumulate_usage(getattr(resp, "usage_metadata", None), accumulated)
 
@@ -293,4 +295,4 @@ async def call_gemini_native(
 
     # Loop exit safeguard (shouldn't reach here).
     return "[gemini: tool loop exhausted]", accumulated
-# 190:66 0:0 1:1
+# 191:66 0:0 1:1

python/services/inference.py

@@ -1,4 +1,4 @@
-# 399:164 0:0 16:15
+# 400:164 0:0 16:15
 # === MODULE_BUILD ===
 # id: a0_service_inference
 #   module_name: inference
@@ -79,6 +79,7 @@
 # Single source of truth for provider specs — loaded from python/config/providers.json.
 # Replaces the old hardcoded PROVIDER_ENDPOINTS dict per the no-string-literals doctrine.
 from .energy_registry import BUILTIN_PROVIDERS
+from .provider_failure import mark_provider_failure
 
 _log = logging.getLogger("a0p.inference")
 
@@ -240,7 +241,7 @@ def _sanitize_provider_error(provider: str, exc: BaseException) -> str:
         if isinstance(exc, _genai_errors.APIError):
             code = getattr(exc, "code", None) or getattr(exc, "status", None) or "?"
             msg = _safe_error_snippet(getattr(exc, "message", None) or str(exc))
-            return f"[{provider} error: {code} {msg}]".rstrip(" ]") + "]"
+            return mark_provider_failure(f"[{provider} error: {code} {msg}]".rstrip(" ]") + "]", provider, exc)
     except ImportError:
         pass
 
@@ -265,17 +266,17 @@ def _sanitize_provider_error(provider: str, exc: BaseException) -> str:
             body_msg = ""
         body_msg = _safe_error_snippet(body_msg)
         if body_msg:
-            return f"[{provider} error: HTTP {code} {body_msg}]"
-        return f"[{provider} error: HTTP {code}]"
+            return mark_provider_failure(f"[{provider} error: HTTP {code} {body_msg}]", provider, exc)
+        return mark_provider_failure(f"[{provider} error: HTTP {code}]", provider, exc)
     if isinstance(exc, httpx.TimeoutException):
-        return f"[{provider} error: request timed out]"
+        return mark_provider_failure(f"[{provider} error: request timed out]", provider, exc)
     if isinstance(exc, httpx.HTTPError):
-        return f"[{provider} error: network error]"
+        return mark_provider_failure(f"[{provider} error: network error]", provider, exc)
     # Generic — include the type plus a sanitized message snippet if non-empty.
     snippet = _safe_error_snippet(str(exc))
     if snippet:
-        return f"[{provider} error: {type(exc).__name__}: {snippet}]"
-    return f"[{provider} error: {type(exc).__name__}]"
+        return mark_provider_failure(f"[{provider} error: {type(exc).__name__}: {snippet}]", provider, exc)
+    return mark_provider_failure(f"[{provider} error: {type(exc).__name__}]", provider, exc)
 
 
 def _canonical_tool_calls(tool_calls: list[dict]) -> str:
@@ -625,6 +626,4 @@ async def _call_anthropic(
         max_tokens=max_tokens, use_tools=use_tools,
         reasoning_effort=reasoning_effort,
         enable_caching=enable_caching)
-
-
-# 399:164 0:0 16:15
+# 400:164 0:0 16:15

python/services/provider_failure.py

@@ -0,0 +1,63 @@
+# 21:28 0:0 0:0
+"""Typed, string-compatible provider failure markers.
+
+Provider adapters historically return sanitized failure text so existing UI and
+direct callers can display a safe error without exposing upstream secrets.
+ProviderFailureText preserves that contract while giving the A0 harness an
+in-process type boundary that cannot be forged by ordinary model text.
+"""
+from __future__ import annotations
+
+# === MODULE_BUILD ===
+# id: a0_service_provider_failure
+#   module_name: provider_failure
+#   module_kind: service
+#   summary: Carries sanitized provider failures as string-compatible typed markers so the harness can route failures without parsing assistant text.
+#   owner: Erin Spencer
+#   public_surface: ProviderFailureText, ProviderCallFailure, mark_provider_failure
+#   internal_surface: none
+#   auth_boundary: none
+#   storage_boundary: none
+#   network_boundary: none
+#   user_data_boundary: none
+#   admin_only: false
+#   tests: tests/test_work_harness_v0.0.0alpha.py
+#   rollout: default_enabled
+#   rollback: Revert marker wrapping; provider adapters return ordinary sanitized strings.
+#   requires: none
+#   since: 2026-09-23
+#   unresolved: none
+# === END MODULE_BUILD ===
+
+from typing import Any
+
+
+class ProviderFailureText(str):
+    """A safe display string carrying only non-secret routing metadata."""
+
+    provider: str
+    error_type: str
+
+    def __new__(cls, text: str, provider: str, error_type: str):
+        obj = str.__new__(cls, text)
+        obj.provider = provider
+        obj.error_type = error_type
+        return obj
+
+
+class ProviderCallFailure(RuntimeError):
+    """Harness-internal exception built from a sanitized provider marker."""
+
+    def __init__(self, content: ProviderFailureText, usage: dict[str, Any] | None = None):
+        super().__init__(f"{content.provider} provider call failed ({content.error_type})")
+        self.provider = content.provider
+        self.error_type = content.error_type
+        self.safe_text = str(content)
+        self.usage = dict(usage or {})
+
+
+def mark_provider_failure(
+    text: str, provider: str, exc: BaseException
+) -> ProviderFailureText:
+    return ProviderFailureText(text, provider, type(exc).__name__)
+# 21:28 0:0 0:0

python/services/providers/open_comp_prov_v0.0.0alpha.py

@@ -1,4 +1,4 @@
-# 364:67 0:0 2:5
+# 365:67 0:0 2:5
 """Generic OpenAI-compatible provider transport.
 
 Provider identity, endpoint, credential name, model, API family, reasoning
@@ -74,6 +74,7 @@
 from openai import AsyncOpenAI
 
 from ._resolver import resolve_model_for_role
+from ..provider_failure import mark_provider_failure
 
 
 def _accumulate_usage(total: dict, usage: dict | None) -> None:
@@ -90,7 +91,7 @@ def _safe_provider_error(provider_name: str, exc: BaseException, api_key: str) -
     from ..inference import _safe_error_snippet
 
     raw = str(exc).replace(api_key, "[redacted]") if api_key else str(exc)
-    return f"[{provider_name} error: {type(exc).__name__}: {safe}]" if (safe := _safe_error_snippet(raw)) else f"[{provider_name} error: {type(exc).__name__}]"
+    return mark_provider_failure(f"[{provider_name} error: {type(exc).__name__}: {safe}]" if (safe := _safe_error_snippet(raw)) else f"[{provider_name} error: {type(exc).__name__}]", provider_name, exc)
 
 
 def _normalize_reasoning_effort(spec: dict, effort: Optional[str]) -> Optional[str]:
@@ -477,4 +478,4 @@ async def call(
         )
     finally:
         reset_caller_provider(caller_provider_token)
-# 364:67 0:0 2:5
+# 365:67 0:0 2:5

python/services/run_context.py

@@ -1,4 +1,4 @@
-# 74:33 0:0 9:0
+# 78:33 0:0 9:0
 # N:M
 """Run-scoped ContextVars for ZFAE recursion tracking.
 
@@ -60,6 +60,9 @@
 current_max_tool_rounds: contextvars.ContextVar[Optional[int]] = contextvars.ContextVar(
     "a0p_max_tool_rounds", default=None,
 )
+current_tool_executions: contextvars.ContextVar[int] = contextvars.ContextVar(
+    "a0p_tool_executions", default=0,
+)
 current_approval_gate_scopes: contextvars.ContextVar[frozenset[str]] = contextvars.ContextVar(
     "approval_gate_scopes", default=frozenset(),
 )
@@ -128,7 +131,8 @@ def snapshot() -> dict:
         "orchestration_mode": current_orchestration_mode.get(),
         "user_tier": current_user_tier.get(),
         "max_tool_rounds": current_max_tool_rounds.get(),
+        "tool_executions": current_tool_executions.get(),
         "approval_gate_scopes": sorted(current_approval_gate_scopes.get()),
     }
 # N:M
-# 74:33 0:0 9:0
+# 78:33 0:0 9:0

python/services/tool_executor.py

@@ -1,4 +1,4 @@
-# 397:101 0:0 12:4
+# 398:101 0:0 12:4
 """ZFAE Tool Executor — thin shim over the per-tool registry.
 
 Tools live in `python/services/tools/*.py` (one file per tool, self-declared
@@ -56,7 +56,7 @@
 from .tools import dispatch as _registry_dispatch
 from .tools import registry as _registry
 from .tools import tool_schemas_chat as _registry_schemas
-from .run_context import current_user_tier
+from .run_context import current_user_tier, current_tool_executions
 
 # ---------------------------------------------------------------------------
 # Per-conversation tool allow-list ContextVar.
@@ -537,6 +537,7 @@ async def _execute_tool_inner(name: str, arguments: dict) -> str:
         if not _tool_tier_allows(name):
             return f"[tool denied — {name} is not available for this account tier]"
         await _approval_denial(name)
+        current_tool_executions.set(current_tool_executions.get() + 1)
         try:
             return await _registry_dispatch(name, **args)
         except KeyError:
@@ -568,4 +569,4 @@ async def _execute_tool_inner(name: str, arguments: dict) -> str:
     "get_active_chat_schemas",
     "get_active_responses_schemas",
 ]
-# 397:101 0:0 12:4
+# 398:101 0:0 12:4