abrichr · Oct 21, 2026 · Oct 21, 2026 · Oct 21, 2026 · Oct 21, 2026
@@ -242,8 +242,10 @@ def handoff(
 
     from openadapt_flow.compile_agent.pipeline import (
         PROPOSED_FILE,
+        RUN1_VALUES_FILE,
         FirstRunCompiler,
         _write_json,
+        _write_run1_values,
         check_run_record,
     )
     from openadapt_flow.compiler.compile import compile_recording
@@ -296,6 +298,9 @@ def handoff(
     shutil.copytree(source, kept)
     for stale in (PROPOSED_FILE, "verify.yaml", "record_checks.json", "admission.json"):
         (compiler.out_dir / stale).unlink(missing_ok=True)
+    # The guarded run is run 1: admission refuses canaries that repeat its
+    # input values, so keep their salted digests as compile-agent's run 1 does.
+    _write_run1_values(compiler.out_dir / RUN1_VALUES_FILE, values)
 
     # 1. Compile, as compile-agent does: distilled, effects mined from the
     # record reads the recording kept.

@@ -79,6 +79,9 @@
 CHECKS_FILE = "record_checks.json"
 TRACE_FILE = "code_action_trace.json"
 AGENT_READS_FILE = "agent_reads.json"
+#: Salted digests of run 1's input values, never the values. ``admit`` uses
+#: them to refuse canaries that repeat run 1's values.
+RUN1_VALUES_FILE = "run1_values.json"
 
 _PLACEHOLDER_RE = re.compile(r"{([A-Za-z_][A-Za-z0-9_]*)}")
 
@@ -600,13 +603,15 @@ def record(
         # earlier draft must not carry over to it.
         for stale in (
             AGENT_READS_FILE,
+            RUN1_VALUES_FILE,
             PROPOSED_FILE,
             CONFIRMED_FILE,
             CHECKS_FILE,
             "admission.json",
             "emitted.json",
         ):
             (self.out_dir / stale).unlink(missing_ok=True)
+        _write_run1_values(self.out_dir / RUN1_VALUES_FILE, values)
         attempt = self.dir / "attempts" / _stamp()
         harnessed = _HarnessedAgent(
             agent,
@@ -1014,6 +1019,9 @@ def admit(
                 "pass canary values: each canary saves them on the record "
                 f"{self.expect}, so choose values that are safe to write there",
             )
+        repeated = _repeats_run1(self.out_dir / RUN1_VALUES_FILE, self.task, rows)
+        if repeated is not None:
+            return self._result("draft", repeated)
         if admitter is not None:
             policy = getattr(admitter, "policy", None)
             count = policy.policy.min_clean_canaries if policy is not None else 1
@@ -1190,6 +1198,62 @@ def _confirmer(confirm: PersonOrCallback, compiler: FirstRunCompiler) -> Optiona
     return confirm(compiler.proposed())
 
 
+def _value_digest(salt: str, name: str, value: Any) -> str:
+    payload = json.dumps([salt, name, str(value)], ensure_ascii=False)
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _write_run1_values(path: Path, values: Mapping[str, Any]) -> None:
+    salt = uuid.uuid4().hex
+    _write_json(
+        path,
+        {
+            "salt": salt,
+            "digests": {
+                name: _value_digest(salt, name, value)
+                for name, value in sorted(values.items())
+            },
+        },
+    )
+
+
+def _repeats_run1(
+    path: Path, task: Any, rows: Sequence[Mapping[str, Any]]
+) -> Optional[str]:
+    """Why these canary rows can't prove the program, or None when they can.
+
+    A canary with run 1's values passes even when the program types the
+    demonstrated text as a constant, so it proves nothing about parameters.
+    Every parameter in every row must differ from run 1's value.
+    """
+
+    if not getattr(task, "params", None):
+        return None
+    if not path.is_file():
+        return (
+            "run 1's input digests are missing, so canary values can't be "
+            "checked against run 1's; record run 1 again"
+        )
+    data = json.loads(path.read_text())
+    salt, digests = data.get("salt"), data.get("digests") or {}
+    if not isinstance(salt, str) or not isinstance(digests, dict):
+        return "run 1's input digests are unreadable; record run 1 again"
+    for number, row in enumerate(rows, start=1):
+        values = task.validate(row)
+        same = sorted(
+            name
+            for name, value in values.items()
+            if name in digests and digests[name] == _value_digest(salt, name, value)
+        )
+        if same:
+            return (
+                f"canary {number} uses run 1's value for {', '.join(same)}; a "
+                "canary with run 1's values passes even when the program types "
+                "the demonstrated text, so choose different values"
+            )
+    return None
+
+
 def compile_first_run(
     prompt: str,
     *,

@@ -18,7 +18,10 @@
 
 A value is not lifted when it was typed inside longer text or never typed
 whole. Those inputs are reported so the reviewer sees them; the program would
-repeat the demonstrated text for them.
+repeat the demonstrated text for them. That includes a value typed whole in one
+step and inside longer text in another (a name typed into its field and again
+into a note): lifting only the whole step would leave the note repeating the
+demonstrated value on every run, with a constant that every check accepts.
 """
 
 from __future__ import annotations
@@ -59,16 +62,16 @@ def to_dict(self) -> dict[str, Any]:
         }
 
 
-def _match(workflow: Workflow, value: str) -> tuple[list[str], bool]:
+def _match(workflow: Workflow, value: str) -> tuple[list[str], list[str]]:
     whole: list[str] = []
-    inside = False
+    inside: list[str] = []
     for step in workflow.steps:
         if step.action != ActionKind.TYPE or step.param or step.text is None:
             continue
         if step.text == value:
             whole.append(step.id)
         elif value in step.text:
-            inside = True
+            inside.append(step.id)
     return whole, inside
 
 
@@ -166,6 +169,13 @@ def lift_params(bundle_dir: Path | str, inputs: Mapping[str, str]) -> LiftReport
                 "typed inside longer text" if inside else "not typed in any step"
             )
             continue
+        if inside:
+            report.unlifted[name] = (
+                f"typed whole in {', '.join(whole)} and inside longer text in "
+                f"{', '.join(inside)}; the longer text would repeat the "
+                "demonstrated value"
+            )
+            continue
         step_id = whole[0]
         if len(whole) > 1:
             picked, why = _pick_repeated(workflow, whole, value)

@@ -0,0 +1,58 @@
+"""A canary must not repeat run 1's input values.
+
+A program that types the demonstrated text as a constant passes a canary that
+uses run 1's values, so such a canary proves nothing about its parameters.
+Run 1's values are kept only as salted digests.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from openadapt_flow.compile_agent.pipeline import _repeats_run1, _write_run1_values
+
+
+class _Task:
+    params = {"patient": "str", "note": "str"}
+
+    @staticmethod
+    def validate(row):
+        return {name: str(value) for name, value in row.items()}
+
+
+def test_run1_values_are_stored_only_as_digests(tmp_path: Path) -> None:
+    path = tmp_path / "run1_values.json"
+    _write_run1_values(path, {"patient": "Jane Doe", "note": "Follow up"})
+    text = path.read_text()
+    assert "Jane Doe" not in text and "Follow up" not in text
+    assert set(json.loads(text)["digests"]) == {"patient", "note"}
+
+
+def test_a_canary_with_run1_values_is_refused(tmp_path: Path) -> None:
+    path = tmp_path / "run1_values.json"
+    _write_run1_values(path, {"patient": "Jane Doe", "note": "Follow up"})
+    why = _repeats_run1(
+        path,
+        _Task,
+        [
+            {"patient": "Ada Byron", "note": "Canary"},
+            {"patient": "Jane Doe", "note": "Canary 2"},
+        ],
+    )
+    assert why is not None
+    assert "canary 2" in why and "patient" in why
+    assert "Jane Doe" not in why
+
+
+def test_canaries_with_new_values_pass(tmp_path: Path) -> None:
+    path = tmp_path / "run1_values.json"
+    _write_run1_values(path, {"patient": "Jane Doe", "note": "Follow up"})
+    assert (
+        _repeats_run1(path, _Task, [{"patient": "Ada Byron", "note": "Canary"}]) is None
+    )
+
+
+def test_missing_run1_digests_fail_closed(tmp_path: Path) -> None:
+    why = _repeats_run1(tmp_path / "absent.json", _Task, [{"patient": "x"}])
+    assert why is not None and "record run 1 again" in why
@@ -11,6 +11,8 @@
 from typing import Any
 
 import pytest
+from hypothesis import given, settings
+from hypothesis import strategies as st
 
 from openadapt_flow.ir import ActionKind, Step, Workflow
 from openadapt_flow.jit import (
@@ -294,26 +296,25 @@ def _bundle(tmp_path: Path, texts: list[str]) -> Path:
 
 
 def test_lift_proposes_inputs_typed_whole_in_one_step(tmp_path: Path) -> None:
-    bundle = _bundle(tmp_path, ["Jane", "Follow up", "x Follow up x", "dup", "dup"])
+    bundle = _bundle(tmp_path, ["Jane", "Follow up", "x Follow up y", "dup", "dup"])
     report = lift_params(
-        bundle, {"name": "Jane", "note": "Follow up", "code": "dup", "missing": "zz"}
+        bundle,
+        {"name": "Jane", "note": "Follow up", "code": "dup", "missing": "zz"},
     )
-    assert [(p.name, p.step_id) for p in report.lifted] == [
-        ("name", "step_000"),
-        ("note", "step_001"),
-    ]
+    assert [(p.name, p.step_id) for p in report.lifted] == [("name", "step_000")]
     assert report.unlifted == {
+        "note": (
+            "typed whole in step_001 and inside longer text in step_002; the "
+            "longer text would repeat the demonstrated value"
+        ),
         "code": (
             "typed in 2 steps (step_003, step_004); no mined record check shows "
             "which field kept it"
         ),
         "missing": "not typed in any step",
     }
     proposals = json.loads((bundle / "param_proposals.json").read_text())["proposals"]
-    assert {p["name"]: p["source"] for p in proposals} == {
-        "name": "task_input",
-        "note": "task_input",
-    }
+    assert {p["name"]: p["source"] for p in proposals} == {"name": "task_input"}
     assert all("Jane" not in json.dumps(p) for p in proposals)  # masked only
 
 
@@ -324,6 +325,49 @@ def test_lift_reports_text_typed_inside_longer_text(tmp_path: Path) -> None:
     assert not (bundle / "param_proposals.json").exists()
 
 
+def test_lift_refuses_a_value_also_echoed_inside_longer_text(tmp_path: Path) -> None:
+    """A name typed into its field and again into a note is not lifted.
+
+    Lifting only the field would leave the note typing the demonstrated name
+    on every later run. The note is a constant, so every check accepts it.
+    """
+
+    bundle = _bundle(tmp_path, ["John Smith", "Called John Smith re: refill"])
+    report = lift_params(bundle, {"patient_name": "John Smith"})
+    assert report.lifted == []
+    assert "inside longer text in step_001" in report.unlifted["patient_name"]
+    assert not (bundle / "param_proposals.json").exists()
+
+
+@settings(max_examples=200, deadline=None)
+@given(
+    value=st.text(alphabet="abcXYZ 19", min_size=1, max_size=6),
+    texts=st.lists(st.text(alphabet="abcXYZ 19", max_size=12), max_size=6),
+)
+def test_lift_never_leaves_an_input_inside_a_lifted_workflow_literal(
+    tmp_path_factory: pytest.TempPathFactory, value: str, texts: list[str]
+) -> None:
+    """Property: once an input is lifted, no other typed step still contains it.
+
+    Every typed step whose text contains the value is either the lifted step
+    or the input is reported unlifted, so a person sees the leftover literal.
+    """
+
+    tmp = tmp_path_factory.mktemp("lift")
+    bundle = _bundle(tmp, texts)
+    report = lift_params(bundle, {"v": value})
+    for lifted in report.lifted:
+        others = [
+            f"step_{i:03d}"
+            for i, text in enumerate(texts)
+            if value in text and f"step_{i:03d}" != lifted.step_id
+        ]
+        assert all(sid in lifted.also_typed_in for sid in others), others
+        assert all(texts[int(sid[5:])] == value for sid in lifted.also_typed_in)
+    if not report.lifted:
+        assert "v" in report.unlifted
+
+
 def _detour_bundle(
     tmp_path: Path, labels: list[str], record: dict[str, str] | None
 ) -> Path: