Multi-turn agent history, compact trace storage, leaner judge evidence, clearer timeline

audits/agentic/semantic_judge.py

@@ -19,6 +19,10 @@
 _MAX_RESULT_CHARS = 1600
 _MAX_LEAF_CHARS = 240
 _MAX_LIST_ITEMS = 10
+# The agent's answers are what task completion is judged on: keep them whole
+# enough that a conclusion at the end is never cut off.
+_MAX_RESPONSE_CHARS = 6000
+_MAX_CONVERSATION_MESSAGES = 20
 
 
 @dataclass
@@ -77,6 +81,9 @@ def _retrieved_sources(trajectory: AgentTrajectory) -> list[str]:
     sources: set[str] = set()
     for step in trajectory.steps:
         documents = step.attributes.get("gen_ai.retrieval.documents")
+        if isinstance(documents, str):
+            # The standard attribute is JSON text: a list of documents, not a source name.
+            documents = _document(documents)
         if isinstance(documents, list):
             documents = [_document(item) for item in documents]
         sources.update(_source_values(documents))
@@ -141,6 +148,77 @@ def build_agentic_judge_prompt_extension(trajectory_summary: str) -> str:
 """
 
 
+def _message_text(message: Any) -> str:
+    """The text parts of one GenAI message (role/parts schema)."""
+    if not isinstance(message, dict):
+        return ""
+    parts = message.get("parts")
+    if not isinstance(parts, list):
+        return str(message.get("content") or "")
+    return "".join(str(p.get("content") or "") for p in parts if isinstance(p, dict) and p.get("type") == "text")
+
+
+def _json_attribute(value: Any) -> Any:
+    if isinstance(value, str):
+        try:
+            return json.loads(value)
+        except (TypeError, ValueError):
+            return None
+    return value
+
+
+def _user_question(text: str) -> tuple[str, bool]:
+    """The user's own words, and whether a RAG prompt template wrapped them.
+
+    Open WebUI's RAG template puts a fixed instruction block and a ``<context>``
+    list of sources before the user's message. The sources are on the retrieval
+    steps already, so the judge gets the message after ``</context>``.
+    """
+    head, marker, tail = text.rpartition("</context>")
+    if marker and "<context>" in head and tail.strip():
+        return tail.strip(), True
+    return text, False
+
+
+def _trace_turns(trajectory: Any) -> list[dict[str, str]]:
+    """The conversation as the trace records it: each user turn and the agent's reply.
+
+    A turn is one root span of the trace (Open WebUI starts one per user message).
+    Its last model call carries the user's message in ``gen_ai.input.messages`` and
+    the agent's reply in ``gen_ai.output.messages``. Only observed span content.
+    """
+    steps = list(getattr(trajectory, "steps", []) or [])
+    parent = {s.span_id: s.parent_span_id for s in steps if s.span_id}
+
+    def root(span_id: str) -> str:
+        seen = set()
+        while parent.get(span_id) in parent and span_id not in seen:
+            seen.add(span_id)
+            span_id = parent[span_id]
+        return span_id
+
+    last_call: dict[str, Any] = {}
+    for step in steps:
+        if step.kind == "inference" and step.purpose == "primary":
+            last_call[root(step.span_id)] = step          # steps are in execution order
+    turns: list[dict[str, str]] = []
+    for step in last_call.values():
+        inputs = _json_attribute(step.attributes.get("gen_ai.input.messages"))
+        outputs = _json_attribute(step.attributes.get("gen_ai.output.messages"))
+        users = [m for m in inputs or [] if isinstance(m, dict) and m.get("role") == "user"]
+        reply = "".join(_message_text(m) for m in outputs or [] if isinstance(m, dict))
+        if users:
+            question, wrapped = _user_question(_message_text(users[-1]))
+            turn = {"role": "user", "content": _bounded(question, _MAX_RESPONSE_CHARS), "span_id": step.span_id}
+            if wrapped:
+                turn["rag_template_removed"] = True
+            turns.append(turn)
+        if reply:
+            turns.append({"role": "assistant", "content": _bounded(reply, _MAX_RESPONSE_CHARS),
+                          "span_id": step.span_id})
+    return turns[-_MAX_CONVERSATION_MESSAGES:]
+
+
 def build_evidence_packet(
     trajectory: Any,
     task: str = "",
@@ -167,15 +245,23 @@ def build_evidence_packet(
             "status": step.status,
         })
     retrieval_sources = _retrieved_sources(trajectory)
-    return {
+    turns = _trace_turns(trajectory)
+    packet = {
         "task": _bounded(task),
         "expectations": list(expectations or []),
-        "final_response": _bounded(final_response),
+        "final_response": _bounded(final_response, _MAX_RESPONSE_CHARS),
+        # Multi-turn scenarios answer the task across turns: every turn, from the trace.
+        "conversation": turns,
         "tool_steps": tool_steps,
         "retrieval_source_ids": retrieval_sources,
         "span_ids": [step.span_id for step in steps],
         "truncated": len(steps) > _MAX_TOOL_STEPS,
     }
+    # The last assistant turn already is the final response: do not send it twice.
+    last = next((t for t in reversed(turns) if t["role"] == "assistant"), None)
+    if last and final_response and last["content"] == packet["final_response"]:
+        del packet["final_response"]
+    return packet
 
 
 def build_agentic_semantic_prompt(packet: dict[str, Any], criteria: list[str]) -> str:
@@ -186,6 +272,11 @@ def build_agentic_semantic_prompt(packet: dict[str, Any], criteria: list[str]) -
     lines += [
         "For each dimension respond PASS, FAIL, or INCONCLUSIVE (insufficient observation).",
         "Cite span_ids from the packet as evidence. Never assume unobserved behavior.",
+        ("`task` is the first user message; `conversation` holds every user turn and agent reply"
+         " as recorded on the trace's model-call spans; the agent's final response is its last"
+         " assistant turn (or `final_response` when given separately). Judge task completion on the whole"
+         " conversation: an earlier turn may already answer the task, and later turns answer"
+         " follow-ups."),
         ('Respond with JSON only: {"dimensions": {"<name>": {"status": "PASS|FAIL|INCONCLUSIVE",'
          ' "rationale": "...", "span_ids": ["..."]}}}'),
     ]

audits/agentic/tests/test_compact.py

@@ -0,0 +1,140 @@
+"""Studio stores GenAI span content once per trace, in Open WebUI's reference format."""
+import json
+
+from audits.agentic.compact import compact_spans, expand_spans
+
+SYSTEM = {"role": "system", "parts": [{"type": "text", "content": "You are the Acme Retail Support Refund Assistant. " * 2}]}
+QUESTION = {"role": "user", "parts": [{"type": "text", "content": "Which document sets the electronics return window?"}]}
+CALL = {"role": "assistant", "parts": [{"type": "text", "content": ""}, {
+    "type": "tool_call", "id": "c1", "name": "query_knowledge_files", "arguments": {"query": "electronics window"}}]}
+CHUNKS = [{"content": "Electronics keep the 15-day refund window from the policy.", "source": "a.md"},
+          {"content": "A defective electronic device is covered for up to 90 days.", "source": "b.md"}]
+ANSWER = {"role": "assistant", "parts": [{"type": "text", "content": "Electronics have a 15-day window (a.md)."}]}
+
+
+def _span(span_id, op, start, trace="t1", **attrs):
+    return {"trace_id": trace, "span_id": span_id, "parent_span_id": "wf", "name": op,
+            "start_time": start, "end_time": start + 0.1,
+            "attributes": {"gen_ai.operation.name": op,
+                           **{k: json.dumps(v) for k, v in attrs.items()}}}
+
+
+def _turn(start=1.0, history=()):
+    history = list(history)
+    return [
+        _span(f"l1-{start}", "chat", start, **{"gen_ai.input.messages": [SYSTEM, *history, QUESTION],
+                                               "gen_ai.output.messages": [{**CALL, "finish_reason": "tool_call"}]}),
+        _span(f"t1-{start}", "execute_tool", start + 1, **{"gen_ai.tool.call.id": "c1",
+                                                            "gen_ai.tool.call.result": {"content": CHUNKS}}),
+        _span(f"l2-{start}", "chat", start + 2, **{
+            "gen_ai.input.messages": [SYSTEM, *history, QUESTION, CALL, {
+                "role": "tool", "parts": [{"type": "tool_call_response", "id": "c1", "response": CHUNKS}]}],
+            "gen_ai.output.messages": [ANSWER]}),
+    ]
+
+
+def _input(span):
+    return json.loads(span["attributes"]["gen_ai.input.messages"])
+
+
+def test_tool_loop_stores_each_piece_once():
+    first, tool, second = compact_spans(_turn())
+    assert len(_input(first)) == 2
+    system, question, call, response = _input(second)
+    assert system == {"role": "system", "parts": [
+        {"type": "reference", "span_id": "l1-1.0", "attribute": "gen_ai.input.messages", "index": 0}]}
+    assert question["parts"][0]["index"] == 1
+    assert call["parts"] == [{"type": "reference", "span_id": "l1-1.0", "attribute": "gen_ai.output.messages", "index": 0}]
+    assert response["parts"][0]["response"] == {"type": "reference", "span_id": "t1-1.0", "attribute": "gen_ai.tool.call.result"}
+    assert json.loads(tool["attributes"]["gen_ai.tool.call.result"]) == {"content": CHUNKS}  # first occurrence kept
+
+
+def test_round_trip_restores_full_content():
+    original = _turn()
+    expanded = expand_spans(compact_spans(original))
+    restored = _input(expanded[2])
+    assert restored[:2] == [SYSTEM, QUESTION]
+    assert restored[2]["parts"][-1]["name"] == "query_knowledge_files"
+    assert restored[3]["parts"][0]["response"] == {"content": CHUNKS}
+
+
+def test_multi_turn_history_refers_to_earlier_turns():
+    turn1 = _turn(1.0)
+    turn2 = _turn(10.0, history=[QUESTION, ANSWER])
+    spans = compact_spans(turn1 + turn2)
+    messages = _input(spans[3])            # turn 2, first model call
+    assert all(m["parts"][0]["type"] == "reference" for m in messages)
+    assert messages[2]["parts"][0] == {"type": "reference", "span_id": "l2-1.0",
+                                       "attribute": "gen_ai.output.messages", "index": 0}
+
+
+def test_compaction_is_idempotent_and_keeps_open_webui_references():
+    once = compact_spans(_turn())
+    assert compact_spans(once) == once
+
+
+def test_other_traces_are_not_referenced():
+    a = _turn(1.0)
+    b = [dict(s, trace_id="t2", span_id=s["span_id"] + "-b") for s in _turn(5.0)]
+    spans = compact_spans(a + b)
+    assert len([m for m in _input(spans[3]) if m["parts"][0]["type"] == "reference"]) == 0
+
+
+def test_evidence_subset_expands_from_the_full_capture():
+    compacted = compact_spans(_turn())
+    (second,) = expand_spans([compacted[2]], pool=compacted)
+    assert _input(second)[0] == SYSTEM
+
+
+def test_unresolvable_reference_is_left_in_place():
+    compacted = compact_spans(_turn())
+    (second,) = expand_spans([compacted[2]])          # no pool: the first call is missing
+    assert _input(second)[0]["parts"][0]["type"] == "reference"
+
+
+def test_workflow_wrapper_is_never_the_referenced_copy():
+    wrapper = _span("wf", "invoke_workflow", 0.5, **{"gen_ai.input.messages": [QUESTION]})
+    spans = compact_spans([wrapper, *_turn()])
+    assert json.loads(spans[0]["attributes"]["gen_ai.input.messages"]) == [QUESTION]
+    assert _input(spans[1])[1] == QUESTION          # LLM-1 keeps the user message in full
+
+
+def test_knowledge_tool_result_refers_to_the_retrieval_documents():
+    """Retrieved chunks are kept once, as the retrieval documents, though the tool span starts first."""
+    docs = [{"source": "a.md", "content": "Electronics keep the 15-day refund window from the policy."}]
+    tool = _span("tool", "execute_tool", 1.0, **{"gen_ai.tool.call.result": {"content": docs}})
+    retrieval = _span("ret", "retrieval", 1.1, **{"gen_ai.retrieval.documents": docs})
+    tool_out, retrieval_out = compact_spans([tool, retrieval])
+    assert json.loads(tool_out["attributes"]["gen_ai.tool.call.result"]) == {
+        "type": "reference", "span_id": "ret", "attribute": "gen_ai.retrieval.documents"}
+    assert json.loads(retrieval_out["attributes"]["gen_ai.retrieval.documents"]) == docs
+
+
+def test_judge_view_keeps_gen_ai_content_once():
+    from audits.agentic.compact import judge_view
+
+    wrapper = _span("wf", "invoke_workflow", 0.5, **{"gen_ai.input.messages": [QUESTION]})
+    spans = [wrapper, *_turn()]
+    for s in spans:
+        s["attributes"].update({"telemetry.sdk.name": "opentelemetry", "service.name": "open-webui"})
+    answer_text = ANSWER["parts"][0]["content"]
+    view = judge_view(spans, pool=spans, conversation=[{"role": "assistant", "content": answer_text}])
+    assert [s["span_id"] for s in view] == ["l1-1.0", "t1-1.0", "l2-1.0"]          # no wrapper
+    assert all(k.startswith("gen_ai.") for s in view for k in s["attributes"])
+    second = view[2]["attributes"]
+    assert list(second)[:2] == ["gen_ai.operation.name", "gen_ai.output.messages"]  # decisions first
+    messages = json.loads(second["gen_ai.input.messages"])
+    assert messages[0] == "(same as span l1-1.0 input.messages message 1, system)"
+    out = json.loads(second["gen_ai.output.messages"])
+    assert out[0]["parts"][0]["content"] == "(the assistant turn shown in the conversation)"
+
+
+def test_judge_view_keeps_the_exception_message():
+    from audits.agentic.compact import judge_view
+
+    span = _span("t9", "execute_tool", 1.0, **{"gen_ai.tool.name": "acme_get_order"})
+    span["attributes"].update({"error.type": "ValueError", "exception.type": "ValueError",
+                               "exception.message": "No order found for 'ACME-9999'.", "service.name": "x"})
+    (view,) = judge_view([span])
+    assert view["attributes"]["exception.message"] == "No order found for 'ACME-9999'."
+    assert "service.name" not in view["attributes"]

