kevin/deepseek-harness
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(pkg): snapshot advanced Python SDK executable flow

Browse Source 
The existing assertion-only executable smoke proved selected outputs but could not detect drift across the integrated Python SDK, JSON-RPC notification stream, and persisted session shape.

Keep this separate from ACP snapshots because it must launch the actual platform-native packaged executable through the Python SDK. The deterministic model drives Cordis dynamic tool mounting, a Code Mode worker dispatch, direct spawn delegation, workflow-worker delegation, plugin disposal, and the parent/child persistence lineage.

Commit four portable goldens for the SDK result and three JSONL logs. Normalize timestamps, temporary paths, opaque session and agent identifiers, and bulky request headers while retaining ordering, tool names and arguments, header deltas, results, lineage, and final responses so all native build legs compare the same behavior.

Run the comparison in the label-gated executable build workflow and document the current coverage in the paired implemented RFC.
This commit is contained in:
Tianyi Cui
2026-07-13 23:31:26 +08:00
parent b49c37f83a
commit cf89226510
9 changed files with 3068 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/smoke-python-runtime.py
+405 -14
View File
@@ -1,9 +1,10 @@
#!/usr/bin/env python3
"""Keyless full-turn smoke for the SDK wrapper and direct NDJSON runtime use."""
"""Keyless full-turn and snapshot smoke for the Python SDK runtime."""
from __future__ import annotations
import argparse
import difflib
import json
import os
import queue
@@ -13,7 +14,10 @@ import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Callable
from typing import TYPE_CHECKING, Callable
if TYPE_CHECKING:
from deepseek_harness import TurnResult
EXPECTED_TEXT = "runtime smoke ok"
@@ -21,6 +25,32 @@ CODE_PROMPT = "Use run_code to compute the packaged worker smoke value."
CODE_WORKER_TEXT = "code worker smoke ok"
WORKFLOW_PROMPT = "Use workflow to compute the packaged worker smoke value without agents."
WORKFLOW_WORKER_TEXT = "workflow worker smoke ok"
SNAPSHOT_PROMPT = "Run the advanced packaged-runtime snapshot scenario."
SNAPSHOT_SESSION_ID = "advanced-executable"
SNAPSHOT_DIRECT_CHILD_PROMPT = "Reply with exactly DIRECT_CHILD_OK and nothing else."
SNAPSHOT_WORKFLOW_CHILD_PROMPT = "Reply with exactly WORKFLOW_CHILD_OK and nothing else."
SNAPSHOT_FINAL_TEXT = "ADVANCED_EXECUTABLE_OK"
SNAPSHOT_MOUNT_CODE = """\
return (ctx) => {
harness.registerTool(ctx, harness.defineTool({
name: 'snapshot_double',
description: 'Double a number for executable snapshot verification.',
parameters: { value: { type: 'number', required: true } },
async execute(args) {
return [{ type: 'text', text: String(args.value * 2) }]
}
}))
}
"""
SNAPSHOT_WORKFLOW_SCRIPT = (
"phase('Delegate')\n"
f"const reply = await agent('{SNAPSHOT_WORKFLOW_CHILD_PROMPT}', {{ label: 'workflow-child' }})\n"
"return { reply }"
)
SNAPSHOT_DIRECTORY = (
Path(__file__).resolve().parent / "snapshots" / "python-sdk-single-exe" / "advanced"
)
SNAPSHOT_FILENAMES = ("result.json", "session.jsonl", "session.1.jsonl", "session.2.jsonl")
CUSTOM_CORDIS = """\
- id: jsonrpc
name: '@deepseek-ai/dsh-jsonrpc'
@@ -41,17 +71,27 @@ CUSTOM_CORDIS = """\
name: '@deepseek-ai/dsh-code-runtime-worker'
- id: subagents
name: '@deepseek-ai/dsh-subagent'
- id: subagent-spawn
name: '@deepseek-ai/dsh-subagent-spawn'
config:
providerName: spawn
- id: subagent-tool
name: '@deepseek-ai/dsh-tool-subagent'
config:
provider: spawn
- id: workflow-engine
name: '@deepseek-ai/dsh-workflow-workerthread'
config:
provider: spawn
- id: workflow-tool
name: '@deepseek-ai/dsh-tool-workflow'
- id: cordis-tool
name: '@deepseek-ai/dsh-tool-cordis'
"""
class MockModelHandler(BaseHTTPRequestHandler):
"""Return deterministic text and worker-tool streaming completions."""
"""Return deterministic text, worker, and orchestration completions."""
requests: list[dict[str, object]] = []
@@ -82,8 +122,11 @@ def completion_chunks(body: dict[str, object]) -> list[dict[str, object]]:
raise AssertionError(f"model request has an invalid latest message: {body}")
if latest.get("role") == "tool":
tool_name = latest_tool_name(messages)
tool_text = json.dumps(latest.get("content"))
call_id, tool_name = latest_tool_call(messages)
tool_text = message_text(latest.get("content"))
advanced = advanced_tool_followup(body, call_id, tool_name, tool_text)
if advanced is not None:
return advanced
if "42" not in tool_text:
raise AssertionError(f"{tool_name} worker returned no expected value: {latest}")
if tool_name == "run_code":
@@ -93,6 +136,17 @@ def completion_chunks(body: dict[str, object]) -> list[dict[str, object]]:
raise AssertionError(f"unexpected tool follow-up: {tool_name}")
prompt = message_text(latest.get("content"))
if prompt == SNAPSHOT_DIRECT_CHILD_PROMPT:
return text_chunks("DIRECT_CHILD_OK")
if prompt == SNAPSHOT_WORKFLOW_CHILD_PROMPT:
return text_chunks("WORKFLOW_CHILD_OK")
if prompt == SNAPSHOT_PROMPT:
assert_advertised_tool(body, "cordis_mount")
return tool_call_chunks(
"advanced-mount",
"cordis_mount",
{"code": SNAPSHOT_MOUNT_CODE},
)
if prompt == CODE_PROMPT:
assert_advertised_tool(body, "run_code")
return tool_call_chunks("call-code-worker", "run_code", {"code": "return 6 * 7"})
@@ -112,6 +166,70 @@ def completion_chunks(body: dict[str, object]) -> list[dict[str, object]]:
return text_chunks(EXPECTED_TEXT)
def advanced_tool_followup(
body: dict[str, object],
call_id: str,
tool_name: str,
tool_text: str,
) -> list[dict[str, object]] | None:
"""Advance the executable snapshot's deterministic parent tool chain."""
if not call_id.startswith("advanced-"):
return None
if call_id == "advanced-mount" and tool_name == "cordis_mount":
if "mounted dyn-1" not in tool_text:
raise AssertionError(f"cordis_mount returned no mount id: {tool_text}")
assert_advertised_tool(body, "run_code")
assert_advertised_tool(body, "snapshot_double")
return tool_call_chunks(
"advanced-code",
"run_code",
{"code": "return await tools.snapshot_double({ value: 21 })"},
)
if call_id == "advanced-code" and tool_name == "run_code":
if "42" not in tool_text:
raise AssertionError(f"run_code returned no dynamic-tool value: {tool_text}")
assert_advertised_tool(body, "subagent")
return tool_call_chunks(
"advanced-direct-child",
"subagent",
{
"description": "Check direct child",
"prompt": SNAPSHOT_DIRECT_CHILD_PROMPT,
},
)
if call_id == "advanced-direct-child" and tool_name == "subagent":
if "DIRECT_CHILD_OK" not in tool_text:
raise AssertionError(f"subagent returned no expected child value: {tool_text}")
assert_advertised_tool(body, "workflow")
return tool_call_chunks(
"advanced-workflow",
"workflow",
{
"script": SNAPSHOT_WORKFLOW_SCRIPT,
"meta": {
"name": "advanced-exe-snapshot",
"description": "exercise one packaged workflow child",
},
},
)
if call_id == "advanced-workflow" and tool_name == "workflow":
if "WORKFLOW_CHILD_OK" not in tool_text:
raise AssertionError(f"workflow returned no expected child value: {tool_text}")
assert_advertised_tool(body, "cordis_unmount")
return tool_call_chunks(
"advanced-unmount",
"cordis_unmount",
{"id": "dyn-1"},
)
if call_id == "advanced-unmount" and tool_name == "cordis_unmount":
if "unmounted dyn-1" not in tool_text:
raise AssertionError(f"cordis_unmount returned no disposal result: {tool_text}")
if "snapshot_double" in advertised_tool_names(body):
raise AssertionError("snapshot_double remained advertised after cordis_unmount")
return text_chunks(SNAPSHOT_FINAL_TEXT)
raise AssertionError(f"unexpected advanced tool follow-up: {call_id} {tool_name}: {tool_text}")
def text_chunks(text: str) -> list[dict[str, object]]:
"""Build a complete streaming text response."""
return [
@@ -147,8 +265,8 @@ def tool_call_chunks(call_id: str, name: str, arguments: dict[str, object]) -> l
]
def latest_tool_name(messages: list[object]) -> str:
"""Find the assistant tool call paired with the latest tool result."""
def latest_tool_call(messages: list[object]) -> tuple[str, str]:
"""Find the assistant call id and name paired with the latest tool result."""
for message in reversed(messages[:-1]):
if not isinstance(message, dict):
continue
@@ -159,8 +277,13 @@ def latest_tool_name(messages: list[object]) -> str:
if not isinstance(call, dict):
continue
function = call.get("function")
if isinstance(function, dict) and isinstance(function.get("name"), str):
return function["name"]
call_id = call.get("id")
if (
isinstance(call_id, str)
and isinstance(function, dict)
and isinstance(function.get("name"), str)
):
return call_id, function["name"]
raise AssertionError(f"tool result has no preceding assistant tool call: {messages}")
@@ -177,8 +300,8 @@ def message_text(content: object) -> str:
return ""
def assert_advertised_tool(body: dict[str, object], expected: str) -> None:
"""Require the packaged deployment to expose the requested tool."""
def advertised_tool_names(body: dict[str, object]) -> set[str]:
"""Return the model-facing tool names advertised on one request."""
tools = body.get("tools")
if not isinstance(tools, list):
raise AssertionError(f"model request advertised no tools: {body}")
@@ -189,6 +312,12 @@ def assert_advertised_tool(body: dict[str, object], expected: str) -> None:
function = tool.get("function")
if isinstance(function, dict) and isinstance(function.get("name"), str):
names.add(function["name"])
return names
def assert_advertised_tool(body: dict[str, object], expected: str) -> None:
"""Require the packaged deployment to expose the requested tool."""
names = advertised_tool_names(body)
if expected not in names:
raise AssertionError(f"model request did not advertise {expected}: {names}")
@@ -211,11 +340,18 @@ class MockModel:
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--scenario", choices=("all", "sdk-default", "sdk-custom", "direct"), default="all")
parser.add_argument(
"--scenario",
choices=("all", "sdk-default", "sdk-custom", "sdk-snapshot", "direct"),
default="all",
)
parser.add_argument("--exe", type=Path)
parser.add_argument("--update-snapshots", action="store_true")
args = parser.parse_args()
if args.scenario in {"all", "sdk-custom", "direct"} and args.exe is None:
parser.error("--exe is required for custom and direct scenarios")
if args.scenario in {"all", "sdk-custom", "sdk-snapshot", "direct"} and args.exe is None:
parser.error("--exe is required for custom, snapshot, and direct scenarios")
if args.update_snapshots and args.scenario not in {"all", "sdk-snapshot"}:
parser.error("--update-snapshots requires --scenario sdk-snapshot or all")
if args.exe is not None and not args.exe.is_file():
parser.error(f"runtime executable does not exist: {args.exe}")
@@ -225,6 +361,9 @@ def main() -> None:
if args.scenario in {"all", "sdk-custom"}:
assert args.exe is not None
smoke_sdk_custom(model.url, args.exe.resolve())
if args.scenario in {"all", "sdk-snapshot"}:
assert args.exe is not None
smoke_sdk_snapshot(model.url, args.exe.resolve(), args.update_snapshots)
if args.scenario in {"all", "direct"}:
assert args.exe is not None
smoke_direct(model.url, args.exe.resolve())
@@ -283,6 +422,49 @@ def smoke_sdk_custom(base_url: str, executable: Path) -> None:
assert_session_log(sessions, root, EXPECTED_TEXT, CODE_WORKER_TEXT, WORKFLOW_WORKER_TEXT)
def smoke_sdk_snapshot(base_url: str, executable: Path, update_snapshots: bool) -> None:
"""Drive and compare the advanced SDK/executable behavioral snapshot."""
from deepseek_harness import DeepSeekHarness
with tempfile.TemporaryDirectory(prefix="dsh-sdk-snapshot-") as temporary:
root = Path(temporary).resolve()
sessions = root / "sessions"
cordis = root / "cordis.yml"
cordis.write_text(CUSTOM_CORDIS)
with DeepSeekHarness(
model="smoke-model",
cwd=str(root),
session_root=str(sessions),
cordis=str(cordis),
runtime_bin=str(executable),
api_key="sk-keyless-smoke",
base_url=base_url,
request_timeout_seconds=60,
) as harness:
result = harness.run(SNAPSHOT_PROMPT, session_id=SNAPSHOT_SESSION_ID)
assert result.status == "ok", result
assert result.final_response == SNAPSHOT_FINAL_TEXT, result.final_response
methods = [notification.method for notification in result.notifications]
if methods.count("subagent.started") != 2 or methods.count("subagent.finished") != 2:
raise AssertionError(f"advanced snapshot emitted unexpected subagent lifecycle: {methods}")
if not any(event.get("type") == "tool/code-dispatch" for event in result.events):
raise AssertionError("advanced snapshot emitted no tool/code-dispatch event")
logs = read_session_logs(sessions)
child_ids = snapshot_child_ids(result)
expected_ids = {SNAPSHOT_SESSION_ID, *child_ids}
if set(logs) != expected_ids:
raise AssertionError(f"advanced snapshot expected parent plus two child logs: {sorted(logs)}")
if "DIRECT_CHILD_OK" not in render_jsonl(logs[child_ids[0]]):
raise AssertionError("first advanced child log has no direct-subagent result")
if "WORKFLOW_CHILD_OK" not in render_jsonl(logs[child_ids[1]]):
raise AssertionError("second advanced child log has no workflow-subagent result")
files = build_snapshot_files(result, logs, child_ids, root)
compare_snapshot_files(files, update_snapshots)
def smoke_direct(base_url: str, executable: Path) -> None:
with tempfile.TemporaryDirectory(prefix="dsh-direct-") as temporary:
root = Path(temporary).resolve()
@@ -399,5 +581,214 @@ def assert_session_log(sessions: Path, cwd: Path, *expected_texts: str) -> None:
raise AssertionError(f"session log has no {expected!r} response: {logs[0]}")
def read_session_logs(sessions: Path) -> dict[str, list[dict[str, object]]]:
"""Parse every persisted JSONL session into a map keyed by header id."""
logs: dict[str, list[dict[str, object]]] = {}
for path in sorted(sessions.rglob("*.jsonl")):
records = [
json.loads(line)
for line in path.read_text(encoding="utf-8").splitlines()
if line
]
if not records or records[0].get("type") != "session":
raise AssertionError(f"session log has no header: {path}")
session_id = records[0].get("id")
if not isinstance(session_id, str):
raise AssertionError(f"session log header has no string id: {path}")
if session_id in logs:
raise AssertionError(f"duplicate persisted session id: {session_id}")
logs[session_id] = records
return logs
def snapshot_child_ids(result: "TurnResult") -> list[str]:
"""Return the two child session ids in their SDK notification order."""
child_ids: list[str] = []
for notification in result.notifications:
if notification.method != "subagent.started":
continue
payload = notification.payload
if payload.get("parentSessionId") != SNAPSHOT_SESSION_ID:
continue
child_id = payload.get("childSessionId")
if isinstance(child_id, str) and child_id not in child_ids:
child_ids.append(child_id)
if len(child_ids) != 2:
raise AssertionError(f"advanced snapshot expected two child session ids: {child_ids}")
return child_ids
def build_snapshot_files(
result: "TurnResult",
logs: dict[str, list[dict[str, object]]],
child_ids: list[str],
cwd: Path,
) -> dict[str, str]:
"""Render the SDK result and three persisted logs into stable goldens."""
replacements = [(str(cwd), "{{cwd}}"), (SNAPSHOT_SESSION_ID, "{{parent}}")]
for index, child_id in enumerate(child_ids, start=1):
replacements.append((child_id, f"{{{{child-{index}}}}}"))
agent_id = snapshot_agent_id(result, child_id)
replacements.append((agent_id, f"{{{{agent-{index}}}}}"))
replacements.sort(key=lambda pair: len(pair[0]), reverse=True)
result_value = {
"session_id": result.session_id,
"status": result.status,
"final_response": result.final_response,
"events": result.events,
"notifications": [
{"method": notification.method, "payload": notification.payload}
for notification in result.notifications
],
"session_root": result.session_root,
}
normalized_result = normalize_snapshot_value(result_value, replacements)
files = {
"result.json": json.dumps(normalized_result, indent=2, ensure_ascii=False) + "\n",
"session.jsonl": render_jsonl(
[normalize_snapshot_value(record, replacements) for record in logs[SNAPSHOT_SESSION_ID]]
),
}
for index, child_id in enumerate(child_ids, start=1):
files[f"session.{index}.jsonl"] = render_jsonl(
[normalize_snapshot_value(record, replacements) for record in logs[child_id]]
)
if tuple(files) != SNAPSHOT_FILENAMES:
raise AssertionError(f"advanced snapshot file set drifted: {tuple(files)}")
return files
def snapshot_agent_id(result: "TurnResult", child_id: str) -> str:
"""Find the successful subagent id paired with one child session."""
for notification in result.notifications:
if notification.method != "subagent.finished":
continue
payload = notification.payload
if payload.get("childSessionId") != child_id:
continue
if payload.get("provider") != "spawn" or payload.get("status") != "ok":
raise AssertionError(f"advanced child did not finish successfully: {payload}")
agent_id = payload.get("agentId")
if isinstance(agent_id, str):
return agent_id
raise AssertionError(f"advanced snapshot has no finished agent for child {child_id}")
def normalize_snapshot_value(
value: object,
replacements: list[tuple[str, str]],
) -> object:
"""Scrub volatile values and bulky request headers without losing behavior."""
if isinstance(value, str):
normalized = value
for actual, token in replacements:
normalized = normalized.replace(actual, token)
return normalized
if isinstance(value, list):
return [normalize_snapshot_value(item, replacements) for item in value]
if not isinstance(value, dict):
return value
normalized = {
key: normalize_snapshot_value(item, replacements)
for key, item in value.items()
}
if normalized.get("type") == "session" and "createdAt" in normalized:
normalized["createdAt"] = 0
if "seq" in normalized and "time" in normalized:
normalized["time"] = 0
scrub_snapshot_header(normalized)
return normalized
def scrub_snapshot_header(value: dict[object, object]) -> None:
"""Tokenize request-header bulk while retaining delta tool names."""
data = value.get("data")
if not isinstance(data, dict):
return
if value.get("type") == "request/header":
header = data.get("header")
if not isinstance(header, dict):
return
if "system" in header:
header["system"] = "{{system}}"
tools = header.get("tools")
if isinstance(tools, list):
header["tools"] = [
tool.get("name") if isinstance(tool, dict) else "{{tools}}"
for tool in tools
]
if isinstance(header.get("messagePrefix"), list):
header["messagePrefix"] = ["{{messagePrefix}}" for _ in header["messagePrefix"]]
return
if value.get("type") != "request/header-delta":
return
system = data.get("system")
if isinstance(system, dict) and isinstance(system.get("insert"), list):
system["insert"] = ["{{system}}" for _ in system["insert"]]
tools = data.get("tools")
if isinstance(tools, dict):
for key in ("added", "changed"):
if isinstance(tools.get(key), list):
tools[key] = [scrub_snapshot_tool_schema(tool) for tool in tools[key]]
if isinstance(data.get("messagePrefix"), list):
data["messagePrefix"] = ["{{messagePrefix}}" for _ in data["messagePrefix"]]
def scrub_snapshot_tool_schema(value: object) -> object:
"""Keep a changed tool's name while tokenizing its schema bulk."""
if not isinstance(value, dict):
return value
return {
key: item if key == "name" else "{{tools}}"
for key, item in value.items()
}
def render_jsonl(records: list[object]) -> str:
"""Render parsed JSON values as compact, newline-terminated JSONL."""
return "".join(
json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n"
for record in records
)
def compare_snapshot_files(files: dict[str, str], update: bool) -> None:
"""Write or exactly compare the advanced executable snapshot files."""
if update:
SNAPSHOT_DIRECTORY.mkdir(parents=True, exist_ok=True)
for name, content in files.items():
(SNAPSHOT_DIRECTORY / name).write_text(content, encoding="utf-8")
print(f"smoke-python-runtime: updated snapshots in {SNAPSHOT_DIRECTORY}")
existing = {
path.name
for path in SNAPSHOT_DIRECTORY.iterdir()
if path.is_file()
} if SNAPSHOT_DIRECTORY.is_dir() else set()
expected = set(SNAPSHOT_FILENAMES)
if existing != expected:
raise AssertionError(
"advanced snapshot files differ: "
f"missing={sorted(expected - existing)}, unexpected={sorted(existing - expected)}"
)
for name, actual in files.items():
expected_text = (SNAPSHOT_DIRECTORY / name).read_text(encoding="utf-8")
if actual == expected_text:
continue
diff = "".join(difflib.unified_diff(
expected_text.splitlines(keepends=True),
actual.splitlines(keepends=True),
fromfile=f"expected/{name}",
tofile=f"actual/{name}",
))
raise AssertionError(
f"advanced executable snapshot mismatch in {name}; "
"rerun with --update-snapshots after reviewing the behavior\n"
f"{diff}"
)
if __name__ == "__main__":
main()