diff --git a/hermes_state_mutation_reset.py b/hermes_state_mutation_reset.py index 6698715004..8ba3ed1e32 100644 --- a/hermes_state_mutation_reset.py +++ b/hermes_state_mutation_reset.py @@ -16,6 +16,8 @@ def reset_in_transaction(db, conn, session_id, payload): raise RuntimeStoreError('runtime_coordination_required') receipt = json.loads(saved[0]) target = validate_local_lineage(conn, receipt) + from hermes_state_runtime_workers import discard_orphan_workers_on_reset, runtime_lineage + discard_orphan_workers_on_reset(conn, runtime_lineage(conn, session_id)) require_target_advanceable(db, conn, list({session_id, target})) parent = conn.execute('SELECT * FROM sessions WHERE id=?', (target,)).fetchone() policy = receipt['policy'] diff --git a/hermes_state_runtime.py b/hermes_state_runtime.py index a02192d019..79a687d9f4 100644 --- a/hermes_state_runtime.py +++ b/hermes_state_runtime.py @@ -145,13 +145,8 @@ def claim_session_input(db, *, epoch: int, session_id: str) -> dict | None: 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') - # A worker follows its physical transcript (compression publish / local reset target) - # while this FIFO stays keyed on the logical owner: scan the whole lineage. - from hermes_state_local_lineage import local_physical_target - lineage = {*_canonical_chain(conn, session_id), *_canonical_chain(conn, local_physical_target(conn, session_id))} - marks = ','.join('?' * len(lineage)) - workers = {row[0] for row in conn.execute( - f"SELECT status FROM worker_executions WHERE session_id IN ({marks}) AND status!='terminal'", tuple(lineage))} + 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: @@ -450,11 +445,13 @@ def register_worker_execution(db, *, epoch: int, execution_id: str, session_id: 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) - if require_idle and conn.execute("SELECT 1 FROM session_admissions WHERE target_session_id=? AND status!='terminal'", (session_id,)).fetchone(): + 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!='terminal'", (sid,)).fetchone() for sid in targets): raise RuntimeStoreError('stale_generation') - if conn.execute("SELECT 1 FROM worker_executions WHERE session_id=? AND status!='terminal'", (session_id,)).fetchone(): + if worker_states(conn, session_id): raise RuntimeStoreError('stale_generation') - if conn.execute("SELECT 1 FROM session_admissions WHERE target_session_id=? AND status='unknown'", (session_id,)).fetchone(): + 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)) diff --git a/hermes_state_runtime_workers.py b/hermes_state_runtime_workers.py new file mode 100644 index 0000000000..c4dfde484a --- /dev/null +++ b/hermes_state_runtime_workers.py @@ -0,0 +1,39 @@ +"""Worker exclusion follows the logical conversation and its physical transcript lineage.""" +import json + + +def runtime_lineage(conn, session_id): + from hermes_state_local import POLICY_PREFIX + from hermes_state_local_lineage import local_physical_target + from hermes_state_sessions import _expand_compression_lineage_ids + from hermes_state_compression import _CHAIN_CAP + from hermes_state_runtime import RuntimeStoreError + targets = {session_id, local_physical_target(conn, session_id)} + row = conn.execute('SELECT chat_id FROM sessions WHERE id=?', (session_id,)).fetchone() + if row and row['chat_id']: + saved = conn.execute('SELECT value FROM state_meta WHERE key=?', (POLICY_PREFIX + row['chat_id'],)).fetchone() + if saved is not None and session_id in json.loads(saved[0]).get('lineage', [row['chat_id']]): + targets.update({row['chat_id'], local_physical_target(conn, row['chat_id'])}) + frontier = list(targets) + for _ in range(_CHAIN_CAP): + frontier = [sid for sid in _expand_compression_lineage_ids(conn, frontier) if sid not in targets] + if not frontier: + return targets + targets.update(frontier) + raise RuntimeStoreError('storage_unavailable') + + +def worker_states(conn, session_id): + return {row[0] for sid in runtime_lineage(conn, session_id) for row in conn.execute( + "SELECT status FROM worker_executions WHERE session_id=? AND status!='terminal'", (sid,))} + + +def discard_orphan_workers_on_reset(conn, session_ids): + """An explicit reset revokes unlinked unknown compute; it never replays its side effects. + + Admission-linked work still requires its own Discard receipt. A running/registered worker + remains a hard reset refusal, and transaction rollback preserves all fences on refusal. + """ + for sid in session_ids: + conn.execute("UPDATE worker_executions SET status='terminal' WHERE session_id=? AND status='unknown' " + "AND execution_id NOT LIKE 'admission-worker:%'", (sid,)) diff --git a/tests/gateway/test_unknown_resolution_physical_worker.py b/tests/gateway/test_unknown_resolution_physical_worker.py index ddd98ba9d4..6d7f5eac9e 100644 --- a/tests/gateway/test_unknown_resolution_physical_worker.py +++ b/tests/gateway/test_unknown_resolution_physical_worker.py @@ -1,6 +1,7 @@ """Resolving an unknown admission must also retire its physical worker assignment.""" from contextlib import closing import os +from pathlib import Path import subprocess import sys from types import SimpleNamespace @@ -32,7 +33,7 @@ def _local_reset(db, epoch, logical, child): source = SessionSource(platform=Platform.LOCAL, chat_id=sid, user_id='human', chat_type='dm') now = _now() entry = SessionEntry('local:' + sid, sid, now, now, origin=source, platform=Platform.LOCAL) - policy = build_policy({'source': 'cli', 'cwd': '/', 'model': 'm', 'toolsets': []}, + policy = build_policy({'source': 'cli', 'cwd': str(Path(db.db_path).parent), 'model': 'm', 'toolsets': []}, {'platform_toolsets': {'cli': []}}, private_secrets={}) commit_local_session(db, epoch=epoch, receipt={ 'profile_id': 'profile', 'principal_id': 'human', 'request_id': 'r', 'session_id': sid, diff --git a/tests/hermes_state/test_worker_lineage_exclusion.py b/tests/hermes_state/test_worker_lineage_exclusion.py new file mode 100644 index 0000000000..72be097499 --- /dev/null +++ b/tests/hermes_state/test_worker_lineage_exclusion.py @@ -0,0 +1,46 @@ +"""A physical worker excludes its whole FIFO lineage; an explicit reset revokes orphan unknowns.""" +import pytest + +from hermes_state import SessionDB +import hermes_state_runtime as rt + + +def test_physical_worker_excludes_logical_fifo_and_second_registration(tmp_path): + from tests.gateway.test_unknown_resolution_physical_worker import _local_reset + with SessionDB(tmp_path / 'state.db') as db: + epoch = rt.begin_runtime_epoch(db, instance_id='owner') + logical = _local_reset(db, epoch, 'logical', 'physical') + rt.register_worker_execution(db, epoch=epoch, execution_id='standalone-worker', session_id='physical', + generation=0, kind='compute', adoption_secret='secret') + rt.admit_session_input(db, epoch=epoch, principal_id='human', session_id=logical, + request_id='later', payload={'text': 'later'}) + assert rt.claim_session_input(db, epoch=epoch, session_id=logical) is None + with pytest.raises(rt.RuntimeStoreError, match='stale_generation'): + rt.register_worker_execution(db, epoch=epoch, execution_id='second-worker', session_id=logical, + generation=db.get_session(logical)['runtime_generation'], kind='compute', adoption_secret='secret') + + +def test_reset_revokes_only_orphan_unknown_worker_and_unblocks_fifo(tmp_path): + from tests.gateway.test_unknown_resolution_physical_worker import _local_reset + with SessionDB(tmp_path / 'state.db') as db: + epoch = rt.begin_runtime_epoch(db, instance_id='owner') + logical = _local_reset(db, epoch, 'logical', 'physical') + generation = db.get_session(logical)['runtime_generation'] + rt.register_worker_execution(db, epoch=epoch, execution_id='standalone-worker', session_id=logical, + generation=generation, kind='compute', adoption_secret='secret') + epoch = rt.begin_runtime_epoch(db, instance_id='restarted') + rt.recover_session_inputs(db, epoch=epoch) + rt.admit_session_input(db, epoch=epoch, principal_id='human', session_id=logical, + request_id='later', payload={'text': 'later'}) + with pytest.raises(rt.RuntimeStoreError, match='unknown_execution'): + rt.claim_session_input(db, epoch=epoch, session_id=logical) + receipt = rt.mutate_runtime_session(db, epoch=epoch, principal_id='human', session_id=logical, + request_id='explicit-reset', expected_revision=db.get_session(logical)['runtime_revision'], + expected_generation=db.get_session(logical)['runtime_generation'], + operation='reset', payload={}) + assert receipt['target_session_id'] != 'physical' + assert db._read_one('SELECT status FROM worker_executions')[0] == 'terminal' + assert rt.claim_session_input(db, epoch=epoch, session_id=logical)['request_id'] == 'later' + with pytest.raises(rt.RuntimeStoreError, match='stale_generation'): + rt.adopt_worker_execution(db, epoch=epoch, execution_id='standalone-worker', + session_id=logical, generation=generation, adoption_secret='secret')