fix(S2-1): canonically bind plan envelope, carry+require full identity, normalize read OSError

Closes the three S2-slice-1 blockers (arbiter). No new scope.

Blocker 1 — the validated-plan's authority envelope was only shape-checked (one type/one
file) and compared to the candidates by RAW dict equality, so a self-consistent forged
pair could smuggle a wrong type/file, wrong constraints, mismatched selected_findings, or
extra nested fields. validate_apply_inputs now builds the CANONICAL projection of
target_api / selection / source_files from the candidates' known keys and requires the
plan to EQUAL it exactly.

Blocker 2 — the code indexed c["event_identity"], but the shared candidates validator did
not require it, so a hash-bound candidate missing an identity field passed validation and
the apply gate raised a bare KeyError (violating "every violation -> ApplyError").
validate_candidates_bundle now requires the full identity set (event/source/handler +
*_identity + *_identity_kind) as strings, and the apply context carries all of them (not
just the human display) for the future span-node guard.

Blocker 3 — the preimage SHA read (_sha_file) sat AFTER `except CollectError`, so an
OSError between the root-confinement check and the read (file vanished / permission /
replaced) leaked as a traceback. The resolve + read are now under one try with an added
`except OSError -> ApplyError`.

tests/test_fix_apply.py grows to 32 checks: canonical-projection mismatches (wrong
type/file/constraints/selected_findings, unknown nested field in target/selection/source),
missing/wrong-typed candidate identity fields, the full-identity-context assertion, and a
source-read OSError (chmod-0, capability-guarded) -> ApplyError. ruff + mypy (ownlang) +
run_tests green; S1 unaffected by the stricter candidates validator.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JxKjqdGEFzq4UzZupw379G

Author: PhysShell

ownlang/fix_apply.py

