refactor(gateway): split session mutation, launch policy and work-record validators

Code-health ratchet on PR #106742 (CC target 20), pure extractions:
- session_mutations.mutate_session (CC 28): import/control authorization moved into
  _authorize_import/_authorize_control, post-commit runtime projections into
  _project_committed. mutate_runtime_session is still looked up through this module.
- session_policy.build_policy (CC 24): model/agent launch overrides moved into
  _apply_launch_overrides, toolset validation/resolution into _resolve_toolsets.
- hosted_room_work_records.validate (CC 26): availability/tasks/receipts/stop checks
  moved into per-section _validate_* helpers, same order and error messages.

(cherry picked from commit 8908d4fb90f3776ace3787d494065bb696a4fc79)

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

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

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=()):