fix(session): harden cron branch source gate (#5477)

api/routes.py

@@ -7535,7 +7535,8 @@ def _load_branch_source_or_refuse(handler, sid: str):
         return get_session(sid)
     except KeyError:
         _foreign_session, _reason = _claim_or_synthesize_cli_session(sid)
-        if _reason == "not_claimable" and _foreign_session is not None and is_cron_session(getattr(_foreign_session, "session_id", ""), str((getattr(_foreign_session, "source_tag", None) or getattr(_foreign_session, "raw_source", None) or getattr(_foreign_session, "source", None) or "")).strip().lower()):
+        _source_kind = str((getattr(_foreign_session, "source_tag", None) or getattr(_foreign_session, "raw_source", None) or getattr(_foreign_session, "source", None) or "")).strip().lower() if _foreign_session is not None else ""
+        if _reason == "not_claimable" and _foreign_session is not None and _source_kind == "cron":
             _foreign_session._branch_source_readonly = True; return _foreign_session
         if _reason == "not_claimable": bad(handler, "Read-only sessions cannot be branched from WebUI", 403); return None
         bad(handler, "Session not found", 404)

static/sessions.js

@@ -2012,11 +2012,8 @@ function _isBranchableReadOnlySession(session) {
     session && session.source_tag,
     session && session.raw_source,
     session && session.source,
-    session && session.session_source,
-    session && session.source_label,
   ].map(v => String(v || '').trim().toLowerCase());
-  const sid = String(session && session.session_id || '').toLowerCase();
-  return sources.includes('cron') || sid.startsWith('cron_');
+  return sources.includes('cron');
 }
 
 function _sourceKeyForSession(session) {

tests/test_465_session_branching.py

@@ -218,8 +218,10 @@ def test_branch_endpoint_consults_foreign_session_guard_on_missing_sidecar():
         "Helper should classify missing-sidecar foreign sessions before returning"
     assert 'if _reason == "not_claimable":' in helper, \
         "Helper should branch on not_claimable foreign ownership"
-    assert 'is_cron_session(' in helper, \
-        "Helper should narrow read-only branch sources to canonical cron sessions"
+    assert '_source_kind == "cron"' in helper, \
+        "Helper should narrow read-only branch sources to resolved cron source metadata"
+    assert 'is_cron_session(' not in helper, \
+        "Helper should not use the cron_ session-id prefix as a branch permission gate"
     assert '_foreign_session._branch_source_readonly = True' in helper, \
         "Helper should mark synthesized read-only sources so branch does not save them"
     assert 'return _foreign_session' in helper, \
@@ -537,7 +539,7 @@ def test_branchable_read_only_helper_accepts_cron_sources():
     assert "session && session.source_tag" in src
     assert "session && session.raw_source" in src
     assert "sources.includes('cron')" in src
-    assert "sid.startsWith('cron_')" in src
+    assert "sid.startsWith('cron_')" not in src
     assert "sid.startsWith('cron-')" not in src
 
 
@@ -595,7 +597,7 @@ def test_forkFromMessage_rejects_read_only_non_cron_sessions_without_loading_or_
 def test_forkFromMessage_allows_read_only_cron_sessions_to_post():
     """Read-only cron message forks should reach the existing keep_count path."""
     result = _commands_harness(
-        "S.session = { session_id: 'cron_1', read_only: true };\n"
+        "S.session = { session_id: 'cron_1', raw_source: 'cron', read_only: true };\n"
         "_oldestIdx = 2;\n"
         "await forkFromMessage(4);\n"
         "console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));"
@@ -611,15 +613,15 @@ def test_forkFromMessage_allows_read_only_cron_sessions_to_post():
     assert payload["renderCalls"] == 1, "cron forkFromMessage should refresh the session list"
 
 
-def test_cmdBranch_rejects_non_canonical_cron_dash_id_without_posting():
-    """Only canonical cron source fields or cron_ ids should unlock read-only branching."""
+def test_cmdBranch_rejects_cron_prefixed_id_without_canonical_source():
+    """Only canonical cron source fields should unlock read-only branching."""
     result = _commands_harness(
-        "S.session = { session_id: 'cron-1', session_source: 'other', read_only: true };\n"
+        "S.session = { session_id: 'cron_spoof_messaging', session_source: 'other', read_only: true };\n"
         "await cmdBranch('');\n"
         "console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));"
     )
     payload = json.loads(result)
-    assert payload["calls"] == [], "cron- id alone should not unlock read-only /branch"
+    assert payload["calls"] == [], "cron_ id alone should not unlock read-only /branch"
     assert payload["toasts"][0][0] == "Read-only sessions cannot be forked."
 
 

tests/test_chat_start_claim_cli_session.py

@@ -1101,6 +1101,34 @@ def test_branch_still_refuses_non_cron_not_claimable_sources(
     assert not (isolated_state_db["sessions_dir"] / f"{SID}.json").exists()
 
 
+def test_branch_refuses_cron_prefixed_non_cron_not_claimable_source(
+    routes_module, monkeypatch, isolated_state_db
+):
+    SID = "cron_spoof_messaging"
+    _make_state_db(
+        isolated_state_db["db"], SID, message_count=1,
+        title="Messaging chat", source="messaging", cwd="/root",
+    )
+    monkeypatch.setattr(
+        routes_module,
+        "_lookup_cli_session_metadata",
+        lambda _sid: {
+            "session_id": SID,
+            "source_tag": "messaging",
+            "raw_source": "messaging",
+            "session_source": "other",
+        },
+    )
+    monkeypatch.setattr(routes_module, "_check_csrf", lambda _handler: True)
+    monkeypatch.setattr(routes_module, "_guard_request_session_visibility", lambda *args, **kwargs: True)
+    handler = _FakePostHandler({"session_id": SID}, path="/api/session/branch")
+    routes_module.handle_post(handler, SimpleNamespace(path="/api/session/branch", query=""))
+    assert handler.status == 403
+    payload = _response_json(handler)
+    assert "read-only" in payload["error"].lower()
+    assert not (isolated_state_db["sessions_dir"] / f"{SID}.json").exists()
+
+
 def test_branch_from_claimable_tui_still_creates_fork(
     routes_module, monkeypatch, isolated_state_db
 ):