"""Ranked labels and exact provider-side data share assignment receipts.""" import time import pytest from hermes_state_runtime import RuntimeStoreError, mutate_worker_execution from tests.hermes_state.test_runtime_worker_lifecycle import setup_store def test_worker_labels_keep_rank_and_reject_late_generations(tmp_path): db, store, scope = setup_store(tmp_path) try: assert store.set_auto_title('owned', 'derived', source='derived') assert store.set_auto_title('owned', 'generated', source='llm') assert not store.set_auto_title('owned', 'late-derived', source='derived') assert store.set_session_title('owned', 'manual') assert not store.set_auto_title('owned', 'late-llm', source='llm') assert store.get_session_title_source('owned') == 'user' assert store.get_next_title_in_lineage('manual') == db.get_next_title_in_lineage('manual') assert store.set_session_title_source('owned', 'llm') now = time.time() + 10 store.touch_session_activity('owned', now, description=' active work ', provenance='unknown') expected = db.get_session('owned') store.touch_session_activity('owned', now - 1, description='older') assert db.get_session('owned')['last_activity_description'] == expected['last_activity_description'] store.clear_session_activity_labels('owned') assert db.get_session('owned')['last_activity_at'] == now assert db.get_session('owned')['last_activity_description'] == '' store.touch_session_activity('owned', now + 1, description='successor') db._execute_write(lambda conn: conn.execute( 'UPDATE sessions SET runtime_generation=runtime_generation+1 WHERE id=?', ('owned',))) for operation, payload in [('session.title', {'title': 'late', 'source': 'user'}), ('session.title_source', {'source': 'derived'}), ('session.activity_clear', {}), ('session.activity', {'ts': now + 2, 'description': 'late', 'provenance': None})]: with pytest.raises(RuntimeStoreError, match='stale_generation'): mutate_worker_execution(db, **scope, sequence=store.journal['next_sequence'], operation=operation, payload=payload) assert db.get_session('owned')['title'] == 'manual' assert db.get_session('owned')['last_activity_description'] == 'successor' finally: store.close() db.close() def test_worker_route_and_api_content_preserve_exact_usage_and_target_fences(tmp_path): db, store, scope = setup_store(tmp_path) try: store.update_system_prompt('owned', 'stable-prefix') store.queue_token_counts('owned', input_tokens=11, output_tokens=7, model='first', billing_provider='provider-one', billing_base_url='https://first.invalid', api_call_count=1) store.update_session_billing_route('owned', provider='provider-two', base_url='https://second.invalid', billing_mode='api') store.queue_token_counts('owned', input_tokens=13, output_tokens=3, model='second', billing_provider='provider-two', billing_base_url='https://second.invalid', api_call_count=1) row = db.get_session('owned') assert (row['input_tokens'], row['output_tokens']) == (24, 10) assert row['billing_provider'] == 'provider-two' # Direct-writer parity: a route write never touches the frozen prompt snapshot (W6). assert row['system_prompt'] == 'stable-prefix' with db._read_ctx() as conn: assert conn.execute('SELECT COUNT(*) FROM system_prompts WHERE prompt=?', ('stable-prefix',)).fetchone()[0] == 1 store.append_messages_batch('owned', [{'role': 'user', 'content': [{'type': 'text', 'text': 'visible'}]}]) assert store.set_latest_user_api_content('owned', 'wrong', 'must-not-land') == 0 assert store.set_latest_user_api_content('owned', [{'type': 'text', 'text': 'visible'}], 'wire-only') == 1 with db._read_ctx() as conn: assert conn.execute('SELECT api_content FROM messages WHERE session_id=?', ('owned',)).fetchone()[0] == 'wire-only' usage = [tuple(r) for r in conn.execute( 'SELECT model,billing_provider,input_tokens,output_tokens FROM session_model_usage WHERE session_id=? ORDER BY model', ('owned',))] assert usage == [('first', 'provider-one', 11, 7), ('second', 'provider-two', 13, 3)] original = store.rpc def lost_ack(method, **params): original(method, **params) raise TimeoutError('lost-ack') store.rpc = lost_ack with pytest.raises(TimeoutError, match='lost-ack'): store.update_session_billing_route('owned', provider='third', base_url='https://third.invalid') store.rpc = original store.retry_pending() assert db.get_session('owned')['billing_provider'] == 'third' for operation, payload in [('session.billing_route', {'provider': 'evil', 'base_url': 'evil', 'billing_mode': None}), ('session.api_content', {'content': 'private', 'api_content': 'evil'}), ('session.title', {'title': 'evil', 'source': 'user'}), ('session.activity_clear', {})]: with pytest.raises(RuntimeStoreError, match='permission_denied'): mutate_worker_execution(db, **dict(scope, session_id='foreign'), sequence=store.journal['next_sequence'], operation=operation, payload=payload) assert db.get_session('foreign')['system_prompt'] == 'private' finally: store.close() db.close() def test_turn_prologue_row_backfills_land_through_the_worker_like_direct(worker): """The real prologue (agent/turn_context.py) updates an already-materialized user row by ``_row_id``: the api_content sidecar after in-place preflight compaction, and the multimodal context part. The managed facade must implement both row-addressed ops, fenced to its own session's active user rows, or the durable row silently diverges from what the model saw (helix4u #4).""" import threading import types from agent.turn_context import _append_multimodal_context, _stamp_api_content_sidecar db, store = worker db.create_session('direct', 'cli') def newest_user(sid): with db._read_ctx() as conn: return tuple(conn.execute("SELECT content, api_content FROM messages WHERE session_id=? AND role='user' " "ORDER BY id DESC LIMIT 1", (sid,)).fetchone()) durable = {} for sid, sdb in (('direct', db), ('owned', store)): agent = types.SimpleNamespace(_session_db=sdb, session_id=sid, _session_persist_lock=threading.Lock(), _persist_user_message_override=None, _last_compaction_in_place=True) text = [{'role': 'user', 'content': 'question'}] sdb.archive_and_compact(sid, text, tail_count=0) _stamp_api_content_sidecar(agent, text, 0, '', 'PLUGIN CONTEXT', preflight_compressed=True) sidecar = newest_user(sid) sdb.append_messages_batch(sid, [{'role': 'assistant', 'content': 'ok'}]) parts = [{'role': 'user', 'content': [{'type': 'text', 'text': 'look'}, {'type': 'image_url', 'image_url': {'url': 'data:image/png;base64,AA=='}}]}] sdb.append_messages_batch(sid, parts) _append_multimodal_context(agent, parts[0], '', 'PLUGIN CONTEXT', preflight_compressed=False) durable[sid] = (sidecar, newest_user(sid)[0]) assert durable['owned'] == durable['direct'] == (('question', 'question\n\nPLUGIN CONTEXT'), 'look\n[screenshot]\nPLUGIN CONTEXT') foreign = db.append_message('foreign', 'user', 'private') assert store.set_user_message_content('owned', foreign, 'overwritten') == 0 assert store.set_message_api_content('owned', foreign, 'private', 'injected') == 0 assert newest_user('foreign') == ('private', None)