fix(state): a malformed unrelated route or local-policy row cannot abort a delete

The deletion fence scans every gateway_routing entry (retire_routes) and every
gateway.local_policy.v1:* creation receipt (retire_local_receipts) to find the ones that name
the sessions being deleted, and called json.loads(...).get(...) on each. One unrelated row that
is a JSON array, an object whose `entry` is not an object, or invalid JSON raised
AttributeError/JSONDecodeError inside the writer transaction, so canonical delete, legacy
delete_session(s) and prune all rolled back for a session that has nothing to do with that row.
The base deletes it, and the production routing loader already skips non-object rows.

A malformed row names no target: it is neither retired nor allowed to abort the sweep, and is
left untouched (its own readers keep refusing it as storage_unavailable). Regression covers the
four malformed shapes x canonical/legacy/prune; 12/12 fail on r6-base, all pass here.

Review: https://github.com/NousResearch/hermes-agent/pull/106742#pullrequestreview-5460639040 (W15)
Review: https://github.com/NousResearch/hermes-agent/pull/106742#pullrequestreview-5458083694 (Session deletion availability)

hermes_state_local.py

@@ -57,6 +57,18 @@ def commit_local_session(db, *, epoch, receipt):
     return db._execute_write(write)
 
 
+def _policy_target(raw):
+    """Current physical target a creation receipt routes to; None for a malformed receipt, which
+    then stays (its own reads already refuse it as ``storage_unavailable``) and cannot abort
+    the delete of an unrelated session."""
+    try:
+        value = json.loads(raw)
+    except (TypeError, ValueError):
+        return None
+    entry = value.get('entry') if isinstance(value, dict) else None
+    return entry.get('session_id') if isinstance(entry, dict) else None
+
+
 def retire_local_receipts(conn, session_ids):
     """Deletion fence half for owner receipts. A local creation policy carries the launch request
     (``request_json``: a cron fire's job and raw prompt), so it goes exactly when its route does:
@@ -69,7 +81,7 @@ def retire_local_receipts(conn, session_ids):
     # Prefix scans (GLOB uses the key index), not one LIKE per id: a prune retires thousands.
     gone = [key for key, value in conn.execute('SELECT key,value FROM state_meta WHERE key GLOB ?',
                                                (POLICY_PREFIX + '*',))
-            if json.loads(value).get('entry', {}).get('session_id') in targets]
+            if _policy_target(value) in targets]
     gone += [key for key, value in conn.execute('SELECT key,value FROM state_meta WHERE key GLOB ?',
                                                 (API_DECLARED_PREFIX + '*',)) if value in targets]
     gone += [API_BINDING_PREFIX + sid for sid in targets]

hermes_state_mutation_retirement.py

@@ -116,6 +116,17 @@ def delete_in_transaction(db, conn, session_id, payload):
     return set(), {'deleted_ids': targets}
 
 
+def entry_session_id(raw):
+    """The ``session_id`` a stored routing/receipt JSON object names, else None. Unrelated rows
+    are not validated here: a malformed one (an array, invalid JSON) names no target, so it can
+    neither be retired nor abort the delete of a different session; the routing loader skips it too."""
+    try:
+        value = json.loads(raw)
+    except (TypeError, ValueError):
+        return None
+    return value.get('session_id') if isinstance(value, dict) else None
+
+
 def retire_routes(conn, session_ids):
     # OR IGNORE: the marker is a fact, not a receipt; re-retiring an id that was recreated
     # beside its tombstone must not abort the delete that removes it again.
@@ -125,6 +136,6 @@ def retire_routes(conn, session_ids):
         conn.execute('DELETE FROM state_meta WHERE key=?', ('gateway.api.settings.v1.' + sid,))
     targets = set(session_ids)
     for row in conn.execute('SELECT scope,session_key,entry_json FROM gateway_routing').fetchall():
-        if json.loads(row['entry_json']).get('session_id') in targets:
+        if entry_session_id(row['entry_json']) in targets:
             conn.execute('DELETE FROM gateway_routing WHERE scope=? AND session_key=?',
                          (row['scope'], row['session_key']))

tests/hermes_state/test_terminal_retirement.py

@@ -186,3 +186,38 @@ def test_deleting_a_session_drops_transcript_copies_from_its_terminal_results(tm
             'final_response': 'a2', 'messages': [], 'completed': True, 'api_calls': 1, 'input_tokens': 5,
             '_messages_are_turn_suffix': True},
             'usage': {'input_tokens': 5, 'output_tokens': 2}}
+
+
+@pytest.mark.parametrize('stray', [
+    "INSERT INTO gateway_routing(scope,session_key,entry_json,updated_at) VALUES('','other','[]',0)",
+    "INSERT INTO gateway_routing(scope,session_key,entry_json,updated_at) VALUES('','other','{bad',0)",
+    "INSERT INTO state_meta(key,value) VALUES('gateway.local_policy.v1:other','[]')",
+    "INSERT INTO state_meta(key,value) VALUES('gateway.local_policy.v1:other','{\"entry\": []}')",
+])
+@pytest.mark.parametrize('delete', ['canonical', 'legacy', 'prune'])
+def test_a_malformed_unrelated_route_or_policy_row_cannot_abort_a_delete(tmp_path, stray, delete):
+    """W15 / routing-array: deletion fences decode every routing entry and creation receipt to
+    find the target's; one unrelated malformed row must not roll back an unrelated delete."""
+    import time
+    with SessionDB(tmp_path / 'state.db') as db:
+        epoch = rt.begin_runtime_epoch(db, instance_id='owner')
+        db.create_session('target', source='api_server')
+        db._execute_write(lambda c: c.execute(stray))
+        if delete == 'canonical':
+            snap = db.get_session('target')
+            rt.mutate_runtime_session(db, epoch=epoch, principal_id='human', session_id='target', request_id='d',
+                operation='delete', payload={}, expected_revision=snap['runtime_revision'],
+                expected_generation=snap['runtime_generation'])
+        elif delete == 'legacy':
+            assert db.delete_session('target')
+        else:
+            old = time.time() - 100 * 86400
+            db._execute_write(lambda c: c.execute('UPDATE sessions SET started_at=?, ended_at=? WHERE id=?',
+                                                  (old, old, 'target')))
+            assert db.prune_sessions(older_than_days=30) == 1
+        assert db.get_session('target') is None
+        from hermes_state_mutation_retirement import retired_session
+        assert retired_session(db, 'target')
+        with db._read_ctx() as c:  # the unrelated row is left exactly as it was
+            assert c.execute("SELECT (SELECT COUNT(*) FROM gateway_routing) + (SELECT COUNT(*) FROM state_meta "
+                             "WHERE key GLOB 'gateway.local_policy.v1:*')").fetchone()[0] == 1