import json from app.journey import build_journey from app.narrator import narrate_contract def _contract(task_id: str, task_kind: str, goal_ids: list[str]) -> dict: return { "task_id": task_id, "contract_ref": f"script-build://task-contracts/sha256/{task_id}", "task_kind": task_kind, "objective": "组织目标并形成可执行的内容结构", "goal_ids": goal_ids, "input_decision_refs": [ { "decision_id": "direction-accepted", "expected_task_kind": "direction", "artifact_ref": { "uri": "script-build://artifact-versions/9", "kind": "direction", }, } ], "base_artifact_ref": None, "criteria": [ {"criterion_id": "complete", "description": "结构覆盖所有核心目标", "hard": True} ], } def test_contract_narrator_uses_only_frozen_contract_facts() -> None: narration = narrate_contract(_contract("task-1", "structure", ["goal-1", "goal-2"])) assert narration.action == "创建整体结构 Task" assert narration.goals == ["goal-1", "goal-2"] assert narration.inputs == ["已接受的创作方向"] assert narration.output == "整体结构" assert narration.criteria == ["结构覆盖所有核心目标"] def test_journey_links_real_lifecycle_and_marks_retry_and_parallel_work() -> None: source = { "snapshot": { "root_trace_id": "root-7", "root_objective": "为选题生成一份可发布的脚本", "tasks": [ { "task_id": "root-task", "parent_task_id": None, "status": "running", "created_at": "2026-07-21T01:00:00+00:00", "current_spec": {"objective": "root", "context_refs": []}, "child_task_ids": ["task-1", "task-2"], }, { "task_id": "task-1", "parent_task_id": "root-task", "status": "needs_replan", "created_at": "2026-07-21T01:01:00+00:00", "current_spec": {"objective": "structure", "context_refs": []}, "child_task_ids": [], }, { "task_id": "task-2", "parent_task_id": "root-task", "status": "running", "created_at": "2026-07-21T01:01:00+00:00", "current_spec": {"objective": "paragraph", "context_refs": []}, "child_task_ids": [], }, ], "attempts": [ { "attempt_id": "attempt-1", "task_id": "task-1", "worker_trace_id": "worker-1", "status": "submitted", "operation_id": "operation-1", "started_at": "2026-07-21T01:02:00+00:00", "duration_ms": 2000, "execution_stats": {"primary_model": "qwen", "total_tokens": 120}, "submission": { "artifact_refs": [ { "uri": "script-build://artifact-versions/11", "kind": "structure", } ] }, }, { "attempt_id": "attempt-2", "task_id": "task-2", "worker_trace_id": "worker-2", "status": "running", "operation_id": "operation-1", "started_at": "2026-07-21T01:02:00+00:00", "execution_stats": {}, "submission": None, }, ], "validations": [ { "validation_id": "validation-1", "task_id": "task-1", "attempt_id": "attempt-1", "status": "completed", "verdict": "failed", "started_at": "2026-07-21T01:03:00+00:00", "duration_ms": 500, "validation_plan": { "mode": "deterministic", "preflight_rule_ids": ["goal-coverage"], }, "criterion_results": [ { "criterion_id": "complete", "verdict": "failed", "reason": "缺少 goal-2", } ], "execution_stats": {"total_tokens": 0}, } ], "decisions": [ { "decision_id": "decision-1", "task_id": "task-1", "attempt_id": "attempt-1", "validation_id": "validation-1", "action": "retry", "reason": "补齐 goal-2 后重新生成", "created_at": "2026-07-21T01:04:00+00:00", } ], "operations": [ { "operation_id": "operation-1", "attempt_ids": ["attempt-1", "attempt-2"], } ], }, "input_summary": { "topic": "一次反转如何改变信念", "persona": {"account_name": "每天心理学", "point_count": 3}, "strategies": [{"name": "先冲突后解法"}], "input_snapshot_id": "11", "input_digest": "sha256:input", }, "contracts": { "task-1": _contract("task-1", "structure", ["goal-1", "goal-2"]), "task-2": _contract("task-2", "paragraph", ["goal-2"]), }, "messages": { "root-7": [ { "role": "assistant", "sequence": 1, "content": { "text": "当前缺少整体结构，需要组织目标。", "tool_calls": [ { "id": "call-1", "function": { "name": "plan_script_tasks", "arguments": json.dumps( {"contracts": [{"task_kind": "structure"}]} ), }, } ], }, }, { "role": "tool", "tool_call_id": "call-1", "content": {"result": json.dumps({"task_ids": ["task-1", "task-2"]})}, }, ], "worker-1": [ { "role": "assistant", "content": { "text": "我先建立结构骨架。", "tool_calls": [ {"function": {"name": "create_script_structure", "arguments": "{}"}} ], }, } ], "worker-2": [], }, "events": [ {"sequence": 10, "payload": {"task_ids": ["task-1", "task-2"]}}, {"sequence": 20, "payload": {"attempt_id": "attempt-1"}}, {"sequence": 21, "payload": {"attempt_id": "attempt-2"}}, {"sequence": 30, "payload": {"validation_id": "validation-1"}}, {"sequence": 40, "payload": {"decision_id": "decision-1"}}, ], } journey = build_journey(7, source) assert [step.step_type for step in journey.steps].count("plan") == 2 execute = next(step for step in journey.steps if step.step_id == "execute:attempt-1") assert execute.parallel_total == 2 assert execute.reasoning_source == "observable" validation = next(step for step in journey.steps if step.step_type == "validate") assert "缺少 goal-2" in validation.reasoning loop = next(edge for edge in journey.edges if edge.relation == "loop") assert loop.source == "decide:decision-1" assert loop.target == "execute:attempt-1" assert any(edge.relation == "validates" for edge in journey.edges) assert [step.sequence for step in journey.steps] == list(range(1, len(journey.steps) + 1))