diff --git a/docs/technical-reference.md b/docs/technical-reference.md index 306da54..ed34a96 100644 --- a/docs/technical-reference.md +++ b/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 diff --git a/examples/fakes/bin/fake-hermes b/examples/fakes/bin/fake-hermes index 238f557..bc7e6a3 100755 --- a/examples/fakes/bin/fake-hermes +++ b/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, " diff --git a/src/multi_agent_dialogue/adapters/hermes.py b/src/multi_agent_dialogue/adapters/hermes.py index c4a4285..e3695c2 100644 --- a/src/multi_agent_dialogue/adapters/hermes.py +++ b/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() diff --git a/tests/test_real_contracts.py b/tests/test_real_contracts.py index 74b711b..9d287ef 100644 --- a/tests/test_real_contracts.py +++ b/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, ...)