fix(state): exclude workers across the complete conversation lineage

A worker on a physical transcript fences its logical FIFO and second registrations. Explicit reset revokes only unlinked unknown workers, preserving live and admission-linked fences.

Reported-by: itsflownium (https://github.com/NousResearch/hermes-agent/pull/106742#issuecomment-6050777316)

Validation: tests/hermes_state/test_worker_lineage_exclusion.py, tests/gateway/test_unknown_resolution_physical_worker.py

Implementation and additional regression probes were AI-assisted with GPT-6 Astra.

(cherry picked from commit b31ed5232056100c55568d67a0c8ab6426886d4f)

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']

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))

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,))

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,

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')