test(integration): make the pinned process-evidence interface executable

.gitattributes

@@ -6,3 +6,4 @@
 /demos/**/data/raw/** text eol=lf
 /demos/**/artifacts/run_manifest.json text eol=lf
 /demos/**/artifacts/*.md text eol=lf
+/tests/fixtures/process_evidence/telemetry_events.jsonl text eol=lf

docs/case-study-process-evidence-to-telemetry.md

@@ -102,3 +102,7 @@ The pinned implementations, synthetic fixtures and producer goldens are public.
 - [Consumer event-time model](https://github.com/stacknil/telemetry-lab/blob/98ae3ffd1edb47d90eb42b3a8625be830582997e/docs/event-time-model.md)
 
 *Prepared from the retained 19 October 2026 integration evidence. No engineering tests or demos were rerun for the documentation review.*
+
+Later work: the [20 October executable integration suite](process-evidence-integration.md)
+adds offline regression tests and a separate producer-to-consumer reproduction runner.
+Its tests and fresh-environment results are separate from the historical observations above.

docs/process-evidence-integration.md

@@ -0,0 +1,119 @@
+# Executable process-evidence integration
+
+This suite checks a saved-evidence producer against the real telemetry consumer.
+It adds executable regression evidence after the [19 October case study](case-study-process-evidence-to-telemetry.md);
+it does not rerun or replace that historical experiment's private receipts.
+
+## Versions and fixture
+
+- Producer: [systems-foundations](https://github.com/stacknil/systems-foundations),
+  pinned at `1d31d4287036cf2bbd3aeb8059d7d332bedef598`,
+  `projects/linux-process-observe`.
+- Consumer: this suite is based on `de542cb3cadba776c8e759f786152f9222fdfdb7`.
+  Reproduce from the reviewed integration revision containing this guide.
+  The historical case used `98ae3ffd1edb47d90eb42b3a8625be830582997e`.
+- Offline input: [`tests/fixtures/process_evidence/telemetry_events.jsonl`](../tests/fixtures/process_evidence/telemetry_events.jsonl),
+  an unmodified, public synthetic producer golden, with [provenance](../tests/fixtures/process_evidence/provenance.json).
+  It is 4,108 LF bytes, SHA-256
+  `1977b98518497c9a07da6393499cb49e6d0be9140ac5bcd708c2a1cd31a3ad51`.
+  Its scoped `.gitattributes` entry prevents checkout newline conversion.
+
+## Layer A: offline regression gate
+
+Install this checkout's development dependencies in a separate Python 3.11+
+virtual environment, then run:
+
+```sh
+python -m pip install -e ".[dev]"
+python -m pip check
+python -m pytest tests/test_process_evidence_integration.py -v
+```
+
+The tests execute the actual loader, normalization, windows, features, rules and
+CLI. They never fetch the producer, access private receipts or mock the consumer.
+The existing ordinary `pytest` CI job includes this layer without extra network
+access during test execution.
+
+## Layer B: actual producer commands and two consumer runs
+
+Git and the same environment's development dependencies are required. Obtain
+the pinned public producer before running the workflow. From the consumer root:
+
+```sh
+git clone https://github.com/stacknil/systems-foundations.git ../process-evidence-producer
+git -C ../process-evidence-producer checkout --detach 1d31d4287036cf2bbd3aeb8059d7d332bedef598
+python scripts/reproduce_process_evidence.py --producer ../process-evidence-producer --output-root ../process-evidence-run
+```
+
+Use a **new** output directory outside both repositories on every invocation.
+The runner refuses collisions and performs no cleanup. It reads immutable Git
+objects, exports only the pinned project with Git newline conversion disabled,
+and executes the exported source. Dirty files in the supplied producer checkout
+cannot select the executed source. The producer uses only the standard library;
+its exported `src` is selected explicitly for each subprocess. No producer fetch,
+package download, live procfs read or network operation occurs inside the runner.
+
+The executable sequence is the following, with `PROJECT` referring to the
+exported pinned project and `OUT` to the fresh output directory:
+
+```sh
+python -m linux_process_observe snapshot --proc-root "$PROJECT/tests/fixtures/baseline/proc" --ss "$PROJECT/tests/fixtures/baseline/ss.txt" --host-id lab-host --observed-at 2026-07-05T00:00:00Z --output-dir "$OUT/baseline"
+python -m linux_process_observe snapshot --proc-root "$PROJECT/tests/fixtures/changed/proc" --ss "$PROJECT/tests/fixtures/changed/ss.txt" --host-id lab-host --observed-at 2026-07-05T00:05:00Z --output-dir "$OUT/changed"
+python -m linux_process_observe diff --before-processes "$OUT/baseline/process_snapshot.json" --after-processes "$OUT/changed/process_snapshot.json" --before-links "$OUT/baseline/process_socket_links.json" --after-links "$OUT/changed/process_socket_links.json" --output-dir "$OUT/diff"
+python -m linux_process_observe adapt --input "$OUT/diff/process_diff.json" --output "$OUT/diff/telemetry_events.jsonl"
+python -m telemetry_lab.cli run window --config "$OUT/one/integration.yaml"
+python -m telemetry_lab.cli run window --config "$OUT/two/integration.yaml"
+```
+
+The runner supplies each module's source path and starts commands from the
+disposable directory. Each consumer config is a fresh copy of `configs/default.yaml`
+with only `input_path` and `output_dir` changed. `integration_receipt.json` records
+actual argv, exit codes, versions and comparisons. Keep generated artifacts outside
+Git; paths in local receipts may reveal local directory names.
+
+## Expected results and boundaries
+
+| Check | Expected contract |
+| --- | --- |
+| Producer | Four successful commands; five JSON objects and report text match pinned golden content; JSONL matches exact bytes and parsed fixture records |
+| Loading and normalization | Seven rows; five required fields; source, target and nested metadata retained |
+| Time | All rows at `2026-07-05T00:05:00Z`; one half-open `[00:05:00Z, 00:06:00Z)` window |
+| Features | Eleven numeric fields; seven events, four sources, seven targets; error, severity and five configured category counts all zero |
+| Rules | Zero alerts with unchanged default rules, thresholds and cooldown |
+| Outputs | Seven artifacts per consumer run; both summary/v1 and manifest/v2 pass their schemas with format checks |
+| Repetition | Feature/alert CSV bytes and selected summary/manifest fields stable; config fingerprints legitimately differ because output paths differ |
+| Plots | Three PNGs per run decode with nonzero dimensions; no cross-platform byte equality claim |
+
+`metadata.time_semantics = snapshot_diff_observed_at` describes comparison
+observation time. Window membership uses `timestamp`; the consumer does not
+infer occurrence time or validate this advisory marker. `source` has the producer
+form `host_id:pid:start_time_ticks`, bounded to same-boot evidence without a boot
+identifier. Preservation proves neither cross-reboot identity nor authentication.
+Executable and endpoint targets remain context, not independent attribution.
+
+Negative and metamorphic controls cover missing required columns/values,
+malformed JSONL and timestamps, changed comparison time, equivalent UTC offsets,
+unknown categories, status changes across the existing error threshold, missing
+or altered advisory metadata, duplicate timestamps, row ordering, half-open
+boundaries and changed raw-byte fingerprints. Unknown categories and numeric
+identifiers remain accepted where existing normalization supports them. Metadata
+changes can leave aggregation unchanged; duplicate records are counted again.
+
+Per-file manifest fingerprints identify exact input/config bytes. Legacy aggregate
+digests separately include labels and normalized text. LF/CRLF changes therefore
+change the exact-byte fingerprint while preserving the legacy digest and features.
+These values provide content identification, not authenticated provenance.
+
+## Architecture and limitations
+
+Layer B stays an explicit standalone workflow. Adding a second repository checkout
+to required PR CI would introduce a network dependency without improving the
+offline consumer gate; using mutable producer `main` would weaken reproducibility.
+The ordinary CI and all release/publication workflows remain unchanged.
+
+No consumer admission policy, default detection rule, schema, golden demo output
+or producer implementation changes. This suite checks structural and bounded
+semantic compatibility and reproducibility, not detection efficacy or production
+readiness. It adds no collector, new demo, security attribution or private evidence.
+Dependency versions are reported rather than locked; other supported environments
+must run these checks rather than infer equivalent PNG bytes or performance.

scripts/reproduce_process_evidence.py

@@ -0,0 +1,182 @@
+"""Execute pinned saved-evidence producer commands and the real consumer twice."""
+from __future__ import annotations
+
+import argparse
+from hashlib import sha256
+from importlib.metadata import version
+import io
+import json
+import os
+from pathlib import Path, PurePosixPath
+import platform
+import subprocess
+import sys
+import tarfile
+
+from jsonschema import Draft202012Validator, FormatChecker
+from PIL import Image
+import yaml
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+from telemetry_lab.io import load_alert_table, load_config, load_feature_table  # noqa: E402
+
+PIN = "1d31d4287036cf2bbd3aeb8059d7d332bedef598"
+PROJECT = "projects/linux-process-observe"
+DIGEST = "1977b98518497c9a07da6393499cb49e6d0be9140ac5bcd708c2a1cd31a3ad51"
+EXPECTED = dict(event_count=7, error_count=0, error_rate=0.0, unique_sources=4,
+                unique_targets=7, high_severity_count=0, login_fail_count=0,
+                login_success_count=0, config_change_count=0, malware_alert_count=0,
+                policy_denied_count=0)
+
+
+def require(condition, message):
+    if not condition:
+        raise ValueError(message)
+
+
+def command(args, *, cwd, env=None):
+    result = subprocess.run([str(arg) for arg in args], cwd=cwd, env=env,
+                            capture_output=True, timeout=120)
+    if result.returncode:
+        raise ValueError("Command failed: " + result.stderr.decode("utf-8", errors="replace"))
+    return result.stdout
+
+
+def export_producer(repository, destination):
+    # Read immutable Git objects, never execute or overwrite the caller's checkout.
+    actual = command(["git", "rev-parse", PIN + "^{commit}"], cwd=repository).decode().strip()
+    require(actual == PIN, "Pinned producer commit is unavailable")
+    data = command(["git", "-c", "core.autocrlf=false", "-c", "core.eol=lf", "archive", "--format=tar", PIN, PROJECT], cwd=repository)
+    with tarfile.open(fileobj=io.BytesIO(data)) as archive:
+        for member in archive:
+            path = PurePosixPath(member.name)
+            require(not path.is_absolute() and ".." not in path.parts and "\\" not in member.name and ":" not in member.name,
+                    "Unsafe producer archive path")
+            # git archive includes the requested project's ancestor directory.
+            if member.isdir() and path.parts == ("projects",):
+                continue
+            require(path.parts[:2] == ("projects", "linux-process-observe"), "Unexpected producer archive scope")
+            if member.isdir():
+                continue
+            require(member.isfile(), "Producer archive must contain regular files")
+            output = destination.joinpath(*path.parts)
+            output.parent.mkdir(parents=True, exist_ok=True)
+            with archive.extractfile(member) as source:
+                output.write_bytes(source.read())
+    return destination / PROJECT
+
+
+def check_result(output, input_path, config_path):
+    summary = json.loads((output / "summary.json").read_text(encoding="utf-8"))
+    manifest = json.loads((output / "run_manifest.json").read_text(encoding="utf-8"))
+    for value, schema in ((summary, "telemetry_summary.schema.json"), (manifest, "run_manifest.v2.schema.json")):
+        contract = json.loads((ROOT / "schemas" / schema).read_text(encoding="utf-8"))
+        Draft202012Validator.check_schema(contract)
+        Draft202012Validator(contract, format_checker=FormatChecker()).validate(value)
+    counts = {key: summary[key] for key in ("normalized_event_count", "window_count", "feature_row_count", "alert_count")}
+    require(counts == dict(normalized_event_count=7, window_count=1, feature_row_count=1, alert_count=0), "Unexpected summary counts")
+    features = load_feature_table(output / "features.csv")
+    require(set(features.columns) == {"window_start", "window_end", *EXPECTED}, "Unexpected feature columns")
+    require(len(features) == 1 and features.iloc[0][list(EXPECTED)].to_dict() == EXPECTED, "Unexpected aggregate features")
+    row = features.iloc[0]
+    require(row.window_start.isoformat() == "2026-07-05T00:05:00+00:00" and row.window_end.isoformat() == "2026-07-05T00:06:00+00:00", "Unexpected observation window")
+    require(load_alert_table(output / "alerts.csv").empty, "Default rules unexpectedly alerted")
+    require(manifest["artifact_schema_versions"] == {"run_manifest": "run-manifest/v2", "telemetry_summary": "telemetry-summary/v1"}, "Unexpected artifact versions")
+    for kind, path in (("input", input_path), ("config", config_path)):
+        require(manifest[kind + "_file_digests"] == {f"external/{kind}/{path.name}": "sha256:" + sha256(path.read_bytes()).hexdigest()}, "Incorrect exact-byte fingerprint")
+        normalized = path.read_text(encoding="utf-8").encode("utf-8")
+        legacy = sha256(path.name.encode() + b"\0" + normalized + b"\0").hexdigest()
+        require(manifest[kind + "_digest"] == "sha256:" + legacy, "Incorrect legacy digest")
+    names = {Path(name).name for name in summary["generated_artifacts"]}
+    expected_names = {"features.csv", "alerts.csv", "summary.json", "run_manifest.json",
+                      "event_count_timeline.png", "error_rate_timeline.png", "alerts_timeline.png"}
+    require(names == {path.name for path in output.iterdir()} == expected_names, "Missing or unexpected output artifact")
+    for name in names:
+        if name.endswith(".png"):
+            with Image.open(output / name) as image:
+                require(image.format == "PNG" and min(image.size) > 0, "Invalid plot")
+                image.verify()
+    return summary, manifest
+
+
+def reproduce(repository, output):
+    repository, output = repository.resolve(), output.resolve()
+    require(not output.is_relative_to(ROOT) and not output.is_relative_to(repository), "Outputs must be outside both repositories")
+    output.mkdir(parents=True, exist_ok=False)  # refuse collisions; never remove existing outputs
+    producer = export_producer(repository, output / "pinned-source")
+    env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1", PYTHONNOUSERSITE="1", MPLBACKEND="Agg",
+               MPLCONFIGDIR=str(output / "plot-config"))
+    receipts = []
+    def execute(module, args, source):
+        invocation = [sys.executable, "-m", module, *map(str, args)]
+        stdout = command(invocation, cwd=output, env=dict(env, PYTHONPATH=str(source / "src")))
+        receipts.append({"argv": invocation, "exit_code": 0, "stdout": stdout.decode("utf-8", errors="replace")})
+    for name, instant in (("baseline", "2026-07-05T00:00:00Z"), ("changed", "2026-07-05T00:05:00Z")):
+        fixture = producer / "tests/fixtures" / name
+        execute("linux_process_observe", ["snapshot", "--proc-root", fixture / "proc", "--ss", fixture / "ss.txt",
+            "--host-id", "lab-host", "--observed-at", instant, "--output-dir", output / name], producer)
+    execute("linux_process_observe", ["diff", "--before-processes", output / "baseline/process_snapshot.json",
+        "--after-processes", output / "changed/process_snapshot.json", "--before-links", output / "baseline/process_socket_links.json",
+        "--after-links", output / "changed/process_socket_links.json", "--output-dir", output / "diff"], producer)
+    input_path = output / "diff/telemetry_events.jsonl"
+    execute("linux_process_observe", ["adapt", "--input", output / "diff/process_diff.json", "--output", input_path], producer)
+    producer_hashes = {}
+    for folder, names in (("baseline", ("process_snapshot.json", "process_socket_links.json")),
+                          ("changed", ("process_snapshot.json", "process_socket_links.json")),
+                          ("diff", ("process_diff.json", "report.md", "telemetry_events.jsonl"))):
+        for name in names:
+            data = (output / folder / name).read_bytes()
+            golden = (producer / "tests/golden" / folder / name).read_bytes()
+            # Compare structured contracts; only the JSONL contract is byte-exact.
+            if name.endswith(".json"):
+                equal = json.loads(data) == json.loads(golden)
+            elif name.endswith(".md"):
+                equal = data.decode().replace("\r\n", "\n") == golden.decode().replace("\r\n", "\n")
+            else:
+                equal = data == golden
+            require(equal, "Producer output differs from pinned golden: " + folder + "/" + name)
+            producer_hashes[folder + "/" + name] = sha256(data).hexdigest()
+    require(producer_hashes["diff/telemetry_events.jsonl"] == DIGEST, "Producer JSONL digest changed")
+    generated = [json.loads(line) for line in input_path.read_text(encoding="utf-8").splitlines()]
+    canonical = [json.loads(line) for line in (ROOT / "tests/fixtures/process_evidence/telemetry_events.jsonl").read_text(encoding="utf-8").splitlines()]
+    require(generated == canonical, "Generated producer records differ semantically from the consumer fixture")
+    results = []
+    for name in ("one", "two"):
+        directory = output / name
+        directory.mkdir()
+        config = load_config(ROOT / "configs/default.yaml")
+        config.update(input_path=str(input_path), output_dir=str(directory / "artifacts"))
+        path = directory / "integration.yaml"
+        path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8", newline="\n")
+        execute("telemetry_lab.cli", ["run", "window", "--config", path], ROOT)
+        results.append(check_result(directory / "artifacts", input_path, path))
+    for name in ("features.csv", "alerts.csv"):
+        require((output / "one/artifacts" / name).read_bytes() == (output / "two/artifacts" / name).read_bytes(), "Structured output is unstable")
+    stable_summary = ("normalized_event_count", "window_count", "feature_row_count", "alert_count", "triggered_rule_names", "triggered_rule_counts", "cooldown_seconds")
+    require({key: results[0][0][key] for key in stable_summary} == {key: results[1][0][key] for key in stable_summary}, "Summary metrics are unstable")
+    for key in ("tool_version", "demo_id", "input_digest", "input_file_digests", "artifact_schema_versions", "execution_mode"):
+        require(results[0][1][key] == results[1][1][key], "Manifest input contract is unstable")
+    report = dict(producer_commit=PIN, consumer_base="de542cb3cadba776c8e759f786152f9222fdfdb7",
+        python=platform.python_version(), platform=platform.system(),
+        packages={name: version(name) for name in ("telemetry-lab", "pandas", "matplotlib", "PyYAML", "jsonschema")},
+        commands=receipts, producer_output_sha256=producer_hashes, schema_validations=4,
+        producer_comparison="five JSON objects; report text with normalized newlines; JSONL exact bytes and parsed records",
+        structured_outputs="stable; output paths/config fingerprints differ by design", png="six decoded PNG smoke checks; no cross-platform byte claim")
+    (output / "integration_receipt.json").write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8", newline="\n")
+    print("PASS: four producer commands, seven golden contracts, two consumer runs, four schema validations")
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--producer", type=Path, required=True, help="local systems-foundations Git repository containing the exact pin")
+    parser.add_argument("--output-root", type=Path, required=True, help="new disposable directory outside both repositories")
+    args = parser.parse_args(argv)
+    try:
+        reproduce(args.producer, args.output_root)
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
+        parser.exit(1, f"FAIL: {error}\n")
+
+
+if __name__ == "__main__":
+    main()