fix(gateway): exclude observations from accepted-turn ownership

evals/gateway_failure_ownership/README.md

@@ -51,7 +51,9 @@ separate identical pre-agent inputs, and a healthy follow-up.
 Two invariant tests are retained. The second uses real SQLite compaction archives
 and durable compression lineage with keyed/keyless input. It proves both parent
 and successor-only ownership; disabling successor lookup makes it RED (3 rows
-instead of 2). It does not run provider-driven compaction. Independent review
+instead of 2). A new ambient observed row must not masquerade as the accepted
+input; that negative control was also RED/GREEN verified. It does not run
+provider-driven compaction. Independent review
 identified baseline-read, content-projection, archive, and successor-only edge
 cases; ownership now avoids content comparison and includes durable lineage.
 

gateway/run_turn.py

@@ -1758,7 +1758,7 @@ class GatewayTurnMixin:
             if prepared.message_text is not None and session_entry is not None:
                 _rows = await self.async_session_store.load_transcript(prepared.persistence_session_id, raw=True)
                 _owned = any(
-                    row.get("role") == "user" and (
+                    row.get("role") == "user" and not row.get("observed") and (
                         row.get("platform_message_id") == str(event.message_id) if event.message_id
                         else row["id"] not in prepared.persisted_user_row_ids
                     ) for row in _rows

tests/gateway/test_failure_writer_ownership.py

@@ -100,6 +100,26 @@ def test_failure_owner_uses_launch_rows_across_compaction(tmp_path):
             )
             assert db.message_count() == before
             assert current_id in {r["id"] for r in store.load_transcript(sid, raw=True)}
+        source = SessionSource(
+            platform=Platform.TELEGRAM, chat_id="observed", user_id="fixture"
+        )
+        entry = store.get_or_create_session(source)
+        sid = entry.session_id
+        prepared = runner._PreparedTurn(
+            [], "", "accepted", "accepted", 1700000000, None, sid
+        )
+        db.append_message(sid, "user", "ambient observation", observed=True)
+        before = db.message_count()
+        await runner._hmwa_agent_error_reply(
+            RuntimeError("controlled pre-agent failure"),
+            MessageEvent(text="accepted", source=source, message_id=None),
+            source,
+            entry,
+            entry.session_key,
+            prepared,
+        )
+        assert db.message_count() == before + 1
+        assert db.get_messages(sid)[-1]["content"] == "accepted"
         db.close()
 
     asyncio.run(check())

website/docs/developer-guide/session-storage.md

@@ -54,7 +54,7 @@ to that turn. Separately accepted identical inputs remain separate, even with
 identical timestamps.
 
 Ownership reads include the launch session's compression lineage and archived
-compaction generations, but not undone rows. This covers both a successful
+compaction generations, but not undone rows or ambient observed messages. This covers both a successful
 parent write before rotation and a successor-only write after a failed parent
 flush. An unreadable keyless baseline stops the turn with the existing
 history-unavailable response rather than assuming an empty history. Normal