From 31c7278e5d43d0795b706807f82778b1664bfc4d Mon Sep 17 00:00:00 2001 From: Pavel Tkachyk Date: Sun, 30 Aug 2026 20:43:38 -0400 Subject: [PATCH] Minimize Hermes capture payloads Signed-off-by: Pavel Tkachyk --- CHANGELOG.md | 8 + README.md | 4 +- docs/extraction-manifest.json | 28 +-- scripts/verify_fresh_migration_run.py | 7 +- scripts/verify_public_plugin_candidate.py | 32 ++- src/substrate_wiki/README.md | 8 +- src/substrate_wiki/__init__.py | 229 ++++------------------ src/substrate_wiki/events.py | 83 ++------ src/substrate_wiki/history.py | 123 +++++------- src/substrate_wiki/plugin.yaml | 3 - tests/test_hardening.py | 80 +++----- tests/test_history.py | 4 +- tests/test_history_replay.py | 23 +-- tests/test_memory_provider.py | 12 +- tests/test_packaging.py | 10 +- 15 files changed, 216 insertions(+), 438 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6642e22..c9c7469 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## Unreleased + +- Upload only completed user/assistant text; exclude tool calls, tool results, system messages, + memory writes, session boundaries, provider scope, hashes, retention metadata, and duplicate + envelope fields. +- Batch historical dialogue up to the request-size limit instead of posting one message per + request, while retaining deterministic IDs and resumable checkpoints. + ## 2.0.3 - Keep approved device polling active across transient hosted transport and edge failures. diff --git a/README.md b/README.md index 180c038..f66d813 100644 --- a/README.md +++ b/README.md @@ -172,7 +172,7 @@ Tenant credentials are stored in native credential custody, with an owner-privat - Implements Hermes provider lifecycle hooks and tools. - Prefetches bounded cited `memory_card` results for automatic recall. -- Captures completed turns, memory writes, and session boundaries. +- Captures only completed user/assistant dialogue turns. - Redacts credential-shaped values before durable local spooling or network transfer. - Replays Hermes history with deterministic event IDs and durable checkpoints. - Keeps spool/checkpoint state beneath the active `$HERMES_HOME/substrate_wiki` profile. @@ -181,7 +181,7 @@ It does **not** expose arbitrary filesystem writes, include Substrate server cod ## Privacy boundary -Visible prompts, assistant output, tool calls, and tool results may be sent to the configured Substrate server after redaction. Redaction is defense in depth, not proof arbitrary sensitive prose is absent. Review the server's access and retention policy before enabling capture. +Visible prompts and assistant output may be sent to the configured Substrate server after redaction. Tool calls, tool results, system messages, memory-write events, session metadata, and provider scope are not uploaded. Redaction is defense in depth, not proof arbitrary sensitive prose is absent. Local failed deliveries remain in a bounded owner-private spool until delivered or explicitly removed. Status, progress, and receipts are content-free. diff --git a/docs/extraction-manifest.json b/docs/extraction-manifest.json index b602541..1bdb7ae 100644 --- a/docs/extraction-manifest.json +++ b/docs/extraction-manifest.json @@ -59,7 +59,7 @@ "class": "standalone_repository_policy_or_test", "path": "CHANGELOG.md", "reason": "Required only by the independent public repository.", - "sha256": "57f9eb855dc41b263892d8229fa9fed046420da8a67d65fc34bd93f02c9eaa9c" + "sha256": "d20f1839056fdc49b98ed686a9a6747267bc62026eb13603577d90a6fbba4f9b" }, { "class": "standalone_repository_policy_or_test", @@ -77,7 +77,7 @@ "class": "standalone_repository_policy_or_test", "path": "README.md", "reason": "Required only by the independent public repository.", - "sha256": "20ac49480369c8531e001466a6537a7dae310bbba7e6079bc217a6181e7d69d7" + "sha256": "1cce63f45c357df4710e24a8351835c4382a47751d57edf4207e234eb350807b" }, { "class": "standalone_repository_policy_or_test", @@ -125,7 +125,7 @@ "class": "standalone_repository_policy_or_test", "path": "scripts/verify_fresh_migration_run.py", "reason": "Required only by the independent public repository.", - "sha256": "49fc5a2ee61eea6bd5d3d4f83ffd6b261bee10450ccbe8b2fde7c9873f89926d" + "sha256": "e185b0328bb7d684a53a57ce1029d8f2a845a608395b087ecff36d372a3ef210" }, { "class": "standalone_plugin_implementation", @@ -353,7 +353,7 @@ { "class": "build-and-install", "destination": "scripts/verify_public_plugin_candidate.py", - "destination_sha256": "bcb89f657bc719e76f7873f71bf5bd302c06fbedd53a3162e6de64337e01b8db", + "destination_sha256": "e0241e4cc230dd50f88a01d23f4c22dac3aff54343e9f3b34ce2ca0906d57e92", "source": "scripts/verify_public_plugin_candidate.py", "source_sha256": "4130935d530075fce1758e2e89bd5d973a722e2293b5b1058cfe0d17f326172b", "transformation": "modified_for_standalone" @@ -361,7 +361,7 @@ { "class": "plugin-package", "destination": "src/substrate_wiki/README.md", - "destination_sha256": "44a11fda85d6d8170d772beadba2149bd114624479344b69a69cd49634309204", + "destination_sha256": "88d492b33ee537567bf85b63dcc7a470319212ac377f4857e27e504a3a1a05a3", "source": "hermes-plugin/substrate_wiki/README.md", "source_sha256": "21b6e60ef34e408fd87570334ba6d2811d2a9ea3c95d512e4d625f958b1cb0dc", "transformation": "modified_for_standalone" @@ -369,7 +369,7 @@ { "class": "plugin-package", "destination": "src/substrate_wiki/__init__.py", - "destination_sha256": "425cd191f723805bed85d965d2d74daeac272dd0660eadf6bac92f1fa02d6f4a", + "destination_sha256": "71c0c19d366759e67c7b999029777295cc8c25503d35fe9cfddc9460bcb34d5a", "source": "hermes-plugin/substrate_wiki/__init__.py", "source_sha256": "a4143022e05a7b93d3aa5799f4319601292e67069ad85d172770c3effe0e5e9d", "transformation": "copied" @@ -401,7 +401,7 @@ { "class": "plugin-package", "destination": "src/substrate_wiki/events.py", - "destination_sha256": "675ba567a391f6b86688876ae26d1cfd9b567fe3312d9ddc03620775596c88e9", + "destination_sha256": "f70c743866c88e3aa9add5746ae558adab175175c31918109fb3ea90dd52c662", "source": "hermes-plugin/substrate_wiki/events.py", "source_sha256": "675ba567a391f6b86688876ae26d1cfd9b567fe3312d9ddc03620775596c88e9", "transformation": "copied" @@ -409,7 +409,7 @@ { "class": "plugin-package", "destination": "src/substrate_wiki/history.py", - "destination_sha256": "7b2065ab5d7b3176b82d18d0a25b8c726c6d6311f9adf2fb2486bfdf8678df8f", + "destination_sha256": "aa4216c382ecd6816aa6b5633e7143a71fb8af66a218a0fafba9837a6d6da28d", "source": "hermes-plugin/substrate_wiki/history.py", "source_sha256": "d758d1b89dad334fc8c352b4e30f6e8e3c0788821c82ecc11bc7e7fcefdef16c", "transformation": "copied" @@ -417,7 +417,7 @@ { "class": "plugin-package", "destination": "src/substrate_wiki/plugin.yaml", - "destination_sha256": "0e1c680eb2822ca6c58315bd91a058468d107065fde0881ab39aafe54bd13690", + "destination_sha256": "e76721f0d7ba3d37958d15fa398ea47bb7c457388e0ad181df5471a1b2ede61c", "source": "hermes-plugin/substrate_wiki/plugin.yaml", "source_sha256": "5bfc8b20bfe99b8dd4fcb35616724e9a4a93bdfd1f3b270cfea1ccff18eaa116", "transformation": "modified_for_standalone" @@ -489,7 +489,7 @@ { "class": "plugin-tests", "destination": "tests/test_hardening.py", - "destination_sha256": "f5f87125f1edd37bff1d44301d6bb0f44cc7faf3ba6122bdbe7569f349fea7a3", + "destination_sha256": "bb9825c1889d919e29a90e43c2be48b84d846fcad813fa3896d194491cc8f386", "source": "tests/contract/test_hermes_plugin_hardening.py", "source_sha256": "6066ff88351167647dc6ffd368fcb2ad666ee722fcd16d2fc04600f274151b17", "transformation": "modified_for_standalone" @@ -497,7 +497,7 @@ { "class": "plugin-tests", "destination": "tests/test_history.py", - "destination_sha256": "3997fc5c3b538df4ded9e994781fd8384b6b3b42ac695e1f1450ac24f6a92e7c", + "destination_sha256": "15a6e7778a92e0d826ba1af61032f222dbf926b5ec2e189c48d7985a70824f19", "source": "tests/contract/test_hermes_history_v12.py", "source_sha256": "dc5cf0ceca8b8c0156bba330f51cc2186fdd2c409861752a7465f1b625228d69", "transformation": "modified_for_standalone" @@ -505,7 +505,7 @@ { "class": "plugin-tests", "destination": "tests/test_history_replay.py", - "destination_sha256": "87272a47ab814803a77a39cafeb1b193b48678bc889b49712b11bace8b7d8c87", + "destination_sha256": "226d30b1ad98ba7603bf3d8f7c86ac0857bf596bb75453d0dd57d677ec1405bd", "source": "tests/contract/test_hermes_history_replay.py", "source_sha256": "ef7636f612cc2d1023856fb8ff39dd7de2cca13336d20b86db5d25d944a3a212", "transformation": "modified_for_standalone" @@ -521,7 +521,7 @@ { "class": "plugin-tests", "destination": "tests/test_memory_provider.py", - "destination_sha256": "2c4517847dfad341063a69afcc737316a74574471fc98a45eaff21cfe4e271fd", + "destination_sha256": "cc967199b8e877a8088937a0c951e80f188ad4c7f29a98067bf649a2f8500b72", "source": "tests/contract/test_hermes_memory_provider.py", "source_sha256": "d95c3c68592e7c45d4eab3c96c8a23726e8ba54116b205b33f78c66239a4ba6e", "transformation": "modified_for_standalone" @@ -537,7 +537,7 @@ { "class": "plugin-tests", "destination": "tests/test_packaging.py", - "destination_sha256": "a3a24a5c987c939514866e72b70d01a32e34f5771089153e1fd12013829d8baf", + "destination_sha256": "9336c62a8011bf4b469574a27dfdee0abe3b1037a7d78baaf95f2840a16c6b51", "source": "tests/contract/test_hermes_plugin_packaging.py", "source_sha256": "c54967830788555b8d7bd5d2f871baeb6b98d7804a2bc7410ab297078bc20ee7", "transformation": "modified_for_standalone" diff --git a/scripts/verify_fresh_migration_run.py b/scripts/verify_fresh_migration_run.py index 8f611e6..974874e 100644 --- a/scripts/verify_fresh_migration_run.py +++ b/scripts/verify_fresh_migration_run.py @@ -56,9 +56,14 @@ def main() -> int: if not isinstance(case_budget, dict): failures.append(f"missing retained budget for case {case_id}") continue + # The retained count reflects the former all-message payload. With the + # minimal user/assistant contract, redaction remains an integrity + # presence check rather than a historical message-volume floor. + current_budget = dict(case_budget) + current_budget["min_redacted_events"] = 1 failures.extend( f"{case_id}/c{run['concurrency']}: {failure}" - for failure in evaluate_budget(run, case_budget) + for failure in evaluate_budget(run, current_budget) ) if failures: raise ValueError("; ".join(failures)) diff --git a/scripts/verify_public_plugin_candidate.py b/scripts/verify_public_plugin_candidate.py index 0f8b62a..b33fa2d 100644 --- a/scripts/verify_public_plugin_candidate.py +++ b/scripts/verify_public_plugin_candidate.py @@ -81,6 +81,7 @@ "8510e5bbf41a3e79a6d5fee1f7816db734307fe7e2a7786b6a61c1ba280520a5", "8793ef3bbab749be6e1089034ba627d24c8b733ffd4311951b590dae01d2ca02", "c8eb7c5157ac7027dbf6ae86235e63dea6a2dd1faab46f0cbcbd536fbdd21ecf", + "1cce63f45c357df4710e24a8351835c4382a47751d57edf4207e234eb350807b", } ), "scripts/benchmark_migration.py": frozenset( @@ -90,16 +91,28 @@ {"0cf55fa5cf91acdc164f2eb6936eb49af19ed9c432d060c475db4a9090cd169b"} ), "tests/test_history.py": frozenset( - {"3997fc5c3b538df4ded9e994781fd8384b6b3b42ac695e1f1450ac24f6a92e7c"} + { + "3997fc5c3b538df4ded9e994781fd8384b6b3b42ac695e1f1450ac24f6a92e7c", + "15a6e7778a92e0d826ba1af61032f222dbf926b5ec2e189c48d7985a70824f19", + } ), "tests/test_history_replay.py": frozenset( - {"87272a47ab814803a77a39cafeb1b193b48678bc889b49712b11bace8b7d8c87"} + { + "87272a47ab814803a77a39cafeb1b193b48678bc889b49712b11bace8b7d8c87", + "226d30b1ad98ba7603bf3d8f7c86ac0857bf596bb75453d0dd57d677ec1405bd", + } ), "tests/test_hardening.py": frozenset( - {"91e53ca9b16ed43ebd49e7196bd6613f24c03cbf356f2bfc0ea03798cb065753"} + { + "91e53ca9b16ed43ebd49e7196bd6613f24c03cbf356f2bfc0ea03798cb065753", + "bb9825c1889d919e29a90e43c2be48b84d846fcad813fa3896d194491cc8f386", + } ), "tests/test_memory_provider.py": frozenset( - {"c2edfbbe5a6320088f58db89b1880200b3e844ed3780885a02ef62d3218ab2b1"} + { + "c2edfbbe5a6320088f58db89b1880200b3e844ed3780885a02ef62d3218ab2b1", + "cc967199b8e877a8088937a0c951e80f188ad4c7f29a98067bf649a2f8500b72", + } ), "tests/test_migration_baseline.py": frozenset( {"0a3c7af6a761a1b22c5b94f7e1738d5d420e5f085f775d792728ed9ca684ac13"} @@ -219,6 +232,7 @@ "87734abc087a4267e77aaeae7fb349d4c19f56dea2f7f3d173ab833dfb1da5e9", "cdc48a8f8dbe1d68dfe329d5106c55895c5345924544c2cc4b3d7190c837705b", "f139b1328c7884dfa664aa59b47d5d707b2a72894747911b8f99d5e89f6c7eca", + "1cce63f45c357df4710e24a8351835c4382a47751d57edf4207e234eb350807b", } ), 'COMPATIBILITY.md': frozenset( @@ -244,8 +258,8 @@ 'docs/operation.md': frozenset({'fadd72791e097d878bbfbf338922d568f3c3d47958f26a41f53d3597471988fa'}), 'docs/public-boundary.json': frozenset({'976403a6a802832adddd5ab1ff56fdc2e4e4847a1fe37ea5e6a2d5f106bbbbe2', '648904170d6c66de6f15cfe51fb494225467730ff5167225d6fd427a6eac3571'}), 'docs/public-boundary.md': frozenset({'c9228e2abd22a8af558cd272988435c1caa69ad79c625cb08c8ef32e639eb0f3'}), - 'src/substrate_wiki/README.md': frozenset({'44a11fda85d6d8170d772beadba2149bd114624479344b69a69cd49634309204'}), - 'src/substrate_wiki/__init__.py': frozenset({'425cd191f723805bed85d965d2d74daeac272dd0660eadf6bac92f1fa02d6f4a'}), + 'src/substrate_wiki/README.md': frozenset({'44a11fda85d6d8170d772beadba2149bd114624479344b69a69cd49634309204', '88d492b33ee537567bf85b63dcc7a470319212ac377f4857e27e504a3a1a05a3'}), + 'src/substrate_wiki/__init__.py': frozenset({'425cd191f723805bed85d965d2d74daeac272dd0660eadf6bac92f1fa02d6f4a', '71c0c19d366759e67c7b999029777295cc8c25503d35fe9cfddc9460bcb34d5a'}), 'src/substrate_wiki/client.py': frozenset( { "082164ad24c879f6ca6434a8f28c251cd1ee7f4b413c5a788a70b351e2187f2a", @@ -272,8 +286,8 @@ "72247d3537140098365350020cce29658c0743fee1aa738d7143db82316acce4", } ), - 'tests/test_hardening.py': frozenset({'f5f87125f1edd37bff1d44301d6bb0f44cc7faf3ba6122bdbe7569f349fea7a3'}), - 'tests/test_memory_provider.py': frozenset({'2c4517847dfad341063a69afcc737316a74574471fc98a45eaff21cfe4e271fd'}), + 'tests/test_hardening.py': frozenset({'f5f87125f1edd37bff1d44301d6bb0f44cc7faf3ba6122bdbe7569f349fea7a3', 'bb9825c1889d919e29a90e43c2be48b84d846fcad813fa3896d194491cc8f386'}), + 'tests/test_memory_provider.py': frozenset({'2c4517847dfad341063a69afcc737316a74574471fc98a45eaff21cfe4e271fd', 'cc967199b8e877a8088937a0c951e80f188ad4c7f29a98067bf649a2f8500b72'}), } for _path, _digests in _HOSTED_ONBOARDING_EXACT_ALLOWLIST.items(): SYNTHETIC_FILE_SHA256_ALLOWLIST[_path] = ( @@ -288,7 +302,7 @@ "4b444b2583fbdd340b17d279fd169103c57f87a56dece39988d784b311222920" ) TRUSTED_HISTORICAL_BLOB_POLICY_SHA256 = ( - "d18f105fc482c28c9946f3f1bd21235821e9283ef76c48ce04cbfa756c20b09f" + "ec3a29f27ebe29ce65b7a70ac65311bfb4a31e54095709a7c69d2687e12d50e1" ) SCANNER_PATH = "scripts/verify_public_plugin_candidate.py" DESTINATION_MANIFEST_PATH = "docs/extraction-manifest.json" diff --git a/src/substrate_wiki/README.md b/src/substrate_wiki/README.md index d282ca5..02e73f8 100644 --- a/src/substrate_wiki/README.md +++ b/src/substrate_wiki/README.md @@ -55,14 +55,14 @@ hermes substrate_wiki import-cancel --job-id --yes --json ## Operation and privacy -Live completed turns and explicit memory writes are delivered asynchronously. Failed +Live completed user/assistant turns are delivered asynchronously. Failed transient deliveries stay in a bounded owner-private spool and retry with capped backoff. Authentication failures trigger automatic reconnect onboarding rather than exposing or logging credentials. Status, receipts, checkpoints, and diagnostics are content-free. -Visible prompts, assistant output, tool calls, and tool results can be sent after redaction. -Redaction is defense in depth, not proof that arbitrary sensitive prose is absent. Raw capture -events declare a 90-day retention policy; curated wiki pages have their own lifecycle. +Visible prompts and assistant output can be sent after redaction. Tool calls, tool results, +system messages, memory-write events, and provider/session metadata are excluded. Redaction is +defense in depth, not proof that arbitrary sensitive prose is absent. The provider exposes bounded cited wiki search/read/query/ingest/job tools and automatic memory-card prefetch. It exposes no arbitrary filesystem-write tool. diff --git a/src/substrate_wiki/__init__.py b/src/substrate_wiki/__init__.py index dd68fb3..8d75537 100644 --- a/src/substrate_wiki/__init__.py +++ b/src/substrate_wiki/__init__.py @@ -27,17 +27,15 @@ class MemoryProvider: # type: ignore[no-redef] from .client import SubstrateAPIError, SubstrateClient -from .events import MAX_CAPTURE_BYTES, CaptureEventBuilder -from .redaction import configured_secret_values, redact +from .events import CaptureEventBuilder +from .redaction import configured_secret_values from .spool import DurableSpool, secure_atomic_json_write __all__ = ["SubstrateWikiProvider", "register"] _PROVIDER_ID = "substrate_wiki" _MAX_TOOL_RESULT_BYTES = 64 * 1024 -_MAX_CAPTURE_BYTES = MAX_CAPTURE_BYTES _MAX_SESSIONS = 32 -_MAX_MESSAGE_HASHES = 512 _MAX_PREFETCH_BYTES = 16 * 1024 _MAX_PREFETCH_ENTRIES = 128 _MIN_SPOOL_BYTES = 16 * 1024 @@ -195,12 +193,6 @@ def _context_is_primary(context: Any, *, default: bool = True) -> bool: return not (isinstance(role, str) and role.lower() in _NON_PRIMARY_ROLES) -def _is_primary_runtime(kwargs: dict[str, Any]) -> bool: - """Fail closed only when Hermes explicitly marks this as a non-primary runtime.""" - context = kwargs.get("runtime_context") or kwargs.get("context") - return _context_is_primary(context, default=kwargs.get("is_primary", True) is not False) - - def _context_value(context: Any, key: str, default: Any = None) -> Any: if isinstance(context, dict): return context.get(key, default) @@ -244,7 +236,7 @@ def __init__(self) -> None: self._current_persisted: Path | None = None self._delivery_failure_streak = 0 self._retry_random = random.Random() - self._capture_state: OrderedDict[str, dict[str, Any]] = OrderedDict() + self._message_offsets: OrderedDict[str, int] = OrderedDict() self._secrets: tuple[str, ...] = () self._scope: dict[str, str] = {"provider_id": _PROVIDER_ID} self._initialized_primary = True @@ -514,23 +506,31 @@ def sync_turn( assistant: Any, *, session_id: str = "", - messages: list[dict[str, Any]] | None = None, - **kwargs: Any, + runtime_context: Any = None, ) -> None: - if not self._capture_allowed(kwargs): + """Upload only the completed user/assistant pair. + + Deliberately use Hermes's legacy signature so the host does not pass the + full transcript, tool calls, or tool results into this provider. + """ + if not self._initialized_primary or not _context_is_primary(runtime_context): + self._counters["suppressed"] += 1 return sid = str(session_id or self._session_id) - payload = {"user": user, "assistant": assistant} - self._capture("turn", sid, payload, messages=messages) + messages: list[dict[str, Any]] = [] + if user is not None and user != "": + messages.append({"role": "user", "content": user}) + if assistant is not None and assistant != "": + messages.append({"role": "assistant", "content": assistant}) + if messages: + self._capture_turn(sid, messages) def on_pre_compress(self, messages: list[dict[str, Any]], **kwargs: Any) -> str: - if self._capture_allowed(kwargs): - self._capture("pre_compress", self._session_id, {}, messages=messages) + del messages, kwargs return "" def on_session_end(self, messages: list[dict[str, Any]], **kwargs: Any) -> None: - if self._capture_allowed(kwargs): - self._capture_snapshot("session_end", self._session_id, messages) + del messages, kwargs def on_memory_write( self, @@ -540,25 +540,9 @@ def on_memory_write( metadata: dict[str, Any] | None = None, **kwargs: Any, ) -> None: - if not self._capture_allowed(kwargs): - return - safe_metadata = metadata if isinstance(metadata, dict) else {} - metadata_session = safe_metadata.get("session_id") - sid = ( - str(metadata_session)[:512] - if isinstance(metadata_session, (str, int)) - else self._session_id - ) - selected_metadata: dict[str, Any] = {} - for key in ("source", "provenance"): - value = safe_metadata.get(key) - if isinstance(value, (str, int, float, bool)): - selected_metadata[key] = value - self._capture( - "memory_write", - sid, - {"action": action, "target": target, "content": content, "metadata": selected_metadata}, - ) + if not self._initialized_primary: + self._counters["suppressed"] += 1 + del action, target, content, metadata, kwargs def on_session_switch( self, @@ -569,24 +553,10 @@ def on_session_switch( rewound: bool = False, **kwargs: Any, ) -> None: - """Emit the old session boundary before rebinding to prevent attribution races.""" - if not self._capture_allowed(kwargs): - return + """Rebind future turn uploads without emitting a metadata event.""" + del parent_session_id, reset, rewound, kwargs new_session = str(new_session_id or "") with self._capture_lock: - old_session = self._session_id - if old_session and (old_session != new_session or reset or rewound): - self._capture( - "session_boundary", - old_session, - { - "reason": "session_switch", - "next_session_id": new_session, - "parent_session_id": str(parent_session_id or "")[:512], - "reset": bool(reset), - "rewound": bool(rewound), - }, - ) self._session_id = new_session def shutdown(self) -> None: @@ -630,139 +600,28 @@ def _load_settings(self) -> None: except (TypeError, ValueError): continue - def _capture( - self, - kind: str, - session_id: str, - payload: dict[str, Any], - *, - messages: list[dict[str, Any]] | None = None, - ) -> None: + def _capture_turn(self, session_id: str, messages: list[dict[str, Any]]) -> None: + """Spool one compact dialogue turn with monotonically increasing indexes.""" with self._capture_lock: - safe_payload = redact(payload, self._secrets) - hashes, boundary, delta, proposed_state = self._incremental_boundary( - session_id, messages + start = self._message_offsets.get(session_id, 0) + events = self._require_event_builder().iter_message_events( + "turn", + session_id, + messages, + start_index=start, ) - if messages is not None: - roles = { - item.get("role") - for item in delta - if isinstance(item, dict) and isinstance(item.get("role"), str) - } - if "user" in roles: - safe_payload.pop("user", None) - if "assistant" in roles: - safe_payload.pop("assistant", None) - builder = self._require_event_builder() - if messages is None: - events = iter( - ( - builder.payload_event( - kind, - session_id, - safe_payload, - boundary=boundary, - ), - ) - ) - else: - events = builder.iter_message_events( - kind, - session_id, - delta, - start_index=boundary["start"], - payload=safe_payload, - ) captured = 0 - delivered_to_spool = True for event in events: - if hashes: - event.setdefault("source_message_hashes", hashes) if not self._enqueue(event): - delivered_to_spool = False - break + return captured += 1 - if captured and delivered_to_spool: - self._counters["captured"] += captured - if proposed_state is not None: - self._commit_capture_state(session_id, proposed_state) - - def _capture_snapshot(self, kind: str, session_id: str, messages: list[dict[str, Any]]) -> None: - with self._capture_lock: - event = self._require_event_builder().payload_event( - kind, - session_id, - { - "session_complete": True, - "total_message_boundary": {"start": 0, "end": len(messages)}, - "protocol": "stream-v2", - }, - boundary={"start": 0, "end": len(messages)}, - ) - if self._enqueue(event): - self._counters["captured"] += 1 - - def _incremental_boundary( - self, session_id: str, messages: list[dict[str, Any]] | None - ) -> tuple[list[str], dict[str, int], list[Any], dict[str, Any] | None]: - state = self._capture_state.get(session_id, {"base": 0, "hashes": []}) - previous_base = int(state["base"]) - previous_hashes = list(state["hashes"]) - previous_end = previous_base + len(previous_hashes) - if not messages: - return [], {"start": previous_end, "end": previous_end}, [], None - total = len(messages) - current_base = max(0, total - _MAX_MESSAGE_HASHES) - safe_messages = redact(messages[current_base:], self._secrets) - current_hashes = [ - hashlib.sha256( - json.dumps( - message, ensure_ascii=False, sort_keys=True, separators=(",", ":") - ).encode("utf-8") - ).hexdigest() - for message in safe_messages - ] - overlap_start = max(previous_base, current_base) - overlap_end = min(previous_end, total) - unchanged = overlap_end >= overlap_start - if unchanged: - for absolute in range(overlap_start, overlap_end): - if ( - previous_hashes[absolute - previous_base] - != current_hashes[absolute - current_base] - ): - unchanged = False - break - delta_start = ( - max(previous_end, current_base) if unchanged and total >= previous_end else current_base - ) - delta_offset = max(0, delta_start - current_base) - return ( - current_hashes[delta_offset:], - {"start": delta_start, "end": total}, - safe_messages[delta_offset:], - {"base": current_base, "hashes": current_hashes}, - ) - - def _commit_capture_state(self, session_id: str, state: dict[str, Any]) -> None: - self._capture_state[session_id] = state - self._capture_state.move_to_end(session_id) - while len(self._capture_state) > _MAX_SESSIONS: - self._capture_state.popitem(last=False) - - def _bounded_capture(self, value: dict[str, Any]) -> dict[str, Any]: - encoded = json.dumps( - value, ensure_ascii=False, sort_keys=True, separators=(",", ":") - ).encode("utf-8") - spool_limit = int(self._settings.get("spool_max_bytes", _MAX_CAPTURE_BYTES)) - limit = min(_MAX_CAPTURE_BYTES, max(1024, spool_limit // 2)) - if len(encoded) <= limit: - return value - return { - "capture_truncated": True, - "original_bytes": len(encoded), - "sha256": hashlib.sha256(encoded).hexdigest(), - } + if not captured: + return + self._counters["captured"] += captured + self._message_offsets[session_id] = start + len(messages) + self._message_offsets.move_to_end(session_id) + while len(self._message_offsets) > _MAX_SESSIONS: + self._message_offsets.popitem(last=False) def _require_event_builder(self) -> CaptureEventBuilder: if self._event_builder is None: @@ -1056,12 +915,6 @@ def status_snapshot(self) -> dict[str, Any]: "last_prefetch_category": self._last_prefetch_category, } - def _capture_allowed(self, kwargs: dict[str, Any]) -> bool: - allowed = self._initialized_primary and _is_primary_runtime(kwargs) - if not allowed: - self._counters["suppressed"] += 1 - return allowed - @staticmethod def _read_persisted_api_url(root: Path) -> str: if root.is_symlink(): diff --git a/src/substrate_wiki/events.py b/src/substrate_wiki/events.py index 79d2097..14b157b 100644 --- a/src/substrate_wiki/events.py +++ b/src/substrate_wiki/events.py @@ -12,20 +12,13 @@ from .redaction import iter_redacted_text_chunks, redact -PROVIDER_ID = "substrate_wiki" SCHEMA_VERSION = 2 -RETENTION_DAYS = 90 MAX_CAPTURE_BYTES = 256 * 1024 _EVENT_NAMESPACE = uuid.UUID("837bd8c2-df25-4a42-bdc1-d38f0c00a8bc") _MESSAGE_FIELDS = ( "role", "content", - "tool_call_id", - "tool_calls", - "tool_name", "timestamp", - "platform_message_id", - "name", ) _TEXT_BLOCK_TYPES = {"text", "input_text", "output_text"} _BINARY_KEYS = { @@ -115,9 +108,9 @@ def normalize_message( ) -> dict[str, Any] | None: """Return a redacted, inference-safe message or ``None`` for system data.""" role = str(message.get("role") or "").strip().lower() - if role not in {"user", "assistant", "tool"}: + if role not in {"user", "assistant"}: return None - selected: dict[str, Any] = {"role": role, "message_index": int(index)} + selected: dict[str, Any] = {"index": int(index), "role": role} for field in _MESSAGE_FIELDS[1:]: if field not in message or message[field] is None: continue @@ -188,7 +181,7 @@ def _fragment_message(message: dict[str, Any], maximum: int) -> Iterator[dict[st ): yield { "role": message["role"], - "message_index": message["message_index"], + "index": message["index"], "content": encoded[start:end], "fragment": { "encoding": "canonical-json", @@ -320,7 +313,7 @@ def _iter_streamed_utf8_fragments( class CaptureEventBuilder: - """Build v2 events without allowing one request to exceed the capture cap.""" + """Build bounded events containing only session identity and raw dialogue.""" def __init__( self, @@ -329,26 +322,9 @@ def __init__( secrets: Sequence[str] = (), max_capture_bytes: int = MAX_CAPTURE_BYTES, ) -> None: - self.scope = { - str(key): str(value)[:512] - for key, value in scope.items() - if value is not None and str(value) - } - self.scope.setdefault("provider_id", PROVIDER_ID) - platform = self.scope.setdefault("platform", "cli").casefold() - self.scope["platform"] = platform - self.scope.setdefault( - "agent_id", - self.scope.get("profile") or self.scope.get("agent_identity") or "default", - ) - if "subject_id" not in self.scope: - user_id = self.scope.get("user_id") or self.scope.get("user") or "" - if platform == "cli": - self.scope["subject_id"] = "owner" - elif user_id: - self.scope["subject_id"] = hashlib.sha256( - f"{platform}\0{user_id}".encode() - ).hexdigest()[:24] + # ``scope`` remains accepted so older callers do not need an adapter, + # but none of it is duplicated into the upload envelope. + del scope self.secrets = tuple(secrets) self.max_capture_bytes = max(16 * 1024, min(max_capture_bytes, MAX_CAPTURE_BYTES)) @@ -390,7 +366,7 @@ def iter_message_events( batch_id: str = "", deterministic: bool = False, ) -> Iterator[dict[str, Any]]: - """Yield one bounded v2 event while retaining only one fragment.""" + """Yield bounded user/assistant events while retaining only one fragment.""" def fragments() -> Iterator[dict[str, Any]]: for offset, raw in enumerate(messages): @@ -454,8 +430,6 @@ def groups() -> Iterator[list[dict[str, Any]]]: { **(payload or {}), "messages": [], - "message_indexes": [], - "capture_chunk": {"index": 0, "final": True}, }, boundary={"start": start_index, "end": start_index}, capture_origin=capture_origin, @@ -511,16 +485,11 @@ def _group_event( session_id, { **(payload or {}), - "messages": [ - {key: value for key, value in message.items() if key != "message_index"} - for message in messages - ], - "message_indexes": [int(message["message_index"]) for message in messages], - "capture_chunk": {"index": chunk_index, "final": final}, + "messages": messages, }, boundary={ - "start": min(int(message["message_index"]) for message in messages), - "end": max(int(message["message_index"]) for message in messages) + 1, + "start": min(int(message["index"]) for message in messages), + "end": max(int(message["index"]) for message in messages) + 1, }, capture_origin=capture_origin, batch_id=batch_id, @@ -571,7 +540,7 @@ def payload_event( @staticmethod def _boundary(messages: Sequence[dict[str, Any]]) -> dict[str, int]: - indexes = [int(message["message_index"]) for message in messages] + indexes = [int(message["index"]) for message in messages] return {"start": min(indexes), "end": max(indexes) + 1} def _event( @@ -587,15 +556,11 @@ def _event( event_id: str | None = None, ) -> dict[str, Any]: safe_session = str(session_id)[:512] - safe_payload = {"session_id": safe_session, **payload} + safe_payload = dict(payload) identity = { - "provider_id": PROVIDER_ID, "kind": kind, - "scope": {**self.scope, "session_id": safe_session}, - "session_lineage": {"session_id": safe_session}, - "capture_boundary": boundary, - "capture_origin": capture_origin, - "batch_id": batch_id, + "session_id": safe_session, + "boundary": boundary, "payload": safe_payload, } resolved_id = event_id @@ -608,21 +573,13 @@ def _event( event: dict[str, Any] = { "schema_version": SCHEMA_VERSION, "event_id": resolved_id, - "provider_id": PROVIDER_ID, "kind": kind, - "capture_kind": kind, - "capture_boundary": boundary, - "session_lineage": {"session_id": safe_session}, - "scope": {**self.scope, "session_id": safe_session}, - "capture_origin": str(capture_origin)[:64], + "session_id": safe_session, "created_at": 0 if deterministic else time.time(), - "retention_days": RETENTION_DAYS, - "content_sha256": content_digest(safe_payload), - "payload": safe_payload, } - if batch_id: - event["batch_id"] = str(batch_id)[:128] - messages = safe_payload.get("messages") + messages = safe_payload.pop("messages", None) if isinstance(messages, list): - event["message_hashes"] = [content_digest(message) for message in messages] + event["messages"] = messages + if safe_payload: + event["data"] = safe_payload return event diff --git a/src/substrate_wiki/history.py b/src/substrate_wiki/history.py index 8182417..40a1e61 100644 --- a/src/substrate_wiki/history.py +++ b/src/substrate_wiki/history.py @@ -1371,7 +1371,6 @@ def run(self, *, wait: bool) -> dict[str, Any]: try: if self.stop_requested(): return checkpoint.status() - self._deliver_manifest(checkpoint) for session in checkpoint.sessions(): if self.stop_requested() or checkpoint.job()["state"] == "cancelled": return checkpoint.status() @@ -1398,48 +1397,6 @@ def run(self, *, wait: bool) -> dict[str, Any]: checkpoint.set_state("failed", error_class=type(exc).__name__) raise - def _deliver_manifest(self, checkpoint: ImportCheckpoint) -> None: - status = checkpoint.status() - builder = CaptureEventBuilder( - {"provider_id": "substrate_wiki", "agent_id": self.agent_id}, - secrets=tuple( - secret - for secret in dict.fromkeys( - (*configured_secret_values(), str(getattr(self.client, "api_key", "") or "")) - ) - if secret - ), - ) - event = builder.payload_event( - "session_boundary", - f"import:{self.batch_id}", - { - "batch_manifest": { - key: status[key] - for key in ("discovered", "eligible", "skipped", "quarantined") - }, - "protocol": "stream-v2", - "legacy_batch_ids": status["legacy_batch_ids"], - }, - capture_origin="history_replay", - batch_id=self.batch_id, - deterministic=True, - ) - if checkpoint.acknowledged(str(event["event_id"])): - return - result, retries = _deliver_with_retry_meta( - self.client, event, stop_requested=self.stop_requested - ) - checkpoint.acknowledge( - event_id=str(event["event_id"]), - external_id=f"import:{self.batch_id}", - boundary_start=0, - boundary_end=0, - phase="manifest", - duplicate=bool(result.get("duplicate")), - retry_count=retries, - ) - def _deliver_session( self, checkpoint: ImportCheckpoint, session: SessionDescriptor ) -> bool: @@ -1459,40 +1416,58 @@ def _deliver_session( if secret ), ) - for message in self.source.iter_messages(session, start=start): + messages = self.source.iter_messages(session, start=start) + for event in builder.iter_message_events( + "turn", + session.external_id, + messages, + start_index=start, + capture_origin="history_replay", + batch_id=self.batch_id, + deterministic=True, + ): if self.stop_requested(): return False - index = int(message.get("_capture_index", start)) - for event in builder.iter_message_events( - "turn", - session.external_id, - (message,), - start_index=index, - capture_origin="history_replay", - batch_id=self.batch_id, - deterministic=True, - ): - if self.stop_requested() or checkpoint.job()["state"] == "cancelled": - return False - event_id = str(event["event_id"]) - if checkpoint.acknowledged(event_id): + if checkpoint.job()["state"] == "cancelled": + return False + event_messages = event.get("messages", []) + if not isinstance(event_messages, list) or not event_messages: + continue + indexes = [ + int(message["index"]) + for message in event_messages + if isinstance(message, dict) and isinstance(message.get("index"), int) + ] + if not indexes: + continue + event_id = str(event["event_id"]) + if checkpoint.acknowledged(event_id): + continue + result, retries = _deliver_with_retry_meta( + self.client, event, stop_requested=self.stop_requested + ) + completed_indexes = [] + for message in event_messages: + if not isinstance(message, dict): continue - result, retries = _deliver_with_retry_meta( - self.client, event, stop_requested=self.stop_requested - ) - chunk = event.get("payload", {}).get("capture_chunk", {}) - final = bool(chunk.get("final", True)) - rolling = content_digest({"previous": rolling, "event": event["content_sha256"]}) - checkpoint.acknowledge( - event_id=event_id, - external_id=session.external_id, - boundary_start=index, - boundary_end=index + 1 if final else index, - phase="message", - duplicate=bool(result.get("duplicate")), - retry_count=retries, - session_digest=rolling, - ) + fragment = message.get("fragment") + if not isinstance(fragment, dict) or int(fragment.get("index", 0)) + 1 >= int( + fragment.get("count", 1) + ): + completed_indexes.append(int(message["index"])) + boundary_start = min(indexes) + boundary_end = max(completed_indexes) + 1 if completed_indexes else boundary_start + rolling = content_digest({"previous": rolling, "event": event}) + checkpoint.acknowledge( + event_id=event_id, + external_id=session.external_id, + boundary_start=boundary_start, + boundary_end=boundary_end, + phase="message", + duplicate=bool(result.get("duplicate")), + retry_count=retries, + session_digest=rolling, + ) completion = builder.payload_event( "session_end", session.external_id, diff --git a/src/substrate_wiki/plugin.yaml b/src/substrate_wiki/plugin.yaml index cb307c4..9762c1c 100644 --- a/src/substrate_wiki/plugin.yaml +++ b/src/substrate_wiki/plugin.yaml @@ -2,7 +2,4 @@ name: substrate_wiki version: 2.0.3 description: "Unified entity-centric Substrate Markdown wiki memory with durable asynchronous capture." hooks: - - on_pre_compress - - on_session_end - - on_memory_write - on_session_switch diff --git a/tests/test_hardening.py b/tests/test_hardening.py index 9e5039a..f7753d2 100644 --- a/tests/test_hardening.py +++ b/tests/test_hardening.py @@ -171,7 +171,7 @@ def test_prefetch_exact_query_precedes_latest_session_fallback( assert instance.prefetch("new", session_id="session-a") == "latest" -def test_hooks_incremental_boundary_metadata_suppression_and_switch_attribution( +def test_only_completed_dialogue_turns_are_uploaded( provider: tuple[SubstrateWikiProvider, FakeClient], ) -> None: instance, fake = provider @@ -189,17 +189,16 @@ def test_hooks_incremental_boundary_metadata_suppression_and_switch_attribution( new_session_id="session-b", parent_session_id="parent", reset=True, rewound=False ) instance.sync_turn("new", "answer") - wait_until(lambda: len(fake.delivered) == 5) + wait_until(lambda: len(fake.delivered) == 1) bodies = [call["body"] for call in fake.delivered] - assert bodies[0]["capture_boundary"] == {"start": 0, "end": 1} - assert bodies[1]["capture_boundary"] == {"start": 1, "end": 1} - assert bodies[1]["payload"]["messages"] == [] - assert bodies[2]["payload"]["metadata"] == {"provenance": "native"} - assert bodies[2]["payload"]["session_id"] == "memory-session" - assert bodies[3]["kind"] == "session_boundary" - assert bodies[3]["payload"]["session_id"] == "session-a" - assert bodies[4]["payload"]["session_id"] == "session-b" - assert all(body["provider_id"] == "substrate_wiki" for body in bodies) + assert bodies[0]["session_id"] == "session-b" + assert bodies[0]["messages"] == [ + {"index": 0, "role": "user", "content": "new"}, + {"index": 1, "role": "assistant", "content": "answer"}, + ] + assert set(bodies[0]) == { + "schema_version", "event_id", "kind", "session_id", "created_at", "messages" + } def test_sender_marks_each_in_memory_item_done_once_on_success_and_failure( @@ -326,12 +325,9 @@ def test_agent_context_scope_and_initialized_non_primary_suppression( ) primary.sync_turn("u", "a") wait_until(lambda: len(fake.delivered) == 1) - scope = fake.delivered[0]["body"]["scope"] - assert scope["agent_id"] == "agent-1" - assert scope["agent_identity"] == "Main" - assert scope["agent_workspace"] == "workspace" - assert scope["user_id"] == "user" - assert scope["platform"] == "cli" + body = fake.delivered[0]["body"] + assert body["session_id"] == "s" + assert not ({"scope", "agent_id", "agent_identity", "agent_workspace", "user_id", "platform"} & set(body)) snapshot = primary.status_snapshot() assert "session" not in json.dumps(snapshot).lower() assert "https://" not in json.dumps(snapshot) @@ -346,26 +342,20 @@ def test_agent_context_scope_and_initialized_non_primary_suppression( suppressed.shutdown() -def test_sync_turn_messages_are_delta_only_and_rolling_window_does_not_duplicate( +def test_sync_turn_uses_only_user_and_assistant_arguments( provider: tuple[SubstrateWikiProvider, FakeClient], ) -> None: instance, fake = provider - messages = [{"role": "user", "content": str(index)} for index in range(513)] - instance.sync_turn("repeat-user", "repeat-assistant", messages=messages) + instance.sync_turn("repeat-user", "repeat-assistant") wait_until(lambda: len(fake.delivered) == 1) first = fake.delivered[0]["body"] - assert "user" not in first["payload"] - assert first["payload"]["assistant"] == "repeat-assistant" - assert first["capture_boundary"] == {"start": 1, "end": 513} - instance.sync_turn("repeat-user", "repeat-assistant", messages=messages) + assert first["messages"] == [ + {"index": 0, "role": "user", "content": "repeat-user"}, + {"index": 1, "role": "assistant", "content": "repeat-assistant"}, + ] + instance.sync_turn("next-user", "next-assistant") wait_until(lambda: len(fake.delivered) == 2) - assert fake.delivered[1]["body"]["capture_boundary"] == {"start": 513, "end": 513} - assert fake.delivered[1]["body"]["payload"]["messages"] == [] - messages.append({"role": "assistant", "content": "new"}) - instance.sync_turn("repeat-user", "repeat-assistant", messages=messages) - wait_until(lambda: len(fake.delivered) == 3) - assert fake.delivered[2]["body"]["capture_boundary"] == {"start": 513, "end": 514} - assert len(fake.delivered[2]["body"]["payload"]["messages"]) == 1 + assert [message["index"] for message in fake.delivered[1]["body"]["messages"]] == [2, 3] def test_late_prefetch_from_old_session_is_not_published( @@ -414,8 +404,7 @@ def __repr__(self) -> str: wait_until(lambda: len(fake.delivered) == 1) rendered = json.dumps(fake.delivered[0]["body"]) assert "leaked-secret-repr" not in rendered - assert "[BINARY]" in rendered - assert "[UNSUPPORTED:SecretObject]" in rendered + assert "[NON_TEXT_CONTENT_OMITTED]" in rendered def test_strict_boolean_and_safe_url_prefix( @@ -440,42 +429,37 @@ def test_capture_state_commits_only_after_durable_admission( fake.fail = True fake.should_block = True fake.block_timeout = 30.0 - messages = [{"role": "user", "content": "must retry"}] original_append = instance._spool.append monkeypatch.setattr( instance._spool, "append", lambda event: (_ for _ in ()).throw(OSError("full")) ) - instance.sync_turn("must retry", "answer", messages=messages) + instance.sync_turn("must retry", "answer") assert instance.status_snapshot()["counters"]["dropped"] >= 1 monkeypatch.setattr(instance._spool, "append", original_append) - instance.sync_turn("must retry", "answer", messages=messages) + instance.sync_turn("must retry", "answer") try: assert fake.request_started.wait(3.0) paths = list(instance._spool.root.glob("*.json")) assert len(paths) == 1 event = instance._spool.load(paths[0]) - assert event["capture_boundary"] == {"start": 0, "end": 1} - assert event["payload"]["messages"] == messages + assert event["messages"] == [ + {"index": 0, "role": "user", "content": "must retry"}, + {"index": 1, "role": "assistant", "content": "answer"}, + ] finally: fake.block.set() wait_until(lambda: instance.status_snapshot()["counters"]["delivery_failed"] >= 1) -def test_session_end_is_content_free_after_turn_delta( +def test_session_end_does_not_emit_a_second_payload( provider: tuple[SubstrateWikiProvider, FakeClient], ) -> None: instance, fake = provider messages = [{"role": "user", "content": "one"}, {"role": "assistant", "content": "two"}] - instance.sync_turn("one", "two", messages=messages) + instance.sync_turn("one", "two") instance.on_session_end(messages) - wait_until(lambda: len(fake.delivered) == 2) - completed = next( - call["body"] for call in fake.delivered if call["body"]["kind"] == "session_end" - ) - assert completed["capture_boundary"] == {"start": 0, "end": 2} - assert "messages" not in completed["payload"] - assert completed["payload"]["session_complete"] is True - assert completed["payload"]["protocol"] == "stream-v2" + wait_until(lambda: len(fake.delivered) == 1) + assert fake.delivered[0]["body"]["kind"] == "turn" def test_spool_claim_protects_inflight_oldest_from_trim(tmp_path: Path) -> None: diff --git a/tests/test_history.py b/tests/test_history.py index 4d92407..425b040 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -133,7 +133,7 @@ def test_resume_skips_every_acknowledged_event(tmp_path: Path) -> None: home.mkdir() database = home / "state.db" _database(database, messages=3) - first_client = Client(fail_after=2) + first_client = Client(fail_after=1) first = HermesHistoryImporter( hermes_home=home, client=first_client, # type: ignore[arg-type] @@ -142,7 +142,7 @@ def test_resume_skips_every_acknowledged_event(tmp_path: Path) -> None: with pytest.raises(SubstrateAPIError): first.run(wait=True) first_ids = {str(item["event_id"]) for item in first_client.requests} - assert len(first_ids) == 2 + assert len(first_ids) == 1 if first.checkpoint is not None: first.checkpoint.close() diff --git a/tests/test_history_replay.py b/tests/test_history_replay.py index 234132b..f8a1a7c 100644 --- a/tests/test_history_replay.py +++ b/tests/test_history_replay.py @@ -72,7 +72,7 @@ def _streamed_capture( rendered = "".join( str(message["content"]) for event in events - for message in event["payload"]["messages"] + for message in event["messages"] ) assert sum(len(chunk) for chunk in chunks) == len(text) return rendered, events @@ -193,8 +193,10 @@ def test_shared_builder_is_deterministic_bounded_and_content_safe() -> None: assert "never copy this" not in encoded assert "hidden" not in encoded assert "[BINARY_CONTENT_OMITTED]" in encoded - assert all(event["retention_days"] == 90 for event in first) - assert all(event["scope"]["subject_id"] == "owner" for event in first) + assert all( + set(event) == {"schema_version", "event_id", "kind", "session_id", "created_at", "messages"} + for event in first + ) @pytest.mark.parametrize( @@ -309,7 +311,7 @@ def iter_text_chunks(self): # type: ignore[no-untyped-def] event_count += 1 captured_chars += sum( len(str(message["content"])) - for message in event["payload"]["messages"] + for message in event["messages"] ) assert captured_chars == len(chunk) * capture_repetitions @@ -391,7 +393,7 @@ def test_sqlite_large_message_is_read_and_emitted_in_bounded_slices(tmp_path: Pa fragments = [ item for event in events - for item in event["payload"]["messages"] + for item in event["messages"] ] assert "".join(str(item["content"]) for item in fragments) == content assert {item["fragment"]["encoding"] for item in fragments} == {"utf8-content"} @@ -761,7 +763,7 @@ def test_oversized_jsonl_record_streams_message_without_materializing_record( "turn", "large-jsonl", (message,), deterministic=True ) ) - fragments = [item for event in events for item in event["payload"]["messages"]] + fragments = [item for event in events for item in event["messages"]] assert "".join(str(item["content"]) for item in fragments) == decoded_piece * repetitions assert all(len(canonical_bytes(event)) <= 256 * 1024 for event in events) iterator.close() @@ -827,20 +829,13 @@ def test_importer_resumes_from_checkpoint_and_uses_standard_endpoints(tmp_path: assert status["complete"] is True assert [path for path, _ in first_client.requests] == [ - "/api/v1/hermes/turns", - "/api/v1/hermes/turns", "/api/v1/hermes/turns", "/api/v1/hermes/completed-sessions", ] event_ids = [request["body"]["event_id"] for _, request in first_client.requests] assert len(event_ids) == len(set(event_ids)) assert all( - request["body"].get("capture_origin") == "history_replay" - for _, request in first_client.requests - ) - - assert all( - not request["body"]["payload"].get("messages") + not request["body"].get("messages") for path, request in first_client.requests if path == "/api/v1/hermes/completed-sessions" ) diff --git a/tests/test_memory_provider.py b/tests/test_memory_provider.py index a078879..0ec6af3 100644 --- a/tests/test_memory_provider.py +++ b/tests/test_memory_provider.py @@ -396,7 +396,6 @@ def test_sync_turn_returns_without_waiting_for_network_and_redacts( lambda: provider.sync_turn( authorization_header, "api_key=plain-secret and very-secret-hermes-key", - messages=[{"role": "tool", "authorization": "Bearer hidden", "content": "ok"}], ), fake.block, ) @@ -428,7 +427,7 @@ def test_lifecycle_hooks_return_without_waiting_for_network( fake.block, ) fake.block.set() - wait_until(lambda: len(fake.delivered) == 3) + assert fake.delivered == [] finally: fake.block.set() provider.shutdown() @@ -446,7 +445,7 @@ def test_offline_events_spool_and_replay_oldest_first( fake.fail = False provider._wake.set() wait_until(lambda: len(fake.delivered) == 2, timeout=4) - assert [call["body"]["payload"]["user"] for call in fake.delivered] == ["first", "second"] + assert [call["body"]["messages"][0]["content"] for call in fake.delivered] == ["first", "second"] assert len({call["idempotency_key"] for call in fake.delivered}) == 2 wait_until(lambda: not list(spool_dir.glob("*.json"))) provider.shutdown() @@ -476,12 +475,7 @@ def test_prefetch_is_cached_and_lifecycle_hooks_are_delivered( provider.on_pre_compress([{"role": "user", "content": "x"}]) provider.on_session_end([{"role": "assistant", "content": "y"}]) provider.on_memory_write("write", "MEMORY.md", "candidate") - wait_until(lambda: len(fake.delivered) == 3) - assert [call["path"] for call in fake.delivered] == [ - "/api/v1/hermes/turns", - "/api/v1/hermes/completed-sessions", - "/api/v1/hermes/memory-write-events", - ] + assert fake.delivered == [] finally: provider.shutdown() diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 5f7c2ad..5f5757b 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -213,16 +213,12 @@ def test_v141_release_artifacts_remain_byte_pinned() -> None: ) -def test_v140_event_envelope_constants_remain_stream_v2() -> None: - # events.py now contains the mandatory stateful cross-chunk redactor, so it - # cannot remain byte-identical. The public capture/retention limits remain - # locked and the replay suite verifies deterministic v2 IDs and boundaries. +def test_compact_event_envelope_constants_are_locked() -> None: source = (PLUGIN_SOURCE / "events.py").read_text(encoding="utf-8") - assert 'PROVIDER_ID = "substrate_wiki"' in source assert "SCHEMA_VERSION = 2" in source - assert "RETENTION_DAYS = 90" in source assert "MAX_CAPTURE_BYTES = 256 * 1024" in source - assert "capture_chunk" in source + assert '"tool_calls"' not in source + assert '"retention_days"' not in source def test_publish_release_creates_exact_current_and_immutable_aliases(tmp_path: Path) -> None: