nesquena-hermes
commit
f10cb2a35c
6 changed files with 313 additions and 31 deletions
|
|
@ -7563,9 +7563,10 @@ 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":
|
||||
bad(handler, "Read-only sessions cannot be branched from WebUI", 403)
|
||||
return None
|
||||
_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)
|
||||
return None
|
||||
|
||||
|
|
@ -13959,7 +13960,7 @@ def handle_post(handler, parsed) -> bool:
|
|||
# /api/session so frontend keep_count values from merged messaging
|
||||
# transcripts do not silently become full sidecar copies.
|
||||
try:
|
||||
source.save()
|
||||
if not getattr(source, "_branch_source_readonly", False): source.save()
|
||||
except Exception:
|
||||
pass
|
||||
cli_meta = _lookup_cli_session_metadata(source.session_id) if _session_requires_cli_metadata_lookup(source) else {}
|
||||
|
|
|
|||
|
|
@ -1807,7 +1807,10 @@ async function cmdBranch(args){
|
|||
const readOnlySession=typeof _isReadOnlySession==='function'
|
||||
? _isReadOnlySession(S.session)
|
||||
: !!(S.session&&(S.session.read_only||S.session.is_read_only));
|
||||
if(readOnlySession){showToast('Read-only sessions cannot be forked.',3000);return;}
|
||||
const branchableReadOnlySession=typeof _isBranchableReadOnlySession==='function'
|
||||
? _isBranchableReadOnlySession(S.session)
|
||||
: false;
|
||||
if(readOnlySession&&!branchableReadOnlySession){showToast('Read-only sessions cannot be forked.',3000);return;}
|
||||
const customTitle=(args||'').trim()||null;
|
||||
try{
|
||||
const data=await api('/api/session/branch',{
|
||||
|
|
@ -1838,7 +1841,10 @@ async function forkFromMessage(msgIdx){
|
|||
const readOnlySession=typeof _isReadOnlySession==='function'
|
||||
? _isReadOnlySession(S.session)
|
||||
: !!(S.session&&(S.session.read_only||S.session.is_read_only));
|
||||
if(readOnlySession){showToast('Read-only sessions cannot be forked.',3000);return;}
|
||||
const branchableReadOnlySession=typeof _isBranchableReadOnlySession==='function'
|
||||
? _isBranchableReadOnlySession(S.session)
|
||||
: false;
|
||||
if(readOnlySession&&!branchableReadOnlySession){showToast('Read-only sessions cannot be forked.',3000);return;}
|
||||
const initialSid = S.session.session_id;
|
||||
// Capture the absolute keep_count before any async work that may
|
||||
// reset _oldestIdx. _oldestIdx is 0 when the full transcript is
|
||||
|
|
|
|||
|
|
@ -2012,6 +2012,16 @@ function _isReadOnlySession(session) {
|
|||
return !!(session && (session.read_only || session.is_read_only));
|
||||
}
|
||||
|
||||
function _isBranchableReadOnlySession(session) {
|
||||
if (!_isReadOnlySession(session)) return false;
|
||||
const sources = [
|
||||
session && session.source_tag,
|
||||
session && session.raw_source,
|
||||
session && session.source,
|
||||
].map(v => String(v || '').trim().toLowerCase());
|
||||
return sources.includes('cron');
|
||||
}
|
||||
|
||||
function _sourceKeyForSession(session) {
|
||||
return (session && (session.raw_source || session.source_tag || session.source || '') || '').toLowerCase();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -13825,7 +13825,10 @@ function renderMessages(options){
|
|||
const readOnlySession=typeof _isReadOnlySession==='function'
|
||||
? _isReadOnlySession(S.session)
|
||||
: !!(S.session&&(S.session.read_only||S.session.is_read_only));
|
||||
const forkBtn = readOnlySession ? '' : `<button class="msg-action-btn" title="${t('fork_from_here')}" onclick="forkFromMessage(${rawIdx+1})">${li('git-branch',13)}</button>`;
|
||||
const branchableReadOnlySession=typeof _isBranchableReadOnlySession==='function'
|
||||
? _isBranchableReadOnlySession(S.session)
|
||||
: false;
|
||||
const forkBtn = (readOnlySession&&!branchableReadOnlySession) ? '' : `<button class="msg-action-btn" title="${t('fork_from_here')}" onclick="forkFromMessage(${rawIdx+1})">${li('git-branch',13)}</button>`;
|
||||
const ttsBtn = !isUser ? `<button class="msg-action-btn msg-tts-btn" title="${t('tts_listen')||'Listen'}" onclick="speakMessage(this)">${li('volume-2',13)}</button>` : '';
|
||||
const tsVal=m._ts||m.timestamp;
|
||||
// _formatInServerTz handles fractional-hour offsets (India +0530 etc.)
|
||||
|
|
|
|||
|
|
@ -25,6 +25,7 @@ import pytest
|
|||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
COMMANDS_JS = ROOT / "static" / "commands.js"
|
||||
SESSIONS_JS = ROOT / "static" / "sessions.js"
|
||||
NODE = shutil.which("node")
|
||||
|
||||
|
||||
|
|
@ -49,6 +50,23 @@ def _extract_async_function(source: str, name: str) -> str:
|
|||
pytest.fail(f"Could not extract complete function body for {name}")
|
||||
|
||||
|
||||
def _extract_function(source: str, name: str) -> str:
|
||||
start = source.find(f"function {name}(")
|
||||
assert start != -1, f"Could not find function {name}"
|
||||
brace = source.find("{", start)
|
||||
assert brace != -1, f"Could not find opening brace for {name}"
|
||||
depth = 0
|
||||
for idx in range(brace, len(source)):
|
||||
ch = source[idx]
|
||||
if ch == "{":
|
||||
depth += 1
|
||||
elif ch == "}":
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
return source[start:idx + 1]
|
||||
pytest.fail(f"Could not extract complete function body for {name}")
|
||||
|
||||
|
||||
def _run_node(script: str) -> str:
|
||||
if NODE is None:
|
||||
pytest.skip("node not on PATH")
|
||||
|
|
@ -69,8 +87,11 @@ def _run_node(script: str) -> str:
|
|||
|
||||
def _commands_harness(body: str) -> str:
|
||||
source = COMMANDS_JS.read_text(encoding="utf-8")
|
||||
session_source = SESSIONS_JS.read_text(encoding="utf-8")
|
||||
cmd_branch = _extract_async_function(source, "cmdBranch")
|
||||
fork_from = _extract_async_function(source, "forkFromMessage")
|
||||
is_read_only = _extract_function(session_source, "_isReadOnlySession")
|
||||
is_branchable_read_only = _extract_function(session_source, "_isBranchableReadOnlySession")
|
||||
return _run_node(
|
||||
"\n".join([
|
||||
"const calls = [];",
|
||||
|
|
@ -93,9 +114,8 @@ def _commands_harness(body: str) -> str:
|
|||
"const loadSession = async (sid) => { loadedSessions.push(sid); };",
|
||||
"const renderSessionList = async () => { renderCalls += 1; };",
|
||||
"const _ensureAllMessagesLoaded = async () => { ensureCalls += 1; };",
|
||||
"function _isReadOnlySession(session) {",
|
||||
" return !!(session && (session.read_only || session.is_read_only));",
|
||||
"}",
|
||||
is_read_only,
|
||||
is_branchable_read_only,
|
||||
cmd_branch,
|
||||
fork_from,
|
||||
"(async () => {",
|
||||
|
|
@ -186,7 +206,8 @@ def test_branch_endpoint_consults_foreign_session_guard_on_missing_sidecar():
|
|||
assert '_load_branch_source_or_refuse(handler, body["session_id"])' in block, \
|
||||
"Branch handler should delegate source-load/refusal to the shared helper"
|
||||
|
||||
# The helper itself must carry the foreign-session classification + 403.
|
||||
# The helper itself must carry the foreign-session classification and pass
|
||||
# read-only cron-like sources to the branch builder without saving them.
|
||||
helper_match = re.search(
|
||||
r'def _load_branch_source_or_refuse\(.*?\)(.*?)(?=\ndef )',
|
||||
src, re.DOTALL
|
||||
|
|
@ -197,10 +218,16 @@ 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 'return bad(handler, "Read-only sessions cannot be branched from WebUI", 403)' in helper \
|
||||
or ('bad(handler, "Read-only sessions cannot be branched from WebUI", 403)' in helper
|
||||
and 'return None' in helper), \
|
||||
"Helper should return a provenance-correct 403 for read-only foreign 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, \
|
||||
"Helper should return synthesized read-only sources to the branch builder"
|
||||
assert 'bad(handler, "Read-only sessions cannot be branched from WebUI", 403)' in helper, \
|
||||
"Helper should keep non-cron not_claimable sources refused"
|
||||
|
||||
|
||||
def test_branch_endpoint_returns_new_session_id():
|
||||
|
|
@ -333,8 +360,8 @@ def test_branch_auto_title():
|
|||
assert '(fork)' in block, "Branch handler should auto-title as '(fork)'"
|
||||
|
||||
|
||||
def test_branch_route_rejects_not_claimable_foreign_sessions_with_403(monkeypatch):
|
||||
"""Direct or stale branch POSTs for read-only foreign sessions must refuse clearly."""
|
||||
def test_branch_route_allows_not_claimable_cron_sessions_to_fork(monkeypatch):
|
||||
"""Direct or stale branch POSTs for read-only cron sessions should create a fork."""
|
||||
handler = _FakeHandler()
|
||||
monkeypatch.setattr(routes, "_check_csrf", lambda _handler: True)
|
||||
monkeypatch.setattr(routes, "read_body", lambda _handler: {"session_id": "cron-1"})
|
||||
|
|
@ -343,14 +370,24 @@ def test_branch_route_rejects_not_claimable_foreign_sessions_with_403(monkeypatc
|
|||
"get_session",
|
||||
lambda _sid, metadata_only=False: (_ for _ in ()).throw(KeyError("Session not found")),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
routes,
|
||||
"_claim_or_synthesize_cli_session",
|
||||
lambda _sid: (object(), "not_claimable"),
|
||||
source = routes.Session(
|
||||
session_id="cron-1",
|
||||
title="Cron Run",
|
||||
workspace=".",
|
||||
model="claude-sonnet",
|
||||
messages=[{"role": "user", "content": "summarize"}],
|
||||
source_tag="cron",
|
||||
raw_source="cron",
|
||||
session_source="other",
|
||||
)
|
||||
monkeypatch.setattr(routes, "_claim_or_synthesize_cli_session", lambda _sid: (source, "not_claimable"))
|
||||
cap = _capture_route(monkeypatch)
|
||||
routes.handle_post(handler, urlparse("/api/session/branch"))
|
||||
assert cap["bad"] == ("Read-only sessions cannot be branched from WebUI", 403)
|
||||
assert "bad" not in cap
|
||||
assert cap["status"] == 200
|
||||
assert cap["ok"]["title"] == "Cron Run (fork)"
|
||||
assert cap["ok"]["parent_session_id"] == "cron-1"
|
||||
assert cap["ok"]["session_id"] in routes.SESSIONS
|
||||
|
||||
|
||||
def test_branch_route_keeps_404_for_truly_missing_sessions(monkeypatch):
|
||||
|
|
@ -485,18 +522,31 @@ def test_fork_button_in_message_actions():
|
|||
|
||||
|
||||
def test_fork_button_is_hidden_for_read_only_sessions():
|
||||
"""Read-only sessions should not render the message-level fork affordance."""
|
||||
"""Non-cron read-only sessions should not render the message-level fork affordance."""
|
||||
src = _read('static/ui.js')
|
||||
assert "const readOnlySession=typeof _isReadOnlySession==='function'" in src, \
|
||||
"ui.js should derive a read-only session flag from the shared helper"
|
||||
assert "const forkBtn = readOnlySession ? '' :" in src, \
|
||||
"fork button should be suppressed when the active session is read-only"
|
||||
assert "const branchableReadOnlySession=typeof _isBranchableReadOnlySession==='function'" in src, \
|
||||
"ui.js should derive a branchable read-only flag from the shared helper"
|
||||
assert "const forkBtn = (readOnlySession&&!branchableReadOnlySession) ? '' :" in src, \
|
||||
"fork button should be suppressed when a read-only session is not branchable"
|
||||
|
||||
|
||||
def test_branchable_read_only_helper_accepts_cron_sources():
|
||||
"""Read-only cron sessions should be forkable follow-up sources."""
|
||||
src = _read('static/sessions.js')
|
||||
assert "function _isBranchableReadOnlySession(session)" in src
|
||||
assert "session && session.source_tag" in src
|
||||
assert "session && session.raw_source" in src
|
||||
assert "sources.includes('cron')" in src
|
||||
assert "sid.startsWith('cron_')" not in src
|
||||
assert "sid.startsWith('cron-')" not in src
|
||||
|
||||
|
||||
def test_cmdBranch_rejects_read_only_sessions_without_posting():
|
||||
"""The /branch command must not POST for read-only sessions."""
|
||||
"""The /branch command must not POST for non-cron read-only sessions."""
|
||||
result = _commands_harness(
|
||||
"S.session = { session_id: 'cron-1', read_only: true };\n"
|
||||
"S.session = { session_id: 'subagent-1', session_source: 'subagent', read_only: true };\n"
|
||||
"await cmdBranch('');\n"
|
||||
"console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));"
|
||||
)
|
||||
|
|
@ -509,10 +559,29 @@ def test_cmdBranch_rejects_read_only_sessions_without_posting():
|
|||
assert payload["toasts"][0][0] == "Read-only sessions cannot be forked."
|
||||
|
||||
|
||||
def test_forkFromMessage_rejects_read_only_sessions_without_loading_or_posting():
|
||||
"""Read-only message forks must stop before the load/post path."""
|
||||
def test_cmdBranch_allows_read_only_cron_sessions_to_post():
|
||||
"""The /branch command should POST for read-only cron sessions."""
|
||||
result = _commands_harness(
|
||||
"S.session = { session_id: 'cron-1', read_only: true };\n"
|
||||
"S.session = { session_id: 'daily-summary', session_source: 'other', raw_source: 'cron', read_only: true };\n"
|
||||
"await cmdBranch('Follow-up');\n"
|
||||
"console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));"
|
||||
)
|
||||
payload = json.loads(result)
|
||||
assert len(payload["calls"]) == 1, "read-only cron /branch should POST once"
|
||||
call = payload["calls"][0]
|
||||
assert call["url"] == "/api/session/branch"
|
||||
assert call["body"]["session_id"] == "daily-summary"
|
||||
assert call["body"]["title"] == "Follow-up"
|
||||
assert payload["ensureCalls"] == 0, "cmdBranch should not trigger message loading"
|
||||
assert payload["loadedSessions"] == ["forked-session"], "cron /branch should load the forked session"
|
||||
assert payload["renderCalls"] == 1, "cron /branch should refresh the session list"
|
||||
assert payload["toasts"][0][0] == "Forked into new session"
|
||||
|
||||
|
||||
def test_forkFromMessage_rejects_read_only_non_cron_sessions_without_loading_or_posting():
|
||||
"""Non-cron read-only message forks must stop before the load/post path."""
|
||||
result = _commands_harness(
|
||||
"S.session = { session_id: 'subagent-1', session_source: 'subagent', read_only: true };\n"
|
||||
"await forkFromMessage(1);\n"
|
||||
"console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));"
|
||||
)
|
||||
|
|
@ -525,6 +594,37 @@ def test_forkFromMessage_rejects_read_only_sessions_without_loading_or_posting()
|
|||
assert payload["toasts"][0][0] == "Read-only sessions cannot be forked."
|
||||
|
||||
|
||||
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', raw_source: 'cron', read_only: true };\n"
|
||||
"_oldestIdx = 2;\n"
|
||||
"await forkFromMessage(4);\n"
|
||||
"console.log(JSON.stringify({ calls, toasts, ensureCalls, loadedSessions, renderCalls }));"
|
||||
)
|
||||
payload = json.loads(result)
|
||||
assert len(payload["calls"]) == 1, "read-only cron forkFromMessage should POST once"
|
||||
call = payload["calls"][0]
|
||||
assert call["url"] == "/api/session/branch"
|
||||
assert call["body"]["session_id"] == "cron_1"
|
||||
assert call["body"]["keep_count"] == 6
|
||||
assert payload["ensureCalls"] == 2, "cron forkFromMessage should preserve the full-load flow"
|
||||
assert payload["loadedSessions"] == ["forked-session"], "cron forkFromMessage should load the fork"
|
||||
assert payload["renderCalls"] == 1, "cron forkFromMessage should refresh the session list"
|
||||
|
||||
|
||||
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_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["toasts"][0][0] == "Read-only sessions cannot be forked."
|
||||
|
||||
|
||||
def test_forkFromMessage_preserves_absolute_keep_count_for_writable_sessions():
|
||||
"""The read-only guard must not break the existing absolute keep_count fix."""
|
||||
result = _commands_harness(
|
||||
|
|
|
|||
|
|
@ -14,10 +14,12 @@ reason branch correctly with monkey-patched state.db / SESSION_INDEX_FILE).
|
|||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
|
@ -298,6 +300,32 @@ def _write_index(path: Path, entries: list[dict]) -> None:
|
|||
path.write_text(json.dumps(entries), encoding="utf-8")
|
||||
|
||||
|
||||
class _FakePostHandler:
|
||||
def __init__(self, body: dict, *, path: str):
|
||||
raw = json.dumps(body).encode("utf-8")
|
||||
self.status = None
|
||||
self.response_headers = {}
|
||||
self.headers = {"Content-Length": str(len(raw))}
|
||||
self.rfile = io.BytesIO(raw)
|
||||
self.wfile = io.BytesIO()
|
||||
self.command = "POST"
|
||||
self.path = path
|
||||
self.client_address = ("127.0.0.1", 12345)
|
||||
|
||||
def send_response(self, status):
|
||||
self.status = status
|
||||
|
||||
def send_header(self, key, value):
|
||||
self.response_headers[key] = value
|
||||
|
||||
def end_headers(self):
|
||||
pass
|
||||
|
||||
|
||||
def _response_json(handler: _FakePostHandler) -> dict:
|
||||
return json.loads(handler.wfile.getvalue().decode("utf-8"))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def routes_module():
|
||||
return pytest.importorskip("api.routes")
|
||||
|
|
@ -451,7 +479,7 @@ def test_helper_materialises_state_db_only_session(
|
|||
assert sess.session_id == SID
|
||||
assert sess.title == "Codex honcho integration"
|
||||
assert sess.model == "MiniMax-M3"
|
||||
assert sess.workspace == "/root" # from CLI metadata
|
||||
assert Path(sess.workspace).name == "root" # from CLI metadata
|
||||
assert len(sess.messages) == 3
|
||||
assert sess.messages[0]["role"] == "user"
|
||||
# Greptile #4911 P1: created_at must be populated from state.db
|
||||
|
|
@ -1023,6 +1051,140 @@ def test_helper_refuses_cron_via_state_db_source(
|
|||
assert sess.read_only is True
|
||||
|
||||
|
||||
def test_branch_from_cron_state_db_returns_writable_fork_without_source_sidecar(
|
||||
routes_module, monkeypatch, isolated_state_db
|
||||
):
|
||||
SID = "20260610_cron_branch_route"
|
||||
_make_state_db(
|
||||
isolated_state_db["db"], SID, message_count=2,
|
||||
title="Cron job", source="cron", cwd="/root",
|
||||
)
|
||||
monkeypatch.setattr(routes_module, "_lookup_cli_session_metadata", lambda _sid: {})
|
||||
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 == 200
|
||||
payload = _response_json(handler)
|
||||
assert payload["parent_session_id"] == SID
|
||||
assert payload["session_id"] != SID
|
||||
assert (isolated_state_db["sessions_dir"] / f'{payload["session_id"]}.json').exists()
|
||||
assert not (isolated_state_db["sessions_dir"] / f"{SID}.json").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("source", ["gateway", "messaging", "external_agent", "unknown", "claude_code"])
|
||||
def test_branch_still_refuses_non_cron_not_claimable_sources(
|
||||
routes_module, monkeypatch, isolated_state_db, source
|
||||
):
|
||||
SID = f"20260610_{source}_branch_route"
|
||||
_make_state_db(
|
||||
isolated_state_db["db"], SID, message_count=1,
|
||||
title=f"{source} chat", source=source, cwd="/root",
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
routes_module,
|
||||
"_lookup_cli_session_metadata",
|
||||
lambda _sid: {
|
||||
"session_id": SID,
|
||||
"source_tag": source,
|
||||
"raw_source": source,
|
||||
"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_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
|
||||
):
|
||||
SID = "20260610_tui_branch_route"
|
||||
_make_state_db(
|
||||
isolated_state_db["db"], SID, message_count=3,
|
||||
title="TUI chat", source="tui", cwd="/root",
|
||||
)
|
||||
monkeypatch.setattr(routes_module, "_lookup_cli_session_metadata", lambda _sid: {})
|
||||
monkeypatch.setattr(routes_module, "_check_csrf", lambda _handler: True)
|
||||
monkeypatch.setattr(routes_module, "_guard_request_session_visibility", lambda *args, **kwargs: True)
|
||||
sess, reason = routes_module._claim_or_synthesize_cli_session(SID)
|
||||
assert reason == "materialized"
|
||||
sess.save()
|
||||
handler = _FakePostHandler({"session_id": SID}, path="/api/session/branch")
|
||||
routes_module.handle_post(handler, SimpleNamespace(path="/api/session/branch", query=""))
|
||||
assert handler.status == 200
|
||||
payload = _response_json(handler)
|
||||
assert payload["parent_session_id"] == SID
|
||||
assert payload["title"] == "TUI chat (fork)"
|
||||
assert (isolated_state_db["sessions_dir"] / f"{SID}.json").exists()
|
||||
assert (isolated_state_db["sessions_dir"] / f'{payload["session_id"]}.json').exists()
|
||||
|
||||
|
||||
def test_chat_start_still_refuses_cron_state_db_source(
|
||||
routes_module, monkeypatch, isolated_state_db
|
||||
):
|
||||
SID = "20260610_cron_chat_start_route"
|
||||
_make_state_db(
|
||||
isolated_state_db["db"], SID, message_count=1,
|
||||
title="Cron job", source="cron", cwd="/root",
|
||||
)
|
||||
monkeypatch.setattr(routes_module, "_lookup_cli_session_metadata", lambda _sid: {})
|
||||
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/chat/start")
|
||||
routes_module.handle_post(handler, SimpleNamespace(path="/api/chat/start", 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_refuses_subagent_view_only_source(
|
||||
routes_module, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(routes_module, "_check_csrf", lambda _handler: True)
|
||||
monkeypatch.setattr(routes_module, "_guard_request_session_visibility", lambda *args, **kwargs: True)
|
||||
monkeypatch.setattr(routes_module, "_session_is_subagent_view_only", lambda _sid: True)
|
||||
handler = _FakePostHandler({"session_id": "subagent-1"}, path="/api/session/branch")
|
||||
routes_module.handle_post(handler, SimpleNamespace(path="/api/session/branch", query=""))
|
||||
assert handler.status == 400
|
||||
payload = _response_json(handler)
|
||||
assert "view-only" in payload["error"].lower()
|
||||
|
||||
|
||||
def test_helper_reason_distinguishes_cli_meta_vs_state_db_source(
|
||||
routes_module,
|
||||
):
|
||||
|
|
|
|||
Loading…
Reference in a new issue