Additional agent DAG tests (#750)

- test_agent_provenance.py: test_session_parent_uri,
  test_session_no_parent_uri, and 6 synthesis tests (types,
  single/multiple parents, document, label)
- test_on_action_callback.py: 3 tests — fires before tool, skipped
  for Final, works when None
- test_callback_message_id.py: 7 tests — message_id on think/observe/
  answer callbacks (streaming + non-streaming) and
  send_final_response
- test_parse_chunk_message_id.py (5 tests) - _parse_chunk propagates
  message_id for thought, observation, answer; handles missing
  gracefully
- test_explainability_parsing.py (+1) -
  test_dispatches_analysis_with_tooluse - Analysis+ToolUse mixin still
  dispatches to Analysis
- test_explainability.py (+1) -
  test_observation_found_via_subtrace_synthesis
- chain walker follows from sub-trace Synthesis to find Observation
  and
  Conclusion in correct order
- test_agent_provenance.py (+8) - session parent_uri (2), synthesis
  single/multiple parents, types, document, label (6)

tests/unit/test_provenance/test_agent_provenance.py

@@ -12,6 +12,7 @@ from trustgraph.provenance.agent import (
     agent_iteration_triples,
     agent_observation_triples,
     agent_final_triples,
+    agent_synthesis_triples,
 )
 
 from trustgraph.provenance.namespaces import (
@@ -21,7 +22,7 @@ from trustgraph.provenance.namespaces import (
     TG_QUERY, TG_THOUGHT, TG_ACTION, TG_ARGUMENTS,
     TG_QUESTION, TG_ANALYSIS, TG_CONCLUSION, TG_DOCUMENT,
     TG_ANSWER_TYPE, TG_REFLECTION_TYPE, TG_THOUGHT_TYPE, TG_OBSERVATION_TYPE,
-    TG_TOOL_USE,
+    TG_TOOL_USE, TG_SYNTHESIS,
     TG_AGENT_QUESTION,
 )
 
@@ -105,6 +106,25 @@ class TestAgentSessionTriples:
         )
         assert len(triples) == 6
 
+    def test_session_parent_uri(self):
+        """Subagent sessions derive from a parent entity (e.g. Decomposition)."""
+        parent = "urn:trustgraph:agent:parent/decompose"
+        triples = agent_session_triples(
+            self.SESSION_URI, "Q", "2024-01-01T00:00:00Z",
+            parent_uri=parent,
+        )
+        derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.SESSION_URI)
+        assert derived is not None
+        assert derived.o.iri == parent
+
+    def test_session_no_parent_uri(self):
+        """Top-level sessions have no wasDerivedFrom."""
+        triples = agent_session_triples(
+            self.SESSION_URI, "Q", "2024-01-01T00:00:00Z"
+        )
+        derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.SESSION_URI)
+        assert derived is None
+
 
 # ---------------------------------------------------------------------------
 # agent_iteration_triples
@@ -358,3 +378,59 @@ class TestAgentFinalTriples:
         )
         doc = find_triple(triples, TG_DOCUMENT, self.FINAL_URI)
         assert doc is None
+
+
+# ---------------------------------------------------------------------------
+# agent_synthesis_triples
+# ---------------------------------------------------------------------------
+
+class TestAgentSynthesisTriples:
+
+    SYNTH_URI = "urn:trustgraph:agent:test-session/synthesis"
+    FINDING_0 = "urn:trustgraph:agent:test-session/finding/0"
+    FINDING_1 = "urn:trustgraph:agent:test-session/finding/1"
+    FINDING_2 = "urn:trustgraph:agent:test-session/finding/2"
+
+    def test_synthesis_types(self):
+        triples = agent_synthesis_triples(self.SYNTH_URI, self.FINDING_0)
+        assert has_type(triples, self.SYNTH_URI, PROV_ENTITY)
+        assert has_type(triples, self.SYNTH_URI, TG_SYNTHESIS)
+        assert has_type(triples, self.SYNTH_URI, TG_ANSWER_TYPE)
+
+    def test_synthesis_single_parent_string(self):
+        """Single parent passed as string."""
+        triples = agent_synthesis_triples(self.SYNTH_URI, self.FINDING_0)
+        derived = find_triples(triples, PROV_WAS_DERIVED_FROM, self.SYNTH_URI)
+        assert len(derived) == 1
+        assert derived[0].o.iri == self.FINDING_0
+
+    def test_synthesis_multiple_parents(self):
+        """Multiple parents for supervisor fan-in."""
+        parents = [self.FINDING_0, self.FINDING_1, self.FINDING_2]
+        triples = agent_synthesis_triples(self.SYNTH_URI, parents)
+        derived = find_triples(triples, PROV_WAS_DERIVED_FROM, self.SYNTH_URI)
+        assert len(derived) == 3
+        derived_uris = {t.o.iri for t in derived}
+        assert derived_uris == set(parents)
+
+    def test_synthesis_single_parent_as_list(self):
+        """Single parent passed as list."""
+        triples = agent_synthesis_triples(self.SYNTH_URI, [self.FINDING_0])
+        derived = find_triples(triples, PROV_WAS_DERIVED_FROM, self.SYNTH_URI)
+        assert len(derived) == 1
+        assert derived[0].o.iri == self.FINDING_0
+
+    def test_synthesis_document(self):
+        triples = agent_synthesis_triples(
+            self.SYNTH_URI, self.FINDING_0,
+            document_id="urn:doc:synth",
+        )
+        doc = find_triple(triples, TG_DOCUMENT, self.SYNTH_URI)
+        assert doc is not None
+        assert doc.o.iri == "urn:doc:synth"
+
+    def test_synthesis_label(self):
+        triples = agent_synthesis_triples(self.SYNTH_URI, self.FINDING_0)
+        label = find_triple(triples, RDFS_LABEL, self.SYNTH_URI)
+        assert label is not None
+        assert label.o.value == "Synthesis"

