frontieragent 长程研究与文件工作 Agent Runtime

300    """Async job model for sub-agent dispatch.
301
302    Core API: submit() / collect() / abort().
303
304    If a SpawnGuard is attached, submit() enforces concurrency/depth/token
305    limits. Without a guard, only basic depth checking applies.
306    """
307
308    def __init__(
309        self,
310        *,
311        event_sink: EventSink | None = None,
312        pause_check_factory: PauseCheckFactory | None = None,
313        agent_registry: AgentRegistry | None = None,
314        resource_manager: Any = None,
315        session_history_dir: Path | None = None,
316    ) -> None:
317        """Construct the bus with optional injected services.
318
319        Missing services fall back to the registry. ``session_history_dir``
320        enables disk offload for messages removed by eager trimming.
321        """
322        self._jobs: dict[str, JobEntry] = {}
323        self._counter: int = 0
324        self._spawn_guard: SpawnGuard | None = None
325        self._sub_agent_profiles: dict[str, dict[str, SubAgentProfile]] = {}
326        self._sessions: dict[str, SubAgentSession] = {}
327        # Per-task evidence / assertion pool — populated when main agents
328        # (or collect_reports) harvest completed SubAgentResults, drained
329        # by the pipeline node at phase end for the report node to consume.
330        self._task_aggregates: dict[str, dict[str, list[dict[str, Any]]]] = {}
331        self._event_sink_injected: EventSink | None = event_sink
332        self._pause_check_factory: PauseCheckFactory | None = pause_check_factory
333        self._agent_registry_injected: AgentRegistry | None = agent_registry
334        self._resource_manager_injected: Any = resource_manager
335        self._session_history_dir: Path | None = session_history_dir

88def deep_merge_overrides(
89    base: dict[str, Any], overrides: dict[str, Any],
90) -> dict[str, Any]:
91    """Deep-merge ``overrides`` onto ``base`` without mutating either.
92
93    Semantics chosen to match SDK caller intent (override one nested
94    knob, leave siblings alone):
95
96    - Nested dicts merge recursively (sibling keys preserved).
97    - Lists are replaced wholesale — partial list merging is almost
98      never what callers want (e.g. ``fallback_api_keys: [new_key]``
99      should *replace* the existing key list, not append).
100    - ``None`` in the override **removes** the key, so callers can
101      disable a defaulted-on feature.
102    - Scalar override replacing a dict in base wins — caller asked
103      for that shape, however unusual.
104    """
105    if not overrides:
106        # Returning a deep copy preserves the contract that the caller
107        # can mutate the result without leaking back into ``base``.
108        return _deep_copy(base)
109
110    out: dict[str, Any] = _deep_copy(base)
111    for key, value in overrides.items():
112        if value is None:
113            out.pop(key, None)
114            continue
115        base_value = out.get(key)
116        if isinstance(base_value, dict) and isinstance(value, dict):
117            out[key] = deep_merge_overrides(base_value, value)
118        else:
119            out[key] = _deep_copy(value)
120    return out

155# ``publish`` used to be the authorization field and ``output_paths`` its
156# attachment. They were never independent: ``validate_publish_contract``
157# rejected true-without-paths and paths-without-true alike, so the boolean
158# carried nothing the manifest did not already carry. Requiring it only added a
159# key the coordinator omitted on roughly a quarter of its assignments (60/236
160# and 63/225 across the two arms of the 30-task APEX replay), where the default
161# silently absorbed it as false -- and an omission next to a correct manifest
162# was a hard rejection rather than a publisher. The manifest is now the grant;
163# ``publish`` survives only so callers that still send it are not rejected.
164#
165# NOTE: this class's docstring is rendered verbatim into the model-facing tool
166# schema, so keep it to what the coordinator needs to read.
167class AgentTeamAssignmentSpec(AssignmentSpec):
168    """One assignment. Carrying ``output_paths`` is what authorizes it to write
169    those ``/outputs`` paths; without them the task runs workspace-only."""
170
171    model_config = ConfigDict(extra="forbid")
172
173    publish: StrictBool | None = Field(
174        default=None,
175        # Marked deprecated in the advertised schema only. Pydantic's own
176        # ``deprecated=True`` warns on every attribute read, and
177        # ``validate_publish_contract`` below reads it twice per assignment --
178        # that is a DeprecationWarning per validated task in every run.
179        json_schema_extra={"deprecated": True},
180        description=(
181            "Deprecated, omit it. output_paths alone grants publication. If "
182            "sent it must agree with the manifest."
183        ),
184    )
185    output_paths: list[StrictStr] = Field(
186        default_factory=list,
187        description=(
188            "Exact absolute final file paths under /outputs. Supplying them IS "
189            "the publication grant, so set them on exactly one final-publisher "
190            "assignment and omit them on every other. Pass a JSON array even "
191            "for one path."
192        ),
193    )
194    replace_manifest: StrictBool = Field(
195        default=False,
196        description=(
197            "Set true only on a follow-up to the existing publisher when the "
198            "required final output formats genuinely changed."
199        ),
200    )
201
202    @property
203    def can_publish(self) -> bool:
204        """Whether this assignment may write ``/outputs``.
205
206        Derived from the manifest alone: a path list is the grant.
207        """
208        return bool(self.output_paths)

