diff --git a/src/contextsafe/evidence.py b/src/contextsafe/evidence.py
index f2dd341..d7ffe75 100644
--- a/src/contextsafe/evidence.py
+++ b/src/contextsafe/evidence.py
@@ -205,6 +205,11 @@ class EvidenceMetadata:
 
     def __post_init__(self) -> None:
         _require_canonical_utc(self.captured_at, "$.captured_at")
+        from contextsafe.preflight import _reject_unsafe_string
+
+        _reject_unsafe_string(self.collector_id, "$.collector_id")
+        _reject_unsafe_string(self.system_id, "$.system_id")
+        _reject_unsafe_string(self.system_version, "$.system_version")
 
     def to_dict(self) -> dict[str, JsonValue]:
         return {
@@ -549,23 +554,31 @@ def parse_evidence_source(
 def parse_evidence_metadata(value: object) -> EvidenceMetadata:
     """Parse deterministic provenance supplied outside the raw source."""
 
+    from contextsafe.preflight import _reject_unsafe_string
+
     data = object_value(value, "$")
     exact_keys(
         data,
         frozenset({"captured_at", "collector_id", "system_id", "system_version"}),
         "$",
     )
+    collector_id = bounded_string(
+        data["collector_id"], "$.collector_id", pattern=SAFE_TOKEN_PATTERN
+    )
+    _reject_unsafe_string(collector_id, "$.collector_id")
+    system_id = bounded_string(data["system_id"], "$.system_id", pattern=ID_PATTERN)
+    _reject_unsafe_string(system_id, "$.system_id")
+    system_version = bounded_string(
+        data["system_version"],
+        "$.system_version",
+        pattern=SAFE_TOKEN_PATTERN,
+    )
+    _reject_unsafe_string(system_version, "$.system_version")
     return EvidenceMetadata(
         captured_at=timestamp_value(data["captured_at"], "$.captured_at"),
-        collector_id=bounded_string(
-            data["collector_id"], "$.collector_id", pattern=SAFE_TOKEN_PATTERN
-        ),
-        system_id=bounded_string(data["system_id"], "$.system_id", pattern=ID_PATTERN),
-        system_version=bounded_string(
-            data["system_version"],
-            "$.system_version",
-            pattern=SAFE_TOKEN_PATTERN,
-        ),
+        collector_id=collector_id,
+        system_id=system_id,
+        system_version=system_version,
     )
 
 
@@ -681,6 +694,20 @@ def parse_evidence_record(value: object) -> EvidenceRecord:
         raise contract_error(
             "unsupported_media_type", "$.media_type", "media type is unsupported"
         )
+    collector_id = bounded_string(
+        data["collector_id"], "$.collector_id", pattern=SAFE_TOKEN_PATTERN
+    )
+    from contextsafe.preflight import _reject_unsafe_string
+
+    _reject_unsafe_string(collector_id, "$.collector_id")
+    system_id = bounded_string(data["system_id"], "$.system_id", pattern=ID_PATTERN)
+    _reject_unsafe_string(system_id, "$.system_id")
+    system_version = bounded_string(
+        data["system_version"],
+        "$.system_version",
+        pattern=SAFE_TOKEN_PATTERN,
+    )
+    _reject_unsafe_string(system_version, "$.system_version")
     record = EvidenceRecord(
         schema_version=schema_version,
         evidence_id=evidence_id,
@@ -695,15 +722,9 @@ def parse_evidence_record(value: object) -> EvidenceRecord:
         ),
         raw_byte_count=_positive_integer(data["raw_byte_count"], "$.raw_byte_count"),
         captured_at=timestamp_value(data["captured_at"], "$.captured_at"),
-        collector_id=bounded_string(
-            data["collector_id"], "$.collector_id", pattern=SAFE_TOKEN_PATTERN
-        ),
-        system_id=bounded_string(data["system_id"], "$.system_id", pattern=ID_PATTERN),
-        system_version=bounded_string(
-            data["system_version"],
-            "$.system_version",
-            pattern=SAFE_TOKEN_PATTERN,
-        ),
+        collector_id=collector_id,
+        system_id=system_id,
+        system_version=system_version,
         boundary_profile_version=enum_string(
             data["boundary_profile_version"],
             "$.boundary_profile_version",
diff --git a/tests/test_evidence_models.py b/tests/test_evidence_models.py
index f904019..2d8de23 100644
--- a/tests/test_evidence_models.py
+++ b/tests/test_evidence_models.py
@@ -19,6 +19,7 @@
     build_evidence_record,
     build_evidence_scope,
     parse_canonical_observation,
+    parse_evidence_metadata,
     parse_evidence_record,
     parse_evidence_source,
 )
@@ -269,6 +270,74 @@ def test_evidence_timestamp_schema_and_models_require_canonical_utc(
         assert raised.value.code == "invalid_timestamp"
 
 
+@pytest.mark.parametrize(
+    ("field", "unsafe_value", "expected_code"),
+    [
+        ("collector_id", "realpatientcanary", "phi_canary_detected"),
+        ("collector_id", "123-45-6789", "direct_identifier_detected"),
+        ("collector_id", "http://patient.org", "direct_identifier_detected"),
+        ("collector_id", "MRN:1234567", "direct_identifier_detected"),
+        ("system_id", "SYS-CANARY-REALPATIENTCANARY", "phi_canary_detected"),
+        ("system_version", "realpatientcanary", "phi_canary_detected"),
+        ("system_version", "2026-08-23", "direct_identifier_detected"),
+    ],
+)
+def test_evidence_metadata_and_parser_reject_phi_and_direct_identifiers(
+    field: str, unsafe_value: str, expected_code: str
+) -> None:
+    valid_payload = {
+        "captured_at": "2026-07-13T12:00:00Z",
+        "collector_id": "TEST-COLLECTOR",
+        "system_id": "SYS-STAGING-EHR",
+        "system_version": "fixture-1.0",
+    }
+    payload = dict(valid_payload)
+    payload[field] = unsafe_value
+
+    with pytest.raises(ContextSafeError) as raised:
+        parse_evidence_metadata(payload)
+    assert raised.value.code == expected_code
+    assert raised.value.path == f"$.{field}"
+
+    collector_id = unsafe_value if field == "collector_id" else "TEST-COLLECTOR"
+    system_id = unsafe_value if field == "system_id" else "SYS-STAGING-EHR"
+    system_version = unsafe_value if field == "system_version" else "fixture-1.0"
+    with pytest.raises(ContextSafeError) as raised:
+        EvidenceMetadata(
+            captured_at=datetime(2026, 7, 13, 12, tzinfo=UTC),
+            collector_id=collector_id,
+            system_id=system_id,
+            system_version=system_version,
+        )
+    assert raised.value.code == expected_code
+    assert raised.value.path == f"$.{field}"
+
+
+def test_parse_evidence_record_rejects_unsafe_metadata_fields(
+    tmp_path: Path,
+    evidence_source_json: dict[str, Any],
+    evidence_scope: Any,
+    evidence_metadata: Any,
+) -> None:
+    source = tmp_path / "source.json"
+    source.write_text(json.dumps(evidence_source_json), encoding="utf-8")
+    record = build_evidence_record(
+        preflight_source(source, evidence_scope), evidence_metadata
+    ).to_dict()
+
+    for field, unsafe_value, expected_code in [
+        ("collector_id", "realpatientcanary", "phi_canary_detected"),
+        ("system_id", "SYS-CANARY-REALPATIENTCANARY", "phi_canary_detected"),
+        ("system_version", "realpatientcanary", "phi_canary_detected"),
+    ]:
+        bad_record = dict(record)
+        bad_record[field] = unsafe_value
+        with pytest.raises(ContextSafeError) as raised:
+            parse_evidence_record(bad_record)
+        assert raised.value.code == expected_code
+        assert raised.value.path == f"$.{field}"
+
+
 @pytest.mark.parametrize(
     ("mutation", "expected_code"),
     [