mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
unsupportedpastels
cc4b3a0532
SQLite evaluates json_extract even behind json_valid(...) AND and raises "malformed JSON". In ROUTE_RETIRED_SQL that aborted whole-index routing saves and the sessions.json import. Each JSON read there now sits in a CASE WHEN json_valid(...) branch: a malformed fence fences like a permanent one, and a malformed marker like an earlier release's unscoped one. _mark_retired's upsert CASE gets the same guard. The other reads (expiring_retirements, _reclaim_expired) are WHERE filters that already read such a row without raising; the test pins them too. Refs #134889 (LC-26) Refs #140762 Refs #140744
434 lines
25 KiB
Python
434 lines
25 KiB
Python
"""Durable deletion fences outside prunable transcript rows."""
|
|
import json
|
|
import time
|
|
from hermes_state_runtime import RuntimeStoreError, admission_fingerprint, mutation_receipt_key, _json, _text
|
|
|
|
# This store's own retired session (its profile is the store's). Written as ``own`` since the
|
|
# shared routing index also carries other profiles' fences; a marker without ``own`` predates
|
|
# that and still fences every route in the index.
|
|
RETIRED_PREFIX = 'gateway.retired_session.v1.'
|
|
# A multiplexed secondary's retired session, in the routing index every served profile shares
|
|
# (the launch home's state.db): ``<prefix><namespace>:<id>``, keyed by the ``agent:<namespace>:``
|
|
# slot of that profile's routing keys. Only routing saves read it, so the launch profile's own
|
|
# creates, binds and generation walks never see another profile's retirement.
|
|
ROUTE_FENCE_PREFIX = 'gateway.retired_route.v1.'
|
|
# A last viewer's close stored until the owner's idle point (gateway/session_viewer_close).
|
|
VIEWER_CLOSE_PREFIX = 'gateway.viewer_close.v1:'
|
|
|
|
|
|
def retired_in(conn, session_id):
|
|
return conn.execute('SELECT 1 FROM state_meta WHERE key=?', (RETIRED_PREFIX + session_id,)).fetchone() is not None
|
|
|
|
|
|
def retired_session(db, session_id):
|
|
with db._read_ctx() as conn:
|
|
return retired_in(conn, session_id)
|
|
|
|
|
|
def create_conflict(conn, session_id):
|
|
"""``(code, message)`` refusing a create of the explicit *session_id*, else None: a live row, or a
|
|
deleted (retired) id. A row recreated beside the fence would list fine and refuse every turn."""
|
|
if conn.execute('SELECT 1 FROM sessions WHERE id=?', (session_id,)).fetchone():
|
|
return 'session_exists', 'Session already exists'
|
|
if retired_in(conn, session_id):
|
|
return 'session_deleted', 'Session was deleted; its id cannot be reused'
|
|
return None
|
|
|
|
|
|
def first_generation(db, derive, *, pinned=None, spent=None):
|
|
"""``(n, derive(n))``: the first generation of a derived session id a NEW request may use. A
|
|
user-deleted id stays retired for good, so a derived id (header-less chat, Bot Chat, hosted-room
|
|
member, A2A context) whose session was deleted moves to its next generation, never recreated.
|
|
``pinned(conn, sid)`` keeps an exact retry on the retired generation holding its identity;
|
|
``spent(conn, sid)`` skips a live one used up another way (a renamed Bot Chat). Read from
|
|
durable rows only, so a restart and every concurrent caller pick the same id."""
|
|
generation = 0
|
|
with db._read_ctx() as conn:
|
|
while True:
|
|
sid = derive(generation)
|
|
if retired_in(conn, sid):
|
|
if pinned is not None and pinned(conn, sid):
|
|
return generation, sid
|
|
elif spent is None or not spent(conn, sid):
|
|
return generation, sid
|
|
generation += 1
|
|
|
|
|
|
def keep_passed_fences(conn, derive, session_id):
|
|
"""Make permanent the fences a walk (``derive`` as for :func:`first_generation`) passed on its
|
|
way to *session_id*, inside the write that binds *session_id*: that successor now holds the
|
|
conversation, so a sweep marker aging out would send a later walk back to the passed
|
|
generation. False when a passed generation has lost its marker since the walk (a sweep's was
|
|
reclaimed): a walk now stops there, so *session_id* must not be bound as its successor."""
|
|
generation = 0
|
|
while (sid := derive(generation)) != session_id:
|
|
if not conn.execute('SELECT 1 FROM state_meta WHERE key=?', (RETIRED_PREFIX + sid,)).fetchone():
|
|
return False
|
|
conn.execute("UPDATE state_meta SET value=json_remove(value, '$.expires') WHERE key=? AND json_valid(value)",
|
|
(RETIRED_PREFIX + sid,))
|
|
generation += 1
|
|
return True
|
|
|
|
|
|
def has_mutation_receipt(db, principal_id, session_id, request_id):
|
|
for value in (principal_id, session_id, request_id):
|
|
_text(value)
|
|
key, _digest = mutation_receipt_key(principal_id, session_id, request_id, operation='model', payload={},
|
|
expected_revision=0)
|
|
with db._read_ctx() as conn:
|
|
return conn.execute('SELECT 1 FROM state_meta WHERE key=?', (key,)).fetchone() is not None
|
|
|
|
|
|
_TRANSCRIPT_FIELDS = frozenset({'messages', 'last_reasoning', 'tools', '_turn_user_row'})
|
|
|
|
|
|
def _compact_result(conn, admission_id):
|
|
"""A retired result keeps the exact-retry outcome (final response, flags, usage/accounting)
|
|
but not the transcript copies a turn result carries (cumulative ``messages``, reasoning, the
|
|
settled ``_turn_user_row``)."""
|
|
from hermes_state_terminal import RESULT_PREFIX, encode_result, load_result
|
|
saved = conn.execute('SELECT value FROM state_meta WHERE key=?', (RESULT_PREFIX + admission_id,)).fetchone()
|
|
if saved is None:
|
|
return
|
|
value = load_result(saved[0])
|
|
result = value.get('result')
|
|
if isinstance(result, dict):
|
|
value['result'] = {**{k: v for k, v in result.items() if k not in _TRANSCRIPT_FIELDS}, 'messages': []}
|
|
conn.execute('UPDATE state_meta SET value=? WHERE key=?', (_json(encode_result(value)), RESULT_PREFIX + admission_id))
|
|
|
|
|
|
def retire_terminal_receipts(conn, session_ids):
|
|
from hermes_state_terminal import ADMISSION_PREFIX, WORKER_PREFIX, identity_key
|
|
for sid in session_ids:
|
|
admissions = conn.execute('SELECT * FROM session_admissions WHERE target_session_id=?', (sid,)).fetchall()
|
|
workers = conn.execute('SELECT * FROM worker_executions WHERE session_id=?', (sid,)).fetchall()
|
|
states = {r['status'] for r in [*admissions, *workers]} - {'terminal'}
|
|
if states:
|
|
raise RuntimeStoreError('unknown_execution' if 'unknown' in states else 'session_busy')
|
|
for raw in admissions:
|
|
row = dict(raw)
|
|
from hermes_state_media import retire_media
|
|
retire_media(conn, json.loads(row['payload_json']), row['request_id'])
|
|
# Keep the digest for exact retries, not another copy of user input/history.
|
|
row['payload_json'] = '{}'
|
|
row['lineage_json'] = '[]'
|
|
row['retired_at'] = time.time()
|
|
_compact_result(conn, row['admission_id'])
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)',
|
|
(ADMISSION_PREFIX + row['admission_id'], _json(row)))
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)',
|
|
(identity_key(row['principal_id'], sid, row['request_id']), json.dumps(row['admission_id'])))
|
|
for raw in workers:
|
|
row = dict(raw)
|
|
receipts = [dict(r) for r in conn.execute(
|
|
'SELECT sequence,payload_digest,result_json FROM worker_receipts WHERE execution_id=? ORDER BY sequence',
|
|
(row['execution_id'],))]
|
|
# A terminal worker can only replay an explicit ``execution.finish`` receipt. Every
|
|
# other result (history/context reads) is user data that must not outlive the delete,
|
|
# even when it is the last one: settlement terminalizes a failed worker without a
|
|
# finish receipt. Digests stay so a late duplicate is still recognised as a conflict.
|
|
closing = admission_fingerprint(canonical_target=sid,
|
|
payload={'operation': 'execution.finish', 'payload': {}})
|
|
for receipt in receipts:
|
|
if receipt['payload_digest'] != closing:
|
|
receipt['result_json'] = None
|
|
row['receipts'] = receipts
|
|
row['retired_at'] = time.time()
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)',
|
|
(WORKER_PREFIX + row['execution_id'], _json(row)))
|
|
conn.execute('DELETE FROM worker_receipts WHERE execution_id=?', (row['execution_id'],))
|
|
conn.execute('DELETE FROM worker_executions WHERE session_id=?', (sid,))
|
|
conn.execute('DELETE FROM session_admissions WHERE target_session_id=?', (sid,))
|
|
|
|
|
|
_LIVE_LEDGER_SQL = """SELECT 1 FROM session_admissions WHERE target_session_id=? AND status IN ('queued','started','unknown')
|
|
UNION ALL SELECT 1 FROM worker_executions WHERE session_id=? AND status IN ('registered','running','unknown') LIMIT 1"""
|
|
|
|
|
|
def retire_sessions(conn, session_ids, *, expiring=()):
|
|
"""The complete deletion fence for ids that are being deleted in this transaction.
|
|
|
|
Terminal receipts alone are not enough: without the ``RETIRED_PREFIX`` marker an exact
|
|
retry of a settled request reports ``not_found`` instead of its terminal receipt, a late
|
|
accounting/constructor backfill recreates the deleted row, and the same request is then
|
|
admitted a second time against it. Every delete path (canonical mutate, legacy
|
|
``delete_session*``, prunes and sweeps) must publish the whole fence, so it lives here once.
|
|
The marker of an id in *expiring* (a housekeeping sweep's) ages out at the retry horizon."""
|
|
from hermes_state_media import retire_spillover
|
|
retire_spillover(conn, session_ids) # before the caller's DELETEs take the pointers away
|
|
retire_terminal_receipts(conn, session_ids)
|
|
retire_mutation_receipts(conn, session_ids)
|
|
retire_routes(conn, session_ids, expiring=expiring)
|
|
from hermes_state_local import retire_local_receipts
|
|
retire_local_receipts(conn, session_ids)
|
|
|
|
|
|
# Mutation-receipt fields that copy user content: the rewound message, a compaction summary and
|
|
# a user-set title (rename/sidebar).
|
|
_MUTATION_TRANSCRIPT_FIELDS = ('target_message', 'summary', 'title')
|
|
# Every session a receipt result can name. A local owner's receipt is keyed by its logical id, but
|
|
# its rewind/compress content comes from the PHYSICAL target (a reset child): deleting only that
|
|
# child must strip it too, or the surviving owner's exact retry replays the deleted text.
|
|
_RECEIPT_SESSION_PATHS = ('$.result.session_id', '$.result.target_message.session_id',
|
|
'$.result.target_session_id', '$.result.previous_target_session_id')
|
|
|
|
|
|
def retire_mutation_receipts(conn, session_ids):
|
|
"""Keep each exact-retry mutation receipt (digest, ids, revision) but drop the user content
|
|
a rewind/compress/rename result carries, so a retry after the delete cannot read it back."""
|
|
if not session_ids:
|
|
return
|
|
named = ' OR '.join(f"json_extract(value,'{path}') IN (SELECT value FROM json_each(?1))"
|
|
for path in _RECEIPT_SESSION_PATHS)
|
|
rows = conn.execute(
|
|
"SELECT key,value FROM state_meta WHERE key GLOB 'gateway.mutation.v1.*' AND "
|
|
f"CASE WHEN json_valid(value) THEN ({named}) END", (json.dumps(list(session_ids)),)).fetchall()
|
|
for key, raw in rows:
|
|
receipt = json.loads(raw)
|
|
result = receipt.get('result')
|
|
if isinstance(result, dict) and any(f in result for f in _MUTATION_TRANSCRIPT_FIELDS):
|
|
receipt['result'] = {k: (None if k in _MUTATION_TRANSCRIPT_FIELDS else v) for k, v in result.items()}
|
|
conn.execute('UPDATE state_meta SET value=? WHERE key=?', (_json(receipt), key))
|
|
|
|
|
|
# Creation ids of the local conversations a gateway door derives and walks by generation
|
|
# (``first_generation``): Bot Chat and A2A contexts.
|
|
_DERIVED_LOCAL_PREFIXES = ('bot-chat:', 'a2a:')
|
|
# The API sessions whose ids a door derives: a hosted-room member (``bot_room``), and the
|
|
# ``api-<16 hex>`` ids of a header-less chat (``_derive_chat_session_id``) and of a deleted chat's or
|
|
# Responses/Runs chain's next generation (``implicit_api_session``). A caller-named ``api_server`` id
|
|
# (``POST /api/sessions``, ``X-Hermes-Session-Id``) is never derived. Matched on the id, not on a
|
|
# binding-receipt field, so it also holds for rows an older release bound. Any id can root a
|
|
# Responses/Runs chain: its fence becomes permanent once a chain moves past it (``keep_passed_fences``).
|
|
_DERIVED_API_SQL = ("(source='bot_room' OR (source='api_server' AND id GLOB 'api-" + '[0-9a-f]' * 16 + "'))")
|
|
|
|
|
|
def _local_conversations(conn):
|
|
"""``(members, derived)``: member id -> every id of its local conversation (the creation id
|
|
(policy, FIFO, generation), each receipt lineage segment and the current target), and the ids
|
|
of every derived (Bot Chat, A2A) conversation. One prefix scan, not a lookup per swept id.
|
|
Malformed receipts name nothing."""
|
|
from hermes_state_local import POLICY_PREFIX
|
|
members, derived = {}, set()
|
|
for (raw,) in conn.execute('SELECT value FROM state_meta WHERE key GLOB ?', (POLICY_PREFIX + '*',)):
|
|
try:
|
|
receipt = json.loads(raw)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if not isinstance(receipt, dict):
|
|
continue
|
|
entry, lineage = receipt.get('entry'), receipt.get('lineage')
|
|
ids = tuple(x for x in [receipt.get('session_id'), *(lineage if isinstance(lineage, list) else []),
|
|
entry.get('session_id') if isinstance(entry, dict) else None] if isinstance(x, str))
|
|
members.update(dict.fromkeys(ids, ids))
|
|
if str(receipt.get('request_id') or '').startswith(_DERIVED_LOCAL_PREFIXES):
|
|
derived.update(ids)
|
|
return members, derived
|
|
|
|
|
|
def _derived_api_sessions(conn, session_ids):
|
|
"""The ids among *session_ids* of derived API sessions (``_DERIVED_API_SQL``)."""
|
|
return {row[0] for row in conn.execute(
|
|
f"SELECT id FROM sessions WHERE {_DERIVED_API_SQL} AND id IN (SELECT value FROM json_each(?))",
|
|
(json.dumps(list(session_ids)),))}
|
|
|
|
|
|
def retire_prunable(conn, session_ids):
|
|
"""Sweep variant of :func:`retire_sessions`: fence the idle sessions and return only those ids.
|
|
A session with live or unknown work is skipped, so one busy row cannot abort a whole
|
|
prune/empty-session sweep (explicit deletes still refuse with ``session_busy``). A local
|
|
reset/compression conversation is swept only whole: while any of its ids is kept or has live
|
|
work (the owner's queued admission continues on the current segment), every one is skipped.
|
|
|
|
A sweep's markers age out at the retry horizon: housekeeping retires thousands of ids a pass
|
|
(cron), and a permanent marker each would grow ``state_meta`` without bound. A derived id's
|
|
stays (Bot Chat, A2A context, header-less chat, a chain's next generation, hosted-room
|
|
member): its conversation resumes on the next generation, and the generation walk stops at the
|
|
first unfenced id, so an aged-out marker would send the next turn back to the swept generation.
|
|
A Bot Chat's mailbox receipts also outlive the horizon. A caller-named API id is not derived:
|
|
its marker ages out, so the caller can use that id again, as for any other swept session."""
|
|
swept, (conversations, derived) = set(session_ids), _local_conversations(conn)
|
|
|
|
def removable(sid):
|
|
whole = conversations.get(sid, (sid,))
|
|
return swept.issuperset(whole) and not any(conn.execute(_LIVE_LEDGER_SQL, (x, x)).fetchone() for x in whole)
|
|
quiet = [sid for sid in session_ids if removable(sid)]
|
|
derived |= _derived_api_sessions(conn, quiet)
|
|
retire_sessions(conn, quiet, expiring=[sid for sid in quiet if sid not in derived])
|
|
return quiet
|
|
|
|
|
|
def delete_in_transaction(db, conn, session_id, payload):
|
|
from hermes_state_mutation_guards import require_idle, delete_targets
|
|
targets = delete_targets(conn, session_id)
|
|
require_idle(db, conn, targets)
|
|
retire_sessions(conn, targets)
|
|
for sid in targets:
|
|
db._bump_conversation_generation(conn, sid, 'session_reset')
|
|
conn.execute('UPDATE sessions SET parent_session_id=NULL, runtime_revision=runtime_revision+1 WHERE parent_session_id=?', (sid,))
|
|
conn.execute('DELETE FROM messages WHERE session_id=?', (sid,))
|
|
conn.executemany('DELETE FROM sessions WHERE id=?', [(sid,) for sid in targets])
|
|
db._delete_unreferenced_system_prompts(conn)
|
|
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 _mark_retired(conn, key, *, expires, **fields):
|
|
# Replace: 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 (and restarts its horizon).
|
|
# An ``expires`` (sweep) marker ages out at the horizon and never shortens a permanent one.
|
|
# A permanent marker without ``own`` (an earlier release's, maybe another profile's shared
|
|
# fence) stays unscoped: it is the only fence of that profile's route. An earlier release's
|
|
# sweep marker came only from this store's own sweep, so it is replaced as written.
|
|
value = {'retired_at': time.time(), **fields}
|
|
if expires:
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE '
|
|
"SET value=excluded.value WHERE json_valid(state_meta.value) "
|
|
"AND json_extract(state_meta.value, '$.expires')",
|
|
(key, json.dumps({**value, 'expires': True})))
|
|
else:
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=CASE '
|
|
"WHEN NOT json_valid(state_meta.value) THEN json_remove(excluded.value, '$.own') "
|
|
"WHEN json_extract(state_meta.value, '$.own') IS NOT NULL "
|
|
"OR json_extract(state_meta.value, '$.expires') THEN excluded.value "
|
|
"ELSE json_remove(excluded.value, '$.own') END", (key, json.dumps(value)))
|
|
|
|
|
|
def retire_routes(conn, session_ids, *, expiring=()):
|
|
expiring = set(expiring)
|
|
for sid in session_ids:
|
|
_mark_retired(conn, RETIRED_PREFIX + sid, expires=sid in expiring, own=True)
|
|
conn.execute('DELETE FROM state_meta WHERE key=?', ('gateway.api.settings.v1.' + sid,))
|
|
conn.execute('DELETE FROM state_meta WHERE key=?', (VIEWER_CLOSE_PREFIX + sid,))
|
|
targets = set(session_ids)
|
|
for row in conn.execute('SELECT scope,session_key,entry_json FROM gateway_routing').fetchall():
|
|
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']))
|
|
|
|
|
|
def route_namespace(session_key):
|
|
"""The ``<ns>`` slot of an ``agent:<ns>:...`` routing key (``main`` for the launch profile, else
|
|
the profile's ``_session_key_namespace`` slot); ``main`` for a key outside that shape."""
|
|
parts = str(session_key or '').split(':')
|
|
return parts[1] if len(parts) > 2 and parts[0] == 'agent' and parts[1] else 'main'
|
|
|
|
|
|
def route_fence_key(namespace, session_id):
|
|
return f'{ROUTE_FENCE_PREFIX}{namespace}:{session_id}'
|
|
|
|
|
|
# True when a routing save must skip an entry, checked only against the profile the route belongs
|
|
# to: this store's own marker for a launch (``main``) route, that profile's fence for a secondary's.
|
|
# A launch-keyed route to an id this store holds no session of can only name another profile's
|
|
# session, so a secondary's fence covers it too (``fence_shared_routes``). A sweep's fence stops
|
|
# fencing at the horizon by itself: only the owner's own store sweeps the owner's marker, and nothing
|
|
# guarantees the launch profile ever sweeps this one. A delete's fence (no ``expires``) never does:
|
|
# its age test is NULL, which ``coalesce`` reads as not expired. An unscoped marker (no ``own``: an
|
|
# earlier release's, written for either) still fences every route. So does a marker or fence that is
|
|
# not valid JSON (no writer stores one): ``json_extract`` raises ``malformed JSON`` on it even behind
|
|
# ``json_valid(...) AND``, so each JSON read sits in a ``CASE WHEN json_valid(...)`` branch.
|
|
# Bound by name (``route_retired_params``): :entry is the route's JSON, :ns its key's
|
|
# ``route_namespace``, :now the save's time.
|
|
ROUTE_RETIRED_SQL = """(
|
|
EXISTS (SELECT 1 FROM state_meta m
|
|
WHERE m.key = :retired_prefix || json_extract(:entry, '$.session_id')
|
|
AND (:ns = 'main' OR NOT CASE WHEN json_valid(m.value)
|
|
THEN json_extract(m.value, '$.own') IS NOT NULL ELSE 0 END))
|
|
OR EXISTS (SELECT 1 FROM state_meta f
|
|
WHERE f.key = :fence_prefix || :ns || ':' || json_extract(:entry, '$.session_id')
|
|
AND NOT CASE WHEN json_valid(f.value)
|
|
THEN coalesce(json_extract(f.value, '$.expires')
|
|
AND json_extract(f.value, '$.retired_at') < :now - :horizon, 0)
|
|
ELSE 0 END
|
|
AND (:ns != 'main' OR NOT EXISTS (
|
|
SELECT 1 FROM sessions s WHERE s.id = json_extract(:entry, '$.session_id')))))"""
|
|
|
|
|
|
def route_retired_params(session_key, entry_json, now):
|
|
"""The named parameters of ``ROUTE_RETIRED_SQL`` for one routing entry saved at *now*."""
|
|
return {'entry': entry_json, 'ns': route_namespace(session_key), 'now': now, 'horizon': RETRY_HORIZON_S,
|
|
'retired_prefix': RETIRED_PREFIX, 'fence_prefix': ROUTE_FENCE_PREFIX}
|
|
|
|
|
|
def route_retired(db, session_key, entry_json):
|
|
"""A routing save (or the legacy sessions.json import) must skip this entry."""
|
|
with db._read_ctx() as conn:
|
|
return bool(conn.execute('SELECT ' + ROUTE_RETIRED_SQL,
|
|
route_retired_params(session_key, entry_json, time.time())).fetchone()[0])
|
|
|
|
|
|
def expiring_retirements(db, session_ids):
|
|
"""The ids among *session_ids* whose own marker in *db* is a sweep's (ages out at the horizon)."""
|
|
with db._read_ctx() as conn:
|
|
return {key[len(RETIRED_PREFIX):] for (key,) in conn.execute(
|
|
"SELECT key FROM state_meta WHERE key IN (SELECT ? || value FROM json_each(?)) "
|
|
"AND json_valid(value) AND json_extract(value, '$.expires')",
|
|
(RETIRED_PREFIX, json.dumps(list(session_ids))))}
|
|
|
|
|
|
def fence_shared_routes(conn, namespace, session_ids, *, expiring=()):
|
|
"""In the shared routing index (*conn*), fence the routes of *session_ids*, retired in the
|
|
profile whose routing keys use *namespace*, and drop them. The fence is keyed by (profile, id)
|
|
and copies the owner's marker: permanent for a delete, aging out with an *expiring* sweep's.
|
|
Unless this store holds a session of that id, the launch (``main``) namespace is fenced too: a
|
|
launch-keyed route to it names the retired session. The sweep fences that aged out here are
|
|
reclaimed in the same write, so the owner's prunes bound them without the launch profile's sweep."""
|
|
_reclaim_expired(conn, ROUTE_FENCE_PREFIX, time.time() - RETRY_HORIZON_S)
|
|
expiring, owned = set(expiring), {}
|
|
for sid in session_ids:
|
|
names = {namespace}
|
|
if not conn.execute('SELECT 1 FROM sessions WHERE id=?', (sid,)).fetchone():
|
|
names.add('main')
|
|
for name in names:
|
|
_mark_retired(conn, route_fence_key(name, sid), expires=sid in expiring)
|
|
owned[sid] = names
|
|
for row in conn.execute('SELECT scope,session_key,entry_json FROM gateway_routing').fetchall():
|
|
if route_namespace(row['session_key']) in owned.get(entry_session_id(row['entry_json']), ()):
|
|
conn.execute('DELETE FROM gateway_routing WHERE scope=? AND session_key=?',
|
|
(row['scope'], row['session_key']))
|
|
|
|
|
|
RETRY_HORIZON_S = 7 * 86400 # exact-retry horizon: a retry within it gets the byte-exact settled reply
|
|
_EXPIRED = "json_valid(value) AND json_extract(value, '$.retired_at') < ?"
|
|
|
|
|
|
def _reclaim_expired(conn, prefix, cutoff):
|
|
"""Drop the sweep (``expires``) markers under *prefix* retired before *cutoff*."""
|
|
conn.execute(f"DELETE FROM state_meta WHERE key GLOB ? AND {_EXPIRED} AND json_extract(value, '$.expires')",
|
|
(prefix + '*', cutoff))
|
|
|
|
|
|
def prune_retired_tombstones(conn, older_than=RETRY_HORIZON_S):
|
|
"""Reclaim deletion tombstones retired more than *older_than* seconds ago: the terminal admission
|
|
with its identity pointer and compacted result, terminal workers and a sweep's (``expires``)
|
|
retired-session marker or shared routing fence. A user delete's marker is the deletion fence itself and is never
|
|
reclaimed: it holds only ``retired_at``, and without it a deleted id could be recreated,
|
|
re-admitted or reissued as a Bot Chat generation.
|
|
Tombstones written before ``retired_at`` existed are stamped now and age from this pass."""
|
|
from hermes_state_terminal import ADMISSION_PREFIX, RESULT_PREFIX, WORKER_PREFIX, identity_key
|
|
prefixes = (ADMISSION_PREFIX, WORKER_PREFIX)
|
|
now = time.time()
|
|
for prefix in prefixes:
|
|
conn.execute("UPDATE state_meta SET value=json_set(value, '$.retired_at', ?) WHERE key GLOB ? "
|
|
"AND json_valid(value) AND json_extract(value, '$.retired_at') IS NULL", (now, prefix + '*'))
|
|
cutoff = now - older_than
|
|
for key, raw in conn.execute(f'SELECT key,value FROM state_meta WHERE key GLOB ? AND {_EXPIRED}',
|
|
(ADMISSION_PREFIX + '*', cutoff)).fetchall():
|
|
row = json.loads(raw)
|
|
conn.execute('DELETE FROM state_meta WHERE key IN (?,?,?)', (key, RESULT_PREFIX + row['admission_id'],
|
|
identity_key(row['principal_id'], row['target_session_id'], row['request_id'])))
|
|
conn.execute(f'DELETE FROM state_meta WHERE key GLOB ? AND {_EXPIRED}', (WORKER_PREFIX + '*', cutoff))
|
|
for prefix in (RETIRED_PREFIX, ROUTE_FENCE_PREFIX):
|
|
_reclaim_expired(conn, prefix, cutoff)
|