diff --git a/agent/turn_context.py b/agent/turn_context.py index 6f36e3524bf4..6ecb93a31343 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -518,8 +518,8 @@ def _stage_turn_user_message( # row; the model still receives role/content unchanged (api_messages strips both). if persist_user_display_kind: user_msg["display_kind"] = persist_user_display_kind - if persist_user_display_metadata: - user_msg["display_metadata"] = persist_user_display_metadata + if persist_user_display_metadata: + user_msg["display_metadata"] = persist_user_display_metadata # The platform message id survives the turn-start flush; restart drain-window # recovery dedups via ``has_platform_message_id`` against this row. if persist_user_platform_id is not None: diff --git a/evals/gateway_failure_ownership/README.md b/evals/gateway_failure_ownership/README.md new file mode 100644 index 000000000000..c08761fbc4a8 --- /dev/null +++ b/evals/gateway_failure_ownership/README.md @@ -0,0 +1,63 @@ +# Gateway failure-writer ownership + +Offline integration: `_handle_message` → session preparation and lease → production +`_run_agent` / TurnRunner → real AIAgent → loopback HTTP/SSE → on-disk SQLite → +gateway persistence/delivery → exception fallback. + +```sh +.venv/bin/python evals/gateway_failure_ownership/probe.py "$PWD" /tmp/ownership.json +scripts/run_tests.sh -j 1 tests/gateway/test_failure_writer_ownership.py +``` + +The first argument selects the production checkout, so the same fixture can A/B +another checkout without editing it. The receipt records its temporary HERMES_HOME. +Inherited credentials are cleared, tools are disabled, and socket connections are +restricted to loopback. No live messaging account or paid provider is contacted. + +## Fault boundaries and controls + +- Real HTTP 400 from the loopback peer; successful replies use HTTP/SSE. +- Controlled AIAgent constructor failure, or voice-policy failure after persistence. + The handler and exception writer remain production code. +- Chat A succeeds with ID `100`; chat B resumes that session and fails construction + with a different input and the same ID. Both inputs must remain, with unmodified + platform IDs for quote/reply lookup. +- A separate Python process commits an unrelated, nonobserved input during a keyless + constructor failure. Both writers' inputs must survive; the gateway lease is not + treated as universal writer authority. +- Separately accepted identical keyed/keyless inputs, identical timestamps, + same-delivery pre-agent retry, provider failure/recovery, and healthy follow-up. +- A normal history read fails closed before agent construction. There is no longer + a raw baseline to read or an extra keyless admission query. +- 5,000 archived 4-KiB rows remain in SQLite. The complete keyless failure handler + must stay below 8 MiB traced peak allocation, with zero raw transcript scans. +- Durable owner markers are unique across accepted inputs and absent from provider + wire messages. The foreign writer supplies its own independent marker. + +The second invariant uses real SQLite archives and a compression tree containing +an earlier `ws_orphan_reap` sibling and a live successor. It checks both published +reroutes and map-free restart routing; root/middle ancestors and successor writes +can establish ownership, while reaped, undone, observed, foreign-marker, and +unmarked rows cannot. These are storage-boundary checks, not provider-driven +compaction. + +## Receipts + +The expanded full-handler matrix has 20 checkpoints (19 deliveries plus marker +propagation). Base `869228cab4a8276d3b4c78da9d9939670c47bd0f`: **7/20**; reviewed +intermediate `653cd72ef63b013798748774a2509f453b566a77`: **1/20**; fixed: **20/20**. +All fault boundaries were reached. Base and fixed runs made 10 loopback requests +and recorded zero external connection attempts. Counts are cumulative exact-row +sequence checks, not independent defect counts. + +The reviewed intermediate lost chat B's input (1 row instead of 2), also lost the +keyless input beside the independent writer, and allocated 28,034,300 bytes in +the archived-history handler. The fixed run retained all 18 expected active user +rows (including the independent writer) and used 107,679 bytes in that handler. +The memory measurement excludes fixture seeding and receipt serialization. + +Exactly two invariant tests are retained. Eleven targeted files passed 63 tests; +broader directory suites and CI are not claimed. No historical rows are rewritten. +Existing rows without a marker cannot establish delivery ownership, so historical +redelivery is not deduplicated by this mechanism. This is exception-path input +arbitration, not global exactly-once delivery or content deduplication. diff --git a/evals/gateway_failure_ownership/foreign_writer.py b/evals/gateway_failure_ownership/foreign_writer.py new file mode 100644 index 000000000000..a7095416d901 --- /dev/null +++ b/evals/gateway_failure_ownership/foreign_writer.py @@ -0,0 +1,13 @@ +"""Independent durable writer used only by the loopback ownership probe.""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(sys.argv[1]).resolve())) +from hermes_state import SessionDB + +db = SessionDB(db_path=Path(sys.argv[2])) +db.append_message( + sys.argv[3], "user", "foreign independent input", observed=False, + display_metadata={"gateway_input_owner": "independent-process-owner"}, +) +db.close() diff --git a/evals/gateway_failure_ownership/probe.py b/evals/gateway_failure_ownership/probe.py new file mode 100644 index 000000000000..1589c260fd4c --- /dev/null +++ b/evals/gateway_failure_ownership/probe.py @@ -0,0 +1,362 @@ +import os, sys, json, asyncio, threading, tempfile, sqlite3, socket, subprocess, tracemalloc +from pathlib import Path +from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler +from types import SimpleNamespace + +ROOT = Path(sys.argv[1]).resolve() +RECEIPT = Path(sys.argv[2]).resolve() +HOME = Path(tempfile.mkdtemp(prefix="hermes-104653-state-")) +# Discard inherited credentials/config, preserve only interpreter essentials. +keep = {k: v for k, v in os.environ.items() if k in ("PATH", "LANG", "LC_ALL", "TZ")} +os.environ.clear() +os.environ.update(keep) +os.environ.update( + HOME=str(HOME), + HERMES_HOME=str(HOME), + HERMES_DISABLE_PLUGINS="1", + NO_PROXY="127.0.0.1,localhost", +) +sys.path.insert(0, str(ROOT)) +os.chdir(HOME) +(HOME / "config.yaml").write_text( + "model:\n provider: openai-compat\n default: fixture-model\n context_length: 131072\nagent:\n max_iterations: 2\ncompression:\n enabled: false\ndatabase:\n journal_mode: delete\n" +) +# Fence all network calls to loopback, including optional discovery/aux paths. +orig_connect = socket.socket.connect +blocked = [] + + +def local_connect(self, address): + if isinstance(address, tuple) and address[0] not in ( + "127.0.0.1", + "::1", + "localhost", + ): + blocked.append(str(address)) + raise OSError("fixture forbids external network") + return orig_connect(self, address) + + +socket.socket.connect = local_connect +requests = [] + + +class Peer(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + requests.append({"path": self.path, "body": body}) + users = [ + m.get("content") for m in body.get("messages", []) if m["role"] == "user" + ] + fail = fault == "http400" + if fail: + data = { + "error": { + "message": "fixture invalid request", + "type": "invalid_request_error", + "code": "fixture_rejected", + } + } + status = 400 + else: + data = { + "id": "chatcmpl-fixture", + "object": "chat.completion", + "created": 1, + "model": "fixture-model", + "choices": [ + { + "index": 0, + "message": {"role": "assistant", "content": "fixture reply"}, + "finish_reason": "stop", + } + ], + "usage": { + "prompt_tokens": 100, + "completion_tokens": 4, + "total_tokens": 104, + }, + } + status = 200 + if body.get("stream") and not fail: + chunks = [ + { + "id": "chatcmpl-fixture", + "object": "chat.completion.chunk", + "created": 1, + "model": "fixture-model", + "choices": [ + { + "index": 0, + "delta": {"role": "assistant", "content": "fixture reply"}, + "finish_reason": None, + } + ], + }, + { + "id": "chatcmpl-fixture", + "object": "chat.completion.chunk", + "created": 1, + "model": "fixture-model", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": data["usage"], + }, + ] + payload = ( + "".join("data: " + json.dumps(c) + "\n\n" for c in chunks) + + "data: [DONE]\n\n" + ).encode() + mime = "text/event-stream" + else: + payload = json.dumps(data).encode() + mime = "application/json" + self.send_response(status) + self.send_header("Content-Type", mime) + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + +server = ThreadingHTTPServer(("127.0.0.1", 0), Peer) +threading.Thread(target=server.serve_forever, daemon=True).start() +from run_agent import AIAgent +from hermes_state import SessionDB +from gateway.session import SessionStore, AsyncSessionStore, SessionSource +from gateway.config import GatewayConfig, Platform +from gateway.platforms.event import MessageEvent +from gateway.run import GatewayRunner + + +import gateway.run as gateway_run + +fault = None +faults = [] +runtime = dict( + api_key="fixture-only", + base_url=f"http://127.0.0.1:{server.server_port}/v1", + provider="openai-compat", + api_mode="chat_completions", +) +gateway_run._resolve_runtime_agent_kwargs = lambda: runtime +import run_agent + + +class ControlledAgent(AIAgent): + def __init__(self, *args, **kwargs): + if fault == "foreign-writer": + result = subprocess.run( + [sys.executable, str(Path(__file__).with_name("foreign_writer.py")), + str(ROOT), str(HOME / "state.db"), sid], + capture_output=True, text=True, stdin=subprocess.DEVNULL, timeout=20, + ) + assert result.returncode == 0, result.stderr + faults.append("foreign-writer-committed") + raise RuntimeError("controlled construction failure after independent write") + if fault in ("pre-agent", "archive-memory"): + faults.append("agent-construction") + raise RuntimeError("controlled agent initialization failure") + kwargs["enabled_toolsets"] = [] + super().__init__(*args, **kwargs) + + +run_agent.AIAgent = ControlledAgent +source = SessionSource( + platform=Platform.TELEGRAM, + chat_id="-100104653", + chat_type="group", + user_id="104653", +) +runner = GatewayRunner(GatewayConfig()) +runner._recover_telegram_topic_thread_id = lambda source: None +runner._is_user_authorized = lambda *a: True + + +def voice_policy(*args, **kwargs): + if fault == "post-agent": + faults.append("voice-policy-after-persistence") + raise RuntimeError("controlled voice policy failure") + return False + + +runner._should_send_voice_reply = voice_policy +from gateway.session_transcript import TranscriptReadError + +original_load = runner.session_store.load_transcript + + +raw_reads = [] +original_get_messages = SessionDB.get_messages + + +def observed_get_messages(self, session_id, *args, **kwargs): + if kwargs.get("include_inactive") or kwargs.get("include_compacted") or any(args[:2]): + raw_reads.append(session_id) + return original_get_messages(self, session_id, *args, **kwargs) + + +SessionDB.get_messages = observed_get_messages + + +def controlled_load(session_id, **kwargs): + if kwargs.get("raw"): + raw_reads.append(session_id) + if fault == "history-read": + faults.append("history-read") + raise TranscriptReadError(session_id) + return original_load(session_id, **kwargs) + + +runner.session_store.load_transcript = controlled_load +entry = runner.session_store.get_or_create_session(source) +sid = entry.session_id + + +def rows(): + with sqlite3.connect(HOME / "state.db") as conn: + conn.row_factory = sqlite3.Row + return [ + dict(row) + for row in conn.execute( + "SELECT id,role,content,platform_message_id,timestamp,display_metadata,active,compacted FROM messages " + "WHERE session_id=? ORDER BY id", + (sid,), + ) + ] + + +async def main(): + global fault, source, entry + from datetime import datetime + + observations = [] + expected = [] + cases = [ + ("chat A input", "100", None, True), + ("chat B distinct input", "100", "pre-agent", True), + ("same happy input", "101", None, True), + ("same happy input", "102", None, True), + ("synthetic happy", None, None, True), + ("synthetic happy", None, None, True), + ("failed provider input", "103", "http400", True), + ("after failure", "104", "post-agent", True), + ("failed provider input", "105", "http400", True), + ("synthetic post-agent", None, "post-agent", True), + ("init failure", "106", "pre-agent", True), + ("init failure", "106", "pre-agent", False), + ("init failure", "107", "pre-agent", True), + ("synthetic init failure", None, "pre-agent", True), + ("synthetic init failure", None, "pre-agent", True), + ("synthetic init failure", None, "history-read", False), + ("keyless independent-writer input", None, "foreign-writer", True), + ("bounded archive input", None, "archive-memory", True), + ("healthy follow-up", "108", None, True), + ] + # Deliberately identical timestamps: independently accepted keyless turns must survive too. + timestamp = datetime.fromtimestamp(1700000000) + for text, pid, fault, accepted in cases: + if text == "chat B distinct input": + from dataclasses import replace + source = replace(source, chat_id="chat-B") + other = runner.session_store.get_or_create_session(source) + entry = runner.session_store.switch_session(other.session_key, sid) + assert entry + event = MessageEvent( + text=text, + source=source, + message_id=pid, + internal=pid is None, + timestamp=timestamp, + ) + if fault == "archive-memory": + # Large archived payloads must stay in SQLite, not become a Python baseline. + with sqlite3.connect(HOME / "state.db") as conn: + conn.executemany( + "INSERT INTO messages (session_id, role, content, timestamp, active, compacted) " + "VALUES (?, 'user', ?, 1, 0, 1)", + ((sid, "x" * 4096) for _ in range(5000)), + ) + tracemalloc.start() + before_raw = len(raw_reads) + before_requests = len(requests) + before_faults = len(faults) + runner._evict_cached_agent(entry.session_key) + response = await runner._handle_message(event) + peak = None + if fault == "archive-memory": + _, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + if fault == "foreign-writer": + expected.append(("foreign independent input", None)) + if accepted: + expected.append((text, pid)) + actual = [ + (r["content"], r["platform_message_id"]) + for r in rows() + if r["role"] == "user" and r["active"] + ] + hit = faults[before_faults:] + if fault == "history-read": + reached = ( + hit == ["history-read"] + and len(requests) == before_requests + and "history is temporarily unavailable" in response + ) + elif fault == "foreign-writer": + reached = hit == ["foreign-writer-committed"] and len(requests) == before_requests + elif fault in ("pre-agent", "archive-memory"): + reached = hit == ["agent-construction"] and len(requests) == before_requests + elif fault == "post-agent": + reached = ( + hit == ["voice-policy-after-persistence"] + and len(requests) > before_requests + ) + else: + reached = len(requests) > before_requests + observations.append( + dict( + text=text, + platform_id=pid, + fault=fault, + accepted=accepted, + expected_users=len(expected), + actual_users=len(actual), + pass_=actual == expected and len(raw_reads) == before_raw + and (peak is None or peak < 8 * 1024 * 1024), + raw_reads=len(raw_reads) - before_raw, + archive_peak_bytes=peak, + reached=reached, + faults=hit, + requests=len(requests) - before_requests, + response=response, + ) + ) + users = [r for r in rows() if r["role"] == "user" and r["active"]] + markers = [json.loads(r["display_metadata"] or "{}").get("gateway_input_owner") for r in users] + observations.append(dict( + text="durable marker propagation", reached=bool(users), + pass_=all(isinstance(m, str) and m for m in markers) and len(set(markers)) == len(markers) + and all("display_metadata" not in m for request in requests + for m in request["body"].get("messages", [])), + )) + receipt = dict( + home=str(HOME), + observations=observations, + rows=[r for r in rows() if r["active"]], + requests=len(requests), + blocked_external_attempts=blocked, + passed=sum(o["pass_"] and o["reached"] for o in observations), + total=len(observations), + ) + RECEIPT.write_text(json.dumps(receipt, indent=2)) + print(json.dumps(receipt, indent=2)) + return receipt["passed"] == receipt["total"] + + +try: + passed = asyncio.run(main()) +finally: + server.shutdown() +sys.exit(0 if passed else 1) diff --git a/gateway/run_turn.py b/gateway/run_turn.py index 3a861ba63200..b42ba50eb5d8 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -1607,6 +1607,8 @@ def _hmwa_user_transcript_entry(event, prepared, ts): } if prepared.persist_user_display_kind: _user_entry["display_kind"] = prepared.persist_user_display_kind + if prepared.persistence_owner: + _user_entry["display_metadata"] = {"gateway_input_owner": prepared.persistence_owner} if getattr(event, "message_id", None): _user_entry["message_id"] = str(event.message_id) return _user_entry @@ -1753,22 +1755,13 @@ async def _hmwa_agent_error_reply(self, e, event, source, session_entry, session # Retain Slack thread/workspace routing so a failed turn cannot leave its status visible. await self._hmwa_stop_typing_for_turn(event, source) logger.exception("Agent error in session %s", session_key) - # Failures before run_conversation() (provider/httpx init) can't persist the inbound turn: - # append the user message here once, unless the latest user row already matches it. + # Replay can coalesce inputs; only this input's durable marker establishes ownership. try: if prepared.message_text is not None and session_entry is not None: - try: - _recent_transcript = await self.async_session_store.load_transcript(session_entry.session_id) - except Exception: - _recent_transcript = [] - _expected_user_content = ( - prepared.persist_user_message if prepared.persist_user_message is not None - else prepared.message_text - ) - _last_user = next( - (_msg for _msg in reversed(_recent_transcript[-10:]) if _msg.get("role") == "user"), None, + _owned = await self.async_session_store.has_input_owner( + prepared.persistence_session_id, prepared.persistence_owner, ) - if _last_user is None or _last_user.get("content") != _expected_user_content: + if not _owned: await self.async_session_store.append_to_transcript( session_entry.session_id, self._hmwa_user_transcript_entry(event, prepared, time.time()), ) @@ -1824,6 +1817,8 @@ class _PreparedTurn: persist_user_message: Any persist_user_timestamp: Any persist_user_display_kind: Optional[str] + persistence_session_id: Optional[str] = None + persistence_owner: Optional[str] = None async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _quick_key, run_generation): """Everything between session resolution and the agent run: session open, task-local env, @@ -1867,6 +1862,9 @@ async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _q # from []. Restore task-local context here (before the broad cleanup finally). try: history = await self.async_session_store.load_transcript(session_entry.session_id) + history = await self._hmwa_run_session_hygiene( + event, source, session_entry, session_key, history, _quick_key, run_generation, + ) except TranscriptReadError: self._clear_session_env(_session_env_tokens) return ( @@ -1875,10 +1873,6 @@ async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _q "Use /reset only if you intentionally want to start a new conversation." ), _session_env_tokens - history = await self._hmwa_run_session_hygiene( - event, source, session_entry, session_key, history, _quick_key, run_generation, - ) - await self._hmwa_first_contact_notes(source, history, turn_sidecar_notes) # Voice channel state rides the user message ONLY when changed (in the system prompt it @@ -1906,9 +1900,16 @@ async def _hmwa_prepare_turn(self, event, source, session_entry, session_key, _q # Bind this run generation to the adapter so deferred post-delivery callbacks are released # by the run that registered them. self._bind_adapter_run_generation(self._adapter_for_source(source), session_key, run_generation) + # Delivery IDs are only unique in their transport namespace. Keyless turns + # need their own identity, even when another process writes to this session. + import uuid + namespace = [source.platform.value, source.profile, source.scope_id, + source.chat_id, source.thread_id, str(event.message_id)] + owner = (str(uuid.uuid5(uuid.NAMESPACE_URL, json.dumps(namespace))) + if event.message_id else str(uuid.uuid4())) return self._PreparedTurn( history, context_prompt, message_text, persist_user_message, persist_user_timestamp, - persist_user_display_kind, + persist_user_display_kind, session_entry.session_id, owner, ), _session_env_tokens async def _handle_message_with_agent(self, event, source, _quick_key: str, run_generation: int): @@ -1959,6 +1960,7 @@ async def _handle_message_with_agent(self, event, source, _quick_key: str, run_g persist_user_message=prepared.persist_user_message, persist_user_timestamp=prepared.persist_user_timestamp, persist_user_display_kind=prepared.persist_user_display_kind, + persist_user_display_metadata={"gateway_input_owner": prepared.persistence_owner}, message_type=event.message_type, ) _turn_seconds = time.monotonic() - _turn_started_monotonic @@ -3773,6 +3775,7 @@ async def _run_agent_inner( channel_prompt: Optional[str] = None, moa_config: Optional[dict] = None, persist_user_message: Optional[Any] = None, persist_user_timestamp: Optional[float] = None, persist_user_display_kind: Optional[str] = None, message_type: Optional[str] = None, + persist_user_display_metadata: Optional[dict] = None, ) -> Dict[str, Any]: """Run the agent; returns the full run_conversation result dict. @@ -3796,6 +3799,7 @@ async def _run_agent_inner( persist_user_message=persist_user_message, persist_user_timestamp=persist_user_timestamp, persist_user_display_kind=persist_user_display_kind, + persist_user_display_metadata=persist_user_display_metadata, ) _status_thread_metadata = self._run_agent_bind_turn_wiring( turn_ctx, turn_runner, source, event_message_id, disp._native_slack_task_cards, diff --git a/gateway/run_turn_runner.py b/gateway/run_turn_runner.py index 6449454b19fe..7d7284e420c6 100644 --- a/gateway/run_turn_runner.py +++ b/gateway/run_turn_runner.py @@ -1430,6 +1430,8 @@ def _run_conversation_with_approval(self, agent, agent_history, observed_group_c # Internal self-injected turn: type the persisted user row so UIs render it as a # timeline notice, not a user bubble (stripped from provider payloads downstream). kwargs["persist_user_display_kind"] = ctx.persist_user_display_kind + if ctx.persist_user_display_metadata: + kwargs["persist_user_display_metadata"] = ctx.persist_user_display_metadata if ctx.moa_config is not None: kwargs["moa_config"] = ctx.moa_config if persist_user_timestamp_override is not None: diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index ff9e795d01fe..50ade91a3a73 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -453,6 +453,29 @@ def rewrite_transcript( self._clear_dirty_transcript(session_id) return True + def has_input_owner(self, session_id: str, owner: str) -> bool: + """Find this accepted input on the canonical live continuation and its ancestors. + + Content and unrelated writers cannot establish ownership. Query only existence; + compaction archives can contain many megabytes that replay never needs to load. + """ + try: + current = self._follow_reroutes(session_id) + db = self._db_for_session_id(current) + current = db.get_compression_tip(current) or current + seen = set() + while current and current not in seen: + seen.add(current) + if db.has_gateway_input_owner(current, owner): + return True + row = db.get_session(current) + if not row or not db._is_compression_child_row(row): + break + current = row["parent_session_id"] + return False + except Exception as e: + raise TranscriptReadError(session_id) from e + def load_transcript(self, session_id: str) -> List[Dict[str, Any]]: """Load all messages from a session's transcript (state.db is canonical). Reads follow the same routing writes use — the in-memory reroute map, then the durable compression tip — diff --git a/gateway/turn_context.py b/gateway/turn_context.py index bfe807ca6560..aa2ce650b918 100644 --- a/gateway/turn_context.py +++ b/gateway/turn_context.py @@ -56,6 +56,7 @@ class TurnContext: # display_kind of the persisted user row for a self-injected turn; DB-only, never sent. # "internal_notification" for async-delegation/background notifications (#82888). persist_user_display_kind: Optional[str] = None + persist_user_display_metadata: Optional[dict] = None user_config: Any = None enabled_toolsets: Any = None disabled_toolsets: Any = None diff --git a/hermes_state_messages.py b/hermes_state_messages.py index 07cb4c69ee57..c81ec6e8cb30 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -1048,6 +1048,15 @@ def message_count(self, session_id: str = None) -> int: sql = "SELECT COUNT(*) FROM messages" + (" WHERE session_id = ?" if session_id else "") return self._read_one(sql, (session_id,) if session_id else ())[0] + def has_gateway_input_owner(self, session_id: str, owner: str) -> bool: + """Probe the accepted-input marker without allocating message bodies or archives.""" + return self._read_one( + "SELECT 1 FROM messages WHERE session_id = ? AND role = 'user' " + "AND observed = 0 AND (active = 1 OR compacted = 1) " + "AND CASE WHEN json_valid(display_metadata) " + "THEN json_extract(display_metadata, '$.gateway_input_owner') END = ? LIMIT 1", + (session_id, owner)) is not None + def has_platform_message_id(self, session_id: str, platform_message_id: str) -> bool: """True when *platform_message_id* exists (partial-index probe; the gateway's transient-failure dedupe). diff --git a/tests/gateway/test_failure_writer_ownership.py b/tests/gateway/test_failure_writer_ownership.py new file mode 100644 index 000000000000..cc10b6ed402a --- /dev/null +++ b/tests/gateway/test_failure_writer_ownership.py @@ -0,0 +1,109 @@ +"""Accepted inbound turns have one durable owner, including exception fallback.""" + +import json +import subprocess +import sys +from pathlib import Path + + +def test_gateway_failure_writer_preserves_accepted_turn_identity(tmp_path): + root = Path(__file__).resolve().parents[2] + receipt = tmp_path / "receipt.json" + result = subprocess.run( + [ + sys.executable, + str(root / "evals/gateway_failure_ownership/probe.py"), + str(root), + str(receipt), + ], + capture_output=True, + text=True, + stdin=subprocess.DEVNULL, + timeout=90, + ) + assert receipt.exists(), result.stdout + result.stderr + data = json.loads(receipt.read_text()) + assert all(row["reached"] for row in data["observations"]), data + assert data["passed"] == data["total"], data["observations"] + assert result.returncode == 0, result.stdout + result.stderr + + +def test_failure_owner_follows_only_live_lineage_markers(tmp_path): + import asyncio + import sqlite3 + from gateway.config import GatewayConfig, Platform + from gateway.platforms.event import MessageEvent + from gateway.run import GatewayRunner + from gateway.session import SessionSource, SessionStore + + async def check(): + store = SessionStore(tmp_path / "sessions", GatewayConfig()) + runner = object.__new__(GatewayRunner) + runner.session_store = store + + async def stop_typing(event, source): + return None + + runner._hmwa_stop_typing_for_turn = stop_typing + # An archived ancestor or the published live child can own a turn. Neither + # a reaped sibling nor undone/observed/foreign-marker rows can own it. + cases = ("ancestor", "middle-ancestor", "live-child", "reaped", "undone", "observed", "foreign", "unmarked") + for index, (location, pid) in enumerate( + (location, pid) for location in cases for pid in (None, "100") + ): + source = SessionSource( + platform=Platform.TELEGRAM, chat_id=f"fixture-{index}", user_id="fixture" + ) + entry = store.get_or_create_session(source) + sid = entry.session_id + db = store._db_for_session_id(sid) + owner = f"accepted-owner-{index}" + metadata = {"gateway_input_owner": owner} + prepared = runner._PreparedTurn( + [], "", "same", [{"type": "text", "text": "same"}], + 1700000000, None, sid, owner, + ) + db.append_message(sid, "user", "same [screenshot]") + middle = sid + "-middle" + db.create_session(middle, source="telegram", parent_session_id=sid) + orphan = sid + "-orphan" + child = sid + "-live" + db.create_session(orphan, source="telegram", parent_session_id=middle) + db.end_session(orphan, "ws_orphan_reap") + db.create_session(child, source="telegram", parent_session_id=middle) + target = {"ancestor": sid, "middle-ancestor": middle, "reaped": orphan}.get(location, child) + marker = {"gateway_input_owner": "foreign-writer"} if location == "foreign" else metadata + current_id = db.append_message( + target, "user", "same [screenshot]", platform_message_id=pid, + observed=location == "observed", + display_metadata=None if location == "unmarked" else marker, + ) + db.end_session(sid, "compression") + db.end_session(middle, "compression") + if location in ("ancestor", "middle-ancestor", "undone"): + with sqlite3.connect(db.db_path) as conn: + conn.execute( + "UPDATE messages SET active=0, compacted=? WHERE id=?", + (int(location != "undone"), current_id), + ) + store._publish_transcript_reroute(sid, child) + owned = location in ("ancestor", "middle-ancestor", "live-child") + assert store.has_input_owner(sid, owner) is owned, location + # A restarted store has no published map and must pick the same live child. + store._transcript_reroutes.clear() + assert store.has_input_owner(sid, owner) is owned, location + before = db.message_count() + await runner._hmwa_agent_error_reply( + RuntimeError("controlled post-compaction failure"), + MessageEvent(text="same", source=source, message_id=pid), + source, entry, entry.session_key, prepared, + ) + assert db.message_count() == before + (not owned), location + assert store.has_input_owner(sid, owner), location + if not owned: + latest = db.get_messages(child)[-1] + assert latest["content"] == prepared.persist_user_message + assert latest["display_metadata"]["gateway_input_owner"] == owner + db.close() + + asyncio.run(check()) diff --git a/website/docs/developer-guide/session-storage.md b/website/docs/developer-guide/session-storage.md index 09fcaca45065..23cdb2b412a5 100644 --- a/website/docs/developer-guide/session-storage.md +++ b/website/docs/developer-guide/session-storage.md @@ -42,6 +42,29 @@ This also applies to synthetic/keyless input; it does not depend on a platform message ID. Existing historical duplicates are not rewritten. The gateway skips its transcript write when the agent reports that it owns persistence. +## Gateway exception-path input ownership + +A gateway exception can occur before agent construction or after its input reaches +SQLite. The gateway gives the accepted input an owner marker in the existing +`display_metadata` sidecar and passes it through the agent's normal persistence +path. Provider messages never contain this metadata. Platform markers namespace +the inbound message ID by platform, profile, scope, chat, and thread; the original +`platform_message_id` remains unchanged for quote/reply resolution. Keyless turns +receive a fresh marker, even for identical text and timestamps. + +The exception writer probes only for that marker, following the published reroute +and canonical live compression successor, then compression ancestors. Active rows +and compaction archives count; undone rows, observed input, and unrelated writers +do not. An unrelated process writing the same session cannot suppress this turn. +No whole-history baseline or archived message-body allocation is needed. Failed +ownership reads do not authorize a speculative append; ordinary history-read +failures retain the existing history-unavailable response. + +Normal agent-owned persistence is unchanged. This is failure-writer arbitration, +not universal exactly-once delivery, content deduplication, or a schema migration. +Historical rows are not rewritten; unmarked historical inputs cannot establish +ownership for a redelivered event. + ## Architecture Overview ```