from __future__ import annotations import asyncio import json from dataclasses import asdict, dataclass from types import SimpleNamespace from typing import Any import pytest from agent.orchestration import ( AgentRole, ArtifactRef, DecisionAction, RoleContextRequest, TaskStatus, ) from script_build_host.application.context_semantics import AdaptiveSemanticSelector from script_build_host.application.mission_workbench import ( MissionWorkbench, WorkbenchError, required_context_source_types, ) from script_build_host.domain.context_broker import ContextCard, estimate_context_tokens from script_build_host.domain.task_contracts import ( AcceptedDecisionRef, ScriptCriterion, ScriptIntentClass, ScriptTaskBudget, ScriptTaskContractV1, ScriptTaskKind, ) class _TaskStore: def __init__(self, ledger: Any) -> None: self.ledger = ledger async def load(self, _root_trace_id: str) -> Any: return self.ledger @pytest.mark.parametrize( ("kind", "required"), [ (ScriptTaskKind.PATTERN_RETRIEVAL, {"topic"}), (ScriptTaskKind.DECODE_RETRIEVAL, {"topic"}), (ScriptTaskKind.EXTERNAL_RETRIEVAL, {"topic"}), (ScriptTaskKind.KNOWLEDGE_RETRIEVAL, {"topic"}), ( ScriptTaskKind.DIRECTION, {"topic", "persona", "section_pattern", "strategy"}, ), (ScriptTaskKind.STRUCTURE, {"direction", "section_pattern"}), (ScriptTaskKind.PARAGRAPH, {"direction", "structure", "persona"}), (ScriptTaskKind.ELEMENT_SET, {"paragraph"}), (ScriptTaskKind.COMPARE, set()), ( ScriptTaskKind.ROOT_DELIVERY, {"direction", "candidate-portfolio", "structured-script"}, ), ], ) def test_every_e2e_role_has_one_authoritative_context_source_policy( kind: ScriptTaskKind, required: set[str] ) -> None: assert set(required_context_source_types(kind)) == required class _Bindings: async def get_by_root(self, root_trace_id: str) -> Any: return SimpleNamespace( root_trace_id=root_trace_id, script_build_id=900049, input_snapshot_id=3, active_direction_artifact_version_id=None, ) class _Snapshots: def __init__(self) -> None: self.value = SimpleNamespace( snapshot_id="3", topic_id=49, topic={"topic": {"result": "健身房中的职场反转故事"}}, account={"account_name": "fixture"}, persona_points=tuple( { "列": ("实质", "形式", "作用", "感受")[index % 4], "点名称": f"persona-{index}", "人设权重分": 1 - index / 100, "帖子覆盖率": 0.8, } for index in range(80) ), section_patterns=tuple( {"模式名称": f"section-{index}", "说明": "转折节奏"} for index in range(40) ), strategies=( {"name": "must", "mode": "always_on", "content": "始终使用可视化叙事"}, {"name": "optional", "mode": "on_demand", "content": "按需策略" * 2_000}, ), model_manifest={"presets": {"script_context_broker": {"model": "fake"}}}, ) async def get(self, _snapshot_id: str, *, script_build_id: int) -> Any: assert script_build_id == 900049 return self.value class _Contracts: def __init__(self, values: dict[str, ScriptTaskContractV1]) -> None: self.values = values async def read(self, _root: str, ref: str) -> Any: return SimpleNamespace(contract=self.values[ref]) @dataclass class _Artifact: text: str raw_artifact_ref: str | None = None def content_payload(self) -> dict[str, Any]: return {"schema_version": "fixture", "full_description": self.text} class _Artifacts: def __init__(self, values: dict[str, _Artifact]) -> None: self.values = values async def read_by_ref(self, ref: ArtifactRef, *, script_build_id: int, **_: Any) -> Any: assert script_build_id == 900049 return SimpleNamespace(artifact=self.values[ref.uri]) class _RawStore: def __init__(self, raw_ref: str, payload: dict[str, Any]) -> None: self.raw_ref = raw_ref self.content = json.dumps(payload, ensure_ascii=False).encode() self.reads = 0 async def read_bytes(self, ref: str) -> bytes: assert ref == self.raw_ref self.reads += 1 return self.content class _ReceiptStore: def __init__(self) -> None: self.events: list[dict[str, Any]] = [] async def append_event(self, _trace_id: str, event: str, payload: dict[str, Any]) -> int: event_id = len(self.events) + 1 self.events.append({"event_id": event_id, "event": event, **payload}) return event_id async def get_events(self, _trace_id: str) -> list[dict[str, Any]]: return list(self.events) def _contract(kind: ScriptTaskKind, index: int) -> ScriptTaskContractV1: return ScriptTaskContractV1( task_kind=kind, scope_ref=f"script-build://scopes/full/part-{index}", intent_class=( ScriptIntentClass.EXPLORE if kind is not ScriptTaskKind.PATTERN_RETRIEVAL else ScriptIntentClass.EXPAND ), objective=f"task objective {index}", input_decision_refs=(), base_artifact_ref=None, write_scope=(f"script-build://writes/part-{index}",), gap_ref=None, output_schema=( "evidence-record/v1" if kind is ScriptTaskKind.PATTERN_RETRIEVAL else "paragraph-artifact/v1" ), criteria=(ScriptCriterion(f"c-{index}", "quality"),), budget=ScriptTaskBudget(), goal_ids=(f"g-{index % 6}",) if kind is not ScriptTaskKind.PATTERN_RETRIEVAL else (), compiler_manifest_digest="sha256:" + "a" * 64, ) def _fixture() -> tuple[MissionWorkbench, Any, _Snapshots, str]: root = SimpleNamespace( task_id="root-task", parent_task_id=None, child_task_ids=[], display_path="1", status=TaskStatus.RUNNING, blocked_reason=None, current_spec=SimpleNamespace(objective="build complete script", context_refs=()), current_spec_version=1, attempt_ids=[], decision_ids=[], ) tasks = {"root-task": root} attempts: dict[str, Any] = {} decisions: dict[str, Any] = {} contracts: dict[str, ScriptTaskContractV1] = {} artifacts: dict[str, _Artifact] = {} for index in range(58): task_id = f"task-{index}" contract_ref = f"script-build://task-contracts/sha256/{index:064x}" accepted = index < 33 attempt_id = f"attempt-{index}" decision_id = f"decision-{index}" artifact_uri = f"script-build://artifact-versions/{index + 1}" task = SimpleNamespace( task_id=task_id, parent_task_id="root-task", child_task_ids=[], display_path=f"1.{index + 1}", status=TaskStatus.COMPLETED if accepted else TaskStatus.PENDING, blocked_reason=None, current_spec=SimpleNamespace( objective=f"objective {index}", context_refs=("script-build://task-kinds/paragraph", contract_ref), ), current_spec_version=1, attempt_ids=[attempt_id] if accepted else [], decision_ids=[decision_id] if accepted else [], ) tasks[task_id] = task contracts[contract_ref] = _contract(ScriptTaskKind.PARAGRAPH, index) if accepted: ref = ArtifactRef( uri=artifact_uri, kind="paragraph", version=str(index + 1), digest="sha256:" + f"{index + 1:064x}", ) attempts[attempt_id] = SimpleNamespace( task_id=task_id, submission=SimpleNamespace(artifact_refs=(ref,), evidence_refs=()), failure=None, status=SimpleNamespace(value="succeeded"), ) decisions[decision_id] = SimpleNamespace( decision_id=decision_id, task_id=task_id, action=DecisionAction.ACCEPT, attempt_id=attempt_id, reason=f"FORGED REASON {index}", ) artifacts[artifact_uri] = _Artifact(f"REAL ARTIFACT CONTENT {index}") raw_ref = "script-build://raw-artifacts/sha256/" + "b" * 64 pattern_index = 58 task_id = "pattern-task" contract_ref = "script-build://task-contracts/sha256/" + "f" * 64 attempt_id = "pattern-attempt" decision_id = "pattern-decision" evidence_ref = ArtifactRef( uri="script-build://artifact-versions/100", kind="evidence", version="100", digest="sha256:" + "c" * 64, ) tasks[task_id] = SimpleNamespace( task_id=task_id, parent_task_id="root-task", child_task_ids=[], display_path="1.59", status=TaskStatus.COMPLETED, blocked_reason=None, current_spec=SimpleNamespace( objective="retrieve patterns", context_refs=("script-build://task-kinds/pattern-retrieval", contract_ref), ), current_spec_version=1, attempt_ids=[attempt_id], decision_ids=[decision_id], ) contracts[contract_ref] = _contract(ScriptTaskKind.PATTERN_RETRIEVAL, pattern_index) attempts[attempt_id] = SimpleNamespace( task_id=task_id, submission=SimpleNamespace(artifact_refs=(), evidence_refs=(evidence_ref,)), failure=None, status=SimpleNamespace(value="succeeded"), ) decisions[decision_id] = SimpleNamespace( decision_id=decision_id, task_id=task_id, action=DecisionAction.ACCEPT, attempt_id=attempt_id, reason="pattern accepted", ) artifacts[evidence_ref.uri] = _Artifact("pattern summary", raw_artifact_ref=raw_ref) ledger = SimpleNamespace( revision=49, root_task_id="root-task", focused_task_id="task-57", tasks=tasks, attempts=attempts, decisions=decisions, operations={}, ) patterns = { "candidates": [ {"name": f"pattern-{index}", "content": "模式详情" * 350} for index in range(193) ] } snapshots = _Snapshots() workbench = MissionWorkbench( task_store=_TaskStore(ledger), bindings=_Bindings(), snapshots=snapshots, artifacts=_Artifacts(artifacts), contracts=_Contracts(contracts), raw_artifacts=_RawStore(raw_ref, patterns), ) return workbench, ledger, snapshots, raw_ref @pytest.mark.asyncio async def test_planner_hot_context_is_bounded_and_every_item_is_pageable() -> None: workbench, _ledger, _snapshots, _ = _fixture() state = await workbench.planner_state("root") assert state.active_total == 26 assert state.active_has_more is True assert state.active_subgraph[0]["task_id"] == "task-57" assert state.accepted_total == 34 encoded = json.dumps(state.to_payload(), ensure_ascii=False) assert len(encoded) < 24_000 assert "FORGED REASON" not in " ".join(str(item["summary"]) for item in state.accepted_frontier) assert any("REAL ARTIFACT CONTENT" in str(item["summary"]) for item in state.accepted_frontier) cursor = None found: list[str] = [] while True: page = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="accepted_decisions", cursor=cursor, ) found.extend(item["handle"] for item in page["page"]["items"]) cursor = page["page"]["next_cursor"] if cursor is None: break assert len(found) == 34 assert len(set(found)) == 34 @pytest.mark.asyncio async def test_duplicate_long_inputs_have_unique_handles_and_budgeted_finite_pagination() -> None: workbench, _ledger, snapshots, _ = _fixture() snapshots.value.persona_points = tuple( {"列": "实质", "点名称": "相同人设" * 350, "人设权重分": 1, "帖子覆盖率": 1} for _ in range(30) ) cursor = None handles: list[str] = [] while True: page = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="inputs", source_types=("persona",), page_size=20, cursor=cursor, ) assert page["receipt"]["estimated_tokens"] <= 2_000 handles.extend(item["handle"] for item in page["page"]["items"]) cursor = page["page"]["next_cursor"] if cursor is None: break assert len(handles) == len(set(handles)) == 30 @pytest.mark.asyncio async def test_role_selected_persona_handles_keep_snapshot_identity_and_are_readable() -> None: workbench, ledger, _snapshots, _ = _fixture() state = await workbench.worker_state( RoleContextRequest( role=AgentRole.WORKER, root_trace_id="root", task_id="task-57", spec_version=1, task_spec={}, attempt_id="persona-attempt", ledger_revision=ledger.revision, ) ) persona_handles = [ item["handle"] for item in state.context_bundle["cards"] if item["source_type"] == "persona" ] assert len(persona_handles) == 12 for handle in persona_handles: page = await workbench.read_mission_context( root_trace_id="root", role="worker", task_id="task-57", attempt_id="persona-attempt", handle=handle, ) assert page["source_handle"] == handle assert page["content_fragment"] @pytest.mark.asyncio async def test_cursor_revision_tamper_and_cross_root_are_rejected() -> None: workbench, ledger, _snapshots, _ = _fixture() first = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="tasks", page_size=5 ) cursor = first["page"]["next_cursor"] assert cursor with pytest.raises(WorkbenchError, match="invalid context cursor"): await workbench.search_mission_context( root_trace_id="root", role="planner", collection="tasks", cursor=cursor[:-1] + ("A" if cursor[-1] != "A" else "B"), page_size=5, ) with pytest.raises(WorkbenchError) as cross_root: await workbench.search_mission_context( root_trace_id="other-root", role="planner", collection="tasks", cursor=cursor, page_size=5, ) assert cross_root.value.code == "CONTEXT_CURSOR_SCOPE_MISMATCH" ledger.revision += 1 with pytest.raises(WorkbenchError) as stale: await workbench.search_mission_context( root_trace_id="root", role="planner", collection="tasks", cursor=cursor, page_size=5, ) assert stale.value.code == "STALE_WORKBENCH_STATE" @pytest.mark.asyncio async def test_inputs_and_193_patterns_are_bounded_but_fully_recoverable() -> None: workbench, ledger, _snapshots, _ = _fixture() inputs = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="inputs", query="optional" ) strategy = next(item for item in inputs["page"]["items"] if item["source_type"] == "strategy") detail = await workbench.read_mission_context( root_trace_id="root", role="planner", handle=strategy["handle"] ) assert estimate_context_tokens(detail) <= 2_000 assert isinstance(detail["next_cursor"], str) with pytest.raises(WorkbenchError) as tampered_detail: await workbench.read_mission_context( root_trace_id="root", role="planner", handle=strategy["handle"], cursor=detail["next_cursor"][:-1] + "x", ) assert tampered_detail.value.code == "CONTEXT_CURSOR_INVALID" fragments = [detail["content_fragment"]] while detail["next_cursor"] is not None: detail = await workbench.read_mission_context( root_trace_id="root", role="planner", handle=strategy["handle"], cursor=detail["next_cursor"], ) fragments.append(detail["content_fragment"]) assert "按需策略" in "".join(fragments) cursor = None immutable_cursor = None found: list[dict[str, Any]] = [] while True: page = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="patterns", cursor=cursor, page_size=20, ) found.extend(page["page"]["items"]) cursor = page["page"]["next_cursor"] immutable_cursor = immutable_cursor or cursor if cursor is None: break assert len(found) == 193 assert len({item["handle"] for item in found}) == 193 raw_prompt = json.dumps( (await workbench.planner_state("root")).to_payload(), ensure_ascii=False ) assert len(raw_prompt) < 24_000 assert "模式详情" * 20 not in raw_prompt ledger.revision += 1 immutable_page = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="patterns", cursor=immutable_cursor, page_size=20, ) assert immutable_page["page"]["total"] == 193 assert workbench.broker._raw_artifacts.reads == 1 @pytest.mark.asyncio async def test_semantic_selector_cannot_escape_allowlist(tmp_path) -> None: calls = 0 async def llm_call(**_: Any) -> dict[str, Any]: nonlocal calls calls += 1 return {"content": json.dumps({"handles": ["bad_handle"]})} selector = AdaptiveSemanticSelector(llm_call=llm_call, cache_root=tmp_path) workbench, ledger, snapshots, _ = _fixture() workbench = MissionWorkbench( task_store=_TaskStore(ledger), bindings=_Bindings(), snapshots=snapshots, artifacts=workbench.broker._artifacts, contracts=workbench.broker._contracts, raw_artifacts=workbench.broker._raw_artifacts, semantic_selector=selector, ) result = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="inputs", query="not present" ) handles = {item["handle"] for item in result["page"]["items"]} assert "bad_handle" not in handles assert result["receipt"]["semantic_calls"] == 1 assert result["receipt"]["fallback_reason"] assert calls == 1 @pytest.mark.asyncio async def test_semantic_selector_caches_valid_allowlisted_order(tmp_path) -> None: calls = 0 async def llm_call(**kwargs: Any) -> dict[str, Any]: nonlocal calls calls += 1 request = json.loads(kwargs["messages"][1]["content"]) handles = [item["handle"] for item in request["candidates"]] return {"content": json.dumps({"handles": list(reversed(handles))})} selector = AdaptiveSemanticSelector(llm_call=llm_call, cache_root=tmp_path) cards = ( ContextCard(handle="one", source_type="input", summary="alpha"), ContextCard(handle="two", source_type="input", summary="beta"), ) first = await selector.select( root_trace_id="root", query="similar", cards=cards, model_config={"model": "fake"} ) second = await selector.select( root_trace_id="root", query="similar", cards=cards, model_config={"model": "fake"} ) assert first.handles == ("two", "one") assert second.handles == first.handles assert second.cache_hit is True assert calls == 1 @pytest.mark.asyncio async def test_semantic_cache_survives_call_limit_and_timeout_falls_back(tmp_path) -> None: calls = 0 async def llm_call(**_: Any) -> dict[str, Any]: nonlocal calls calls += 1 return {"content": json.dumps({"handles": ["one", "two"]})} cards = ( ContextCard(handle="one", source_type="input", summary="alpha"), ContextCard(handle="two", source_type="input", summary="beta"), ) selector = AdaptiveSemanticSelector( llm_call=llm_call, cache_root=tmp_path, mission_call_limit=1 ) first = await selector.select( root_trace_id="root", query="same", cards=cards, model_config={"model": "fake"} ) restarted_selector = AdaptiveSemanticSelector( llm_call=llm_call, cache_root=tmp_path, mission_call_limit=1 ) cached = await restarted_selector.select( root_trace_id="root", query="same", cards=cards, model_config={"model": "fake"} ) limited = await restarted_selector.select( root_trace_id="root", query="different", cards=cards, model_config={"model": "fake"} ) assert first.calls == 1 assert cached.cache_hit is True assert limited.fallback_reason == "semantic_call_limit" assert calls == 1 async def slow_call(**_: Any) -> dict[str, Any]: await asyncio.sleep(0.05) return {"content": "{}"} timeout_selector = AdaptiveSemanticSelector( llm_call=slow_call, cache_root=tmp_path / "timeout", timeout_seconds=0.001, ) timed_out = await timeout_selector.select( root_trace_id="timeout-root", query="slow", cards=cards, model_config={"model": "fake"}, ) assert timed_out.calls == 1 assert timed_out.fallback_reason == "TimeoutError:semantic_fallback" @pytest.mark.asyncio async def test_semantic_rerank_never_removes_candidates_beyond_its_window(tmp_path) -> None: semantic_first: list[str] = [] async def llm_call(**kwargs: Any) -> dict[str, Any]: request = json.loads(kwargs["messages"][1]["content"]) handles = [item["handle"] for item in request["candidates"]] semantic_first[:] = [handles[-1]] return {"content": json.dumps({"handles": list(reversed(handles))})} selector = AdaptiveSemanticSelector(llm_call=llm_call, cache_root=tmp_path) workbench, ledger, snapshots, _ = _fixture() workbench = MissionWorkbench( task_store=_TaskStore(ledger), bindings=_Bindings(), snapshots=snapshots, artifacts=workbench.broker._artifacts, contracts=workbench.broker._contracts, raw_artifacts=workbench.broker._raw_artifacts, semantic_selector=selector, ) cursor = None handles: list[str] = [] while True: result = await workbench.search_mission_context( root_trace_id="root", role="planner", collection="inputs", query="unmatched semantic query", page_size=20, cursor=cursor, ) handles.extend(item["handle"] for item in result["page"]["items"]) cursor = result["page"]["next_cursor"] if cursor is None: break assert len(handles) == 123 assert len(set(handles)) == 123 assert handles[0] == semantic_first[0] @pytest.mark.asyncio async def test_build_900049_shape_automatic_refresh_stays_bounded_and_reduces_by_half() -> None: workbench, ledger, _snapshots, _ = _fixture() refreshes = [await workbench.render("root", ledger) for _ in range(11)] # The framework refreshes every five turns and marks each result output-only-once, # so only the latest bounded projection remains in the provider prompt. assert len(set(refreshes)) == 1 bounded = len(refreshes[-1]) legacy_round = json.dumps( { "task_tree": [task.current_spec.objective for task in ledger.tasks.values()], "accepted_artifacts": ["x" * 4_000 for _ in range(33)], "pattern_payload": ["模式详情" * 350 for _ in range(193)], }, ensure_ascii=False, ) legacy = len(legacy_round) * 54 assert bounded < legacy * 0.50 assert bounded < 24_000 @pytest.mark.asyncio async def test_root_validator_gets_full_closure_as_required_pages_when_large() -> None: workbench, ledger, snapshots, _ = _fixture() contract_ref = "script-build://task-contracts/sha256/" + "d" * 64 contract = SimpleNamespace( task_kind=ScriptTaskKind.ROOT_DELIVERY, scope_ref="script-build://scopes/full", intent_class=ScriptIntentClass.DELIVER, objective="deliver", goal_ids=("g-1",), criteria=(ScriptCriterion("root", "complete"),), input_decision_refs=(), candidate_closure_decision_refs=(), comparison_decision_refs=(), base_artifact_ref=None, to_payload=lambda: {"task_kind": "root-delivery"}, ) ledger.tasks["root-task"].current_spec = SimpleNamespace( objective="deliver", context_refs=("script-build://task-kinds/root-delivery", contract_ref), ) workbench.broker._contracts.values[contract_ref] = contract refs = tuple( ArtifactRef( uri=f"script-build://artifact-versions/{200 + index}", kind=kind, version=str(200 + index), digest="sha256:" + f"{200 + index:064x}", ) for index, kind in enumerate(("direction", "candidate-portfolio", "structured-script")) ) for index, ref in enumerate(refs): workbench.broker._artifacts.values[ref.uri] = _Artifact( f"ROOT CLOSURE {index} " + "正文" * 12_000 ) class RootTools: async def validation_evidence_refs(self, *, context: dict[str, Any]): assert context["task_id"] == "root-task" return refs root_workbench = MissionWorkbench( task_store=workbench.broker._task_store, bindings=_Bindings(), snapshots=snapshots, artifacts=workbench.broker._artifacts, contracts=workbench.broker._contracts, root_tools=RootTools(), ) request = RoleContextRequest( role=AgentRole.VALIDATOR, root_trace_id="root", task_id="root-task", spec_version=1, task_spec={}, attempt_id="root-attempt", ledger_revision=ledger.revision, snapshot_id="root-snapshot", artifact_snapshot={ "snapshot_id": "root-snapshot", "artifact_refs": [asdict(refs[-1])], "evidence_refs": [], }, ) state = await root_workbench.validator_state(request) assert state.artifact["full_content_in_bootstrap"] is False assert len(state.required_handles) == 3 assert state.context_bundle["sufficiency"]["status"] == "ambiguous" assert ( root_workbench.pending_required_context( root_trace_id="root", task_id="root-task", attempt_id="root-attempt" ) == 3 ) for handle in state.required_handles: cursor = None while True: page = await root_workbench.read_mission_context( root_trace_id="root", role="validator", task_id="root-task", attempt_id="root-attempt", handle=handle, cursor=cursor, ) cursor = page["next_cursor"] if cursor is None: break assert ( root_workbench.pending_required_context( root_trace_id="root", task_id="root-task", attempt_id="root-attempt" ) == 0 ) @pytest.mark.asyncio async def test_root_worker_can_expand_structured_script_and_validator_reads_survive_restart() -> ( None ): workbench, ledger, snapshots, _ = _fixture() contract_ref = "script-build://task-contracts/sha256/" + "e" * 64 contract = SimpleNamespace( task_kind=ScriptTaskKind.ROOT_DELIVERY, scope_ref="script-build://scopes/full", intent_class=ScriptIntentClass.DELIVER, objective="deliver", goal_ids=("g-1",), criteria=(ScriptCriterion("root", "complete"),), input_decision_refs=(), candidate_closure_decision_refs=(), comparison_decision_refs=(), base_artifact_ref=None, to_payload=lambda: {"task_kind": "root-delivery"}, ) ledger.tasks["root-task"].current_spec = SimpleNamespace( objective="deliver", context_refs=("script-build://task-kinds/root-delivery", contract_ref), ) workbench.broker._contracts.values[contract_ref] = contract refs = tuple( ArtifactRef( uri=f"script-build://artifact-versions/{500 + index}", kind=kind, version=str(500 + index), digest="sha256:" + f"{500 + index:064x}", ) for index, kind in enumerate(("direction", "candidate-portfolio", "structured-script")) ) for index, ref in enumerate(refs): workbench.broker._artifacts.values[ref.uri] = _Artifact( f"ROOT SOURCE {index} " + "正文" * 12_000 ) class RootTools: async def validation_evidence_refs(self, *, context: dict[str, Any]): assert context["task_id"] == "root-task" return refs receipt_store = _ReceiptStore() root_workbench = MissionWorkbench( task_store=workbench.broker._task_store, bindings=_Bindings(), snapshots=snapshots, artifacts=workbench.broker._artifacts, contracts=workbench.broker._contracts, root_tools=RootTools(), receipt_store=receipt_store, ) worker_request = RoleContextRequest( role=AgentRole.WORKER, root_trace_id="root", task_id="root-task", spec_version=1, task_spec={}, attempt_id="root-attempt", ledger_revision=ledger.revision, ) worker = await root_workbench.worker_state(worker_request) source_types = {item["source_type"] for item in worker.context_bundle["cards"]} assert {"direction", "candidate-portfolio", "structured-script"}.issubset(source_types) structured_handle = next( item["handle"] for item in worker.context_bundle["cards"] if item["source_type"] == "structured-script" ) detail = await root_workbench.read_mission_context( root_trace_id="root", role="worker", task_id="root-task", attempt_id="root-attempt", handle=structured_handle, ) assert "ROOT SOURCE 2" in detail["content_fragment"] validator_request = RoleContextRequest( role=AgentRole.VALIDATOR, root_trace_id="root", task_id="root-task", spec_version=1, task_spec={}, attempt_id="root-attempt", ledger_revision=ledger.revision, snapshot_id="root-snapshot", artifact_snapshot={ "snapshot_id": "root-snapshot", "artifact_refs": [asdict(refs[-1])], "evidence_refs": [], }, ) validator_state = await root_workbench.validator_state(validator_request) restarted = MissionWorkbench( task_store=workbench.broker._task_store, bindings=_Bindings(), snapshots=snapshots, artifacts=workbench.broker._artifacts, contracts=workbench.broker._contracts, root_tools=RootTools(), receipt_store=receipt_store, ) with pytest.raises(WorkbenchError) as incomplete: await restarted.verify_context_exhausted( root_trace_id="root", task_id="root-task", attempt_id="root-attempt" ) assert incomplete.value.code == "CONTEXT_NOT_EXHAUSTED" for handle in validator_state.required_handles: cursor = None while True: page = await root_workbench.read_mission_context( root_trace_id="root", role="validator", task_id="root-task", attempt_id="root-attempt", handle=handle, cursor=cursor, ) cursor = page["next_cursor"] if cursor is None: break await restarted.verify_context_exhausted( root_trace_id="root", task_id="root-task", attempt_id="root-attempt" ) @pytest.mark.asyncio async def test_compare_worker_receives_all_and_only_frozen_candidates() -> None: workbench, ledger, _snapshots, _ = _fixture() task = ledger.tasks["task-57"] contract_ref = "script-build://task-contracts/sha256/" + "9" * 64 refs = tuple( AcceptedDecisionRef( decision_id=f"decision-{index}", artifact_ref=ledger.attempts[f"attempt-{index}"].submission.artifact_refs[0], scope_ref=f"script-build://scopes/full/part-{index}", expected_task_kind=ScriptTaskKind.PARAGRAPH, ) for index in range(25) ) direction_ref = AcceptedDecisionRef( decision_id="decision-32", artifact_ref=ledger.attempts["attempt-32"].submission.artifact_refs[0], scope_ref="script-build://scopes/direction", expected_task_kind=ScriptTaskKind.DIRECTION, ) contract = ScriptTaskContractV1( task_kind=ScriptTaskKind.COMPARE, scope_ref="script-build://scopes/full/compare", intent_class=ScriptIntentClass.COMPARE, objective="compare every frozen candidate", # Production contracts add Direction as a normal input. It is protected # policy context, not one of the candidates being compared. input_decision_refs=(direction_ref,), base_artifact_ref=None, write_scope=("script-build://writes/compare",), gap_ref=None, output_schema="comparison-artifact/v1", criteria=(ScriptCriterion("compare", "compare all candidates"),), budget=ScriptTaskBudget(), goal_ids=("g-1",), comparison_decision_refs=refs, compiler_manifest_digest="sha256:" + "a" * 64, ) workbench.broker._contracts.values[contract_ref] = contract task.current_spec = SimpleNamespace( objective=contract.objective, context_refs=("script-build://task-kinds/compare", contract_ref), ) state = await workbench.worker_state( RoleContextRequest( role=AgentRole.WORKER, root_trace_id="root", task_id=task.task_id, spec_version=1, task_spec={}, attempt_id="compare-attempt", ledger_revision=ledger.revision, ) ) cards = state.context_bundle["cards"] assert len(cards) == 25 assert {item["source_type"] for item in cards} == {"paragraph"} assert set(state.context_bundle["required_handles"]) == {item["handle"] for item in cards} @pytest.mark.asyncio async def test_element_worker_gets_target_text_existing_dimensions_and_link_gaps() -> None: workbench, ledger, snapshots, _ = _fixture() task = ledger.tasks["task-57"] contract_ref = "script-build://task-contracts/sha256/" + "8" * 64 paragraph_ref = AcceptedDecisionRef( decision_id="decision-0", artifact_ref=ledger.attempts["attempt-0"].submission.artifact_refs[0], scope_ref="script-build://scopes/full/part-0", expected_task_kind=ScriptTaskKind.PARAGRAPH, ) contract = ScriptTaskContractV1( task_kind=ScriptTaskKind.ELEMENT_SET, scope_ref="script-build://scopes/full/part-0/elements", intent_class=ScriptIntentClass.EXPAND, objective="extract linked elements", input_decision_refs=(paragraph_ref,), base_artifact_ref=None, write_scope=("script-build://writes/part-0/elements",), gap_ref=None, output_schema="element-set-artifact/v1", criteria=(ScriptCriterion("elements", "cover paragraph"),), budget=ScriptTaskBudget(), goal_ids=("g-1",), compiler_manifest_digest="sha256:" + "a" * 64, ) workbench.broker._contracts.values[contract_ref] = contract task.current_spec = SimpleNamespace( objective=contract.objective, context_refs=("script-build://task-kinds/element-set", contract_ref), ) class Candidates: calls = 0 async def read_attempt_workspace(self, *, context: dict[str, Any]): self.calls += 1 assert context["task_id"] == task.task_id base = { "level": 1, "theme": "theme", "form": "form", "function": "function", "feeling": "feeling", "description": "description", "is_active": True, } return { "paragraphs": [ { **base, "paragraph_id": 1, "name": "opening", "paragraph_index": 1, "content_range": {"from": 0, "to": 20}, "full_description": "real paragraph text", }, { **base, "paragraph_id": 2, "name": "ending", "paragraph_index": 2, "content_range": {"from": 21, "to": 40}, "full_description": "second paragraph text", }, ], "elements": [ { "element_id": 1, "name": "conflict", "dimension_primary": "实质", "dimension_secondary": "叙事冲突", "is_active": True, } ], "paragraph_element_links": [{"paragraph_id": 1, "element_id": 1}], } element_workbench = MissionWorkbench( task_store=workbench.broker._task_store, bindings=_Bindings(), snapshots=snapshots, artifacts=workbench.broker._artifacts, contracts=workbench.broker._contracts, candidates=Candidates(), ) state = await element_workbench.worker_state( RoleContextRequest( role=AgentRole.WORKER, root_trace_id="root", task_id=task.task_id, spec_version=1, task_spec={}, attempt_id="element-attempt", ledger_revision=ledger.revision, ) ) assert len(state.paragraph_targets) == 2 assert all( "paragraph_id" not in item and "content_excerpt" not in item for item in state.paragraph_targets ) assert state.workspace_summary["unlinked_paragraph_count"] == 1 assert state.workspace_summary["unlinked_element_count"] == 0 assert state.workspace_summary["unlinked_paragraph_target_keys"] == [ state.paragraph_targets[1]["paragraph_target_key"] ] assert state.workspace_summary["unlinked_element_handles"] == [] cards = state.context_bundle["cards"] assert {item["source_type"] for item in cards} == {"paragraph", "existing_element"} assert any("实质/叙事冲突" in item["summary"] for item in cards) element_card = next(item for item in cards if item["source_type"] == "existing_element") assert element_card["metadata"]["linked_paragraph_target_keys"] == [ state.paragraph_targets[0]["paragraph_target_key"] ] for _ in range(49): await element_workbench.read_mission_context( root_trace_id="root", role="worker", task_id=task.task_id, attempt_id="element-attempt", handle=state.paragraph_targets[0]["paragraph_target_key"], section="outline", ) assert element_workbench.broker._candidates.calls == 1 @pytest.mark.asyncio async def test_direction_worker_always_receives_always_on_strategy_before_optional_inputs() -> None: workbench, ledger, _snapshots, _ = _fixture() task = ledger.tasks["task-57"] contract_ref = "script-build://task-contracts/sha256/" + "7" * 64 contract = ScriptTaskContractV1( task_kind=ScriptTaskKind.DIRECTION, scope_ref="script-build://scopes/direction", intent_class=ScriptIntentClass.EXPLORE, objective="choose a direction", input_decision_refs=(), base_artifact_ref=None, write_scope=("script-build://writes/direction",), gap_ref=None, output_schema="script-direction/v1", criteria=(ScriptCriterion("direction", "coherent direction"),), budget=ScriptTaskBudget(), goal_ids=(), compiler_manifest_digest="sha256:" + "a" * 64, ) workbench.broker._contracts.values[contract_ref] = contract task.current_spec = SimpleNamespace( objective=contract.objective, context_refs=("script-build://task-kinds/direction", contract_ref), ) state = await workbench.worker_state( RoleContextRequest( role=AgentRole.WORKER, root_trace_id="root", task_id=task.task_id, spec_version=1, task_spec={}, attempt_id="direction-attempt", ledger_revision=ledger.revision, ) ) cards = state.context_bundle["cards"] assert cards[0]["source_type"] == "strategy" assert cards[0]["metadata"]["mode"] == "always_on" assert {item["source_type"] for item in cards}.issubset( {"topic", "persona", "section_pattern", "strategy"} )