audits/agentic/tests/test_semantic_judge_execution.py

@@ -149,3 +149,68 @@ def test_packet_reads_retrieval_documents_sent_as_json_strings():
         ]})])
     packet = build_evidence_packet(trajectory)
     assert packet["retrieval_source_ids"] == ["customer_faq.md", "f-1", "shipping.md"]
+
+
+def test_packet_takes_every_turn_from_the_trace_and_keeps_long_answers():
+    """Each turn's user message and agent reply come from its model-call span; a long
+    final answer is not cut before its conclusion."""
+    import json
+
+    from audits.agentic.semantic_judge import (
+        build_agentic_semantic_prompt,
+        build_evidence_packet,
+    )
+    from audits.agentic.trajectory import normalize
+
+    answer = "The 15-day electronics window applies. " * 30 + "So your January 2nd deadline stands."
+
+    def message(role, text):
+        return {"role": role, "parts": [{"type": "text", "content": text}]}
+
+    def chat(span_id, turn, start, inputs, reply):
+        return {"span_id": span_id, "parent_span_id": turn, "name": "chat", "start_time": start,
+                "attributes": {"gen_ai.operation.name": "chat",
+                               "gen_ai.input.messages": json.dumps(inputs),
+                               "gen_ai.output.messages": json.dumps([message("assistant", reply)])}}
+
+    first = message("user", "How long do I have to return my headphones?")
+    second = message("user", "Is there a high-value exception?")
+    spans = [
+        {"span_id": "wf1", "name": "invoke_workflow", "start_time": 1.0,
+         "attributes": {"gen_ai.operation.name": "invoke_workflow"}},
+        chat("c1", "wf1", 1.1, [first], "15 days from delivery: until January 2nd."),
+        {"span_id": "wf2", "name": "invoke_workflow", "start_time": 5.0,
+         "attributes": {"gen_ai.operation.name": "invoke_workflow"}},
+        chat("c2", "wf2", 5.1, [first, message("assistant", "15 days."), second], answer),
+    ]
+    packet = build_evidence_packet(normalize(spans), task="How long do I have?", final_response=answer)
+    turns = packet["conversation"]
+    assert "final_response" not in packet                           # same as the last turn: sent once
+    assert [(t["role"], t["span_id"]) for t in turns] == [
+        ("user", "c1"), ("assistant", "c1"), ("user", "c2"), ("assistant", "c2")]
+    assert turns[2]["content"] == "Is there a high-value exception?"
+    assert turns[3]["content"].endswith("deadline stands.")
+    assert "whole" in build_agentic_semantic_prompt(packet, ["task_completion"])
+
+
+def test_retrieval_sources_read_the_standard_json_documents_attribute():
+    """gen_ai.retrieval.documents is JSON text: the packet lists source names, not the whole JSON."""
+    import json
+
+    from audits.agentic.semantic_judge import build_evidence_packet
+    from audits.agentic.trajectory import normalize
+
+    docs = [{"source": "refunds.md", "file_id": "f1", "content": "Electronics: 15 days."},
+            {"source": "matrix.md", "content": "Defective: 90 days."}]
+    span = {"span_id": "r1", "name": "retrieval", "start_time": 1.0,
+            "attributes": {"gen_ai.operation.name": "retrieval", "gen_ai.retrieval.documents": json.dumps(docs)}}
+    packet = build_evidence_packet(normalize([span]))
+    assert packet["retrieval_source_ids"] == ["f1", "matrix.md", "refunds.md"]
+
+
+def test_user_turn_drops_the_rag_template_but_keeps_the_question():
+    from audits.agentic.semantic_judge import _user_question
+
+    wrapped = "### Task:\nRespond using the context.\n<context>\n<source id=1>…</source>\n</context>\n\nIs it 15 days?"
+    assert _user_question(wrapped) == ("Is it 15 days?", True)
+    assert _user_question("Is it 15 days?") == ("Is it 15 days?", False)

