|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +import asyncio |
| 4 | +from collections.abc import Mapping |
| 5 | +from typing import Any |
| 6 | + |
| 7 | +import pytest |
| 8 | + |
| 9 | +from frontier_agent.core.loop_types import TurnContext, notify_observers |
| 10 | +from plugins.tools import task_board as tb |
| 11 | +from workflows._shared.citation_contract import ( |
| 12 | + finalize_report_with_canonical_references, |
| 13 | +) |
| 14 | +from workflows.agent_team.nodes import fast_reporter_v1 |
| 15 | +from workflows.agent_team.nodes.reporter import agent_team_reporter |
| 16 | +from workflows.agent_team.observers import bare_text_finalize as btf |
| 17 | +from workflows.agent_team.observers.bare_text_finalize import ( |
| 18 | + BareTextFinalizeObserver, |
| 19 | +) |
| 20 | +from workflows.agent_team.spec import SWARM_SPEC |
| 21 | +from workflows.agent_team.spec_report import AGENT_TEAM_REPORT_SPEC |
| 22 | + |
| 23 | +_TASK = "task" |
| 24 | +_ANSWER = "Deployment complete; 100% of attacks blocked [1]." |
| 25 | +_REFERENCES = [{"url": "https://example.com/a", "title": "Example"}] |
| 26 | +_BYPASS_ERR = ( |
| 27 | + "Cannot finish: task board has unresolved item(s) ['t1']. " |
| 28 | + "For each, call update_task(...)" |
| 29 | +) |
| 30 | + |
| 31 | + |
| 32 | +def _seed_board(resolutions: dict[str, str]) -> None: |
| 33 | + tb._BOARDS[_TASK] = { |
| 34 | + "seq": len(resolutions), |
| 35 | + "tasks": { |
| 36 | + tid: { |
| 37 | + "description": f"work {tid}", |
| 38 | + "resolution": resolution, |
| 39 | + "owners": [], |
| 40 | + } |
| 41 | + for tid, resolution in resolutions.items() |
| 42 | + }, |
| 43 | + } |
| 44 | + |
| 45 | + |
| 46 | +def _context(turn: int, max_turns: int = 20) -> TurnContext: |
| 47 | + return TurnContext( |
| 48 | + turn=turn, |
| 49 | + max_turns=max_turns, |
| 50 | + task_id=_TASK, |
| 51 | + role_id="coordinator", |
| 52 | + ai_text=_ANSWER, |
| 53 | + thinking="", |
| 54 | + tool_calls=[], |
| 55 | + messages=[], |
| 56 | + usage=None, |
| 57 | + metadata={}, |
| 58 | + ) |
| 59 | + |
| 60 | + |
| 61 | +def _dispatch(ctx: TurnContext) -> list: |
| 62 | + try: |
| 63 | + return asyncio.run( |
| 64 | + notify_observers( |
| 65 | + [BareTextFinalizeObserver()], |
| 66 | + "on_llm_response", |
| 67 | + ctx, |
| 68 | + ), |
| 69 | + ) |
| 70 | + finally: |
| 71 | + tb._BOARDS.pop(_TASK, None) |
| 72 | + |
| 73 | + |
| 74 | +def test_mid_run_unfinished_board_still_blocks() -> None: |
| 75 | + """Away from the last turn the gate wins: rejected, never latched.""" |
| 76 | + _seed_board({"t1": "open"}) |
| 77 | + ctx = _context(turn=5) |
| 78 | + |
| 79 | + interventions = _dispatch(ctx) |
| 80 | + |
| 81 | + (intervention,) = interventions |
| 82 | + assert intervention.continue_to_next_turn is True |
| 83 | + assert intervention.stop_reason is None |
| 84 | + assert "final_answer" not in ctx.metadata |
| 85 | + |
| 86 | + |
| 87 | +def test_final_turn_bypass_stores_marker_and_warning() -> None: |
| 88 | + """The bypass must leave machine- and human-readable traces in metadata. |
| 89 | +
|
| 90 | + ``finalize_gate`` blocks while board items are open, but the final turn |
| 91 | + accepts the answer anyway (never lose it to max_turns). The observer |
| 92 | + stores the gate message (``finalize_gate_bypassed``) and a ready-to-append |
| 93 | + warning built while the board is still live |
| 94 | + (``finalize_gate_warning``); delivery nodes append the warning after any |
| 95 | + finalization that would otherwise strip it (issue #19, review on #48). |
| 96 | + The latched answer itself stays clean — appending here would be defeated |
| 97 | + by the reporter's References cleanup. |
| 98 | + """ |
| 99 | + _seed_board({"t1": "open", "t2": "open"}) |
| 100 | + ctx = _context(turn=19) |
| 101 | + |
| 102 | + interventions = _dispatch(ctx) |
| 103 | + |
| 104 | + (intervention,) = interventions |
| 105 | + assert intervention.stop_reason == "final_answer" |
| 106 | + assert ctx.metadata.get("finalize_gate_bypassed") |
| 107 | + warning = str(ctx.metadata.get("finalize_gate_warning") or "") |
| 108 | + assert "unfinished" in warning.casefold() |
| 109 | + assert "t1" in warning and "t2" in warning |
| 110 | + # The answer is latched verbatim; the warning travels separately. |
| 111 | + assert ctx.metadata["final_answer"] == _ANSWER |
| 112 | + |
| 113 | + |
| 114 | +def test_final_turn_clean_board_latches_untouched() -> None: |
| 115 | + """Gate passes → the answer is latched verbatim, no marker, no warning.""" |
| 116 | + _seed_board({"t1": "resolved", "t2": "cancelled"}) |
| 117 | + ctx = _context(turn=19) |
| 118 | + |
| 119 | + interventions = _dispatch(ctx) |
| 120 | + |
| 121 | + (intervention,) = interventions |
| 122 | + assert intervention.stop_reason == "final_answer" |
| 123 | + assert ctx.metadata["final_answer"] == _ANSWER |
| 124 | + assert "finalize_gate_bypassed" not in ctx.metadata |
| 125 | + assert "finalize_gate_warning" not in ctx.metadata |
| 126 | + |
| 127 | + |
| 128 | +def test_references_finalizer_strips_a_trailing_warning() -> None: |
| 129 | + """Documents WHY the warning must be re-attached after finalization. |
| 130 | +
|
| 131 | + ``finalize_report_with_canonical_references`` drops everything from the |
| 132 | + ``References`` heading to the end of the body — a warning appended after |
| 133 | + that section does not survive the reporter's citation cleanup. |
| 134 | +
|
| 135 | + ``strip_trailing_references`` refuses to cut when the heading starts |
| 136 | + before 30% of the body (mid-body sections are legitimate), so the body |
| 137 | + must be realistically long — as in a real report, where the reviewer's |
| 138 | + reproduction showed the warning being stripped. |
| 139 | + """ |
| 140 | + body = ( |
| 141 | + "Attackers probed hidden paths, enumerated backup archives, and " |
| 142 | + "fuzzed administrative endpoints across several weeks of access " |
| 143 | + "logs before the intrusion was detected. " * 3 |
| 144 | + + "[1]\n\n" |
| 145 | + "## References\n\n" |
| 146 | + "[1] https://example.com/a\n" |
| 147 | + ) |
| 148 | + warning = "\n\n---\n\n> ⚠ Unfinished work at submission: task t1." |
| 149 | + finalized = finalize_report_with_canonical_references( |
| 150 | + body + warning, |
| 151 | + references=_REFERENCES, |
| 152 | + language="en", |
| 153 | + ) |
| 154 | + assert "Unfinished" not in finalized |
| 155 | + assert "https://example.com/a" in finalized # canonical block re-appended |
| 156 | + |
| 157 | + |
| 158 | +def test_append_bypass_warning_is_idempotent() -> None: |
| 159 | + """Delivery boundaries append exactly once, even on repeated calls.""" |
| 160 | + warning = "\n\n---\n\n> ⚠ Unfinished work at submission: task t1." |
| 161 | + source: Mapping[str, Any] = {"finalize_gate_warning": warning} |
| 162 | + |
| 163 | + once = btf.append_bypass_warning("Report body.", source) |
| 164 | + twice = btf.append_bypass_warning(once, source) |
| 165 | + assert once.endswith(warning) |
| 166 | + assert twice == once |
| 167 | + # No stored warning → text untouched. |
| 168 | + assert btf.append_bypass_warning("Report body.", {}) == "Report body." |
| 169 | + |
| 170 | + |
| 171 | +@pytest.mark.asyncio |
| 172 | +async def test_reporter_reappends_warning_after_finalization( |
| 173 | + monkeypatch: pytest.MonkeyPatch, |
| 174 | +) -> None: |
| 175 | + """The reporter's output must carry the warning exactly once (review #48). |
| 176 | +
|
| 177 | + ``_run_fast_reporter`` is the post-finalization report: References |
| 178 | + cleanup already ran inside the chain, so the node itself is the seam |
| 179 | + that re-attaches the stored bypass warning. |
| 180 | + """ |
| 181 | + finalized_report = ( |
| 182 | + "Rewritten report with citation [1].\n\n" |
| 183 | + "## References\n\n- [1] https://example.com/a\n" |
| 184 | + ) |
| 185 | + |
| 186 | + async def _stub_reporter( |
| 187 | + _state: dict[str, Any], _ctx: Any, |
| 188 | + ) -> str: |
| 189 | + return finalized_report |
| 190 | + |
| 191 | + monkeypatch.setattr(fast_reporter_v1, "_run_fast_reporter", _stub_reporter) |
| 192 | + warning = "\n\n---\n\n> ⚠ Unfinished work at submission: task t1." |
| 193 | + state: dict[str, Any] = { |
| 194 | + "reporter_backend": "fast", |
| 195 | + "metadata": {}, |
| 196 | + "finalize_gate_bypassed": _BYPASS_ERR, |
| 197 | + "finalize_gate_warning": warning, |
| 198 | + } |
| 199 | + |
| 200 | + out = await agent_team_reporter(state, None) # type: ignore[arg-type] |
| 201 | + |
| 202 | + final_answer = str(out.get("final_answer") or "") |
| 203 | + assert final_answer.endswith(warning.rstrip("\n")) or warning in final_answer |
| 204 | + assert final_answer.count("Unfinished work at submission") == 1 |
| 205 | + # The marker must pass through so downstream consumers keep it. |
| 206 | + assert out.get("finalize_gate_bypassed") == _BYPASS_ERR |
| 207 | + |
| 208 | + |
| 209 | +def _node(spec: Any, node_id: str) -> Any: |
| 210 | + return next(node for node in spec.nodes if node.node_id == node_id) |
| 211 | + |
| 212 | + |
| 213 | +@pytest.mark.parametrize("spec", [SWARM_SPEC, AGENT_TEAM_REPORT_SPEC]) |
| 214 | +def test_specs_publish_and_forward_the_bypass_marker(spec: Any) -> None: |
| 215 | + """main_agent must publish the marker; the reporter must receive it. |
| 216 | +
|
| 217 | + Without these list entries the marker dies in loop-local metadata and |
| 218 | + consumers only ever see ``answer_status="complete"`` (review on #48). |
| 219 | + """ |
| 220 | + marker_keys = {"finalize_gate_bypassed", "finalize_gate_warning"} |
| 221 | + main_fields = set(_node(spec, "main_agent").output_fields) |
| 222 | + reporter = _node(spec, "agent_team_reporter") |
| 223 | + assert marker_keys <= main_fields |
| 224 | + assert marker_keys <= set(reporter.context_policy.include_fields) |
| 225 | + assert marker_keys <= set(reporter.output_fields) |
0 commit comments