verify: sealed overlays (sealed_from) — candidates can't weaken hande…

examples/papers/agentcoder.py

@@ -5,10 +5,15 @@
 The test designer writes tests *without seeing the implementation* — tests
 written after the code inherit its blind spots. So both seats work in
 parallel before the freeze (h5i stamps both artifacts independent), then
-the programmer applies the granted test suite as sealed-phase materials and
-must satisfy it. The test executor role is not an LLM at all:
-``conductor.verify`` runs the suite neutrally, and each failure loops back
-to the programmer as a constructed review.
+the programmer is shown the test suite as sealed-phase materials and must
+satisfy it. The test executor role is not an LLM at all:
+``conductor.verify`` runs the suite neutrally with the designer's artifact
+as a **sealed overlay** (``sealed_from=tests``): the designer's test files
+are overlaid over the programmer's candidate at verify time, so the
+programmer *cannot* weaken or skip them — an edit to a sealed path is
+discarded and surfaced as ``sealed_overridden`` tamper evidence, and the
+programmer needn't even copy the tests into its tree. Each failure loops
+back to the programmer as a constructed review.
 
     python examples/papers/agentcoder.py ["<task>"]   # default: implement quicksort with pytest
 """
@@ -54,18 +59,27 @@ async def main(task: str) -> None:
         )
         await c.freeze()
 
-        # The programmer takes on the independent test suite as materials.
+        # The programmer sees the independent test suite as materials. It
+        # may copy the tests locally to iterate, but the copy carries no
+        # authority: verification below overlays the designer's originals.
         candidate = await programmer.work(
-            "Apply the granted teammate artifact — an independently designed "
-            "test suite for your task — into your worktree alongside your "
-            "implementation. Do not weaken or delete tests; adjust your "
-            "implementation until it honestly satisfies them.",
+            "The granted teammate artifact is an independently designed test "
+            "suite for your task. Make your implementation honestly satisfy "
+            "it — you may copy the tests into your worktree to run them "
+            "locally, but the neutral verifier always uses the designer's "
+            "originals, so editing them cannot help you.",
             materials=[tests],
         )
 
-        # Test executor loop: neutral runs, failures loop back as reviews.
+        # Test executor loop: neutral runs against the SEALED designer
+        # tests, failures loop back as reviews.
         for iteration in range(1, MAX_ITERATIONS + 1):
-            verification = await c.verify(candidate, VERIFY)
+            verification = await c.verify(candidate, VERIFY, sealed_from=tests)
+            if verification.sealed_overridden:
+                print(
+                    "note: candidate edits to sealed test paths were "
+                    f"discarded: {', '.join(verification.sealed_overridden)}"
+                )
             if verification.applies_cleanly and verification.tests_passed:
                 print(f"iteration {iteration}: test executor is green")
                 break
@@ -81,8 +95,9 @@ async def main(task: str) -> None:
                     round=candidate.round,
                     body=(
                         "Verdict: REVISE\n\nThe independently designed test "
-                        f"suite failed in a neutral worktree:\n{evidence}\n\n"
-                        "Fix the implementation; do not touch the tests."
+                        "suite (sealed — your copy of it carries no authority) "
+                        f"failed in a neutral worktree:\n{evidence}\n\n"
+                        "Fix the implementation; editing tests cannot help."
                     ),
                     referenced_artifacts=(candidate.id,),
                 ),

src/h5i/orchestra/_conductor.py

@@ -353,10 +353,25 @@ async def verify(
         command: Sequence[str],
         *,
         isolation: str | None = None,
+        sealed_from: "Artifact | str | None" = None,
     ) -> Verification:
         """Neutrally re-execute ``command`` against the artifact owner's
         latest submission in a fresh sandboxed worktree — never the author's