130async def collect_reports(timeout: int = 1800) -> str:
131    """Collect reports from any sub-agents that have finished their tasks.
132
133    Reports also drain automatically between turns when agents finish
134    *during* a turn, but agents still running at the start of a new turn
135    won't appear unless you call this. Block-wait here is the right move
136    whenever you need a running agent's output to proceed.
137
138    Behaviour:
139      1. Non-blocking drain: any sub-agent that already finished returns
140         its ``<report>`` immediately, regardless of ``timeout``.
141      2. If the drain found nothing, classify the session state:
142         * no agents at all → ``"create_subagent first"``.
143         * agents created but never assigned → ``[status] no_work_queued``,
144           call ``assign_task`` (do NOT poll again).
145         * all submitted work already collected → ``[status] all_collected``,
146           synthesize or assign a follow-up (do NOT poll again).
147         * at least one session running → wait up to ``timeout`` seconds
148           for the FIRST completion, then drain any others that
149           completed during the wait.
150
151    Reports come back as
152    ``<report agent="NAME" status="complete|incomplete|failed|paused"
153    reason="...">…</report>`` — treat ``incomplete`` / ``failed`` as
154    partial evidence, not a final answer.
155

30class TraceObserver(BaseObserver):
31    """Awaited (critical) so trace lines stay ordered; never returns an
32    intervention. Append-only JSONL; failures are swallowed (a broken trace
33    must not break a run)."""
34
35    critical = True
36
37    def __init__(self, path: str, *, mode: str = "", cwd: str = "") -> None:
38        self.path = path
39        self._write({"t": "start", "mode": mode, "cwd": cwd})
40
41    def _write(self, rec: dict[str, Any]) -> None:
42        try:
43            with open(self.path, "a", encoding="utf-8") as f:
44                f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")
45                f.flush()
46        except Exception:
47            pass
48
49    async def on_llm_response(self, ctx: TurnContext) -> Intervention | None:
50        self._write({
51            "t": "llm", "turn": ctx.turn,
52            "text": ctx.ai_text or "",
53            "tool_calls": list(ctx.tool_calls or []),
54        })
55        return None
56
57    async def on_tool_result(

86def test_native_workflow_modes_are_explicit_and_use_shipped_profiles(monkeypatch):
87    """The TUI must not silently emulate a team with its generic ReAct loop."""
88    monkeypatch.setenv("OPENAI_API_KEY", "test")
89    monkeypatch.setenv("OPENAI_BASE_URL", "https://example.test/v1")
90    monkeypatch.setenv("OPENAI_MODEL", "test-model")
91
92    react = P.get_profile("react")
93    team = P.get_profile("agent_team")
94
95    assert (react.workflow, react.workflow_profile) == (
96        "stateful-react-agent", "tui",
97    )
98    assert (team.workflow, team.workflow_profile) == (
99        "agent_team", "tui",
100    )
101
102
103def test_native_workflow_modes_accept_local_openai_compatible_provider(monkeypatch):
104    """The GPU Compose path needs no real API key for its local SGLang server."""
105    monkeypatch.setenv("OPENAI_PROVIDER", "local")
106    monkeypatch.setenv("OPENAI_API_KEY", "EMPTY")
107    monkeypatch.setenv("OPENAI_BASE_URL", "http://model:30000/v1")
108    monkeypatch.setenv("OPENAI_MODEL", "local-model")
109    monkeypatch.setenv("OPENAI_CONTEXT_WINDOW", "32768")
110    monkeypatch.setenv("OPENAI_MAX_TOKENS", "8192")
111
112    for mode in ("react", "agent_team"):
113        profile = P.get_profile(mode)
114        assert profile.provider == "local"
115        assert profile.runtime_config(profile.model_config, mode=mode).ok
116        assert profile.model_config.api_key == "EMPTY"
117        assert profile.model_config.base_url == "http://model:30000/v1"
118        assert profile.model_config.model == "local-model"
119        assert profile.model_config.context_window == 32768
120        assert profile.model_config.max_tokens == 8192