"""DSH headless 适配器的离线合同测试。""" from __future__ import annotations import json import pathlib import tempfile import unittest from jsonschema import Draft202012Validator ROOT = next( parent for parent in (pathlib.Path(__file__).resolve().parent, *pathlib.Path(__file__).resolve().parents) if (parent / "AGENTS.md").is_file() and (parent / ".git").exists() ) if str(ROOT) not in __import__("sys").path: __import__("sys").path.insert(0, str(ROOT)) from framework.adapters.dsh.normalization import ( # noqa: E402 DshNormalizationError, normalize_dsh_session, ) from framework.adapters.dsh.runner import ( # noqa: E402 DshError, DshExecutionPolicy, DshHeadlessRunner, DshProcessResult, build_dsh_argv, build_dsh_patch, ) from framework.primitives.artifacts import read_jsonl # noqa: E402 from framework.primitives.execution import FrameworkExecutionRequest # noqa: E402 SESSION_LINES = [ { "type": "session", "version": 0, "id": "session-test-1", "createdAt": 1, "cwd": "/tmp/work", "delegationDepth": 0, }, { "type": "request/header", "data": { "header": {"config": {"provider": "probe", "model": "probe-model"}}, "reason": "initial", }, }, {"type": "request/context", "data": {"provider": "probe", "model": "probe-model"}}, {"type": "turn/start", "data": {"turn": 1}}, {"type": "step/start", "data": {"turn": 1, "step": 1}}, { "type": "user/message", "data": { "content": [{"type": "text", "text": "task"}], "source": {"kind": "user"}, "role": "user", }, }, { "type": "assistant/chunk", "data": { "turn": 1, "step": 1, "chunk": {"type": "finish", "reason": {"kind": "stop"}}, }, }, { "type": "assistant/message", "data": { "turn": 1, "step": 1, "message": { "role": "assistant", "content": [{"type": "text", "text": '{"ok":true}'}], "source": {"kind": "model", "provider": "probe", "model": "probe-model"}, }, "usage": {"inputTokens": 4, "outputTokens": 2, "costUsd": 0.01}, }, }, {"type": "tool/call", "data": {"turn": 1, "step": 1, "callId": "c1", "name": "read", "arguments": '{"path":"x"}'}}, { "type": "tool/result", "data": { "turn": 1, "step": 1, "message": { "source": {"kind": "tool", "callId": "c1"}, "content": [{"type": "tool-result", "content": [{"type": "text", "text": "ok"}], "isError": False}], }, }, }, {"type": "step/end", "data": {"turn": 1, "step": 1}}, {"type": "turn/end", "data": {"turn": 1, "reason": {"kind": "completed"}}}, {"type": "future/new-event", "data": {"opaque": True}}, ] class RecordingSink: def __init__(self) -> None: self.rows: list[tuple[str, dict]] = [] def emit(self, event_type: str, **kwargs): self.rows.append((event_type, kwargs)) return len(self.rows) class DshAdapterTest(unittest.TestCase): def request(self, *, tools: tuple[str, ...] = (), session_mode: str = "fresh") -> FrameworkExecutionRequest: return FrameworkExecutionRequest( system_prompt="system", user_content="task", tool_allowlist=tools, session_mode=session_mode, ) def write_session(self, root: pathlib.Path) -> pathlib.Path: path = root / "--tmp-work--" / "session-test-1" / "session.jsonl" path.parent.mkdir(parents=True) path.write_text( "".join(json.dumps(line, ensure_ascii=False) + "\n" for line in SESSION_LINES), encoding="utf-8", ) return path def test_normalize_session_preserves_unknown_and_builds_artifact(self): with tempfile.TemporaryDirectory() as directory: root = pathlib.Path(directory) source = self.write_session(root / "source") artifact = root / "framework-events.jsonl" outcome = normalize_dsh_session(source, artifact, run_id="run-1", framework_version="0.1.1-rc.2") self.assertEqual(outcome.session_id, "session-test-1") self.assertEqual(outcome.final_text, '{"ok":true}') self.assertEqual(outcome.model_calls[0].actual_model_id, "probe/probe-model") self.assertEqual(outcome.model_calls[0].cost_usd, 0.01) self.assertEqual(outcome.tool_calls[0].args, {"path": "x"}) self.assertEqual(outcome.turns, 1) self.assertEqual(outcome.unknown_event_types, ("future/new-event",)) events = read_jsonl(artifact) self.assertEqual(len(events), len(SESSION_LINES)) self.assertEqual(events[-1]["kind"], "unknown") self.assertEqual(events[-1]["runId"], "run-1") self.assertTrue(outcome.artifact_sha256.startswith("sha256:")) def test_compressed_session_fails_closed(self): with tempfile.TemporaryDirectory() as directory: root = pathlib.Path(directory) source = root / "session.jsonl.zstd" source.write_bytes(b"not used") with self.assertRaisesRegex(DshNormalizationError, "COMPRESSED_ARTIFACT_UNSUPPORTED"): normalize_dsh_session(source, root / "events.jsonl") def test_patch_and_argv_are_explicit_and_toolless(self): policy = DshExecutionPolicy(provider="probe", model="probe-model", cwd="/tmp/work") patch = build_dsh_patch(self.request(), policy, session_root="/tmp/session-root") rows = {row["id"]: row for row in patch} self.assertEqual(rows["agent-default-model"]["config"], {"provider": "probe", "model": "probe-model"}) self.assertEqual(rows["session-persistence-jsonl"]["config"]["compression"], "none") self.assertTrue(rows["tool-fs"]["disabled"]) argv = build_dsh_argv(self.request(), policy, "/tmp/patch.json") self.assertEqual(argv[:4], ["dsh", "--profile", "headless", "--patch"]) self.assertEqual(pathlib.Path(argv[4]).name, "patch.json") self.assertEqual(argv[-2], "--") self.assertEqual(argv[-1], "task") def test_nonempty_tools_and_continue_fail_before_launch(self): policy = DshExecutionPolicy(provider="p", model="m") with self.assertRaises(DshError) as tool_error: build_dsh_argv(self.request(tools=("read",)), policy, "/tmp/p.json") self.assertEqual(tool_error.exception.error_code, "DSH_TOOL_POLICY_UNSUPPORTED") with self.assertRaises(DshError) as session_error: build_dsh_argv(self.request(session_mode="continue"), policy, "/tmp/p.json") self.assertEqual(session_error.exception.error_code, "DSH_SESSION_CONTINUE_UNSUPPORTED") def test_runner_uses_flushed_session_as_result(self): with tempfile.TemporaryDirectory() as directory: root = pathlib.Path(directory) def launcher(argv, _timeout, _cwd, _env): patch_path = pathlib.Path(argv[argv.index("--patch") + 1]) patch = json.loads(patch_path.read_text(encoding="utf-8")) session_root = pathlib.Path( next(row for row in patch if row["id"] == "session-persistence-jsonl")["config"]["root"] ) self.write_session(session_root) return DshProcessResult(returncode=0, stdout=b"ignored stdout\n") sink = RecordingSink() outcome = DshHeadlessRunner(launcher=launcher, framework_version="0.1.1-rc.2").run( self.request(), DshExecutionPolicy(provider="probe", model="probe-model", cwd="/tmp/work"), sink, artifact_dir=root / "run", run_id="run-1", ) self.assertEqual(outcome.final_text, '{"ok":true}') self.assertEqual(outcome.session_id, "session-test-1") self.assertEqual([row[0] for row in sink.rows], [ "agent.started", "model.completed", "tool.completed", "agent.completed" ]) result = outcome.as_framework_result().as_dict() schema = json.loads((ROOT / "framework/primitives/schemas/framework_execution_result.schema.json").read_text(encoding="utf-8")) Draft202012Validator(schema).validate(result) self.assertEqual(result["status"], "completed") self.assertEqual(result["requestedModel"], "probe/probe-model") self.assertTrue((root / "run/framework-events.jsonl").is_file()) def test_runner_propagates_process_failure(self): with tempfile.TemporaryDirectory() as directory: sink = RecordingSink() def launcher(_argv, _timeout, _cwd, _env): return DshProcessResult(returncode=1, stderr=b"MISSING_CREDENTIAL: sk-secret-token-123456") with self.assertRaises(DshError) as context: DshHeadlessRunner(launcher=launcher).run( self.request(), DshExecutionPolicy(provider="probe", model="probe-model"), sink, artifact_dir=directory, ) self.assertEqual(context.exception.error_code, "DSH_EXIT_NONZERO") self.assertNotIn("sk-secret", str(context.exception)) if __name__ == "__main__": unittest.main(verbosity=2)