From 7d75ab7a31a7b488231b5cb060c39b5f4908ac16 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:22:02 -0700
Subject: [PATCH 01/29] feat(harness): add continuous work harness

---
 python/services/work_harness.py | 342 ++++++++++++++++++++++++++++++++
 1 file changed, 342 insertions(+)
 create mode 100644 python/services/work_harness.py

diff --git a/python/services/work_harness.py b/python/services/work_harness.py
new file mode 100644
index 00000000..94a66286
--- /dev/null
+++ b/python/services/work_harness.py
@@ -0,0 +1,342 @@
+# 218:82 0:0 1:3
+"""Continuous A0 work harness.
+
+The conversation is the durable work identity. Model/provider selection is an
+internal execution detail unless the caller explicitly pins a model.
+
+Usage guidance:
+    history, state = prepare_continuity(messages, attachments, previous_state)
+    content, usage, provider_id = await run_single_turn(
+        model_id="deepseek",
+        messages=history,
+        user_id=user_id,
+        system_prompt=system_prompt,
+        pin_requested_provider=False,
+        use_tools=True,
+    )
+
+Persist `state` on the owning conversation. Pass it back on the next turn.
+The bounded checkpoint is continuity aid, not transcript authority; the full
+persisted message transcript remains authoritative and replayable.
+"""
+from __future__ import annotations
+
+# === MODULE_BUILD ===
+# id: a0_service_work_harness
+#   module_name: work_harness
+#   module_kind: service
+#   summary: Keeps one durable conversation identity across bounded context and internally routed provider execution.
+#   owner: Erin Spencer
+#   public_surface: prepare_continuity, run_single_turn, continuity_state_summary
+#   internal_surface: _checkpoint_line, _fallback_candidates, _safe_to_replay
+#   auth_boundary: none
+#   storage_boundary: read
+#   network_boundary: external
+#   user_data_boundary: read
+#   admin_only: false
+#   tests: tests/test_work_harness_v0.0.0alpha.py
+#   rollout: default_enabled
+#   rollback: Revert the chat integration and remove this service; persisted harness_state can remain inert.
+#   requires: a0_service_agent_instance, a0_service_model_catalog
+#   since: 2026-09-23
+#   unresolved: provider quota APIs are not uniformly queryable before a call; unsafe post-tool replay remains hmmm
+# === END MODULE_BUILD ===
+
+# === BOUNDARIES ===
+# id: work_harness_runtime_boundary
+#   summary: Reads user conversation text and may call configured external model providers; it performs no persistence itself.
+#   auth_boundary: none
+#   storage_boundary: read
+#   network_boundary: external
+#   user_data_boundary: read
+#   admin_only: false
+#   pii: possible
+#   secrets: none
+#   owner: platform-runtime
+# === END BOUNDARIES ===
+
+# === CONTRACTS ===
+# id: harness_continuity_preserves_identity
+#   given: a conversation exceeds the exact-history window
+#   then: the original task anchor and recent messages remain in-role while older turns are represented by one bounded historical checkpoint tied to the same conversation
+#   class: correctness
+#
+# id: harness_checkpoint_is_bounded
+#   given: arbitrarily many old text turns are compacted
+#   then: the continuity checkpoint stays within CHECKPOINT_MAX_CHARS and declares the persisted transcript authoritative
+#   class: resource
+#
+# id: harness_explicit_pin_never_falls_back
+#   given: a caller explicitly pins a model/provider and that call fails
+#   then: the harness propagates the failure without silently substituting another provider
+#   class: correctness
+#
+# id: harness_auto_fallback_replays_only_safe_turns
+#   given: an unpinned call fails before provider execution, or a tool-free call fails transiently
+#   then: the harness may retry an entitled configured provider; tool-capable post-dispatch failures are not automatically replayed
+#   class: safety
+# === END CONTRACTS ===
+
+from typing import Any, Optional
+
+import httpx
+
+from .agent_instance import AgentInstance
+from .energy_registry import cheap_provider
+from .model_catalog import list_models_for_user
+
+
+EXACT_HISTORY_LIMIT = 40
+RECENT_MESSAGES = 24
+CHECKPOINT_MAX_CHARS = 12_000
+CHECKPOINT_ENTRY_MAX_CHARS = 1_000
+CHECKPOINT_VERSION = 1
+
+
+def _history_entry(message: dict[str, Any], attachments: dict[int, list[dict]]) -> dict[str, Any]:
+    entry: dict[str, Any] = {
+        "role": message["role"],
+        "content": str(message.get("content") or ""),
+    }
+    mid = message.get("id")
+    atts = attachments.get(mid, []) if isinstance(mid, int) else []
+    if atts:
+        entry["attachments"] = [
+            {"storage_url": a.get("storage_url"), "mime_type": a.get("mime_type")}
+            for a in atts
+        ]
+    return entry
+
+
+def _checkpoint_line(message: dict[str, Any], attachment_count: int) -> str:
+    content = " ".join(str(message.get("content") or "").split())
+    if len(content) > CHECKPOINT_ENTRY_MAX_CHARS:
+        content = content[: CHECKPOINT_ENTRY_MAX_CHARS - 1].rstrip() + "…"
+    role = str(message.get("role") or "unknown").upper()
+    mid = message.get("id", "?")
+    suffix = f" [attachments:{attachment_count}]" if attachment_count else ""
+    return f"{role}[{mid}]{suffix}: {content}"
+
+
+def _bounded_checkpoint(text: str) -> str:
+    if len(text) <= CHECKPOINT_MAX_CHARS:
+        return text
+    marker = "[earlier checkpoint text omitted; persisted transcript remains authoritative]\n"
+    keep = CHECKPOINT_MAX_CHARS - len(marker)
+    return marker + text[-keep:]
+
+
+def prepare_continuity(
+    messages: list[dict[str, Any]],
+    attachments_by_message: Optional[dict[int, list[dict]]] = None,
+    previous_state: Optional[dict[str, Any]] = None,
+) -> tuple[list[dict[str, Any]], Optional[dict[str, Any]]]:
+    """Build bounded provider history without changing conversation identity.
+
+    The latest 24 messages and the first task-anchor message remain exact.
+    Older messages enter a bounded, explicitly historical checkpoint. The raw
+    database transcript remains the authority for lossless inspection.
+    """
+    attachments = attachments_by_message or {}
+    eligible = [m for m in messages if m.get("role") in ("user", "assistant")]
+    if len(eligible) <= EXACT_HISTORY_LIMIT:
+        return [_history_entry(m, attachments) for m in eligible], None
+
+    older = eligible[1:-RECENT_MESSAGES]
+    recent = eligible[-RECENT_MESSAGES:]
+    previous = previous_state if isinstance(previous_state, dict) else {}
+    previous_valid = (
+        previous.get("version") == CHECKPOINT_VERSION
+        and isinstance(previous.get("content"), str)
+        and isinstance(previous.get("through_message_id"), int)
+    )
+
+    old_ids = [m.get("id") for m in older if isinstance(m.get("id"), int)]
+    through = previous.get("through_message_id") if previous_valid else None
+    if previous_valid and through is not None and old_ids and through <= old_ids[-1]:
+        additions = [m for m in older if isinstance(m.get("id"), int) and m["id"] > through]
+        checkpoint_text = previous["content"]
+    else:
+        additions = older
+        checkpoint_text = ""
+
+    lines = [
+        _checkpoint_line(m, len(attachments.get(m.get("id"), [])))
+        for m in additions
+    ]
+    if lines:
+        checkpoint_text = "\n".join(part for part in (checkpoint_text, *lines) if part)
+    checkpoint_text = _bounded_checkpoint(checkpoint_text)
+
+    state = {
+        "version": CHECKPOINT_VERSION,
+        "through_message_id": old_ids[-1] if old_ids else 0,
+        "compacted_messages": len(older),
+        "content": checkpoint_text,
+        "hmmm": (
+            "bounded continuity is intentionally lossy; the persisted transcript "
+            "remains authoritative and old attachment payloads are represented only by count"
+        ),
+    }
+    checkpoint_message = {
+        "role": "user",
+        "content": (
+            "[A0 HARNESS CONTINUITY — historical transcript excerpts, not a new request. "
+            "USER lines are earlier user statements; ASSISTANT lines are earlier assistant output.]\n"
+            + checkpoint_text
+        ),
+    }
+    history = [
+        _history_entry(eligible[0], attachments),
+        checkpoint_message,
+        *[_history_entry(m, attachments) for m in recent],
+    ]
+    return history, state
+
+
+def continuity_state_summary(state: Optional[dict[str, Any]]) -> Optional[dict[str, Any]]:
+    """Return non-content checkpoint provenance safe to duplicate into usage metadata."""
+    if not state:
+        return None
+    return {
+        "version": state.get("version"),
+        "through_message_id": state.get("through_message_id"),
+        "compacted_messages": state.get("compacted_messages"),
+        "hmmm": state.get("hmmm"),
+    }
+
+
+def _preflight_failure(exc: BaseException) -> bool:
+    text = str(exc).lower()
+    if isinstance(exc, PermissionError):
+        return any(token in text for token in ("tier", "disabled", "provider"))
+    if isinstance(exc, ValueError):
+        return any(token in text for token in ("unknown model", "no routed model", "no call path"))
+    if isinstance(exc, RuntimeError):
+        return any(token in text for token in ("unavailable:", "api key", "env var", "no spec"))
+    return False
+
+
+def _transient_transport_failure(exc: BaseException) -> bool:
+    if isinstance(exc, (httpx.TimeoutException, httpx.NetworkError)):
+        return True
+    if isinstance(exc, httpx.HTTPStatusError):
+        try:
+            return exc.response.status_code == 429 or exc.response.status_code >= 500
+        except Exception:
+            return False
+    text = str(exc).lower()
+    return any(token in text for token in ("rate limit", "quota", "overloaded", "timed out", "timeout"))
+
+
+def _safe_to_replay(exc: BaseException, *, use_tools: bool) -> bool:
+    if _preflight_failure(exc):
+        return True
+    return not use_tools and _transient_transport_failure(exc)
+
+
+async def _fallback_candidates(user_id: Optional[str], seed_provider: Optional[str]) -> list[str]:
+    roster = await list_models_for_user(user_id)
+    rows = [
+        p for p in roster.get("providers", [])
+        if p.get("enabled") and p.get("key_present") and not p.get("tier_blocked")
+        and p.get("provider_id") != seed_provider
+    ]
+    cheap = cheap_provider()
+    rows.sort(key=lambda p: (p.get("provider_id") != cheap, str(p.get("provider_id") or "")))
+    candidates: list[str] = []
+    for provider in rows:
+        models = provider.get("models") or []
+        primary = next((m for m in models if m.get("is_primary") and not m.get("disabled")), None)
+        if primary and primary.get("model_id"):
+            candidates.append(str(primary["model_id"]))
+    return candidates
+
+
+async def run_single_turn(
+    *,
+    model_id: str,
+    messages: list[dict[str, Any]],
+    user_id: Optional[str],
+    system_prompt: Optional[str],
+    pin_requested_provider: bool,
+    use_tools: bool,
+    max_tokens: int = 8000,
+    reasoning_effort: Optional[str] = None,
+) -> tuple[str, dict[str, Any], str]:
+    """Run one turn while keeping provider substitution behind the harness.
+
+    Explicit pins never fall back. Auto mode may recover from failures known to
+    occur before dispatch. A tool-free call may also recover from transient
+    transport/quota errors. Tool-capable post-dispatch failures remain hmmm
+    because replay could duplicate an external side effect.
+    """
+    attempts: list[str] = [model_id]
+    primary = AgentInstance.from_model(
+        model_id=model_id,
+        user_id=user_id,
+        enforce_tier=False,
+        enforce_enabled=False,
+    )
+    primary.use_tools = use_tools
+    seed_provider = await primary.ensure_resolved()
+
+    try:
+        content, usage = await primary.run(
+            messages,
+            system_prompt_override=system_prompt,
+            max_tokens=max_tokens,
+            reasoning_effort=reasoning_effort,
+            pin_requested_provider=pin_requested_provider,
+            enforce_routed_tier=True,
+        )
+        actual_provider = primary.provider_id or seed_provider
+    except Exception as first_exc:
+        if pin_requested_provider:
+            raise
+        if not _safe_to_replay(first_exc, use_tools=use_tools):
+            if use_tools and _transient_transport_failure(first_exc):
+                raise RuntimeError(
+                    "hmmm: provider failed after tool-capable execution began; "
+                    "automatic provider replay was withheld to avoid duplicate side effects"
+                ) from first_exc
+            raise
+
+        last_exc: BaseException = first_exc
+        candidates = await _fallback_candidates(user_id, seed_provider)
+        for fallback_model in candidates:
+            attempts.append(fallback_model)
+            inst = AgentInstance.from_model(
+                model_id=fallback_model,
+                user_id=user_id,
+                enforce_tier=True,
+                enforce_enabled=True,
+            )
+            inst.use_tools = use_tools
+            try:
+                content, usage = await inst.run(
+                    messages,
+                    system_prompt_override=system_prompt,
+                    max_tokens=max_tokens,
+                    reasoning_effort=reasoning_effort,
+                    pin_requested_provider=True,
+                    enforce_routed_tier=True,
+                )
+                actual_provider = inst.provider_id or await inst.ensure_resolved()
+                break
+            except Exception as exc:
+                last_exc = exc
+                if not _safe_to_replay(exc, use_tools=use_tools):
+                    raise
+        else:
+            raise last_exc
+
+    traced = dict(usage or {})
+    traced["harness"] = {
+        "mode": "explicit-pin" if pin_requested_provider else "continuous-auto",
+        "attempted_models": attempts,
+        "fallback_count": max(0, len(attempts) - 1),
+        "actual_provider": actual_provider,
+    }
+    return content, traced, actual_provider
+# 218:82 0:0 1:3

