diff --git a/notes/process-evidence-schema.md b/notes/process-evidence-schema.md
index ba4e37c..94aedd4 100644
--- a/notes/process-evidence-schema.md
+++ b/notes/process-evidence-schema.md
@@ -76,6 +76,27 @@ Saved `ss -H -a -n -t -u -p` output can connect a TCP or UDP socket to a PID whe
 
 Missing process context is not converted into false certainty. The socket remains in the artifact with `linked=false`.
 
+`process_name` on a socket link is the name reported by `ss`, including when it
+differs from the PID-associated procfs process's `name`. This preserves both
+observations. A name disagreement can follow legitimate changes and does not prove
+PID reuse or misconduct; agreement cannot rule out reuse between saved captures.
+
+`linked=true` describes an unambiguous PID association in the supplied records.
+It does not independently verify that `ss` observed the procfs execution instance.
+The link's `process_id`, executable, UID and GID are procfs-derived context. Duplicate
+PID candidates fail instead of selecting the first or last record. During `diff`,
+process IDs must agree with their declared host/PID/start fields, and linked records
+must reference the paired process record with matching PID, executable, UID and GID.
+An unlinked record must not claim that procfs identity or context. Contradictory raw
+artifacts remain available; a failed comparison produces no downstream diff.
+
+The telemetry adapter checks change-identity self-consistency. Its diff input does
+not contain the complete paired procfs artifacts, so it cannot repeat cohort
+validation or prove instance ownership. Names are contextual and excluded from the
+existing link identity: a name-only difference does not create a link-change event.
+Rejecting name mismatches as an identity test would both lose legitimate observations
+and still miss PID/name reuse. No confidence or verification field is introduced.
+
 `wildcard` means the socket was bound to an any-address value such as `0.0.0.0` or `::`. It does not prove external reachability because firewall, namespace, routing, and host policy are outside this snapshot.
 
 ## Diff interpretation
diff --git a/projects/linux-process-observe/README.md b/projects/linux-process-observe/README.md
index a8b2650..c30f5b1 100644
--- a/projects/linux-process-observe/README.md
+++ b/projects/linux-process-observe/README.md
@@ -95,7 +95,10 @@ The JSONL can be supplied as `input_path` to telemetry-lab's existing `run windo
 - `parent_process_id` links to the same snapshot identity when the parent PID is present in the saved procfs export; otherwise it is `null`.
 - UID/GID fields come from the saved procfs `status` record. Parent PID and start time come from `stat`.
 - The executable path is context, not proof of the executable file's contents or integrity.
-- Socket links use PID context reported by `ss`. A socket is retained with `linked=false` when no process context is available.
+- Socket links use PID context reported by `ss`. `linked=true` means an unambiguous PID association within the supplied records, not independent verification of the socket's execution instance. A socket is retained with `linked=false` when no matching procfs context is available.
+- A link's `process_name` preserves the name reported by `ss`; the selected process's name remains in `process_snapshot.json`. A disagreement is retained as context, without proving PID reuse or malicious behavior. Names can legitimately change, and matching names cannot exclude PID reuse between source captures.
+- Link `process_id`, executable, UID and GID come from the PID-associated procfs record. Duplicate PID candidates are rejected rather than reconciled by input order. `diff` rejects process IDs inconsistent with their host/PID/start fields, dangling linked identities, and links whose procfs-derived fields disagree with the paired process artifact, even when timestamps match.
+- The standalone telemetry adapter validates identity self-consistency; it does not have the paired procfs artifact and cannot independently revalidate a socket's process attribution. Link-name-only changes do not create new events: the link diff remains added/removed by the existing identity and endpoint key.
 - `exposure=loopback`, `wildcard`, or `specific` describes the local bind address; it is not a firewall or reachability verdict.
 
 ## Validation Status
diff --git a/projects/linux-process-observe/src/linux_process_observe/adapters.py b/projects/linux-process-observe/src/linux_process_observe/adapters.py
index 3934551..ae56365 100644
--- a/projects/linux-process-observe/src/linux_process_observe/adapters.py
+++ b/projects/linux-process-observe/src/linux_process_observe/adapters.py
@@ -4,7 +4,7 @@
 from typing import Any
 import json
 
