fix(process): reject invalid temporal cohorts

CHANGELOG.md

@@ -8,8 +8,12 @@ All notable changes to this project will be documented in this file.
 
 ### Changed
 
+- Preserve up to microsecond `observed_at` precision when normalizing process evidence snapshots to UTC, and reject finer input precision instead of silently truncating it.
+
 ### Fixed
 
+- Reject process diffs when process/socket-link artifacts do not share a snapshot instant or when the before instant is not strictly earlier than the after instant.
+
 ## [v0.3.0] - 2026-08-12
 
 408-to-Security Bridge

notes/process-evidence-schema.md

@@ -10,7 +10,7 @@ This lab combines:
 - PID
 - `/proc/<pid>/stat` start time in clock ticks
 
-The resulting `process_id` is `host_id:pid:start_time_ticks`. It is a snapshot identity, not a cryptographic identity and not a guarantee that two independently collected records observed exactly the same execution context.
+The resulting `process_id` is `host_id:pid:start_time_ticks`. Because start time is measured from system boot and the envelope has no boot identifier, it is a same-boot snapshot identity rather than a cross-reboot durable identity. It is not a cryptographic identity and not a guarantee that two independently collected records observed exactly the same execution context.
 
 ## Evidence envelope
 
@@ -29,11 +29,27 @@ Every JSON artifact uses:
 - `schema` identifies the shared outer contract.
 - `source` names the saved evidence family, such as `procfs` or `procfs+ss`.
 - `host_id` is a stable, sanitized host identifier supplied by the operator.
-- `observed_at` is an explicit timezone-aware collection time normalized to UTC.
+- `observed_at` is an explicit timezone-aware collection time normalized to UTC. Fractional inputs retain up to microsecond precision; more than six fractional-second digits are rejected rather than truncated.
 - `records` contains artifact-specific normalized records.
 
 The envelope is deliberately small so LogLens, telemetry-lab, or later evidence experiments can consume the same outer shape without pretending that every record family has the same inner fields.
 
+## Temporal cohort contract
+
+`process_diff.json` is directional evidence, so its four inputs must form one comparable temporal cohort:
+
+```text
+before process instant == before socket-link instant
+after process instant  == after socket-link instant
+before instant < after instant
+```
+
+Observation times are compared as timezone-aware instants. For example, `2026-07-05T00:00:00Z` and `2026-07-05T08:00:00+08:00` are equivalent. Equal or reversed before/after instants and mismatched process/socket-link pairs fail closed before any records are compared.
+
+Subsecond precision is retained so two captures within one second can still satisfy strict ordering. If two captures have the same normalized instant, the lab cannot infer their order and refuses to produce a directional diff.
+
+Paired timestamps are cohort provenance, not atomicity proof. Procfs and saved `ss` inputs may still have been collected sequentially, and socket evidence provides PID context rather than an execution-instance identity independent of procfs.
+
 ## Process record
 
 Process records preserve:

projects/linux-process-observe/README.md

@@ -45,7 +45,9 @@ The three system-evidence `.json` artifacts use the same envelope:
 }
 ```
 
-`observed_at` is required, must include a timezone, and is normalized to UTC. Record ordering is deterministic.
+`observed_at` is required, must include a timezone, and is normalized to UTC. Whole-second inputs keep the existing `...00Z` form; fractional inputs retain up to microsecond precision so captures within the same second remain orderable. Inputs with more than six fractional-second digits fail closed instead of being silently truncated. Record ordering is deterministic.
+
+A diff is produced only for a valid temporal cohort: the process and socket-link artifacts within each snapshot must represent the same observation instant, and the before instant must be strictly earlier than the after instant. Equivalent timezone representations compare as instants, not raw strings. Matching timestamps describe the intended snapshot cohort; they do not prove procfs and `ss` were collected atomically.
 
 ## Workflow
 
@@ -89,6 +91,7 @@ The JSONL can be supplied as `input_path` to telemetry-lab's existing `run windo
 ## Identity And Link Semantics
 
 - `process_id` is `host_id:pid:start_time_ticks`; PID alone is not treated as durable identity because Linux can reuse it.
+- `start_time_ticks` is measured from system boot. Because the envelope has no boot identifier, `process_id` is a same-boot snapshot identity, not a cross-reboot durable identity.
 - `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.
@@ -97,7 +100,7 @@ The JSONL can be supplied as `input_path` to telemetry-lab's existing `run windo
 
 ## Validation Status
 
-Pytest covers procfs parsing, `ss` parsing, process identity, socket linking, malformed inputs, timezone normalization, golden artifacts, diffs, reports, and the CLI workflow.
+Pytest covers procfs parsing, `ss` parsing, process identity, socket linking, malformed inputs, timezone and subsecond normalization, fail-closed temporal cohort validation, golden artifacts, diffs, reports, and the CLI workflow.
 The adapter adds golden JSONL coverage, an unlinked-socket source fallback test, malformed diff coverage, and CLI error reporting coverage.
 
 ## Non-Goals

projects/linux-process-observe/src/linux_process_observe/diff.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any
 
-from .models import EVIDENCE_SCHEMA, EvidenceEnvelope
+from .models import EVIDENCE_SCHEMA, EvidenceEnvelope, parse_observed_at
 
 
 def build_diff_envelope(
@@ -45,15 +46,38 @@ def build_diff_envelope(
     )
 
 
-def _validate_artifacts(*envelopes: EvidenceEnvelope) -> None:
+def _validate_artifacts(
+    before_processes: EvidenceEnvelope,
+    after_processes: EvidenceEnvelope,
+    before_links: EvidenceEnvelope,
+    after_links: EvidenceEnvelope,
+) -> None:
+    envelopes = (before_processes, after_processes, before_links, after_links)
     host_ids = {item.host_id for item in envelopes}
     if len(host_ids) != 1:
         raise ValueError("all artifacts must use the same host_id")
-    if envelopes[0].source != "procfs" or envelopes[1].source != "procfs":
+    if before_processes.source != "procfs" or after_processes.source != "procfs":
         raise ValueError("process artifacts must use source=procfs")
-    if envelopes[2].source != "procfs+ss" or envelopes[3].source != "procfs+ss":
+    if before_links.source != "procfs+ss" or after_links.source != "procfs+ss":
         raise ValueError("link artifacts must use source=procfs+ss")
 
+    before_time = _snapshot_observed_at("before", before_processes, before_links)
+    after_time = _snapshot_observed_at("after", after_processes, after_links)
+    if before_time >= after_time:
+        raise ValueError("before observed_at must be earlier than after observed_at")
+
+
+def _snapshot_observed_at(
+    phase: str,
+    processes: EvidenceEnvelope,
+    links: EvidenceEnvelope,
+) -> datetime:
+    process_time = parse_observed_at(processes.observed_at)
+    link_time = parse_observed_at(links.observed_at)
+    if process_time != link_time:
+        raise ValueError(f"{phase} process and link artifacts must represent the same observation time")
+    return process_time
+
 
 def _records(envelope: EvidenceEnvelope, record_type: str) -> list[dict[str, Any]]:
     records: list[dict[str, Any]] = []

projects/linux-process-observe/src/linux_process_observe/models.py

@@ -4,9 +4,25 @@
 from datetime import datetime
 from typing import Any, Mapping
 import json
+import re
 
 
 EVIDENCE_SCHEMA = "stacknil.system-evidence.v1"
+_FRACTIONAL_COMPONENT = re.compile(r"[.,](\d+)")
+
+
+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")
+    if any(len(digits) > 6 for digits in _FRACTIONAL_COMPONENT.findall(value)):
+        raise ValueError("observed_at supports at most 6 fractional second digits")
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ValueError("observed_at must be ISO 8601") from exc
+    if parsed.tzinfo is None or parsed.utcoffset() is None:
+        raise ValueError("observed_at must include a timezone")
+    return parsed
 
 
 @dataclass(frozen=True, slots=True)
@@ -108,14 +124,7 @@ def from_mapping(cls, payload: Mapping[str, Any]) -> "EvidenceEnvelope":
             raise ValueError("source must be a non-empty string")
         if not isinstance(host_id, str) or not host_id.strip():
             raise ValueError("host_id must be a non-empty string")
-        if not isinstance(observed_at, str) or not observed_at.strip():
-            raise ValueError("observed_at must be a non-empty string")
-        try:
-            parsed_time = datetime.fromisoformat(observed_at.replace("Z", "+00:00"))
-        except ValueError as exc:
-            raise ValueError("observed_at must be ISO 8601") from exc
-        if parsed_time.tzinfo is None or parsed_time.utcoffset() is None:
-            raise ValueError("observed_at must include a timezone")
+        parse_observed_at(observed_at)
         if not isinstance(records, list) or not all(isinstance(item, dict) for item in records):
             raise ValueError("records must be a list of objects")
 

projects/linux-process-observe/src/linux_process_observe/snapshot.py

@@ -1,13 +1,20 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from datetime import datetime, timezone
+from datetime import timezone
 from ipaddress import ip_address
 from pathlib import Path
 from typing import Any
 import json
 
-from .models import EVIDENCE_SCHEMA, EvidenceEnvelope, ProcessRecord, ProcessSocketLink, SocketObservation
+from .models import (
+    EVIDENCE_SCHEMA,
+    EvidenceEnvelope,
+    ProcessRecord,
+    ProcessSocketLink,
+    SocketObservation,
+    parse_observed_at,
+)
 from .parsers.procfs import parse_procfs_root
 from .parsers.ss_text import parse_ss_text
 
@@ -131,12 +138,12 @@ def write_envelope(envelope: EvidenceEnvelope, path: str | Path) -> None:
 
 def normalize_observed_at(value: str) -> str:
     try:
-        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+        parsed = parse_observed_at(value)
     except ValueError as exc:
-        raise EvidenceInputError("observed-at", "-", exc.__class__.__name__, "observed_at must be ISO 8601") from exc
-    if parsed.tzinfo is None or parsed.utcoffset() is None:
-        raise EvidenceInputError("observed-at", "-", "ValueError", "observed_at must include a timezone")
-    return parsed.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
+        raise EvidenceInputError("observed-at", "-", exc.__class__.__name__, str(exc)) from exc
+    normalized = parsed.astimezone(timezone.utc)
+    timespec = "microseconds" if normalized.microsecond else "seconds"
+    return normalized.isoformat(timespec=timespec).replace("+00:00", "Z")
 
 
 def _socket_role(state: str) -> str:

projects/linux-process-observe/tests/test_cli.py

@@ -66,6 +66,36 @@ def test_cli_reports_malformed_procfs_with_input_context(tmp_path: Path, capsys)
     assert not (tmp_path / "process_snapshot.json").exists()
 
 
+def test_cli_diff_rejects_reversed_snapshot_order(tmp_path: Path, capsys) -> None:
+    baseline_dir = tmp_path / "baseline"
+    changed_dir = tmp_path / "changed"
+    diff_dir = tmp_path / "diff"
+    assert _snapshot("baseline", "2026-07-05T00:10:00Z", baseline_dir) == 0
+    assert _snapshot("changed", "2026-07-05T00:05:00Z", changed_dir) == 0
+
+    exit_code = main(
+        [
+            "diff",
+            "--before-processes",
+            str(baseline_dir / "process_snapshot.json"),
+            "--after-processes",
+            str(changed_dir / "process_snapshot.json"),
+            "--before-links",
+            str(baseline_dir / "process_socket_links.json"),
+            "--after-links",
+            str(changed_dir / "process_socket_links.json"),
+            "--output-dir",
+            str(diff_dir),
+        ]
+    )
+    captured = capsys.readouterr()
+
+    assert exit_code == 1
+    assert "error command=diff input=artifacts" in captured.err
+    assert "before observed_at must be earlier than after observed_at" in captured.err
+    assert not diff_dir.exists()
+
+
 def _snapshot(name: str, observed_at: str, output_dir: Path) -> int:
     return main(
         [

projects/linux-process-observe/tests/test_diff.py

@@ -63,6 +63,101 @@ def test_diff_rejects_duplicate_socket_link_identity() -> None:
         )
 
 
+@pytest.mark.parametrize(
+    ("phase", "mismatched_time"),
+    [
+        ("before", "2026-07-05T00:01:00Z"),
+        ("after", "2026-07-05T00:06:00Z"),
+    ],
+)
+def test_diff_rejects_mismatched_process_and_link_observation_times(
+    phase: str,
+    mismatched_time: str,
+) -> None:
+    before_processes, before_links = _build("baseline", "2026-07-05T00:00:00Z")
+    after_processes, after_links = _build("changed", "2026-07-05T00:05:00Z")
+    if phase == "before":
+        before_links = _with_observed_at(before_links, mismatched_time)
+    else:
+        after_links = _with_observed_at(after_links, mismatched_time)
+
+    with pytest.raises(ValueError, match=f"{phase} process and link artifacts must represent the same observation time"):
+        build_diff_envelope(
+            before_processes=before_processes,
+            after_processes=after_processes,
+            before_links=before_links,
+            after_links=after_links,
+        )
+
+
+@pytest.mark.parametrize(
+    ("before_time", "after_time"),
+    [
+        ("2026-07-05T00:05:00Z", "2026-07-05T00:05:00Z"),
+        ("2026-07-05T00:10:00Z", "2026-07-05T00:05:00Z"),
+    ],
+    ids=["equal", "reversed"],
+)
+def test_diff_rejects_non_increasing_observation_times(before_time: str, after_time: str) -> None:
+    before_processes, before_links = _build("baseline", before_time)
+    after_processes, after_links = _build("changed", after_time)
+
+    with pytest.raises(ValueError, match="before observed_at must be earlier than after observed_at"):
+        build_diff_envelope(
+            before_processes=before_processes,
+            after_processes=after_processes,
+            before_links=before_links,
+            after_links=after_links,
+        )
+
+
+def test_diff_compares_observation_times_as_instants() -> None:
+    before_processes, before_links = _build("baseline", "2026-07-05T00:00:00Z")
+    after_processes, after_links = _build("changed", "2026-07-05T00:05:00Z")
+    before_links = _with_observed_at(before_links, "2026-07-05T08:00:00+08:00")
+    after_links = _with_observed_at(after_links, "2026-07-05T08:05:00+08:00")
+
+    diff = build_diff_envelope(
+        before_processes=before_processes,
+        after_processes=after_processes,
+        before_links=before_links,
+        after_links=after_links,
+    )
+
+    assert diff.to_dict() == _load_json(GOLDEN / "diff" / "process_diff.json")
+
+
+def test_diff_accepts_strictly_ordered_subsecond_snapshots() -> None:
+    before_processes, before_links = _build("baseline", "2026-07-05T00:00:00.100000Z")
+    after_processes, after_links = _build("changed", "2026-07-05T00:00:00.200000Z")
+
+    diff = build_diff_envelope(
+        before_processes=before_processes,
+        after_processes=after_processes,
+        before_links=before_links,
+        after_links=after_links,
+    )
+
+    assert diff.observed_at == "2026-07-05T00:00:00.200000Z"
+
+
+def test_diff_rejects_observation_times_beyond_microsecond_precision() -> None:
+    before_processes, before_links = _build("baseline", "2026-07-05T00:00:00Z")
+    after_processes, after_links = _build("changed", "2026-07-05T00:05:00Z")
+    before_processes = _with_observed_at(before_processes, "2026-07-05T00:00:00.1234561Z")
+    before_links = _with_observed_at(before_links, "2026-07-05T00:00:00.1234561Z")
+    after_processes = _with_observed_at(after_processes, "2026-07-05T00:00:00.1234569Z")
+    after_links = _with_observed_at(after_links, "2026-07-05T00:00:00.1234569Z")
+
+    with pytest.raises(ValueError, match="observed_at supports at most 6 fractional second digits"):
+        build_diff_envelope(
+            before_processes=before_processes,
+            after_processes=after_processes,
+            before_links=before_links,
+            after_links=after_links,
+        )
+
+
 def _build(name: str, observed_at: str) -> tuple[EvidenceEnvelope, EvidenceEnvelope]:
     return build_snapshot_artifacts(
         proc_root=FIXTURES / name / "proc",
@@ -72,5 +167,15 @@ def _build(name: str, observed_at: str) -> tuple[EvidenceEnvelope, EvidenceEnvel
     )
 
 
+def _with_observed_at(envelope: EvidenceEnvelope, observed_at: str) -> EvidenceEnvelope:
+    return EvidenceEnvelope(
+        schema=envelope.schema,
+        source=envelope.source,
+        host_id=envelope.host_id,
+        observed_at=observed_at,
+        records=envelope.records,
+    )
+
+
 def _load_json(path: Path) -> dict[str, object]:
     return json.loads(path.read_text(encoding="utf-8"))

projects/linux-process-observe/tests/test_snapshot.py

@@ -47,6 +47,28 @@ def test_snapshot_requires_timezone_aware_observation_time() -> None:
         )
 
 
+def test_snapshot_preserves_subsecond_observation_time() -> None:
+    processes, links = build_snapshot_artifacts(
+        proc_root=FIXTURES / "baseline" / "proc",
+        ss_path=FIXTURES / "baseline" / "ss.txt",
+        host_id="lab-host",
+        observed_at="2026-07-05T08:00:00.123456+08:00",
+    )
+
+    assert processes.observed_at == "2026-07-05T00:00:00.123456Z"
+    assert links.observed_at == "2026-07-05T00:00:00.123456Z"
+
+
+def test_snapshot_rejects_observation_time_beyond_microsecond_precision() -> None:
+    with pytest.raises(EvidenceInputError, match="observed_at supports at most 6 fractional second digits"):
+        build_snapshot_artifacts(
+            proc_root=FIXTURES / "baseline" / "proc",
+            ss_path=FIXTURES / "baseline" / "ss.txt",
+            host_id="lab-host",
+            observed_at="2026-07-05T00:00:00.1234567Z",
+        )
+
+
 def test_loaded_envelope_requires_timezone_aware_observation_time() -> None:
     with pytest.raises(ValueError, match="observed_at must include a timezone"):
         EvidenceEnvelope.from_mapping(