608 lines
23 KiB
Python
608 lines
23 KiB
Python
"""Regression coverage for #6220 historical tool transcript ownership.
|
|
|
|
Only an unambiguous ``assistant.tool_calls[].id`` ->
|
|
``role=tool.tool_call_id`` turn may move from the legacy settled renderer into
|
|
an Assistant Turn Anchor scene. Ambiguous or richer historical shapes stay on
|
|
the compatibility path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
UI_JS = (ROOT / "static" / "ui.js").read_text(encoding="utf-8")
|
|
ANCHOR_JS = ROOT / "static" / "assistant_turn_anchors.js"
|
|
NODE = shutil.which("node")
|
|
|
|
|
|
_EXTRACT_FUNC_JS = r"""
|
|
function extractFunc(name){
|
|
const start = src.indexOf('function ' + name);
|
|
if(start === -1) throw new Error(name + ' not found');
|
|
const params = src.indexOf('(', start);
|
|
let depth = 0, close = -1;
|
|
for(let i=params; i<src.length; i++){
|
|
if(src[i] === '(') depth++;
|
|
else if(src[i] === ')'){
|
|
depth--;
|
|
if(depth === 0){ close = i; break; }
|
|
}
|
|
}
|
|
const brace = src.indexOf('{', close);
|
|
depth = 0;
|
|
for(let i=brace; i<src.length; i++){
|
|
if(src[i] === '{') depth++;
|
|
else if(src[i] === '}'){
|
|
depth--;
|
|
if(depth === 0) return src.slice(start, i + 1);
|
|
}
|
|
}
|
|
throw new Error(name + ' body did not close');
|
|
}
|
|
""".strip()
|
|
|
|
|
|
def _run_hydration(messages: list[dict], *, repeat: bool = False, throw_helper: bool = False) -> dict:
|
|
assert NODE, "node is required for Anchor hydration behavior tests"
|
|
script = f"""
|
|
const fs = require('fs');
|
|
const vm = require('vm');
|
|
const src = fs.readFileSync({json.dumps(str(ROOT / 'static' / 'ui.js'))}, 'utf8');
|
|
{_EXTRACT_FUNC_JS}
|
|
|
|
const sandbox = {{console}};
|
|
sandbox.window = sandbox;
|
|
sandbox.globalThis = sandbox;
|
|
vm.createContext(sandbox);
|
|
vm.runInContext(
|
|
fs.readFileSync({json.dumps(str(ANCHOR_JS))}, 'utf8'),
|
|
sandbox,
|
|
{{filename:'assistant_turn_anchors.js'}}
|
|
);
|
|
sandbox._redactToolTargetLabel = value => String(value)
|
|
.replace(/([A-Za-z0-9_]*(?:TOKEN|API[_-]?KEY|SECRET|PASSWORD)[A-Za-z0-9_]*\\s*=\\s*)(?:"[^"]*"|'[^']*'|\\S+)/gi, '$1[redacted]');
|
|
|
|
for(const name of [
|
|
'_clipCliToolSnippet',
|
|
'_cliToolResultText',
|
|
'_cliLooksLikePatchDiff',
|
|
'_cliToolResultSnippet',
|
|
'_prefixedCliDiffLines',
|
|
'_firstOwnedValue',
|
|
'_cliPatchSnippetFromArgs',
|
|
'_cliToolCardSnippet',
|
|
'_cliToolCardHasDiffSnippet',
|
|
'_anchorSceneToolCallFromRow',
|
|
'_toolArgsSnapshot',
|
|
'_assistantReasoningPayloadText',
|
|
'_idLinkedHistoricalMessageText',
|
|
'_idLinkedHistoricalMessageHasVisibleText',
|
|
'_idLinkedHistoricalMessageRef',
|
|
'_idLinkedHistoricalToolArguments',
|
|
'_idLinkedHistoricalToolResultRaw',
|
|
'_idLinkedHistoricalRedactSnippet',
|
|
'_idLinkedHistoricalHasVisibleSidecar',
|
|
'_idLinkedHistoricalTurnScene',
|
|
'_hydrateIdLinkedHistoricalToolScenes',
|
|
]){{
|
|
const fn = src.indexOf('function ' + name) >= 0 ? extractFunc(name) : null;
|
|
if(fn) vm.runInContext(fn, sandbox, {{filename:'ui.js:' + name}});
|
|
}}
|
|
|
|
if({str(throw_helper).lower()}){{
|
|
vm.runInContext(
|
|
"_toolArgsSnapshot = function(){{ throw new Error('synthetic event build failure'); }}",
|
|
sandbox
|
|
);
|
|
}}
|
|
|
|
const messages = {json.dumps(messages)};
|
|
sandbox.messages = messages;
|
|
const first = vm.runInContext(
|
|
"_hydrateIdLinkedHistoricalToolScenes(messages, {{sessionId:'sid-6220', mode:'compact_worklog'}})",
|
|
sandbox
|
|
);
|
|
const afterFirst = JSON.stringify(messages);
|
|
const second = {str(repeat).lower()}
|
|
? vm.runInContext(
|
|
"_hydrateIdLinkedHistoricalToolScenes(messages, {{sessionId:'sid-6220', mode:'compact_worklog'}})",
|
|
sandbox
|
|
)
|
|
: null;
|
|
const owners = messages
|
|
.map((message, index) => message && message._anchor_activity_scene ? {{index, scene:message._anchor_activity_scene}} : null)
|
|
.filter(Boolean);
|
|
sandbox.owners = owners;
|
|
const toolCards = vm.runInContext(
|
|
"owners.map(owner => owner.scene.activity_rows.map(row => _anchorSceneToolCallFromRow(row, {{settled:true}})))",
|
|
sandbox
|
|
);
|
|
console.log(JSON.stringify({{first, second, unchanged:afterFirst===JSON.stringify(messages), owners, toolCards}}));
|
|
"""
|
|
result = subprocess.run(
|
|
[NODE, "-e", script],
|
|
cwd=ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def _run_carry_forward_then_hydration(messages: list[dict]) -> dict:
|
|
assert NODE, "node is required for Anchor hydration behavior tests"
|
|
script = f"""
|
|
const fs = require('fs');
|
|
const vm = require('vm');
|
|
const uiSrc = fs.readFileSync({json.dumps(str(ROOT / 'static' / 'ui.js'))}, 'utf8');
|
|
const messagesSrc = fs.readFileSync({json.dumps(str(ROOT / 'static' / 'messages.js'))}, 'utf8');
|
|
let src = uiSrc;
|
|
{_EXTRACT_FUNC_JS}
|
|
|
|
function extractFuncFrom(source, name){{
|
|
const previous = src;
|
|
src = source;
|
|
try{{ return extractFunc(name); }}
|
|
finally{{ src = previous; }}
|
|
}}
|
|
|
|
const sandbox = {{console}};
|
|
sandbox.window = sandbox;
|
|
sandbox.globalThis = sandbox;
|
|
sandbox._EPHEMERAL_TURN_FIELDS = [
|
|
'_turnUsage',
|
|
'_turnDuration',
|
|
'_turnTps',
|
|
'_gatewayRouting',
|
|
'_statusCard',
|
|
'_anchor_stream_id',
|
|
'_anchor_activity_scene',
|
|
];
|
|
vm.createContext(sandbox);
|
|
vm.runInContext(
|
|
fs.readFileSync({json.dumps(str(ANCHOR_JS))}, 'utf8'),
|
|
sandbox,
|
|
{{filename:'assistant_turn_anchors.js'}}
|
|
);
|
|
sandbox._redactToolTargetLabel = value => String(value);
|
|
|
|
for(const name of [
|
|
'_clipCliToolSnippet',
|
|
'_cliToolResultText',
|
|
'_cliLooksLikePatchDiff',
|
|
'_cliToolResultSnippet',
|
|
'_prefixedCliDiffLines',
|
|
'_firstOwnedValue',
|
|
'_cliPatchSnippetFromArgs',
|
|
'_cliToolCardSnippet',
|
|
'_cliToolCardHasDiffSnippet',
|
|
'_anchorSceneToolCallFromRow',
|
|
'_toolArgsSnapshot',
|
|
'_assistantReasoningPayloadText',
|
|
'_idLinkedHistoricalMessageText',
|
|
'_idLinkedHistoricalMessageHasVisibleText',
|
|
'_idLinkedHistoricalMessageRef',
|
|
'_idLinkedHistoricalToolArguments',
|
|
'_idLinkedHistoricalToolResultRaw',
|
|
'_idLinkedHistoricalRedactSnippet',
|
|
'_idLinkedHistoricalHasVisibleSidecar',
|
|
'_idLinkedHistoricalTurnScene',
|
|
'_hydrateIdLinkedHistoricalToolScenes',
|
|
]){{
|
|
const fn = uiSrc.indexOf('function ' + name) >= 0 ? extractFuncFrom(uiSrc, name) : null;
|
|
if(fn) vm.runInContext(fn, sandbox, {{filename:'ui.js:' + name}});
|
|
}}
|
|
for(const name of [
|
|
'_messageIdentityKey',
|
|
'_isHistoricalAnchorActivityScene',
|
|
'_carryForwardEphemeralTurnFields',
|
|
]){{
|
|
const fn = messagesSrc.indexOf('function ' + name) >= 0 ? extractFuncFrom(messagesSrc, name) : null;
|
|
if(fn) vm.runInContext(fn, sandbox, {{filename:'messages.js:' + name}});
|
|
}}
|
|
|
|
const original = {json.dumps(messages)};
|
|
const prev = JSON.parse(JSON.stringify(original));
|
|
sandbox.prev = prev;
|
|
vm.runInContext(
|
|
"_hydrateIdLinkedHistoricalToolScenes(prev, {{sessionId:'sid-6220', mode:'compact_worklog'}})",
|
|
sandbox
|
|
);
|
|
const next = JSON.parse(JSON.stringify(original));
|
|
sandbox.next = next;
|
|
vm.runInContext("_carryForwardEphemeralTurnFields(prev, next)", sandbox);
|
|
const carried = next
|
|
.map((message, index) => message && message._anchor_activity_scene ? {{index, turn_id:message._anchor_activity_scene.identity.turn_id, tools:message._anchor_activity_scene.activity_rows.map(row => row.tool_call_id)}} : null)
|
|
.filter(Boolean);
|
|
const hydrated = vm.runInContext(
|
|
"_hydrateIdLinkedHistoricalToolScenes(next, {{sessionId:'sid-6220', mode:'compact_worklog'}})",
|
|
sandbox
|
|
);
|
|
const owners = next
|
|
.map((message, index) => message && message._anchor_activity_scene ? {{index, turn_id:message._anchor_activity_scene.identity.turn_id, tools:message._anchor_activity_scene.activity_rows.map(row => row.tool_call_id)}} : null)
|
|
.filter(Boolean);
|
|
console.log(JSON.stringify({{carried, hydrated, owners}}));
|
|
"""
|
|
result = subprocess.run(
|
|
[NODE, "-e", script],
|
|
cwd=ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
assert result.returncode == 0, result.stderr
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def _linked_turn(call_id: str = "call-1") -> list[dict]:
|
|
return [
|
|
{"role": "user", "content": "Inspect the workspace"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": call_id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": "terminal",
|
|
"arguments": '{"command":"pwd"}',
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": call_id,
|
|
"content": '{"output":"/workspace"}',
|
|
},
|
|
{"role": "assistant", "content": "The workspace is `/workspace`."},
|
|
]
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_id_linked_historical_tool_turn_hydrates_one_anchor_scene_idempotently():
|
|
data = _run_hydration(_linked_turn(), repeat=True)
|
|
|
|
assert data["first"] == 1
|
|
assert data["second"] == 0
|
|
assert data["unchanged"] is True
|
|
assert [owner["index"] for owner in data["owners"]] == [3]
|
|
|
|
scene = data["owners"][0]["scene"]
|
|
assert scene["version"] == "activity_scene_v1"
|
|
assert scene["final_answer"] == "The workspace is `/workspace`."
|
|
assert scene["final_message_ref"] == "raw_idx:3"
|
|
assert scene["identity"]["session_id"] == "sid-6220"
|
|
assert scene["identity"]["source_message_refs"] == ["raw_idx:1", "raw_idx:3"]
|
|
assert len(scene["activity_rows"]) == 1
|
|
|
|
row = scene["activity_rows"][0]
|
|
assert row["role"] == "tool"
|
|
assert row["source_event_type"] == "tool_complete"
|
|
assert row["tool_call_id"] == "call-1"
|
|
assert row["tool"]["name"] == "terminal"
|
|
assert row["tool"]["args"] == {"command": "pwd"}
|
|
assert row["tool"]["snippet"] == "/workspace"
|
|
assert row["tool"]["done"] is True
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_id_linked_historical_final_answer_preserves_raw_string_whitespace():
|
|
messages = _linked_turn()
|
|
messages[-1]["content"] = " indented code\n"
|
|
|
|
data = _run_hydration(messages)
|
|
|
|
assert data["first"] == 1
|
|
assert data["owners"][0]["scene"]["final_answer"] == " indented code\n"
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_id_linked_historical_final_answer_joins_text_parts_like_renderer():
|
|
messages = _linked_turn()
|
|
messages[-1]["content"] = [
|
|
{"type": "text", "text": "First paragraph."},
|
|
{"type": "input_text", "text": "ignored input"},
|
|
{"type": "text", "text": "Second paragraph."},
|
|
]
|
|
|
|
data = _run_hydration(messages)
|
|
|
|
assert data["first"] == 1
|
|
assert data["owners"][0]["scene"]["final_answer"] == "First paragraph.\nSecond paragraph."
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_id_linked_historical_multi_tool_turn_preserves_declaration_order():
|
|
messages = [
|
|
{"role": "user", "content": "Inspect both files"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": "call-first",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "terminal",
|
|
"arguments": '{"command":"pwd"}',
|
|
},
|
|
},
|
|
{
|
|
"id": "call-second",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "read_file",
|
|
"arguments": '{"path":"README.md"}',
|
|
},
|
|
},
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call-second", "content": "readme body"},
|
|
{"role": "tool", "tool_call_id": "call-first", "content": "/workspace"},
|
|
{"role": "assistant", "content": "Both checks completed."},
|
|
]
|
|
|
|
data = _run_hydration(messages, repeat=True)
|
|
|
|
assert data["first"] == 1
|
|
assert data["second"] == 0
|
|
rows = data["owners"][0]["scene"]["activity_rows"]
|
|
assert [row["tool_call_id"] for row in rows] == ["call-first", "call-second"]
|
|
assert [row["tool"]["name"] for row in rows] == ["terminal", "read_file"]
|
|
assert [row["tool"]["snippet"] for row in rows] == ["/workspace", "readme body"]
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
@pytest.mark.parametrize(
|
|
"messages",
|
|
[
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [
|
|
{"id": "call-1", "function": {"name": "one", "arguments": "{}"}},
|
|
{"id": "call-1", "function": {"name": "two", "arguments": "{}"}},
|
|
]},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "first"},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "second"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": "I will inspect it first.", "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "reasoning": "Need to inspect it.", "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "user", "content": "What happened?"},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": [{"type": "tool_use", "id": "tool-1", "name": "terminal", "input": {}}]},
|
|
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "tool-1", "content": "ok"}]},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{not-json"}}]},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}],
|
|
"_partial_tool_calls": [{"name": "terminal"}],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
],
|
|
)
|
|
def test_ambiguous_or_richer_historical_tool_shapes_keep_legacy_ownership(messages):
|
|
data = _run_hydration(messages)
|
|
|
|
assert data["first"] == 0
|
|
assert data["owners"] == []
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
@pytest.mark.parametrize("content", [
|
|
{"stdout": "...", "env": "API_KEY=result-secret-123"},
|
|
[{"stdout": "ok"}],
|
|
None,
|
|
])
|
|
def test_non_string_historical_tool_results_keep_legacy_ownership(content):
|
|
messages = _linked_turn()
|
|
messages[2]["content"] = content
|
|
|
|
data = _run_hydration(messages)
|
|
|
|
assert data["first"] == 0
|
|
assert data["owners"] == []
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_string_historical_tool_result_is_redacted_before_anchor_scene():
|
|
messages = _linked_turn()
|
|
messages[2]["content"] = '{"output":"API_KEY=result-secret-123"}'
|
|
|
|
data = _run_hydration(messages)
|
|
|
|
assert data["first"] == 1
|
|
row = data["owners"][0]["scene"]["activity_rows"][0]
|
|
assert "result-secret-123" not in json.dumps(row)
|
|
assert "API_KEY=[redacted]" in row["tool"]["snippet"]
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
@pytest.mark.parametrize(
|
|
"messages",
|
|
[
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "system", "content": "visible system status"},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "developer", "content": "visible developer note"},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "function", "name": "legacy_func", "content": "legacy function output"},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}]},
|
|
{"role": "assistant", "content": None, "tool_calls": []},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"attachments": [{"name": "evidence.txt"}],
|
|
"tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"_statusCard": {"title": "Visible status"},
|
|
"tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": "{}"}}],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
[
|
|
{"role": "user", "content": "Run it"},
|
|
{"role": "assistant", "content": None, "tool_calls": [{"id": "call-1", "function": {"name": "terminal", "arguments": ""}}]},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": "ok"},
|
|
{"role": "assistant", "content": "Done"},
|
|
],
|
|
],
|
|
)
|
|
def test_non_whitelisted_intermediate_records_keep_legacy_ownership(messages):
|
|
data = _run_hydration(messages)
|
|
|
|
assert data["first"] == 0
|
|
assert data["owners"] == []
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_diff_historical_tool_result_preserves_diff_card_projection():
|
|
messages = _linked_turn()
|
|
messages[2]["content"] = "diff --git a/demo.txt b/demo.txt\n@@\n-old\n+new\n"
|
|
|
|
data = _run_hydration(messages)
|
|
|
|
assert data["first"] == 1
|
|
row = data["owners"][0]["scene"]["activity_rows"][0]
|
|
assert row["payload"]["is_diff"] is True
|
|
assert row["tool"]["is_diff"] is True
|
|
assert data["toolCards"][0][0]["is_diff"] is True
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_event_building_helper_exceptions_keep_legacy_ownership():
|
|
data = _run_hydration(_linked_turn(), throw_helper=True)
|
|
|
|
assert data["first"] == 0
|
|
assert data["owners"] == []
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_reused_ids_in_adjacent_complete_turns_remain_turn_scoped():
|
|
data = _run_hydration(_linked_turn("call-reused") + _linked_turn("call-reused"))
|
|
|
|
assert data["first"] == 2
|
|
assert [owner["index"] for owner in data["owners"]] == [3, 7]
|
|
assert [owner["scene"]["activity_rows"][0]["tool_call_id"] for owner in data["owners"]] == ["call-reused", "call-reused"]
|
|
assert data["owners"][0]["scene"]["identity"]["turn_id"] != data["owners"][1]["scene"]["identity"]["turn_id"]
|
|
|
|
|
|
@pytest.mark.skipif(NODE is None, reason="node is required for Anchor hydration tests")
|
|
def test_duplicate_final_answers_rebuild_historical_scenes_after_refresh():
|
|
first_turn = _linked_turn("call-first")
|
|
second_turn = _linked_turn("call-second")
|
|
first_turn[1]["tool_calls"][0]["function"]["name"] = "first_tool"
|
|
first_turn[2]["content"] = "first result"
|
|
first_turn[3]["content"] = "Done"
|
|
second_turn[1]["tool_calls"][0]["function"]["name"] = "second_tool"
|
|
second_turn[2]["content"] = "second result"
|
|
second_turn[3]["content"] = "Done"
|
|
|
|
data = _run_carry_forward_then_hydration(first_turn + second_turn)
|
|
|
|
assert data["carried"] == []
|
|
assert data["hydrated"] == 2
|
|
assert [owner["index"] for owner in data["owners"]] == [3, 7]
|
|
assert [owner["tools"] for owner in data["owners"]] == [["call-first"], ["call-second"]]
|
|
assert data["owners"][0]["turn_id"] != data["owners"][1]["turn_id"]
|
|
|
|
|
|
def test_render_messages_hydrates_before_visibility_and_fallback_ownership_scan():
|
|
start = UI_JS.index("function renderMessages(options)")
|
|
body = UI_JS[start : UI_JS.index("\nfunction ", start + 1)]
|
|
|
|
hydrate = "_hydrateIdLinkedHistoricalToolScenes(S.messages"
|
|
visible = "const visWithIdx=_getVisibleMessagesWithIdx();"
|
|
fallback = "const anchorOwnedAssistantRawIdxs=new Set();"
|
|
|
|
assert hydrate in body
|
|
assert body.index(hydrate) < body.index(visible) < body.index(fallback)
|
|
assert "if(!S.busy" in body[: body.index(visible)]
|