From c98f8640d6572ac46e0e337257bd57296edf4ca7 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:22:05 -0700
Subject: [PATCH 02/29] test(harness): cover continuity and safe failover

---
 tests/test_work_harness_v0.0.0alpha.py | 190 +++++++++++++++++++++++++
 1 file changed, 190 insertions(+)
 create mode 100644 tests/test_work_harness_v0.0.0alpha.py

diff --git a/tests/test_work_harness_v0.0.0alpha.py b/tests/test_work_harness_v0.0.0alpha.py
new file mode 100644
index 00000000..8bece0ec
--- /dev/null
+++ b/tests/test_work_harness_v0.0.0alpha.py
@@ -0,0 +1,190 @@
+# 131:29 0:0 0:1
+"""Witnesses for the continuous A0 work harness.
+
+Usage:
+    uv run pytest -q tests/test_work_harness_v0.0.0alpha.py
+"""
+from __future__ import annotations
+
+# === CHECKS ===
+# id: check_harness_continuity_preserves_identity
+#   proves: harness_continuity_preserves_identity, harness_checkpoint_is_bounded
+#   call: self::test_continuity_keeps_anchor_recent_and_bounded_checkpoint
+#   requires: python3
+#   timeout: 10
+#   mutates: none
+#   cleanup: none
+#
+# id: check_harness_explicit_pin_never_falls_back
+#   proves: harness_explicit_pin_never_falls_back
+#   call: self::test_explicit_pin_never_falls_back
+#   requires: python3
+#   timeout: 10
+#   mutates: none
+#   cleanup: none
+#
+# id: check_harness_auto_fallback_replays_only_safe_turns
+#   proves: harness_auto_fallback_replays_only_safe_turns
+#   call: self::test_auto_mode_falls_back_on_preflight_failure
+#   requires: python3
+#   timeout: 10
+#   mutates: none
+#   cleanup: none
+# === END CHECKS ===
+
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+
+from python.services import work_harness
+
+
+def _messages(n: int) -> list[dict]:
+    return [
+        {
+            "id": i + 1,
+            "role": "user" if i % 2 == 0 else "assistant",
+            "content": f"message-{i + 1} " + ("x" * 80),
+        }
+        for i in range(n)
+    ]
+
+
+def test_continuity_keeps_anchor_recent_and_bounded_checkpoint() -> None:
+    messages = _messages(80)
+    history, state = work_harness.prepare_continuity(messages)
+    assert state is not None
+    assert history[0]["content"] == messages[0]["content"]
+    assert history[-1]["content"] == messages[-1]["content"]
+    assert history[1]["role"] == "user"
+    assert "A0 HARNESS CONTINUITY" in history[1]["content"]
+    assert state["through_message_id"] == messages[-work_harness.RECENT_MESSAGES - 1]["id"]
+    assert state["compacted_messages"] == 55
+    assert len(state["content"]) <= work_harness.CHECKPOINT_MAX_CHARS
+    assert "persisted transcript" in state["hmmm"]
+
+
+def test_continuity_incrementally_advances_checkpoint() -> None:
+    first, state = work_harness.prepare_continuity(_messages(60))
+    assert state is not None
+    _, next_state = work_harness.prepare_continuity(_messages(62), previous_state=state)
+    assert next_state is not None
+    assert next_state["through_message_id"] > state["through_message_id"]
+    assert next_state["compacted_messages"] > state["compacted_messages"]
+    assert first[0]["content"].startswith("message-1")
+
+
+class _FakeInstance:
+    failures: dict[str, BaseException] = {}
+    calls: list[str] = []
+
+    def __init__(self, model_id: str):
+        self.model_id = model_id
+        self.provider_id = None
+        self.use_tools = True
+
+    async def ensure_resolved(self) -> str:
+        self.provider_id = "seed-provider" if self.model_id == "seed" else "fallback-provider"
+        return self.provider_id
+
+    async def run(self, messages, **kwargs):
+        self.calls.append(self.model_id)
+        failure = self.failures.get(self.model_id)
+        if failure is not None:
+            raise failure
+        self.provider_id = "fallback-provider" if self.model_id != "seed" else "seed-provider"
+        return "ok", {"input_tokens": 1, "output_tokens": 1, "provider_id": self.provider_id}
+
+
+@pytest.fixture(autouse=True)
+def _fake_instances(monkeypatch: pytest.MonkeyPatch):
+    _FakeInstance.failures = {}
+    _FakeInstance.calls = []
+
+    def factory(model_id: str, **kwargs):
+        return _FakeInstance(model_id)
+
+    monkeypatch.setattr(work_harness.AgentInstance, "from_model", factory)
+    monkeypatch.setattr(work_harness, "cheap_provider", lambda: "fallback-provider")
+    monkeypatch.setattr(
+        work_harness,
+        "list_models_for_user",
+        AsyncMock(return_value={
+            "providers": [{
+                "provider_id": "fallback-provider",
+                "enabled": True,
+                "key_present": True,
+                "tier_blocked": False,
+                "models": [{"model_id": "fallback", "is_primary": True, "disabled": False}],
+            }]
+        }),
+    )
+
+
+@pytest.mark.asyncio
+async def test_auto_mode_falls_back_on_preflight_failure() -> None:
+    _FakeInstance.failures["seed"] = RuntimeError(
+        "seed-provider unavailable: env var SEED_API_KEY is not set"
+    )
+    content, usage, provider = await work_harness.run_single_turn(
+        model_id="seed",
+        messages=[{"role": "user", "content": "continue"}],
+        user_id="u1",
+        system_prompt=None,
+        pin_requested_provider=False,
+        use_tools=True,
+    )
+    assert content == "ok"
+    assert provider == "fallback-provider"
+    assert _FakeInstance.calls == ["seed", "fallback"]
+    assert usage["harness"]["fallback_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_explicit_pin_never_falls_back() -> None:
+    _FakeInstance.failures["seed"] = RuntimeError(
+        "seed-provider unavailable: env var SEED_API_KEY is not set"
+    )
+    with pytest.raises(RuntimeError, match="unavailable"):
+        await work_harness.run_single_turn(
+            model_id="seed",
+            messages=[{"role": "user", "content": "continue"}],
+            user_id="u1",
+            system_prompt=None,
+            pin_requested_provider=True,
+            use_tools=False,
+        )
+    assert _FakeInstance.calls == ["seed"]
+
+
+@pytest.mark.asyncio
+async def test_tool_capable_transient_failure_is_not_replayed() -> None:
+    _FakeInstance.failures["seed"] = httpx.TimeoutException("timed out")
+    with pytest.raises(RuntimeError, match="hmmm: provider failed"):
+        await work_harness.run_single_turn(
+            model_id="seed",
+            messages=[{"role": "user", "content": "continue"}],
+            user_id="u1",
+            system_prompt=None,
+            pin_requested_provider=False,
+            use_tools=True,
+        )
+    assert _FakeInstance.calls == ["seed"]
+
+
+@pytest.mark.asyncio
+async def test_tool_free_transient_failure_can_fall_back() -> None:
+    _FakeInstance.failures["seed"] = httpx.TimeoutException("timed out")
+    content, usage, provider = await work_harness.run_single_turn(
+        model_id="seed",
+        messages=[{"role": "user", "content": "continue"}],
+        user_id="u1",
+        system_prompt=None,
+        pin_requested_provider=False,
+        use_tools=False,
+    )
+    assert content == "ok"
+    assert provider == "fallback-provider"
+    assert usage["harness"]["fallback_count"] == 1
+# 131:29 0:0 0:1

From d82a9b117d3b340f4e184467e93abae12368d22d Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:22:41 -0700
Subject: [PATCH 03/29] fix(harness): map conversation runtime state

---
 python/models.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/python/models.py b/python/models.py
index a2d23e5e..a1a1d732 100644
--- a/python/models.py
+++ b/python/models.py
@@ -1,4 +1,4 @@
-# 358:22 0:0 5:0
+# 361:23 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"))
 
@@ -448,3 +451,4 @@ class GeneratedImage(Base):
     tags = Column(JSONB, server_default=text("'[]'::jsonb"))
     skill_origin = Column(Text)
 # 358:22 0:0 5:0
+# 361:23 0:0 5:0

From 1cbbf698b0545d1278a90f78b919a47e1017a80d Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:22:43 -0700
Subject: [PATCH 04/29] feat(harness): persist conversation harness state

---
 shared/schema.ts | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/shared/schema.ts b/shared/schema.ts
index 220d28b0..46541700 100644
--- a/shared/schema.ts
+++ b/shared/schema.ts
@@ -1,4 +1,4 @@
-// 697:26 0:141 0:0
+// 698:27 0:141 0:0
 import { sql } from "drizzle-orm";
 import { pgTable, text, varchar, serial, integer, timestamp, jsonb, real, boolean, uniqueIndex, index, numeric } from "drizzle-orm/pg-core";
 import { createInsertSchema } from "drizzle-zod";
@@ -20,6 +20,7 @@ export const conversations = pgTable("conversations", {
   enabledTools: jsonb("enabled_tools").$type<string[] | null>(),
   maxToolRounds: integer("max_tool_rounds"),
   inferenceMode: varchar("inference_mode", { length: 20 }).default("agentic"),
+  harnessState: jsonb("harness_state").$type<Record<string, unknown> | null>(),
   createdAt: timestamp("created_at").default(sql`CURRENT_TIMESTAMP`).notNull(),
   updatedAt: timestamp("updated_at").default(sql`CURRENT_TIMESTAMP`).notNull(),
 }, (t) => [
@@ -823,3 +824,4 @@ export const settings = pgTable("settings", {
 
 export type Setting = typeof settings.$inferSelect;
 // 697:26 0:141 0:0
+// 698:27 0:141 0:0

From d9907f97c9aef3122968ad23f691b85277badddf Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:22:45 -0700
Subject: [PATCH 05/29] feat(harness): persist continuity checkpoint

---
 python/storage/core.py | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/python/storage/core.py b/python/storage/core.py
index 17faf94d..eed92a41 100644
--- a/python/storage/core.py
+++ b/python/storage/core.py
@@ -1,4 +1,4 @@
-# 393:53 0:0 4:2
+# 399:58 0:0 4:2
 from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional
 from sqlalchemy import select, update, delete, func, desc, asc, or_
@@ -117,6 +117,18 @@ async def update_conversation_title(self, id: int, title: str) -> None:
                 .values(title=title, updated_at=datetime.utcnow())
             )
 
+    async def update_conversation_harness_state(self, id: int, state: Dict[str, Any]) -> None:
+        """Persist the bounded continuity state for one conversation.
+
+        Usage: work_harness prepares the state; chat owns authorization and calls
+        this only after it has resolved the conversation to the authenticated owner.
+        """
+        async with get_session() as session:
+            await session.execute(
+                update(Conversation).where(Conversation.id == id)
+                .values(harness_state=state, updated_at=datetime.utcnow())
+            )
+
     async def delete_conversation(self, id: int) -> None:
         async with get_session() as session:
             await session.execute(delete(Conversation).where(Conversation.id == id))
@@ -505,3 +517,4 @@ async def delete_custom_tool(self, id: int) -> None:
 #   class: security
 # === END CONTRACTS ===
 # 393:53 0:0 4:2
+# 399:58 0:0 4:2

From 8be86777e5f62e0914e92ce2324f583026623466 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:22:48 -0700
Subject: [PATCH 06/29] feat(harness): route chat through continuous harness

---
 python/routes/chat.py | 62 +++++++++++++++++--------------------------
 1 file changed, 24 insertions(+), 38 deletions(-)

diff --git a/python/routes/chat.py b/python/routes/chat.py
index 814202f8..5f48fd25 100644
--- a/python/routes/chat.py
+++ b/python/routes/chat.py
@@ -1,4 +1,4 @@
-# 685:192 2:7 2:16
+# 677:186 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,
@@ -963,3 +948,4 @@ async def _grant_scope_if_valid(scope: str) -> tuple[bool, str]:
 #   class: security
 # === END CONTRACTS ===
 # 685:192 2:7 2:16
+# 677:186 2:7 2:16

From 758c3103fc1dee2b28375a9fc9ca1b6f7a3480cf Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:24:03 -0700
Subject: [PATCH 07/29] test(harness): follow provider pin through harness seam

---
 tests/test_open_comp_rout_v0.0.0alpha.py | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/tests/test_open_comp_rout_v0.0.0alpha.py b/tests/test_open_comp_rout_v0.0.0alpha.py
index c6f2bccd..418abea7 100644
--- a/tests/test_open_comp_rout_v0.0.0alpha.py
+++ b/tests/test_open_comp_rout_v0.0.0alpha.py
@@ -1,4 +1,4 @@
-# 399:1 0:0 0:0
+# 402:2 0:0 0:0
 """Routing and catalog tests for registry-driven compatible providers."""
 
 import ast
@@ -69,16 +69,19 @@ def test_approval_replays_preserve_explicit_provider_pin() -> None:
         for call in replay_calls
     )
 
-    instance_runs = [
+    harness_calls = [
         node
         for node in ast.walk(tree)
         if isinstance(node, ast.Call)
-        and isinstance(node.func, ast.Attribute)
-        and node.func.attr == "run"
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "run_single_turn"
     ]
+    assert len(harness_calls) == 1
     assert any(
-        any(keyword.arg == "pin_requested_provider" for keyword in call.keywords)
-        for call in instance_runs
+        keyword.arg == "pin_requested_provider"
+        and isinstance(keyword.value, ast.Name)
+        and keyword.value.id == "provider_pin_requested"
+        for keyword in harness_calls[0].keywords
     )
 
     pending_writes = [
@@ -509,3 +512,4 @@ async def test_catalog_resolver_pricing_and_missing_key_are_fail_closed(
             use_tools=False,
         )
 # 399:1 0:0 0:0
+# 402:2 0:0 0:0

From 99d4a6dda2b237f60b4c6f94a17553429814e06a Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:25:17 -0700
Subject: [PATCH 08/29] test(harness): register canonical contract witnesses

---
 python/tests/test_work_harness_contracts.py | 44 +++++++++++++++++++++
 1 file changed, 44 insertions(+)
 create mode 100644 python/tests/test_work_harness_contracts.py

diff --git a/python/tests/test_work_harness_contracts.py b/python/tests/test_work_harness_contracts.py
new file mode 100644
index 00000000..d03a9d7f
--- /dev/null
+++ b/python/tests/test_work_harness_contracts.py
@@ -0,0 +1,44 @@
+# 22:15 0:0 0:0
+"""Canonical contract-runner witness for the continuous A0 harness.
+
+Usage:
+    uv run pytest -q python/tests/test_work_harness_contracts.py
+The witness delegates to the focused behavioral suite so the contract graph and
+ordinary pytest execution share one source of executable evidence.
+"""
+
+# === CHECKS ===
+# id: check_work_harness_contracts
+#   proves: harness_continuity_preserves_identity, harness_checkpoint_is_bounded, harness_explicit_pin_never_falls_back, harness_auto_fallback_replays_only_safe_turns
+#   call: self::test_work_harness_contracts
+#   requires: python3, pytest
+#   timeout: 30
+#   mutates: none
+#   cleanup: none
+# === END CHECKS ===
+
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+
+def test_work_harness_contracts() -> None:
+    root = Path(__file__).resolve().parents[2]
+    environment = dict(os.environ)
+    environment["PYTHONDONTWRITEBYTECODE"] = "1"
+    subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "pytest",
+            "-q",
+            "-p",
+            "no:cacheprovider",
+            "tests/test_work_harness_v0.0.0alpha.py",
+        ],
+        cwd=root,
+        env=environment,
+        check=True,
+    )
+# 22:15 0:0 0:0

From 8b4a56e3a0cc7dd4c26447e2ca0394ce6ae8a631 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:28:40 -0700
Subject: [PATCH 09/29] chore(msdmd): normalize file annotations

---
 python/models.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/python/models.py b/python/models.py
index a1a1d732..39cbf671 100644
--- a/python/models.py
+++ b/python/models.py
@@ -1,4 +1,4 @@
-# 361:23 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
@@ -450,5 +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:23 0:0 5:0
+# 361:22 0:0 5:0

From 34968522a3bc6073ab956a181b98b29c5833c07d Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:28:42 -0700
Subject: [PATCH 10/29] chore(msdmd): normalize file annotations

---
 python/routes/chat.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/python/routes/chat.py b/python/routes/chat.py
index 5f48fd25..33141792 100644
--- a/python/routes/chat.py
+++ b/python/routes/chat.py
@@ -1,4 +1,4 @@
-# 677:186 2:7 2:16
+# 677:185 2:7 2:16
 import time
 import traceback
 from fastapi import APIRouter, HTTPException, Request
@@ -947,5 +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:186 2:7 2:16
+# 677:185 2:7 2:16

From 32cd0e7618650031c1da62861b6f563d8f1dc767 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:28:45 -0700
Subject: [PATCH 11/29] chore(msdmd): normalize file annotations

---
 python/storage/core.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/python/storage/core.py b/python/storage/core.py
index eed92a41..5bdf0660 100644
--- a/python/storage/core.py
+++ b/python/storage/core.py
@@ -1,4 +1,4 @@
-# 399:58 0:0 4:2
+# 399:57 0:0 4:2
 from datetime import datetime, timedelta
 from typing import Any, Dict, List, Optional
 from sqlalchemy import select, update, delete, func, desc, asc, or_
@@ -516,5 +516,4 @@ async def delete_custom_tool(self, id: int) -> None:
 #          owner field from the request body)
 #   class: security
 # === END CONTRACTS ===
-# 393:53 0:0 4:2
-# 399:58 0:0 4:2
+# 399:57 0:0 4:2

From ad525bd7134d70f0e9f9cf968cbbca3aad6d2bbb Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:28:47 -0700
Subject: [PATCH 12/29] chore(msdmd): normalize file annotations

---
 shared/schema.ts | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/shared/schema.ts b/shared/schema.ts
index 46541700..1a8d211f 100644
--- a/shared/schema.ts
+++ b/shared/schema.ts
@@ -1,4 +1,4 @@
-// 698:27 0:141 0:0
+// 698:26 0:141 0:0
 import { sql } from "drizzle-orm";
 import { pgTable, text, varchar, serial, integer, timestamp, jsonb, real, boolean, uniqueIndex, index, numeric } from "drizzle-orm/pg-core";
 import { createInsertSchema } from "drizzle-zod";
@@ -823,5 +823,4 @@ export const settings = pgTable("settings", {
 }, (t) => [uniqueIndex("uq_settings_user_key").on(t.userId, t.key)]);
 
 export type Setting = typeof settings.$inferSelect;
-// 697:26 0:141 0:0
-// 698:27 0:141 0:0
+// 698:26 0:141 0:0

From 0975d5e9fd164f63d7fc1192460e386bcae33235 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:28:49 -0700
Subject: [PATCH 13/29] refactor(test): keep routing witness within module
 budget

---
 tests/test_open_comp_rout_v0.0.0alpha.py | 21 +++++++--------------
 1 file changed, 7 insertions(+), 14 deletions(-)

diff --git a/tests/test_open_comp_rout_v0.0.0alpha.py b/tests/test_open_comp_rout_v0.0.0alpha.py
index 418abea7..12052210 100644
--- a/tests/test_open_comp_rout_v0.0.0alpha.py
+++ b/tests/test_open_comp_rout_v0.0.0alpha.py
@@ -1,4 +1,4 @@
-# 402:2 0:0 0:0
+# 396:1 0:0 0:0
 """Routing and catalog tests for registry-driven compatible providers."""
 
 import ast
@@ -70,19 +70,13 @@ def test_approval_replays_preserve_explicit_provider_pin() -> None:
     )
 
     harness_calls = [
-        node
-        for node in ast.walk(tree)
-        if isinstance(node, ast.Call)
-        and isinstance(node.func, ast.Name)
-        and node.func.id == "run_single_turn"
+        node for node in ast.walk(tree)
+        if isinstance(node, ast.Call) and getattr(node.func, "id", None) == "run_single_turn"
     ]
     assert len(harness_calls) == 1
-    assert any(
-        keyword.arg == "pin_requested_provider"
-        and isinstance(keyword.value, ast.Name)
-        and keyword.value.id == "provider_pin_requested"
-        for keyword in harness_calls[0].keywords
-    )
+    pin_kw = next(k for k in harness_calls[0].keywords if k.arg == "pin_requested_provider")
+    assert isinstance(pin_kw.value, ast.Name)
+    assert pin_kw.value.id == "provider_pin_requested"
 
     pending_writes = [
         node
@@ -511,5 +505,4 @@ async def test_catalog_resolver_pricing_and_missing_key_are_fail_closed(
             provider_id="deepseek",
             use_tools=False,
         )
-# 399:1 0:0 0:0
-# 402:2 0:0 0:0
+# 396:1 0:0 0:0

From 702b62e10b99753fda436a4a7345d8d82fb7dc7a Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:28:52 -0700
Subject: [PATCH 14/29] fix(harness): ensure production continuity column

---
 python/main.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/python/main.py b/python/main.py
index 17f830eb..c580cc3e 100644
--- a/python/main.py
+++ b/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

From 08b09d67ebebf417dc88982a1f67aaa47bb27400 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:31:16 -0700
Subject: [PATCH 15/29] feat(harness): track run tool executions

---
 python/services/run_context.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/python/services/run_context.py b/python/services/run_context.py
index 74f83fed..64ded221 100644
--- a/python/services/run_context.py
+++ b/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

From da25e3457f8940870e230018afff4677e268e305 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:31:19 -0700
Subject: [PATCH 16/29] feat(harness): mark tool execution boundary

---
 python/services/tool_executor.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/python/services/tool_executor.py b/python/services/tool_executor.py
index 3cd9c896..529f0900 100644
--- a/python/services/tool_executor.py
+++ b/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

From a2582cf458871d83739071f8b78dd19ef4376cef Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:31:21 -0700
Subject: [PATCH 17/29] feat(harness): fail over safely before tool execution

---
 python/services/work_harness.py | 34 +++++++++++++++++++++++++--------
 1 file changed, 26 insertions(+), 8 deletions(-)

diff --git a/python/services/work_harness.py b/python/services/work_harness.py
index 94a66286..15e02471 100644
--- a/python/services/work_harness.py
+++ b/python/services/work_harness.py
@@ -1,4 +1,4 @@
-# 218:82 0:0 1:3
+# 236:82 0:0 1:3
 """Continuous A0 work harness.
 
 The conversation is the durable work identity. Model/provider selection is an
@@ -84,6 +84,7 @@
 from .agent_instance import AgentInstance
 from .energy_registry import cheap_provider
 from .model_catalog import list_models_for_user
+from .run_context import current_tool_executions
 
 
 EXACT_HISTORY_LIMIT = 40
@@ -229,10 +230,10 @@ def _transient_transport_failure(exc: BaseException) -> bool:
     return any(token in text for token in ("rate limit", "quota", "overloaded", "timed out", "timeout"))
 
 
-def _safe_to_replay(exc: BaseException, *, use_tools: bool) -> bool:
+def _safe_to_replay(exc: BaseException, *, tool_executions: int) -> bool:
     if _preflight_failure(exc):
         return True
-    return not use_tools and _transient_transport_failure(exc)
+    return tool_executions == 0 and _transient_transport_failure(exc)
 
 
 async def _fallback_candidates(user_id: Optional[str], seed_provider: Optional[str]) -> list[str]:
@@ -272,6 +273,7 @@ async def run_single_turn(
     because replay could duplicate an external side effect.
     """
     attempts: list[str] = [model_id]
+    tool_token = current_tool_executions.set(0)
     primary = AgentInstance.from_model(
         model_id=model_id,
         user_id=user_id,
@@ -293,13 +295,17 @@ async def run_single_turn(
         actual_provider = primary.provider_id or seed_provider
     except Exception as first_exc:
         if pin_requested_provider:
+            current_tool_executions.reset(tool_token)
             raise
-        if not _safe_to_replay(first_exc, use_tools=use_tools):
-            if use_tools and _transient_transport_failure(first_exc):
+        executed = current_tool_executions.get()
+        if not _safe_to_replay(first_exc, tool_executions=executed):
+            if executed > 0 and _transient_transport_failure(first_exc):
+                current_tool_executions.reset(tool_token)
                 raise RuntimeError(
-                    "hmmm: provider failed after tool-capable execution began; "
+                    "hmmm: provider failed after a tool execution boundary; "
                     "automatic provider replay was withheld to avoid duplicate side effects"
                 ) from first_exc
+            current_tool_executions.reset(tool_token)
             raise
 
         last_exc: BaseException = first_exc
@@ -326,17 +332,29 @@ async def run_single_turn(
                 break
             except Exception as exc:
                 last_exc = exc
-                if not _safe_to_replay(exc, use_tools=use_tools):
+                executed = current_tool_executions.get()
+                if not _safe_to_replay(exc, tool_executions=executed):
+                    if executed > 0 and _transient_transport_failure(exc):
+                        current_tool_executions.reset(tool_token)
+                        raise RuntimeError(
+                            "hmmm: fallback provider failed after a tool execution boundary; "
+                            "further replay was withheld to avoid duplicate side effects"
+                        ) from exc
+                    current_tool_executions.reset(tool_token)
                     raise
         else:
+            current_tool_executions.reset(tool_token)
             raise last_exc
 
+    tool_executions = current_tool_executions.get()
+    current_tool_executions.reset(tool_token)
     traced = dict(usage or {})
     traced["harness"] = {
         "mode": "explicit-pin" if pin_requested_provider else "continuous-auto",
         "attempted_models": attempts,
         "fallback_count": max(0, len(attempts) - 1),
         "actual_provider": actual_provider,
+        "tool_executions": tool_executions,
     }
     return content, traced, actual_provider
-# 218:82 0:0 1:3
+# 236:82 0:0 1:3

From 4a6c396bdbdaf1548a5452d5321fe0a7db81c9b3 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:31:23 -0700
Subject: [PATCH 18/29] test(harness): distinguish pre-tool and post-tool
 failures

---
 tests/test_work_harness_v0.0.0alpha.py | 32 ++++++++++++++++++++++----
 1 file changed, 28 insertions(+), 4 deletions(-)

diff --git a/tests/test_work_harness_v0.0.0alpha.py b/tests/test_work_harness_v0.0.0alpha.py
index 8bece0ec..f8f98703 100644
--- a/tests/test_work_harness_v0.0.0alpha.py
+++ b/tests/test_work_harness_v0.0.0alpha.py
@@ -1,4 +1,4 @@
-# 131:29 0:0 0:1
+# 152:29 0:0 0:1
 """Witnesses for the continuous A0 work harness.
 
 Usage:
@@ -159,9 +159,33 @@ async def test_explicit_pin_never_falls_back() -> None:
 
 
 @pytest.mark.asyncio
-async def test_tool_capable_transient_failure_is_not_replayed() -> None:
+async def test_agentic_transient_failure_before_tools_can_fall_back() -> None:
     _FakeInstance.failures["seed"] = httpx.TimeoutException("timed out")
-    with pytest.raises(RuntimeError, match="hmmm: provider failed"):
+    content, usage, provider = await work_harness.run_single_turn(
+        model_id="seed",
+        messages=[{"role": "user", "content": "continue"}],
+        user_id="u1",
+        system_prompt=None,
+        pin_requested_provider=False,
+        use_tools=True,
+    )
+    assert content == "ok"
+    assert provider == "fallback-provider"
+    assert _FakeInstance.calls == ["seed", "fallback"]
+    assert usage["harness"]["tool_executions"] == 0
+
+
+@pytest.mark.asyncio
+async def test_agentic_transient_failure_after_tool_boundary_is_not_replayed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def unsafe_run(self, messages, **kwargs):
+        self.calls.append(self.model_id)
+        work_harness.current_tool_executions.set(1)
+        raise httpx.TimeoutException("timed out")
+
+    monkeypatch.setattr(_FakeInstance, "run", unsafe_run)
+    with pytest.raises(RuntimeError, match="hmmm: provider failed after a tool execution boundary"):
         await work_harness.run_single_turn(
             model_id="seed",
             messages=[{"role": "user", "content": "continue"}],
@@ -187,4 +211,4 @@ async def test_tool_free_transient_failure_can_fall_back() -> None:
     assert content == "ok"
     assert provider == "fallback-provider"
     assert usage["harness"]["fallback_count"] == 1
-# 131:29 0:0 0:1
+# 152:29 0:0 0:1

From c3fd041df5890287b8c3af7bd9d939d69e7e28d7 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:32:09 -0700
Subject: [PATCH 19/29] fix(harness): reset execution evidence at one boundary

---
 python/services/work_harness.py | 136 ++++++++++++++++----------------
 1 file changed, 66 insertions(+), 70 deletions(-)

diff --git a/python/services/work_harness.py b/python/services/work_harness.py
index 15e02471..d609f334 100644
--- a/python/services/work_harness.py
+++ b/python/services/work_harness.py
@@ -1,4 +1,4 @@
-# 236:82 0:0 1:3
+# 232:82 0:0 1:3
 """Continuous A0 work harness.
 
 The conversation is the durable work identity. Model/provider selection is an
@@ -267,13 +267,12 @@ async def run_single_turn(
 ) -> tuple[str, dict[str, Any], str]:
     """Run one turn while keeping provider substitution behind the harness.
 
-    Explicit pins never fall back. Auto mode may recover from failures known to
-    occur before dispatch. A tool-free call may also recover from transient
-    transport/quota errors. Tool-capable post-dispatch failures remain hmmm
+    Explicit pins never fall back. Auto mode may recover from pre-dispatch
+    failures, including transient quota/transport failures before any registry
+    tool executes. After a tool boundary, transient failure remains hmmm
     because replay could duplicate an external side effect.
     """
     attempts: list[str] = [model_id]
-    tool_token = current_tool_executions.set(0)
     primary = AgentInstance.from_model(
         model_id=model_id,
         user_id=user_id,
@@ -282,72 +281,69 @@ async def run_single_turn(
     )
     primary.use_tools = use_tools
     seed_provider = await primary.ensure_resolved()
-
+    tool_token = current_tool_executions.set(0)
     try:
-        content, usage = await primary.run(
-            messages,
-            system_prompt_override=system_prompt,
-            max_tokens=max_tokens,
-            reasoning_effort=reasoning_effort,
-            pin_requested_provider=pin_requested_provider,
-            enforce_routed_tier=True,
-        )
-        actual_provider = primary.provider_id or seed_provider
-    except Exception as first_exc:
-        if pin_requested_provider:
-            current_tool_executions.reset(tool_token)
-            raise
-        executed = current_tool_executions.get()
-        if not _safe_to_replay(first_exc, tool_executions=executed):
-            if executed > 0 and _transient_transport_failure(first_exc):
-                current_tool_executions.reset(tool_token)
-                raise RuntimeError(
-                    "hmmm: provider failed after a tool execution boundary; "
-                    "automatic provider replay was withheld to avoid duplicate side effects"
-                ) from first_exc
-            current_tool_executions.reset(tool_token)
-            raise
-
-        last_exc: BaseException = first_exc
-        candidates = await _fallback_candidates(user_id, seed_provider)
-        for fallback_model in candidates:
-            attempts.append(fallback_model)
-            inst = AgentInstance.from_model(
-                model_id=fallback_model,
-                user_id=user_id,
-                enforce_tier=True,
-                enforce_enabled=True,
+        try:
+            content, usage = await primary.run(
+                messages,
+                system_prompt_override=system_prompt,
+                max_tokens=max_tokens,
+                reasoning_effort=reasoning_effort,
+                pin_requested_provider=pin_requested_provider,
+                enforce_routed_tier=True,
             )
-            inst.use_tools = use_tools
-            try:
-                content, usage = await inst.run(
-                    messages,
-                    system_prompt_override=system_prompt,
-                    max_tokens=max_tokens,
-                    reasoning_effort=reasoning_effort,
-                    pin_requested_provider=True,
-                    enforce_routed_tier=True,
-                )
-                actual_provider = inst.provider_id or await inst.ensure_resolved()
-                break
-            except Exception as exc:
-                last_exc = exc
-                executed = current_tool_executions.get()
-                if not _safe_to_replay(exc, tool_executions=executed):
-                    if executed > 0 and _transient_transport_failure(exc):
-                        current_tool_executions.reset(tool_token)
-                        raise RuntimeError(
-                            "hmmm: fallback provider failed after a tool execution boundary; "
-                            "further replay was withheld to avoid duplicate side effects"
-                        ) from exc
-                    current_tool_executions.reset(tool_token)
-                    raise
-        else:
-            current_tool_executions.reset(tool_token)
-            raise last_exc
-
-    tool_executions = current_tool_executions.get()
-    current_tool_executions.reset(tool_token)
+            actual_provider = primary.provider_id or seed_provider
+        except Exception as first_exc:
+            if pin_requested_provider:
+                raise
+            executed = current_tool_executions.get()
+            if not _safe_to_replay(first_exc, tool_executions=executed):
+                if executed > 0 and _transient_transport_failure(first_exc):
+                    raise RuntimeError(
+                        "hmmm: provider failed after a tool execution boundary; "
+                        "automatic provider replay was withheld to avoid duplicate side effects"
+                    ) from first_exc
+                raise
+
+            last_exc: BaseException = first_exc
+            candidates = await _fallback_candidates(user_id, seed_provider)
+            for fallback_model in candidates:
+                attempts.append(fallback_model)
+                try:
+                    inst = AgentInstance.from_model(
+                        model_id=fallback_model,
+                        user_id=user_id,
+                        enforce_tier=True,
+                        enforce_enabled=True,
+                    )
+                    inst.use_tools = use_tools
+                    content, usage = await inst.run(
+                        messages,
+                        system_prompt_override=system_prompt,
+                        max_tokens=max_tokens,
+                        reasoning_effort=reasoning_effort,
+                        pin_requested_provider=True,
+                        enforce_routed_tier=True,
+                    )
+                    actual_provider = inst.provider_id or await inst.ensure_resolved()
+                    break
+                except Exception as exc:
+                    last_exc = exc
+                    executed = current_tool_executions.get()
+                    if not _safe_to_replay(exc, tool_executions=executed):
+                        if executed > 0 and _transient_transport_failure(exc):
+                            raise RuntimeError(
+                                "hmmm: fallback provider failed after a tool execution boundary; "
+                                "further replay was withheld to avoid duplicate side effects"
+                            ) from exc
+                        raise
+            else:
+                raise last_exc
+
+        tool_executions = current_tool_executions.get()
+    finally:
+        current_tool_executions.reset(tool_token)
+
     traced = dict(usage or {})
     traced["harness"] = {
         "mode": "explicit-pin" if pin_requested_provider else "continuous-auto",
@@ -357,4 +353,4 @@ async def run_single_turn(
         "tool_executions": tool_executions,
     }
     return content, traced, actual_provider
-# 236:82 0:0 1:3
+# 232:82 0:0 1:3

From 6321fa2696489d19e88ee01b7aa2a1629eef9b57 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:32:11 -0700
Subject: [PATCH 20/29] test(harness): prove execution counter cleanup

---
 tests/test_work_harness_v0.0.0alpha.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/tests/test_work_harness_v0.0.0alpha.py b/tests/test_work_harness_v0.0.0alpha.py
index f8f98703..d937f695 100644
--- a/tests/test_work_harness_v0.0.0alpha.py
+++ b/tests/test_work_harness_v0.0.0alpha.py
@@ -1,4 +1,4 @@
-# 152:29 0:0 0:1
+# 153:29 0:0 0:1
 """Witnesses for the continuous A0 work harness.
 
 Usage:
@@ -195,6 +195,7 @@ async def unsafe_run(self, messages, **kwargs):
             use_tools=True,
         )
     assert _FakeInstance.calls == ["seed"]
+    assert work_harness.current_tool_executions.get() == 0
 
 
 @pytest.mark.asyncio
@@ -211,4 +212,4 @@ async def test_tool_free_transient_failure_can_fall_back() -> None:
     assert content == "ok"
     assert provider == "fallback-provider"
     assert usage["harness"]["fallback_count"] == 1
-# 152:29 0:0 0:1
+# 153:29 0:0 0:1

From 13798385db0d4eb90dc8a252c3270d58c1171932 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:32:56 -0700
Subject: [PATCH 21/29] docs(harness): align replay and mutation boundaries

---
 python/services/work_harness.py | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/python/services/work_harness.py b/python/services/work_harness.py
index d609f334..4c8f8ab4 100644
--- a/python/services/work_harness.py
+++ b/python/services/work_harness.py
@@ -32,7 +32,7 @@
 #   auth_boundary: none
 #   storage_boundary: read
 #   network_boundary: external
-#   user_data_boundary: read
+#   user_data_boundary: write
 #   admin_only: false
 #   tests: tests/test_work_harness_v0.0.0alpha.py
 #   rollout: default_enabled
@@ -44,11 +44,11 @@
 
 # === BOUNDARIES ===
 # id: work_harness_runtime_boundary
-#   summary: Reads user conversation text and may call configured external model providers; it performs no persistence itself.
-#   auth_boundary: none
+#   summary: Reads conversation state and may invoke configured providers; tool-enabled calls can cross downstream mutating-tool boundaries while this service itself performs no persistence.
+#   auth_boundary: delegates tier enforcement and tool approvals to existing runtime gates
 #   storage_boundary: read
 #   network_boundary: external
-#   user_data_boundary: read
+#   user_data_boundary: write
 #   admin_only: false
 #   pii: possible
 #   secrets: none
@@ -72,8 +72,8 @@
 #   class: correctness
 #
 # id: harness_auto_fallback_replays_only_safe_turns
-#   given: an unpinned call fails before provider execution, or a tool-free call fails transiently
-#   then: the harness may retry an entitled configured provider; tool-capable post-dispatch failures are not automatically replayed
+#   given: an unpinned call fails before any registry tool executes, including a transient provider quota or transport failure
+#   then: the harness may retry an entitled configured provider; after a registry tool executes, transient failures are not automatically replayed
 #   class: safety
 # === END CONTRACTS ===
 

From 2ae8ccbb92a927174657012ef7da17fd4602baf1 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:32:58 -0700
Subject: [PATCH 22/29] test(harness): witness registry execution boundary

---
 tests/test_work_harness_v0.0.0alpha.py | 26 +++++++++++++++++++++++---
 1 file changed, 23 insertions(+), 3 deletions(-)

diff --git a/tests/test_work_harness_v0.0.0alpha.py b/tests/test_work_harness_v0.0.0alpha.py
index d937f695..357e5c82 100644
--- a/tests/test_work_harness_v0.0.0alpha.py
+++ b/tests/test_work_harness_v0.0.0alpha.py
@@ -1,4 +1,4 @@
-# 153:29 0:0 0:1
+# 167:29 0:0 0:1
 """Witnesses for the continuous A0 work harness.
 
 Usage:
@@ -37,7 +37,7 @@
 import httpx
 import pytest
 
-from python.services import work_harness
+from python.services import tool_executor, work_harness
 
 
 def _messages(n: int) -> list[dict]:
@@ -122,6 +122,26 @@ def factory(model_id: str, **kwargs):
     )
 
 
+
+
+
+@pytest.mark.asyncio
+async def test_registry_dispatch_marks_execution_boundary(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def fake_dispatch(name: str, **kwargs):
+        return f"ok:{name}"
+
+    token = work_harness.current_tool_executions.set(0)
+    monkeypatch.setattr(tool_executor, "_registry_dispatch", fake_dispatch)
+    try:
+        result = await tool_executor._execute_tool_inner("harness_probe", {})
+        assert result == "ok:harness_probe"
+        assert work_harness.current_tool_executions.get() == 1
+    finally:
+        work_harness.current_tool_executions.reset(token)
+
+
 @pytest.mark.asyncio
 async def test_auto_mode_falls_back_on_preflight_failure() -> None:
     _FakeInstance.failures["seed"] = RuntimeError(
@@ -212,4 +232,4 @@ async def test_tool_free_transient_failure_can_fall_back() -> None:
     assert content == "ok"
     assert provider == "fallback-provider"
     assert usage["harness"]["fallback_count"] == 1
-# 153:29 0:0 0:1
+# 167:29 0:0 0:1

From aaad3247ee4c660c4c5911302c773e21535e8ec2 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:38:52 -0700
Subject: [PATCH 23/29] feat(harness): type sanitized provider failures

---
 python/services/provider_failure.py | 63 +++++++++++++++++++++++++++++
 1 file changed, 63 insertions(+)
 create mode 100644 python/services/provider_failure.py

diff --git a/python/services/provider_failure.py b/python/services/provider_failure.py
new file mode 100644
index 00000000..8ae20e7b
--- /dev/null
+++ b/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

From 141f4cecda1be1a1073bbe8de889f79a889dbb16 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:38:54 -0700
Subject: [PATCH 24/29] feat(harness): mark sanitized provider failures

---
 python/services/inference.py | 21 ++++++++++-----------
 1 file changed, 10 insertions(+), 11 deletions(-)

diff --git a/python/services/inference.py b/python/services/inference.py
index 50be4af1..dabe49f2 100644
--- a/python/services/inference.py
+++ b/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

From 823f7b798af5ef87d18e1a13ff8c0a3475514973 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:38:57 -0700
Subject: [PATCH 25/29] feat(harness): type compatible provider failures

---
 python/services/providers/open_comp_prov_v0.0.0alpha.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/python/services/providers/open_comp_prov_v0.0.0alpha.py b/python/services/providers/open_comp_prov_v0.0.0alpha.py
index 216d9b0b..6fe0f72b 100644
--- a/python/services/providers/open_comp_prov_v0.0.0alpha.py
+++ b/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

From 02563686f1dbd8581c1b0cb761686d192f250663 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:38:59 -0700
Subject: [PATCH 26/29] feat(harness): type Gemini provider failures

---
 python/services/gemini_native.py | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/python/services/gemini_native.py b/python/services/gemini_native.py
index 96957d14..c39d5488 100644
--- a/python/services/gemini_native.py
+++ b/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

From 253214f8b6e776db3427e8370ce79d93b47d71cb Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:39:01 -0700
Subject: [PATCH 27/29] feat(harness): fail over typed provider failures

---
 python/services/work_harness.py | 96 ++++++++++++++++++++++++++++-----
 1 file changed, 83 insertions(+), 13 deletions(-)

diff --git a/python/services/work_harness.py b/python/services/work_harness.py
index 4c8f8ab4..209891ce 100644
--- a/python/services/work_harness.py
+++ b/python/services/work_harness.py
@@ -1,4 +1,4 @@
-# 232:82 0:0 1:3
+# 298:82 0:0 1:3
 """Continuous A0 work harness.
 
 The conversation is the durable work identity. Model/provider selection is an
@@ -84,6 +84,7 @@
 from .agent_instance import AgentInstance
 from .energy_registry import cheap_provider
 from .model_catalog import list_models_for_user
+from .provider_failure import ProviderCallFailure, ProviderFailureText
 from .run_context import current_tool_executions
 
 
@@ -231,11 +232,35 @@ def _transient_transport_failure(exc: BaseException) -> bool:
 
 
 def _safe_to_replay(exc: BaseException, *, tool_executions: int) -> bool:
+    if isinstance(exc, ProviderCallFailure):
+        return tool_executions == 0
     if _preflight_failure(exc):
         return True
     return tool_executions == 0 and _transient_transport_failure(exc)
 
 
+def _marked_failure(content: str, usage: dict[str, Any]) -> Optional[ProviderCallFailure]:
+    if isinstance(content, ProviderFailureText):
+        return ProviderCallFailure(content, usage)
+    return None
+
+
+def _boundary_result(
+    *, provider: str, attempts: list[str], tool_executions: int,
+    reason: str, usage: Optional[dict[str, Any]] = None,
+) -> tuple[str, dict[str, Any], str]:
+    traced = dict(usage or {})
+    traced["harness"] = {
+        "mode": "continuous-auto",
+        "attempted_models": list(attempts),
+        "fallback_count": max(0, len(attempts) - 1),
+        "actual_provider": provider,
+        "tool_executions": tool_executions,
+        "status": "hmmm",
+    }
+    return f"hmmm: {reason}", traced, provider
+
+
 async def _fallback_candidates(user_id: Optional[str], seed_provider: Optional[str]) -> list[str]:
     roster = await list_models_for_user(user_id)
     rows = [
@@ -293,16 +318,29 @@ async def run_single_turn(
                 enforce_routed_tier=True,
             )
             actual_provider = primary.provider_id or seed_provider
+            marked = _marked_failure(content, usage)
+            if marked is not None and not pin_requested_provider:
+                raise marked
         except Exception as first_exc:
             if pin_requested_provider:
                 raise
             executed = current_tool_executions.get()
             if not _safe_to_replay(first_exc, tool_executions=executed):
-                if executed > 0 and _transient_transport_failure(first_exc):
-                    raise RuntimeError(
-                        "hmmm: provider failed after a tool execution boundary; "
-                        "automatic provider replay was withheld to avoid duplicate side effects"
-                    ) from first_exc
+                if executed > 0 and (
+                    isinstance(first_exc, ProviderCallFailure)
+                    or _transient_transport_failure(first_exc)
+                ):
+                    return _boundary_result(
+                        provider=getattr(first_exc, "provider", seed_provider),
+                        attempts=attempts,
+                        tool_executions=executed,
+                        reason=(
+                            "provider failure followed a tool execution boundary; automatic "
+                            "replay was withheld because the preceding tool may have mutated state. "
+                            "Conversation state is preserved for continuation."
+                        ),
+                        usage=getattr(first_exc, "usage", None),
+                    )
                 raise
 
             last_exc: BaseException = first_exc
@@ -326,19 +364,51 @@ async def run_single_turn(
                         enforce_routed_tier=True,
                     )
                     actual_provider = inst.provider_id or await inst.ensure_resolved()
+                    marked = _marked_failure(content, usage)
+                    if marked is not None:
+                        raise marked
                     break
                 except Exception as exc:
                     last_exc = exc
                     executed = current_tool_executions.get()
                     if not _safe_to_replay(exc, tool_executions=executed):
-                        if executed > 0 and _transient_transport_failure(exc):
-                            raise RuntimeError(
-                                "hmmm: fallback provider failed after a tool execution boundary; "
-                                "further replay was withheld to avoid duplicate side effects"
-                            ) from exc
+                        if executed > 0 and (
+                            isinstance(exc, ProviderCallFailure)
+                            or _transient_transport_failure(exc)
+                        ):
+                            return _boundary_result(
+                                provider=getattr(exc, "provider", inst.provider_id or fallback_model),
+                                attempts=attempts,
+                                tool_executions=executed,
+                                reason=(
+                                    "fallback provider failure followed a tool execution boundary; "
+                                    "further replay was withheld because the preceding tool may "
+                                    "have mutated state. Conversation state is preserved for continuation."
+                                ),
+                                usage=getattr(exc, "usage", None),
+                            )
                         raise
             else:
-                raise last_exc
+                if isinstance(last_exc, ProviderCallFailure):
+                    return _boundary_result(
+                        provider=last_exc.provider,
+                        attempts=attempts,
+                        tool_executions=current_tool_executions.get(),
+                        reason=(
+                            "every eligible provider failed before a tool execution boundary; "
+                            f"last sanitized provider response was {last_exc.safe_text}"
+                        ),
+                        usage=last_exc.usage,
+                    )
+                return _boundary_result(
+                    provider=seed_provider,
+                    attempts=attempts,
+                    tool_executions=current_tool_executions.get(),
+                    reason=(
+                        "every eligible provider failed before a tool execution boundary; "
+                        f"last failure type was {type(last_exc).__name__}"
+                    ),
+                )
 
         tool_executions = current_tool_executions.get()
     finally:
@@ -353,4 +423,4 @@ async def run_single_turn(
         "tool_executions": tool_executions,
     }
     return content, traced, actual_provider
-# 232:82 0:0 1:3
+# 298:82 0:0 1:3

From c06e12e662e201e0c2e13fa0ef24575fd4729bf2 Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:39:04 -0700
Subject: [PATCH 28/29] test(harness): prove typed provider failover and hmmm
 continuation

---
 tests/test_work_harness_v0.0.0alpha.py | 66 +++++++++++++++++++++-----
 1 file changed, 55 insertions(+), 11 deletions(-)

diff --git a/tests/test_work_harness_v0.0.0alpha.py b/tests/test_work_harness_v0.0.0alpha.py
index 357e5c82..685da12f 100644
--- a/tests/test_work_harness_v0.0.0alpha.py
+++ b/tests/test_work_harness_v0.0.0alpha.py
@@ -1,4 +1,4 @@
-# 167:29 0:0 0:1
+# 206:29 0:0 0:1
 """Witnesses for the continuous A0 work harness.
 
 Usage:
@@ -38,6 +38,7 @@
 import pytest
 
 from python.services import tool_executor, work_harness
+from python.services.provider_failure import ProviderFailureText, mark_provider_failure
 
 
 def _messages(n: int) -> list[dict]:
@@ -142,6 +143,46 @@ async def fake_dispatch(name: str, **kwargs):
         work_harness.current_tool_executions.reset(token)
 
 
+def test_provider_failure_marker_is_string_compatible() -> None:
+    marker = mark_provider_failure("[seed error: quota]", "seed-provider", RuntimeError("quota"))
+    assert isinstance(marker, str)
+    assert isinstance(marker, ProviderFailureText)
+    assert marker.provider == "seed-provider"
+    assert marker.error_type == "RuntimeError"
+    assert str(marker) == "[seed error: quota]"
+
+
+@pytest.mark.asyncio
+async def test_auto_mode_falls_back_on_typed_provider_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def marked_run(self, messages, **kwargs):
+        self.calls.append(self.model_id)
+        if self.model_id == "seed":
+            return (
+                mark_provider_failure(
+                    "[seed error: RateLimitError]", "seed-provider", RuntimeError("quota")
+                ),
+                {"input_tokens": 1},
+            )
+        self.provider_id = "fallback-provider"
+        return "ok", {"input_tokens": 1, "output_tokens": 1}
+
+    monkeypatch.setattr(_FakeInstance, "run", marked_run)
+    content, usage, provider = await work_harness.run_single_turn(
+        model_id="seed",
+        messages=[{"role": "user", "content": "continue"}],
+        user_id="u1",
+        system_prompt=None,
+        pin_requested_provider=False,
+        use_tools=True,
+    )
+    assert content == "ok"
+    assert provider == "fallback-provider"
+    assert _FakeInstance.calls == ["seed", "fallback"]
+    assert usage["harness"]["fallback_count"] == 1
+
+
 @pytest.mark.asyncio
 async def test_auto_mode_falls_back_on_preflight_failure() -> None:
     _FakeInstance.failures["seed"] = RuntimeError(
@@ -205,15 +246,18 @@ async def unsafe_run(self, messages, **kwargs):
         raise httpx.TimeoutException("timed out")
 
     monkeypatch.setattr(_FakeInstance, "run", unsafe_run)
-    with pytest.raises(RuntimeError, match="hmmm: provider failed after a tool execution boundary"):
-        await work_harness.run_single_turn(
-            model_id="seed",
-            messages=[{"role": "user", "content": "continue"}],
-            user_id="u1",
-            system_prompt=None,
-            pin_requested_provider=False,
-            use_tools=True,
-        )
+    content, usage, provider = await work_harness.run_single_turn(
+        model_id="seed",
+        messages=[{"role": "user", "content": "continue"}],
+        user_id="u1",
+        system_prompt=None,
+        pin_requested_provider=False,
+        use_tools=True,
+    )
+    assert content.startswith("hmmm:")
+    assert "may have mutated state" in content
+    assert provider == "seed-provider"
+    assert usage["harness"]["status"] == "hmmm"
     assert _FakeInstance.calls == ["seed"]
     assert work_harness.current_tool_executions.get() == 0
 
@@ -232,4 +276,4 @@ async def test_tool_free_transient_failure_can_fall_back() -> None:
     assert content == "ok"
     assert provider == "fallback-provider"
     assert usage["harness"]["fallback_count"] == 1
-# 167:29 0:0 0:1
+# 206:29 0:0 0:1

From aa0ed96108f6858d6a60b81fabb863f63c9c930a Mon Sep 17 00:00:00 2001
From: Erin Spencer <erin.eps.hovel@gmail.com>
Date: Wed, 23 Sep 2026 11:39:45 -0700
Subject: [PATCH 29/29] fix(harness): require zero tool executions for replay

---
 python/services/work_harness.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/python/services/work_harness.py b/python/services/work_harness.py
index 209891ce..856b1411 100644
--- a/python/services/work_harness.py
+++ b/python/services/work_harness.py
@@ -232,11 +232,11 @@ def _transient_transport_failure(exc: BaseException) -> bool:
 
 
 def _safe_to_replay(exc: BaseException, *, tool_executions: int) -> bool:
-    if isinstance(exc, ProviderCallFailure):
-        return tool_executions == 0
-    if _preflight_failure(exc):
+    if tool_executions != 0:
+        return False
+    if isinstance(exc, ProviderCallFailure) or _preflight_failure(exc):
         return True
-    return tool_executions == 0 and _transient_transport_failure(exc)
+    return _transient_transport_failure(exc)
 
 
 def _marked_failure(content: str, usage: dict[str, Any]) -> Optional[ProviderCallFailure]:
@@ -377,7 +377,7 @@ async def run_single_turn(
                             or _transient_transport_failure(exc)
                         ):
                             return _boundary_result(
-                                provider=getattr(exc, "provider", inst.provider_id or fallback_model),
+                                provider=getattr(exc, "provider", fallback_model),
                                 attempts=attempts,
                                 tool_executions=executed,
                                 reason=(