增加结束守卫

.gitignore

@@ -19,6 +19,7 @@ build/
 logs/
 tests/*
 !tests/__init__.py
+!tests/supply_agent/
 !tests/supply_infra/
 tests/supply_infra/*
 !tests/supply_infra/__init__.py

agents/find_agent/__init__.py

@@ -24,6 +24,7 @@ _PUBLISH_TIMEOUT_SECONDS = 120.0
 def run_find_agent(
     user_input: str,
     *,
+    run_id: str,
     settings: Settings | None = None,
     model: str | None = None,
 ) -> AgentResult:
@@ -37,6 +38,7 @@ def run_find_agent(
         arun_find_agent(
             agent,
             user_input,
+            run_id=run_id,
             timeout_seconds=find_agent_timeout_seconds(),
         )
     )

agents/find_agent/async_runner.py

@@ -5,6 +5,9 @@ import asyncio
 import logging
 import threading
 
+from agents.find_agent.completion_guard import (
+    configure_find_agent_completion_guard,
+)
 from supply_agent.agent.core import Agent
 from supply_agent.types import AgentResult
 
@@ -68,9 +71,11 @@ async def arun_find_agent(
     agent: Agent,
     user_input: str,
     *,
+    run_id: str,
     timeout_seconds: float,
 ) -> AgentResult:
     """在单个事件循环内运行 find_agent 核心循环并确保资源释放（不含 OSS 发布）。"""
+    configure_find_agent_completion_guard(agent, run_id)
     try:
         return await asyncio.wait_for(
             agent.arun_core(user_input),

agents/find_agent/completion_guard.py

@@ -0,0 +1,60 @@
+"""find_agent 基于持久化运行状态的结束守卫。"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING
+
+from supply_agent.types import CompletionGuard, Message
+from supply_infra.services.video_discovery_service import (
+    get_video_discovery_service,
+)
+
+if TYPE_CHECKING:
+    from supply_agent.agent.core import Agent
+
+_TERMINAL_STATUSES = {"finished", "failed"}
+
+
+def create_find_completion_guard(run_id: str) -> CompletionGuard:
+    """创建只允许已进入 finished / failed 终态的 find_agent 结束守卫。"""
+    normalized_run_id = str(run_id or "").strip()[:64]
+
+    def guard(_response: Message, _messages: Sequence[Message]) -> str | None:
+        if not normalized_run_id:
+            return (
+                "当前用户消息中缺少 run_id，无法确认运行状态。"
+                "请不要直接输出最终结果。"
+            )
+
+        run = get_video_discovery_service().lookup_run(normalized_run_id)
+        if run is None:
+            return (
+                f"run_id={normalized_run_id} 不存在，无法确认运行状态。"
+                "请不要直接输出最终结果。"
+            )
+
+        status = str(run.get("status") or "").strip()
+        if status in _TERMINAL_STATUSES:
+            return None
+
+        return (
+            f"当前 run.status 仍为 {status or 'unknown'}。"
+            "请继续处理；正常完成后先调用 "
+            "update_video_discovery_run_status 将状态更新为 finished，"
+            "无法完成时将状态更新为 failed，再输出最终结果。"
+        )
+
+    return guard
+
+
+def configure_find_agent_completion_guard(agent: Agent, run_id: str) -> None:
+    """在 find_agent 循环创建前按需注入状态结束守卫。"""
+    if getattr(agent, "completion_guard", None) is not None:
+        return
+    agent.completion_guard = create_find_completion_guard(run_id)
+
+
+__all__ = [
+    "configure_find_agent_completion_guard",
+    "create_find_completion_guard",
+]

agents/find_agent/demand_run.py

@@ -378,7 +378,7 @@ def discover_videos_for_demand(
     try:
         from agents.find_agent.run_outcome import evaluate_find_agent_run
 
-        agent_result = run_find_agent(user_input)
+        agent_result = run_find_agent(user_input, run_id=run_id)
         outcome = evaluate_find_agent_run(run_id, agent_result)
         if not outcome.succeeded:
             stop_reason = outcome.failure_reason or "incomplete"

agents/find_agent/prompt/system_prompt.md

@@ -210,13 +210,9 @@
 
 - 根搜索默认形成 2～3 个语义不同的词；每个生产性首页最多继续 1 页，除非第二页仍能
   显著提升判断质量；
-- 搜索结果先按需求相关性、分享规模/效率和主推荐潜力做廉价预筛，默认最多选择8条进入
-  `douyin_detail` 和双画像阶段；
-- 内容相关性与分享动机主要依据标题、描述、`topic_list`、话题标签和详情字段判断，
-  不得依赖或声称使用了视频画面、语音、字幕解析；
+- 搜索结果先按需求相关性、分享规模/效率和主推荐潜力做廉价预筛，然后进入 `douyin_detail` 和双画像阶段；
+- 内容相关性与分享动机主要依据标题、描述、`topic_list`、话题标签和详情字段判断
 - 执行新搜索、翻页、详情或画像后，应重新保存受影响候选；
-- 不得用“接下来我会继续”作为最终回答；
-- 已按“工具故障终止规则”判断无法继续时，不再尝试正常流程，直接输出失败摘要。
 
 # 结束流程
 

agents/find_agent/run.py

@@ -12,13 +12,16 @@ def main() -> None:
 
     while True:
         try:
+            run_id = input("run_id> ").strip()
+            if not run_id or run_id.lower() in ("exit", "quit", "q"):
+                break
             user_input = input("You> ").strip()
         except (EOFError, KeyboardInterrupt):
             print("\nBye.")
             break
         if not user_input or user_input.lower() in ("exit", "quit", "q"):
             break
-        result = run_find_agent(user_input)
+        result = run_find_agent(user_input, run_id=run_id)
         print(f"\nAgent> {result.content}\n")
 
 

supply_agent/__init__.py

@@ -4,6 +4,13 @@ from supply_agent.agent.core import Agent
 from supply_agent.config import Settings
 from supply_agent.skills.registry import SkillRegistry
 from supply_agent.tools.registry import ToolRegistry
+from supply_agent.types import CompletionGuard
 
 __version__ = "0.1.0"
-__all__ = ["Agent", "Settings", "ToolRegistry", "SkillRegistry"]
+__all__ = [
+    "Agent",
+    "CompletionGuard",
+    "Settings",
+    "ToolRegistry",
+    "SkillRegistry",
+]

supply_agent/agent/__init__.py

@@ -1,5 +1,6 @@
 """Agent module."""
 
 from supply_agent.agent.core import Agent
+from supply_agent.types import CompletionGuard
 
-__all__ = ["Agent"]
+__all__ = ["Agent", "CompletionGuard"]

supply_agent/agent/core.py

@@ -9,7 +9,14 @@ from supply_agent.logging.logger import AgentLogger
 from supply_agent.logging.publish import publish_run_artifacts
 from supply_agent.skills.registry import SkillRegistry
 from supply_agent.tools.registry import ToolRegistry
-from supply_agent.types import AgentEvent, AgentEventType, AgentResult, Message, Role
+from supply_agent.types import (
+    AgentEvent,
+    AgentEventType,
+    AgentResult,
+    CompletionGuard,
+    Message,
+    Role,
+)
 
 
 DEFAULT_SYSTEM_PROMPT = """\
@@ -60,6 +67,7 @@ class Agent:
         temperature: float | None = None,
         reasoning_effort: str | None = None,
         logger: AgentLogger | None = None,
+        completion_guard: CompletionGuard | None = None,
     ) -> None:
         self.name = name
         self.settings = settings or get_settings()
@@ -77,6 +85,7 @@ class Agent:
         self.system_prompt = system_prompt or DEFAULT_SYSTEM_PROMPT
         self.max_iterations = max_iterations or self.settings.agent_max_iterations
         self.temperature = temperature
+        self.completion_guard = completion_guard
 
         if model:
             self.llm.set_model(model)
@@ -133,6 +142,7 @@ class Agent:
             temperature=self.temperature,
             logger=self.logger,
             active_skills=self._active_skills,
+            completion_guard=self.completion_guard,
         )
 
     def _finish_run(self, result: AgentResult) -> None:

supply_agent/agent/loop.py

@@ -10,6 +10,7 @@ from supply_agent.types import (
     AgentEvent,
     AgentEventType,
     AgentResult,
+    CompletionGuard,
     Message,
     Role,
     ToolCall,
@@ -21,7 +22,12 @@ if TYPE_CHECKING:
     from supply_agent.logging.logger import AgentLogger
 
 # Outcome of one assistant turn before tools / completion.
-_TurnKind = Literal["done", "tools"]
+_TurnKind = Literal["done", "tools", "continue"]
+
+_DEFAULT_COMPLETION_GUARD_FEEDBACK = (
+    "当前结果尚未满足结束条件。请根据已有上下文继续任务；"
+    "满足结束条件后再输出最终回答。"
+)
 
 
 class AgentLoop:
@@ -44,6 +50,7 @@ class AgentLoop:
         logger: AgentLogger | None = None,
         active_skills: list[str] | None = None,
         system_message_builder: Callable[[], Message] | None = None,
+        completion_guard: CompletionGuard | None = None,
     ) -> None:
         self.llm = llm
         self.tools = tools
@@ -53,6 +60,7 @@ class AgentLoop:
         self.max_iterations = max_iterations
         self.temperature = temperature
         self.logger = logger
+        self.completion_guard = completion_guard
         # Use `is not None` so an empty shared list from Agent is kept by identity.
         self.active_skills = active_skills if active_skills is not None else []
         self.tool_calls_made = 0
@@ -143,8 +151,25 @@ class AgentLoop:
         self.messages.append(response)
         if response.tool_calls:
             return "tools", None
+        feedback = self._completion_guard_feedback(response)
+        if feedback is not None:
+            self._append_completion_guard_feedback(feedback)
+            return "continue", None
         return "done", self._build_result(response.content or "", iterations)
 
+    def _completion_guard_feedback(self, response: Message) -> str | None:
+        if self.completion_guard is None:
+            return None
+        feedback = self.completion_guard(response, tuple(self.messages))
+        if feedback is None:
+            return None
+        if not isinstance(feedback, str):
+            raise TypeError("completion_guard 必须返回 str 或 None")
+        return feedback.strip() or _DEFAULT_COMPLETION_GUARD_FEEDBACK
+
+    def _append_completion_guard_feedback(self, feedback: str) -> None:
+        self.messages.append(Message(role=Role.USER, content=feedback))
+
     def _nudge_best_answer_sync(self, iterations: int) -> AgentResult:
         self.messages.append(
             Message(
@@ -197,6 +222,8 @@ class AgentLoop:
             kind, done = self._handle_assistant_message(response, iterations)
             if kind == "done" and done is not None:
                 return done
+            if kind == "continue":
+                continue
 
             assert response.tool_calls is not None
             for tc in response.tool_calls:
@@ -224,6 +251,8 @@ class AgentLoop:
             kind, done = self._handle_assistant_message(response, iterations)
             if kind == "done" and done is not None:
                 return done
+            if kind == "continue":
+                continue
 
             assert response.tool_calls is not None
             for tc in response.tool_calls:
@@ -264,6 +293,8 @@ class AgentLoop:
                     data=done.model_dump(),
                 )
                 return
+            if kind == "continue":
+                continue
 
             assert response.tool_calls is not None
             for tc in response.tool_calls:
@@ -328,6 +359,8 @@ class AgentLoop:
                     data=done.model_dump(),
                 )
                 return
+            if kind == "continue":
+                continue
 
             assert response.tool_calls is not None
             for tc in response.tool_calls:

supply_agent/types.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
+from collections.abc import Callable, Sequence
 from enum import StrEnum
-from typing import Any, Literal
+from typing import Any, TypeAlias
 
 from pydantic import BaseModel, Field
 
@@ -38,6 +39,19 @@ class Message(BaseModel):
         return data
 
 
+CompletionGuard: TypeAlias = Callable[
+    [Message, Sequence[Message]],
+    str | None,
+]
+"""Completion callback.
+
+Return ``None`` to accept an assistant response without tool calls as the final
+answer. Return feedback text to reject completion and continue the agent loop;
+the feedback is appended as a user message before the next iteration. The guard
+does not run for the forced final answer after ``max_iterations`` is reached.
+"""
+
+
 class ToolCall(BaseModel):
     """A tool invocation requested by the model."""
 

tests/supply_agent/test_agent_loop.py

@@ -0,0 +1,229 @@
+"""AgentLoop shared-path behavior tests."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from supply_agent.agent.loop import AgentLoop
+from supply_agent.tools.base import tool
+from supply_agent.tools.registry import ToolRegistry
+from supply_agent.types import (
+    AgentEventType,
+    Message,
+    Role,
+    ToolCall,
+)
+
+
+class _FakeLLM:
+    """Deterministic LLM that returns scripted assistant messages."""
+
+    def __init__(self, responses: list[Message]) -> None:
+        self._responses = list(responses)
+        self.calls: list[dict[str, Any]] = []
+
+    def chat(
+        self,
+        messages: list[Message],
+        tools: list[Any] | None = None,
+        temperature: float | None = None,
+        *,
+        iteration: int = 0,
+    ) -> Message:
+        self.calls.append(
+            {
+                "messages": list(messages),
+                "tools": tools,
+                "iteration": iteration,
+            }
+        )
+        if not self._responses:
+            raise AssertionError("FakeLLM: no more scripted responses")
+        return self._responses.pop(0)
+
+    async def achat(
+        self,
+        messages: list[Message],
+        tools: list[Any] | None = None,
+        temperature: float | None = None,
+        *,
+        iteration: int = 0,
+    ) -> Message:
+        return self.chat(
+            messages,
+            tools=tools,
+            temperature=temperature,
+            iteration=iteration,
+        )
+
+
+def _system() -> Message:
+    return Message(role=Role.SYSTEM, content="base system")
+
+
+def _assistant_text(content: str) -> Message:
+    return Message(role=Role.ASSISTANT, content=content)
+
+
+def _assistant_tools(*calls: ToolCall) -> Message:
+    return Message(role=Role.ASSISTANT, content=None, tool_calls=list(calls))
+
+
+@pytest.fixture
+def skill_tools() -> ToolRegistry:
+    registry = ToolRegistry()
+
+    @tool(name="load_skill")
+    def load_skill(name: str) -> str:
+        return f"skill-body:{name}"
+
+    registry.register(load_skill, name="load_skill")
+    return registry
+
+
+def test_stream_loads_skill_into_system_message(skill_tools: ToolRegistry) -> None:
+    active_skills: list[str] = []
+    system_parts = {"value": "base system"}
+
+    def builder() -> Message:
+        parts = [system_parts["value"]]
+        for name in active_skills:
+            parts.append(f"LOADED:{name}")
+        return Message(role=Role.SYSTEM, content="\n\n".join(parts))
+
+    llm = _FakeLLM(
+        [
+            _assistant_tools(
+                ToolCall(
+                    id="c1",
+                    name="load_skill",
+                    arguments='{"name": "demo"}',
+                )
+            ),
+            _assistant_text("done"),
+        ]
+    )
+    loop = AgentLoop(
+        llm=llm,  # type: ignore[arg-type]
+        tools=skill_tools,
+        system_message=builder(),
+        system_message_builder=builder,
+        messages=[Message(role=Role.USER, content="hi")],
+        max_iterations=5,
+        active_skills=active_skills,
+    )
+
+    events = list(loop.stream())
+    assert any(e.type == AgentEventType.DONE for e in events)
+    assert "demo" in active_skills
+    # Second LLM call must see refreshed system prompt with skill content.
+    assert "LOADED:demo" in (llm.calls[1]["messages"][0].content or "")
+
+
+def test_stream_nudges_best_answer_on_max_iterations() -> None:
+    registry = ToolRegistry()
+
+    @tool(name="noop")
+    def noop() -> str:
+        return '{"ok": true}'
+
+    registry.register(noop, name="noop")
+
+    llm = _FakeLLM(
+        [
+            _assistant_tools(ToolCall(id="c1", name="noop", arguments="{}")),
+            _assistant_text("final after nudge"),
+        ]
+    )
+    loop = AgentLoop(
+        llm=llm,  # type: ignore[arg-type]
+        tools=registry,
+        system_message=_system(),
+        messages=[Message(role=Role.USER, content="hi")],
+        max_iterations=1,
+    )
+
+    events = list(loop.stream())
+    done = next(e for e in events if e.type == AgentEventType.DONE)
+    assert done.data["content"] == "final after nudge"
+    assert len(llm.calls) == 2
+    assert "best answer" in (llm.calls[1]["messages"][-1].content or "").lower()
+
+
+def test_run_and_stream_share_max_iteration_nudge() -> None:
+    registry = ToolRegistry()
+
+    @tool(name="noop")
+    def noop() -> str:
+        return '{"ok": true}'
+
+    registry.register(noop, name="noop")
+
+    def make_loop(responses: list[Message]) -> AgentLoop:
+        return AgentLoop(
+            llm=_FakeLLM(responses),  # type: ignore[arg-type]
+            tools=registry,
+            system_message=_system(),
+            messages=[Message(role=Role.USER, content="hi")],
+            max_iterations=1,
+        )
+
+    run_result = make_loop(
+        [
+            _assistant_tools(ToolCall(id="c1", name="noop", arguments="{}")),
+            _assistant_text("from-run"),
+        ]
+    ).run()
+    stream_events = list(
+        make_loop(
+            [
+                _assistant_tools(ToolCall(id="c1", name="noop", arguments="{}")),
+                _assistant_text("from-stream"),
+            ]
+        ).stream()
+    )
+    stream_done = next(e for e in stream_events if e.type == AgentEventType.DONE)
+
+    assert run_result.content == "from-run"
+    assert stream_done.data["content"] == "from-stream"
+    assert run_result.iterations == stream_done.data["iterations"] == 1
+
+
+@pytest.mark.asyncio
+async def test_astream_loads_skill(skill_tools: ToolRegistry) -> None:
+    active_skills: list[str] = []
+
+    def builder() -> Message:
+        parts = ["base"]
+        for name in active_skills:
+            parts.append(f"LOADED:{name}")
+        return Message(role=Role.SYSTEM, content="\n\n".join(parts))
+
+    llm = _FakeLLM(
+        [
+            _assistant_tools(
+                ToolCall(
+                    id="c1",
+                    name="load_skill",
+                    arguments='{"name": "async-demo"}',
+                )
+            ),
+            _assistant_text("ok"),
+        ]
+    )
+    loop = AgentLoop(
+        llm=llm,  # type: ignore[arg-type]
+        tools=skill_tools,
+        system_message=builder(),
+        system_message_builder=builder,
+        messages=[Message(role=Role.USER, content="hi")],
+        max_iterations=5,
+        active_skills=active_skills,
+    )
+
+    events = [event async for event in loop.astream()]
+    assert any(e.type == AgentEventType.DONE for e in events)
+    assert "async-demo" in active_skills
+    assert "LOADED:async-demo" in (llm.calls[1]["messages"][0].content or "")

tests/supply_agent/test_completion_guard.py

@@ -0,0 +1,205 @@
+"""Tests for the optional agent completion guard."""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import pytest
+
+from agents.find_agent.completion_guard import (
+    configure_find_agent_completion_guard,
+    create_find_completion_guard,
+)
+from supply_agent import Agent
+from supply_agent.agent.loop import AgentLoop
+from supply_agent.tools.registry import ToolRegistry
+from supply_agent.types import Message, Role
+
+
+class _FakeLLM:
+    def __init__(self, responses: list[Message]) -> None:
+        self.responses = list(responses)
+
+    def chat(self, *_args, **_kwargs) -> Message:
+        return self.responses.pop(0)
+
+    async def achat(self, *_args, **_kwargs) -> Message:
+        return self.responses.pop(0)
+
+
+def _loop(
+    responses: list[Message],
+    *,
+    max_iterations: int = 3,
+    completion_guard=None,
+) -> AgentLoop:
+    return AgentLoop(
+        llm=_FakeLLM(responses),  # type: ignore[arg-type]
+        tools=ToolRegistry(),
+        system_message=Message(role=Role.SYSTEM, content="system"),
+        messages=[Message(role=Role.USER, content="task")],
+        max_iterations=max_iterations,
+        completion_guard=completion_guard,
+    )
+
+
+def test_completion_guard_is_disabled_by_default() -> None:
+    agent = Agent()
+    loop = _loop([Message(role=Role.ASSISTANT, content="final")])
+
+    result = loop.run()
+
+    assert agent.completion_guard is None
+    assert result.content == "final"
+    assert result.iterations == 1
+
+
+def test_agent_forwards_configured_completion_guard_to_loop() -> None:
+    def guard(_response: Message, _messages: Sequence[Message]) -> None:
+        return None
+
+    agent = Agent(completion_guard=guard)
+
+    assert agent.completion_guard is guard
+    assert agent._create_loop([]).completion_guard is guard
+
+
+def test_completion_guard_feedback_continues_sync_loop() -> None:
+    calls = 0
+
+    def guard(_response: Message, messages: Sequence[Message]) -> str | None:
+        nonlocal calls
+        calls += 1
+        assert messages[-1].role == Role.ASSISTANT
+        return "run.status 仍为 running，请先更新为终态" if calls == 1 else None
+
+    loop = _loop(
+        [
+            Message(role=Role.ASSISTANT, content="尚未完成"),
+            Message(role=Role.ASSISTANT, content="最终结果"),
+        ],
+        completion_guard=guard,
+    )
+
+    result = loop.run()
+
+    assert result.content == "最终结果"
+    assert result.iterations == 2
+    assert result.messages[-2].content == "run.status 仍为 running，请先更新为终态"
+
+
+def test_completion_guard_feedback_continues_stream_loop() -> None:
+    calls = 0
+
+    def guard(_response: Message, _messages: Sequence[Message]) -> str | None:
+        nonlocal calls
+        calls += 1
+        return "not ready" if calls == 1 else None
+
+    loop = _loop(
+        [
+            Message(role=Role.ASSISTANT, content="premature"),
+            Message(role=Role.ASSISTANT, content="done"),
+        ],
+        completion_guard=guard,
+    )
+
+    events = list(loop.stream())
+
+    assert events[-1].data["content"] == "done"
+    assert events[-1].data["iterations"] == 2
+
+
+@pytest.mark.asyncio
+async def test_completion_guard_feedback_continues_async_loop() -> None:
+    calls = 0
+
+    def guard(_response: Message, _messages: Sequence[Message]) -> str | None:
+        nonlocal calls
+        calls += 1
+        return "not ready" if calls == 1 else None
+
+    loop = _loop(
+        [
+            Message(role=Role.ASSISTANT, content="premature"),
+            Message(role=Role.ASSISTANT, content="done"),
+        ],
+        completion_guard=guard,
+    )
+
+    result = await loop.arun()
+
+    assert result.content == "done"
+    assert result.iterations == 2
+
+
+def test_iteration_limit_takes_priority_over_completion_guard() -> None:
+    def guard(_response: Message, _messages: Sequence[Message]) -> str:
+        return "run.status 仍为 running"
+
+    loop = _loop(
+        [
+            Message(role=Role.ASSISTANT, content="premature"),
+            Message(role=Role.ASSISTANT, content="forced final"),
+        ],
+        max_iterations=1,
+        completion_guard=guard,
+    )
+
+    result = loop.run()
+
+    assert result.content == "forced final"
+    assert result.iterations == 1
+
+
+@pytest.mark.parametrize("status", ["finished", "failed"])
+def test_find_agent_completion_guard_accepts_terminal_status(
+    monkeypatch: pytest.MonkeyPatch,
+    status: str,
+) -> None:
+    class _Service:
+        @staticmethod
+        def lookup_run(_run_id: str) -> dict[str, str]:
+            return {"status": status}
+
+    monkeypatch.setattr(
+        "agents.find_agent.completion_guard.get_video_discovery_service",
+        lambda: _Service(),
+    )
+    guard = create_find_completion_guard("run-1")
+
+    assert guard(
+        Message(role=Role.ASSISTANT, content="final"),
+        (),
+    ) is None
+
+
+def test_find_agent_completion_guard_rejects_running_status(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class _Service:
+        @staticmethod
+        def lookup_run(_run_id: str) -> dict[str, str]:
+            return {"status": "running"}
+
+    monkeypatch.setattr(
+        "agents.find_agent.completion_guard.get_video_discovery_service",
+        lambda: _Service(),
+    )
+    guard = create_find_completion_guard("run-1")
+
+    feedback = guard(
+        Message(role=Role.ASSISTANT, content="premature"),
+        (),
+    )
+
+    assert feedback is not None
+    assert "run.status 仍为 running" in feedback
+    assert "update_video_discovery_run_status" in feedback
+
+
+def test_configure_find_agent_completion_guard_only_when_missing() -> None:
+    agent = Agent()
+
+    configure_find_agent_completion_guard(agent, "scheduled-run")
+
+    assert agent.completion_guard is not None

tests/supply_agent/test_publish_hook.py

@@ -0,0 +1,64 @@
+"""Publish hook stays free of supply_infra at import time."""
+
+from __future__ import annotations
+
+import importlib
+from pathlib import Path
+
+from supply_agent.logging.publish import (
+    get_run_artifact_publisher,
+    publish_run_artifacts,
+    set_run_artifact_publisher,
+)
+
+
+def test_publish_module_does_not_import_supply_infra() -> None:
+    source = importlib.util.find_spec("supply_agent.logging.publish")
+    assert source is not None and source.origin is not None
+    text = Path(source.origin).read_text(encoding="utf-8")
+    assert "import supply_infra" not in text
+    assert "from supply_infra" not in text
+
+
+def test_publish_is_noop_without_publisher() -> None:
+    previous = get_run_artifact_publisher()
+    try:
+        set_run_artifact_publisher(None)
+
+        class _Logger:
+            pass
+
+        assert publish_run_artifacts(_Logger()) is None  # type: ignore[arg-type]
+    finally:
+        set_run_artifact_publisher(previous)
+
+
+def test_set_run_artifact_publisher_is_invoked() -> None:
+    previous = get_run_artifact_publisher()
+    seen: list[object] = []
+
+    def _publisher(logger: object) -> str:
+        seen.append(logger)
+        return "https://example.com/log.html"
+
+    try:
+        set_run_artifact_publisher(_publisher)
+        marker = object()
+        assert publish_run_artifacts(marker) == "https://example.com/log.html"  # type: ignore[arg-type]
+        assert seen == [marker]
+    finally:
+        set_run_artifact_publisher(previous)
+
+
+def test_infra_registers_publisher_hook() -> None:
+    previous = get_run_artifact_publisher()
+    try:
+        set_run_artifact_publisher(None)
+        import supply_infra  # noqa: F401
+        from supply_infra.agent_logging.register import register_agent_logging_hooks
+        from supply_infra.agent_logging.publish import publish_run_artifacts_to_oss
+
+        register_agent_logging_hooks()
+        assert get_run_artifact_publisher() is publish_run_artifacts_to_oss
+    finally:
+        set_run_artifact_publisher(previous)

tests/supply_agent/test_tool_errors.py

@@ -0,0 +1,44 @@
+"""Tests for tool error detection."""
+
+from __future__ import annotations
+
+from supply_agent.tools.base import tool
+from supply_agent.tools.errors import content_indicates_error
+from supply_agent.tools.registry import ToolRegistry
+
+
+def test_content_indicates_error_truthy_error_field() -> None:
+    assert content_indicates_error('{"error": "boom"}')
+    assert content_indicates_error('{\n  "error": "boom",\n  "title": "x"\n}')
+
+
+def test_content_indicates_error_ignores_null_or_missing() -> None:
+    assert not content_indicates_error('{"error": null, "ok": true}')
+    assert not content_indicates_error('{"error": "", "ok": true}')
+    assert not content_indicates_error('{"ok": true}')
+    assert not content_indicates_error("plain text success")
+    assert not content_indicates_error('["not", "an", "object"]')
+
+
+def test_content_indicates_error_does_not_use_prefix_heuristic() -> None:
+    # Would be true under startswith('{"error"'), but is a success payload.
+    assert not content_indicates_error('{"error_code": 0, "message": "ok"}')
+
+
+def test_registry_execute_sets_is_error_from_json() -> None:
+    registry = ToolRegistry()
+
+    @tool(name="fail")
+    def fail() -> str:
+        return '{\n  "error": "bad args",\n  "input_error": true\n}'
+
+    @tool(name="ok")
+    def ok() -> str:
+        return '{"error": null, "items": []}'
+
+    registry.register(fail, name="fail")
+    registry.register(ok, name="ok")
+
+    assert registry.execute("1", "fail", "{}").is_error is True
+    assert registry.execute("2", "ok", "{}").is_error is False
+    assert registry.execute("3", "missing", "{}").is_error is True

tests/supply_infra/scheduler/test_discover_videos_from_demands.py

@@ -685,6 +685,7 @@ async def test_find_agent_timeout_closes_async_client() -> None:
         await arun_find_agent(
             agent,  # type: ignore[arg-type]
             "test",
+            run_id="test-run",
             timeout_seconds=0.01,
         )