-from .models import EVIDENCE_SCHEMA, EvidenceEnvelope
+from .models import EVIDENCE_SCHEMA, EvidenceEnvelope, socket_link_identity, validate_process_record_identity
 
 
 _EVENT_TYPES = {
@@ -90,10 +90,12 @@ def _map_record(
     if event_prefix == "process":
         for process_record in (before, after):
             if process_record is not None:
-                _validate_process_identity(process_record, identity, index)
+                _validate_process_identity(process_record, identity, index, diff.host_id)
         source = _required_string(selected, "process_id", index)
         target = _required_string(selected, "executable", index)
     else:
+        if socket_link_identity(selected) != identity:
+            raise ValueError(f"record {index} socket identity must match observation")
         source = _socket_source(diff.host_id, selected, index)
         target = _socket_target(selected, index)
 
@@ -166,11 +168,12 @@ def _change_records(
 
 
 def _validate_process_identity(
-    record: dict[str, Any], identity: str, index: int
+    record: dict[str, Any], identity: str, index: int, host_id: str
 ) -> None:
     process_id = _required_string(record, "process_id", index)
     if process_id != identity:
         raise ValueError(f"record {index} process_id must match identity")
+    validate_process_record_identity(record, host_id)
 
 
 def _socket_source(host_id: str, record: dict[str, Any], index: int) -> str:
diff --git a/projects/linux-process-observe/src/linux_process_observe/diff.py b/projects/linux-process-observe/src/linux_process_observe/diff.py
index bb75ca8..f2e9d3d 100644
--- a/projects/linux-process-observe/src/linux_process_observe/diff.py
+++ b/projects/linux-process-observe/src/linux_process_observe/diff.py
@@ -3,7 +3,10 @@
 from datetime import datetime
 from typing import Any
 
-from .models import EVIDENCE_SCHEMA, EvidenceEnvelope, parse_observed_at
+from .models import (
+    EVIDENCE_SCHEMA, EvidenceEnvelope, parse_observed_at,
+    socket_link_identity, validate_process_record_identity,
+)
 
 
 def build_diff_envelope(
@@ -15,10 +18,10 @@ def build_diff_envelope(
 ) -> EvidenceEnvelope:
     _validate_artifacts(before_processes, after_processes, before_links, after_links)
 
-    before_process_map = _records_by_key(before_processes, record_type="process", key="process_id")
-    after_process_map = _records_by_key(after_processes, record_type="process", key="process_id")
-    before_link_map = _links_by_identity(before_links)
-    after_link_map = _links_by_identity(after_links)
+    before_process_map = _processes_by_identity(before_processes)
+    after_process_map = _processes_by_identity(after_processes)
+    before_link_map = _links_by_identity(before_links, before_process_map)
+    after_link_map = _links_by_identity(after_links, after_process_map)
 
     changes: list[dict[str, Any]] = []
     for identity in sorted(after_process_map.keys() - before_process_map.keys()):
@@ -88,37 +91,55 @@ def _records(envelope: EvidenceEnvelope, record_type: str) -> list[dict[str, Any
     return records
 
 
-def _records_by_key(envelope: EvidenceEnvelope, *, record_type: str, key: str) -> dict[str, dict[str, Any]]:
+def _processes_by_identity(envelope: EvidenceEnvelope) -> dict[str, dict[str, Any]]:
     records: dict[str, dict[str, Any]] = {}
-    for record in _records(envelope, record_type):
-        value = record.get(key)
+    pids: set[int] = set()
+    for record in _records(envelope, "process"):
+        value = record.get("process_id")
         if not isinstance(value, str) or not value:
-            raise ValueError(f"{record_type} record missing {key}")
+            raise ValueError("process record missing process_id")
         if value in records:
-            raise ValueError(f"duplicate {record_type} {key}: {value}")
+            raise ValueError(f"duplicate process process_id: {value}")
+        validate_process_record_identity(record, envelope.host_id)
+        if record["pid"] in pids:
+            raise ValueError(f"duplicate process pid: {record['pid']}")
+        pids.add(record["pid"])
         records[value] = record
     return records
 
 
-def _link_identity(record: dict[str, Any]) -> str:
-    process = record.get("process_id") or f"pid={record.get('pid', 'unknown')}"
-    return (
-        f"{process}|{record.get('protocol')}|{record.get('state')}|"
-        f"{record.get('local_address')}:{record.get('local_port')}|"
-        f"{record.get('remote_address')}:{record.get('remote_port')}"
-    )
-
-
-def _links_by_identity(envelope: EvidenceEnvelope) -> dict[str, dict[str, Any]]:
+def _links_by_identity(
+    envelope: EvidenceEnvelope, processes: dict[str, dict[str, Any]]
+) -> dict[str, dict[str, Any]]:
     links: dict[str, dict[str, Any]] = {}
     for record in _records(envelope, "process_socket_link"):
-        identity = _link_identity(record)
+        _validate_link_context(record, processes)
+        identity = socket_link_identity(record)
         if identity in links:
             raise ValueError(f"duplicate process_socket_link identity: {identity}")
         links[identity] = record
     return links
 
 
+def _validate_link_context(record: dict[str, Any], processes: dict[str, dict[str, Any]]) -> None:
+    if record.get("linked") is False:
+        if any(record.get(field) is not None for field in ("process_id", "executable", "uid", "gid")):
+            raise ValueError("unlinked socket must not claim procfs identity or context")
+        return
+    if record.get("linked") is not True:
+        raise ValueError("socket linked must be boolean")
+    process_id = record.get("process_id")
+    process = processes.get(process_id) if isinstance(process_id, str) else None
+    if process is None:
+        raise ValueError("linked socket process_id must reference the paired process artifact")
+    for field in ("pid", "executable", "uid", "gid"):
+        if (
+            field not in process or field not in record
+            or type(record[field]) is not type(process[field]) or record[field] != process[field]
+        ):
+            raise ValueError(f"linked socket {field} must match procfs process")
+
+
 def _field_changes(before: dict[str, Any], after: dict[str, Any]) -> dict[str, dict[str, Any]]:
     immutable = {"record_type", "process_id", "pid", "start_time_ticks"}
     changes: dict[str, dict[str, Any]] = {}
diff --git a/projects/linux-process-observe/src/linux_process_observe/models.py b/projects/linux-process-observe/src/linux_process_observe/models.py
index 10ce5a4..e9ae1ee 100644
--- a/projects/linux-process-observe/src/linux_process_observe/models.py
+++ b/projects/linux-process-observe/src/linux_process_observe/models.py
@@ -11,6 +11,25 @@
 _FRACTIONAL_COMPONENT = re.compile(r"[.,](\d+)")
 
 
+def validate_process_record_identity(record: Mapping[str, Any], host_id: str) -> None:
+    """Check declared fields, without asserting independent instance verification."""
+    pid = record.get("pid")
+    ticks = record.get("start_time_ticks")
+    if type(pid) is not int or type(ticks) is not int:
+        raise ValueError("process identity requires integer pid and start_time_ticks")
+    if record.get("process_id") != f"{host_id}:{pid}:{ticks}":
+        raise ValueError("process_id must match host_id, pid and start_time_ticks")
+
+
+def socket_link_identity(record: Mapping[str, Any]) -> str:
+    process = record.get("process_id") or f"pid={record.get('pid', 'unknown')}"
+    return (
+        f"{process}|{record.get('protocol')}|{record.get('state')}|"
+        f"{record.get('local_address')}:{record.get('local_port')}|"
+        f"{record.get('remote_address')}:{record.get('remote_port')}"
+    )
+
+
 def parse_observed_at(value: str) -> datetime:
     if not isinstance(value, str) or not value.strip():
         raise ValueError("observed_at must be a non-empty string")
diff --git a/projects/linux-process-observe/src/linux_process_observe/parsers/procfs.py b/projects/linux-process-observe/src/linux_process_observe/parsers/procfs.py
index dbdd251..4d620ff 100644
--- a/projects/linux-process-observe/src/linux_process_observe/parsers/procfs.py
+++ b/projects/linux-process-observe/src/linux_process_observe/parsers/procfs.py
@@ -24,7 +24,11 @@ def parse_procfs_root(root: str | Path, *, host_id: str) -> list[ProcessRecord]:
             key=lambda item: int(item.name),
         )
     ]
-    processes_by_pid = {item.pid: item for item in records}
+    processes_by_pid: dict[int, ProcessRecord] = {}
+    for item in records:
+        if item.pid in processes_by_pid:
+            raise ValueError(f"duplicate procfs pid: {item.pid}")
+        processes_by_pid[item.pid] = item
     return [
         replace(
             item,
diff --git a/projects/linux-process-observe/src/linux_process_observe/snapshot.py b/projects/linux-process-observe/src/linux_process_observe/snapshot.py
index 2dbc63d..8284fc3 100644
--- a/projects/linux-process-observe/src/linux_process_observe/snapshot.py
+++ b/projects/linux-process-observe/src/linux_process_observe/snapshot.py
@@ -73,7 +73,11 @@ def build_snapshot_artifacts(
 def link_processes_to_sockets(
     processes: list[ProcessRecord], observations: list[SocketObservation]
 ) -> list[ProcessSocketLink]:
-    processes_by_pid = {item.pid: item for item in processes}
+    processes_by_pid: dict[int, ProcessRecord] = {}
+    for item in processes:
+        if item.pid in processes_by_pid:
+            raise ValueError(f"duplicate procfs pid: {item.pid}")
+        processes_by_pid[item.pid] = item
     links: list[ProcessSocketLink] = []
 
     for socket in observations:
@@ -86,7 +90,7 @@ def link_processes_to_sockets(
                     linked=process is not None,
                     process_id=process.process_id if process else None,
                     pid=process.pid if process else (ref.pid if ref else None),
-                    process_name=process.name if process else (ref.name if ref else None),
+                    process_name=ref.name if ref else None,
                     executable=process.executable if process else None,
                     uid=process.uid if process else None,
                     gid=process.gid if process else None,
diff --git a/projects/linux-process-observe/tests/test_attribution_integrity.py b/projects/linux-process-observe/tests/test_attribution_integrity.py
new file mode 100644
index 0000000..34fcfab
--- /dev/null
+++ b/projects/linux-process-observe/tests/test_attribution_integrity.py
@@ -0,0 +1,221 @@
+import json
+from pathlib import Path
+
+import pytest
+
+from linux_process_observe.adapters import build_telemetry_events
+from linux_process_observe.cli import main
+from linux_process_observe.diff import build_diff_envelope
+from linux_process_observe.models import EVIDENCE_SCHEMA, EvidenceEnvelope, ProcessRecord
+from linux_process_observe.parsers.procfs import parse_procfs_root
+from linux_process_observe.parsers.ss_text import parse_ss_text
+from linux_process_observe.snapshot import build_snapshot_artifacts, link_processes_to_sockets, write_envelope
+
+
+HOST = "example-host"
+T0 = "2026-07-05T00:00:00Z"
+T1 = "2026-07-05T00:05:00Z"
+
+
+def _process(pid=120, ticks=1200, name="worker-a"):
+    return ProcessRecord("process", f"{HOST}:{pid}:{ticks}", pid, 0, None, name,
+                         "S", 1001, 1001, 1001, 1001, f"/opt/example/bin/{name}",
+                         [f"/opt/example/bin/{name}"], ticks)
+
+
+def _ss(refs=(("worker-a", 120),)):
+    users = ",".join(f'("{name}",pid={pid},fd={3+i})' for i, (name, pid) in enumerate(refs))
+    return f"tcp LISTEN 0 64 127.0.0.1:9000 0.0.0.0:* users:({users})\n"
+
+
+def _pair(time, processes=None, observations=None):
+    processes = [_process()] if processes is None else processes
+    observations = parse_ss_text(_ss()) if observations is None else observations
+    return (EvidenceEnvelope(EVIDENCE_SCHEMA, "procfs", HOST, time, [p.to_dict() for p in processes]),
+            EvidenceEnvelope(EVIDENCE_SCHEMA, "procfs+ss", HOST, time,
+                             [link.to_dict() for link in link_processes_to_sockets(processes, observations)]))
+
+
+def _diff(before, after):
+    return build_diff_envelope(before_processes=before[0], before_links=before[1],
+                               after_processes=after[0], after_links=after[1])
+
+
+def _export(root: Path, process: ProcessRecord, directory=None):
+    target = root / (directory or str(process.pid))
+    target.mkdir(parents=True)
+    (target / "status").write_text(f"Name:\t{process.name}\nState:\tS\nPPid:\t0\nUid:\t1001 1001 1001 1001\nGid:\t1001 1001 1001 1001\n", encoding="utf-8")
+    fields = ["S", "0"] + ["0"] * 17 + [str(process.start_time_ticks)]
+    (target / "stat").write_text(f"{process.pid} ({process.name}) " + " ".join(fields), encoding="utf-8")
+    (target / "exe").write_text(process.executable, encoding="utf-8")
+    (target / "cmdline").write_bytes(process.executable.encode() + b"\0")
+
+
+@pytest.mark.parametrize("reported_name", ["worker-a", "worker-b"])
+def test_reported_ss_name_is_preserved_through_cli_diff_and_adapt(tmp_path, reported_name):
+    proc = tmp_path / "proc"
+    _export(proc, _process())
+    ss = tmp_path / "ss.txt"
+    ss.write_text(_ss(((reported_name, 120),)), encoding="utf-8")
+    after = build_snapshot_artifacts(proc_root=proc, ss_path=ss, host_id=HOST, observed_at=T1)
+    link = after[1].records[0]
+    assert link["process_name"] == reported_name
+    assert link["linked"] is True
+    assert link["process_id"] == _process().process_id
+    assert (link["executable"], link["uid"], link["gid"]) == (_process().executable, 1001, 1001)
+    before = _pair(T0, [], [])
+    paths = []
+    for label, pair in [("before", before), ("after", after)]:
+        for kind, envelope in zip(["processes", "links"], pair):
+            path = tmp_path / f"{label}-{kind}.json"
+            write_envelope(envelope, path)
+            paths.extend([f"--{label}-{kind}", str(path)])
+    output = tmp_path / "diff"
+    assert main(["diff", *paths, "--output-dir", str(output)]) == 0
+    telemetry = output / "events.jsonl"
+    assert main(["adapt", "--input", str(output / "process_diff.json"), "--output", str(telemetry)]) == 0
+    rows = [json.loads(line) for line in telemetry.read_text().splitlines()]
+    assert [row["event_type"] for row in rows] == ["process_added", "socket_link_added"]
+    assert {row["source"] for row in rows} == {_process().process_id}
+    assert {row["metadata"]["time_semantics"] for row in rows} == {"snapshot_diff_observed_at"}
+    assert "Process/socket links: +1 / -0" in (output / "report.md").read_text()
+
+
+def test_same_name_does_not_add_an_instance_verification_claim():
+    processes, links = _pair(T1)
+    assert links.records[0]["linked"] is True
+    assert "verified" not in links.records[0]
+    assert "confidence" not in links.records[0]
+    assert _diff(_pair(T0), (processes, links)).records == []
+
+
+def test_observed_start_change_and_shared_socket_references_remain_distinct():
+    diff = _diff(_pair(T0), _pair(T1, [_process(ticks=2200)]))
+    assert [(row["record_type"], row["change_type"]) for row in diff.records] == [
+        ("process_change", "added"), ("process_change", "removed"),
+        ("process_socket_link_change", "added"), ("process_socket_link_change", "removed")]
+    processes = [_process(), _process(pid=220, ticks=2200, name="worker-b")]
+    observations = parse_ss_text(_ss((("worker-b", 220), ("worker-a", 120))))
+    assert link_processes_to_sockets(processes, observations) == link_processes_to_sockets(list(reversed(processes)), observations)
+    assert [row.pid for row in link_processes_to_sockets(processes, observations)] == [120, 220]
+
+
+@pytest.mark.parametrize("reverse", [False, True])
+def test_duplicate_pid_candidates_are_rejected_without_selecting_by_order(reverse):
+    processes = [_process(), _process(ticks=2200, name="worker-b")]
+    if reverse:
+        processes.reverse()
+    with pytest.raises(ValueError, match="duplicate procfs pid"):
+        link_processes_to_sockets(processes, parse_ss_text(_ss()))
+
+
+def test_numeric_directory_aliases_do_not_create_two_instances_for_one_pid(tmp_path):
+    _export(tmp_path, _process(), "120")
+    _export(tmp_path, _process(ticks=2200, name="worker-b"), "0120")
+    with pytest.raises(ValueError, match="duplicate procfs pid"):
+        parse_procfs_root(tmp_path, host_id=HOST)
+
+
+@pytest.mark.parametrize("phase", ["before", "after"])
+def test_diff_rejects_process_id_that_disagrees_with_observed_fields(phase):
+    before, after = _pair(T0), _pair(T1)
+    selected = before if phase == "before" else after
+    selected[0].records[0]["process_id"] = "example-host:999:9999"
+    with pytest.raises(ValueError, match="process_id must match host_id, pid and start_time_ticks"):
+        _diff(before, after)
+
+
+def test_diff_rejects_multiple_valid_identities_for_the_same_pid():
+    before, after = _pair(T0), _pair(T1)
+    after[0].records.append(_process(ticks=2200).to_dict())
+    with pytest.raises(ValueError, match="duplicate process pid"):
+        _diff(before, after)
+
+
+@pytest.mark.parametrize("phase", ["before", "after"])
+@pytest.mark.parametrize(("field", "value"), [("pid", 999), ("executable", "/opt/example/bin/other"), ("uid", 0), ("gid", 0)])
+def test_matching_times_do_not_admit_contradictory_procfs_context(phase, field, value):
+    before, after = _pair(T0), _pair(T1)
+    selected = before if phase == "before" else after
+    selected[1].records[0][field] = value
+    with pytest.raises(ValueError, match=f"linked socket {field} must match procfs process"):
+        _diff(before, after)
+
+
+@pytest.mark.parametrize("phase", ["before", "after"])
+def test_matching_times_do_not_admit_a_dangling_linked_identity(phase):
+    before, after = _pair(T0), _pair(T1)
+    selected = before if phase == "before" else after
+    selected[1].records[0]["process_id"] = "example-host:120:9999"
+    with pytest.raises(ValueError, match="linked socket process_id must reference the paired process artifact"):
+        _diff(before, after)
+
+
+@pytest.mark.parametrize("field", ["process_id", "executable", "uid", "gid"])
+def test_unlinked_socket_cannot_claim_procfs_identity_or_context(field):
+    before, after = _pair(T0), _pair(T1, observations=parse_ss_text(_ss((("unknown", 999),))))
+    after[1].records[0][field] = _process().to_dict().get(field, 1001)
+    with pytest.raises(ValueError, match="unlinked socket must not claim procfs identity or context"):
+        _diff(before, after)
+
+
+def test_adapter_rejects_a_self_consistent_label_with_contradictory_process_fields():
+    diff = _diff(_pair(T0, [], []), _pair(T1))
+    change = diff.records[0]
+    change["identity"] = change["after"]["process_id"] = "example-host:999:9999"
+    with pytest.raises(ValueError, match="process_id must match host_id, pid and start_time_ticks"):
+        build_telemetry_events(diff)
+
+
+def test_adapter_rejects_socket_change_identity_that_disagrees_with_observation():
+    diff = _diff(_pair(T0, [], []), _pair(T1))
+    diff.records[-1]["identity"] = "different-observation"
+    with pytest.raises(ValueError, match="socket identity must match observation"):
+        build_telemetry_events(diff)
+
+
+def test_ss_name_only_change_preserves_context_without_inventing_a_link_event():
+    before = _pair(T0)
+    after = _pair(T1, observations=parse_ss_text(_ss((("worker-b", 120),))))
+    assert after[1].records[0]["process_name"] == "worker-b"
+    assert after[1].records[0]["linked"] is True
+    assert _diff(before, after).records == []
+
+
+def test_absent_pid_retains_an_unlinked_observation_and_pid_fallback():
+    after = _pair(T1, observations=parse_ss_text(_ss((("worker-b", 999),))))
+    link = after[1].records[0]
+    assert link["linked"] is False and link["pid"] == 999
+    assert all(link[field] is None for field in ("process_id", "executable", "uid", "gid"))
+    events = build_telemetry_events(_diff(_pair(T0, [], []), after))
+    assert events[-1]["source"] == f"{HOST}:pid:999"
+
+
+@pytest.mark.parametrize(("field", "value"), [("pid", True), ("start_time_ticks", "1200")])
+def test_non_integer_identity_components_are_rejected(field, value):
+    before, after = _pair(T0), _pair(T1)
+    after[0].records[0][field] = value
+    with pytest.raises(ValueError, match="integer pid and start_time_ticks"):
+        _diff(before, after)
+
+
+@pytest.mark.parametrize(("field", "value"), [("uid", 0), ("process_id", "example-host:120:9999")])
+def test_cli_rejects_contradictory_artifacts_without_creating_outputs(tmp_path, capsys, field, value):
+    before, after = _pair(T0), _pair(T1)
+    after[1].records[0][field] = value
+    args = []
+    inputs = {}
+    for phase, pair in [("before", before), ("after", after)]:
+        for kind, envelope in zip(["processes", "links"], pair):
+            path = tmp_path / f"{phase}-{kind}.json"
+            write_envelope(envelope, path)
+            inputs[path] = path.read_bytes()
+            args.extend([f"--{phase}-{kind}", str(path)])
+    output = tmp_path / "not-created"
+    assert main(["diff", *args, "--output-dir", str(output)]) == 1
+    captured = capsys.readouterr()
+    assert captured.out == ""
+    assert "error command=diff input=artifacts" in captured.err
+    assert "Traceback" not in captured.err
+    assert not output.exists()
+    assert all(path.read_bytes() == data for path, data in inputs.items())