Add e2e suite 27: retried LLM turn must keep conversation history

e2e/test_suite27_retry_context.py

@@ -0,0 +1,306 @@
+"""Suite 27: Retry preserves agent conversation context.
+
+Feature NOT covered by Suites 1-26: what an agent's LLM turn receives as input when the
+workflow is RETRIED, as opposed to freshly scheduled.
+
+Conversation history is not carried by ${...} references between tasks — the server's
+LLM_CHAT_COMPLETE task mapper reconstructs it at scheduling time by walking the workflow's
+completed tasks. Retry is a different code path that re-resolves the task definition's
+inputParameters instead, so it can silently hand the model a bare [system, user] template
+with none of the tool calls that already succeeded. The model, seeing what looks like a new
+request, re-issues a tool call it has already made — a duplicate execution of a potentially
+side-effecting tool, not just a display quirk.
+
+Tracked as orkes-io/orkes-conductor#3876. The same shape exists in conductor-oss's
+WorkflowExecutorUtils.taskToBeRescheduled, so this suite is meaningful against either server.
+
+Assertions are algorithmic — no LLM output parsing. The check is on the retried task's
+inputData.messages, read from the Conductor workflow API.
+
+The LLM-retry test is xfail, not skipped: it fails against current servers, which is the
+point of the suite, but it must keep RUNNING so that an XPASS reports the server-side fix.
+A skip would report nothing, in this repo's CI or in downstream's.
+
+No mocks. Real server, real LLM.
+"""
+
+import json
+import os
+import time
+
+import pytest
+import requests
+from conductor.ai.agents import Agent, tool
+
+from conftest import BASE_URL, get_workflow
+
+pytestmark = [pytest.mark.e2e]
+
+# Non-strict so the expected failure is green today and an XPASS — the server fix landing —
+# is also green, just visible in the summary. Strict would invert the problem: the suite
+# would go red the moment the bug is fixed.
+KNOWN_BROKEN = pytest.mark.xfail(
+    reason=(
+        "orkes-io/orkes-conductor#3876: a retried LLM_CHAT_COMPLETE task is dispatched "
+        "without its conversation history. XPASS means the server fix has landed — drop "
+        "this marker."
+    ),
+    strict=False,
+)
+
+MODEL = os.environ.get("CONDUCTOR_AGENT_LLM_MODEL", "openai/gpt-4o-mini")
+
+LLM_TASK_TYPE = "LLM_CHAT_COMPLETE"
+CUSTOMER_TOOL = "get_customer_info"
+
+# Distinctive token from get_customer_info's result. Asserting on it proves the ACTUAL prior
+# exchange was reconstructed — a merely tool-shaped message would say nothing about which call
+# it describes.
+TOOL_RESULT_MARKER = "Seattle"
+
+# Seconds the second tool stalls. The condition under test is "retry an LLM turn that was
+# interrupted", and with a fast model the whole agent finishes in a few seconds. Holding the
+# second tool open keeps the workflow RUNNING long enough to interrupt it deterministically.
+# It does not affect what is asserted — the retry happens on an LLM task, not on this one.
+TOOL_DELAY_S = float(os.environ.get("SUITE27_TOOL_DELAY_S", "8"))
+
+ARM_TIMEOUT_S = 120
+POLL_S = 0.1
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Deterministic tools
+# ═══════════════════════════════════════════════════════════════════════════
+
+@tool
+def get_customer_info(customer_id: str) -> dict:
+    """Look up a customer's account information, including their home city, by customer ID."""
+    return {"customer_id": customer_id, "name": "Jane Doe", "city": "Seattle"}
+
+
+@tool
+def get_weather(city: str) -> str:
+    """Get the current temperature, wind speed, and humidity for a city."""
+    time.sleep(TOOL_DELAY_S)
+    return f"{city}: 17.7C, wind 6.4 km/h, humidity 73%"
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Server helpers
+# ═══════════════════════════════════════════════════════════════════════════
+
+def _terminate(execution_id: str, reason: str) -> None:
+    resp = requests.delete(
+        f"{BASE_URL}/api/workflow/{execution_id}", params={"reason": reason}, timeout=15
+    )
+    if resp.status_code >= 400:
+        resp = requests.post(
+            f"{BASE_URL}/api/workflow/{execution_id}/terminate",
+            params={"reason": reason},
+            timeout=15,
+        )
+    resp.raise_for_status()
+
+
+def _retry(execution_id: str) -> None:
+    resp = requests.post(
+        f"{BASE_URL}/api/workflow/{execution_id}/retry",
+        params={"resumeSubworkflowTasks": "false"},
+        timeout=15,
+    )
+    resp.raise_for_status()
+
+
+def _llm_tasks(wf: dict) -> list:
+    return [t for t in wf.get("tasks", []) if t.get("taskType") == LLM_TASK_TYPE]
+
+
+def _roles(messages) -> list:
+    return [str(m.get("role")) for m in (messages or []) if isinstance(m, dict)]
+
+
+def _wait_for_interruptible_llm_turn(execution_id: str):
+    """
+    Block until the first tool has COMPLETED and a LATER LLM turn is non-terminal.
+
+    Interrupting before that reproduces a different (working) case from the issue — retrying a
+    plain tool task, which needs no history. Returns the in-flight LLM task, or None on timeout.
+
+    The tool is matched by taskType: agentspan names tool task refs call_<id>_0__<iter>, so
+    matching on referenceTaskName never fires.
+    """
+    deadline = time.time() + ARM_TIMEOUT_S
+    while time.time() < deadline:
+        wf = get_workflow(execution_id)
+        if wf.get("status") in ("COMPLETED", "FAILED", "TERMINATED", "TIMED_OUT"):
+            return None
+
+        tool_done = any(
+            t.get("status") == "COMPLETED"
+            and CUSTOMER_TOOL in (t.get("taskType") or "", t.get("taskDefName") or "")
+            for t in wf.get("tasks", [])
+        )
+        if tool_done:
+            inflight = [
+                t for t in _llm_tasks(wf) if t.get("status") in ("IN_PROGRESS", "SCHEDULED")
+            ]
+            if inflight:
+                return inflight[-1]
+        time.sleep(POLL_S)
+    return None
+
+
+def _find_retried_llm_task(execution_id: str, before_task_ids: set):
+    """
+    The LLM task the retry produced: a taskId not present before the retry.
+
+    Identity must come from taskId, never from the input. The interrupted original still holds
+    full history in its inputData, so selecting "the task whose messages have history" would
+    find it every time and pass against a broken server.
+    """
+    deadline = time.time() + 30
+    while time.time() < deadline:
+        wf = get_workflow(execution_id)
+        fresh = [t for t in _llm_tasks(wf) if t.get("taskId") not in before_task_ids]
+        with_input = [t for t in fresh if (t.get("inputData") or {}).get("messages")]
+        if with_input:
+            return with_input[-1]
+        time.sleep(POLL_S)
+    return None
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Tests
+# ═══════════════════════════════════════════════════════════════════════════
+
+@pytest.fixture(scope="module")
+def context_agent():
+    return Agent(
+        name="s27_retry_context",
+        model=MODEL,
+        tools=[get_customer_info, get_weather],
+        instructions=(
+            "First call get_customer_info to find the customer's city, then "
+            "call get_weather for that city, then summarize both."
+        ),
+    )
+
+
+@KNOWN_BROKEN
+def test_retried_llm_turn_keeps_conversation_history(runtime, context_agent):
+    """A retried LLM turn must receive the same reconstructed history as a fresh dispatch."""
+    handle = runtime.start(
+        context_agent, "What's the weather like where customer 4471 lives?"
+    )
+    execution_id = handle.execution_id
+
+    target = _wait_for_interruptible_llm_turn(execution_id)
+    if target is None:
+        pytest.skip(
+            "agent never reached an interruptible second LLM turn "
+            "(finished too fast, or the model did not call the tool)"
+        )
+
+    # Sanity: the fresh dispatch of this turn DID get history. That is the behaviour the
+    # retried task has to match, and it also proves the fixture reached the intended state.
+    assert _roles((target.get("inputData") or {}).get("messages")) != ["system", "user"], (
+        "fixture problem: the in-flight LLM turn had no history to begin with"
+    )
+
+    _terminate(execution_id, "suite27-retry-context")
+
+    before = {t.get("taskId") for t in get_workflow(execution_id).get("tasks", [])}
+    _retry(execution_id)
+
+    retried = _find_retried_llm_task(execution_id, before)
+    assert retried is not None, "retry produced no new LLM_CHAT_COMPLETE task"
+
+    messages = (retried.get("inputData") or {}).get("messages")
+    roles = _roles(messages)
+    blob = json.dumps(messages or [], default=str)
+
+    # Structural: some tool exchange survived.
+    has_tool_exchange = any(
+        isinstance(m, dict)
+        and (m.get("toolCalls") or m.get("tool_calls") or str(m.get("role", "")).lower() == "tool")
+        for m in (messages or [])
+    )
+    assert has_tool_exchange, (
+        f"retried LLM task lost the conversation history — roles={roles}. "
+        f"Only [system, user] means the retry path skipped the history assembly the task "
+        f"mapper performs at scheduling time (orkes-io/orkes-conductor#3876)."
+    )
+
+    # Content: it is the ACTUAL prior exchange, not merely something tool-shaped.
+    assert CUSTOMER_TOOL in blob, (
+        f"retried history has a tool exchange but not the '{CUSTOMER_TOOL}' call that ran; "
+        f"roles={roles}"
+    )
+    assert TOOL_RESULT_MARKER in blob, (
+        f"retried history references '{CUSTOMER_TOOL}' but not its result "
+        f"('{TOOL_RESULT_MARKER}'); roles={roles}"
+    )
+
+
+def test_retried_tool_task_is_unaffected(runtime, context_agent):
+    """
+    Control from the issue: retrying an interrupted TOOL task behaves correctly.
+
+    A tool task's input is genuinely static, so re-resolving the definition reproduces it
+    exactly. This must keep passing — a fix for the LLM case must not disturb it.
+    """
+    handle = runtime.start(
+        context_agent, "What's the weather like where customer 4471 lives?"
+    )
+    execution_id = handle.execution_id
+
+    # Wait for the slow tool to be in flight, then interrupt it.
+    deadline = time.time() + ARM_TIMEOUT_S
+    target = None
+    while time.time() < deadline:
+        wf = get_workflow(execution_id)
+        if wf.get("status") in ("COMPLETED", "FAILED", "TERMINATED", "TIMED_OUT"):
+            break
+        inflight = [
+            t
+            for t in wf.get("tasks", [])
+            if "get_weather" in (t.get("taskType") or "", t.get("taskDefName") or "")
+            and t.get("status") in ("IN_PROGRESS", "SCHEDULED")
+        ]
+        if inflight:
+            target = inflight[-1]
+            break
+        time.sleep(POLL_S)
+
+    if target is None:
+        pytest.skip("agent never reached an interruptible get_weather task")
+
+    original_input = dict(target.get("inputData") or {})
+
+    _terminate(execution_id, "suite27-retry-tool-control")
+
+    before = {t.get("taskId") for t in get_workflow(execution_id).get("tasks", [])}
+    _retry(execution_id)
+
+    deadline = time.time() + 30
+    retried = None
+    while time.time() < deadline:
+        wf = get_workflow(execution_id)
+        fresh = [
+            t
+            for t in wf.get("tasks", [])
+            if t.get("taskId") not in before
+            and "get_weather" in (t.get("taskType") or "", t.get("taskDefName") or "")
+        ]
+        if fresh:
+            retried = fresh[-1]
+            break
+        time.sleep(POLL_S)
+
+    assert retried is not None, "retry produced no new get_weather task"
+
+    retried_input = retried.get("inputData") or {}
+    assert retried_input.get("city") == original_input.get("city"), (
+        f"retried tool task should re-run with the same arguments: "
+        f"before={original_input.get('city')!r} after={retried_input.get('city')!r}"
+    )