diff --git a/gateway/hosted_room_work_records.py b/gateway/hosted_room_work_records.py index be9818d345..dd3346482f 100644 --- a/gateway/hosted_room_work_records.py +++ b/gateway/hosted_room_work_records.py @@ -79,6 +79,67 @@ def _sha(value): raise WorkRecordError("work record digest is invalid") +def _validate_tasks(record, tasks): + task_ids = set() + for task in tasks: + _fields(task, _TASK_FIELDS) + for name in ("task_id", "thread_id", "turn_id", "member_id", "profile"): + _id(task[name]) + for name in ("settlement_id", "cancel_id"): + if task[name] is not None: + _id(task[name]) + for name in ("execution_generation", "cancel_generation"): + _integer(task[name]) + _integer(task["source_event_seq"], minimum=1) + _sha(task["payload_sha256"]) + if task["source_event_seq"] > record["history"]["seq"] or task["phase"] not in _PHASES or task["task_id"] in task_ids: + raise WorkRecordError("work record task conflicts") + task_ids.add(task["task_id"]) + + +def _validate_receipts(record, receipts, v2): + receipt_ids = set() + for receipt in receipts: + _fields(receipt, _RECEIPT_FIELDS) + for name in _RECEIPT_FIELDS - {"authority_epoch", "execution_generation"}: + _id(receipt[name]) + _integer(receipt["execution_generation"], minimum=1) + _integer(receipt["authority_epoch"], minimum=1) + if (receipt["room_id"] != record["room_id"] + or receipt["authority_gateway_id"] != receipt["home_install_id"] + or (not v2 and (receipt["home_install_id"] != record["home_install_id"] or receipt["authority_epoch"] != 1))): + raise WorkRecordError("work record receipt lineage conflicts") + key = tuple(receipt[k] for k in sorted(_RECEIPT_FIELDS - {"run_id", "session_id"})) + if key in receipt_ids: + raise WorkRecordError("work record receipt is duplicated") + receipt_ids.add(key) + + +def _validate_stop(record): + _fields(record["stop"], {"closing", "revocation_complete", "seq", "cancel_id"}) + for key in ("closing", "revocation_complete"): + if type(record["stop"][key]) is not bool: + raise WorkRecordError("work record closing fact is invalid") + _integer(record["stop"]["seq"]) + if record["stop"]["seq"] > record["history"]["seq"]: + raise WorkRecordError("work record stop is beyond its prefix") + if record["stop"]["cancel_id"] is not None: + _id(record["stop"]["cancel_id"]) + + +def _validate_availability(record): + if record["availability"] not in {"available", "unavailable"} or record["reason"] not in { + None, "task_store_missing", "unsupported_task", "bounds_exceeded"}: + raise WorkRecordError("work record availability is invalid") + if (record["availability"] == "available") != (record["reason"] is None): + raise WorkRecordError("work record availability conflicts") + tasks, receipts = record["tasks"], record["receipts"] + if not isinstance(tasks, list) or len(tasks) > MAX_TASKS or not isinstance(receipts, list) or len(receipts) > MAX_RECEIPTS: + raise WorkRecordError("work record list exceeds its bound") + if record["availability"] == "unavailable" and (tasks or receipts): + raise WorkRecordError("unavailable work record must not appear complete") + + def validate(record: dict) -> dict: v2 = isinstance(record, dict) and type(record.get("version")) is int and record["version"] == 2 _fields(record, _FIELDS | {"lineage_sha256", "incompleteness"} if v2 else _FIELDS) @@ -101,54 +162,10 @@ def validate(record: dict) -> dict: _sha(record["history"]["event_sha256"]) if record["limitations"] != LIMITATIONS: raise WorkRecordError("work record limitations are required") - if record["availability"] not in {"available", "unavailable"} or record["reason"] not in { - None, "task_store_missing", "unsupported_task", "bounds_exceeded"}: - raise WorkRecordError("work record availability is invalid") - if (record["availability"] == "available") != (record["reason"] is None): - raise WorkRecordError("work record availability conflicts") - tasks, receipts = record["tasks"], record["receipts"] - if not isinstance(tasks, list) or len(tasks) > MAX_TASKS or not isinstance(receipts, list) or len(receipts) > MAX_RECEIPTS: - raise WorkRecordError("work record list exceeds its bound") - if record["availability"] == "unavailable" and (tasks or receipts): - raise WorkRecordError("unavailable work record must not appear complete") - task_ids, receipt_ids = set(), set() - for task in tasks: - _fields(task, _TASK_FIELDS) - for name in ("task_id", "thread_id", "turn_id", "member_id", "profile"): - _id(task[name]) - for name in ("settlement_id", "cancel_id"): - if task[name] is not None: - _id(task[name]) - for name in ("execution_generation", "cancel_generation"): - _integer(task[name]) - _integer(task["source_event_seq"], minimum=1) - _sha(task["payload_sha256"]) - if task["source_event_seq"] > record["history"]["seq"] or task["phase"] not in _PHASES or task["task_id"] in task_ids: - raise WorkRecordError("work record task conflicts") - task_ids.add(task["task_id"]) - for receipt in receipts: - _fields(receipt, _RECEIPT_FIELDS) - for name in _RECEIPT_FIELDS - {"authority_epoch", "execution_generation"}: - _id(receipt[name]) - _integer(receipt["execution_generation"], minimum=1) - _integer(receipt["authority_epoch"], minimum=1) - if (receipt["room_id"] != record["room_id"] - or receipt["authority_gateway_id"] != receipt["home_install_id"] - or (not v2 and (receipt["home_install_id"] != record["home_install_id"] or receipt["authority_epoch"] != 1))): - raise WorkRecordError("work record receipt lineage conflicts") - key = tuple(receipt[k] for k in sorted(_RECEIPT_FIELDS - {"run_id", "session_id"})) - if key in receipt_ids: - raise WorkRecordError("work record receipt is duplicated") - receipt_ids.add(key) - _fields(record["stop"], {"closing", "revocation_complete", "seq", "cancel_id"}) - for key in ("closing", "revocation_complete"): - if type(record["stop"][key]) is not bool: - raise WorkRecordError("work record closing fact is invalid") - _integer(record["stop"]["seq"]) - if record["stop"]["seq"] > record["history"]["seq"]: - raise WorkRecordError("work record stop is beyond its prefix") - if record["stop"]["cancel_id"] is not None: - _id(record["stop"]["cancel_id"]) + _validate_availability(record) + _validate_tasks(record, record["tasks"]) + _validate_receipts(record, record["receipts"], v2) + _validate_stop(record) _sha(record["digest"]) if record["digest"] != digest({k: v for k, v in record.items() if k not in {"revision", "digest"}}): raise WorkRecordError("work record content digest conflicts") diff --git a/gateway/session_mutations.py b/gateway/session_mutations.py index ab3327d56a..a384053246 100644 --- a/gateway/session_mutations.py +++ b/gateway/session_mutations.py @@ -12,6 +12,42 @@ _METADATA = frozenset({'rename', 'archive', 'sidebar'}) _FIELDS = frozenset({'session_id', 'request_id', 'expected_revision', 'operation', 'payload'}) +def _authorize_import(authority, actor, ref, params): + """Import authorization; returns the imported session ids the write must claim.""" + if actor.profile_id != authority.profile_id or ref.profile_id != authority.profile_id: + raise RuntimeStoreError('profile_mismatch') + if 'session:create' not in actor.capabilities: + raise RuntimeStoreError('permission_denied') + from hermes_state_mutations import validate_action + validate_action('import', params['payload']) + normalized, errors = authority.db._validate_import_payload(params['payload']['sessions']) + if errors: + raise RuntimeStoreError('invalid_params') + return tuple(item['session']['id'] for item in normalized) + + +def _authorize_control(authority, actor, ref, params): + """Control authorization; True when the session is cold history the write must claim.""" + from hermes_state_mutation_retirement import has_mutation_receipt + if actor.profile_id != authority.profile_id or ref.profile_id != authority.profile_id: + raise RuntimeStoreError('profile_mismatch') + if 'session:control' not in actor.capabilities: + raise RuntimeStoreError('permission_denied') + cold_history = False + # A receipt authorizes only its original principal's exact retry; the + # transaction still verifies the entire digest and current epoch. + if not has_mutation_receipt(authority.db, actor.subject, ref.session_id, params['request_id']): + if ref.session_id not in authority.sessions: + from hermes_state_mutation_binding import authorize_history + with authority.db._read_ctx() as conn: + from gateway.session_local_migration import require_history_claim + require_history_claim(authority, conn, actor, ref.session_id) + cold_history = authorize_history(conn, actor, ref.session_id) + if not cold_history: + authority.authorize(actor, ref, 'session:control') + return cold_history + + async def mutate_session(authority, actor, ref, params): if (set(params) - {'expected_generation'} != _FIELDS or params['session_id'] != ref.session_id): @@ -20,34 +56,11 @@ async def mutate_session(authority, actor, ref, params): if not isinstance(operation, str): raise RuntimeStoreError('invalid_params') cold_history = False + imported_ids = None if operation == 'import': - if actor.profile_id != authority.profile_id or ref.profile_id != authority.profile_id: - raise RuntimeStoreError('profile_mismatch') - if 'session:create' not in actor.capabilities: - raise RuntimeStoreError('permission_denied') - from hermes_state_mutations import validate_action - validate_action(operation, params['payload']) - normalized, errors = authority.db._validate_import_payload(params['payload']['sessions']) - if errors: - raise RuntimeStoreError('invalid_params') - imported_ids = tuple(item['session']['id'] for item in normalized) + imported_ids = _authorize_import(authority, actor, ref, params) else: - from hermes_state_mutation_retirement import has_mutation_receipt - if actor.profile_id != authority.profile_id or ref.profile_id != authority.profile_id: - raise RuntimeStoreError('profile_mismatch') - if 'session:control' not in actor.capabilities: - raise RuntimeStoreError('permission_denied') - # A receipt authorizes only its original principal's exact retry; the - # transaction still verifies the entire digest and current epoch. - if not has_mutation_receipt(authority.db, actor.subject, ref.session_id, params['request_id']): - if ref.session_id not in authority.sessions: - from hermes_state_mutation_binding import authorize_history - with authority.db._read_ctx() as conn: - from gateway.session_local_migration import require_history_claim - require_history_claim(authority, conn, actor, ref.session_id) - cold_history = authorize_history(conn, actor, ref.session_id) - if not cold_history: - authority.authorize(actor, ref, 'session:control') + cold_history = _authorize_control(authority, actor, ref, params) if operation == 'branch' and 'session:create' not in actor.capabilities: raise RuntimeStoreError('permission_denied') authority._require_admission_open() @@ -109,6 +122,17 @@ async def mutate_session(authority, actor, ref, params): _authorize_write=authorize_write if cold_history or operation == 'import' else None) # Post-commit projections are idempotent reads of the committed receipt, so exact # retries repeat them (like delete's retirement); only the one-shot event is fenced. + _project_committed(authority, ref, operation, result) + if live is not None: + if operation == 'rewind': + authority.runner._evict_cached_agent(live.route) + if applied: + live.event_stream.publish(ref.session_id, result, event_type='session.updated') + return result + + +def _project_committed(authority, ref, operation, result): + """Publish a committed receipt's runtime projections (safe to repeat on exact retry).""" if operation == 'model': from gateway.session_local import publish_local_policy publish_local_policy(authority, ref.session_id) @@ -142,9 +166,3 @@ async def mutate_session(authority, actor, ref, params): evict = getattr(authority.runner, '_evict_cached_agent', None) if callable(evict): evict(candidate.route) - if live is not None: - if operation == 'rewind': - authority.runner._evict_cached_agent(live.route) - if applied: - live.event_stream.publish(ref.session_id, result, event_type='session.updated') - return result diff --git a/gateway/session_policy.py b/gateway/session_policy.py index f8564efb69..7f44d28fba 100644 --- a/gateway/session_policy.py +++ b/gateway/session_policy.py @@ -85,8 +85,6 @@ def present_sections(config): def build_policy(params, config, *, private_secrets=None, profile_terminal=True): - from hermes_cli.tools_config import _get_platform_tools - from toolsets import validate_toolset from agent.runtime_cwd import resolve_agent_cwd from tools.terminal_scope import build_profile_terminal_scope, default_terminal_scope from hermes_constants import get_hermes_home @@ -112,6 +110,23 @@ def build_policy(params, config, *, private_secrets=None, profile_terminal=True) from gateway.session_local_editor import validate_editor validate_editor(source, params.get('editor')) config = present_sections(json.loads(json.dumps(config))) + _apply_launch_overrides(params, config) + enabled = _resolve_toolsets(params, config, source, safe_mode) + terminal = build_profile_terminal_scope(get_hermes_home()) if profile_terminal else default_terminal_scope() + terminal['TERMINAL_CWD'] = cwd + request = {k: v for k, v in params.items() if k not in {'request_id', 'api_key'}} + request.setdefault('source', 'cli') + from gateway.session_local_mcp import private_editor_request + private_editor_request(request, private_secrets) + _extract_config_secrets(config, private_secrets) + _extract_config_secrets(terminal, private_secrets, (None,)) + return LocalSessionPolicy(source, SURFACES[source], cwd, model, tuple(sorted(enabled)), + json.dumps(config), json.dumps(request, sort_keys=True), json.dumps(terminal), + safe_mode=safe_mode, ignore_user_config=ignore_user_config) + + +def _apply_launch_overrides(params, config): + """Validate the explicit model/agent launch overrides and fold them into *config*.""" from urllib.parse import urlsplit from hermes_constants import parse_reasoning_effort for key in ('provider', 'base_url'): @@ -139,6 +154,12 @@ def build_policy(params, config, *, private_secrets=None, profile_terminal=True) config.setdefault('agent', {})['reasoning_effort'] = params['reasoning'] # An explicit launch level wins over a per-model default, just like CLI. config['agent'].pop('reasoning_overrides', None) + + +def _resolve_toolsets(params, config, source, safe_mode): + """Validate explicit toolsets and return the session's enabled toolset set.""" + from hermes_cli.tools_config import _get_platform_tools + from toolsets import validate_toolset explicit = params.get('toolsets') if 'toolsets' in params: if (not isinstance(explicit, list) or any(not isinstance(x, str) or not validate_toolset(x) for x in explicit) @@ -166,17 +187,7 @@ def build_policy(params, config, *, private_secrets=None, profile_terminal=True) # Plugin toolsets are user customizations; the safe worker never imports them. from hermes_cli.tools_config import _get_plugin_toolset_keys enabled -= _get_plugin_toolset_keys() - terminal = build_profile_terminal_scope(get_hermes_home()) if profile_terminal else default_terminal_scope() - terminal['TERMINAL_CWD'] = cwd - request = {k: v for k, v in params.items() if k not in {'request_id', 'api_key'}} - request.setdefault('source', 'cli') - from gateway.session_local_mcp import private_editor_request - private_editor_request(request, private_secrets) - _extract_config_secrets(config, private_secrets) - _extract_config_secrets(terminal, private_secrets, (None,)) - return LocalSessionPolicy(source, SURFACES[source], cwd, model, tuple(sorted(enabled)), - json.dumps(config), json.dumps(request, sort_keys=True), json.dumps(terminal), - safe_mode=safe_mode, ignore_user_config=ignore_user_config) + return enabled def _extract_config_secrets(value, private, path=()):