@@ -114,10 +114,32 @@ def validate_apply_inputs(
     # Hash binding: the candidates are the plan's own input.
     if bundle_sha256(candidates) != validated_plan["input_bundle_sha256"]:
         raise ApplyError("candidates sha256 does not match validated-plan.input_bundle_sha256")
-    if validated_plan["target_api"] != candidates["target_api"]:
-        raise ApplyError("validated-plan.target_api does not match candidates.target_api")
-    if validated_plan["source_files"] != candidates["source_files"]:
-        raise ApplyError("validated-plan.source_files does not match candidates.source_files")
+
+    # The plan's materialized authority envelope must EQUAL the canonical projection of
+    # the candidates' known keys — not a raw dict copy. This rejects a self-consistent
+    # forged pair that smuggled extra nested fields, a wrong type/file, wrong constraints,
+    # or mismatched selected_findings into the plan.
+    the_type = candidates["selection"]["allowed_types"][0]
+    the_source = candidates["source_files"][0]
+    expected_target_api = {"subscribe": candidates["target_api"]["subscribe"]}
+    expected_selection = {
+        "allowed_types": [{"full_name": the_type["full_name"], "file": the_type["file"]}],
+        "selected_findings": candidates["selection"].get("selected_findings"),
+        "constraints": {
+            "max_types_changed": 1,
+            "max_files_changed": 1,
+            "allow_helper_changes": False,
+            "allow_config_changes": False,
+            "allow_suppressions": False,
+        },
+    }
+    expected_source_files = [{"path": the_source["path"], "sha256": the_source["sha256"]}]
+    if validated_plan["target_api"] != expected_target_api:
+        raise ApplyError("validated-plan.target_api is not the canonical projection of candidates")
+    if validated_plan["selection"] != expected_selection:
+        raise ApplyError("validated-plan.selection is not the canonical projection of candidates")
+    if validated_plan["source_files"] != expected_source_files:
+        raise ApplyError("validated-plan.source_files is not the canonical projection")
 
     candidate_by_id = {c["finding_id"]: c for c in candidates["candidates"]}
     seen: set[str] = set()
@@ -142,6 +164,9 @@ def validate_apply_inputs(
                 raise ApplyError(
                     f"{fid}: convert_acquire requires an inotify_property_changed contract"
                 )
+            # The FULL identity contract the future span-node guard needs (not just the
+            # human display). validate_candidates_bundle guarantees all are present + str,
+            # so the direct index cannot KeyError.
             convert.append({
                 "finding_id": fid,
                 "file": c["file"],
@@ -150,7 +175,11 @@ def validate_apply_inputs(
                 "event": c["event"],
                 "event_identity": c["event_identity"],
                 "source": c["source"],
+                "source_identity": c["source_identity"],
+                "source_identity_kind": c["source_identity_kind"],
                 "handler": c["handler"],
+                "handler_identity": c["handler_identity"],
+                "handler_identity_kind": c["handler_identity_kind"],
             })
         else:
             manual.append(fid)
@@ -167,13 +196,18 @@ def validate_apply_inputs(
         if start_b < end_a:
             raise ApplyError("overlapping convert_acquire spans")
 
-    # Source guard: confine to root, then match the pristine preimage SHA.
+    # Source guard: confine to root, then match the pristine preimage SHA. A read that
+    # fails between the confinement check and the hash (file vanished / permission /
+    # replaced) is normalized to ApplyError, never a leaked OSError.
     source = candidates["source_files"][0]
     try:
         canonical, abs_path = _resolve_source(root, source["path"])
+        actual_sha = _sha_file(abs_path)
     except CollectError as exc:
         raise ApplyError(f"source path: {exc}") from exc
-    if _sha_file(abs_path) != source["sha256"]:
+    except OSError as exc:
+        raise ApplyError(f"cannot read source file: {exc}") from exc
+    if actual_sha != source["sha256"]:
         raise ApplyError(f"STALE SOURCE / PREIMAGE MISMATCH for {canonical}")
 
     return {

ownlang/fix_candidates.py

@@ -416,7 +416,10 @@ def validate_candidates_bundle(bundle: Any) -> None:
             raise CollectError(f"{cctx}: containing_type must be {type_name!r}")
         if _field(c, "file", "str", cctx) != source_path:
             raise CollectError(f"{cctx}: file must be {source_path!r}")
-        for k in ("event", "source", "handler"):
+        # Display + full identity set — required so downstream (S2 apply) can index them
+        # without a KeyError and the span-node guard gets the real identity contract.
+        for k in ("event", "source", "handler", "event_identity", "source_identity",
+                  "source_identity_kind", "handler_identity", "handler_identity_kind"):
             _field(c, k, "str", cctx)
         contract = _field(c, "event_contract", "str", cctx)
         if contract not in _CONTRACTS:

tests/test_fix_apply.py

@@ -125,9 +125,52 @@ def raises(vplan: dict, cands: dict, root: str) -> bool:
         check(ctx["source_file"] == _REL
               and ctx["target_subscribe"] == "WeakEvents.AddPropertyChanged",
               "context carries source + target")
-        check(ctx["convert_acquire"][0]["source"] == "_pub"
-              and ctx["convert_acquire"][0]["handler"] == "OnChanged",
-              "convert target carries candidate identity")
+        target0 = ctx["convert_acquire"][0]
+        check(target0["source"] == "_pub" and target0["handler"] == "OnChanged",
+              "convert target carries candidate display identity")
+        check(set(target0) >= {"source_identity", "source_identity_kind", "handler_identity",
+                               "handler_identity_kind", "event_identity", "containing_type"},
+              "convert target carries the FULL identity contract")
+
+        # Blocker 1: the plan envelope must be the canonical projection of candidates.
+        def vgood() -> dict:
+            return _vplan(cands, [_decision(_FID_A, "convert_acquire", 100),
+                                  _decision(_FID_B, "manual_review", 200)])
+        v = vgood()
+        v["selection"]["allowed_types"][0]["full_name"] = "Other.Type"
+        check(raises(v, cands, root), "wrong selected type refused")
+        v = vgood()
+        v["selection"]["allowed_types"][0]["file"] = "N/Other.cs"
+        check(raises(v, cands, root), "wrong selected type file refused")
+        v = vgood()
+        v["selection"]["constraints"]["max_types_changed"] = 2
+        check(raises(v, cands, root), "wrong constraints refused")
+        v = vgood()
+        v["selection"]["selected_findings"] = [_FID_A]
+        check(raises(v, cands, root), "selected_findings mismatch refused")
+        v = vgood()
+        v["target_api"]["extra"] = 1
+        check(raises(v, cands, root), "unknown nested target_api field refused")
+        v = vgood()
+        v["selection"]["extra"] = 1
+        check(raises(v, cands, root), "unknown nested selection field refused")
+        v = vgood()
+        v["source_files"][0]["extra"] = 1
+        check(raises(v, cands, root), "unknown nested source_files field refused")
+
+        # Blocker 2: a candidate missing an identity field is a controlled ApplyError.
+        for missing in ("event_identity", "source_identity", "handler_identity",
+                        "handler_identity_kind"):
+            bc_cand = _cand(_FID_A, inpc, "inotify_property_changed", 100)
+            del bc_cand[missing]
+            bc = _bundle([bc_cand], sha)
+            bv = _vplan(bc, [_decision(_FID_A, "convert_acquire", 100)])
+            check(raises(bv, bc, root), f"candidate missing {missing} -> ApplyError")
+        bad_type_cand = _cand(_FID_A, inpc, "inotify_property_changed", 100)
+        bad_type_cand["source_identity"] = 42
+        bt = _bundle([bad_type_cand], sha)
+        btv = _vplan(bt, [_decision(_FID_A, "convert_acquire", 100)])
+        check(raises(btv, bt, root), "candidate identity of wrong type -> ApplyError")
 
         # hash binding: mutate candidates after the plan was built
         mutated = _bundle([_cand(_FID_A, inpc, "inotify_property_changed", 100),
@@ -202,6 +245,28 @@ def raises(vplan: dict, cands: dict, root: str) -> bool:
         # the file does not exist under root2 -> _resolve_source refuses (not a regular file)
         check(raises(ev, esc, root2), "missing/uncontained source refused")
 
+    # Blocker 3: an OSError reading the source (perms) is normalized to ApplyError.
+    with tempfile.TemporaryDirectory() as root3:
+        p = os.path.join(root3, "N", "C.cs")
+        os.makedirs(os.path.dirname(p), exist_ok=True)
+        body = b"class C {}\n"
+        with open(p, "wb") as fh:
+            fh.write(body)
+        os.chmod(p, 0)
+        unreadable = True
+        try:
+            with open(p, "rb"):
+                unreadable = False  # perms not enforced (Windows / running as root) -> skip
+        except OSError:
+            pass
+        if unreadable:
+            sha3 = "sha256:" + hashlib.sha256(body).hexdigest()
+            oc = _bundle([_cand(_FID_A, ("convert_acquire", "manual_review"),
+                                "inotify_property_changed", 100)], sha3)
+            ov = _vplan(oc, [_decision(_FID_A, "convert_acquire", 100)])
+            check(raises(ov, oc, root3), "source read OSError -> ApplyError")
+        os.chmod(p, 0o644)  # restore so the tempdir cleans up
+
     print(f"fix-apply (S2 slice 1): {ok} ok, {bad} bad")
     return bad