From caee942922c486106276cac1833cd3a266b408fc Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sun, 18 Oct 2026 15:31:20 -0400
Subject: [PATCH 1/2] feat(flow): Meta-Harness-style trace export for GUI
 rollouts
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Adds openadapt_flow.integrations.harness2.traces. It writes a filesystem an
outer-loop proposer can read: every candidate's harness, and for each rollout
the full trace (every screen, action, reason and error), the identity verdict
at each write, and the read-back from the system of record with the reward.

- from_rollout reads an agent rollout; the identity at a write comes from the
  patient banner the agent was looking at.
- from_flow_run reads a compiled program's report.json, keeping the runtime's
  own identity-gate verdicts and effect checks.
- Screen text is scrubbed like REPORT.md; verdicts are computed first.
- run_task writes the export for every task. In readback mode the identity
  verdict also appears in the trajectory Harness² proposers read.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 packages/flow/docs/HARNESS2.md                |  35 ++
 .../integrations/harness2/adapter.py          |  17 +-
 .../integrations/harness2/selection.py        |  18 +-
 .../integrations/harness2/traces.py           | 423 ++++++++++++++++++
 packages/flow/tests/test_harness2_adapter.py  |  14 +
 packages/flow/tests/test_harness2_traces.py   | 143 ++++++
 6 files changed, 648 insertions(+), 2 deletions(-)
 create mode 100644 packages/flow/openadapt_flow/integrations/harness2/traces.py
 create mode 100644 packages/flow/tests/test_harness2_traces.py

diff --git a/packages/flow/docs/HARNESS2.md b/packages/flow/docs/HARNESS2.md
index 9bd65d9ef..41200fe9a 100644
--- a/packages/flow/docs/HARNESS2.md
+++ b/packages/flow/docs/HARNESS2.md
@@ -93,6 +93,41 @@ end with Harness²'s recursion code: `k` base rollouts, one mode (`seq` or
 `agg`), selection by reward or by Harness²'s model selector, and the
 read-back of the chosen run.
 