chat/config.py

@@ -57,7 +57,7 @@ def is_disabled(value: str | None) -> bool:
 ENABLED = not is_disabled(MODE)
 
 #: The base path the Open WebUI subpath build serves at. Must match the
-#: WEBUI_SUBPATH baked into the image (0.12.4.5-subpath -> /chat). Defined
+#: WEBUI_SUBPATH baked into the image (0.12.4.9-subpath -> /chat). Defined
 #: first: UPSTREAM below bakes it into the default.
 SUBPATH = os.environ.get("SIMPLEAUDIT_CHAT_SUBPATH", "/chat")
 if not SUBPATH.startswith("/"):

chat/proxy.py

@@ -585,7 +585,7 @@ def start_open_webui(studio_port: int | None = None) -> subprocess.Popen:
 OWUI_WHEEL_URL = os.environ.get(
     "SIMPLEAUDIT_CHAT_WHEEL",
     "https://github.com/SushantGautam/open-webui/releases/download/"
-    "v0.12.4.5-subpath/open_webui-0.12.4.5-py3-none-any.whl",
+    "v0.12.4.9-subpath/open_webui-0.12.4.9-py3-none-any.whl",
 )
 OWUI_PACKAGE = os.environ.get("SIMPLEAUDIT_CHAT_PACKAGE", "open-webui")
 

