mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
109a8f7c03
Compaction carries a tool result forward as a shortened stub with the same message_uid and timestamp. Tool results keep their payload in the display key on purpose (the archived output must never fold into its stub), so the stub got a display slot of its own and Desktop painted the tool result twice: the full archived original and the stub (#132939, #135655). The display reads now skip a tool row when an archived, display-visible tool row of the same session, message_uid and timestamp holds an earlier slot. The rule reads identity only (no body comparison) and changes no stored row: the stub's display_metadata and the model history stay as they are. An in-place prune keeps its row's slot, so its stub still wins its slot as before. Every display reader applies it: the indexed page and its count, the timeline and messages-around reads, and the Python dedupe behind session.resume, lineage reads and the runtime worker's snapshot read. The SQL probe seeks (session_id, active, timestamp), so a page read stays per-row cheap.
478 lines
25 KiB
Python
478 lines
25 KiB
Python
"""Closed compression operations on the worker receipt's connection.
|
|
|
|
No handler opens a connection or invokes a self-committing SessionDB method.
|
|
"""
|
|
from typing import Any, Dict, List, Optional
|
|
import json
|
|
from functools import partial
|
|
import math
|
|
import time
|
|
|
|
from hermes_state_runtime import RuntimeStoreError, _text
|
|
from hermes_state_compression import _claim_lease_row
|
|
|
|
|
|
def _fields(payload, required, optional=()):
|
|
if not isinstance(payload, dict) or set(payload) - set(required) - set(optional) or set(required) - set(payload):
|
|
raise RuntimeStoreError('invalid_params')
|
|
|
|
|
|
def _number(value):
|
|
if type(value) not in (int, float) or not math.isfinite(value):
|
|
raise RuntimeStoreError('invalid_params')
|
|
return value
|
|
|
|
|
|
def worker_compression_lock(db, conn, sid, payload, *, action):
|
|
if action == 'holder':
|
|
_fields(payload, ())
|
|
row = conn.execute('SELECT holder FROM compression_locks WHERE session_id=? AND expires_at>?',
|
|
(sid, time.time())).fetchone()
|
|
return {'value': row[0] if row else None}
|
|
_fields(payload, ('holder',), () if action == 'release' else ('ttl_seconds',))
|
|
holder = _text(payload['holder'])
|
|
now = time.time()
|
|
ttl = _number(payload.get('ttl_seconds', 300.0))
|
|
if not 0.1 <= ttl <= 3600:
|
|
raise RuntimeStoreError('invalid_params')
|
|
if action == 'acquire':
|
|
from hermes_state import _compression_lock_holder_process_is_dead
|
|
value = _claim_lease_row(conn, 'compression_locks', 'session_id', sid, holder, now, now + ttl,
|
|
lambda h, e: e < now or _compression_lock_holder_process_is_dead(h))[0]
|
|
elif action == 'renew':
|
|
value = conn.execute('UPDATE compression_locks SET expires_at=? WHERE session_id=? AND holder=?',
|
|
(now + ttl, sid, holder)).rowcount > 0
|
|
else:
|
|
conn.execute('DELETE FROM compression_locks WHERE session_id=? AND holder=?', (sid, holder))
|
|
value = None
|
|
return {'value': value}
|
|
|
|
|
|
def worker_cooldown_record(db, conn, sid, payload):
|
|
_fields(payload, ('cooldown_until', 'error'))
|
|
deadline = _number(payload['cooldown_until'])
|
|
error = payload['error']
|
|
if error is not None and not isinstance(error, str):
|
|
raise RuntimeStoreError('invalid_params')
|
|
conn.execute('UPDATE sessions SET compression_failure_cooldown_until = CASE '
|
|
'WHEN compression_failure_cooldown_until IS NOT NULL AND compression_failure_cooldown_until > ? '
|
|
'THEN compression_failure_cooldown_until ELSE ? END, compression_failure_error=? WHERE id=?',
|
|
(deadline, deadline, error, sid))
|
|
return {'value': None}
|
|
|
|
|
|
def worker_cooldown_restore(db, conn, sid, payload):
|
|
_fields(payload, ('snapshot',))
|
|
snapshot = payload['snapshot']
|
|
_fields(snapshot, ('session_exists', 'cooldown_until', 'error'))
|
|
if type(snapshot['session_exists']) is not bool:
|
|
raise RuntimeStoreError('invalid_params')
|
|
if not snapshot['session_exists']:
|
|
raise RuntimeError('cannot restore absent compression cooldown row: session now exists')
|
|
deadline, error = snapshot['cooldown_until'], snapshot['error']
|
|
if deadline is not None:
|
|
_number(deadline)
|
|
if error is not None and not isinstance(error, str):
|
|
raise RuntimeStoreError('invalid_params')
|
|
conn.execute('UPDATE sessions SET compression_failure_cooldown_until=?, compression_failure_error=? WHERE id=?',
|
|
(deadline, error, sid))
|
|
actual = conn.execute('SELECT compression_failure_cooldown_until,compression_failure_error FROM sessions WHERE id=?',
|
|
(sid,)).fetchone()
|
|
if actual is None or tuple(actual) != (deadline, error):
|
|
raise RuntimeError('compression cooldown rollback verification failed')
|
|
return {'value': None}
|
|
|
|
|
|
def worker_cooldown_clear(db, conn, sid, payload):
|
|
_fields(payload, ())
|
|
return worker_cooldown_restore(db, conn, sid, {'snapshot': {
|
|
'session_exists': True, 'cooldown_until': None, 'error': None}})
|
|
|
|
|
|
def worker_compression_counter(db, conn, sid, payload, *, column):
|
|
_fields(payload, ('value',))
|
|
value = payload['value']
|
|
if value is not None:
|
|
_number(value)
|
|
if column != 'compression_recovery_deadline' and (type(value) is not int or value < 0):
|
|
raise RuntimeStoreError('invalid_params')
|
|
conn.execute(f'UPDATE sessions SET {column}=? WHERE id=?', (value, sid))
|
|
return {'value': None}
|
|
|
|
|
|
class CompressionSnapshot:
|
|
"""Reuse production read algorithms on the receipt connection, never a second handle.
|
|
|
|
Only read primitives are provided; this object cannot start/commit a write.
|
|
Wire handlers below select concrete projections, not caller-named methods.
|
|
"""
|
|
from hermes_state_compression import SessionCompressionMixin as _C
|
|
from hermes_state_messages import SessionMessagesMixin as _M
|
|
from hermes_state_sessions import SessionSessionsMixin as _S
|
|
|
|
get_compression_chain = _C.get_compression_chain
|
|
get_compression_tip = _C.get_compression_tip
|
|
get_compression_lineage = _C.get_compression_lineage
|
|
_is_compression_child_row = _C._is_compression_child_row
|
|
_session_lineage_root_to_tip = _S._session_lineage_root_to_tip
|
|
_is_explicit_branch_session = _S._is_explicit_branch_session
|
|
declared_scope_identity = _S.declared_scope_identity
|
|
_resume_lineage_ids = _M._resume_lineage_ids
|
|
get_conversation_root = _M.get_conversation_root
|
|
resolve_resume_session_id = _M.resolve_resume_session_id
|
|
latest_conversation_boundary = _M.latest_conversation_boundary
|
|
_is_explicit_fork_child_row = _M._is_explicit_fork_child_row
|
|
|
|
def __init__(self, db, conn):
|
|
self.db, self.conn = db, conn
|
|
|
|
def _read_ctx(self):
|
|
from contextlib import nullcontext
|
|
return nullcontext(self.conn)
|
|
|
|
def _read_one(self, sql, params=()):
|
|
return self.conn.execute(sql, params).fetchone()
|
|
|
|
def _read_all(self, sql, params=()):
|
|
return self.conn.execute(sql, params).fetchall()
|
|
|
|
def get_session(self, sid):
|
|
from hermes_state_worker_context import worker_context
|
|
return worker_context(self.db, self.conn, sid, {})['session']
|
|
|
|
def authorize(self, assigned, target):
|
|
_text(target)
|
|
# Attribution ancestry is readable; unrelated siblings and foreign trees are not.
|
|
allowed = self._session_lineage_root_to_tip(assigned)
|
|
allowed += self.get_compression_chain(assigned)
|
|
if target not in allowed:
|
|
raise RuntimeStoreError('permission_denied')
|
|
|
|
|
|
def worker_lineage_context(db, conn, sid, payload):
|
|
_fields(payload, ('target',))
|
|
view = CompressionSnapshot(db, conn)
|
|
target = payload['target']
|
|
view.authorize(sid, target)
|
|
return {'session': view.get_session(target)}
|
|
|
|
|
|
def worker_lineage(db, conn, sid, payload):
|
|
_fields(payload, ('target',))
|
|
view = CompressionSnapshot(db, conn)
|
|
target = payload['target']
|
|
view.authorize(sid, target)
|
|
return {'readable_ids': view._session_lineage_root_to_tip(sid) + view.get_compression_chain(sid),
|
|
'lineage': view.get_compression_lineage(target),
|
|
'root': view.get_conversation_root(target),
|
|
'tip': view.get_compression_tip(target),
|
|
'resume': view.resolve_resume_session_id(target),
|
|
'identity': view.declared_scope_identity(target)}
|
|
|
|
|
|
def worker_boundary(db, conn, sid, payload):
|
|
_fields(payload, ('session_key', 'source'))
|
|
view = CompressionSnapshot(db, conn)
|
|
row = view.get_session(sid)
|
|
if (row['session_key'], row['source']) != (payload['session_key'], payload['source']):
|
|
raise RuntimeStoreError('permission_denied')
|
|
return {'value': view.latest_conversation_boundary(payload['session_key'], payload['source'])}
|
|
|
|
|
|
def worker_history(db, conn, sid, payload):
|
|
_fields(payload, ('target', 'include_ancestors', 'include_inactive', 'repair_alternation',
|
|
'include_row_ids', 'include_compacted'))
|
|
if any(type(v) is not bool for k, v in payload.items() if k != 'target'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
view = CompressionSnapshot(db, conn)
|
|
target = payload['target']
|
|
view.authorize(sid, target)
|
|
ids = view._resume_lineage_ids(target) if payload['include_ancestors'] else [target]
|
|
active = db._active_clause(payload['include_inactive'], payload['include_compacted'])
|
|
display = payload['include_compacted'] and not payload['include_inactive']
|
|
rows = conn.execute(f'SELECT {"session_id, compacted, display_order, " if display else ""}{db._CONVERSATION_ROW_COLUMNS} FROM messages '
|
|
f'WHERE session_id IN ({",".join("?" for _ in ids)}){active} ORDER BY id', ids).fetchall()
|
|
if display: # the owner's display reads' rule: a message whose newest copy was removed stays hidden
|
|
from hermes_state_messages import SessionMessagesMixin
|
|
rows = SessionMessagesMixin._with_removed_identities(view, rows, ids)
|
|
if payload['include_compacted']:
|
|
rows = db._dedupe_display_generations(rows, drop_removed=display)
|
|
return {'messages': db._rows_to_conversation(rows, session_id=target,
|
|
include_ancestors=payload['include_ancestors'], repair_alternation=payload['repair_alternation'],
|
|
include_row_ids=payload['include_row_ids'])}
|
|
|
|
|
|
|
|
|
|
|
|
def archive_on_connection(db, conn, session_id: str, compacted_messages: list[dict[str, Any]], model_config_patch: Optional[dict[str, Any]]=None, watermark: Optional[int]=None, lock_holder: Optional[str]=None, tail_count: int=0, carried_messages: Optional[list[dict[str, Any]]]=None, covered_ids: Optional[list[int]]=None, unresolved_held: Optional[list[dict[str, Any]]]=None, held_for_cas: Optional[list[dict[str, Any]]]=None):
|
|
# The owner's in-place compaction body, on this receipt's connection: one implementation, so
|
|
# main's coverage/merge/timestamp/display-order fixes reach worker and mutation verbs too.
|
|
return db._archive_and_compact_on_conn(conn, session_id, compacted_messages,
|
|
model_config_patch=model_config_patch, watermark=watermark, lock_holder=lock_holder,
|
|
tail_count=tail_count, carried_messages=carried_messages, covered_ids=covered_ids,
|
|
unresolved_held=unresolved_held, held_for_cas=held_for_cas)
|
|
|
|
def publish_on_connection(db, conn, *, parent_session_id: str, child_session_id: str, source: str, messages: list[dict[str, Any]], model: str | None=None, model_config: dict[str, Any] | None=None, system_prompt: str | None=None, cwd: str | None=None, profile_name: str | None=None, compression_lock_holder: str | None=None, require_compression_lease: bool=True, require_lease_refresh: bool=False, lease_ttl_seconds: float=300.0, watermark: Optional[int]=None, watermark_ceiling: Optional[int]=None):
|
|
# The owner's rotation body (lease, closure, child row, timestamps, tail clone, local target,
|
|
# auto-archive lift) on this receipt's connection, so the two paths cannot drift.
|
|
db._publish_compression_child_on_conn(conn, parent_session_id=parent_session_id,
|
|
child_session_id=child_session_id, source=source, messages=messages, model=model,
|
|
model_config=model_config, system_prompt=system_prompt, cwd=cwd, profile_name=profile_name,
|
|
compression_lock_holder=compression_lock_holder, require_compression_lease=require_compression_lease,
|
|
require_lease_refresh=require_lease_refresh, lease_ttl_seconds=lease_ttl_seconds,
|
|
watermark=watermark, watermark_ceiling=watermark_ceiling)
|
|
|
|
def worker_watermark(db, conn, sid, payload):
|
|
_fields(payload, ())
|
|
return {'value': conn.execute('SELECT COALESCE(MAX(id),0) FROM messages WHERE session_id=? AND active=1',
|
|
(sid,)).fetchone()[0]}
|
|
|
|
|
|
def _handoff_messages(conn, sid, messages):
|
|
from hermes_state_runtime import HANDOFF_MESSAGE_FIELDS
|
|
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) - HANDOFF_MESSAGE_FIELDS or msg.get('role') not in ('user', 'assistant', 'system', 'tool'):
|
|
raise RuntimeStoreError('invalid_params')
|
|
if '_row_id' in msg and not conn.execute('SELECT 1 FROM messages WHERE session_id=? AND id=?',
|
|
(sid, msg['_row_id'])).fetchone():
|
|
raise RuntimeStoreError('permission_denied')
|
|
|
|
|
|
def _watermarks(payload):
|
|
for field in ('watermark', 'watermark_ceiling'):
|
|
value = payload.get(field)
|
|
if value is not None and (type(value) is not int or value < 0):
|
|
raise RuntimeStoreError('invalid_params')
|
|
|
|
|
|
def worker_archive(db, conn, sid, payload):
|
|
from hermes_state_worker_context import SIDECAR_KEYS
|
|
_fields(payload, ('messages', 'model_config_patch', 'watermark', 'lock_holder', 'tail_count'),
|
|
optional=('carried_messages', 'covered_ids', 'unresolved_held', 'held_for_cas'))
|
|
_handoff_messages(conn, sid, payload['messages'])
|
|
carried = payload.get('carried_messages') or []
|
|
if not isinstance(carried, list) or any(not isinstance(m, dict) for m in carried):
|
|
raise RuntimeStoreError('invalid_params')
|
|
covered = payload.get('covered_ids')
|
|
if covered is not None and (not isinstance(covered, list)
|
|
or any(type(i) is not int or i <= 0 for i in covered)):
|
|
raise RuntimeStoreError('invalid_params')
|
|
unresolved = payload.get('unresolved_held')
|
|
if unresolved is not None and (not isinstance(unresolved, list)
|
|
or any(not isinstance(m, dict) for m in unresolved)):
|
|
raise RuntimeStoreError('invalid_params')
|
|
# The lease-less CAS fence (#124102): dicts projected to the closed handoff fields plus the
|
|
# persistence marker the owner's compare reads; validated like the other held-row payloads.
|
|
cas = payload.get('held_for_cas')
|
|
if cas is not None and (not isinstance(cas, list)
|
|
or any(not isinstance(m, dict) for m in cas)):
|
|
raise RuntimeStoreError('invalid_params')
|
|
_watermarks(payload)
|
|
# Micro-compaction and proactive prune archive in place without a lease, like the owner path.
|
|
if payload['lock_holder'] is not None:
|
|
_text(payload['lock_holder'])
|
|
patch = payload['model_config_patch']
|
|
if patch is not None and (not isinstance(patch, dict) or set(patch) - SIDECAR_KEYS):
|
|
raise RuntimeStoreError('invalid_params')
|
|
if type(payload['tail_count']) is not int or payload['tail_count'] < 0:
|
|
raise RuntimeStoreError('invalid_params')
|
|
value = archive_on_connection(db, conn, sid, payload['messages'], model_config_patch=patch,
|
|
watermark=payload['watermark'], lock_holder=payload['lock_holder'], tail_count=payload['tail_count'],
|
|
carried_messages=carried, covered_ids=covered, unresolved_held=unresolved, held_for_cas=cas)
|
|
from hermes_state_runtime import row_annotations
|
|
return {'value': value, 'annotations': row_annotations(payload['messages'])}
|
|
|
|
|
|
def _migrate_worker_automation(db, conn, parent, child):
|
|
from hermes_cli.goals import GoalState
|
|
from hermes_cli.heartbeat import HeartbeatState
|
|
from hermes_cli.loops import LoopState
|
|
for family, state_type in (('goal', GoalState), ('heartbeat', HeartbeatState), ('loop', LoopState)):
|
|
old_key, new_key = f'{family}:{parent}', f'{family}:{child}'
|
|
row = conn.execute('SELECT value FROM state_meta WHERE key=?', (old_key,)).fetchone()
|
|
if row is None:
|
|
continue
|
|
state = state_type.from_json(row[0])
|
|
if state.status == 'cleared':
|
|
continue
|
|
prior = conn.execute('SELECT value FROM state_meta WHERE key=?', (new_key,)).fetchone()
|
|
if prior is not None and (family != 'heartbeat' or state_type.from_json(prior[0]).status != 'cleared'):
|
|
continue
|
|
db.set_meta(new_key, state.to_json(), cursor=conn)
|
|
state.status = 'cleared'
|
|
db.set_meta(old_key, state.to_json(), cursor=conn)
|
|
|
|
|
|
def worker_publish(db, conn, sid, payload):
|
|
from hermes_state_sessions import _parse_model_config
|
|
_fields(payload, ('child_session_id', 'source', 'messages', 'model', 'model_config', 'system_prompt',
|
|
'cwd', 'profile_name', 'compression_lock_holder', 'require_compression_lease',
|
|
'require_lease_refresh', 'lease_ttl_seconds', 'watermark', 'watermark_ceiling'))
|
|
child = _text(payload['child_session_id'])
|
|
_text(payload['compression_lock_holder'])
|
|
_watermarks(payload)
|
|
if payload['require_compression_lease'] is not True or type(payload['require_lease_refresh']) is not bool:
|
|
raise RuntimeStoreError('invalid_params')
|
|
if not 0.1 <= _number(payload['lease_ttl_seconds']) <= 3600:
|
|
raise RuntimeStoreError('invalid_params')
|
|
_handoff_messages(conn, sid, payload['messages'])
|
|
parent = CompressionSnapshot(db, conn).get_session(sid)
|
|
for key in ('source', 'cwd', 'profile_name'):
|
|
if payload[key] is not None and payload[key] != parent[key]:
|
|
raise RuntimeStoreError('permission_denied')
|
|
for key in ('model', 'system_prompt'):
|
|
if payload[key] is not None and not isinstance(payload[key], str):
|
|
raise RuntimeStoreError('invalid_params')
|
|
cfg = payload['model_config']
|
|
if cfg is not None and not isinstance(cfg, dict):
|
|
raise RuntimeStoreError('invalid_params')
|
|
cfg = dict(cfg or {})
|
|
prior = _parse_model_config(parent['model_config'])
|
|
for key in ('_branched_from', '_delegate_from', '_reset_from'):
|
|
if key in cfg and cfg[key] != prior.get(key):
|
|
raise RuntimeStoreError('permission_denied')
|
|
if key in prior:
|
|
cfg[key] = prior[key]
|
|
if conn.execute('SELECT 1 FROM sessions WHERE id=?', (child,)).fetchone():
|
|
raise RuntimeStoreError('permission_denied')
|
|
worker = conn.execute("SELECT * FROM worker_executions WHERE session_id=? AND status IN ('registered','running')", (sid,)).fetchone()
|
|
if worker is None:
|
|
raise RuntimeStoreError('stale_generation')
|
|
publish_on_connection(db, conn, parent_session_id=sid, **dict(payload, model_config=cfg))
|
|
_migrate_worker_automation(db, conn, sid, child)
|
|
conn.execute('UPDATE sessions SET runtime_generation=? WHERE id=?', (worker['generation'], child))
|
|
conn.execute('UPDATE worker_executions SET session_id=? WHERE execution_id=?', (child, worker['execution_id']))
|
|
# The FIFO/retry key is immutable; only physical lineage follows rotation.
|
|
conn.execute("UPDATE session_admissions SET lineage_json=json_insert(lineage_json,'$[#]',?) "
|
|
"WHERE json_extract(lineage_json,'$[#-1]')=? AND generation=? AND owner_epoch=? "
|
|
"AND status IN ('started','unknown')",
|
|
(child, sid, worker['generation'], worker['owner_epoch']))
|
|
from hermes_state_runtime import row_annotations
|
|
return {'value': None, 'worker_assignment': {'parent': sid, 'session_id': child},
|
|
'annotations': row_annotations(payload['messages'])}
|
|
|
|
|
|
def worker_receipt_assignment(conn, execution_id, session_id, generation, sequence, digest):
|
|
"""Only an identical already-committed publication can retry its old assignment."""
|
|
from hermes_state_runtime import _worker_assignment
|
|
receipt = conn.execute('SELECT payload_digest,result_json FROM worker_receipts WHERE execution_id=? AND sequence=?',
|
|
(execution_id, sequence)).fetchone()
|
|
if receipt and receipt['payload_digest'] == digest:
|
|
assignment = json.loads(receipt['result_json']).get('worker_assignment')
|
|
if assignment and assignment['parent'] == session_id:
|
|
current = conn.execute('SELECT session_id FROM worker_executions WHERE execution_id=?', (execution_id,)).fetchone()
|
|
return _worker_assignment(conn, execution_id, current['session_id'], generation)
|
|
return _worker_assignment(conn, execution_id, session_id, generation)
|
|
|
|
|
|
def worker_claim_origin(db, params):
|
|
"""Keep the producer claim bound to its original identity after physical rotation."""
|
|
with db._read_ctx() as conn:
|
|
row = conn.execute("SELECT result_json FROM worker_receipts WHERE execution_id=? "
|
|
"AND json_extract(result_json,'$.worker_assignment.parent') IS NOT NULL ORDER BY sequence LIMIT 1",
|
|
(params['execution_id'],)).fetchone()
|
|
return json.loads(row[0])['worker_assignment']['parent'] if row else params['session_id']
|
|
|
|
|
|
def worker_retry_target(db, params):
|
|
"""Transport verification accepts old targets only for exact publication replay."""
|
|
from hermes_state_keys import admission_fingerprint
|
|
if params.get('operation') != 'compression.publish':
|
|
return params['session_id']
|
|
digest = admission_fingerprint(canonical_target=params['session_id'],
|
|
payload={'operation': params['operation'], 'payload': params['payload']})
|
|
with db._read_ctx() as conn:
|
|
row = worker_receipt_assignment(conn, params['execution_id'], params['session_id'],
|
|
params['generation'], params['sequence'], digest)
|
|
return row['session_id']
|
|
|
|
|
|
from hermes_state_common import _COMPRESSION_LOCK_ROW_SQL as _LOCK_ROW_SQL, _ENDED_ROW_SQL, _ended_by_compression
|
|
|
|
def reopen_on_connection(db, conn, session_id):
|
|
if not _ended_by_compression(conn.execute(_ENDED_ROW_SQL, (session_id,)).fetchone()):
|
|
return False
|
|
child = conn.execute('\n SELECT 1\n FROM sessions\n WHERE parent_session_id = ?\n ' + db._NON_CONTINUATION_CHILD_FILTER_SQL.format(alias='') + '\n LIMIT 1\n ', (session_id,) * 4).fetchone()
|
|
if child is not None:
|
|
return False
|
|
now = time.time()
|
|
lock_row = conn.execute(_LOCK_ROW_SQL, (session_id,)).fetchone()
|
|
if lock_row is not None:
|
|
expires_at = lock_row['expires_at']
|
|
if expires_at is None or float(expires_at) >= now:
|
|
return False
|
|
deleted = conn.execute('DELETE FROM compression_locks WHERE session_id = ? AND holder = ? AND expires_at = ?', (session_id, lock_row['holder'], expires_at))
|
|
if deleted.rowcount != 1:
|
|
return False
|
|
updated = conn.execute("UPDATE sessions SET ended_at = NULL, end_reason = NULL WHERE id = ? AND ended_at IS NOT NULL AND end_reason = 'compression'", (session_id,))
|
|
return updated.rowcount == 1
|
|
|
|
|
|
def worker_reopen(db, conn, sid, payload):
|
|
_fields(payload, ())
|
|
return {'value': reopen_on_connection(db, conn, sid)}
|
|
|
|
|
|
def worker_compression_cleanup(db, conn, sid, payload):
|
|
_fields(payload, ('target', 'holder'))
|
|
target = _text(payload['target'])
|
|
holder = _text(payload['holder'])
|
|
view = CompressionSnapshot(db, conn)
|
|
view.authorize(sid, target)
|
|
if db._session_turn_lease_key_on_conn(conn, sid) != db._session_turn_lease_key_on_conn(conn, target):
|
|
raise RuntimeStoreError('permission_denied')
|
|
conn.execute('DELETE FROM compression_locks WHERE session_id=? AND holder=?', (target, holder))
|
|
return {'value': None}
|
|
|
|
|
|
def worker_compression_append(db, conn, sid, payload):
|
|
_fields(payload, ('messages', 'compression_lock_holder', 'turn_lease_holder', 'turn_lease_ttl_seconds'))
|
|
_handoff_messages(conn, sid, payload['messages'])
|
|
for key in ('compression_lock_holder', 'turn_lease_holder'):
|
|
if payload[key] is not None:
|
|
_text(payload[key])
|
|
if not 0.1 <= _number(payload['turn_lease_ttl_seconds']) <= 3600:
|
|
raise RuntimeStoreError('invalid_params')
|
|
count = db._append_messages_in_transaction(conn, sid, **payload)
|
|
from hermes_state_runtime import row_annotations
|
|
return {'count': count, 'annotations': row_annotations(payload['messages'])}
|
|
|
|
|
|
def worker_turn_cleanup(db, conn, sid, payload):
|
|
_fields(payload, ('target', 'holder'))
|
|
target = _text(payload['target'])
|
|
holder = _text(payload['holder'])
|
|
CompressionSnapshot(db, conn).authorize(sid, target)
|
|
key = db._session_turn_lease_key_on_conn(conn, sid)
|
|
if key != db._session_turn_lease_key_on_conn(conn, target):
|
|
raise RuntimeStoreError('permission_denied')
|
|
conn.execute('DELETE FROM session_turn_leases WHERE conversation_id=? AND holder=?', (key, holder))
|
|
return {'released': True}
|
|
|
|
|
|
WORKER_COMPRESSION_HANDLERS = {
|
|
'turn.cleanup': worker_turn_cleanup,
|
|
'compression.append': worker_compression_append,
|
|
'compression.reopen': worker_reopen,
|
|
'compression.cleanup': worker_compression_cleanup,
|
|
'compression.watermark': worker_watermark,
|
|
'compression.archive': worker_archive,
|
|
'compression.publish': worker_publish,
|
|
'compression.context': worker_lineage_context,
|
|
'compression.lineage': worker_lineage,
|
|
'compression.boundary': worker_boundary,
|
|
'compression.history': worker_history,
|
|
**{'compression.lock.' + action: partial(worker_compression_lock, action=action)
|
|
for action in ('acquire', 'renew', 'release', 'holder')},
|
|
'compression.cooldown.record': worker_cooldown_record,
|
|
'compression.cooldown.restore': worker_cooldown_restore,
|
|
'compression.cooldown.clear': worker_cooldown_clear,
|
|
**{'compression.' + name: partial(worker_compression_counter, column=column)
|
|
for name, column in (('fallback', 'compression_fallback_streak'),
|
|
('ineffective', 'compression_ineffective_count'),
|
|
('recovery', 'compression_recovery_deadline'))},
|
|
}
|