test(harness): conformance scaffold + CI integration job skeleton

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: declan-scale

.github/workflows/harness-integration.yml

@@ -0,0 +1,33 @@
+name: Harness Integration
+
+on:
+  pull_request:
+    paths:
+      - "src/agentex/lib/core/harness/**"
+      - "src/agentex/lib/adk/_modules/**"
+      - ".github/workflows/harness-integration.yml"
+
+jobs:
+  conformance:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@d4b2f3b6ecc6e67c4457f6d3e41ec42d3d0fcb86 # v5.4.2
+        with:
+          version: '0.10.2'
+
+      - name: Bootstrap
+        run: ./scripts/bootstrap
+
+      - name: Conformance suite
+        run: uv run pytest tests/lib/core/harness/ -v
+
+  # Live integration matrix (harness x {sync, async, temporal}) is added per-harness
+  # in the migration plans. Placeholder job keeps the workflow valid until then.
+  live-matrix:
+    runs-on: ubuntu-latest
+    if: false  # enabled once the first harness's test agents land
+    steps:
+      - run: echo "populated by migration PRs"

tests/lib/core/harness/conformance/__init__.py

@@ -0,0 +1,39 @@
+"""Shared conformance engine: every harness tap registers fixtures here.
+
+A fixture is (name, list[StreamTaskMessage]). The runner asserts that span
+derivation over the events is identical regardless of delivery channel, which is
+the cross-channel guarantee from the spec.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from agentex.lib.core.harness.span_derivation import SpanDeriver
+from agentex.lib.core.harness.types import SpanSignal, StreamTaskMessage
+
+
+@dataclass
+class Fixture:
+    name: str
+    events: list[StreamTaskMessage]
+
+
+_REGISTRY: list[Fixture] = []
+
+
+def register(fixture: Fixture) -> None:
+    _REGISTRY.append(fixture)
+
+
+def all_fixtures() -> list[Fixture]:
+    return list(_REGISTRY)
+
+
+def derive_all(events: list[StreamTaskMessage]) -> list[SpanSignal]:
+    d = SpanDeriver()
+    out: list[SpanSignal] = []
+    for e in events:
+        out.extend(d.observe(e))
+    out.extend(d.flush())
+    return out

tests/lib/core/harness/conformance/runner.py

@@ -0,0 +1,28 @@
+import pytest
+
+from tests.lib.core.harness.conformance.runner import Fixture, derive_all, register, all_fixtures
+from agentex.types.task_message_update import (
+    StreamTaskMessageStart, StreamTaskMessageDone, StreamTaskMessageFull,
+)
+from agentex.types.tool_request_content import ToolRequestContent
+from agentex.types.tool_response_content import ToolResponseContent
+
+register(Fixture(
+    name="builtin-single-tool",
+    events=[
+        StreamTaskMessageStart(type="start", index=0,
+            content=ToolRequestContent(type="tool_request", author="agent",
+                                       tool_call_id="c", name="Bash", arguments={})),
+        StreamTaskMessageDone(type="done", index=0),
+        StreamTaskMessageFull(type="full", index=1,
+            content=ToolResponseContent(type="tool_response", author="agent",
+                                        tool_call_id="c", name="Bash", content="ok")),
+    ],
+))
+
+
+@pytest.mark.parametrize("fixture", all_fixtures(), ids=lambda f: f.name)
+def test_span_derivation_is_deterministic(fixture):
+    # Deriving twice over the same events yields identical signals (the property
+    # that makes yield vs auto-send equivalent, since both observe the same stream).
+    assert derive_all(fixture.events) == derive_all(fixture.events)