mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
5e327d2b26
An attachment send that reached the owner but whose landed check missed comes back as text under its original submission id. Enter on it was refused `admission_conflict` (same id, other payload): the words stayed in the composer under a "Prompt failed" card, although the send had run once. `admission_conflict` alone proves nothing (route, binding and lineage checks raise it too), so the three checks that DO prove "this principal's id on this target is already admitted" now name it: `field=request_id` (hermes_state_runtime.ADMITTED_REQUEST_CONFLICT) in admit_session_input, retry_terminal_admission and the attachment-bytes check. Desktop settles exactly that refusal as the dedupe it is (a `terminal` receipt: bubble dropped, journal retired, draft cleared, no new id). Every other conflict, and a backend that names no field, stays a visible failure. An edited restored draft already drops its id (matchingDraftRetry), so a deliberate edit is a new message.
1006 lines
56 KiB
Python
1006 lines
56 KiB
Python
"""Canonical runtime transactions; callers authorize and resolve session identity.
|
|
|
|
No execution, transport, or authority objects live here. Every mutation uses the
|
|
SessionDB transaction owner, including its inode guard and SQLite retry policy.
|
|
"""
|
|
import json
|
|
import uuid
|
|
from typing import NamedTuple, Optional
|
|
|
|
from agent.conversation_compression_archive import ABSORBED_ROW_IDS
|
|
from agent.message_metadata import (
|
|
CANONICAL_ROW, DATA_USE, DB_ROW_SNAPSHOT, MESSAGE_UID, PERSISTENCE_ONLY_MESSAGE_FIELDS, TOOL_CALL_UID, TOOL_CALL_UIDS,
|
|
)
|
|
from hermes_state_keys import admission_fingerprint
|
|
|
|
|
|
class SessionGoneEvent(NamedTuple):
|
|
"""The ``on_session_delete`` event a committed runtime mutation owes its plugins."""
|
|
session_ids: tuple
|
|
reason: str
|
|
|
|
|
|
class RuntimeStoreError(ValueError):
|
|
"""A refusal by the state authority. ``reason`` is the bounded wire code (callers compare it as a
|
|
string); ``field`` and ``detail`` name the check that fired and why, so a client and the gateway
|
|
log can tell the reasons' many checks apart instead of reporting the bare code."""
|
|
|
|
def __init__(self, reason: str, *, field: str | None = None, detail: str | None = None):
|
|
self.reason = reason
|
|
self.field = _one_line(field, 80)
|
|
self.detail = _one_line(detail, 160)
|
|
super().__init__(reason)
|
|
|
|
|
|
#: The one ``admission_conflict`` that proves the request was already admitted: this principal's
|
|
#: request id on this target holds an admission with another payload. Every other conflict (route,
|
|
#: binding, lineage) carries no such proof, so a client tells them apart by this field.
|
|
ADMITTED_REQUEST_CONFLICT = {'field': 'request_id', 'detail': 'this request id was already admitted with another payload'}
|
|
|
|
|
|
def _one_line(value, limit):
|
|
"""A refusal's ``field``/``detail`` as the log line and the client frame carry it: either may hold
|
|
client text, so whitespace collapses to single spaces, other non-printables drop (no forged log
|
|
lines), and the length is bounded below the client's frame limit."""
|
|
if value is None:
|
|
return None
|
|
text = ''.join(ch for ch in ' '.join(str(value).split()) if ch.isprintable())
|
|
return text[:limit] or None
|
|
|
|
|
|
def _text(value):
|
|
if not isinstance(value, str) or not value or len(value) > 1024:
|
|
raise RuntimeStoreError('invalid_params')
|
|
return value
|
|
|
|
|
|
def _json(value):
|
|
if not isinstance(value, dict):
|
|
raise RuntimeStoreError('invalid_params')
|
|
try:
|
|
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False)
|
|
except (TypeError, ValueError, UnicodeError) as exc:
|
|
raise RuntimeStoreError('invalid_params') from exc
|
|
|
|
|
|
def _epoch(conn, epoch):
|
|
row = conn.execute('SELECT epoch FROM runtime_epoch WHERE singleton=1').fetchone()
|
|
if type(epoch) is not int or row is None or row[0] != epoch:
|
|
raise RuntimeStoreError('stale_epoch')
|
|
|
|
|
|
def _session(conn, session_id):
|
|
row = conn.execute('SELECT * FROM sessions WHERE id=?', (session_id,)).fetchone()
|
|
if row is None:
|
|
raise RuntimeStoreError('not_found')
|
|
return row
|
|
|
|
|
|
def _admission(conn, admission_id):
|
|
row = conn.execute('SELECT * FROM session_admissions WHERE admission_id=?', (admission_id,)).fetchone()
|
|
if row is None:
|
|
from hermes_state_terminal import terminal_admission
|
|
row = terminal_admission(conn, admission_id)
|
|
if row is None:
|
|
raise RuntimeStoreError('not_found')
|
|
return row
|
|
|
|
|
|
def _retire_admission_workers(conn, row, owner_epoch):
|
|
"""Retire the worker rows this admission's generation owns on BOTH the logical
|
|
FIFO owner and its current physical transcript (reset/compression move the private
|
|
worker assignment to the child while target_session_id stays the logical ID)."""
|
|
lineage = json.loads(row['lineage_json'])
|
|
targets = list({row['target_session_id'], lineage[-1] if lineage else row['target_session_id']})
|
|
scope = f"session_id IN ({','.join('?' * len(targets))}) AND generation=? AND owner_epoch=?"
|
|
# Settlement closes an execution that never wrote execution.finish: its read copies go too.
|
|
from hermes_state_runtime_workers import compact_terminal_receipts
|
|
for (execution_id,) in conn.execute(f"SELECT execution_id FROM worker_executions WHERE {scope} "
|
|
"AND status!='terminal'", (*targets, row['generation'], owner_epoch)).fetchall():
|
|
compact_terminal_receipts(conn, execution_id)
|
|
conn.execute(f"UPDATE worker_executions SET status='terminal' WHERE {scope}",
|
|
(*targets, row['generation'], owner_epoch))
|
|
|
|
|
|
def _row(row):
|
|
if row is None:
|
|
return None
|
|
result = dict(row)
|
|
result['payload'] = json.loads(result.pop('payload_json'))
|
|
result.pop('payload_digest')
|
|
result.pop('lineage_json')
|
|
result.pop('retired_at', None) # tombstone retention stamp, not admission state
|
|
return result
|
|
|
|
|
|
def begin_runtime_epoch(db, *, instance_id: str) -> int:
|
|
"""Caller MUST hold the profile ownership lock, not merely an endpoint ticket."""
|
|
_text(instance_id)
|
|
def write(conn):
|
|
conn.execute('''INSERT INTO runtime_epoch(singleton,epoch,instance_id) VALUES(1,1,?)
|
|
ON CONFLICT(singleton) DO UPDATE SET epoch=runtime_epoch.epoch+1, instance_id=excluded.instance_id''', (instance_id,))
|
|
return conn.execute('SELECT epoch FROM runtime_epoch WHERE singleton=1').fetchone()[0]
|
|
return db._execute_write(write)
|
|
|
|
|
|
# Main's busy-queue cap (GatewayRunner._BUSY_QUEUE_MAX_PENDING): queued inputs one session holds.
|
|
ADMISSION_QUEUE_MAX = 32
|
|
|
|
|
|
def admit_session_input(db, *, epoch: int, principal_id: str, session_id: str,
|
|
request_id: str, payload: dict, intent: str = 'queue',
|
|
_authorize_write=None, _replay_guard=None) -> dict:
|
|
"""Admit input; the trusted private guard raises to refuse a NEW write.
|
|
|
|
The guard receives the owning transaction connection, must not commit it or
|
|
perform external effects, and may run again on SQLite retry. Exact existing
|
|
and terminal replays bypass it: they cannot create or change accepted work.
|
|
``_replay_guard(conn)`` runs on the exact live replay instead, inside the same
|
|
transaction (a replacement's binding to the admission it retired).
|
|
"""
|
|
for value in (principal_id, session_id, request_id):
|
|
_text(value)
|
|
if intent not in ('queue', 'steer', 'redirect'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
encoded = _json(payload)
|
|
from hermes_state_terminal import retry_terminal_admission
|
|
retired = retry_terminal_admission(db, epoch=epoch, principal_id=principal_id, session_id=session_id,
|
|
request_id=request_id, payload=payload, intent=intent)
|
|
if retired is not None:
|
|
return retired
|
|
digest = admission_fingerprint(canonical_target=session_id, payload={'input': json.loads(encoded), 'intent': intent})
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
_session(conn, session_id)
|
|
old = conn.execute('''SELECT * FROM session_admissions
|
|
WHERE principal_id=? AND target_session_id=? AND request_id=?''', (principal_id, session_id, request_id)).fetchone()
|
|
if old is not None:
|
|
if old['payload_digest'] != digest:
|
|
raise RuntimeStoreError('admission_conflict', **ADMITTED_REQUEST_CONFLICT)
|
|
if _replay_guard is not None:
|
|
_replay_guard(conn)
|
|
return _row(old)
|
|
queued = conn.execute('''SELECT COUNT(*) FROM session_admissions WHERE target_session_id=?
|
|
AND status='queued' ''', (session_id,)).fetchone()[0]
|
|
# Main's cap holds in both states: a FIFO paused behind an unknown turn runs every accepted
|
|
# input after Discard, so it must not grow without bound while nobody resolves it.
|
|
if queued >= ADMISSION_QUEUE_MAX:
|
|
raise RuntimeStoreError('queue_full')
|
|
if _authorize_write is not None:
|
|
_authorize_write(conn)
|
|
admission_id = uuid.uuid4().hex
|
|
conn.execute('''INSERT INTO session_admissions(admission_id,request_id,principal_id,
|
|
target_session_id,lineage_json,payload_json,payload_digest,intent,status,owner_epoch)
|
|
VALUES(?,?,?,?,?,?,?,?,'queued',?)''',
|
|
(admission_id, request_id, principal_id, session_id, json.dumps([session_id]), encoded, digest, intent, epoch))
|
|
return _row(_admission(conn, admission_id))
|
|
return db._execute_write(write)
|
|
|
|
|
|
def get_session_admission(db, *, admission_id: str) -> dict | None:
|
|
with db._read_ctx() as conn:
|
|
from hermes_state_terminal import terminal_admission
|
|
row = conn.execute('SELECT * FROM session_admissions WHERE admission_id=?', (admission_id,)).fetchone()
|
|
return _row(row if row is not None else terminal_admission(conn, admission_id))
|
|
|
|
|
|
def find_session_admission(db, *, principal_id: str, session_id: str, request_id: str) -> dict | None:
|
|
"""The admission at its exact durable identity: the UNIQUE (principal, target, request) key."""
|
|
with db._read_ctx() as conn:
|
|
return _row(conn.execute('SELECT * FROM session_admissions WHERE principal_id=? AND target_session_id=? '
|
|
'AND request_id=?', (principal_id, session_id, request_id)).fetchone())
|
|
|
|
|
|
def list_session_admissions(db, *, session_id: str, pending_only: bool = True) -> list[dict]:
|
|
with db._read_ctx() as conn:
|
|
# Pending is an explicit index-range IN, not ``!='terminal'``: a negated status cannot
|
|
# bound the (target, status, seq) index, so every probe scanned the session's history.
|
|
sql = ("SELECT * FROM session_admissions WHERE target_session_id=? "
|
|
+ ("AND status IN ('queued','started','unknown') " if pending_only else "") + "ORDER BY seq")
|
|
return [_row(row) for row in conn.execute(sql, (session_id,))]
|
|
|
|
|
|
_ANY_HEAD = object()
|
|
|
|
|
|
def claim_session_input(db, *, epoch: int, session_id: str, _guard=None, head=_ANY_HEAD) -> dict | bool | None:
|
|
"""Claim the FIFO head. ``_guard(conn)`` (owner admission gate) raises to refuse inside the
|
|
write transaction, so a drain that began while this waited on the writer still wins.
|
|
``head`` is the admission id the caller's preclaim checks validated (None: it saw nothing
|
|
queued); if the head is another row by the time this write runs, nothing is claimed and False
|
|
tells the caller to re-read and validate the row that is now first."""
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
if _guard is not None:
|
|
_guard(conn)
|
|
session = _session(conn, session_id)
|
|
blocked = conn.execute("SELECT status FROM session_admissions WHERE target_session_id=? AND status IN ('started','unknown')", (session_id,)).fetchall()
|
|
if any(row[0] == 'unknown' for row in blocked):
|
|
raise RuntimeStoreError('unknown_execution')
|
|
from hermes_state_runtime_workers import worker_states
|
|
workers = worker_states(conn, session_id)
|
|
if 'unknown' in workers:
|
|
raise RuntimeStoreError('unknown_execution')
|
|
if blocked or workers:
|
|
return None
|
|
row = conn.execute("SELECT * FROM session_admissions WHERE target_session_id=? AND status='queued' ORDER BY seq LIMIT 1", (session_id,)).fetchone()
|
|
if head is not _ANY_HEAD and (row['admission_id'] if row is not None else None) != head:
|
|
return False
|
|
if row is None:
|
|
return None
|
|
generation = session['runtime_generation'] + 1
|
|
conn.execute('UPDATE sessions SET runtime_generation=? WHERE id=?', (generation, session_id))
|
|
changed = conn.execute("UPDATE session_admissions SET status='started',owner_epoch=?,generation=? WHERE admission_id=? AND status='queued'", (epoch, generation, row['admission_id']))
|
|
if changed.rowcount != 1:
|
|
raise RuntimeStoreError('stale_generation')
|
|
claimed = _row(_admission(conn, row['admission_id']))
|
|
# The transcript high-water mark at the claim (``messages.id`` is AUTOINCREMENT across every
|
|
# session): a row above it on this conversation was written by this turn, so settlement can
|
|
# tell this turn's own failed-turn boundary from an earlier turn's (#134889 GW-21).
|
|
claimed['transcript_mark'] = conn.execute('SELECT COALESCE(MAX(id), 0) FROM messages').fetchone()[0]
|
|
return claimed
|
|
return db._execute_write(write)
|
|
|
|
|
|
def settle_session_input(db, *, epoch: int, admission_id: str, generation: int, outcome: str,
|
|
result: dict | None = None, _transfers=None, _terminal_write=None) -> dict:
|
|
"""``_terminal_write(conn, row)`` is the owner's same-transaction effect (a transcript notice,
|
|
``gateway.session_results.finish_result``); like ``resolve_unknown_session_input``'s it must
|
|
not commit, may run again on SQLite retry, and raising rolls the settlement back."""
|
|
if outcome not in ('completed', 'interrupted', 'rejected', 'failed'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
from hermes_state_terminal import compact_result, encode_result
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
row = _admission(conn, admission_id)
|
|
session = _session(conn, row['target_session_id'])
|
|
if (row['status'] != 'started' or row['owner_epoch'] != epoch
|
|
or type(generation) is not int or row['generation'] != generation
|
|
or session['runtime_generation'] != generation):
|
|
raise RuntimeStoreError('stale_generation')
|
|
_retire_admission_workers(conn, row, epoch)
|
|
if _terminal_write is not None:
|
|
_terminal_write(conn, row)
|
|
if result is not None:
|
|
from hermes_state_terminal import RESULT_PREFIX
|
|
encoded = _json(encode_result(compact_result(result, user_message=json.loads(row['payload_json']).get('text'))))
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?) '
|
|
'ON CONFLICT(key) DO UPDATE SET value=excluded.value',
|
|
(RESULT_PREFIX + admission_id, encoded))
|
|
conn.execute("UPDATE session_admissions SET status='terminal',outcome=? WHERE admission_id=?", (outcome, admission_id))
|
|
settle_steer_transfers(conn, epoch=epoch, generation=generation, transfers=_transfers, outcome=outcome)
|
|
conn.execute('UPDATE sessions SET runtime_revision=runtime_revision+1 WHERE id=?', (row['target_session_id'],))
|
|
return _row(_admission(conn, admission_id))
|
|
return db._execute_write(write)
|
|
|
|
|
|
_CANCEL_QUEUED = "UPDATE session_admissions SET status='terminal',outcome='cancelled' WHERE admission_id=?"
|
|
|
|
|
|
def cancel_session_input(db, *, epoch: int, admission_id: str) -> dict:
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
row = _admission(conn, admission_id)
|
|
if row['status'] == 'unknown':
|
|
raise RuntimeStoreError('unknown_execution')
|
|
# A steer in transfer reads as its running claim and settles with it: as started, not cancellable.
|
|
if row['status'] == 'started' or row['outcome'] == STEERING:
|
|
raise RuntimeStoreError('stale_generation')
|
|
if row['status'] == 'queued':
|
|
conn.execute(_CANCEL_QUEUED, (admission_id,))
|
|
return _row(_admission(conn, admission_id))
|
|
return db._execute_write(write)
|
|
|
|
|
|
# A queued admission handed to its FIFO's running claim: terminal (never claimable, outside the queue
|
|
# cap) but not yet an outcome. It is bound to that claim's ``(owner_epoch, generation)`` and settles
|
|
# with it: ``steered`` once the claim completes with the text delivered, ``interrupted`` otherwise, or
|
|
# back to ``queued`` (its own FIFO place) when the text never reached the claim's agent.
|
|
STEERING = 'steering'
|
|
|
|
|
|
def steer_replayed(row) -> bool:
|
|
"""``row`` is a committed transfer (still in flight, or settled ``steered``): an exact retry of
|
|
``prompt.steer_queued`` replays it."""
|
|
return row['status'] == 'terminal' and row['outcome'] in (STEERING, 'steered')
|
|
|
|
|
|
def _requeue_transfer(conn, *, admission_id: str, epoch: int, generation: int) -> bool:
|
|
"""A transfer still bound to the claim ``(epoch, generation)`` goes back to its FIFO place (``seq``)."""
|
|
return conn.execute("UPDATE session_admissions SET status='queued',outcome=NULL,generation=NULL "
|
|
"WHERE admission_id=? AND status='terminal' AND outcome=? AND owner_epoch=? "
|
|
"AND generation=?", (admission_id, STEERING, epoch, generation)).rowcount == 1
|
|
|
|
|
|
def steer_session_input(db, *, epoch: int, admission_id: str, generation: int) -> tuple[dict, bool]:
|
|
"""Hand a still-queued admission to its FIFO's running claim: queued -> terminal ``steering``
|
|
bound to ``generation`` in one write, only while ``generation`` is the started claim this owner
|
|
holds, so the drain can never also claim it. ``(row, True)`` when this call committed the
|
|
transfer, ``(row, False)`` for an exact retry of one already committed (never delivered twice)."""
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
row = _admission(conn, admission_id)
|
|
if steer_replayed(row):
|
|
return _row(row), False
|
|
if row['status'] != 'queued':
|
|
raise RuntimeStoreError('admission_started')
|
|
if type(generation) is not int or conn.execute(
|
|
"SELECT 1 FROM session_admissions WHERE target_session_id=? AND status='started' "
|
|
"AND owner_epoch=? AND generation=?", (row['target_session_id'], epoch, generation)).fetchone() is None:
|
|
raise RuntimeStoreError('session_idle')
|
|
conn.execute("UPDATE session_admissions SET status='terminal',outcome=?,owner_epoch=?,generation=? "
|
|
"WHERE admission_id=?", (STEERING, epoch, generation, admission_id))
|
|
return _row(_admission(conn, admission_id)), True
|
|
return db._execute_write(write)
|
|
|
|
|
|
def settle_steer_transfers(conn, *, epoch: int, generation: int, transfers, outcome: str) -> None:
|
|
"""In the receiving claim's terminal write: each admission bound to it settles ``steered`` only if
|
|
its text reached the claim's agent (``transfers[id]``) and the claim completed; a delivered text
|
|
under any other ending is ``interrupted`` (never replayed); an undelivered one is queued again."""
|
|
for admission_id, delivered in (transfers or {}).items():
|
|
if not delivered:
|
|
_requeue_transfer(conn, admission_id=admission_id, epoch=epoch, generation=generation)
|
|
continue
|
|
conn.execute("UPDATE session_admissions SET outcome=? WHERE admission_id=? AND status='terminal' "
|
|
"AND outcome=? AND owner_epoch=? AND generation=?",
|
|
('steered' if outcome == 'completed' else 'interrupted', admission_id, STEERING, epoch, generation))
|
|
|
|
|
|
def revert_steer_input(db, *, epoch: int, admission_id: str, generation: int) -> bool:
|
|
"""A transfer whose text never reached the claim's agent: the admission is queued again, in its
|
|
original FIFO place (its ``seq``). False when the claim's settlement already decided it."""
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
return _requeue_transfer(conn, admission_id=admission_id, epoch=epoch, generation=generation)
|
|
return db._execute_write(write)
|
|
|
|
|
|
REPLACEMENT_PREFIX = 'gateway.queued_replacement.v1.'
|
|
|
|
|
|
def replacement_binding(principal_id: str, session_id: str, request_id: str) -> str:
|
|
"""The durable key binding a replacement's identity to the admission it retired."""
|
|
return REPLACEMENT_PREFIX + admission_fingerprint(canonical_target=session_id,
|
|
payload={'principal': principal_id, 'request': request_id})
|
|
|
|
|
|
def retire_replaced_input(conn, admission_id: str, binding: str) -> None:
|
|
"""Inside the replacement's own admission write: the replaced row, still queued, is cancelled
|
|
in the same transaction that admits its successor at the FIFO tail, which is bound to it."""
|
|
row = _admission(conn, admission_id)
|
|
if row['status'] != 'queued':
|
|
raise RuntimeStoreError('admission_started')
|
|
conn.execute(_CANCEL_QUEUED, (admission_id,))
|
|
conn.execute('INSERT OR REPLACE INTO state_meta(key,value) VALUES(?,?)', (binding, json.dumps(admission_id)))
|
|
|
|
|
|
def check_replaced_input(conn, admission_id: str, binding: str) -> None:
|
|
"""The exact replay of a replacement answers only the request that retired ``admission_id``."""
|
|
saved = conn.execute('SELECT value FROM state_meta WHERE key=?', (binding,)).fetchone()
|
|
if saved is None or json.loads(saved[0]) != admission_id:
|
|
raise RuntimeStoreError('admission_conflict')
|
|
|
|
|
|
def rehome_queued_inputs(db, *, epoch: int, old_session_id: str, new_session_id: str,
|
|
principal_prefix: str) -> list[str]:
|
|
"""Re-target the old owner's still-queued ``principal_prefix*`` admissions onto
|
|
``new_session_id`` in one transaction (the compression-exhaustion auto-reset). Each row keeps
|
|
its admission id, request id and seq, so FIFO order and retry dedupe survive; the digest is
|
|
re-keyed to the new target. Nothing moves while the old FIFO holds an ``unknown`` row (the
|
|
operator's to resume). Returns the moved admission ids in FIFO order."""
|
|
for value in (old_session_id, new_session_id, principal_prefix):
|
|
_text(value)
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
_session(conn, new_session_id)
|
|
if conn.execute("SELECT 1 FROM session_admissions WHERE target_session_id=? AND status='unknown'",
|
|
(old_session_id,)).fetchone():
|
|
return []
|
|
rows = conn.execute("SELECT * FROM session_admissions WHERE target_session_id=? AND status='queued' "
|
|
"AND substr(principal_id,1,?)=? ORDER BY seq",
|
|
(old_session_id, len(principal_prefix), principal_prefix)).fetchall()
|
|
for row in rows:
|
|
digest = admission_fingerprint(canonical_target=new_session_id, payload={
|
|
'input': json.loads(row['payload_json']), 'intent': row['intent']})
|
|
conn.execute("UPDATE session_admissions SET target_session_id=?,lineage_json=?,payload_digest=? "
|
|
"WHERE admission_id=? AND status='queued'",
|
|
(new_session_id, json.dumps([new_session_id]), digest, row['admission_id']))
|
|
if rows:
|
|
for sid in (old_session_id, new_session_id):
|
|
conn.execute('UPDATE sessions SET runtime_revision=runtime_revision+1 WHERE id=?', (sid,))
|
|
return [row['admission_id'] for row in rows]
|
|
return db._execute_write(write)
|
|
|
|
|
|
def retarget_local_automation(db, *, epoch: int, session_id: str, principal_id: str,
|
|
old_targets, new_target: str) -> list[str]:
|
|
"""Point a LOCAL owner's still-queued ``principal_id`` automation rows (a Bot Chat's queued
|
|
``hermes peer dm`` turns, plugin and completion notices) that name a pre-reset transcript in
|
|
``old_targets`` at ``new_target``, in one transaction (the compression-exhaustion auto-reset).
|
|
A LOCAL FIFO's owner is its receipt id and survives the reset; only the descriptor's ``target``
|
|
goes stale, and the drain refuses a stale one (``admission_conflict``), pausing the FIFO for
|
|
good. Admission id, request id and seq are kept; the digest is re-keyed to the new payload.
|
|
Nothing moves behind an ``unknown`` head. Returns the retargeted admission ids in FIFO order."""
|
|
for value in (session_id, principal_id, new_target, *old_targets):
|
|
_text(value)
|
|
old_targets = set(old_targets)
|
|
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
_session(conn, new_target)
|
|
if conn.execute("SELECT 1 FROM session_admissions WHERE target_session_id=? AND status='unknown'",
|
|
(session_id,)).fetchone():
|
|
return []
|
|
moved = []
|
|
for row in conn.execute("SELECT * FROM session_admissions WHERE target_session_id=? AND status='queued' "
|
|
"AND principal_id=? ORDER BY seq", (session_id, principal_id)).fetchall():
|
|
payload = json.loads(row['payload_json'])
|
|
descriptor = payload.get('local_automation_v1') if isinstance(payload, dict) else None
|
|
if not isinstance(descriptor, dict) or descriptor.get('target') not in old_targets:
|
|
continue
|
|
descriptor['target'] = new_target
|
|
digest = admission_fingerprint(canonical_target=session_id, payload={'input': payload, 'intent': row['intent']})
|
|
conn.execute("UPDATE session_admissions SET payload_json=?,payload_digest=? WHERE admission_id=? AND status='queued'",
|
|
(_json(payload), digest, row['admission_id']))
|
|
moved.append(row['admission_id'])
|
|
if moved:
|
|
conn.execute('UPDATE sessions SET runtime_revision=runtime_revision+1 WHERE id=?', (session_id,))
|
|
return moved
|
|
return db._execute_write(write)
|
|
|
|
|
|
def recover_session_inputs(db, *, epoch: int) -> int:
|
|
"""Never replay started work. Live worker adoption is a separate explicit operation."""
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
from hermes_state_runtime_workers import retire_stale_children
|
|
retire_stale_children(conn, epoch)
|
|
conn.execute("UPDATE worker_executions SET status='unknown' WHERE status IN ('registered','running') AND owner_epoch!=?", (epoch,))
|
|
# A steer handed to a lost claim may or may not have reached its agent: never replayed.
|
|
conn.execute("UPDATE session_admissions SET outcome='interrupted' WHERE status='terminal' AND outcome=? "
|
|
"AND owner_epoch!=?", (STEERING, epoch))
|
|
return conn.execute("UPDATE session_admissions SET status='unknown' WHERE status='started' AND owner_epoch!=?", (epoch,)).rowcount
|
|
return db._execute_write(write)
|
|
|
|
|
|
def mutation_receipt_key(principal_id, session_id, request_id, *, operation, payload, expected_revision,
|
|
expected_generation=None):
|
|
"""``(state_meta key, digest)`` of a mutation request's durable exact-retry receipt."""
|
|
key = 'gateway.mutation.v1.' + admission_fingerprint(
|
|
canonical_target=session_id, payload={'principal': principal_id, 'request': request_id})
|
|
fingerprint = {'operation': operation, 'payload': payload, 'expected_revision': expected_revision}
|
|
if expected_generation is not None:
|
|
fingerprint['expected_generation'] = expected_generation
|
|
return key, admission_fingerprint(canonical_target=session_id, payload=fingerprint)
|
|
|
|
|
|
def replay_mutation_receipt(conn, key, digest):
|
|
"""The stored result of an exact retry, ``None`` when unseen; a different request refuses."""
|
|
old = conn.execute('SELECT value FROM state_meta WHERE key=?', (key,)).fetchone()
|
|
if old is None:
|
|
return None
|
|
receipt = json.loads(old[0])
|
|
if receipt['digest'] != digest:
|
|
raise RuntimeStoreError('admission_conflict')
|
|
return receipt['result']
|
|
|
|
|
|
def mutate_runtime_session(db, *, epoch: int, principal_id: str, session_id: str,
|
|
request_id: str, expected_revision: int,
|
|
operation: str, payload: dict, expected_generation: int | None = None,
|
|
_live_guard=None, _authorize_write=None, _prepare_only=False, _prepared=None) -> dict:
|
|
"""Commit a closed metadata edit and its retry receipt in the same transaction.
|
|
|
|
Caller authorizes the principal and resolves the canonical session. These
|
|
metadata edits do not stop execution. Existing direct writers must migrate
|
|
before this seam can provide universal revision fencing.
|
|
"""
|
|
for value in (principal_id, session_id, request_id):
|
|
_text(value)
|
|
if type(expected_revision) is not int or expected_revision < 0:
|
|
raise RuntimeStoreError('invalid_params')
|
|
from hermes_state_mutations import validate_action, apply_action
|
|
validate_action(operation, payload)
|
|
if expected_generation is not None and (type(expected_generation) is not int or expected_generation < 0):
|
|
raise RuntimeStoreError('invalid_params')
|
|
if operation in {'delete', 'rewind', 'reset', 'branch', 'model', 'compress'} and expected_generation is None:
|
|
raise RuntimeStoreError('invalid_params')
|
|
# Snapshot caller data before waiting for the writer lock.
|
|
payload = json.loads(_json(payload))
|
|
key, digest = mutation_receipt_key(principal_id, session_id, request_id, operation=operation, payload=payload,
|
|
expected_revision=expected_revision, expected_generation=expected_generation)
|
|
|
|
def write(conn) -> tuple[dict, Optional[SessionGoneEvent]]:
|
|
"""The receipt result plus the on_session_delete event this commit owes (None on replay)."""
|
|
_epoch(conn, epoch)
|
|
replay = replay_mutation_receipt(conn, key, digest)
|
|
if replay is not None:
|
|
return replay, None
|
|
session = conn.execute('SELECT * FROM sessions WHERE id=?', (session_id,)).fetchone()
|
|
if session is None and operation != 'import':
|
|
raise RuntimeStoreError('not_found')
|
|
revision = session['runtime_revision'] if session else 0
|
|
if revision != expected_revision:
|
|
raise RuntimeStoreError('revision_conflict')
|
|
if expected_generation is not None and (session is None or session['runtime_generation'] != expected_generation):
|
|
raise RuntimeStoreError('stale_generation')
|
|
if _live_guard is not None:
|
|
from hermes_state_mutation_guards import delete_targets
|
|
targets = delete_targets(conn, session_id) if operation == 'delete' else [session_id]
|
|
_live_guard(targets)
|
|
if _authorize_write is not None:
|
|
_authorize_write(conn)
|
|
if _prepare_only:
|
|
from hermes_state_mutation_prepared import local_snapshot
|
|
snapshot = local_snapshot(db, conn, session_id)
|
|
if operation == 'branch':
|
|
# The branch commit's own idle rule (queued input too), before the owner starts servers.
|
|
from hermes_state_mutation_guards import require_idle
|
|
require_idle(db, conn, list({session_id, snapshot['target']}))
|
|
return {'snapshot': snapshot}, None
|
|
affected, projection = apply_action(db, conn, session_id, operation, payload, prepared=_prepared)
|
|
event = None
|
|
if operation == 'delete':
|
|
event = SessionGoneEvent(tuple(projection.get('deleted_ids') or (session_id,)), 'delete')
|
|
elif payload.get('archived') is True and session is not None and not session['archived']:
|
|
event = SessionGoneEvent(tuple(affected) or (session_id,), 'archive')
|
|
conn.executemany('UPDATE sessions SET runtime_revision=runtime_revision+1 WHERE id=?',
|
|
[(target,) for target in affected])
|
|
updated = conn.execute('SELECT * FROM sessions WHERE id=?', (session_id,)).fetchone()
|
|
result = {'session_id': session_id, 'revision': updated['runtime_revision'] if updated else expected_revision + 1,
|
|
'execution_generation': updated['runtime_generation'] if updated else expected_generation,
|
|
'operation': operation, **projection}
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)',
|
|
(key, _json({'digest': digest, 'result': result})))
|
|
return result, event
|
|
result, event = db._execute_write(write)
|
|
if event is not None:
|
|
from hermes_state_common import emit_session_deleted
|
|
emit_session_deleted(db, event.session_ids, reason=event.reason, primary=session_id)
|
|
return result
|
|
|
|
|
|
_IMPORT_KEY = 'gateway.prompt_admissions_import.v1'
|
|
|
|
|
|
def _canonical_chain(conn, session_id):
|
|
# Same selector as SessionDB.get_compression_chain, on OUR transaction connection.
|
|
from hermes_state_compression import _CHAIN_STEP_SQL
|
|
_session(conn, session_id)
|
|
chain = [session_id]
|
|
for _ in range(100):
|
|
child = conn.execute(_CHAIN_STEP_SQL, (chain[-1],)).fetchone()
|
|
if child is None:
|
|
return chain
|
|
if child['id'] in chain:
|
|
raise RuntimeStoreError('admission_conflict')
|
|
chain.append(child['id'])
|
|
raise RuntimeStoreError('admission_conflict')
|
|
|
|
|
|
def _import_legacy_row(conn, row, epoch, principal_id):
|
|
for field in ('admission_id', 'target_session_id', 'root'):
|
|
_text(row[field])
|
|
lineage = json.loads(row['lineage'])
|
|
if not isinstance(lineage, list) or row['target_session_id'] not in lineage:
|
|
raise RuntimeStoreError('invalid_params')
|
|
payload = json.loads(row['payload'])
|
|
encoded = _json(payload)
|
|
intent = payload.get('intent', 'queue')
|
|
intent = 'redirect' if intent == 'interrupt' else intent
|
|
if intent not in ('queue', 'steer', 'redirect') or row['status'] not in ('queued', 'started', 'unknown', 'terminal'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
chain = _canonical_chain(conn, row['target_session_id'])
|
|
target = chain[-1]
|
|
status = 'unknown' if row['status'] == 'started' else row['status']
|
|
generation = row['generation']
|
|
if generation is None and status == 'unknown':
|
|
generation = _session(conn, target)['runtime_generation'] + 1
|
|
if generation is not None and (type(generation) is not int or generation < 0):
|
|
raise RuntimeStoreError('invalid_params')
|
|
digest = admission_fingerprint(canonical_target=target, payload={'input': payload, 'intent': intent})
|
|
conn.execute("""INSERT INTO session_admissions(admission_id,request_id,principal_id,
|
|
target_session_id,lineage_json,payload_json,payload_digest,intent,status,outcome,owner_epoch,generation)
|
|
VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", (row['admission_id'], row['admission_id'], principal_id,
|
|
target, json.dumps(chain), encoded, digest, intent, status, row['outcome'], epoch, generation))
|
|
if generation is not None:
|
|
conn.execute('UPDATE sessions SET runtime_generation=MAX(runtime_generation,?) WHERE id=?', (generation, target))
|
|
|
|
|
|
def import_legacy_session_admissions(db, *, epoch: int, source_path, principal_id: str,
|
|
writers_drained: bool) -> int:
|
|
"""Import a frozen legacy snapshot; original file remains rollback evidence.
|
|
|
|
The caller proves the old writers are drained before invoking this function.
|
|
Fingerprinting uses SQL snapshot data, never a raw open/close on a live inode.
|
|
"""
|
|
from contextlib import closing
|
|
from pathlib import Path
|
|
import hashlib
|
|
import sqlite3
|
|
from hermes_cli.sqlite_safe_read import connect_tracked
|
|
if writers_drained is not True:
|
|
raise RuntimeStoreError('invalid_params')
|
|
_text(principal_id)
|
|
source_path = Path(source_path).resolve(strict=True)
|
|
with closing(connect_tracked(source_path.as_uri() + '?mode=ro', uri=True)) as source:
|
|
source.row_factory = sqlite3.Row
|
|
source.execute('PRAGMA query_only=ON')
|
|
source.execute('BEGIN')
|
|
rows = [dict(r) for r in source.execute('SELECT * FROM admissions ORDER BY seq')]
|
|
executions = []
|
|
if source.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='executions'").fetchone():
|
|
executions = [dict(r) for r in source.execute('SELECT * FROM executions ORDER BY root')]
|
|
fingerprint = hashlib.sha256(_json({'rows': rows, 'executions': executions}).encode('utf-8')).hexdigest()
|
|
marker = _json({'source': str(source_path), 'fingerprint': fingerprint, 'imported_id_count': len(rows), 'principal_id': principal_id})
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
old = conn.execute('SELECT value FROM state_meta WHERE key=?', (_IMPORT_KEY,)).fetchone()
|
|
if old is not None:
|
|
if old[0] != marker:
|
|
raise RuntimeStoreError('admission_conflict')
|
|
return 0
|
|
for execution in executions:
|
|
generation = execution['generation']
|
|
if type(generation) is not int or generation < 0:
|
|
raise RuntimeStoreError('invalid_params')
|
|
target = _canonical_chain(conn, _text(execution['root']))[-1]
|
|
conn.execute('UPDATE sessions SET runtime_generation=MAX(runtime_generation,?) WHERE id=?', (generation, target))
|
|
for row in rows:
|
|
_import_legacy_row(conn, row, epoch, principal_id)
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?)', (_IMPORT_KEY, marker))
|
|
return len(rows)
|
|
return db._execute_write(write)
|
|
|
|
|
|
def resolve_unknown_session_input(db, *, epoch: int, admission_id: str, generation: int,
|
|
_terminal_write=None, _captured=None) -> dict:
|
|
"""Explicit operator acknowledgement; resolves uncertainty, never requeues it.
|
|
|
|
``_terminal_write(conn, row)`` is the owner's same-transaction effect (the discarded turn's
|
|
transcript boundary): it must not commit, may run again on SQLite retry, and raising rolls
|
|
the whole resolution back, so the FIFO can never advance past an unclosed lost turn.
|
|
``_captured=(outcome, stored_result)`` is the exact result this owner captured for the turn
|
|
before its settlement write failed; it commits with the resolution (only for an ``unknown``
|
|
row stamped by this same owner epoch), so the finished answer is not thrown away."""
|
|
from hermes_state_terminal import RESULT_PREFIX
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
row = _admission(conn, admission_id)
|
|
if (type(generation) is int and row['generation'] == generation and row['status'] == 'terminal'
|
|
and (row['outcome'] == 'interrupted' or conn.execute(
|
|
'SELECT 1 FROM state_meta WHERE key=?', (RESULT_PREFIX + admission_id,)).fetchone())):
|
|
return _row(row)
|
|
if type(generation) is not int or row['status'] != 'unknown' or row['generation'] != generation:
|
|
raise RuntimeStoreError('stale_generation')
|
|
outcome = 'interrupted'
|
|
if _captured is not None and row['owner_epoch'] == epoch:
|
|
outcome, stored = _captured
|
|
if outcome not in ('completed', 'interrupted', 'failed'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
from hermes_state_terminal import compact_result, encode_result
|
|
conn.execute('INSERT INTO state_meta(key,value) VALUES(?,?) '
|
|
'ON CONFLICT(key) DO UPDATE SET value=excluded.value',
|
|
(RESULT_PREFIX + admission_id, _json(encode_result(compact_result(
|
|
stored, user_message=json.loads(row['payload_json']).get('text'))))))
|
|
if _terminal_write is not None:
|
|
_terminal_write(conn, row)
|
|
_retire_admission_workers(conn, row, row['owner_epoch'])
|
|
conn.execute("UPDATE session_admissions SET status='terminal',outcome=? WHERE admission_id=?",
|
|
(outcome, admission_id))
|
|
conn.execute('UPDATE sessions SET runtime_revision=runtime_revision+1 WHERE id=?', (row['target_session_id'],))
|
|
return _row(_admission(conn, admission_id))
|
|
return db._execute_write(write)
|
|
|
|
|
|
def _worker_public(row):
|
|
result = dict(row)
|
|
result.pop('adoption_digest')
|
|
return result
|
|
|
|
|
|
def _worker_assignment(conn, execution_id, session_id, generation):
|
|
row = conn.execute('SELECT * FROM worker_executions WHERE execution_id=?', (execution_id,)).fetchone()
|
|
if row is None:
|
|
raise RuntimeStoreError('not_found')
|
|
if row['session_id'] != session_id:
|
|
raise RuntimeStoreError('permission_denied')
|
|
if (type(generation) is not int or row['generation'] != generation
|
|
or _session(conn, session_id)['runtime_generation'] != generation):
|
|
raise RuntimeStoreError('stale_generation')
|
|
return row
|
|
|
|
|
|
def _linked_worker_admission(conn, session_id, generation, owner_epoch, statuses):
|
|
placeholders = ','.join('?' for _ in statuses)
|
|
rows = conn.execute(f"""SELECT * FROM session_admissions
|
|
WHERE generation=? AND owner_epoch=? AND status IN ({placeholders})
|
|
AND (target_session_id=? OR EXISTS (
|
|
SELECT 1 FROM json_each(session_admissions.lineage_json) WHERE value=?
|
|
)) ORDER BY seq""",
|
|
(generation, owner_epoch, *statuses, session_id, session_id)).fetchall()
|
|
if len(rows) > 1:
|
|
raise RuntimeStoreError('admission_conflict')
|
|
return rows[0] if rows else None
|
|
|
|
|
|
def _secret_digest(secret):
|
|
import hashlib
|
|
_text(secret)
|
|
return hashlib.sha256(secret.encode('utf-8')).hexdigest()
|
|
|
|
|
|
def register_worker_execution(db, *, epoch: int, execution_id: str, session_id: str,
|
|
generation: int, kind: str, adoption_secret: str, require_idle: bool = False) -> dict:
|
|
for value in (execution_id, session_id):
|
|
_text(value)
|
|
if kind not in ('cron', 'child', 'compute', 'kanban'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
digest = _secret_digest(adoption_secret)
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
session = _session(conn, session_id)
|
|
if type(generation) is not int or session['runtime_generation'] != generation:
|
|
raise RuntimeStoreError('stale_generation')
|
|
old = conn.execute('SELECT * FROM worker_executions WHERE execution_id=?', (execution_id,)).fetchone()
|
|
if old is not None:
|
|
if (old['session_id'], old['generation'], old['kind'], old['owner_epoch'], old['adoption_digest']) != (session_id, generation, kind, epoch, digest):
|
|
raise RuntimeStoreError('admission_conflict')
|
|
return _worker_public(old)
|
|
from hermes_state_runtime_workers import runtime_lineage, worker_states
|
|
targets = runtime_lineage(conn, session_id)
|
|
if require_idle and any(conn.execute("SELECT 1 FROM session_admissions WHERE target_session_id=? AND status IN ('queued','started','unknown')", (sid,)).fetchone() for sid in targets):
|
|
raise RuntimeStoreError('stale_generation')
|
|
if worker_states(conn, session_id):
|
|
raise RuntimeStoreError('stale_generation')
|
|
if any(conn.execute("SELECT 1 FROM session_admissions WHERE target_session_id=? AND status='unknown'", (sid,)).fetchone() for sid in targets):
|
|
raise RuntimeStoreError('unknown_execution')
|
|
conn.execute("""INSERT INTO worker_executions(execution_id,session_id,kind,owner_epoch,generation,status,adoption_digest)
|
|
VALUES(?,?,?,?,?,'registered',?)""", (execution_id, session_id, kind, epoch, generation, digest))
|
|
return _worker_public(_worker_assignment(conn, execution_id, session_id, generation))
|
|
return db._execute_write(write)
|
|
|
|
|
|
def adopt_worker_execution(db, *, epoch: int, execution_id: str, session_id: str,
|
|
generation: int, adoption_secret: str) -> dict:
|
|
"""Authority first verifies the original producer claim and live worker proof."""
|
|
import hmac
|
|
digest = _secret_digest(adoption_secret)
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
row = _worker_assignment(conn, execution_id, session_id, generation)
|
|
if row['status'] == 'terminal':
|
|
raise RuntimeStoreError('stale_generation')
|
|
if not hmac.compare_digest(row['adoption_digest'], digest):
|
|
raise RuntimeStoreError('permission_denied')
|
|
# An admission-reserved worker reports to the reserving owner over that owner's pipes and
|
|
# dies with them, so only its same-epoch handshake adopts it. A late handshake reaching a
|
|
# restarted owner would re-arm the recovered `unknown` admission as `started` with nobody
|
|
# left to settle it (and Discard refuses `started`); refuse it so recovery's verdict stands.
|
|
if execution_id.startswith('admission-worker:') and row['owner_epoch'] != epoch:
|
|
raise RuntimeStoreError('stale_epoch')
|
|
linked = _linked_worker_admission(
|
|
conn, session_id, generation, row['owner_epoch'], ('started', 'unknown'))
|
|
if linked is not None:
|
|
changed = conn.execute("""UPDATE session_admissions SET owner_epoch=?,status='started'
|
|
WHERE admission_id=? AND generation=? AND owner_epoch=?
|
|
AND status IN ('started','unknown')""",
|
|
(epoch, linked['admission_id'], generation, row['owner_epoch']))
|
|
if changed.rowcount != 1:
|
|
raise RuntimeStoreError('stale_generation')
|
|
changed = conn.execute(
|
|
"UPDATE worker_executions SET owner_epoch=?,status='running' WHERE execution_id=?",
|
|
(epoch, execution_id))
|
|
if changed.rowcount != 1:
|
|
raise RuntimeStoreError('stale_generation')
|
|
return _worker_public(_worker_assignment(conn, execution_id, session_id, generation))
|
|
return db._execute_write(write)
|
|
|
|
|
|
_MESSAGE_FIELDS = frozenset({
|
|
'role', 'content', 'tool_name', 'tool_calls', 'tool_call_id', 'token_count',
|
|
'finish_reason', 'reasoning', 'reasoning_content', 'reasoning_details',
|
|
'codex_reasoning_items', 'codex_message_items', 'platform_message_id', 'message_id',
|
|
'observed', 'effect_disposition', '_compressed_summary', 'timestamp', 'api_content',
|
|
'display_kind', 'display_metadata', '_row_id', '_canonical_content', DB_ROW_SNAPSHOT, CANONICAL_ROW, DATA_USE,
|
|
# Main's durable message identity (message_uid / merge witness / tool-call uids) and the alternation
|
|
# repair's row counts ride every persisted dict; the owner's insert and coverage read them.
|
|
}) | PERSISTENCE_ONLY_MESSAGE_FIELDS | {ABSORBED_ROW_IDS}
|
|
# Compaction handoffs (archive/publish) also carry the summary's user-turn flag. This is the closed
|
|
# boundary the owner validates AND the projection the worker facade applies to live dicts first.
|
|
HANDOFF_MESSAGE_FIELDS = _MESSAGE_FIELDS | {'_compressed_summary_has_user_turn'}
|
|
# Row state the owner's insert and transcript repair stamp on each message (main mutates the caller's
|
|
# dict in place; a worker gets it back as an annotation). None = absent, so a stale adoption is cleared.
|
|
# The owner-minted identity rides too: a worker dict without it re-mints a new uid on every later
|
|
# archive/rotation copy, so the logical message changes identity across generations.
|
|
_ROW_ANNOTATION_KEYS = ('_row_id', 'timestamp', DB_ROW_SNAPSHOT, CANONICAL_ROW,
|
|
MESSAGE_UID, TOOL_CALL_UIDS, TOOL_CALL_UID)
|
|
|
|
|
|
def row_annotations(messages):
|
|
return [{key: msg.get(key) for key in _ROW_ANNOTATION_KEYS} for msg in messages]
|
|
|
|
|
|
def _worker_append(db, conn, session_id, payload):
|
|
if set(payload) - {'messages', 'turn_lease_holder'} or 'messages' not in payload:
|
|
raise RuntimeStoreError('invalid_params')
|
|
messages = payload['messages']
|
|
if not isinstance(messages, list) or len(messages) > 1000:
|
|
raise RuntimeStoreError('invalid_params')
|
|
for msg in messages:
|
|
if (not isinstance(msg, dict) or set(msg) - _MESSAGE_FIELDS
|
|
or msg.get('role') not in ('user', 'assistant', 'system', 'tool')):
|
|
raise RuntimeStoreError('invalid_params')
|
|
# Existing row annotations can adopt only rows in THIS transcript.
|
|
if '_row_id' in msg and not conn.execute(
|
|
'SELECT 1 FROM messages WHERE id=? AND session_id=?',
|
|
(msg['_row_id'], session_id)).fetchone():
|
|
raise RuntimeStoreError('permission_denied')
|
|
holder = payload.get('turn_lease_holder')
|
|
if holder is not None:
|
|
_text(holder)
|
|
count = db._append_messages_in_transaction(conn, session_id, messages, turn_lease_holder=holder)
|
|
return {'count': count, 'annotations': row_annotations(messages)}
|
|
|
|
|
|
def _worker_turn(db, conn, session_id, payload, operation):
|
|
import time
|
|
from hermes_state_compression import _claim_lease_row
|
|
from hermes_state import _compression_lock_holder_process_is_dead
|
|
allowed = {'holder'} if operation == 'turn.release' else {'holder', 'ttl_seconds'}
|
|
if set(payload) != allowed:
|
|
raise RuntimeStoreError('invalid_params')
|
|
holder = _text(payload['holder'])
|
|
ttl = payload.get('ttl_seconds', 300)
|
|
if type(ttl) not in (float, int) or not 0.1 <= ttl <= 3600:
|
|
raise RuntimeStoreError('invalid_params')
|
|
key = db._session_turn_lease_key_on_conn(conn, session_id)
|
|
now = time.time()
|
|
if operation == 'turn.acquire':
|
|
value = _claim_lease_row(conn, 'session_turn_leases', 'conversation_id', key,
|
|
holder, now, now + ttl,
|
|
lambda h, e: float(e) <= now or _compression_lock_holder_process_is_dead(h))[0]
|
|
elif operation == 'turn.renew':
|
|
value = conn.execute('UPDATE session_turn_leases SET expires_at=? WHERE conversation_id=? AND holder=?',
|
|
(now + ttl, key, holder)).rowcount > 0
|
|
else:
|
|
conn.execute('DELETE FROM session_turn_leases WHERE conversation_id=? AND holder=?', (key, holder))
|
|
value = None
|
|
return {'value': value}
|
|
|
|
|
|
def _worker_usage(db, conn, session_id, payload, *, auxiliary=False):
|
|
from hermes_state_usage import _MODEL_USAGE_FIELDS, _TOKEN_COUNTERS
|
|
source = None
|
|
if not auxiliary:
|
|
# ``source`` is the worker platform (turn_usage._agent_session_source). A Desktop
|
|
# managed worker passes ``gui``; created_surface maps that to ``desktop``. The row
|
|
# is usually stamped at admission. A compression child can still be NULL (or an
|
|
# auxiliary-usage ``unknown``), and this worker never create_session's it, so fill
|
|
# only an unset column, as the creator upsert does.
|
|
raw = payload.get('source')
|
|
source = raw if isinstance(raw, str) else None
|
|
payload = {k: v for k, v in payload.items() if k != 'source'}
|
|
allowed = (_MODEL_USAGE_FIELDS - {'billing_mode', 'actual_cost_usd', 'cost_status', 'cost_source'} | {'task'}) if auxiliary else (_MODEL_USAGE_FIELDS | {'pricing_version', 'absolute', 'task'})
|
|
if set(payload) - allowed:
|
|
raise RuntimeStoreError('invalid_params')
|
|
for key, value in payload.items():
|
|
if key in (*_TOKEN_COUNTERS, 'api_call_count'):
|
|
valid = type(value) is int and 0 <= value <= 2**53
|
|
elif key in ('estimated_cost_usd', 'actual_cost_usd'):
|
|
valid = value is None or (type(value) in (int, float) and 0 <= value <= 1e12)
|
|
elif key == 'absolute':
|
|
valid = type(value) is bool
|
|
else:
|
|
valid = value is None or (isinstance(value, str) and len(value) <= 4096)
|
|
if not valid:
|
|
raise RuntimeStoreError('invalid_params')
|
|
if auxiliary:
|
|
if not payload.get('task'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
db._record_model_usage(conn, session_id, **({'api_call_count': 1} | payload))
|
|
else:
|
|
db._update_token_counts_in_transaction(conn, session_id, **payload)
|
|
# The managed worker's platform is the policy label (gui), not the in-process
|
|
# agent surface (desktop). Stamp the same created_surface the insert uses.
|
|
from gateway.session_policy import created_surface
|
|
from hermes_state_compression import _UNSET_CREATED_SQL, stamp_compression_ancestors
|
|
surfaced = created_surface(source)
|
|
if surfaced:
|
|
conn.execute(
|
|
"UPDATE sessions SET created_source=? WHERE id=? AND "
|
|
+ _UNSET_CREATED_SQL.format(col="created_source"),
|
|
(surfaced, session_id))
|
|
stamp_compression_ancestors(conn, session_id)
|
|
return {'value': None}
|
|
|
|
|
|
def _worker_finish(db, conn, session_id, payload):
|
|
if payload:
|
|
raise RuntimeStoreError('invalid_params')
|
|
return {'status': 'terminal'}
|
|
|
|
|
|
def mutate_worker_execution(db, *, epoch, execution_id, session_id, generation,
|
|
sequence, operation, payload):
|
|
"""One closed durable mutation and receipt; never call a self-committing API here."""
|
|
from hermes_state_worker_context import (worker_context, worker_prompt, worker_row_role, worker_sidecars,
|
|
worker_tail_role, worker_tool_names)
|
|
from hermes_state_worker_compression import WORKER_COMPRESSION_HANDLERS
|
|
from hermes_state_worker_lifecycle import WORKER_LIFECYCLE_HANDLERS
|
|
handlers = {
|
|
**WORKER_LIFECYCLE_HANDLERS,
|
|
**WORKER_COMPRESSION_HANDLERS,
|
|
'session.context': worker_context,
|
|
'session.prompt': worker_prompt,
|
|
'session.sidecars': worker_sidecars,
|
|
'session.tools': worker_tool_names,
|
|
'session.tail_role': worker_tail_role,
|
|
'session.row_role': worker_row_role,
|
|
'transcript.append': _worker_append,
|
|
'execution.finish': _worker_finish,
|
|
'usage.main': _worker_usage,
|
|
'usage.auxiliary': lambda db, conn, sid, p: _worker_usage(db, conn, sid, p, auxiliary=True),
|
|
**{name: (lambda db, conn, sid, p, op=name: _worker_turn(db, conn, sid, p, op))
|
|
for name in ('turn.acquire', 'turn.renew', 'turn.release')},
|
|
}
|
|
return apply_worker_receipt(db, epoch=epoch, execution_id=execution_id, session_id=session_id,
|
|
generation=generation, sequence=sequence, operation=operation,
|
|
payload=payload, handlers=handlers)
|
|
|
|
|
|
def apply_worker_receipt(db, *, epoch, execution_id, session_id, generation, sequence,
|
|
operation, payload, handlers):
|
|
"""The one worker receipt transaction every ``worker.persist`` family shares.
|
|
|
|
An exact (execution, sequence, digest) retry returns the stored result; otherwise the
|
|
epoch, assignment, live status and next sequence are fenced, ``handlers[operation]`` runs on
|
|
the transaction connection (it must not commit), and its result is stored as the receipt.
|
|
``execution.finish`` is the only operation that makes the execution terminal.
|
|
"""
|
|
from hermes_state_worker_compression import worker_receipt_assignment
|
|
if type(sequence) is not int or sequence < 1 or not isinstance(operation, str) or operation not in handlers:
|
|
raise RuntimeStoreError('invalid_params')
|
|
encoded = _json(payload)
|
|
if len(encoded.encode('utf-8', errors='surrogatepass')) > 4 * 1024 * 1024:
|
|
raise RuntimeStoreError('invalid_params')
|
|
digest = admission_fingerprint(canonical_target=session_id,
|
|
payload={'operation': operation, 'payload': json.loads(encoded)})
|
|
def write(conn):
|
|
_epoch(conn, epoch)
|
|
row = worker_receipt_assignment(conn, execution_id, session_id, generation, sequence, digest)
|
|
if row['owner_epoch'] != epoch:
|
|
raise RuntimeStoreError('stale_epoch')
|
|
old = conn.execute('SELECT * FROM worker_receipts WHERE execution_id=? AND sequence=?',
|
|
(execution_id, sequence)).fetchone()
|
|
if old is not None:
|
|
if old['payload_digest'] != digest:
|
|
raise RuntimeStoreError('admission_conflict')
|
|
from hermes_state_runtime_workers import retired_receipt
|
|
if retired_receipt(old['result_json']):
|
|
raise RuntimeStoreError('stale_generation') # a terminal execution's read copy
|
|
return json.loads(old['result_json'])
|
|
if row['status'] not in ('registered', 'running'):
|
|
raise RuntimeStoreError('stale_generation')
|
|
if sequence != row['last_sequence'] + 1:
|
|
raise RuntimeStoreError('invalid_params')
|
|
# Each SQLite retry gets fresh rows; rolled-back annotations must not escape.
|
|
changes = conn.total_changes
|
|
result = handlers[operation](db, conn, session_id, json.loads(encoded))
|
|
# The CAS fence moves only when the handler changed durable state (or the execution
|
|
# closes): a read-only receipt (context, history, lineage, lifecycle) must not make a
|
|
# client's mutation built on the pre-read revision fail revision_conflict.
|
|
changed = conn.total_changes != changes or operation == 'execution.finish'
|
|
conn.execute('INSERT INTO worker_receipts(execution_id,sequence,payload_digest,result_json) VALUES(?,?,?,?)',
|
|
(execution_id, sequence, digest, json.dumps(result, ensure_ascii=True, allow_nan=False)))
|
|
conn.execute("UPDATE worker_executions SET last_sequence=?,status=? WHERE execution_id=?",
|
|
(sequence, 'terminal' if operation == 'execution.finish' else 'running', execution_id))
|
|
if changed:
|
|
conn.execute('UPDATE sessions SET runtime_revision=runtime_revision+1 WHERE id=?', (session_id,))
|
|
if operation == 'execution.finish':
|
|
from hermes_state_runtime_workers import compact_terminal_receipts
|
|
compact_terminal_receipts(conn, execution_id)
|
|
return result
|
|
return db._execute_write(write, patience_s=db._TRANSCRIPT_WRITE_PATIENCE_S)
|