feat(hermes): prove the message boundary of a one-shot turn (issue #5 proposal 1)

docs/technical-reference.md

@@ -268,6 +268,21 @@ all-rows behavior (every row is a main-conversation call there). The
 column check matches SQLite semantics case-insensitively, so a schema
 declaring `"Task"` still triggers the filter.
 
+**Message boundary**: every row in the matched session must carry a
+timestamp inside the invocation window — a row outside it means the
+session saw activity this launch cannot account for (late finalization
+writes, a reused/continued session, a concurrent writer) and the turn
+fails closed. The turn is the final ACTIVE text-bearing assistant
+message; interim active drafts before it (tool narration, superseded
+text) and inactive tail rows (compaction ghosts) are normal one-shot
+noise, but ANY active row after the final assistant message — a
+follow-up user prompt, a tool/system row, an empty assistant stub —
+means the session continued beyond this turn and completion is
+refused. The accepted span (`first_message_id`, `final_message_id`,
+plus `session_row_count` covering every row in the session) is
+recorded in the proof as `message_boundary` so review can see exactly
+which rows bounded the turn.
+
 **Turn briefings**: every actor receives one `## Goal` / `## Checks` /
 `## Boundaries` / `## Report` task file (the exact sections
 `fable-session` requires; one shape serves every transport). From the

examples/fakes/bin/fake-hermes

@@ -28,6 +28,10 @@ a dirty terminal: ended_at set, end_reason='error'), FAKE_ENDED
 FAKE_API_CALLS (usage api_call_count, default 1), FAKE_STDOUT_LIE
 (stdout differs from the database), FAKE_EXTRA_SESSION (second session
 with the same source), FAKE_MULTI_MODEL (second observed model),
