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 diff --git a/python/models.py b/python/models.py index a2d23e5e..39cbf671 100644 --- a/python/models.py +++ b/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 diff --git a/python/routes/chat.py b/python/routes/chat.py index 814202f8..33141792 100644 --- a/python/routes/chat.py +++ b/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 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 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 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 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 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 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 diff --git a/python/services/work_harness.py b/python/services/work_harness.py new file mode 100644 index 00000000..856b1411 --- /dev/null +++ b/python/services/work_harness.py @@ -0,0 +1,426 @@ +# 298: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: write +# 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 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: write +# 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 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 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 +from .provider_failure import ProviderCallFailure, ProviderFailureText +from .run_context import current_tool_executions + + +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, *, tool_executions: int) -> bool: + if tool_executions != 0: + return False + if isinstance(exc, ProviderCallFailure) or _preflight_failure(exc): + return True + return _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 = [ + 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 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] + 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() + tool_token = current_tool_executions.set(0) + try: + 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 + 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 ( + 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 + 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() + 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 ( + isinstance(exc, ProviderCallFailure) + or _transient_transport_failure(exc) + ): + return _boundary_result( + provider=getattr(exc, "provider", 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: + 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: + 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 +# 298:82 0:0 1:3 diff --git a/python/storage/core.py b/python/storage/core.py index 17faf94d..5bdf0660 100644 --- a/python/storage/core.py +++ b/python/storage/core.py @@ -1,4 +1,4 @@ -# 393:53 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_ @@ -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)) @@ -504,4 +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:57 0:0 4:2 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 diff --git a/shared/schema.ts b/shared/schema.ts index 220d28b0..1a8d211f 100644 --- a/shared/schema.ts +++ b/shared/schema.ts @@ -1,4 +1,4 @@ -// 697:26 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"; @@ -20,6 +20,7 @@ export const conversations = pgTable("conversations", { enabledTools: jsonb("enabled_tools").$type(), maxToolRounds: integer("max_tool_rounds"), inferenceMode: varchar("inference_mode", { length: 20 }).default("agentic"), + harnessState: jsonb("harness_state").$type | null>(), createdAt: timestamp("created_at").default(sql`CURRENT_TIMESTAMP`).notNull(), updatedAt: timestamp("updated_at").default(sql`CURRENT_TIMESTAMP`).notNull(), }, (t) => [ @@ -822,4 +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:26 0:141 0:0 diff --git a/tests/test_open_comp_rout_v0.0.0alpha.py b/tests/test_open_comp_rout_v0.0.0alpha.py index c6f2bccd..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 @@ -# 399:1 0:0 0:0 +# 396:1 0:0 0:0 """Routing and catalog tests for registry-driven compatible providers.""" import ast @@ -69,17 +69,14 @@ def test_approval_replays_preserve_explicit_provider_pin() -> None: for call in replay_calls ) - instance_runs = [ - node - for node in ast.walk(tree) - if isinstance(node, ast.Call) - and isinstance(node.func, ast.Attribute) - and node.func.attr == "run" + harness_calls = [ + node for node in ast.walk(tree) + if isinstance(node, ast.Call) and getattr(node.func, "id", None) == "run_single_turn" ] - assert any( - any(keyword.arg == "pin_requested_provider" for keyword in call.keywords) - for call in instance_runs - ) + assert len(harness_calls) == 1 + 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 @@ -508,4 +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 +# 396:1 0:0 0:0 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..685da12f --- /dev/null +++ b/tests/test_work_harness_v0.0.0alpha.py @@ -0,0 +1,279 @@ +# 206: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 tool_executor, work_harness +from python.services.provider_failure import ProviderFailureText, mark_provider_failure + + +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_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) + + +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( + "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_agentic_transient_failure_before_tools_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=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) + 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 + + +@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 +# 206:29 0:0 0:1