tests/test_analysis_pipeline.py
Rama main · réplica de lectura del repositorio original.
"""
Tests for `src.pipeline.analysis_pipeline`.
Acceptance:
- is_analysis_use_case recognises the right modes
- AnalysisLayerNode runs tasks in parallel
- tasks with failed dependencies are skipped
- dependency-output injection via {dep:ID} and param_deps works
- the retry logic kicks in for transient errors
- the decomposer builds working plans per use case
- AnalysisPipelineRunner wraps it all in a DAG
"""
import asyncio
import unittest
from dataclasses import dataclass, field
from src.pipeline.analysis_pipeline import (
ANALYSIS_USE_CASES,
AnalysisLayerNode,
AnalysisPipelineRunner,
Decomposer,
is_analysis_use_case,
)
from src.pipeline.models import SubTask, TaskPlan, TaskResult, TaskState
from tests._helpers import async_test
@dataclass
class _MinimalCtx:
"""Minimal ctx stub with the fields analysis_pipeline touches."""
query: str = "Erkläre Quantenmechanik"
use_case: str = "explainer"
task_plan: TaskPlan = None
task_results: dict = field(default_factory=dict)
preflight_data: dict = field(default_factory=dict)
status: str = "running"
error_message: str = ""
started_at: str = ""
finished_at: str = ""
node_results: list = field(default_factory=list)
class _MockLLMClient:
"""Mock LLM with a fixed answer per prompt substring."""
def __init__(self, responses: dict, fail_count: dict = None):
# responses: dict[str_substr] -> str_response
# fail_count: dict[str_substr] -> int (how often the call should
# fail before it succeeds)
self.responses = responses
self.fail_count = dict(fail_count or {})
self.calls: list[str] = []
async def harvest_complete(self, messages):
prompt = messages[0]["content"]
self.calls.append(prompt)
for k, v in self.responses.items():
if k in prompt:
if self.fail_count.get(k, 0) > 0:
self.fail_count[k] -= 1
raise RuntimeError(f"mock failure ({self.fail_count[k]} remaining)")
return v
raise KeyError(f"No mock response matches the prompt: {prompt[:80]!r}")
async def primary_complete_json(self, messages):
# Real behaviour: returns a dict, not a string. We parse the
# configured string answer as JSON, or return an empty dict for
# broken answers.
text = await self.harvest_complete(messages)
try:
import json as _json
data = _json.loads(text)
return data if isinstance(data, dict) else {}
except (ValueError, TypeError):
return {}
# ───────────────────────────────────────────────────────────────────
# is_analysis_use_case
# ───────────────────────────────────────────────────────────────────
class TestIsAnalysisUseCase(unittest.TestCase):
def test_recognized_modes(self):
for mode in ANALYSIS_USE_CASES:
self.assertTrue(is_analysis_use_case(mode), f"mode {mode!r} should be an analysis")
def test_unrecognized_modes(self):
for mode in ["web", "institution", "product_comparison", "unknown", ""]:
self.assertFalse(is_analysis_use_case(mode), f"mode {mode!r} is NOT an analysis")
# ───────────────────────────────────────────────────────────────────
# AnalysisLayerNode — layer execution
# ───────────────────────────────────────────────────────────────────
class TestLayerExecution(unittest.TestCase):
@async_test
async def test_single_task_executes(self):
task = SubTask(
id="T1", phase="p1", description="Test",
prompt_template="Analysiere: {topic}",
prompt_params={"topic": "Beispiel"},
)
llm = _MockLLMClient(responses={"Analysiere": "ANALYSE-OUTPUT"})
ctx = _MinimalCtx()
node = AnalysisLayerNode(layer_idx=0, tasks=[task], llm_client=llm)
meta = await node.run(ctx)
self.assertEqual(meta["done"], 1)
self.assertIn("T1", ctx.task_results)
self.assertEqual(ctx.task_results["T1"].state, TaskState.DONE)
self.assertEqual(ctx.task_results["T1"].output, "ANALYSE-OUTPUT")
@async_test
async def test_parallel_execution_in_layer(self):
"""Three tasks in one layer run in parallel."""
tasks = [
SubTask(id=f"T{i}", phase="p", description="x",
prompt_template=f"Marker_T{i}",
prompt_params={})
for i in range(3)
]
# every task has its own marker in the prompt
llm = _MockLLMClient(responses={
"Marker_T0": "out-0",
"Marker_T1": "out-1",
"Marker_T2": "out-2",
})
ctx = _MinimalCtx()
node = AnalysisLayerNode(layer_idx=0, tasks=tasks, llm_client=llm)
meta = await node.run(ctx)
self.assertEqual(meta["done"], 3)
self.assertEqual(ctx.task_results["T0"].output, "out-0")
self.assertEqual(ctx.task_results["T1"].output, "out-1")
self.assertEqual(ctx.task_results["T2"].output, "out-2")
@async_test
async def test_skipped_when_dependency_failed(self):
"""A task with a failed dependency is skipped."""
# T1 is already marked FAILED in ctx
ctx = _MinimalCtx(task_results={
"T1": TaskResult(task_id="T1", state=TaskState.FAILED,
error="Erstes Failure"),
})
# T2 depends on T1
task = SubTask(
id="T2", phase="p", description="x",
prompt_template="x", depends_on=["T1"],
)
llm = _MockLLMClient(responses={}) # must not be called
node = AnalysisLayerNode(layer_idx=1, tasks=[task], llm_client=llm)
meta = await node.run(ctx)
self.assertEqual(meta["skipped"], 1)
self.assertEqual(meta["executed"], 0)
self.assertEqual(ctx.task_results["T2"].state, TaskState.SKIPPED)
self.assertIn("T1", ctx.task_results["T2"].error)
# the LLM was NOT called
self.assertEqual(len(llm.calls), 0)
# ───────────────────────────────────────────────────────────────────
# Dependency-Output-Injection
# ───────────────────────────────────────────────────────────────────
class TestDepInjection(unittest.TestCase):
@async_test
async def test_dep_marker_substituted(self):
"""{dep:T1} is replaced by T1.output."""
ctx = _MinimalCtx(task_results={
"T1": TaskResult(task_id="T1", state=TaskState.DONE,
output="ERSTER_OUTPUT"),
})
task = SubTask(
id="T2", phase="p", description="x",
prompt_template="Aufbauend auf: {dep:T1} weitermachen",
depends_on=["T1"],
)
llm = _MockLLMClient(responses={"ERSTER_OUTPUT": "T2-OUT"})
node = AnalysisLayerNode(layer_idx=1, tasks=[task], llm_client=llm)
await node.run(ctx)
# the LLM got the substituted prompt
self.assertIn("ERSTER_OUTPUT", llm.calls[0])
self.assertNotIn("{dep:T1}", llm.calls[0])
@async_test
async def test_param_dep_substituted(self):
"""param_deps fills template variables from dependency outputs."""
ctx = _MinimalCtx(task_results={
"T1": TaskResult(task_id="T1", state=TaskState.DONE,
output="VORHERIGE_ANALYSE"),
})
task = SubTask(
id="T2", phase="p", description="x",
prompt_template="Bewerte: {analyse}",
param_deps={"analyse": "T1"},
depends_on=["T1"],
)
llm = _MockLLMClient(responses={"VORHERIGE_ANALYSE": "T2-OUT"})
node = AnalysisLayerNode(layer_idx=1, tasks=[task], llm_client=llm)
await node.run(ctx)
self.assertIn("VORHERIGE_ANALYSE", llm.calls[0])
@async_test
async def test_concat_param_dep_joins_outputs(self):
"""concat_param_deps concatenates several dependency outputs."""
ctx = _MinimalCtx(task_results={
"T1": TaskResult(task_id="T1", state=TaskState.DONE,
output="OUTPUT_EINS"),
"T2": TaskResult(task_id="T2", state=TaskState.DONE,
output="OUTPUT_ZWEI"),
})
task = SubTask(
id="T3", phase="p", description="x",
prompt_template="Synthese aus: {teile}",
concat_param_deps={"teile": ["T1", "T2"]},
depends_on=["T1", "T2"],
)
llm = _MockLLMClient(responses={"Synthese aus": "T3-OUT"})
node = AnalysisLayerNode(layer_idx=1, tasks=[task], llm_client=llm)
await node.run(ctx)
self.assertIn("OUTPUT_EINS", llm.calls[0])
self.assertIn("OUTPUT_ZWEI", llm.calls[0])
# ───────────────────────────────────────────────────────────────────
# Retry logic
# ───────────────────────────────────────────────────────────────────
class TestRetryLogic(unittest.TestCase):
@async_test
async def test_retry_succeeds_after_failures(self):
"""The first attempts fail, the third succeeds."""
task = SubTask(
id="T1", phase="p", description="x",
prompt_template="MARKER", max_retries=3,
)
llm = _MockLLMClient(
responses={"MARKER": "ERFOLG"},
fail_count={"MARKER": 2}, # the first 2 attempts fail
)
ctx = _MinimalCtx()
node = AnalysisLayerNode(layer_idx=0, tasks=[task], llm_client=llm)
await node.run(ctx)
result = ctx.task_results["T1"]
self.assertEqual(result.state, TaskState.DONE)
self.assertEqual(result.output, "ERFOLG")
self.assertEqual(result.retries_used, 2)
# the LLM was called 3 times
self.assertEqual(len(llm.calls), 3)
@async_test
async def test_failure_after_all_retries(self):
"""If there is no success even after max_retries: FAILED."""
task = SubTask(
id="T1", phase="p", description="x",
prompt_template="MARKER", max_retries=1,
)
llm = _MockLLMClient(
responses={"MARKER": "WIRD-NIE-ERREICHT"},
fail_count={"MARKER": 99}, # always fail
)
ctx = _MinimalCtx()
node = AnalysisLayerNode(layer_idx=0, tasks=[task], llm_client=llm)
await node.run(ctx)
result = ctx.task_results["T1"]
self.assertEqual(result.state, TaskState.FAILED)
self.assertIn("mock failure", result.error)
# max_retries=1 → 2 attempts
self.assertEqual(len(llm.calls), 2)
# ───────────────────────────────────────────────────────────────────
# Decomposer
# ───────────────────────────────────────────────────────────────────
class TestDecomposer(unittest.TestCase):
"""Tests for the decomposer."""
@async_test
async def test_unknown_use_case_returns_empty_plan(self):
"""Unknown use case → empty plan + warning."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="?", use_case="non_existent", preflight_data={},
)
self.assertIsInstance(plan, TaskPlan)
self.assertEqual(plan.tasks, [])
@async_test
async def test_explainer_with_llm_concept_extraction(self):
"""Explainer makes an LLM call and builds the plan from the concept list."""
# the LLM returns JSON with three concepts
llm = _MockLLMClient({
"Identify 3-5 core concepts": (
'{"concepts": ["Welle-Teilchen-Dualität", '
'"Heisenberg-Unschärfe", "Quantenverschränkung"]}'
),
})
decomposer = Decomposer(llm_client=llm)
plan = await decomposer.decompose(
query="Quantenmechanik",
use_case="explainer",
preflight_data={
"topic": "Quantenmechanik",
"audience": "Studierende",
"length": "kurz (1-2 Seiten)",
},
)
# expected structure: 3 concepts → 3 explanation + 3 example + 1 synthesis
self.assertEqual(plan.use_case, "explainer")
self.assertEqual(len(plan.tasks), 7)
phases = [t.phase for t in plan.tasks]
self.assertEqual(phases.count("explanation"), 3)
self.assertEqual(phases.count("example"), 3)
self.assertEqual(phases.count("synthesis"), 1)
# topology: examples depend on explanations, synthesis on everything
synth = [t for t in plan.tasks if t.phase == "synthesis"][0]
self.assertEqual(len(synth.depends_on), 6)
@async_test
async def test_explainer_falls_back_on_invalid_concepts(self):
"""If the LLM delivers no JSON concepts: fall back to the topic."""
llm = _MockLLMClient({
"Identify 3-5 core concepts": "kein JSON",
})
decomposer = Decomposer(llm_client=llm)
plan = await decomposer.decompose(
query="Test",
use_case="explainer",
preflight_data={"topic": "Test", "audience": "x", "length": "y"},
)
# at least one concept (fallback to the topic)
n_explanations = sum(1 for t in plan.tasks if t.phase == "explanation")
self.assertGreaterEqual(n_explanations, 1)
@async_test
async def test_peer_review_full_review(self):
"""Peer review with the focus 'full' generates four aspects."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="Review",
use_case="peer_review",
preflight_data={
"manuscript_summary": "Ein Manuskript...",
"discipline": "Soziologie",
"review_focus": "full",
},
)
self.assertEqual(plan.use_case, "peer_review")
# 4 aspects + 1 recommendation
self.assertEqual(len(plan.tasks), 5)
# the recommendation depends on all aspects
rec = [t for t in plan.tasks if t.id == "REC"][0]
self.assertEqual(len(rec.depends_on), 4)
@async_test
async def test_peer_review_focused_methodology(self):
"""Peer review with the focus 'methodology' generates only one aspect."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="peer_review",
preflight_data={
"manuscript_summary": "x",
"discipline": "x",
"review_focus": "methodology",
},
)
# 1 aspect + 1 recommendation
self.assertEqual(len(plan.tasks), 2)
@async_test
async def test_decision_analysis_builds_matrix(self):
"""Decision analysis builds an options × criteria matrix."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="decision_analysis",
preflight_data={
"decision": "Hardware-Kauf",
"options": "Option A\nOption B\nOption C",
"criteria": "Kosten\nLeistung",
},
)
# 3 options × 2 criteria = 6 cells + 1 recommendation
self.assertEqual(len(plan.tasks), 7)
cells = [t for t in plan.tasks if t.phase == "cell_evaluation"]
self.assertEqual(len(cells), 6)
@async_test
async def test_decision_analysis_uses_default_criteria(self):
"""Without criteria: default list."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="decision_analysis",
preflight_data={
"decision": "Test",
"options": "A\nB",
# criteria empty
},
)
# 2 options × 4 default criteria = 8 + 1
self.assertEqual(len(plan.tasks), 9)
@async_test
async def test_grant_proposal_six_sections_plus_coherence(self):
"""Grant proposal: 6 sections + coherence check."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="grant_proposal",
preflight_data={
"research_question": "Test-Frage",
"funder": "DFG",
"duration": "36 Monate",
},
)
sections = [t for t in plan.tasks if t.phase == "section_drafting"]
self.assertEqual(len(sections), 6)
# the coherence check depends on all sections
coh = [t for t in plan.tasks if t.id == "COHERENCE"][0]
self.assertEqual(len(coh.depends_on), 6)
@async_test
async def test_research_design_phase_chain(self):
"""Research design: gap → hypotheses → (method, limitations) → integration."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="research_design",
preflight_data={
"research_question": "Test?",
"discipline": "Soziologie",
"design_preference": "qualitativ",
},
)
ids = {t.id for t in plan.tasks}
self.assertEqual(ids, {"GAP", "HYPO", "METHOD", "LIMIT", "DESIGN"})
# check the topology
method = next(t for t in plan.tasks if t.id == "METHOD")
self.assertEqual(method.depends_on, ["HYPO"])
design = next(t for t in plan.tasks if t.id == "DESIGN")
self.assertEqual(set(design.depends_on),
{"GAP", "HYPO", "METHOD", "LIMIT"})
# the topological order can be computed
layers = plan.topological_order()
self.assertGreater(len(layers), 1)
@async_test
async def test_literature_review_with_custom_questions(self):
"""Literature review uses the user's key_questions."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="literature_review",
preflight_data={
"topic": "BAföG-Reform",
"discipline": "Bildungswissenschaft",
"key_questions": "Frage 1\nFrage 2",
},
)
questions = [t for t in plan.tasks if t.phase == "question_synthesis"]
self.assertEqual(len(questions), 2)
# The plan includes obtaining literature and a meta-synthesis, so that
# the review does not come from the model's knowledge alone.
ids = [t.id for t in plan.tasks]
self.assertEqual(ids, ["QUERIES", "SEARCH", "Q_1", "Q_2", "META"])
search = next(t for t in plan.tasks if t.id == "SEARCH")
self.assertEqual(search.executor, "literature_search")
self.assertTrue(search.search_config["expand_citations"])
# every question synthesis must see the literature found
for q in questions:
self.assertEqual(q.param_deps.get("hits"), "SEARCH")
@async_test
async def test_literature_review_default_questions(self):
"""Without key_questions: three default questions."""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="literature_review",
preflight_data={"topic": "x", "discipline": "x"},
)
questions = [t for t in plan.tasks if t.phase == "question_synthesis"]
self.assertEqual(len(questions), 3)
@async_test
async def test_literature_finder_searches_and_assesses(self):
"""The mode does not stop at the search queries.
SEARCH runs the queries against the literature APIs, ASSESS
evaluates the hits.
"""
decomposer = Decomposer(llm_client=_MockLLMClient({}))
plan = await decomposer.decompose(
query="x",
use_case="literature_finder",
preflight_data={"topic": "Test"},
)
ids = [t.id for t in plan.tasks]
self.assertEqual(
ids, ["STRATEGY", "QUERIES", "SEARCH", "ASSESS", "SELECTION"])
search = next(t for t in plan.tasks if t.id == "SEARCH")
self.assertEqual(search.executor, "literature_search")
self.assertTrue(search.show_in_report)
# SELECTION renders the selection with the bibliographic details from
# the API — without an LLM call, so that authors, journal and DOI are
# not copied by the model.
sel = next(t for t in plan.tasks if t.id == "SELECTION")
self.assertEqual(sel.executor, "render_selection")
# estimated_calls only counts LLM tasks — search and rendering cost
# no model request.
self.assertEqual(plan.estimated_calls, 3)
@async_test
async def test_decomposer_swallows_internal_errors(self):
"""If an LLM call crashes, the decomposer returns an empty plan instead
of crashing the pipeline."""
class _Broken:
async def primary_complete_json(self, msgs, **kw):
raise RuntimeError("LLM down")
async def harvest_complete(self, msgs, **kw):
raise RuntimeError("LLM down")
decomposer = Decomposer(llm_client=_Broken())
plan = await decomposer.decompose(
query="x",
use_case="explainer",
preflight_data={"topic": "x", "audience": "y", "length": "z"},
)
# empty plan, no crash
self.assertEqual(plan.tasks, [])
@async_test
async def test_parse_lines_helper(self):
"""_parse_lines cleans up bullets, numbers and empty lines."""
out = Decomposer._parse_lines(
"- Erste Option\n"
"* Zweite Option\n"
"1. Dritte Option\n"
"\n"
" 4) Vierte Option \n"
"• Fünfte Option\n"
)
self.assertEqual(out, [
"Erste Option", "Zweite Option", "Dritte Option",
"Vierte Option", "Fünfte Option",
])
@async_test
async def test_all_use_cases_produce_resolvable_prompts(self):
"""Regression test: all analysis use cases build plans whose prompt
templates can be resolved completely.
Prevents bugs of the form: prompt_template contains {var}, but the
variable is not defined in prompt_params/param_deps/
concat_param_deps. Such bugs would show up at run time as unresolved
{var} markers in the LLM prompt — hard to debug.
"""
import re
from src.pipeline.analysis_pipeline import (
ANALYSIS_USE_CASES, AnalysisLayerNode,
)
# typical mandatory inputs per use case (all mandatory fields filled)
inputs = {
"explainer": {
"topic": "Test-Thema", "audience": "Studierende",
"length": "kurz",
},
"peer_review": {
"manuscript_summary": "Manuskript",
"discipline": "Soziologie",
"review_focus": "full",
},
"decision_analysis": {
"decision": "Test", "options": "A\nB", "criteria": "K1",
},
"grant_proposal": {
"research_question": "?", "funder": "DFG",
"duration": "36 Monate",
},
"research_design": {
"research_question": "?", "discipline": "x",
},
"literature_review": {
"topic": "x", "discipline": "y",
},
"literature_finder": {"topic": "x"},
}
# mock LLM for use cases with a preliminary LLM call (explainer)
llm = _MockLLMClient({
"Identify 3-5 core concepts": (
'{"concepts": ["A", "B"]}'
),
})
decomposer = Decomposer(llm)
for uc in sorted(ANALYSIS_USE_CASES):
plan = await decomposer.decompose("q", uc, inputs[uc])
self.assertGreater(
len(plan.tasks), 0,
f"{uc}: empty plan — decomposer broken?",
)
# pre-fill all dependency tasks with state DONE
results = {
dep: TaskResult(
task_id=dep, state=TaskState.DONE,
output=f"[{dep}]",
)
for t in plan.tasks for dep in t.depends_on
}
for t in plan.tasks:
try:
resolved = AnalysisLayerNode._resolve_prompt(t, results)
except Exception as e:
self.fail(
f"{uc}/{t.id}: _resolve_prompt crash: {e}",
)
# no unresolved {var} markers
unresolved = re.findall(r'\{[a-z_]+\}', resolved)
self.assertEqual(
unresolved, [],
f"{uc}/{t.id}: unresolved vars {unresolved} in the "
f"resolved prompt:\n{resolved[:200]}",
)
# no unresolved {dep:...} markers
unresolved_deps = re.findall(r'\{dep:[^}]+\}', resolved)
self.assertEqual(
unresolved_deps, [],
f"{uc}/{t.id}: unresolved deps {unresolved_deps}",
)
# ───────────────────────────────────────────────────────────────────
# AnalysisPipelineRunner
# ───────────────────────────────────────────────────────────────────
class TestAnalysisPipelineRunner(unittest.TestCase):
async def _noop_cb(self, *args, **kwargs):
pass
@async_test
async def test_decompose_only_sets_plan_and_status(self):
runner = AnalysisPipelineRunner(
llm_client=_MockLLMClient({
"Identify 3-5 core concepts": (
'{"concepts": ["A", "B"]}'
),
}),
progress_callback=self._noop_cb,
)
ctx = _MinimalCtx()
ctx.use_case = "explainer"
ctx.preflight_data = {
"topic": "T", "audience": "A", "length": "L",
}
await runner.decompose_only(ctx)
# with the real decomposer: the plan is filled
self.assertIsNotNone(ctx.task_plan)
self.assertGreater(len(ctx.task_plan.tasks), 0)
self.assertEqual(ctx.status, "plan_ready")
@async_test
async def test_run_with_unknown_use_case_errors_clearly(self):
"""Unknown use case → empty plan → clear error message."""
runner = AnalysisPipelineRunner(
llm_client=_MockLLMClient({}),
progress_callback=self._noop_cb,
)
ctx = _MinimalCtx()
ctx.use_case = "non_existent_use_case"
await runner.run(ctx, skip_decompose=False)
self.assertEqual(ctx.status, "error")
self.assertIn("decomposer", ctx.error_message)
self.assertIn("not implemented", ctx.error_message)
@async_test
async def test_run_with_prefab_plan_executes_layers(self):
"""With a hard-wired TaskPlan the DAG execution runs."""
# 3 tasks: T1 → T2, T1 → T3 (T2 and T3 depend on T1)
plan = TaskPlan(
use_case="explainer",
tasks=[
SubTask(id="T1", phase="p", description="x",
prompt_template="MARKER_1"),
SubTask(id="T2", phase="p", description="x",
prompt_template="MARKER_2 — {dep:T1}",
depends_on=["T1"]),
SubTask(id="T3", phase="p", description="x",
prompt_template="MARKER_3 — {dep:T1}",
depends_on=["T1"]),
],
)
ctx = _MinimalCtx()
ctx.task_plan = plan
llm = _MockLLMClient(responses={
"MARKER_1": "out-T1",
"MARKER_2": "out-T2",
"MARKER_3": "out-T3",
})
runner = AnalysisPipelineRunner(
llm_client=llm, progress_callback=self._noop_cb,
)
await runner.run(ctx, skip_decompose=True)
self.assertEqual(ctx.status, "done")
# all 3 tasks: DONE
self.assertEqual(ctx.task_results["T1"].state, TaskState.DONE)
self.assertEqual(ctx.task_results["T2"].state, TaskState.DONE)
self.assertEqual(ctx.task_results["T3"].state, TaskState.DONE)
# T2 and T3 got the T1 output substituted
self.assertIn("out-T1", llm.calls[1]) # layer 2, call 1 (T2 or T3)
self.assertIn("out-T1", llm.calls[2]) # layer 2, call 2
@async_test
async def test_stop_check_aborts_pipeline(self):
"""An external stop check prevents execution."""
plan = TaskPlan(
use_case="explainer",
tasks=[SubTask(id="T1", phase="p", description="x",
prompt_template="MARKER")],
)
ctx = _MinimalCtx()
ctx.task_plan = plan
llm = _MockLLMClient({"MARKER": "ok"})
runner = AnalysisPipelineRunner(
llm_client=llm, progress_callback=self._noop_cb,
stop_check=lambda: True, # always True
)
await runner.run(ctx, skip_decompose=True)
# The pipeline did not run — status either cancelled or
# T1 SKIPPED. Acceptable: what matters is that the LLM was not called
# (the exact semantics depend on the engine; mainly: the stop signal
# was respected)
# at least: not "done"
self.assertNotEqual(ctx.status, "done",
f"status should not be 'done' with stop_check, "
f"was: {ctx.status}")
@async_test
async def test_stop_check_aborts_mid_pipeline(self):
"""A stop signal in the middle of the pipeline (after the first task)
must take effect before all further tasks have run.
stop_check must be called periodically, not only once before the
start — otherwise the user could not stop long pipelines.
"""
# 5 tasks in a chain, every LLM call waits 0.05 s
plan = TaskPlan(
use_case="explainer",
tasks=[
SubTask(id=f"T{i}", phase="p", description=f"T{i}",
prompt_template=f"M{i} {{dep:T{i-1}}}"
if i > 0 else f"M{i}",
depends_on=[f"T{i-1}"] if i > 0 else [])
for i in range(5)
],
)
ctx = _MinimalCtx()
ctx.task_plan = plan
import time as _time
call_counter = [0]
class _SlowLLM:
async def harvest_complete(self, msgs, **kw):
call_counter[0] += 1
await asyncio.sleep(0.05)
return f"out-{call_counter[0]}"
primary_complete = harvest_complete
# stop after 130 ms — should stop after ~3 tasks
start = _time.monotonic()
runner = AnalysisPipelineRunner(
llm_client=_SlowLLM(), progress_callback=self._noop_cb,
stop_check=lambda: _time.monotonic() - start > 0.13,
)
await runner.run(ctx, skip_decompose=True)
# stopped: neither all 5 tasks ran, nor none — and the tasks that
# ran succeeded (a failing double would also stop the chain early)
done = [t for t, r in ctx.task_results.items() if r.state == TaskState.DONE]
self.assertTrue(done, "no task completed — the test double is broken")
self.assertLess(call_counter[0], 5,
"all 5 tasks ran — the stop signal "
"did NOT take effect")
self.assertGreater(call_counter[0], 0,
"not a single task ran — "
"the pipeline stopped too early")
@async_test
async def test_final_report_synthesis_from_final_task(self):
"""With status=done, ctx.final_report must be filled from the final
task's output (otherwise ctx.final_report would stay empty).
"""
# plan: T1 → T2, with T2 as the final task
plan = TaskPlan(
use_case="explainer",
tasks=[
SubTask(id="T1", phase="p1", description="Vorarbeit",
prompt_template="MARKER_1"),
SubTask(id="T2", phase="p2", description="Synthese",
prompt_template="MARKER_2 {dep:T1}",
depends_on=["T1"]),
],
)
ctx = _MinimalCtx()
ctx.task_plan = plan
ctx.preflight_data = {"topic": "Mein Thema"}
llm = _MockLLMClient({
"MARKER_1": "Vorarbeit-Output",
"MARKER_2": "FINALE-SYNTHESE",
})
runner = AnalysisPipelineRunner(
llm_client=llm, progress_callback=self._noop_cb,
)
await runner.run(ctx, skip_decompose=True)
self.assertEqual(ctx.status, "done")
self.assertIsNotNone(ctx.final_report)
# the final task output must be included
self.assertIn("FINALE-SYNTHESE", ctx.final_report)
# title from preflight_data
self.assertIn("Mein Thema", ctx.final_report)
@async_test
async def test_final_report_warns_on_failed_tasks(self):
"""If tasks fail, the report must make that transparent instead of
silently finishing with incomplete content.
"""
plan = TaskPlan(
use_case="explainer",
tasks=[
SubTask(id="T1", phase="p", description="Wichtige Vorarbeit",
prompt_template="MARKER_FAIL", max_retries=0),
SubTask(id="T2", phase="p", description="Final",
prompt_template="MARKER_OK"),
],
)
ctx = _MinimalCtx()
ctx.task_plan = plan
ctx.preflight_data = {"topic": "Test"}
# T1 raises, T2 runs normally
class _SelectiveLLM:
calls = []
async def harvest_complete(self, msgs, **kw):
content = msgs[0]["content"]
_SelectiveLLM.calls.append(content)
if "MARKER_FAIL" in content:
raise RuntimeError("simulated error")
return "T2-Output"
primary_complete = harvest_complete
runner = AnalysisPipelineRunner(
llm_client=_SelectiveLLM(), progress_callback=self._noop_cb,
)
await runner.run(ctx, skip_decompose=True)
self.assertEqual(ctx.status, "done")
# the report should contain the error notice
self.assertIn("Note", ctx.final_report)
self.assertIn("Wichtige Vorarbeit", ctx.final_report)
if __name__ == "__main__":
unittest.main()