tests/unit/test_provenance/test_explainability.py

@@ -558,3 +558,96 @@ class TestExplainabilityClientDetectSessionType:
         mock_flow = MagicMock()
         client = ExplainabilityClient(mock_flow, retry_delay=0.0)
         assert client.detect_session_type("urn:trustgraph:docrag:abc") == "docrag"
+
+
+class TestChainWalkerFollowsSubTraceTerminal:
+    """Test that _follow_provenance_chain continues from a sub-trace's
+    Synthesis to find downstream entities like Observation."""
+
+    def test_observation_found_via_subtrace_synthesis(self):
+        """
+        DAG: Question -> Analysis -> GraphRAG Question -> Synthesis -> Observation
+        The walker should find Analysis, the sub-trace, then follow from
+        Synthesis to discover Observation.
+        """
+        # Entity triples (s, p, o)
+        entity_data = {
+            "urn:agent:q": [
+                ("urn:agent:q", RDF_TYPE, TG_AGENT_QUESTION),
+                ("urn:agent:q", TG_QUERY, "test"),
+            ],
+            "urn:agent:analysis": [
+                ("urn:agent:analysis", RDF_TYPE, TG_ANALYSIS),
+                ("urn:agent:analysis", PROV_WAS_DERIVED_FROM, "urn:agent:q"),
+            ],
+            "urn:graphrag:q": [
+                ("urn:graphrag:q", RDF_TYPE, TG_QUESTION),
+                ("urn:graphrag:q", RDF_TYPE, TG_GRAPH_RAG_QUESTION),
+                ("urn:graphrag:q", TG_QUERY, "test"),
+                ("urn:graphrag:q", PROV_WAS_DERIVED_FROM, "urn:agent:analysis"),
+            ],
+            "urn:graphrag:synth": [
+                ("urn:graphrag:synth", RDF_TYPE, TG_SYNTHESIS),
+                ("urn:graphrag:synth", PROV_WAS_DERIVED_FROM, "urn:graphrag:q"),
+            ],
+            "urn:agent:obs": [
+                ("urn:agent:obs", RDF_TYPE, TG_OBSERVATION_TYPE),
+                ("urn:agent:obs", PROV_WAS_DERIVED_FROM, "urn:graphrag:synth"),
+            ],
+            "urn:agent:conclusion": [
+                ("urn:agent:conclusion", RDF_TYPE, TG_CONCLUSION),
+                ("urn:agent:conclusion", PROV_WAS_DERIVED_FROM, "urn:agent:obs"),
+            ],
+        }
+
+        # Build a mock flow that answers triples queries
+        # Query by s= returns that entity's triples
+        # Query by p=wasDerivedFrom, o=X returns entities derived from X
+        def mock_triples_query(s=None, p=None, o=None, **kwargs):
+            if s and not p:
+                # Fetch entity triples
+                tuples = entity_data.get(s, [])
+                return _make_wire_triples(tuples)
+            elif p == PROV_WAS_DERIVED_FROM and o:
+                # Find entities derived from o
+                results = []
+                for uri, tuples in entity_data.items():
+                    for _, pred, obj in tuples:
+                        if pred == PROV_WAS_DERIVED_FROM and obj == o:
+                            results.append((uri, pred, obj))
+                return _make_wire_triples(results)
+            return []
+
+        mock_flow = MagicMock()
+        mock_flow.triples_query.side_effect = mock_triples_query
+
+        client = ExplainabilityClient(mock_flow, retry_delay=0.0, max_retries=2)
+
+        # Mock fetch_graphrag_trace to return a trace with a synthesis
+        synth_entity = Synthesis(uri="urn:graphrag:synth", entity_type="synthesis")
+        client.fetch_graphrag_trace = MagicMock(return_value={
+            "question": Question(uri="urn:graphrag:q", entity_type="question",
+                                 question_type="graph-rag"),
+            "synthesis": synth_entity,
+        })
+
+        trace = client.fetch_agent_trace(
+            "urn:agent:q",
+            graph="urn:graph:retrieval",
+        )
+
+        # Should have found all steps
+        step_types = [
+            type(s).__name__ if not isinstance(s, dict) else s.get("type")
+            for s in trace["steps"]
+        ]
+
+        assert "Analysis" in step_types, f"Missing Analysis in {step_types}"
+        assert "sub-trace" in step_types, f"Missing sub-trace in {step_types}"
+        assert "Observation" in step_types, f"Missing Observation in {step_types}"
+        assert "Conclusion" in step_types, f"Missing Conclusion in {step_types}"
+
+        # Observation should come after the sub-trace
+        subtrace_idx = step_types.index("sub-trace")
+        obs_idx = step_types.index("Observation")
+        assert obs_idx > subtrace_idx, "Observation should appear after sub-trace"