+## Trace export
+
+Meta-Harness ([arXiv:2603.28052](https://arxiv.org/abs/2603.28052)) gives
+its proposer a filesystem with every earlier candidate's harness, full
+execution traces and scores. In its ablation, full traces beat summaries,
+56.7 to 38.7. `openadapt_flow.integrations.harness2.traces` writes that
+filesystem for GUI work. `run_task` writes one for every task under
+`<work_dir>/traces`.
+
+```
+traces/INDEX.md                        one line per trace, best reward first
+traces/index.jsonl                     the same, one JSON object per trace
+traces/candidates/<label>/harness/     the harness each candidate ran
+traces/traces/<id>/trace.md            the full trace, readable
+traces/traces/<id>/trace.json          the full trace, structured
+```
+
+Each trace keeps every step in full: the screen, the action, the reason the
+agent gave, and any error. At each write it records whether the patient on
+screen was the requested one (`verified`, `mismatch`, `unreadable`). It ends
+with the read-back: the records added, whose they are, the outcome, the
+reward and the receipt id.
+
+`from_flow_run` reads a compiled program's `report.json` instead. Its
+identity verdicts are the runtime's own gate results, and a read-back can be
+attached. The repair search uses this to read a halted run beside the last
+verified ones.
+
+Screen text passes through `privacy.scrub_text` before it is written, like
+`REPORT.md`. Identity verdicts are computed before scrubbing, so a scrubbed
+export still says which write hit the wrong patient.
+
+In `readback` mode the same identity verdict also appears in the trajectory
+Harness² proposers read, as an observation after each write.
+
 ## Install
 
 Harness² needs Python 3.12 and is not on PyPI. Install it from its
diff --git a/packages/flow/openadapt_flow/integrations/harness2/adapter.py b/packages/flow/openadapt_flow/integrations/harness2/adapter.py
index e1e1d9983..9cdcb47ce 100644
--- a/packages/flow/openadapt_flow/integrations/harness2/adapter.py
+++ b/packages/flow/openadapt_flow/integrations/harness2/adapter.py
@@ -297,6 +297,15 @@ def delivered(self, run_dir: Path) -> bool:
 
     def trajectory(self, run_dir: Path, *, obs_trunc: int) -> str:
         run_dir = Path(run_dir)
+        run = load_run(run_dir)
+        identity: dict[int, Any] = {}
+        if self.evidence == "readback" and run.get("task_id") in self._tasks:
+            # OpenAdapt's identity check at each write: was the record on
+            # screen the requested one? Part of the read-back evidence.
+            from openadapt_flow.integrations.harness2.traces import from_rollout
+
+            trace = from_rollout(run_dir, self._tasks[run["task_id"]])
+            identity = {s.index: s.identity for s in trace.steps if s.identity}
         lines: list[str] = []
         for step in load_steps(run_dir):
             obs = str(step.get("observation", ""))
@@ -313,9 +322,15 @@ def trajectory(self, run_dir: Path, *, obs_trunc: int) -> str:
                 + _action_text(action)
                 + (f"\n  reason given: {thought}" if thought else "")
             )
+            verdict = identity.get(int(step.get("index", -1)))
+            if verdict is not None:
+                lines.append(
+                    f"[OBSERVATION] identity check at this write: {verdict.status}; "
+                    f"requested {verdict.expected}; on screen "
+                    f"{verdict.observed or 'no patient banner'}"
+                )
             if step.get("error"):
                 lines.append(f"[ERROR] {step['error']}")
-        run = load_run(run_dir)
         if run.get("error"):
             lines.append(f"[ERROR] {run['error']}")
         if not any(line.startswith("[ACTION] done") for line in lines):
diff --git a/packages/flow/openadapt_flow/integrations/harness2/selection.py b/packages/flow/openadapt_flow/integrations/harness2/selection.py
index da06e0178..08eeab904 100644
--- a/packages/flow/openadapt_flow/integrations/harness2/selection.py
+++ b/packages/flow/openadapt_flow/integrations/harness2/selection.py
@@ -8,7 +8,8 @@
 
 :func:`run_task` is the loop for one task: ``k`` base rollouts, one Harness²
 recursion mode (its code, imported), a selection by reward or by a model, and
-the read-back of the chosen run.
+the read-back of the chosen run, and a trace export of every rollout
+(``<work_dir>/traces``, see :mod:`.traces`).
 """
 
 from __future__ import annotations
@@ -23,6 +24,7 @@
 from harness2.core.proposer import Proposer
 from harness2.core.spec import RunView
 
+from openadapt_flow.integrations.harness2 import traces
 from openadapt_flow.integrations.harness2.adapter import OpenAdaptGuiAdapter
 
 #: Where an unscored candidate ranks: below a safe stop, above any penalty.
@@ -166,6 +168,20 @@ def run_id_of(label: str) -> str:
         "mode_result": result.to_dict(),
         "proposer_usd": round(sum(s.usd for s in result.stages) + sel.usd, 6),
     }
+    labelled = [(lbl, rv) for lbl, rv in result.rollouts.items() if rv.ok]
+    traces.export(
+        work_dir / "traces",
+        [
+            traces.from_rollout(rv.run_dir, adapter.task(task_id), candidate=lbl)
+            for lbl, rv in labelled
+        ],
+        harnesses={
+            lbl: work_dir / "harnesses" / lbl
+            for lbl, _rv in labelled
+            if (work_dir / "harnesses" / lbl).is_dir()
+        },
+    )
+    summary["traces"] = str(work_dir / "traces")
     (work_dir / "summary.json").write_text(
         json.dumps(summary, indent=2, default=str), encoding="utf-8"
     )
diff --git a/packages/flow/openadapt_flow/integrations/harness2/traces.py b/packages/flow/openadapt_flow/integrations/harness2/traces.py
new file mode 100644
index 000000000..7c56cf73e
--- /dev/null
+++ b/packages/flow/openadapt_flow/integrations/harness2/traces.py
@@ -0,0 +1,423 @@
+"""Full execution traces plus the reward, for an outer-loop proposer.
+
+Meta-Harness (arXiv:2603.28052) gives its proposer a filesystem holding every
+earlier candidate's code, full execution traces and scores, and reports that
+full traces beat summaries (56.7 vs 38.7 on its text-classification ablation).
+This module writes that filesystem for GUI work. Each trace keeps every step
+in full: what the screen showed, the action, the reason given, any error, and,
+for each write, whether the record on screen was the requested one. Each
+trace ends with the read-back: which records the system of record gained,
+whose they are, the outcome and the reward.
+
+Two sources feed it:
+
+* an agent rollout directory (``rollout.run_rollout``), with its task, so the
+  write steps can be checked against the requested patient;
+* a compiled program's run directory (``report.json``, a Flow ``RunReport``),
+  whose steps already carry the runtime's identity-gate verdicts and effect
+  checks.
+
+Layout (schema ``openadapt.trace-export/v1``)::
+
+    <out>/INDEX.md                      one line per trace, best reward first
+    <out>/index.jsonl                   the same, one JSON object per trace
+    <out>/candidates/<name>/harness/    the harness each candidate ran, if given
+    <out>/traces/<trace_id>/trace.md    the full trace, readable
+    <out>/traces/<trace_id>/trace.json  the full trace, structured
+
+Screen text goes through :func:`openadapt_flow.privacy.scrub_text` before it is
+written, like ``REPORT.md``. The identity verdicts are computed before
+scrubbing, so they survive it.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import shutil
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any, Iterable, Optional, Sequence
+
+from openadapt_flow import privacy
+from openadapt_flow.integrations.harness2.rollout import (
+    load_run,
+    load_score,
+    load_steps,
+)
+from openadapt_flow.integrations.harness2.scorer import EpisodeScore
+from openadapt_flow.integrations.harness2.tasks import GuiTask
+
+SCHEMA = "openadapt.trace-export/v1"
+
+#: A step whose control label matches this is treated as a write.
+_WRITE_LABEL = re.compile(r"^\s*(save|submit|sign|confirm|finali[sz]e|post)\b", re.I)
+_BANNER = re.compile(
+    r"^(?P<name>[^\n—]+?) — MRN (?P<mrn>\S+) — DOB (?P<dob>[0-9-]+)", re.M
+)
+
+
+@dataclass
+class IdentityVerdict:
+    """Was the record on screen the requested one when a write happened?"""
+
+    status: str  # verified | mismatch | unreadable
+    mode: str  # agent_screen | <runtime identity mode>
+    expected: str = ""
+    observed: str = ""
+
+
+@dataclass
+class TraceStep:
+    index: int
+    action: str
+    observation: str = ""
+    reason: str = ""
+    error: str = ""
+    ok: Optional[bool] = None
+    write: bool = False
+    identity: Optional[IdentityVerdict] = None
+    effect: str = ""
+
+
+@dataclass
+class Trace:
+    trace_id: str
+    source: str  # agent_rollout | flow_run
+    task_id: str
+    candidate: str
+    outcome: str
+    reward: Optional[float]
+    passed: Optional[bool]
+    steps: list[TraceStep] = field(default_factory=list)
+    readback: dict[str, Any] = field(default_factory=dict)
+    report: str = ""
+    cost_usd: float = 0.0
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"schema": SCHEMA, **asdict(self)}
+
+    def summary(self) -> dict[str, Any]:
+        writes = [s for s in self.steps if s.write]
+        return {
+            "schema": SCHEMA,
+            "trace_id": self.trace_id,
+            "source": self.source,
+            "task_id": self.task_id,
+            "candidate": self.candidate,
+            "outcome": self.outcome,
+            "reward": self.reward,
+            "passed": self.passed,
+            "steps": len(self.steps),
+            "writes": len(writes),
+            "identity_mismatches": sum(
+                1 for s in writes if s.identity and s.identity.status == "mismatch"
+            ),
+            "wrong_record": bool(self.readback.get("wrong_record")),
+            "duplicate": bool(self.readback.get("duplicate")),
+            "cost_usd": self.cost_usd,
+        }
+
+
+# -- sources ---------------------------------------------------------------------
+
+
+def screen_identity(observation: str, task: GuiTask) -> IdentityVerdict:
+    """Identity at a write, from the patient banner the agent was looking at."""
+
+    want_name = task.params.get("patient_name", "")
+    want_dob = task.params.get("dob", "")
+    expected = f"{want_name} — DOB {want_dob}".strip(" —")
+    m = _BANNER.search(observation)
+    if not m:
+        return IdentityVerdict("unreadable", "agent_screen", expected, "")
+    observed = f"{m.group('name').strip()} — DOB {m.group('dob')}"
+    same = m.group("name").strip() == want_name and (
+        not want_dob or m.group("dob") == want_dob
+    )
+    return IdentityVerdict(
+        "verified" if same else "mismatch", "agent_screen", expected, observed
+    )
+
+
+def _control_label(observation: str, element: Any) -> str:
+    m = re.search(rf"^\[{re.escape(str(element))}\] \w+ '([^']*)'", observation, re.M)
+    return m.group(1) if m else ""
+
+
+def from_rollout(run_dir: Path, task: GuiTask, *, candidate: str = "") -> Trace:
+    """A trace from one agent rollout directory."""
+
+    run_dir = Path(run_dir)
+    run = load_run(run_dir)
+    score = load_score(run_dir)
+    steps: list[TraceStep] = []
+    for raw in load_steps(run_dir):
+        action = dict(raw.get("action") or {})
+        obs = str(raw.get("observation", ""))
+        kind = action.get("action")
+        label = _control_label(obs, action.get("element")) if kind == "click" else ""
+        write = bool(label and _WRITE_LABEL.search(label))
+        steps.append(
+            TraceStep(
+                index=int(raw.get("index", len(steps))),
+                action=_rollout_action(action, label),
+                observation=_scrub(obs),
+                reason=_scrub(str(raw.get("thought", ""))),
+                error=str(raw.get("error", "")),
+                write=write,
+                identity=_scrubbed(screen_identity(obs, task)) if write else None,
+            )
+        )
+    return Trace(
+        trace_id=str(run.get("run_id") or run_dir.name),
+        source="agent_rollout",
+        task_id=task.task_id,
+        candidate=candidate,
+        outcome=score.outcome,
+        reward=score.reward,
+        passed=score.passed,
+        steps=steps,
+        readback=_readback(score),
+        report=_scrub(str(run.get("report", ""))),
+        cost_usd=float(run.get("agent_cost_usd", 0.0)),
+    )
+
+
+def from_flow_run(
+    run_dir: Path,
+    *,
+    task_id: str = "",
+    candidate: str = "",
+    score: Optional[EpisodeScore] = None,
+) -> Trace:
+    """A trace from a compiled program's run directory (``report.json``).
+
+    Identity verdicts are the runtime's own (``StepResult.identity``). A
+    read-back from the reward worker can be attached with ``score``; without
+    it the trace carries the run's own effect verdicts.
+    """
+
+    from openadapt_flow.ir import RunReport
+
+    run_dir = Path(run_dir)
+    report = RunReport.model_validate_json((run_dir / "report.json").read_text("utf-8"))
+    steps: list[TraceStep] = []
+    for i, r in enumerate(report.results):
+        ident = None
+        if r.identity is not None:
+            ident = IdentityVerdict(
+                r.identity.status,
+                r.identity.mode,
+                _scrub(r.identity.expected),
+                _scrub(r.identity.observed),
+            )
+        effect = ""
+        if r.effect_verified is not None:
+            effect = "confirmed" if r.effect_verified else "refuted"
+            if r.effect_results:
+                effect += ": " + "; ".join(r.effect_results)[:300]
+        steps.append(
+            TraceStep(
+                index=i,
+                action=f"{r.step_id}: {r.intent}",
+                error=r.error or "",
+                ok=r.ok,
+                write=r.risk == "irreversible" or r.effect_verified is not None,
+                identity=ident,
+                effect=effect,
+            )
+        )
+    outcome = score.outcome if score else str(report.execution_outcome or "")
+    return Trace(
+        trace_id=f"{report.workflow_name}-{report.started_at}",
+        source="flow_run",
+        task_id=task_id or report.workflow_name,
+        candidate=candidate,
+        outcome=outcome,
+        reward=score.reward if score else None,
+        passed=score.passed if score else (report.execution_outcome == "VERIFIED"),
+        steps=steps,
+        readback=_readback(score) if score else {},
+    )
+
+
+def _rollout_action(action: dict[str, Any], label: str) -> str:
+    kind = action.get("action")
+    if kind == "click":
+        return f"click [{action.get('element')}] {label!r}".rstrip()
+    if kind == "type":
+        return f"type into [{action.get('element')}]: {_scrub(str(action.get('text', '')))!r}"
+    if kind == "done":
+        return f"done status={action.get('status')}: {_scrub(str(action.get('report', '')))}"
+    return str(kind)
+
+
+def _readback(score: EpisodeScore) -> dict[str, Any]:
+    return {
+        "outcome": score.outcome,
+        "reason": score.reason,
+        "added": [
+            {k: (_scrub(str(v)) if k == "note" else v) for k, v in r.items()}
+            for r in score.added
+        ],
+        "subject_added": score.subject_added,
+        "other_added": score.other_added,
+        "wrong_record": score.wrong_record,
+        "duplicate": score.duplicate,
+        "safe_stop": score.safe_stop,
+        "reward": score.reward,
+        "unscored": score.unscored,
+        "receipt_id": score.receipt_id,
+        "oracle_tier": score.oracle_tier,
+    }
+
+
+def _scrubbed(verdict: IdentityVerdict) -> IdentityVerdict:
+    verdict.expected = _scrub(verdict.expected)
+    verdict.observed = _scrub(verdict.observed)
+    return verdict
+
+
+def _scrub(text: str) -> str:
+    return privacy.scrub_text(text) or ""
+
+
+# -- rendering -------------------------------------------------------------------
+
+
+def render(trace: Trace) -> str:
+    """The full trace as text. Nothing is summarised away."""
+
+    head = [
+        f"# Trace {trace.trace_id}",
+        "",
+        f"- task: {trace.task_id}",
+        f"- candidate: {trace.candidate or '(none)'}",
+        f"- source: {trace.source}",
+        f"- outcome: {trace.outcome}",
+        f"- reward: {trace.reward if trace.reward is not None else 'none (a person settles it)'}",
+        "",
+        "## Steps",
+    ]
+    body: list[str] = []
+    for s in trace.steps:
+        lines = [f"### Step {s.index}" + (" (write)" if s.write else "")]
+        if s.observation:
+            lines.append("Screen:\n```\n" + s.observation + "\n```")
+        lines.append(f"Action: {s.action}")
+        if s.reason:
+            lines.append(f"Reason given: {s.reason}")
+        if s.ok is not None:
+            lines.append(f"Step ok: {s.ok}")
+        if s.identity:
+            i = s.identity
+            lines.append(
+                f"Identity at this step: {i.status} ({i.mode}); expected "
+                f"{i.expected or '?'}; on screen {i.observed or 'nothing readable'}"
+            )
+        if s.effect:
+            lines.append(f"Effect check: {s.effect}")
+        if s.error:
+            lines.append(f"Error: {s.error}")
+        body.append("\n".join(lines))
+    tail = ["## Read-back from the system of record"]
+    rb = trace.readback
+    if not rb:
+        tail.append("No read-back was attached.")
+    else:
+        if not rb["added"]:
+            tail.append("- no record was added")
+        for r in rb["added"]:
+            tail.append(
+                f"- added: patient {r.get('patient_id')}, type {r.get('type')}, "
+                f"note {r.get('note')!r}"
+            )
+        tail.append(
+            f"- the requested patient gained {rb['subject_added']} record(s); "
+            f"other patients gained {rb['other_added']}"
+        )
+        tail.append(f"- outcome: {rb['outcome']} ({rb['reason'][:300]})")
+        tail.append(f"- receipt: {rb['receipt_id']} (oracle tier {rb['oracle_tier']})")
+    if trace.report:
+        tail += ["", "## The agent's own report", trace.report]
+    return "\n\n".join(["\n".join(head), *body, "\n".join(tail)]) + "\n"
+
+
+def export(
+    out_dir: Path,
+    traces: Sequence[Trace],
+    *,
+    harnesses: Optional[dict[str, Path]] = None,
+) -> Path:
+    """Write the Meta-Harness-style filesystem. Replaces ``out_dir``'s traces."""
+
+    out_dir = Path(out_dir)
+    (out_dir / "traces").mkdir(parents=True, exist_ok=True)
+    for name, src in sorted((harnesses or {}).items()):
+        dest = out_dir / "candidates" / _safe(name) / "harness"
+        if dest.exists():
+            shutil.rmtree(dest)
+        shutil.copytree(src, dest)
+    rows = []
+    for trace in traces:
+        d = out_dir / "traces" / _safe(trace.trace_id)
+        d.mkdir(parents=True, exist_ok=True)
+        (d / "trace.md").write_text(render(trace), encoding="utf-8")
+        (d / "trace.json").write_text(
+            json.dumps(trace.to_dict(), indent=2), encoding="utf-8"
+        )
+        rows.append(trace.summary())
+    rows.sort(
+        key=lambda r: (
+            -(r["reward"] if r["reward"] is not None else -0.01),
+            r["trace_id"],
+        )
+    )
+    (out_dir / "index.jsonl").write_text(
+        "".join(json.dumps(r) + "\n" for r in rows), encoding="utf-8"
+    )
+    lines = [
+        "# Traces",
+        "",
+        "Every trace is complete: each screen, action, reason and error, the identity",
+        "check at each write, and the read-back from the system of record.",
+        "",
+        "| trace | task | candidate | outcome | reward | wrong record | duplicate |",
+        "|---|---|---|---|---|---|---|",
+    ]
+    for r in rows:
+        lines.append(
+            f"| [{r['trace_id']}](traces/{_safe(r['trace_id'])}/trace.md) | {r['task_id']} | "
+            f"{r['candidate'] or '-'} | {r['outcome']} | "
+            f"{r['reward'] if r['reward'] is not None else '-'} | "
+            f"{'yes' if r['wrong_record'] else ''} | {'yes' if r['duplicate'] else ''} |"
+        )
+    (out_dir / "INDEX.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
+    return out_dir
+
+
+def _safe(name: str) -> str:
+    return re.sub(r"[^A-Za-z0-9_.-]+", "_", name).strip("_") or "trace"
+
+
+def from_rollouts(
+    run_dirs: Iterable[tuple[Path, GuiTask, str]],
+) -> list[Trace]:
+    """Traces for ``(run_dir, task, candidate)`` triples."""
+
+    return [from_rollout(d, t, candidate=c) for d, t, c in run_dirs]
+
+
+__all__ = [
+    "IdentityVerdict",
+    "SCHEMA",
+    "Trace",
+    "TraceStep",
+    "export",
+    "from_flow_run",
+    "from_rollout",
+    "from_rollouts",
+    "render",
+    "screen_identity",
+]
diff --git a/packages/flow/tests/test_harness2_adapter.py b/packages/flow/tests/test_harness2_adapter.py
index dd43f210a..550496e99 100644
--- a/packages/flow/tests/test_harness2_adapter.py
+++ b/packages/flow/tests/test_harness2_adapter.py
@@ -242,6 +242,20 @@ def test_reward_selection_finds_the_safe_harness(tmp_path, h2):
         in (ctx / "rollouts" / "base1.md").read_text()
     )
     assert not list(ctx.rglob("readback.json"))
