From 57bae2e6ae6a3aaaf3a720dab6b59c198cedc23b Mon Sep 17 00:00:00 2001 From: m4 Date: Sun, 13 Sep 2026 15:12:17 +0800 Subject: [PATCH] feat(llm): scoped model runtime, history rebuild and gateway proxy streaming --- EvoScientist/llm/adapter_registry.py | 2 + EvoScientist/llm/contracts.py | 54 +- EvoScientist/llm/gateway_proxy.py | 25 +- EvoScientist/llm/gemini_interactions.py | 155 ++++- EvoScientist/llm/history_initialization.py | 112 ++++ EvoScientist/llm/history_rebuild.py | 244 +++++++ EvoScientist/llm/host_execution_registry.py | 370 +++++++++++ EvoScientist/llm/invocation/contract.py | 3 +- EvoScientist/llm/model_config.py | 52 +- EvoScientist/llm/models.py | 2 +- EvoScientist/llm/runtime.py | 691 +++++++++++++++++++- 11 files changed, 1656 insertions(+), 54 deletions(-) create mode 100644 EvoScientist/llm/history_initialization.py create mode 100644 EvoScientist/llm/history_rebuild.py create mode 100644 EvoScientist/llm/host_execution_registry.py diff --git a/EvoScientist/llm/adapter_registry.py b/EvoScientist/llm/adapter_registry.py index 895d58b..eb8ebcb 100644 --- a/EvoScientist/llm/adapter_registry.py +++ b/EvoScientist/llm/adapter_registry.py @@ -398,6 +398,8 @@ class AdapterRegistration: ) if reasoning not in {None, "off"}: result["reasoning"] = {"effort": reasoning} + if self.adapter_id in {"openai", "xai"}: + result["reasoning"]["summary"] = "auto" elif self.adapter_id == "dashscope" and thinking: result["reasoning"] = {"effort": "medium"} elif self.adapter_id == "google-gemini": diff --git a/EvoScientist/llm/contracts.py b/EvoScientist/llm/contracts.py index 7157227..020466c 100644 --- a/EvoScientist/llm/contracts.py +++ b/EvoScientist/llm/contracts.py @@ -68,6 +68,14 @@ def now_ms() -> int: def _unsigned(value: Any) -> dict[str, Any]: payload = asdict(value) payload.pop("signature", None) + if payload.get("predecessor_owner_epoch") == 0: + payload.pop("predecessor_owner_epoch") + for name in ("continuation_pending_hash", "continuation_decision_hash"): + if payload.get(name) == "": + payload.pop(name) + for name in ("execution_id", "predecessor_execution_id", "predecessor_checkpoint_id"): + if payload.get(name) == "": + payload.pop(name) return payload @@ -95,6 +103,11 @@ class RoutePreparationGrant: key_id: str signature: str contract_version: int = CONTRACT_VERSION + predecessor_execution_id: str = "" + predecessor_checkpoint_id: str = "" + predecessor_owner_epoch: int = 0 + continuation_pending_hash: str = "" + continuation_decision_hash: str = "" def unsigned_payload(self) -> dict[str, Any]: return _unsigned(self) @@ -195,6 +208,7 @@ class PreparedRunQuote: key_id: str signature: str contract_version: int = CONTRACT_VERSION + execution_id: str = "" def unsigned_payload(self) -> dict[str, Any]: return _unsigned(self) @@ -622,11 +636,7 @@ class HmacGrantAuthority: payload = {**defaults, **kwargs} if "roles" in payload: payload["roles"] = tuple(sorted({str(role) for role in payload["roles"]})) - unsigned = { - key_name: value - for key_name, value in payload.items() - if key_name != "signature" - } + unsigned = _unsigned(contract_class(signature="", **payload)) signature = sign_contract(contract_class.__name__, unsigned, key) return contract_class(signature=signature, **payload) @@ -766,6 +776,22 @@ class AgentExecutionProfile: return cls.web_v3() +@dataclass(frozen=True, slots=True) +class ModelFactoryResult: + """Explicit transfer of exclusively owned closeable resources to one run. + + Plain custom factory return values are borrowed. Do not list shared clients. + The runtime closes only these exact objects, never their reachable children. + Built-in factories register newly allocated transports during construction. + Cleanup tries every resource even after failures, at most three times per + resource. Failed resources and their exception history remain owned; no + successful terminal event is emitted until all resources close. + """ + + model: Any + owned_clients: tuple[Any, ...] = () + + @dataclass(frozen=True, slots=True) class AgentModelSet: main_agent: Any @@ -864,7 +890,23 @@ class EvoWebRun(Protocol): self, after_sequence: int | None = None ) -> AsyncIterator[EvoRuntimeEvent]: ... - async def cancel(self, reason: str) -> str: ... + async def cancel(self, reason: str, *, owner_epoch: int | None = None, + boot_id: str | None = None) -> str: ... + + async def wait_stopped(self, timeout: float | None = 2.0) -> str: + """Return a terminal outcome only after owned resource cleanup. + + A bounded timeout returns ``unknown`` without cancelling execution or + cleanup. Cancelling the waiter also leaves cleanup owned by the run. + Host checkpointers and workspace backends are borrowed, never closed. + ``awaiting_input`` ends this execution, not the checkpoint workflow; + a separate run may resume it. Its run_terminal payload includes the + final checkpoint_thread_id, checkpoint_id, checkpoint_ns and JSON + pending_interrupts (id/value records). Consumers must not interpret + every non-failed outcome as completed. Final checkpoint read failure + yields failed with FINAL_CHECKPOINT_READ_FAILED, never completed. + """ + ... def event_payload(value: Any) -> dict[str, Any]: diff --git a/EvoScientist/llm/gateway_proxy.py b/EvoScientist/llm/gateway_proxy.py index dbd42ed..be2ebd6 100644 --- a/EvoScientist/llm/gateway_proxy.py +++ b/EvoScientist/llm/gateway_proxy.py @@ -174,7 +174,30 @@ def _chunk_to_message(chunk: dict[str, Any]) -> BaseMessage: parsed = messages_from_dict([message_dict]) if len(parsed) != 1: raise RuntimeError("AI4SCI_MODEL_STREAM_DELTA_INVALID") - return parsed[0] + message = parsed[0] + summary = delta.get("reasoning_summary") + if isinstance(summary, str) and summary: + content = list(message.content) if isinstance(message.content, list) else [] + existing_summary = "".join( + str(part.get("text") or "") + for block in content + if isinstance(block, Mapping) and block.get("type") == "reasoning" + for part in (block.get("summary") or []) + if isinstance(part, Mapping) and part.get("type") == "summary_text" + ) + if summary.startswith(existing_summary): + summary_delta = summary[len(existing_summary):] + elif existing_summary.startswith(summary): + summary_delta = "" + else: + summary_delta = summary + if summary_delta: + content.append({ + "type": "reasoning", + "summary": [{"type": "summary_text", "text": summary_delta}], + }) + message.content = content + return message def proxy_from_config( diff --git a/EvoScientist/llm/gemini_interactions.py b/EvoScientist/llm/gemini_interactions.py index 7ed85c1..c819fd1 100644 --- a/EvoScientist/llm/gemini_interactions.py +++ b/EvoScientist/llm/gemini_interactions.py @@ -3,7 +3,9 @@ from __future__ import annotations import json +import logging from collections.abc import AsyncIterator, Mapping, Sequence +from types import SimpleNamespace from typing import Any from langchain_core.language_models.chat_models import BaseChatModel @@ -20,6 +22,60 @@ from langchain_core.tools import BaseTool from langchain_core.utils.function_calling import convert_to_openai_tool from pydantic import Field, SecretStr +logger = logging.getLogger(__name__) + + +def _cleanup_outcome( + primary: BaseException | None, errors: list[BaseException] +) -> None: + if primary is not None or not errors: + return + for error in errors: + if not isinstance(error, Exception): + raise error + raise RuntimeError("MODEL_PROVIDER_CLEANUP_ERROR") from None + + +def _record_cleanup_error(resource: str) -> None: + # Never log exception messages, reprs or tracebacks containing provider secrets. + try: + logger.warning("MODEL_PROVIDER_CLEANUP_ERROR resource=%s", resource) + except Exception: + # A broken logging handler must not replace the provider exception either. + pass + + +def _close_owned(client: Any, stream: Any, primary: BaseException | None) -> None: + errors: list[BaseException] = [] + for resource, target in (("stream", stream), ("client", client)): + if target is None: + continue + try: + target.close() + except BaseException as error: + errors.append(error) + _record_cleanup_error(resource) + _cleanup_outcome(primary, errors) + + +async def _aclose_owned( + client: Any, stream: Any, primary: BaseException | None +) -> None: + errors: list[BaseException] = [] + for resource in ("stream", "async_client", "client"): + try: + if resource == "stream": + if stream is not None: + await stream.close() + elif resource == "async_client": + await client.aio.aclose() + else: + client.close() + except BaseException as error: + errors.append(error) + _record_cleanup_error(resource) + _cleanup_outcome(primary, errors) + class GeminiInteractionsChatModel(BaseChatModel): """Minimal native bridge that preserves signed Provider content blocks.""" @@ -96,7 +152,7 @@ class GeminiInteractionsChatModel(BaseChatModel): "generation_config": generation_config, "tools": list(self.bound_tools), "store": False, - "stream": False, + "stream": True, } def _generate( @@ -108,8 +164,20 @@ class GeminiInteractionsChatModel(BaseChatModel): request = self._request(messages) if stop: request["generation_config"]["stop_sequences"] = stop - response = self._client().interactions.create(**request) - return _chat_result(response) + client = self._client() + stream = None + state = _InteractionAccumulator() + primary = None + try: + stream = client.interactions.create(**request) + for event in stream: + state.accept(_dump(event)) + return state.result() + except BaseException as error: + primary = error + raise + finally: + _close_owned(client, stream, primary) async def _agenerate( self, @@ -120,8 +188,20 @@ class GeminiInteractionsChatModel(BaseChatModel): request = self._request(messages) if stop: request["generation_config"]["stop_sequences"] = stop - response = await self._client().aio.interactions.create(**request) - return _chat_result(response) + client = self._client() + stream = None + state = _InteractionAccumulator() + primary = None + try: + stream = await client.aio.interactions.create(**request) + async for event in stream: + state.accept(_dump(event)) + return state.result() + except BaseException as error: + primary = error + raise + finally: + await _aclose_owned(client, stream, primary) async def _astream( self, @@ -133,8 +213,22 @@ class GeminiInteractionsChatModel(BaseChatModel): request["stream"] = True if stop: request["generation_config"]["stop_sequences"] = stop - stream = await self._client().aio.interactions.create(**request) + client = self._client() + stream = None + primary = None + try: + stream = await client.aio.interactions.create(**request) + async for chunk in self._astream_events(stream): + yield chunk + except BaseException as error: + primary = error + raise + finally: + await _aclose_owned(client, stream, primary) + + async def _astream_events(self, stream: Any) -> AsyncIterator[ChatGenerationChunk]: blocks: dict[int, dict[str, Any]] = {} + completed = False async for event in stream: payload = _dump(event) event_type = payload.get("event_type") @@ -175,6 +269,7 @@ class GeminiInteractionsChatModel(BaseChatModel): if event_type == "error": raise RuntimeError("MODEL_PROVIDER_PROTOCOL_ERROR") if event_type == "interaction.complete": + completed = True interaction = payload.get("interaction") or {} ordered_blocks = [blocks[index] for index in sorted(blocks)] yield ChatGenerationChunk( @@ -188,9 +283,7 @@ class GeminiInteractionsChatModel(BaseChatModel): provider_request_id=interaction.get("id"), ), response_metadata={ - "model_name": str( - (interaction.get("model") or {}).get("id") or "" - ), + "model_name": _interaction_model_name(interaction), "finish_reason": str( interaction.get("status") or "unknown" ), @@ -198,6 +291,50 @@ class GeminiInteractionsChatModel(BaseChatModel): ) ) + if not completed: + raise RuntimeError("MODEL_PROVIDER_PROTOCOL_ERROR") + + +class _InteractionAccumulator: + def __init__(self) -> None: + self.blocks: dict[int, dict[str, Any]] = {} + self.interaction: dict[str, Any] | None = None + + def accept(self, payload: dict[str, Any]) -> None: + kind = payload.get("event_type") + if kind == "content.start": + self.blocks[int(payload["index"])] = dict(payload.get("content") or {}) + elif kind == "content.delta": + _merge_stream_delta( + self.blocks.setdefault(int(payload["index"]), {}), + dict(payload.get("delta") or {}), + ) + elif kind == "error": + raise RuntimeError("MODEL_PROVIDER_PROTOCOL_ERROR") + elif kind == "interaction.complete": + self.interaction = payload.get("interaction") or {} + + def result(self) -> ChatResult: + if self.interaction is None: + raise RuntimeError("MODEL_PROVIDER_PROTOCOL_ERROR") + interaction = self.interaction + return _chat_result( + SimpleNamespace( + outputs=[self.blocks[index] for index in sorted(self.blocks)], + usage=interaction.get("usage"), + id=interaction.get("id"), + status=interaction.get("status", "unknown"), + model=SimpleNamespace(id=_interaction_model_name(interaction)), + ) + ) + + +def _interaction_model_name(interaction: Mapping[str, Any]) -> str: + model = interaction.get("model") + if isinstance(model, Mapping): + return str(model.get("id") or "") + return str(model or "") + def _message_content(message: BaseMessage) -> list[dict[str, Any]]: if isinstance(message, AIMessage): diff --git a/EvoScientist/llm/history_initialization.py b/EvoScientist/llm/history_initialization.py new file mode 100644 index 0000000..b2a6ccf --- /dev/null +++ b/EvoScientist/llm/history_initialization.py @@ -0,0 +1,112 @@ +"""Local durable initialization gate. All graph access must use this gate. + +The file lock covers graph writes as well as owner changes; a row CAS alone +cannot fence an old writer while it is suspended in a graph await. +""" +import asyncio +import fcntl +import sqlite3 +from contextlib import asynccontextmanager, closing +from pathlib import Path + +from langgraph.graph import START, END + + +class SqliteInitializationStore: + def __init__(self, path): + if str(path) == ":memory:": + raise ValueError("initialization requires a durable file") + self.path = str(Path(path).resolve()) + with closing(self.connect()) as db, db: + db.execute("""CREATE TABLE IF NOT EXISTS history_initialization ( + key TEXT PRIMARY KEY, attempt TEXT NOT NULL, digest TEXT NOT NULL, + owner TEXT NOT NULL, fence INTEGER NOT NULL, status TEXT NOT NULL)""") + + def connect(self): + db = sqlite3.connect(self.path) + db.row_factory = sqlite3.Row + return db + + def get(self, key): + with closing(self.connect()) as db: + row = db.execute("SELECT * FROM history_initialization WHERE key=?", (key,)).fetchone() + return dict(row) if row else None + + @asynccontextmanager + async def locked(self, key): + with open(self.path + "." + key.split(":")[-1] + ".lock", "a") as lock: + while True: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except BlockingIOError: + await asyncio.sleep(0.01) + try: + yield + finally: + fcntl.flock(lock, fcntl.LOCK_UN) + + def claim(self, key, attempt, digest, owner, fence): + if not isinstance(attempt, str) or not attempt.strip() or not isinstance(owner, str) or not owner.strip(): + raise ValueError("attempt and owner required") + if type(fence) is not int or fence < 1: + raise ValueError("positive owner fence required") + with closing(self.connect()) as db, db: + db.execute("BEGIN IMMEDIATE") + row = db.execute("SELECT * FROM history_initialization WHERE key=?", (key,)).fetchone() + if row: + if row["attempt"] != attempt or row["digest"] != digest: + raise ValueError("history attempt/content conflict") + if fence < row["fence"] or (fence == row["fence"] and owner != row["owner"]): + raise ValueError("stale history owner fence") + if row["status"] not in ("INITIALIZING", "READY"): + raise ValueError("history checkpoint already consumed") + db.execute("UPDATE history_initialization SET owner=?, fence=? WHERE key=?", (owner, fence, key)) + else: + db.execute("INSERT INTO history_initialization VALUES (?,?,?,?,?,?)", + (key, attempt, digest, owner, fence, "INITIALIZING")) + + def status(self, key, attempt, owner, fence, before, after): + with closing(self.connect()) as db, db: + changed = db.execute("""UPDATE history_initialization SET status=? + WHERE key=? AND attempt=? AND owner=? AND fence=? AND status=?""", + (after, key, attempt, owner, fence, before)).rowcount + if changed != 1: + raise ValueError("history requires READY and current owner fence") + + +async def initialize(graph, key, messages, digest, store, attempt, owner, fence): + config = {"configurable": {"thread_id": key}} + marker = {"history_attempt": attempt, "history_digest": digest} + async with store.locked(key): + current = await graph.aget_state(config) + if store.get(key) is None and current.created_at is not None: + raise ValueError("unowned history checkpoint already exists") + store.claim(key, attempt, digest, owner, fence) + if current.created_at is None: + await graph.aupdate_state(dict(config, metadata=dict(marker, history_stage="START")), + {"messages": list(messages)}, as_node=START) + current = await graph.aget_state(config) + metadata = current.metadata or {} + if any(metadata.get(k) != v for k, v in marker.items()): + raise ValueError("unowned history checkpoint") + if current.values.get("messages", []) != list(messages): + raise ValueError("history checkpoint content changed") + stage = metadata.get("history_stage") + if stage == "START": + await graph.aupdate_state(dict(config, metadata=dict(marker, history_stage="END")), + None, as_node=END) + current = await graph.aget_state(config) + elif stage != "END": + raise ValueError("unknown history initialization stage") + if current.next or current.tasks: + raise ValueError("history checkpoint is not READY") + store.status(key, attempt, owner, fence, store.get(key)["status"], "READY") + return config + + +async def invoke(graph, key, store, attempt, owner, fence, input): + async with store.locked(key): + # Consume before execution: uncertain invocation must not be replayed. + store.status(key, attempt, owner, fence, "READY", "CONSUMED") + return await graph.ainvoke(input, {"configurable": {"thread_id": key}}) \ No newline at end of file diff --git a/EvoScientist/llm/history_rebuild.py b/EvoScientist/llm/history_rebuild.py new file mode 100644 index 0000000..3fb084d --- /dev/null +++ b/EvoScientist/llm/history_rebuild.py @@ -0,0 +1,244 @@ +"""Committed history to a fresh turn checkpoint, never an old stack resume. + +The host supplies authorized, revision-consistent records and owns the turn +fence across creation and invocation. This module does not read or modify PG. +""" +from __future__ import annotations + +import copy +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Sequence + +from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage +from langchain_core.runnables import RunnableConfig +from langgraph.graph import END, START + +from .patches import _sanitize_openai_tool_history, _validate_openai_tool_history +from .history_initialization import SqliteInitializationStore, initialize, invoke + + +@dataclass(frozen=True) +class HistoryScope: + tenant_id: str + thread_id: str + turn_id: str + workspace_id: str + graph_version: str + tool_version: str + history_revision: int + + def validate(self) -> None: + for value in (self.tenant_id, self.thread_id, self.turn_id, + self.workspace_id, self.graph_version, self.tool_version): + if not isinstance(value, str) or not value.strip(): + raise ValueError("history scope fields must be nonempty strings") + if type(self.history_revision) is not int or self.history_revision < 0: + raise ValueError("invalid history revision") + + +@dataclass(frozen=True) +class HistoryRecord: + message_id: str + revision: int + message: BaseMessage + partial: bool = False + file_refs: tuple[str, ...] = () + + +@dataclass(frozen=True) +class NormalizedHistory: + scope: HistoryScope + records: tuple[HistoryRecord, ...] + messages: tuple[BaseMessage, ...] + + +def _check_scope(source: HistoryScope, expected: HistoryScope) -> None: + source.validate() + expected.validate() + if source != expected: + raise ValueError("history scope/version/revision mismatch") + + +def _text(content: Any) -> str: + # Only text is replayed. Media bytes stay outside model history; authorized + # file references are supplied separately by the host's record adapter. + if isinstance(content, str): + return "[inline media omitted]" if "base64," in content or "data:" in content else content + if isinstance(content, list): + return "\n".join(_text(b.get("text", "")) for b in content + if isinstance(b, dict) and b.get("type") == "text") + return "" + + +def normalize_history(records: Sequence[HistoryRecord], *, source: HistoryScope, + expected: HistoryScope) -> NormalizedHistory: + """Pure normalization of already-authorized stable-ID committed records. + + Complete parallel tool pairs survive; unmatched calls become sourced facts, + never invented ToolMessages. Raw messages and provider metadata are untouched. + """ + _check_scope(source, expected) + seen: set[str] = set() + messages = [] + notes = [] + for record in records: + if (not isinstance(record.message_id, str) or not record.message_id.strip() + or record.message_id in seen or type(record.revision) is not int + or record.revision < 0 or not isinstance(record.message, BaseMessage)): + raise ValueError("invalid or duplicate history record") + seen.add(record.message_id) + message = copy.deepcopy(record.message) + message.id = record.message_id + messages.append(message) + calls = {} + results = set() + for message in messages: + for call in getattr(message, "tool_calls", []): + call_id = call.get("id") + if not call_id or call_id in calls: + raise ValueError("missing or duplicate tool call ID") + calls[call_id] = call.get("name") + if isinstance(message, ToolMessage): + call_id = message.tool_call_id + if not call_id or call_id not in calls or call_id in results: + raise ValueError("unassociated or duplicate tool result") + if message.name is not None and message.name != calls[call_id]: + raise ValueError("tool result name conflicts with call ID") + results.add(call_id) + repaired = _sanitize_openai_tool_history(messages) + retained = {m.id: m for m in repaired} + for record, original in zip(records, messages): + repaired_message = retained.get(record.message_id) + kept = {c.get("id") for c in getattr(repaired_message, "tool_calls", [])} + missing = [c.get("id") or "unidentified" for c in getattr(original, "tool_calls", []) + if c.get("id") not in kept] + provenance = f"source={record.message_id}@{record.revision}" + if missing or getattr(original, "invalid_tool_calls", []): + notes.append(f"[{provenance}] incomplete tool calls (result unavailable; execution unknown): {missing}") + if record.partial: + notes.append(f"[{provenance}] partial result, not a completed answer") + for ref in record.file_refs: + if not isinstance(ref, str) or not ref.strip() or "data:" in ref or "base64" in ref: + raise ValueError("invalid file reference") + notes.append(f"[{provenance}] file reference: {ref}") + for message in repaired: + message.content = _text(message.content) + message.additional_kwargs = {} + message.response_metadata = {} + if hasattr(message, "artifact"): + message.artifact = None + # Tool arguments are historical context, not executable input. Reject + # embedded binary payload rather than stringify it into provider text. + if "base64" in json.dumps(getattr(message, "tool_calls", [])): + raise ValueError("inline binary tool arguments are not replayable") + if notes: + repaired.append(HumanMessage(content="Historical context notes:\n" + "\n".join(notes), + id="history-notes:" + str(source.history_revision))) + _validate_openai_tool_history(repaired) + return NormalizedHistory(source, tuple(copy.deepcopy(records)), tuple(repaired)) + + +def committed_history_input(history: dict, current: dict, *, thread_id: str, + run_id: str, checkpoint_exists: bool = False) -> dict: + """Project PG display records into the existing worker's message channel. + + Display tool items are sourced facts, not provider tool protocol. The + reducer sentinel is only used when the host attests no compatible checkpoint + exists. Later turns append without resetting the summarizer's cut indexes. + No second invocation or checkpoint mutation is performed by the HTTP host. + """ + from langchain_core.messages import AIMessage, RemoveMessage, convert_to_messages + from langgraph.graph.message import REMOVE_ALL_MESSAGES + + if (history.get("schema") != "ai4sci.committed-history.v1" + or history.get("thread_id") != thread_id + or not isinstance(history.get("records"), list) + or not isinstance(current, dict) + or not isinstance(current.get("messages"), list)): + raise ValueError("invalid committed history input") + scope = HistoryScope( + tenant_id=history["user_uid"], thread_id=thread_id, turn_id=run_id, + workspace_id=thread_id, graph_version="worker-input-v1", + tool_version="display-facts-v2", history_revision=history["conversation_revision"], + ) + records = [] + for row in sorted(history["records"], key=lambda r: (r["message_index"], r["message_id"])): + if row["message_id"] == history.get("excluded_message_id"): + continue + payload = row["payload"] + role = row["role"] + if role not in {"user", "assistant"}: + raise ValueError("unsupported committed history role") + text = [] + refs = [] + if role == "user": + text.append(_text(payload.get("content", ""))) + for attachment in payload.get("attached_files") or []: + if isinstance(attachment, dict): + ref = attachment.get("virtual_path") or attachment.get("path") + if isinstance(ref, str) and ref.startswith("/workspace/"): + refs.append(ref) + else: + for item in sorted(payload.get("items") or [], key=lambda i: i["item_sequence"]): + if item.get("status") == "superseded": + continue + kind = item.get("type") + source = f"[{row['message_id']}@{row['revision']}:{item['item_id']}]" + if kind == "message": + text.extend(part["text"] for part in item.get("content", []) + if part.get("type") in {"output_text", "refusal"}) + elif kind in {"tool_call", "tool_output", "agent_status", "summarization"}: + fields = {key: item[key] for key in + ("name", "input", "output", "result", "summary", "status") if key in item} + text.append(f"{source} historical {kind}: " + json.dumps(fields, ensure_ascii=False)) + elif kind == "artifact" and str(item.get("virtual_path", "")).startswith("/workspace/"): + refs.append(item["virtual_path"]) + message = (HumanMessage if role == "user" else AIMessage)(content="\n".join(text)) + records.append(HistoryRecord(row["message_id"], row["revision"], message, + partial=bool(payload.get("incomplete")), file_refs=tuple(refs))) + normalized = normalize_history(records, source=scope, expected=scope) + incoming = convert_to_messages(copy.deepcopy(current["messages"])) + for index, message in enumerate(incoming): + # Stable IDs ensure replay cannot duplicate the current turn either. + message.id = f"current:{run_id}:{index}" + if checkpoint_exists: + return {**current, "messages": [m.model_dump(mode="json") for m in incoming]} + messages = [RemoveMessage(id=REMOVE_ALL_MESSAGES), *normalized.messages, *incoming] + # Old summary cut indexes refer to the replaced checkpoint message list. + # The existing summarizer/budget middleware recomputes them for this input. + return {**current, "messages": [m.model_dump(mode="json") for m in messages], + "_summarization_event": None} + + +def history_key(scope: HistoryScope) -> str: + scope.validate() + identity = json.dumps(list(vars(scope).values()), separators=(",", ":")) + return "history-v1:" + hashlib.sha256(identity.encode()).hexdigest() + + +async def invoke_history_checkpoint(graph, *, scope, store, attempt, owner, fence, input): + return await invoke(graph, history_key(scope), store, attempt, owner, fence, input) + + +async def create_history_checkpoint(graph: Any, history: NormalizedHistory, *, + expected: HistoryScope, store=None, attempt=None, + owner=None, fence=None) -> RunnableConfig: + """Seed through a durable gate; dispatch only via invoke_history_checkpoint. + + A matching attempt/content may recover initialization, never execution. + The host owns authorization and issues monotonically increasing fences. + """ + _check_scope(history.scope, expected) + normalized = normalize_history(history.records, source=history.scope, expected=expected) + key = history_key(expected) + if store is not None: + payload = [{"id": r.message_id, "revision": r.revision, + "message": r.message.model_dump(mode="json"), + "partial": r.partial, "files": r.file_refs} for r in history.records] + digest = hashlib.sha256(json.dumps(payload, sort_keys=True, + separators=(",", ":")).encode()).hexdigest() + return await initialize(graph, key, normalized.messages, digest, + store, attempt, owner, fence) + raise ValueError("durable initialization store is required") \ No newline at end of file diff --git a/EvoScientist/llm/host_execution_registry.py b/EvoScientist/llm/host_execution_registry.py new file mode 100644 index 0000000..8bb6b06 --- /dev/null +++ b/EvoScientist/llm/host_execution_registry.py @@ -0,0 +1,370 @@ +"""Opt-in host identity prototype. Records never prove resource quiescence. + +Trusted local host API, not an authenticated remote control endpoint. A future +PG host store can implement this protocol without introducing a dispatcher. +Only the runtime resource owner may attest cleanup and release its claim. +""" +from contextlib import closing, contextmanager +from pathlib import Path +import sqlite3 +import json +import hashlib +import os +from typing import Protocol + +from .contracts import EvoRuntimeError, canonical_json_v1 + + +class HostExecutionRegistry(Protocol): + def claim_checkpoint(self, execution_id: str, *, store_id: str, checkpoint_thread_id: str, subject_id: str) -> None: ... + def release_unbound_checkpoint(self, execution_id: str) -> None: ... + def continuation(self, execution_id: str) -> dict: ... + def prepare_terminal(self, execution_id: str, *, event: dict) -> None: ... + def terminal_intent(self, execution_id: str) -> dict | None: ... + def confirm_terminal(self, execution_id: str, *, digest: str) -> None: ... + def finish(self, execution_id: str, *, outcome: str, checkpoint_id: str = "") -> None: ... + def bind(self, *, execution_id: str, grant_id: str, digest: str, + thread_id: str, turn_id: str, predecessor_execution_id: str = "", + predecessor_checkpoint_id: str = "", predecessor_owner_epoch: int = 0, + continuation_pending_hash: str = "", continuation_decision_hash: str = "") -> None: ... + def lookup_grant(self, grant_id: str, digest: str) -> dict | None: ... + def inspect(self, execution_id: str) -> dict: ... + def transfer_control(self, execution_id: str, *, expected_epoch: int, new_epoch: int) -> int: ... + def require_control(self, execution_id: str, *, owner_epoch: int) -> None: ... + def inspect_control(self, execution_id: str, *, owner_epoch: int | None, boot_id: str | None) -> dict: ... + def accept_cancel(self, execution_id: str, *, owner_epoch: int | None, boot_id: str | None, reason: str) -> int: ... + + +class SQLiteHostRegistry: + def __init__(self, path: str | Path, *, host_id: str, boot_id: str): + self.path, self.host_id, self.boot_id = str(path), host_id, boot_id + directory = Path(path).parent + directory.mkdir(mode=0o700, parents=True, exist_ok=True) + # Exclusive creation gives new user-content stores safe defaults without + # changing permissions on an existing host's directory or database. + try: + fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + pass + else: + os.close(fd) + with self._transaction() as db: + db.execute("BEGIN IMMEDIATE") + db.execute("""CREATE TABLE IF NOT EXISTS executions ( + execution_id TEXT PRIMARY KEY, grant_id TEXT UNIQUE NOT NULL, + digest TEXT NOT NULL, thread_id TEXT NOT NULL, + turn_id TEXT NOT NULL, host_id TEXT NOT NULL, boot_id TEXT NOT NULL, + owner_epoch INTEGER NOT NULL DEFAULT 1)""") + schema = db.execute("SELECT sql FROM sqlite_master WHERE name='executions'").fetchone()[0] + if "UNIQUE(thread_id, turn_id)" in schema: + db.execute("ALTER TABLE executions RENAME TO legacy_executions") + db.execute("""CREATE TABLE executions ( + execution_id TEXT PRIMARY KEY, grant_id TEXT UNIQUE NOT NULL, + digest TEXT NOT NULL, thread_id TEXT NOT NULL, turn_id TEXT NOT NULL, + host_id TEXT NOT NULL, boot_id TEXT NOT NULL, + owner_epoch INTEGER NOT NULL DEFAULT 1)""") + db.execute("INSERT INTO executions SELECT * FROM legacy_executions") + db.execute("DROP TABLE legacy_executions") + db.execute("""CREATE TABLE IF NOT EXISTS cancel_intents ( + intent_id INTEGER PRIMARY KEY AUTOINCREMENT, + execution_id TEXT NOT NULL, owner_epoch INTEGER NOT NULL, + boot_id TEXT NOT NULL, reason TEXT NOT NULL)""") + db.execute("""CREATE TABLE IF NOT EXISTS terminal_evidence ( + execution_id TEXT PRIMARY KEY, outcome TEXT NOT NULL)""") + db.execute("""CREATE TABLE IF NOT EXISTS terminal_intents ( + execution_id TEXT PRIMARY KEY, event_json TEXT NOT NULL, + digest TEXT NOT NULL, phase TEXT NOT NULL, + cleanup_boot_id TEXT NOT NULL)""") + db.execute("""CREATE TABLE IF NOT EXISTS pending_continuations ( + execution_id TEXT PRIMARY KEY, checkpoint_id TEXT NOT NULL, + consumed_by TEXT UNIQUE)""") + columns = {row[1] for row in db.execute("PRAGMA table_info(pending_continuations)")} + for name in ("pending_hash", "decision_hash"): + if name not in columns: + db.execute(f"ALTER TABLE pending_continuations ADD COLUMN {name} TEXT NOT NULL DEFAULT ''") + db.execute("""CREATE TABLE IF NOT EXISTS active_claims ( + execution_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, + turn_id TEXT NOT NULL, UNIQUE(thread_id, turn_id))""") + db.execute("""INSERT OR IGNORE INTO active_claims + SELECT execution_id, thread_id, turn_id FROM executions + WHERE execution_id NOT IN (SELECT execution_id FROM terminal_evidence)""") + db.execute("""CREATE TABLE IF NOT EXISTS checkpoint_scopes ( + store_id TEXT NOT NULL, checkpoint_thread_id TEXT NOT NULL, + checkpoint_ns TEXT NOT NULL, subject_id TEXT NOT NULL, + PRIMARY KEY(store_id, checkpoint_thread_id, checkpoint_ns))""") + db.execute("""CREATE TABLE IF NOT EXISTS checkpoint_writers ( + execution_id TEXT PRIMARY KEY, store_id TEXT NOT NULL, + checkpoint_thread_id TEXT NOT NULL, checkpoint_ns TEXT NOT NULL, + host_id TEXT NOT NULL, boot_id TEXT NOT NULL, + UNIQUE(store_id, checkpoint_thread_id, checkpoint_ns))""") + db.execute("""CREATE TABLE IF NOT EXISTS execution_checkpoint_scopes ( + execution_id TEXT PRIMARY KEY, store_id TEXT NOT NULL, + checkpoint_thread_id TEXT NOT NULL, checkpoint_ns TEXT NOT NULL)""") + + def claim_checkpoint(self, execution_id: str, *, store_id: str, + checkpoint_thread_id: str, subject_id: str) -> None: + """Root Graph writer, including child namespaces; no expiry/takeover.""" + scope = (store_id, checkpoint_thread_id, "") + if not all((execution_id, store_id, checkpoint_thread_id, subject_id)): + raise EvoRuntimeError("CHECKPOINT_SCOPE_INVALID") + with self._transaction() as db: + db.execute("BEGIN IMMEDIATE") + owner = db.execute("SELECT subject_id FROM checkpoint_scopes WHERE store_id=? AND checkpoint_thread_id=? AND checkpoint_ns=?", scope).fetchone() + if owner and owner[0] != subject_id: + raise EvoRuntimeError("CHECKPOINT_SUBJECT_MISMATCH") + if db.execute("SELECT 1 FROM checkpoint_writers WHERE store_id=? AND checkpoint_thread_id=? AND checkpoint_ns=?", scope).fetchone(): + raise EvoRuntimeError("CHECKPOINT_WRITER_BUSY") + if db.execute("SELECT 1 FROM active_claims WHERE execution_id NOT IN (SELECT execution_id FROM execution_checkpoint_scopes)").fetchone(): + raise EvoRuntimeError("CHECKPOINT_LEGACY_WRITER_UNKNOWN") + db.execute("INSERT OR IGNORE INTO checkpoint_scopes VALUES (?, ?, ?, ?)", (*scope, subject_id)) + db.execute("INSERT INTO checkpoint_writers VALUES (?, ?, ?, ?, ?, ?)", + (execution_id, *scope, self.host_id, self.boot_id)) + + def release_unbound_checkpoint(self, execution_id: str) -> None: + """Only preparation failure, before bind/construction/writes began.""" + with self._transaction() as db: + db.execute("BEGIN IMMEDIATE") + if db.execute("SELECT 1 FROM executions WHERE execution_id=?", (execution_id,)).fetchone(): + raise EvoRuntimeError("CHECKPOINT_WRITER_ALREADY_BOUND") + db.execute("DELETE FROM checkpoint_writers WHERE execution_id=? AND host_id=? AND boot_id=?", + (execution_id, self.host_id, self.boot_id)) + + def _connect(self): + db = sqlite3.connect(self.path, timeout=2) + db.row_factory = sqlite3.Row + db.execute("PRAGMA synchronous=FULL") + return db + + @contextmanager + def _transaction(self): + try: + with closing(self._connect()) as db, db: + yield db + except sqlite3.OperationalError as exc: + code = getattr(exc, "sqlite_errorcode", 0) & 255 + if code in {sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED}: + raise EvoRuntimeError("HOST_REGISTRY_BUSY") from exc + raise + + def bind(self, *, execution_id: str, grant_id: str, digest: str, + thread_id: str, turn_id: str, predecessor_execution_id: str = "", + predecessor_checkpoint_id: str = "", predecessor_owner_epoch: int = 0, + continuation_pending_hash: str = "", continuation_decision_hash: str = "") -> None: + with self._transaction() as db: + db.execute("BEGIN IMMEDIATE") + if db.execute("SELECT 1 FROM executions WHERE execution_id=? OR grant_id=?", + (execution_id, grant_id)).fetchone(): + raise EvoRuntimeError("EXECUTION_IDENTITY_CONFLICT") + writer = db.execute("SELECT * FROM checkpoint_writers WHERE execution_id=?", (execution_id,)).fetchone() + if writer and (writer["host_id"] != self.host_id or writer["boot_id"] != self.boot_id): + raise EvoRuntimeError("EXECUTION_BOOT_MISMATCH") + if writer and predecessor_execution_id: + previous = db.execute("SELECT * FROM execution_checkpoint_scopes WHERE execution_id=?", (predecessor_execution_id,)).fetchone() + if previous is None or any(previous[k] != writer[k] for k in ("store_id", "checkpoint_thread_id", "checkpoint_ns")): + raise EvoRuntimeError("CONTINUATION_CHECKPOINT_SCOPE_MISMATCH") + if db.execute("SELECT 1 FROM active_claims WHERE thread_id=? AND turn_id=?", + (thread_id, turn_id)).fetchone(): + raise EvoRuntimeError("TURN_EXECUTION_UNKNOWN") + prior = db.execute("SELECT 1 FROM executions WHERE thread_id=? AND turn_id=?", + (thread_id, turn_id)).fetchone() + if prior or predecessor_execution_id or predecessor_checkpoint_id: + if not predecessor_execution_id: + raise EvoRuntimeError("CONTINUATION_REQUIRED") + pending = db.execute("""SELECT p.*, e.owner_epoch, e.host_id FROM pending_continuations p + JOIN executions e USING(execution_id) + WHERE p.execution_id=? AND e.thread_id=? AND e.turn_id=?""", + (predecessor_execution_id, thread_id, turn_id)).fetchone() + if (pending is not None and pending["consumed_by"] is not None + and pending["checkpoint_id"] == predecessor_checkpoint_id): + failed = db.execute( + "SELECT 1 FROM terminal_evidence WHERE execution_id=? AND outcome='failed'", + (pending["consumed_by"],), + ).fetchone() + if failed: + # Never unconsume a decision on failure. A new grant alone + # is insufficient; recovery needs fresh pending authority. + raise EvoRuntimeError("CONTINUATION_CONSUMED_FAILURE_REQUIRES_REAUTHORIZATION") + if (pending is None or not predecessor_checkpoint_id + or pending["checkpoint_id"] != predecessor_checkpoint_id + or pending["consumed_by"] is not None): + raise EvoRuntimeError("CONTINUATION_INVALID") + if (pending["host_id"] != self.host_id or not pending["pending_hash"] + or pending["owner_epoch"] != predecessor_owner_epoch + or isinstance(predecessor_owner_epoch, bool) + or pending["pending_hash"] != continuation_pending_hash + or len(continuation_decision_hash) != 64): + raise EvoRuntimeError("CONTINUATION_AUTHORIZATION_INVALID") + if db.execute("SELECT 1 FROM cancel_intents WHERE execution_id=?", + (predecessor_execution_id,)).fetchone(): + raise EvoRuntimeError("CONTINUATION_CANCELLED") + db.execute("UPDATE pending_continuations SET consumed_by=?, decision_hash=? WHERE execution_id=?", + (execution_id, continuation_decision_hash, predecessor_execution_id)) + db.execute("INSERT INTO executions VALUES (?, ?, ?, ?, ?, ?, ?, 1)", + (execution_id, grant_id, digest, thread_id, turn_id, + self.host_id, self.boot_id)) + db.execute("INSERT INTO active_claims VALUES (?, ?, ?)", + (execution_id, thread_id, turn_id)) + if writer: + db.execute("INSERT INTO execution_checkpoint_scopes VALUES (?, ?, ?, ?)", + (execution_id, writer["store_id"], writer["checkpoint_thread_id"], writer["checkpoint_ns"])) + + def finish(self, execution_id: str, *, outcome: str, checkpoint_id: str = "") -> None: + """Trusted resource-owner attestation, not transferable control authority.""" + self._finish(execution_id, outcome=outcome, checkpoint_id=checkpoint_id) + + def prepare_terminal(self, execution_id: str, *, event: dict) -> None: + """Original resource owner only, AFTER all owned cleanup returns.""" + body = canonical_json_v1(event).decode() + digest = hashlib.sha256(body.encode()).hexdigest() + if (event.get("run_id") != execution_id or event.get("kind") != "run" + or event.get("payload", {}).get("kind") != "run_terminal"): + raise EvoRuntimeError("EXECUTION_TERMINAL_CONFLICT") + with self._transaction() as db: + db.execute("BEGIN IMMEDIATE") + row = db.execute("SELECT * FROM executions WHERE execution_id=?", (execution_id,)).fetchone() + if row is None or row['host_id'] != self.host_id or row['boot_id'] != self.boot_id: + raise EvoRuntimeError("EXECUTION_BOOT_MISMATCH") + prior = db.execute("SELECT digest FROM terminal_intents WHERE execution_id=?", (execution_id,)).fetchone() + if prior and prior['digest'] != digest: + raise EvoRuntimeError("EXECUTION_TERMINAL_CONFLICT") + db.execute("INSERT OR IGNORE INTO terminal_intents VALUES (?, ?, ?, 'prepared', ?)", + (execution_id, body, digest, self.boot_id)) + + def terminal_intent(self, execution_id: str) -> dict | None: + with closing(self._connect()) as db: + row = db.execute("""SELECT i.* FROM terminal_intents i JOIN executions e USING(execution_id) + WHERE execution_id=? AND e.host_id=?""", (execution_id, self.host_id)).fetchone() + if row is None: + return None + return {**dict(row), 'event': json.loads(row['event_json']), 'cleanup_confirmed': True} + + def confirm_terminal(self, execution_id: str, *, digest: str) -> None: + with self._transaction() as db: + row = db.execute("""SELECT i.* FROM terminal_intents i JOIN executions e USING(execution_id) + WHERE execution_id=? AND e.host_id=?""", (execution_id, self.host_id)).fetchone() + if row is None or row['digest'] != digest: + raise EvoRuntimeError("EXECUTION_TERMINAL_CONFLICT") + db.execute("UPDATE terminal_intents SET phase='sink_confirmed' WHERE execution_id=? AND phase='prepared'", + (execution_id,)) + + def _finish(self, execution_id: str, *, outcome: str, checkpoint_id: str) -> None: + if outcome not in {"completed", "cancelled", "failed", "awaiting_input"}: + raise EvoRuntimeError("EXECUTION_OUTCOME_INVALID") + with self._transaction() as db: + db.execute("BEGIN IMMEDIATE") + row = db.execute("SELECT * FROM executions WHERE execution_id=?", + (execution_id,)).fetchone() + intent = db.execute("SELECT * FROM terminal_intents WHERE execution_id=?", (execution_id,)).fetchone() + if row is None or row["host_id"] != self.host_id or (row["boot_id"] != self.boot_id and + (intent is None or intent['phase'] not in {'sink_confirmed', 'registry_finished'})): + raise EvoRuntimeError("EXECUTION_BOOT_MISMATCH") + if intent is not None: + payload = json.loads(intent['event_json'])['payload'] + if (intent['phase'] not in {'sink_confirmed', 'registry_finished'} + or payload['outcome'] != outcome + or str(payload.get('checkpoint_id') or '') != checkpoint_id): + raise EvoRuntimeError("EXECUTION_TERMINAL_CONFLICT") + prior = db.execute("SELECT outcome FROM terminal_evidence WHERE execution_id=?", + (execution_id,)).fetchone() + if prior is not None and prior["outcome"] != outcome: + raise EvoRuntimeError("EXECUTION_TERMINAL_CONFLICT") + db.execute("INSERT OR IGNORE INTO terminal_evidence VALUES (?, ?)", + (execution_id, outcome)) + if outcome == "awaiting_input" and checkpoint_id: + pending = db.execute("SELECT checkpoint_id FROM pending_continuations WHERE execution_id=?", + (execution_id,)).fetchone() + if pending is not None and pending["checkpoint_id"] != checkpoint_id: + raise EvoRuntimeError("EXECUTION_TERMINAL_CONFLICT") + identity = {} + if intent is not None: + payload = json.loads(intent['event_json'])['payload'] + identity = {k: payload.get(k) for k in ( + "checkpoint_thread_id", "checkpoint_id", "checkpoint_ns", "pending_interrupts")} + pending_hash = hashlib.sha256(canonical_json_v1(identity)).hexdigest() if identity else "" + db.execute("INSERT OR IGNORE INTO pending_continuations " + "(execution_id, checkpoint_id, consumed_by, pending_hash, decision_hash) VALUES (?, ?, NULL, ?, '')", + (execution_id, checkpoint_id, pending_hash)) + db.execute("DELETE FROM active_claims WHERE execution_id=?", (execution_id,)) + db.execute("DELETE FROM checkpoint_writers WHERE execution_id=?", (execution_id,)) + db.execute("UPDATE terminal_intents SET phase='registry_finished' WHERE execution_id=?", (execution_id,)) + + def continuation(self, execution_id: str) -> dict: + with closing(self._connect()) as db: + row = db.execute("""SELECT p.*, e.owner_epoch FROM pending_continuations p + JOIN executions e USING(execution_id) WHERE execution_id=? AND e.host_id=?""", + (execution_id, self.host_id)).fetchone() + if row is None: + raise EvoRuntimeError("CONTINUATION_INVALID") + return dict(row) + + def lookup_grant(self, grant_id: str, digest: str) -> dict | None: + with closing(self._connect()) as db: + row = db.execute("SELECT * FROM executions WHERE grant_id=?", (grant_id,)).fetchone() + if row is None: + return None + if row["digest"] != digest: + raise EvoRuntimeError("CONTRACT_REPLAYED") + return dict(row) + + def transfer_control(self, execution_id: str, *, expected_epoch: int, new_epoch: int) -> int: + if new_epoch <= expected_epoch: + raise EvoRuntimeError("OWNER_EPOCH_STALE") + with self._transaction() as db: + changed = db.execute( + "UPDATE executions SET owner_epoch=? WHERE execution_id=? AND owner_epoch=? AND host_id=? AND boot_id=?", + (new_epoch, execution_id, expected_epoch, self.host_id, self.boot_id), + ).rowcount + if changed != 1: + raise EvoRuntimeError("OWNER_EPOCH_STALE") + return new_epoch + + def require_control(self, execution_id: str, *, owner_epoch: int) -> None: + with closing(self._connect()) as db: + row = db.execute("SELECT owner_epoch FROM executions WHERE execution_id=?", + (execution_id,)).fetchone() + if row is None or row["owner_epoch"] != owner_epoch: + raise EvoRuntimeError("OWNER_EPOCH_STALE") + + def inspect(self, execution_id: str) -> dict: + with closing(self._connect()) as db: + row = db.execute("SELECT * FROM executions WHERE execution_id=?", (execution_id,)).fetchone() + terminal = db.execute("SELECT outcome FROM terminal_evidence WHERE execution_id=?", + (execution_id,)).fetchone() + cancel = db.execute("SELECT 1 FROM cancel_intents WHERE execution_id=? LIMIT 1", + (execution_id,)).fetchone() + return {**(dict(row) if row else {"execution_id": execution_id}), + "cancel_requested": cancel is not None, + "recovery_action": "inspect_only", + "status": terminal["outcome"] if terminal else "unknown", + "resources_confirmed_exited": terminal is not None, + "source": "host_binding_only" if row else "no_host_binding"} + + def _control_row(self, db, execution_id, owner_epoch, boot_id): + if owner_epoch is None: + raise EvoRuntimeError("OWNER_EPOCH_REQUIRED") + row = db.execute("SELECT * FROM executions WHERE execution_id=?", (execution_id,)).fetchone() + if row is None: + raise EvoRuntimeError("EXECUTION_UNKNOWN") + if row["host_id"] != self.host_id or row["boot_id"] != self.boot_id or boot_id != self.boot_id: + raise EvoRuntimeError("EXECUTION_BOOT_MISMATCH") + if isinstance(owner_epoch, bool) or row["owner_epoch"] != owner_epoch: + raise EvoRuntimeError("OWNER_EPOCH_STALE") + return row + + def inspect_control(self, execution_id: str, *, owner_epoch: int | None, boot_id: str | None) -> dict: + with closing(self._connect()) as db: + row = self._control_row(db, execution_id, owner_epoch, boot_id) + return self.inspect(execution_id) + + def accept_cancel(self, execution_id: str, *, owner_epoch: int | None, boot_id: str | None, reason: str) -> int: + # COMMIT is the linearization point shared with transfer's conditional UPDATE. + # Accepted commands survive a later transfer; no write lock crosses an await. + with self._transaction() as db: + db.execute("BEGIN IMMEDIATE") + self._control_row(db, execution_id, owner_epoch, boot_id) + cursor = db.execute( + "INSERT INTO cancel_intents (execution_id, owner_epoch, boot_id, reason) VALUES (?, ?, ?, ?)", + (execution_id, owner_epoch, boot_id, reason), + ) + assert cursor.lastrowid is not None + return cursor.lastrowid \ No newline at end of file diff --git a/EvoScientist/llm/invocation/contract.py b/EvoScientist/llm/invocation/contract.py index 16a594e..13bc79a 100644 --- a/EvoScientist/llm/invocation/contract.py +++ b/EvoScientist/llm/invocation/contract.py @@ -83,8 +83,9 @@ def compile_invocation_plan( """Validate and freeze adapter output before constructing a provider SDK.""" params = dict(sdk_params) - streaming = purpose == "main_agent" + streaming = True params["streaming"] = streaming + params["disable_streaming"] = False if runtime_provider == "openai" and streaming: params["stream_usage"] = True token_fields = tuple( diff --git a/EvoScientist/llm/model_config.py b/EvoScientist/llm/model_config.py index 5bea899..7986ca0 100644 --- a/EvoScientist/llm/model_config.py +++ b/EvoScientist/llm/model_config.py @@ -14,7 +14,7 @@ import stat import tempfile import time from collections.abc import Mapping, Sequence -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass, field, replace from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any @@ -477,12 +477,49 @@ class EvoModelConfig: pools = _parse_endpoint_pools(raw.get("endpoint_pools"), providers) health = _parse_health(raw.get("route_health")) selectors = _parse_selectors(raw.get("route_selectors"), providers, pools) + # Schema2's historical non_streaming Web tool routes and native routes + # are administrator declarations, not probes or HTTP streaming policy. + declared_tool_models = { + (selector.provider, selector.model) + for selector in selectors.values() + if selector.tool_call_transport in {"native", "non_streaming"} + } + providers = { + provider_id: replace( + provider, + models={ + model_id: replace( + model, + capabilities={ + **model.capabilities, + "tools": (provider_id, model_id) in declared_tool_models, + }, + ) + for model_id, model in provider.models.items() + }, + ) + for provider_id, provider in providers.items() + } main_routes, title_selector_id = _parse_purpose_routes( raw.get("purpose_routes"), selectors ) limits = _parse_call_limits(raw.get("purpose_call_limits")) web_runtime = _parse_web_runtime(raw.get("web_runtime")) fallbacks = _parse_fallbacks(raw.get("tool_protocol_fallbacks"), selectors) + execution_selectors = set(main_routes.selectable.values()) | { + title_selector_id + } + for primary, candidates in fallbacks.items(): + execution_selectors.add(primary) + execution_selectors.update(candidates) + for selector_id in sorted(execution_selectors): + selector = selectors[selector_id] + if (selector.provider, selector.model) not in declared_tool_models: + raise EvoRuntimeError( + "LLM_ROUTE_CONFIGURATION_REQUIRED", + "schema2 tool capability migration requires an explicit " + f"native or historical non_streaming declaration for {selector_id}", + ) config = cls( config_revision=revision, config_identity_key_id=identity_key_id, @@ -1973,6 +2010,7 @@ def _parse_providers(value: Any) -> Mapping[str, ProviderConfig]: _string_set(access.get("allowed_plans"), "allowed_plans"), _string_set(access.get("allowed_roles"), "allowed_roles"), quote, + capabilities={"tools": False, "text": True}, ) result[key] = ProviderConfig( key, @@ -2137,10 +2175,12 @@ def _parse_selectors( transport = _text( item.get("tool_call_transport"), "selector.tool_call_transport" ) - if transport != "non_streaming": + # Legacy values remain readable for signed route evidence, but do not + # control HTTP streaming. Tool support and HTTP mode are independent. + if transport not in {"native", "streaming", "non_streaming"}: raise EvoRuntimeError( "LLM_ROUTE_CONFIGURATION_REQUIRED", - "Web tool routes must be non_streaming", + "Web tool routes require native tool support", ) _validate_params( provider.params, name=f"providers.{provider_id}.params", allowed=allowed @@ -3180,14 +3220,16 @@ def convert_v2_to_v3_draft( "api_mode": selector.api_mode if selector else "chat_completions", - "tool_call_transport": "native", + "tool_call_transport": "native" + if model.capabilities.get("tools", False) + else "disabled", }, "capabilities": { "text": True, "vision": model.supports_vision, "video": False, "documents": False, - "tools": True, + "tools": bool(model.capabilities.get("tools", False)), "structured_output": False, "thinking": model.supports_reasoning, }, diff --git a/EvoScientist/llm/models.py b/EvoScientist/llm/models.py index 0ad38a8..f21dbc1 100644 --- a/EvoScientist/llm/models.py +++ b/EvoScientist/llm/models.py @@ -572,7 +572,7 @@ def get_chat_model( _original_provider = provider base_url_default, api_key_env = _OPENAI_ROUTED_PROVIDERS[provider] if provider == "custom-openai": - base_url = os.environ.get("CUSTOM_OPENAI_BASE_URL", "") + base_url = str(kwargs.get("base_url") or os.environ.get("CUSTOM_OPENAI_BASE_URL", "")) if not base_url: raise ValueError( "CUSTOM_OPENAI_BASE_URL environment variable is required when using " diff --git a/EvoScientist/llm/runtime.py b/EvoScientist/llm/runtime.py index 64a7361..0a9d06f 100644 --- a/EvoScientist/llm/runtime.py +++ b/EvoScientist/llm/runtime.py @@ -15,6 +15,7 @@ import traceback import uuid from collections import defaultdict from collections.abc import AsyncIterator, Callable, Mapping, Sequence +from contextvars import ContextVar from dataclasses import asdict, dataclass, field, replace from typing import Any @@ -35,6 +36,7 @@ from .contracts import ( ModelAttemptEvent, ModelCatalog, ModelCatalogEntry, + ModelFactoryResult, PreparedRunQuote, PricingQuote, RouteCallBound, @@ -46,6 +48,7 @@ from .contracts import ( now_ms, ) from .crypto import HmacKeyRing, canonical_json_v1 +from .host_execution_registry import HostExecutionRegistry from .invocation import InvocationPlan, compile_invocation_plan from .model_config import ( EvoModelConfig, @@ -65,6 +68,7 @@ from .user_options import ( _BIGINT_MAX = 2**63 - 1 logger = logging.getLogger(__name__) +_construction_owner: ContextVar[Any] = ContextVar("evo_construction_owner", default=None) _ROUTE_SEMANTICS_INFO = "ai4sci/route-semantics-hash/v3" _ROUTE_FINGERPRINT_INFO = "ai4sci/route-fingerprint/v3" _ENDPOINT_FINGERPRINT_INFO = "ai4sci/endpoint-fingerprint/v3" @@ -382,6 +386,7 @@ class EvoModelRuntime: ] | None = None, runtime_instance_id: str | None = None, + host_registry: HostExecutionRegistry | None = None, ) -> None: self.store = store self.admission_verifier = admission_verifier @@ -399,6 +404,7 @@ class EvoModelRuntime: self.model_factory = model_factory or self._default_model_factory self.agent_factory = agent_factory or self._default_agent_factory self.runtime_instance_id = runtime_instance_id or str(uuid.uuid4()) + self.host_registry = host_registry self._route_health = RouteHealthBook() self._provider_health = RouteHealthBook() self._pool = _SmoothWeightedRoundRobin() @@ -464,6 +470,7 @@ class EvoModelRuntime: ) -> PreparedRunQuote: self.admission_verifier.require_preparation(grant) self._validate_preparation_echo(grant, agent_input) + await self._validate_continuation(grant, agent_input, host) tool_payload = self._tool_registry_payload(host) request_key = (grant.subject_id, grant.request_id, grant.turn_id) request_digest = self._preparation_request_digest( @@ -616,7 +623,14 @@ class EvoModelRuntime: for routes in purpose_routes.values() for route in routes } + execution_id = str(uuid.uuid4()) snapshot_payload = { + "execution_id": execution_id, + "predecessor_execution_id": grant.predecessor_execution_id, + "predecessor_checkpoint_id": grant.predecessor_checkpoint_id, + "predecessor_owner_epoch": grant.predecessor_owner_epoch, + "continuation_pending_hash": grant.continuation_pending_hash, + "continuation_decision_hash": grant.continuation_decision_hash, "request_id": grant.request_id, "turn_id": grant.turn_id, "thread_id": grant.thread_id, @@ -640,6 +654,7 @@ class EvoModelRuntime: preparation_id = str(uuid.uuid4()) issued_at = now_ms() quote = self.quote_authority.sign_quote( + execution_id=execution_id, preparation_id=preparation_id, request_id=grant.request_id, turn_id=grant.turn_id, @@ -737,6 +752,142 @@ class EvoModelRuntime: self._prepared_requests[request_key] = preparation_id return quote + def inspect(self, execution_id: str, *, owner_epoch: int | None = None, + boot_id: str | None = None) -> dict: + if self.host_registry is None: + return {"execution_id": execution_id, "status": "unknown", + "resources_confirmed_exited": False, "source": "registry_disabled"} + state = self.host_registry.inspect_control( + execution_id, owner_epoch=owner_epoch, boot_id=boot_id) + run = next((h.run for h in self._prepared.values() + if h.run is not None and h.run.run_id == execution_id), None) + if run is not None: + state["terminal_error_code"] = run._terminal_error_code + return state + + async def cancel(self, execution_id: str, *, reason: str, + owner_epoch: int | None = None, boot_id: str | None = None) -> str: + run = next((run for _, run in self._started_grants.values() + if run.run_id == execution_id), None) + if run is None: + raise EvoRuntimeError("EXECUTION_UNKNOWN") + if self.host_registry is not None: + self.host_registry.accept_cancel( + execution_id, owner_epoch=owner_epoch, boot_id=boot_id, reason=reason) + # Consume the committed intent before yielding. Later transfers do not + # revoke already accepted cancellation; the run owns its cleanup task. + run._request_cancel(reason) + return await run.wait_stopped(timeout=2.0) + + async def recover_terminal(self, execution_id: str, *, sink: Any) -> str: + """Explicit host repair of durable cleanup evidence; never executes Graph.""" + registry = self.host_registry + intent = registry.terminal_intent(execution_id) if registry is not None else None + if registry is None or intent is None: + raise EvoRuntimeError("TERMINAL_CLEANUP_EVIDENCE_REQUIRED") + event = EvoRuntimeEvent(**intent['event']) + digest = hashlib.sha256(canonical_json_v1(asdict(event))).hexdigest() + if digest != intent['digest']: + raise EvoRuntimeError("EVO_EVENT_CONFLICT") + if intent['phase'] == 'prepared': + try: + confirmation = await sink.confirm(event.event_id, digest) + if confirmation == 'absent': + await sink.commit(event) + confirmation = await sink.confirm(event.event_id, digest) + except Exception as exc: + raise EvoRuntimeError("EVENT_COMMIT_INDETERMINATE") from exc + if confirmation != 'committed': + raise EvoRuntimeError("EVO_EVENT_CONFLICT" if confirmation == 'conflict' + else "EVENT_COMMIT_INDETERMINATE") + await asyncio.to_thread(registry.confirm_terminal, execution_id, digest=digest) + outcome = str(event.payload['outcome']) + await asyncio.to_thread(registry.finish, execution_id, outcome=outcome, + checkpoint_id=str(event.payload.get('checkpoint_id') or '')) + run = next((h.run for h in self._prepared.values() + if h.run is not None and h.run.run_id == execution_id), None) + if run is not None and run._terminal_task is not None and run._terminal_task.done(): + async with run._condition: + if not any(item.event_id == event.event_id for item in run._journal): + run._journal.append(event) + run._journal_bytes += len(canonical_json_v1(asdict(event))) + run._sequence = max(run._sequence, event.sequence) + run._terminal_event = event + run._state = 'TERMINAL' + run._terminal_error_code = None + run._condition.notify_all() + if run._prepared.state == 'CONSTRUCTION_FAILED': + async with self._registry_lock: + self._retire_prepared_locked(run._prepared) + return outcome + + async def _checkpoint_store_id(self, host): + # No portable durable identity exists in BaseCheckpointSaver. Only the + # inspected SQLite implementation is supported by this opt-in protocol. + from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver + import os + if type(host.checkpointer) is not AsyncSqliteSaver: + raise EvoRuntimeError("CHECKPOINT_STORE_UNSUPPORTED") + async with host.checkpointer.conn.execute("PRAGMA database_list") as cursor: + databases = await cursor.fetchall() + filename = next((row[2] for row in databases if row[1] == "main"), "") + if not filename: + raise EvoRuntimeError("CHECKPOINT_STORE_UNSUPPORTED") + info = os.stat(filename) + return f"sqlite:{info.st_dev}:{info.st_ino}" + + async def _validate_continuation(self, grant, agent_input, host): + if self.host_registry is None: + return + from langgraph.types import Command + command = agent_input.message + if not grant.predecessor_execution_id: + if isinstance(command, Command): + raise EvoRuntimeError("CONTINUATION_REQUIRED") + return + pending = self.host_registry.continuation(grant.predecessor_execution_id) + if (pending["owner_epoch"] != grant.predecessor_owner_epoch + or isinstance(grant.predecessor_owner_epoch, bool) + or pending["pending_hash"] != grant.continuation_pending_hash + or pending["checkpoint_id"] != grant.predecessor_checkpoint_id + or grant.checkpoint_snapshot_id != grant.predecessor_checkpoint_id): + raise EvoRuntimeError("CONTINUATION_AUTHORIZATION_INVALID") + if not isinstance(command, Command) or command.update or command.goto or command.graph: + raise EvoRuntimeError("CONTINUATION_DECISION_INVALID") + resume = command.resume + if (not isinstance(resume, dict) or set(resume) != {"decisions"} + or hashlib.sha256(canonical_json_v1(resume)).hexdigest() != grant.continuation_decision_hash): + raise EvoRuntimeError("CONTINUATION_DECISION_INVALID") + intent = self.host_registry.terminal_intent(grant.predecessor_execution_id) + if intent is None or intent["phase"] != "registry_finished": + raise EvoRuntimeError("CONTINUATION_INVALID") + payload = intent["event"]["payload"] + identity = {k: payload[k] for k in ( + "checkpoint_thread_id", "checkpoint_id", "checkpoint_ns", "pending_interrupts")} + if identity["checkpoint_thread_id"] != agent_input.checkpoint_thread_id or identity["checkpoint_ns"] != "": + raise EvoRuntimeError("CONTINUATION_INVALID") + checkpoint = await host.checkpointer.aget_tuple({"configurable": { + "thread_id": agent_input.checkpoint_thread_id, "checkpoint_ns": identity["checkpoint_ns"]}}) + if (checkpoint is None or checkpoint.config["configurable"]["checkpoint_id"] != identity["checkpoint_id"] + or checkpoint.checkpoint["channel_values"].get("_verified_review_mode", {}).get("mode") != "manual"): + raise EvoRuntimeError("CONTINUATION_CHECKPOINT_STALE") + actual = [] + for _, channel, value in checkpoint.pending_writes or (): + if channel == "__interrupt__": + for item in value: + projected = {"id": item.id, "value": item.value} + if projected not in actual: + actual.append(projected) + if actual != identity["pending_interrupts"] or not actual: + raise EvoRuntimeError("CONTINUATION_PENDING_STALE") + actions = [action for item in actual for action in item["value"].get("action_requests", [])] + decisions = resume["decisions"] + if (len(actual) != 1 or not actions or not isinstance(decisions, list) + or len(decisions) != len(actions) + or any(not isinstance(d, dict) or d.get("type") not in {"approve", "reject"} + or set(d) - {"type", "message"} for d in decisions)): + raise EvoRuntimeError("CONTINUATION_DECISION_INVALID") + async def start_web_run(self, admission: AdmissionGrant) -> EvoWebRun: self.admission_verifier.require_admission(admission) digest = canonical_json_v1(admission.unsigned_payload()) @@ -746,6 +897,14 @@ class EvoModelRuntime: if replay[0] != digest: raise EvoRuntimeError("CONTRACT_REPLAYED") return replay[1] + if self.host_registry is not None: + binding = self.host_registry.lookup_grant( + admission.grant_id, hashlib.sha256(digest).hexdigest() + ) + if binding is not None: + if self.host_registry.inspect(binding["execution_id"])["status"] == "failed": + raise EvoRuntimeError("EXECUTION_FAILED_RETRY_REQUIRES_NEW_ADMISSION") + raise EvoRuntimeError("EXECUTION_UNKNOWN") handle = self._prepared.get(admission.preparation_id) if handle is None: raise EvoRuntimeError("RUN_LOST") @@ -761,22 +920,59 @@ class EvoModelRuntime: self._validate_still_fresh(handle) if handle.host.runtime_event_sink is None: raise EvoRuntimeError("EVENT_INGRESS_UNAVAILABLE") - model_set = self._build_model_set( - handle.snapshot, admission.reasoning_effort - ) - agent = self.agent_factory(handle.snapshot, handle.host, model_set) run = _EvoWebRun( runtime=self, admission=admission, prepared=handle, - agent=agent, - model_set=model_set, + agent=None, + model_set=AgentModelSet(None, None, None), ) + if self.host_registry is not None: + store_id = await self._checkpoint_store_id(handle.host) + self.host_registry.claim_checkpoint( + run.run_id, store_id=store_id, + checkpoint_thread_id=handle.input.checkpoint_thread_id, + subject_id=admission.subject_id) + try: + await self._validate_continuation(handle.grant, handle.input, handle.host) + if self.host_registry is not None: + self.host_registry.bind( + execution_id=run.run_id, grant_id=admission.grant_id, + digest=hashlib.sha256(digest).hexdigest(), + thread_id=admission.thread_id, turn_id=admission.turn_id, + predecessor_execution_id=handle.grant.predecessor_execution_id, + predecessor_checkpoint_id=handle.grant.predecessor_checkpoint_id, + predecessor_owner_epoch=handle.grant.predecessor_owner_epoch, + continuation_pending_hash=handle.grant.continuation_pending_hash, + continuation_decision_hash=handle.grant.continuation_decision_hash, + ) + except BaseException: + if self.host_registry is not None: + self.host_registry.release_unbound_checkpoint(run.run_id) + raise + token = _construction_owner.set(run) + try: + model_set = self._build_model_set( + handle.snapshot, admission.reasoning_effort + ) + run._model_set = model_set + run._agent = self.agent_factory(handle.snapshot, handle.host, model_set) + except BaseException: + # Retain the owner on the handle even if rollback itself fails. + handle.run = run + handle.state = "CONSTRUCTION_FAILED" + run._start_terminal("failed", error_code="RUN_CONSTRUCTION_FAILED") + await run._ensure_cleanup() + await run.wait_stopped(timeout=0.25) + raise + finally: + _construction_owner.reset(token) handle.state = "STARTED" handle.run = run handle.start_grant_digest = digest async with self._registry_lock: self._started_grants[admission.grant_id] = (digest, run) + run._start() return run async def cancel_prepared_run(self, preparation_id: str, *, reason: str) -> bool: @@ -1089,7 +1285,7 @@ class EvoModelRuntime: invocation_identity = route_fingerprint(config, route, fingerprint_key) supports_tools = bool(model.capabilities.get("tools", False)) effective_tool_transport = ( - route.tool_call_transport if supports_tools else "disabled" + "native" if supports_tools else "disabled" ) identity = RouteIdentity( config_revision=config.config_revision, @@ -1269,6 +1465,14 @@ class EvoModelRuntime: provider=route.runtime_provider, **self._model_factory_kwargs(route), ) + if isinstance(model, ModelFactoryResult): + owner = _construction_owner.get() + if owner is None and model.owned_clients: + raise EvoRuntimeError("MODEL_RESOURCE_OWNER_MISSING") + if owner is not None: + for client in model.owned_clients: + owner._owned_clients[id(client)] = client + model = model.model return self._attach_route_metadata(model, route, purpose, bound) @staticmethod @@ -1355,13 +1559,10 @@ class EvoModelRuntime: invocation_plan=plan, ) + params.pop("output_token_limit", None) params.update( { "max_tokens": bound.max_output_tokens, - "disable_streaming": "tool_calling", - "streaming": False - if purpose != "main_agent" - else params.get("streaming", False), "use_responses_api": route.identity.api_mode == "responses", } ) @@ -1645,8 +1846,26 @@ class EvoModelRuntime: from .gemini_interactions import create_gemini_interactions_model return create_gemini_interactions_model(**kwargs) - from .models import get_chat_model + from .models import _OPENAI_ROUTED_PROVIDERS, get_chat_model + # LangChain caches default OpenAI HTTP clients process-wide. A run must + # never close that cache underneath another run's active stream. + if kwargs.get("provider") == "openai" or kwargs.get("provider") in _OPENAI_ROUTED_PROVIDERS: + import httpx + + # Supplying owned clients disables LangChain's automatic usage opt-in. + # Keep explicit caller configuration and borrowed transports intact. + kwargs.setdefault("stream_usage", True) + if "http_client" not in kwargs: + kwargs["http_client"] = httpx.Client() + owner = _construction_owner.get() + if owner is not None: + owner._owned_clients[id(kwargs["http_client"])] = kwargs["http_client"] + if "http_async_client" not in kwargs: + kwargs["http_async_client"] = httpx.AsyncClient() + owner = _construction_owner.get() + if owner is not None: + owner._owned_clients[id(kwargs["http_async_client"])] = kwargs["http_async_client"] return get_chat_model(**kwargs) @staticmethod @@ -1676,7 +1895,7 @@ class _EvoWebRun: self._host = prepared.host self._agent = agent self._model_set = model_set - self._run_id = str(uuid.uuid4()) + self._run_id = prepared.quote.execution_id or str(uuid.uuid4()) self._state = "ACTIVE" self._sequence = 0 self._journal: list[EvoRuntimeEvent] = [] @@ -1686,6 +1905,12 @@ class _EvoWebRun: self._budget_lock = asyncio.Lock() self._terminal_lock = asyncio.Lock() self._agent_task: asyncio.Task[None] | None = None + self._checkpoint_writes_stopped = True + from ..stream.stop import StreamStop + self._stream_stop = StreamStop() if runtime.host_registry is not None else None + self._event_iterator = None + self._stop_task = None + self._stop_errors = [] self._stream_wakeup_task: asyncio.Task[None] | None = None self._attempt_counts: dict[str, int] = defaultdict(int) self._remaining = admission.provider_run_reserve_microunits @@ -1698,6 +1923,78 @@ class _EvoWebRun: ) self._attempt_callback = _RuntimeAttemptCallback(self) self._terminal_event: EvoRuntimeEvent | None = None + self._terminal_task: asyncio.Task[EvoRuntimeEvent] | None = None + self._terminal_error_code: str | None = None + self._terminal_phase = "cleanup" + self._cancel_task: asyncio.Task[None] | None = None + self._cleanup_task: asyncio.Task[None] | None = None + self._resource_close_tasks: dict[int, asyncio.Task[None]] = {} + self._owned_clients: dict[int, Any] = {} + self._cleanup_failures: dict[int, list[BaseException]] = defaultdict(list) + + async def _close_owned_client(self, key: int, client: Any) -> None: + # A pending attempt owns the resource until it really returns. In + # particular, cancelling an observer cannot stop a synchronous thread. + for attempt in range(3): + try: + close = getattr(client, "aclose", None) or getattr(client, "close") + if asyncio.iscoroutinefunction(close): + result = close() + else: + result = await asyncio.to_thread(close) + if hasattr(result, "__await__"): + await result + except BaseException as exc: + self._cleanup_failures[key].append(exc) + if isinstance(exc, asyncio.CancelledError): + return + else: + self._owned_clients.pop(key, None) + return + if attempt < 2: + await asyncio.sleep(0) + + async def _close_owned_clients(self) -> None: + self._resource_close_tasks = { + key: asyncio.create_task( + self._close_owned_client(key, client), + name=f"evo-resource-close-{self._run_id}-{key}", + ) + for key, client in tuple(self._owned_clients.items()) + } + if self._resource_close_tasks: + await asyncio.gather(*self._resource_close_tasks.values()) + + + async def _ensure_cleanup(self, *, timeout: float | None = 0.25) -> None: + if self._cleanup_task is None: + self._cleanup_task = asyncio.create_task( + self._close_owned_clients(), name=f"evo-cleanup-{self._run_id}" + ) + # asyncio.wait bounds observation without cancelling the run-owned + # coordinator or any close operation, including cancellation-resistant + # async clients. Later observers can confirm eventual completion. + await asyncio.wait({self._cleanup_task}, timeout=timeout) + if not self._cleanup_task.done(): + raise TimeoutError("Run resource close remains unconfirmed") + self._cleanup_task.result() + if self._owned_clients: + errors = [] + for key in self._owned_clients: + task = self._resource_close_tasks.get(key) + failures = self._cleanup_failures.get(key, []) + errors.append( + failures[-1] if task is not None and task.done() and failures + else TimeoutError("Run resource close remains unconfirmed") + ) + raise BaseExceptionGroup("Run resource cleanup unconfirmed", errors) + + def _start(self) -> None: + if self._agent_task is None and self._terminal_event is None: + self._agent_task = asyncio.create_task( + self._run_with_timeout(), name=f"evo-run-{self._run_id}" + ) + self._agent_task.add_done_callback(self._wake_stream_waiters) @property def run_id(self) -> str: @@ -1714,11 +2011,7 @@ class _EvoWebRun: cursor = 0 if after_sequence is None else after_sequence if cursor < 0 or cursor > self._sequence: raise EvoRuntimeError("EVENT_CURSOR_INVALID") - if self._agent_task is None: - self._agent_task = asyncio.create_task( - self._run_with_timeout(), name=f"evo-run-{self._run_id}" - ) - self._agent_task.add_done_callback(self._wake_stream_waiters) + async for event in self._replay(cursor): yield event @@ -1729,15 +2022,48 @@ class _EvoWebRun: async with self._condition: self._condition.notify_all() - async def cancel(self, reason: str) -> str: + async def cancel(self, reason: str, *, owner_epoch: int | None = None, + boot_id: str | None = None) -> str: + if self._runtime.host_registry is not None: + return await self._runtime.cancel( + self.run_id, reason=reason, owner_epoch=owner_epoch, boot_id=boot_id) + self._request_cancel(reason) + return await self.wait_stopped(timeout=2.0) + + def _request_cancel(self, reason: str) -> None: + if self._terminal_task is not None or self._terminal_event is not None: + return + if self._cancel_task is None: + self._state = "CANCELLING" + if self._agent_task is not None and not self._agent_task.done(): + self._agent_task.cancel() + self._cancel_task = asyncio.create_task( + self._finish_cancel(reason), name=f"evo-stop-{self._run_id}" + ) + + async def wait_stopped(self, timeout: float | None = 2.0) -> str: if self._terminal_event is not None: return str(self._terminal_event.payload.get("outcome") or "terminal") + try: + async with asyncio.timeout(timeout): + task = self._cancel_task or self._agent_task + if task is not None: + await asyncio.shield(task) + if self._terminal_task is not None: + await asyncio.shield(self._terminal_task) + except Exception: + # Observation is not an ownership transfer or a stop proof. + return "unknown" + if self._terminal_event is None: + return "unknown" + return str(self._terminal_event.payload.get("outcome") or "terminal") + + async def _finish_cancel(self, reason: str) -> None: if self._agent_task is not None: - self._agent_task.cancel() + await asyncio.gather(self._agent_task, return_exceptions=True) await self._terminal_locked( "cancelled", error_code="RUN_CANCELLED", reason=reason ) - return "cancelled" async def _run_with_timeout(self) -> None: try: @@ -1763,9 +2089,13 @@ class _EvoWebRun: "billing_fencing_token": self._admission.billing_fencing_token, }, ) - from ..stream.events import stream_agent_events + from ..stream.events import ( + _snapshot_has_pending_interrupt, + stream_agent_events, + ) - async for source in stream_agent_events( + self._checkpoint_writes_stopped = False + self._event_iterator = stream_agent_events( self._agent, self._input.message, self._input.checkpoint_thread_id, @@ -1782,21 +2112,66 @@ class _EvoWebRun: configurable={ "turn_fencing_token": self._admission.turn_fencing_token, "turn_lease_owner": self._admission.request_id, + "checkpoint_writer_strict_close": self._runtime.host_registry is not None, + **({"checkpoint_id": self._prepared.grant.predecessor_checkpoint_id, + "checkpoint_ns": ""} + if self._prepared.grant.predecessor_execution_id else {}), }, error_mode="raise", - ): - if str(source.get("type") or "") == "done": - continue - await self._append_locked("agent", dict(source)) + **({"stop_owner": self._stream_stop} if self._stream_stop else {}), + ) + try: + async for source in self._event_iterator: + if str(source.get("type") or "") == "done": + continue + await self._append_locked("agent", dict(source)) + finally: + if self._stop_task is None: + self._stop_task = asyncio.create_task( + self._stop_checkpoint_writer(), name=f"evo-writer-stop-{self._run_id}") + await asyncio.shield(self._stop_task) + try: + final_state = await self._agent.aget_state( + {"configurable": {"thread_id": self._input.checkpoint_thread_id}} + ) + checkpoint = final_state.config["configurable"] + if not checkpoint.get("checkpoint_id"): + raise ValueError("Final checkpoint identity missing") + pending = [] + for item in ( + *getattr(final_state, "interrupts", ()), + *(item for task in getattr(final_state, "tasks", ()) + for item in getattr(task, "interrupts", ())), + ): + projected = {"id": item.id, "value": item.value} + if projected not in pending: + pending.append(projected) + checkpoint_details = json.loads(json.dumps({ + "checkpoint_thread_id": checkpoint["thread_id"], + "checkpoint_id": checkpoint["checkpoint_id"], + "checkpoint_ns": checkpoint.get("checkpoint_ns", ""), + "pending_interrupts": pending, + }, allow_nan=False)) + awaiting_input = _snapshot_has_pending_interrupt(final_state) + except Exception: + await self._terminal_locked( + "failed", error_code="FINAL_CHECKPOINT_READ_FAILED" + ) + return + if awaiting_input: + await self._terminal_locked( + "awaiting_input", checkpoint_details=checkpoint_details + ) + return if self._admission.title_policy == "best_effort": await self._run_title() else: await self._append_locked( "title", {"kind": "skipped", "reason": "DISABLED"} ) - await self._terminal_locked("completed") + await self._terminal_locked("completed", checkpoint_details=checkpoint_details) except asyncio.CancelledError: - return + raise except Exception as exc: logger.exception( "evo run %s failed before model provider error was reported", @@ -1812,6 +2187,21 @@ class _EvoWebRun: error_details=error_details, ) + async def _stop_checkpoint_writer(self) -> None: + while True: + try: + if self._event_iterator is not None: + await self._event_iterator.aclose() + if self._stream_stop is not None: + await asyncio.shield(self._stream_stop.start([])) + from ..stream.stop import sqlite_barrier + await sqlite_barrier(self._host.checkpointer) + self._checkpoint_writes_stopped = True + return + except Exception as exc: + self._stop_errors.append(exc) + await asyncio.sleep(0.05) + async def _run_title(self) -> None: if self._model_set.title is None: await self._append_locked( @@ -2048,6 +2438,9 @@ class _EvoWebRun: payload_digest = hashlib.sha256( canonical_json_v1(asdict(event)) ).hexdigest() + registry = self._runtime.host_registry + if control_terminal and registry is not None: + await asyncio.to_thread(registry.prepare_terminal, self.run_id, event=asdict(event)) commit_error: Exception | None = None try: result = await sink.commit(event) @@ -2069,6 +2462,8 @@ class _EvoWebRun: raise EvoRuntimeError("EVO_EVENT_CONFLICT") else: raise EvoRuntimeError("EVENT_COMMIT_INDETERMINATE") + if control_terminal and registry is not None: + await asyncio.to_thread(registry.confirm_terminal, self.run_id, digest=payload_digest) async with self._condition: self._sequence = sequence self._journal_bytes += event_size @@ -2077,13 +2472,65 @@ class _EvoWebRun: return event async def _terminal_locked( + self, + outcome: str, + **details: Any, + ) -> EvoRuntimeEvent: + # Freeze the first terminal intent before yielding. Agent timeout/cancel + # cannot cancel its compensation or substitute a different outcome. + return await asyncio.shield(self._start_terminal(outcome, **details)) + + def _start_terminal(self, outcome: str, **details: Any) -> asyncio.Task[EvoRuntimeEvent]: + if self._terminal_task is None: + self._terminal_task = asyncio.create_task( + self._finalize_terminal(outcome, **details), + name=f"evo-terminal-{self._run_id}", + ) + self._terminal_task.add_done_callback(self._observe_terminal) + return self._terminal_task + + def _observe_terminal(self, task: asyncio.Task[EvoRuntimeEvent]) -> None: + error = None if task.cancelled() else task.exception() + if task.cancelled() or error is not None: + self._terminal_error_code = ( + "RUN_TERMINAL_CANCELLED" if task.cancelled() else + "RUN_TERMINAL_CLEANUP_UNCONFIRMED" if self._terminal_phase == "cleanup" else + "RUN_TERMINAL_COMMIT_UNCONFIRMED") + logger.error("owner terminal task failed: %s", self._terminal_error_code) + + async def _finish_registry(self, outcome: str, checkpoint_id: str = "") -> None: + delay = 0.1 + while self._runtime.host_registry is not None: + try: + await asyncio.to_thread( + self._runtime.host_registry.finish, self.run_id, + outcome=outcome, checkpoint_id=checkpoint_id, + ) + return + except Exception: + # Product-owned compensation, never an Agent retry. Keep the + # claim and fixed outcome until the original owner confirms it. + logger.warning("registry finalization pending for %s", self.run_id) + await asyncio.sleep(delay) + delay = min(delay * 2, 5.0) + + async def _finalize_terminal( self, outcome: str, *, error_code: str | None = None, reason: str | None = None, error_details: Mapping[str, Any] | None = None, + checkpoint_details: Mapping[str, Any] | None = None, ) -> EvoRuntimeEvent: + # Terminalization is run-owned. Only public observers have deadlines; + # a slow close must not permanently fail the finalization task. + if self._stop_task is not None: + await asyncio.shield(self._stop_task) + await self._ensure_cleanup(timeout=None) + if self._runtime.host_registry is not None and not self._checkpoint_writes_stopped: + raise EvoRuntimeError("CHECKPOINT_WRITER_STOP_UNCONFIRMED") + self._terminal_phase = "commit" async with self._terminal_lock: if self._terminal_event is not None: return self._terminal_event @@ -2103,9 +2550,15 @@ class _EvoWebRun: "error_details": dict(error_details or {}), "purpose_attempt_counts": dict(self._attempt_counts), "provider_reserve_remaining_microunits": self._remaining, + **dict(checkpoint_details or {}), }, control_terminal=True, ) + await self._finish_registry( + outcome, str((checkpoint_details or {}).get("checkpoint_id") or "")) + if self._prepared.state == "CONSTRUCTION_FAILED": + async with self._runtime._registry_lock: + self._runtime._retire_prepared_locked(self._prepared) async with self._condition: self._state = "TERMINAL" self._terminal_event = event @@ -2155,7 +2608,13 @@ class _RuntimeAttemptCallback(AsyncCallbackHandler): purpose, route = self._route_for(metadata) try: callback_payload = _callback_messages_payload(messages) - input_bound = _provider_input_token_bound(callback_payload) + # Structured output and bound tools live in invocation params, not messages. + invocation = _kwargs.get("invocation_params", {}) + input_parameters = _effective_callback_input_parameters(invocation) + input_bound = _provider_input_token_bound({ + "messages": callback_payload, + **input_parameters, + }) if purpose == "main_agent" and self.run._force_context_repair_pending: self.run._force_context_repair_pending = False raise EvoRuntimeError( @@ -2535,6 +2994,12 @@ class _ProviderInputBound: def _decode_media_payload(payload: str, mime: str) -> tuple[bytes, str]: if payload.startswith("data:"): + # Bound the encoded allocation before split/decode using the same image + # ingress limit, not the ordinary JSON/schema text budget. + from EvoScientist.document_extract import MAX_IMAGE_BYTES + + if len(payload) > 4 * ((MAX_IMAGE_BYTES + 2) // 3) + 256: + raise EvoRuntimeError("TOKEN_BOUND_UNAVAILABLE") try: header, encoded = payload.split(",", 1) resolved_mime = header[5:].split(";", 1)[0] or mime @@ -2640,7 +3105,7 @@ def _project_media_for_bound(value: Any, metrics: dict[str, int]) -> Any: def _provider_input_token_bound(payload: Any) -> _ProviderInputBound: metrics = {"media_tokens": 0, "media_blocks": 0, "largest_media_bytes": 0} - projected = _project_media_for_bound(payload, metrics) + projected = _callback_input_parameters(payload, media_metrics=metrics) text_tokens = _payload_token_bound(projected) media_tokens = metrics["media_tokens"] return _ProviderInputBound( @@ -2652,6 +3117,170 @@ def _provider_input_token_bound(payload: Any) -> _ProviderInputBound: ) +_INPUT_PROJECTION_MAX_DEPTH = 64 +_INPUT_PROJECTION_MAX_NODES = 100_000 +_INPUT_PROJECTION_MAX_BYTES = 8 * 1024 * 1024 + + +def _effective_callback_input_parameters(invocation: Any) -> dict[str, Any]: + """Project already-flattened OpenAI callback params, then SDK shallow merge. + + ChatOpenAI._default_params expands model_kwargs; _get_invocation_params + overlays call kwargs. OpenAI BaseClient._build_request uses _merge_mappings + with extra_json last (not a recursive merge). A remaining model_kwargs is + therefore NOT a second defaults layer we can safely interpret. + Unknown body extensions and explicit alternative context transports fail closed. + Responses use_previous_response_id may generate an ID after this callback; + that SDK-managed context is outside this projection's visibility, not banned. + """ + fields = { + "tools", "functions", "response_format", "tool_choice", "function_call", + "system", "instructions", + } + if type(invocation) is not dict or any(type(key) is not str for key in invocation): + raise EvoRuntimeError("MODEL_INPUT_PROJECTION_INVALID") + if any(key in invocation for key in ( + "model_kwargs", "messages", "input", "text", "previous_response_id", + "conversation", "prompt", + )): + raise EvoRuntimeError("MODEL_INPUT_PROJECTION_INVALID") + extra = invocation.get("extra_body") + if extra is None: + extra = {} + # These non-input controls are emitted by adapter_registry/models. Validate + # their data without dropping or altering them in the actual SDK request. + controls = {"enable_thinking", "thinking_budget", "thinking"} + if type(extra) is not dict or any( + type(key) is not str or key not in fields | controls for key in extra + ): + raise EvoRuntimeError("MODEL_INPUT_PROJECTION_INVALID") + effective = {key: invocation[key] for key in fields if key in invocation} + effective.update(extra) + validated = _callback_input_parameters(effective) + return {key: value for key, value in validated.items() if key in fields} + + +def _callback_input_parameters( + value: Any, *, media_metrics: dict[str, int] | None = None +) -> Any: + """Validate JSON/schema trees before serialization; never format rejected values. + + Limits cover depth (root=0), visited nodes including keys, and aggregate + JSON scalar/container bytes. Shared acyclic values are counted each time. + BaseModel classes are trusted application code: schema hooks execute before + their output can be bounded. This is not a sandbox for untrusted classes. + """ + from pydantic import BaseModel + + active: set[int] = set() + nodes = 0 + size = 0 + + def visit(item: Any, depth: int, path: tuple[str, ...] = (), + media: tuple[tuple[str, ...], str] | None = None) -> Any: + nonlocal nodes, size + nodes += 1 + if depth > _INPUT_PROJECTION_MAX_DEPTH or nodes > _INPUT_PROJECTION_MAX_NODES: + raise ValueError + kind = type(item) + if item is None or kind in (bool, int, float, str): + if kind is str and media is not None and path == media[0]: + from EvoScientist.document_extract import MAX_IMAGE_BYTES, prepare_image_bytes + + if len(item) > 4 * ((MAX_IMAGE_BYTES + 2) // 3) + 256: + raise ValueError + raw, mime = _decode_media_payload(item, media[1]) + if not mime.startswith("image/"): + raise ValueError + # Parse using the ingress pixel/byte limits. Do not replace the + # actual request with the downsampled result or with the marker. + prepare_image_bytes(raw, "callback-image") + assert media_metrics is not None + media_metrics["media_tokens"] += (len(raw) + 2) // 3 + 512 + media_metrics["media_blocks"] += 1 + media_metrics["largest_media_bytes"] = max( + media_metrics["largest_media_bytes"], len(raw) + ) + item = f"" + elif kind is str and media_metrics is not None and item.startswith("data:"): + raise ValueError + if isinstance(item, float) and not math.isfinite(item): + raise ValueError + if isinstance(item, str) and len(item) > _INPUT_PROJECTION_MAX_BYTES: + raise ValueError + size += len(json.dumps(item, ensure_ascii=True, allow_nan=False)) + if size > _INPUT_PROJECTION_MAX_BYTES: + raise ValueError + return item + is_schema = isinstance(item, type) and issubclass(item, BaseModel) + if kind not in (dict, list) and not is_schema: + raise ValueError + identity = id(item) + if identity in active: + raise ValueError + active.add(identity) + try: + if isinstance(item, type) and issubclass(item, BaseModel): + return visit(item.model_json_schema(), depth + 1) + if not isinstance(item, (dict, list)): + raise ValueError + if len(item) > _INPUT_PROJECTION_MAX_NODES - nodes: + raise ValueError + size += 2 + 2 * len(item) + if size > _INPUT_PROJECTION_MAX_BYTES: + raise ValueError + if isinstance(item, list): + return [visit(child, depth + 1, path + ("[]",)) for child in item] + # Only content-list image blocks (or a direct block-list caller) can + # exempt payload bytes. Tools/schema objects never acquire this role. + content_block = path in { + ("[]",), + ("messages", "[]", "[]", "data", "content", "[]"), + ("messages", "[]", "content", "[]"), + } + if media_metrics is not None and content_block: + block_type = item.get("type") + mime = item.get("mime_type", "image/unknown") + if type(item.get("inline_data")) is dict: + inline_mime = item["inline_data"].get("mime_type") + if type(inline_mime) is str and inline_mime.startswith("image/"): + media = (path + ("inline_data", "data"), inline_mime) + if type(block_type) is str and block_type in {"image", "image_url", "input_image"}: + location = None + if type(item.get("base64")) is str: + location = ("base64",) + elif (type(item.get("image_url")) is dict + and type(item["image_url"].get("url")) is str + and item["image_url"]["url"].startswith("data:")): + location = ("image_url", "url") + elif (type(item.get("image_url")) is str + and item["image_url"].startswith("data:")): + location = ("image_url",) + elif type(item.get("url")) is str and item["url"].startswith("data:"): + location = ("url",) + elif type(item.get("source")) is dict and item["source"].get("type") == "base64": + location = ("source", "data") + mime = item["source"].get("media_type", mime) + if location is not None: + if type(mime) is not str: + raise ValueError + media = (path + location, mime) + result = {} + for key, child in item.items(): + if type(key) is not str: + raise ValueError + visit(key, depth + 1) + result[key] = visit(child, depth + 1, path + (key,), media) + return result + finally: + active.remove(identity) + + try: + return visit(value, 0) + except Exception: + raise EvoRuntimeError("MODEL_INPUT_PROJECTION_INVALID") from None + + def _callback_messages_payload(messages: Sequence[Sequence[Any]]) -> list[list[Any]]: """Convert LangChain callback messages to canonical-JSON-safe values."""