chat/tests/test_proxy.py

@@ -381,8 +381,8 @@ class WheelMarkerTests(SimpleTestCase):
     def test_default_wheel_is_content_capture_release(self):
         # Pinned exactly: a version bump must change this assertion deliberately,
         # not pass silently via a substring of an older tag.
-        self.assertIn("v0.12.4.5-subpath", proxy.OWUI_WHEEL_URL)
-        self.assertIn("open_webui-0.12.4.5-py3-none-any.whl", proxy.OWUI_WHEEL_URL)
+        self.assertIn("v0.12.4.9-subpath", proxy.OWUI_WHEEL_URL)
+        self.assertIn("open_webui-0.12.4.9-py3-none-any.whl", proxy.OWUI_WHEEL_URL)
 
     def test_install_requirement_includes_observability_extra(self):
         requirement = proxy._owui_install_requirement()

deploy/openwebui-subpath/Caddyfile.subpath

@@ -3,7 +3,7 @@
 #
 # The routing shape enabled by the subpath fork built via
 # deploy/openwebui-subpath/build.sh (published as
-# ghcr.io/sushantgautam/open-webui:v0.12.4.5-subpath). It replaces the old
+# ghcr.io/sushantgautam/open-webui:v0.12.4.9-subpath). It replaces the old
 # cross-origin iframe model (Caddyfile.chat on :8801) with one origin / one
 # port:
 #

deploy/openwebui-subpath/README.md

@@ -154,7 +154,7 @@ pip caveats): [`DESIGN.md`](./DESIGN.md).
 publishing to ghcr.io (root on push, `-subpath` on `webui_subpath` dispatch).
 
 The cutover is done: `docker-compose.yml` pulls
-`ghcr.io/sushantgautam/open-webui:v0.12.4.5-subpath` (set `WEBUI_SUBPATH=/chat`),
+`ghcr.io/sushantgautam/open-webui:v0.12.4.9-subpath` (set `WEBUI_SUBPATH=/chat`),
 `chat-proxy` (Caddy, no profile) is the **only published port** and runs
 `Caddyfile.subpath`, Studio's wrapper page moved to `/playground/`, and
 `Caddyfile.chat` + the separate cross-origin port are retired. Embedded