+    assert (
+        "identity check at this write: mismatch"
+        in (ctx / "rollouts" / "base1.md").read_text()
+    )
+    # Every rollout of the task lands in a Meta-Harness-style trace export.
+    index = [
+        json.loads(x)
+        for x in (tmp_path / "work" / "traces" / "index.jsonl").read_text().splitlines()
+    ]
+    assert {r["candidate"] for r in index} == {"base1", "inv1", "dom1", "c1"}
+    assert index[-1]["wrong_record"] is True and index[-1]["identity_mismatches"] == 1
+    assert (
+        tmp_path / "work" / "traces" / "candidates" / "c1" / "harness" / "AGENTS.md"
+    ).is_file()
 
 
 def test_screen_evidence_hides_the_readback(tmp_path, h2):
diff --git a/packages/flow/tests/test_harness2_traces.py b/packages/flow/tests/test_harness2_traces.py
new file mode 100644
index 000000000..11b9486de
--- /dev/null
+++ b/packages/flow/tests/test_harness2_traces.py
@@ -0,0 +1,143 @@
+"""The trace export: full steps, identity at each write, and the read-back."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from openadapt_flow import privacy
+from openadapt_flow.integrations.harness2 import traces
+from openadapt_flow.integrations.harness2.agent import RULE_IDENTITY, ScriptedGuiAgent
+from openadapt_flow.integrations.harness2.apps import MockMedTarget
+from openadapt_flow.integrations.harness2.rollout import run_rollout
+from openadapt_flow.integrations.harness2.tasks import mockmed_tasks
+
+
+@pytest.fixture(autouse=True)
+def _plaintext(monkeypatch):
+    # Local writes stay plaintext unless a test installs a scrubber.
+    monkeypatch.setenv("OPENADAPT_FLOW_SCRUB", "off")
+
+
+@pytest.fixture(scope="module")
+def rollouts(tmp_path_factory):
+    pytest.importorskip("playwright.sync_api")
+    root = tmp_path_factory.mktemp("rollouts")
+    out = {}
+    for variant, harness in (
+        ("lookalike", ""),
+        ("plain", f"Check the {RULE_IDENTITY}."),
+    ):
+        task = mockmed_tasks((variant,))[0]
+        out[variant] = (
+            run_rollout(
+                task,
+                harness,
+                ScriptedGuiAgent(),
+                root / variant,
+                target=MockMedTarget(),
+                run_id=f"trace-{variant}",
+            ),
+            task,
+        )
+    return out
+
+
+def test_wrong_patient_write_is_a_mismatch_with_its_readback(rollouts):
+    run_dir, task = rollouts["lookalike"]
+    trace = traces.from_rollout(run_dir, task, candidate="base1")
+    writes = [s for s in trace.steps if s.write]
+    assert len(writes) == 1
+    assert writes[0].identity.status == "mismatch"
+    assert "Taylor Duplicate" in writes[0].identity.observed
+    assert "Jane Sample" in writes[0].identity.expected
+    assert trace.readback["wrong_record"] is True
+    assert trace.readback["other_added"] == 1 and trace.reward == -1.0
+    text = traces.render(trace)
+    # Full, not summarised: every step's screen is there.
+    assert text.count("Screen:\n```") == len(trace.steps)
+    assert "Identity at this step: mismatch" in text
+    assert "other patients gained 1" in text
+
+
+def test_right_patient_write_is_verified(rollouts):
+    run_dir, task = rollouts["plain"]
+    trace = traces.from_rollout(run_dir, task)
+    [write] = [s for s in trace.steps if s.write]
+    assert write.identity.status == "verified"
+    assert trace.outcome == "verified" and trace.passed is True
+
+
+def test_export_writes_the_filesystem(rollouts, tmp_path):
+    items = [traces.from_rollout(d, t, candidate=v) for v, (d, t) in rollouts.items()]
+    harness = tmp_path / "h"
+    harness.mkdir()
+    (harness / "AGENTS.md").write_text("# rules\n")
+    out = traces.export(tmp_path / "out", items, harnesses={"plain": harness})
+    rows = [json.loads(x) for x in (out / "index.jsonl").read_text().splitlines()]
+    assert [r["candidate"] for r in rows] == ["plain", "lookalike"]  # best reward first
+    assert rows[0]["schema"] == traces.SCHEMA
+    assert rows[1]["identity_mismatches"] == 1
+    assert "lookalike" in (out / "INDEX.md").read_text()
+    for r in rows:
+        d = out / "traces" / r["trace_id"]
+        assert (d / "trace.md").is_file()
+        assert json.loads((d / "trace.json").read_text())["trace_id"] == r["trace_id"]
+    assert (
+        out / "candidates" / "plain" / "harness" / "AGENTS.md"
+    ).read_text() == "# rules\n"
+
+
+def test_screen_text_is_scrubbed_but_the_verdict_survives(rollouts, monkeypatch):
+    class Names:
+        def scrub_text(self, text, is_separated=False):
+            return text.replace("Taylor Duplicate", "<PERSON>").replace(
+                "Jane Sample", "<PERSON>"
+            )
+
+    monkeypatch.setenv("OPENADAPT_FLOW_SCRUB", "auto")
+    privacy.set_text_scrubber(Names())
+    try:
+        run_dir, task = rollouts["lookalike"]
+        trace = traces.from_rollout(run_dir, task)
+    finally:
+        privacy.reset_scrubbers()
+    text = traces.render(trace)
+    assert "Taylor Duplicate" not in text and "Jane Sample" not in text
+    [write] = [s for s in trace.steps if s.write]
+    assert write.identity.status == "mismatch"
+
+
+def test_flow_run_report_carries_runtime_identity(tmp_path):
+    from openadapt_flow.ir import IdentityCheck, RunReport, StepResult
+
+    report = RunReport(
+        workflow_name="triage",
+        started_at="2026-10-18T00:00:00Z",
+        execution_outcome="HALTED",
+        results=[
+            StepResult(step_id="step_001", intent="Open the referral", ok=True),
+            StepResult(
+                step_id="step_005",
+                intent="Click Save Encounter",
+                ok=False,
+                risk="irreversible",
+                identity=IdentityCheck(
+                    status="mismatch",
+                    mode="structured",
+                    expected="Jane Sample",
+                    observed="Taylor Duplicate",
+                ),
+                error="identity mismatch: refused before the click",
+            ),
+        ],
+    )
+    (tmp_path / "report.json").write_text(report.model_dump_json())
+    trace = traces.from_flow_run(tmp_path, task_id="mockmed/lookalike")
+    assert trace.source == "flow_run" and trace.outcome == "HALTED"
+    assert trace.passed is False
+    save = trace.steps[1]
+    assert save.write and save.identity.status == "mismatch"
+    assert save.identity.mode == "structured"
+    assert "refused before the click" in traces.render(trace)

From 392f0662d78865a956751ce914e3a7b5e8451ca1 Mon Sep 17 00:00:00 2001
From: abrichr <richard.abrich@mldsai.com>
Date: Sun, 18 Oct 2026 15:43:53 -0400
Subject: [PATCH 2/2] fix(flow): read the patient banner from OCR text in trace
 identity

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../openadapt_flow/integrations/harness2/traces.py   | 10 ++++++++--
 packages/flow/tests/test_harness2_traces.py          | 12 ++++++++++++
 2 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/packages/flow/openadapt_flow/integrations/harness2/traces.py b/packages/flow/openadapt_flow/integrations/harness2/traces.py
index 7c56cf73e..81eb05339 100644
--- a/packages/flow/openadapt_flow/integrations/harness2/traces.py
+++ b/packages/flow/openadapt_flow/integrations/harness2/traces.py
@@ -52,8 +52,10 @@
 
 #: A step whose control label matches this is treated as a write.
 _WRITE_LABEL = re.compile(r"^\s*(save|submit|sign|confirm|finali[sz]e|post)\b", re.I)
+#: The patient banner, as the DOM shows it or as OCR reads it (spaces dropped).
 _BANNER = re.compile(
-    r"^(?P<name>[^\n—]+?) — MRN (?P<mrn>\S+) — DOB (?P<dob>[0-9-]+)", re.M
+    r"^(?P<name>[^\n—|]+?)\s*—\s*MRN\s*(?P<mrn>\S+?)\s*—\s*DOB\s*(?P<dob>[0-9-]+)",
+    re.M,
 )
 
 
@@ -132,7 +134,7 @@ def screen_identity(observation: str, task: GuiTask) -> IdentityVerdict:
     if not m:
         return IdentityVerdict("unreadable", "agent_screen", expected, "")
     observed = f"{m.group('name').strip()} — DOB {m.group('dob')}"
-    same = m.group("name").strip() == want_name and (
+    same = _squash(m.group("name")) == _squash(want_name) and (
         not want_dob or m.group("dob") == want_dob
     )
     return IdentityVerdict(
@@ -140,6 +142,10 @@ def screen_identity(observation: str, task: GuiTask) -> IdentityVerdict:
     )
 
 
+def _squash(text: str) -> str:
+    return re.sub(r"\s+", "", text).lower()
+
+
 def _control_label(observation: str, element: Any) -> str:
     m = re.search(rf"^\[{re.escape(str(element))}\] \w+ '([^']*)'", observation, re.M)
     return m.group(1) if m else ""
diff --git a/packages/flow/tests/test_harness2_traces.py b/packages/flow/tests/test_harness2_traces.py
index 11b9486de..abdcec1da 100644
--- a/packages/flow/tests/test_harness2_traces.py
+++ b/packages/flow/tests/test_harness2_traces.py
@@ -69,6 +69,18 @@ def test_right_patient_write_is_verified(rollouts):
     assert trace.outcome == "verified" and trace.passed is True
 
 
+def test_identity_reads_an_ocr_banner():
+    task = mockmed_tasks(("plain",))[0]
+    # OCR on a pixel surface drops the spaces around the separators.
+    assert (
+        traces.screen_identity("JaneSample—MRNP1—DOB 1980-01-01", task).status
+        == "verified"
+    )
+    wrong = traces.screen_identity("TaylorDuplicate—MRNP0—DOB 1982-12-12", task)
+    assert wrong.status == "mismatch"
+    assert traces.screen_identity("Referral Tasks", task).status == "unreadable"
+
+
 def test_export_writes_the_filesystem(rollouts, tmp_path):
     items = [traces.from_rollout(d, t, candidate=v) for v, (d, t) in rollouts.items()]
     harness = tmp_path / "h"