+FAKE_LATE_ASSISTANT (extra ACTIVE assistant row timestamped an hour
+after the run — outside the invocation window),
+FAKE_FOLLOWUP_USER (extra ACTIVE user row after the final assistant
+message — the session continued beyond the turn),
 FAKE_SPAWN_MARKER, FAKE_EXIT, FAKE_WORD_COUNT.
 """
 
@@ -189,6 +193,26 @@ def main() -> int:
             "api_call_count) VALUES (?, ?, ?, ?)",
             (session_id, model, provider, api_calls),
         )
+        if os.environ.get("FAKE_LATE_ASSISTANT"):
+            # A later ACTIVE text-bearing assistant row timestamped well
+            # outside the invocation window: the session saw activity the
+            # launch cannot account for.
+            con.execute(
+                "INSERT INTO messages (session_id, role, content, timestamp, "
+                "active) VALUES (?, ?, ?, ?, ?)",
+                (session_id, "assistant",
+                 "LATE assistant row — written after the invocation window.",
+                 started + 3600.0, 1),
+            )
+        if os.environ.get("FAKE_FOLLOWUP_USER"):
+            # An active user row AFTER the final assistant message: the
+            # session continued with a new prompt beyond this turn.
+            con.execute(
+                "INSERT INTO messages (session_id, role, content, timestamp, "
+                "active) VALUES (?, ?, ?, ?, ?)",
+                (session_id, "user", "One more thing: ...",
+                 started + 0.004, 1),
+            )
         if os.environ.get("FAKE_MULTI_MODEL"):
             con.execute(
                 "INSERT INTO session_model_usage (session_id, model, "

src/multi_agent_dialogue/adapters/hermes.py

@@ -15,7 +15,11 @@
 provider, observed model set, and the final ACTIVE assistant message
 from the ``sessions``, ``messages``, and ``session_model_usage`` tables
 — exactly one session with this turn's unique ``--source`` started
-inside the invocation window.
+inside the invocation window. The message boundary is proven too:
+every session row must sit inside the invocation window, the turn is
+the final ACTIVE text-bearing assistant message, and any active row
+after it (a continued session) fails closed; the accepted
+first/final message ids are recorded in the proof.
 
 Compatible Hermes session records may leave ``sessions.ended_at`` and
 ``sessions.end_reason`` NULL after a successful one-shot. Completion is
@@ -212,6 +216,7 @@ def execute(self, context: PrepareContext, packet: CommandPacket,
             "observed_models": observed["models"],
             "observed_providers": observed["providers"],
             "api_call_count": observed["api_call_count"],
+            "message_boundary": observed["message_boundary"],
             "invocation_window": [started_before, ended_after],
             # The proof names what actually established completion instead
             # of claiming a DB terminal state when none was persisted.
@@ -360,20 +365,77 @@ def _observe_session(db_path: Path, source: str, started_before: float,
                 )
 
             try:
-                final_row = con.execute(
-                    "SELECT content FROM messages WHERE session_id = ? "
-                    "AND role = 'assistant' AND active = 1 "
-                    "AND content IS NOT NULL AND content != '' "
-                    "ORDER BY id DESC LIMIT 1",
+                message_rows = con.execute(
+                    "SELECT id, role, content, timestamp, active "
+                    "FROM messages WHERE session_id = ? ORDER BY id",
                     (session_id,),
-                ).fetchone()
+                ).fetchall()
             except sqlite3.Error as exc:
                 raise AdapterError(f"{where}: cannot query messages: {exc}") from exc
-            if final_row is None or not str(final_row[0]).strip():
+
+            # Message boundary: every row this session contains must
+            # belong to THIS invocation. A row outside the window means
+            # the session saw activity the launch cannot account for
+            # (late finalization writes, a reused/continued session, a
+            # concurrent writer) and the boundary between "the turn" and
+            # "later activity" is unproven.
+            for msg_id, msg_role, _content, msg_ts, _active in message_rows:
+                try:
+                    ts = float(msg_ts)
+                except (TypeError, ValueError) as exc:
+                    raise AdapterError(
+                        f"{where}: session {session_id} message id "
+                        f"{msg_id} ({msg_role}) has no usable timestamp; "
+                        "the message boundary is unproven"
+                    ) from exc
+                if not (
+                    started_before - WINDOW_SLACK_SECONDS
+                    <= ts
+                    <= ended_after + WINDOW_SLACK_SECONDS
+                ):
+                    raise AdapterError(
+                        f"{where}: session {session_id} message id "
+                        f"{msg_id} ({msg_role}) sits outside this "
+                        "invocation window; the session saw activity this "
+                        "launch cannot account for and the message "
+                        "boundary is unproven"
+                    )
+
+            # The turn is the final ACTIVE text-bearing assistant message.
+            # Interim active drafts before it (tool narration, superseded
+            # text) are normal one-shot noise; an inactive tail row
+            # (compaction ghost) is normal too. What is NOT acceptable:
+            # ANY active row after it — a follow-up user prompt, a tool
+            # or system row, an empty assistant stub — because the session
+            # then continued past the answer and "the turn" no longer
+            # bounds the exchange.
+            final_id = None
+            final_text: str | None = None
+            for msg_id, msg_role, content, _ts, active in message_rows:
+                if (
+                    msg_role == "assistant"
+                    and active == 1
+                    and content is not None
+                    and str(content).strip()
+                ):
+                    final_id, final_text = msg_id, str(content)
+            if final_id is None or final_text is None:
                 raise AdapterError(
                     f"{where}: session {session_id} has no final active "
                     "assistant message; there is no turn to publish"
                 )
+            later_active = [
+                (msg_id, msg_role)
+                for msg_id, msg_role, _c, _t, active in message_rows
+                if active == 1 and msg_id > final_id
+            ]
+            if later_active:
+                raise AdapterError(
+                    f"{where}: session {session_id} shows active row(s) "
+                    f"{later_active} after the final assistant message id "
+                    f"{final_id}; the session continued beyond this turn "
+                    "and the message boundary is unproven"
+                )
             return {
                 "session_id": session_id,
                 "started_at": float(started_at),
@@ -382,7 +444,16 @@ def _observe_session(db_path: Path, source: str, started_before: float,
                 "models": sorted(models),
                 "providers": sorted(providers),
                 "api_call_count": api_calls,
-                "final_message": str(final_row[0]),
+                "final_message": final_text,
+                # Message-boundary span recorded in the proof so review
+                # can see exactly which rows bounded the turn. An empty
+                # session already failed the no-final-message check above,
+                # so message_rows is non-empty here.
+                "message_boundary": {
+                    "first_message_id": message_rows[0][0],
+                    "final_message_id": final_id,
+                    "session_row_count": len(message_rows),
+                },
             }
         finally:
             con.close()

tests/test_real_contracts.py

@@ -601,6 +601,46 @@ def test_incomplete_session_fails_closed(self) -> None:
             runner.launch(dialogue, "hermes-north")
         self.assertEqual(dialogue.state()["turn_index"], 0)
 
+    def test_late_assistant_message_fails_closed(self) -> None:
+        # Canary from the design review: a later ACTIVE assistant message
+        # outside the invocation window → the turn is rejected and nothing
+        # commits.
+        dialogue = self.make_dialogue(env_north={"FAKE_LATE_ASSISTANT": "1"})
+        with self.assertRaises(engine.ProtocolError):
+            runner.launch(dialogue, "hermes-north")
+        self.assertEqual(dialogue.state()["turn_index"], 0)
+        self.assertEqual(dialogue.state()["completed_turns"], [])
+
+    def test_followup_user_message_fails_closed(self) -> None:
+        dialogue = self.make_dialogue(env_north={"FAKE_FOLLOWUP_USER": "1"})
+        with self.assertRaises(engine.ProtocolError):
+            runner.launch(dialogue, "hermes-north")
+        self.assertEqual(dialogue.state()["turn_index"], 0)
+
+    def test_message_boundary_recorded_in_proof(self) -> None:
+        dialogue = self.make_dialogue()
+        runner.launch(dialogue, "hermes-north")
+        proof = self.evidence_for(dialogue, 0)["proof"]
+        boundary = proof["message_boundary"]
+        # user prompt + interim draft + final + inactive ghost tail.
+        self.assertEqual(boundary["session_row_count"], 4)
+        self.assertLess(
+            boundary["first_message_id"], boundary["final_message_id"]
+        )
+        # The final id is the max ACTIVE text-bearing assistant row; the
+        # inactive ghost tail never counts.
+        con = sqlite3.connect(proof["state_db"])
+        try:
+            row = con.execute(
+                "SELECT MAX(id) FROM messages WHERE session_id = ? "
+                "AND role = 'assistant' AND active = 1 "
+                "AND content IS NOT NULL AND content != ''",
+                (proof["session_id"],),
+            ).fetchone()
+        finally:
+            con.close()
+        self.assertEqual(boundary["final_message_id"], row[0])
+
 
 class HermesOneShotTerminalTests(HermesTestCase):
     """Compatibility when a clean one-shot leaves terminal fields NULL."""
@@ -800,6 +840,63 @@ def test_start_outside_window_is_rejected_even_with_null_terminal(self) -> None:
         self.build_db(started_at=self.before - 120.0)
         self.assert_refused("outside this")
 
+    # -- message boundary -------------------------------------------------
+
+    def add_message_row(self, role: str, content: str, timestamp: float,
+                        active: int = 1) -> None:
+        con = sqlite3.connect(self.db)
+        try:
+            con.execute(
+                "INSERT INTO messages (session_id, role, content, timestamp, "
+                "active) VALUES ('sess-1', ?, ?, ?, ?)",
+                (role, content, timestamp, active),
+            )
+            con.commit()
+        finally:
+            con.close()
+
+    def test_message_outside_invocation_window_is_rejected(self) -> None:
+        self.build_db()
+        self.add_message_row("assistant", "late write", self.after + 120.0)
+        self.assert_refused("outside this invocation window")
+
+    def test_user_message_after_final_assistant_is_rejected(self) -> None:
+        self.build_db()
+        self.add_message_row("user", "follow-up prompt", self.before + 2.0)
+        self.assert_refused("message boundary")
+
+    def test_active_tool_row_after_final_assistant_is_rejected(self) -> None:
+        # ANY active tail row — not just user prompts — breaks the
+        # boundary: the session continued past the answer.
+        self.build_db()
+        self.add_message_row("tool", "tool output", self.before + 2.0)
+        self.assert_refused("continued beyond this turn")
+
+    def test_inactive_tail_row_is_accepted(self) -> None:
+        # Compaction ghosts (inactive rows after the final message) are
+        # normal one-shot noise and stay accepted.
+        self.build_db()
+        self.add_message_row("assistant", "ghost", self.before + 2.0,
+                             active=0)
+        observed = self.observe()
+        self.assertEqual(observed["final_message"], "The final answer.")
+        self.assertEqual(observed["message_boundary"]["session_row_count"], 2)
+
+    def test_message_without_usable_timestamp_is_rejected(self) -> None:
+        self.build_db()
+        # REAL affinity stores unconvertible text as-is; float() then
+        # fails at observation time.
+        self.add_message_row("assistant", "x", "not-a-number")  # type: ignore[arg-type]
+        self.assert_refused("no usable timestamp")
+
+    def test_message_boundary_is_recorded(self) -> None:
+        self.build_db()
+        boundary = self.observe()["message_boundary"]
+        self.assertEqual(boundary["session_row_count"], 1)
+        self.assertEqual(
+            boundary["first_message_id"], boundary["final_message_id"]
+        )
+
 
 class HermesUsageTaskFilterTests(HermesTerminalFieldMatrixTests):
     """Auxiliary usage rows (title generation, vision, compression, ...)