-        box."""
+        box.
+
+        By default the command runs against the candidate's own submitted
+        tree — appropriate when tests are part of the deliverable under
+        review. ``sealed_from`` applies a **sealed overlay** instead: pass
+        the sealing agent's ``Artifact`` (or a submission id / team agent
+        id), and that submission's base..commit diff is overlaid over the
+        candidate before the command runs, so the candidate cannot weaken
+        checks it was handed — its edits to sealed paths are discarded and
+        recorded as ``Verification.sealed_overridden`` tamper evidence.
+        Typically the sealed content is an independently designed test set;
+        the mechanism is content-agnostic (golden files, scoring harnesses).
+        The sealing agent must differ from the candidate's owner
+        (self-sealing fails closed), and the built-in verdict refuses to
+        compare candidates verified against divergent sealed sets."""
         if isinstance(command, (str, bytes)):
             raise TypeError(
                 "verify(command=...) takes an argv sequence like "
@@ -368,6 +383,10 @@ async def verify(
         }
         if isolation is not None:
             params["isolation"] = isolation
+        if sealed_from is not None:
+            params["sealed_from"] = (
+                sealed_from.id if isinstance(sealed_from, Artifact) else sealed_from
+            )
         return Verification.from_raw(await self._request("conductor.verify", params))
 
     async def judge(self, policy: _policy.Policy = _policy.tests_then_smallest_diff) -> Verdict:

src/h5i/orchestra/_types.py

@@ -171,8 +171,24 @@ class Verification:
     isolation: str
     capture_id: str | None
     failure: str | None
+    #: sealed-overlay mode (non-None = sealed): the submission id whose
+    #: base..commit diff was overlaid over the candidate before the command
+    #: ran — the candidate's edits to those paths could not weaken the check
+    sealed_from: str | None
+    #: sealed mode only: tree OID of the sealing submission (the content
+    #: digest the verdict compares across candidates)
+    sealed_tree_oid: str | None
+    #: sealed mode only: the paths pinned by the overlay
+    sealed_paths: tuple[str, ...]
+    #: sealed paths where a candidate edit was discarded by the overlay —
+    #: tamper evidence, surfaced rather than silently dropped
+    sealed_overridden: tuple[str, ...]
     raw: Mapping[str, Any] = field(repr=False)
 
+    @property
+    def sealed(self) -> bool:
+        return self.sealed_from is not None
+
     @classmethod
     def from_raw(cls, raw: Mapping[str, Any]) -> "Verification":
         return cls(
@@ -186,6 +202,10 @@ def from_raw(cls, raw: Mapping[str, Any]) -> "Verification":
             isolation=_s(raw, "isolation", "unknown"),
             capture_id=raw.get("capture_id"),
             failure=raw.get("failure"),
+            sealed_from=raw.get("sealed_from"),
+            sealed_tree_oid=raw.get("sealed_tree_oid"),
+            sealed_paths=tuple(raw.get("sealed_paths") or ()),
+            sealed_overridden=tuple(raw.get("sealed_overridden") or ()),
             raw=dict(raw),
         )
 

src/h5i/orchestra/patterns.py

@@ -150,16 +150,22 @@ async def verify_and_judge(
     *,
     verify: Sequence[str] | None = None,
     isolation: str | None = None,
+    sealed_from: Artifact | str | None = None,
     judge: Policy | None = None,
 ) -> Verdict | None:
     """The shared pattern tail: neutrally verify each artifact — one at a
     time, verify worktrees share on-disk state and parallel creation is racy
     — then record a verdict. An explicit ``judge`` always wins; otherwise,
     when a verifier ran, the CLI's finalize rule (`tests_then_smallest_diff`)
-    applies; with neither, nothing is recorded and ``None`` comes back."""
+    applies; with neither, nothing is recorded and ``None`` comes back.
+    ``sealed_from`` seals part of every verified tree to one submission's
+    content (see ``Conductor.verify``) — typically an independent test set;
+    its owner must not be one of the candidates."""
     if verify is not None:
         for artifact in artifacts:
-            await c.verify(artifact, verify, isolation=isolation)
+            await c.verify(
+                artifact, verify, isolation=isolation, sealed_from=sealed_from
+            )
     if judge is not None:
         return await c.judge(judge)
     if verify is not None:
@@ -190,6 +196,7 @@ async def ensemble(
     rounds: int = 1,
     verify: Sequence[str] | None = None,
     isolation: str | None = None,
+    sealed_from: Artifact | str | None = None,
     judge: Policy | None = None,
     approve: Callable[[Review], bool] = approves,
 ) -> EnsembleOutcome:
@@ -227,7 +234,12 @@ async def ensemble(
 
     # 4-5. Neutral verification + verdict, the shared tail.
     verdict = await verify_and_judge(
-        c, list(latest.values()), verify=verify, isolation=isolation, judge=judge
+        c,
+        list(latest.values()),
+        verify=verify,
+        isolation=isolation,
+        sealed_from=sealed_from,
+        judge=judge,
     )
 
     return EnsembleOutcome(
@@ -255,6 +267,7 @@ async def integrate(
     *,
     verify: Sequence[str] | None = None,
     isolation: str | None = None,
+    sealed_from: Artifact | str | None = None,
 ) -> IntegrateOutcome:
     """The multi-implementer merge seat: seal the round, then one integrator
     fuses ``parts`` in its own env — granted their diffs as materials,
@@ -273,7 +286,9 @@ async def integrate(
     )
     verification = None
     if verify is not None:
-        verification = await c.verify(merged, verify, isolation=isolation)
+        verification = await c.verify(
+            merged, verify, isolation=isolation, sealed_from=sealed_from
+        )
     return IntegrateOutcome(merged=merged, verification=verification)
 
 
@@ -317,6 +332,7 @@ async def arena(
     *,
     verify: Sequence[str] | None = None,
     isolation: str | None = None,
+    sealed_from: Artifact | str | None = None,
     judge: Policy | None = None,
 ) -> ArenaOutcome:
     """Independent attempts, ranked: N agents try the same task with no
@@ -330,7 +346,12 @@ async def arena(
     )
     await c.freeze()
     verdict = await verify_and_judge(
-        c, artifacts, verify=verify, isolation=isolation, judge=judge
+        c,
+        artifacts,
+        verify=verify,
+        isolation=isolation,
+        sealed_from=sealed_from,
+        judge=judge,
     )
     rows = await c.compare()
     return ArenaOutcome(artifacts=artifacts, rows=rows, verdict=verdict)

tests/test_conductor.py

@@ -99,6 +99,50 @@ async def test_work_round_trip_preserves_unknown_fields():
         await c.close()
 
 
+async def test_verify_sealed_from_passes_id_and_parses_seal_fields():
+    mock = MockOrchestra()
+    mock.on("agent.hire", lambda p: {"agent_id": p["name"], "env_id": f"env/{p['name']}/x"})
+    mock.on("agent.work", lambda p: artifact_raw(p["agent"], id_=f"sha:{p['agent']}"))
+    mock.on("conductor.verify", lambda p: {
+        "id": "v1", "submission_id": p["artifact"]["id"], "owner_agent": "coder",
+        "round": 1, "command": p["command"], "applies_cleanly": True,
+        "tests_passed": False, "isolation": "workspace",
+        "sealed_from": p["sealed_from"], "sealed_tree_oid": "tree:tests",
+        "sealed_paths": ["tests.sh"], "sealed_overridden": ["tests.sh"],
+    })
+    c = await launch_conductor(mock)
+    try:
+        coder = await c.hire("coder", runtime="claude")
+        designer = await c.hire("designer", runtime="codex")
+        candidate = await coder.work("implement")
+        tests = await designer.work("design tests")
+
+        # An Artifact seals by its submission id; a plain string passes through.
+        v = await c.verify(candidate, ["sh", "tests.sh"], sealed_from=tests)
+        assert mock.calls_to("conductor.verify")[-1]["sealed_from"] == tests.id
+        await c.verify(candidate, ["sh", "tests.sh"], sealed_from="designer")
+        assert mock.calls_to("conductor.verify")[-1]["sealed_from"] == "designer"
+
+        # The seal evidence round-trips into the typed Verification.
+        assert v.sealed
+        assert v.sealed_from == tests.id
+        assert v.sealed_tree_oid == "tree:tests"
+        assert v.sealed_paths == ("tests.sh",)
+        assert v.sealed_overridden == ("tests.sh",)
+
+        # Unsealed requests omit the parameter entirely.
+        mock.on("conductor.verify", lambda p: {
+            "id": "v2", "submission_id": p["artifact"]["id"], "owner_agent": "coder",
+            "round": 1, "command": p["command"], "applies_cleanly": True,
+            "tests_passed": True, "isolation": "workspace",
+        })
+        plain = await c.verify(candidate, ["sh", "tests.sh"])
+        assert "sealed_from" not in mock.calls_to("conductor.verify")[-1]
+        assert not plain.sealed and plain.sealed_paths == ()
+    finally:
+        await c.close()
+
+
 async def test_step_commit_replay_and_abort():
     mock = MockOrchestra()
     journal: dict[str, object] = {}

tests/test_patterns.py

@@ -288,6 +288,25 @@ async def test_verify_and_judge_fallback_policy_and_none():
         await c.close()
 
 
+async def test_verify_and_judge_threads_sealed_from():
+    mock = MockOrchestra()
+    wire_basics(mock)
+    c = await launch_conductor(mock)
+    try:
+        coder = await c.hire("coder")
+        designer = await c.hire("designer")
+        candidate = await coder.work("implement")
+        tests = await designer.work("design tests")
+        await c.freeze()
+        await patterns.verify_and_judge(
+            c, [candidate], verify=["sh", "tests.sh"], sealed_from=tests
+        )
+        (call,) = mock.calls_to("conductor.verify")
+        assert call["sealed_from"] == tests.id
+    finally:
+        await c.close()
+
+
 async def test_judge_panel_custom_aggregate():
     mock = MockOrchestra()
     wire_basics(mock)