diff --git a/tests/plugins_tests/test_ecpa_plugin_evidence.py b/tests/plugins_tests/test_ecpa_plugin_evidence.py index ebda1490420c..0f0596dc7173 100644 --- a/tests/plugins_tests/test_ecpa_plugin_evidence.py +++ b/tests/plugins_tests/test_ecpa_plugin_evidence.py @@ -13,6 +13,7 @@ import pytest import vllm.plugins as plugins +from vllm.entrypoints.cli.main import _bind_serve_process_identity from vllm.plugins import evidence Payload = dict[str, Any] @@ -75,6 +76,21 @@ def expected_event_id(payload: Payload) -> str: return hashlib.sha256(material).hexdigest() +def test_serve_cli_binds_parent_before_general_plugin_loading(): + _bind_serve_process_identity(["vllm", "serve", "model"]) + process = evidence.capture_scope()["process"] + assert (process["role"], process["ordinal"], process["assignment_source"]) == ( + "api-server", + 0, + "host", + ) + + +def test_non_serve_cli_does_not_claim_api_server_role(): + _bind_serve_process_identity(["vllm", "bench"]) + assert evidence.capture_scope()["process"]["assignment_source"] == "environment" + + def test_event_id_and_deduplication_bind_assignment_source(monkeypatch): events: list[Payload] = [] enable(events) diff --git a/vllm/entrypoints/cli/main.py b/vllm/entrypoints/cli/main.py index c85bec4f1809..bfd6f022d3bf 100644 --- a/vllm/entrypoints/cli/main.py +++ b/vllm/entrypoints/cli/main.py @@ -14,6 +14,15 @@ logger = init_logger(__name__) +def _bind_serve_process_identity(argv: list[str]) -> None: + """Bind the single-process serve parent before parser plugins execute.""" + if argv[1:2] != ["serve"]: + return + from vllm.plugins.evidence import bind_process_identity + + bind_process_identity("api-server", 0) + + def maybe_run_omni() -> bool: # If `--omni` arg is passed to the CLI, delegate to vLLM Omni's entrypoint handling if "--omni" not in sys.argv: @@ -40,6 +49,10 @@ def main(): if maybe_run_omni(): return + # Parser construction loads general plugins. Bind before those imports so + # the parent event is host-assigned rather than `unknown`/environment. + _bind_serve_process_identity(sys.argv) + import vllm.entrypoints.cli.benchmark.main import vllm.entrypoints.cli.collect_env import vllm.entrypoints.cli.download_kernels diff --git a/vllm/plugins/native_attention_operator_observer.py b/vllm/plugins/native_attention_operator_observer.py new file mode 100644 index 000000000000..6da91c1cfa3e --- /dev/null +++ b/vllm/plugins/native_attention_operator_observer.py @@ -0,0 +1,675 @@ +"""Opt-in source contacts; callback return is not attention/formal qualification.""" + +from __future__ import annotations + +import hashlib +import json +import math +import time +from enum import IntEnum + +SCHEMA = "ecpa-native-cached-attention-operator-source-candidate/v1" +_publisher = None +_scope = None +_current_stream = None +_get_format = None +_read_prefix = None +_materialized = {} +_metadata = {} +_observed = set() + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + + +def sha(value): + return hashlib.sha256(value).hexdigest() + + +def require(value, reason): + if not value: + raise ValueError(reason) + + +def enabled(): + return _publisher is not None + + +def register_source_candidate( + publisher, *, scope, current_stream, get_format, read_prefix=None +): + """Explicit new source-only opt-in. No existing source producer is upgraded.""" + global _publisher, _scope, _current_stream, _get_format, _read_prefix + require(not enabled(), "attention source callback already registered") + require( + all(callable(f) for f in (publisher, scope, current_stream, get_format)), + "attention source callbacks missing", + ) + _publisher, _scope = publisher, scope + _current_stream, _get_format = current_stream, get_format + _read_prefix = read_prefix or read_destination_prefix + + +def remember_acknowledged_materialization(fact, ack): + """Called ONLY after actual artifact publish validates its Unix source ACK.""" + if not enabled(): + return + scope = _scope() + require( + fact["native_producer"] == scope["process"], "materialization worker changed" + ) + require( + fact["classification"] == "REAL_RUNTIME_UNQUALIFIED", + "source candidate requires actual native runtime materialization", + ) + require( + ack["schema"] == "ecpa-native-semantic-collector-ack/v1" + and ack["formal_producer_registration"] is False + and ack["ack_sha256"] + == sha(canonical({k: v for k, v in ack.items() if k != "ack_sha256"})), + "actual source ACK is absent or corrupt", + ) + require(ack["phase"]["expected_ranks"] == [0, 1], "source ACK rank scope changed") + binding = fact["binding"] + key = (binding["runtime_request_id"], binding["layer_name"]) + require(len(_materialized) < 64, "bounded fresh materialization scope exceeded") + require(key not in _materialized, "ambiguous materialization for request/layer") + # Actual ACK raw digest binds the transport envelope, whereas this digest + # binds its privately archived typed payload; they are intentionally distinct. + _materialized[key] = { + "fact": fact, + "fact_sha256": sha(canonical(fact)), + "ack": ack, + "registered_monotonic_ns": time.monotonic_ns(), + } + + +def bind_actual_input_batch(runner, metadata, *, eager): + """Source-owned IDs and row mapping after real attention metadata construction.""" + if not enabled(): + return + _metadata.clear() + # These are explicit unsupported paths, not inferred successful consumption. + if not eager or isinstance(metadata, list): + return + require(isinstance(metadata, dict), "per-layer metadata mapping unsupported") + req_ids = runner.input_batch.req_ids[: runner.input_batch.num_reqs] + if len(req_ids) != 1: + return + request_id = req_ids[0] + require( + runner.input_batch.req_id_to_index[request_id] == 0, + "actual InputBatch request row differs", + ) + require(runner.requests[request_id] is not None, "actual runner request absent") + for layer_name, value in metadata.items(): + if (request_id, layer_name) not in _materialized: + continue + state = getattr(getattr(value, "attn_state", None), "name", None) + if state != "DecodeOnly": + continue + require( + value.num_prefills == 0 and value.num_decodes == 1, + "cached decode metadata includes another request", + ) + require(value.actual_seq_lengths_q == [1], "decode is not one actual token") + tables = runner.input_batch.block_table.block_tables + require(len(tables) == 1, "only actual single attention group supported") + table = tables[0] + width = int(table.num_blocks_per_row[0]) + require(0 < width <= table.block_table.gpu.shape[1], "actual row width invalid") + cpu_blocks = table.block_table.np[0, :width].tolist() + require( + cpu_blocks == list(runner.requests[request_id].block_ids[0]), + "actual CPU row and request metadata differ", + ) + lengths = value.seq_lens.detach().cpu().tolist() + require( + lengths == list(value.seq_lens_list) and len(lengths) == 1, + "actual native decode sequence metadata fields diverged", + ) + _metadata[(id(value), layer_name)] = { + "object": value, + "runtime_request_id": request_id, + "layer_name": layer_name, + "metadata_row": 0, + "actual_cpu_block_ids": cpu_blocks, + "metadata_seq_lengths_kv": lengths, + "metadata_actual_seq_lengths_q": list(value.actual_seq_lengths_q), + "bound_monotonic_ns": time.monotonic_ns(), + } + + +# Explicit exact vendor codes, excluding UNDEFINED. New/unknown formats are not +# silently mapped to a known storage layout by integer coercion. +_VENDOR_FORMAT_SOURCE_SHA256 = ( + "8abe7bea6817f25d2360aee783afae16fe7fe2aaa46a04e7b7b865e7f2039488" +) +_NPU_FORMAT_CODES = frozenset( + {0, 1, 2, 3, 4, 12, 16, 27, 29, 30, 32, 33, 35, 47, 50, 51, 52, 53, 54} +) + + +class SourceObservationRejected(ValueError): + """Bounded real rejected descriptors, never an OP receipt or source ACK.""" + + def __init__(self, reason, fields, *, stage): + self.diagnostic = { + "schema": "ecpa-native-attention-source-rejection/v1", + "classification": "SOURCE_REJECTION_ONLY", + "stage": stage, + "reason": str(reason)[:256], + "actual_fields": fields, + "OP_receipt": False, + "formal_registration": False, + } + encoded = canonical(self.diagnostic) + require(len(encoded) <= 8192, "source rejection diagnostic exceeds bound") + super().__init__(str(reason)[:256] + "; diagnostic=" + encoded.decode()) + + +def _field_descriptor(value): + return { + "python_type": (type(value).__module__ + "." + type(value).__qualname__)[:80], + "value_repr": repr(value)[:80], + } + + +def canonical_native_format(value): + """Only a plain int or this actual vendor Format is a wire integer.""" + if type(value) is int: + code = value + else: + if not isinstance(value, IntEnum): + raise SourceObservationRejected( + "native NPU format type unsupported", + {"npu_format": _field_descriptor(value)}, + stage="cache-view-format", + ) + try: + from pathlib import Path + + import torch_npu.npu._format as vendor_format + + Format = vendor_format.Format + actual_vendor_source = Path(vendor_format.__file__).read_bytes() + actual_source_sha = sha(actual_vendor_source) + if actual_source_sha != _VENDOR_FORMAT_SOURCE_SHA256: + raise SourceObservationRejected( + "native vendor format source bytes changed", + { + "actual_source_sha256": _field_descriptor(actual_source_sha), + "expected_source_sha256": _field_descriptor( + _VENDOR_FORMAT_SOURCE_SHA256 + ), + "npu_format": _field_descriptor(value), + }, + stage="cache-view-format", + ) + except (ImportError, OSError) as exc: + raise SourceObservationRejected( + "native vendor format source unavailable", + {"npu_format": _field_descriptor(value)}, + stage="cache-view-format", + ) from exc + if Format is None or type(value) is not Format: + raise SourceObservationRejected( + "native NPU format type unsupported", + {"npu_format": _field_descriptor(value)}, + stage="cache-view-format", + ) + code = int(value) + if code not in _NPU_FORMAT_CODES: + raise SourceObservationRejected( + "native NPU format code unsupported", + {"npu_format": _field_descriptor(value)}, + stage="cache-view-format", + ) + return code + + +def _cache_rejection_fields(observed, materialization): + fields = {} + for label, rows in ( + ("actual", observed["native_arguments"]["cache_views"]), + ( + "reference", + {k: v["after"] for k, v in materialization["cache_views"].items()}, + ), + ): + for kind, view in rows.items(): + fields[label + "." + kind] = { + name: _field_descriptor(value) + for name, value in view.items() + if name + in { + "device", + "dtype", + "shape", + "stride", + "data_ptr", + "storage_ptr", + "storage_offset", + "npu_format", + } + } + fields["runtime_request_id"] = _field_descriptor(observed["runtime_request_id"]) + fields["layer_name"] = _field_descriptor(observed["layer_name"]) + for name in ("block_table_rows", "sequence_lengths"): + fields["native_arguments." + name] = _field_descriptor( + observed["native_arguments"][name] + ) + for name in ( + "actual_cpu_block_ids", + "metadata_seq_lengths_kv", + "metadata_bound_monotonic_ns", + "native_entry_readback_started_ns", + "native_entry_readback_completed_ns", + "native_call_started_ns", + "native_call_returned_ns", + "native_stream_completed_ns", + "monotonic_ns", + ): + fields[name] = _field_descriptor(observed[name]) + return fields + + +def cache_view(tensor): + return { + "device": str(tensor.device), + "dtype": str(tensor.dtype), + "shape": list(tensor.shape), + "stride": list(tensor.stride()), + "data_ptr": tensor.data_ptr(), + "storage_ptr": tensor.untyped_storage().data_ptr(), + "storage_offset": tensor.storage_offset(), + "npu_format": canonical_native_format(_get_format(tensor)) + if str(tensor.device).startswith("npu:") + else None, + } + + +def read_destination_prefix(cache, fact, kind): + """Actual logical block exports at OP entry; independent of KV gather helper.""" + import torch + + reference_shape = fact["cache_views"][kind]["after"]["shape"] + reference = fact["cache_views"][kind]["after"] + actual_shape, actual_stride = list(cache.shape), list(cache.stride()) + require( + (actual_shape == reference_shape and actual_stride == reference["stride"]) + or ( + actual_shape + == [ + reference_shape[0], + reference_shape[1], + reference_shape[2] * reference_shape[3], + ] + and actual_stride == [reference["stride"][0], reference["stride"][1], 1] + ), + "native cache layout is not actual typed 4D or exact flatten", + ) + require( + str(cache.dtype) in {"torch.float16", "torch.bfloat16"} + and str(cache.dtype) == reference["dtype"], + "native prefix reader supports actual binary16 cache dtype only", + ) + logical = cache.view(reference_shape) + remaining = len(fact["keep_indices"]) + block_size = fact["block_size"] + digest = hashlib.sha256() + for block in fact["destination_blocks"]: + if remaining <= 0: + break + count = min(remaining, block_size) + # Copy the actual logical block through framework device-to-CPU. No + # compression workspace or previous digest supplies these bytes. + raw = logical[block].detach().to(device="cpu", non_blocking=False).contiguous() + raw = raw.view(torch.uint8).numpy().tobytes() + token_bytes = reference_shape[2] * reference_shape[3] * 2 + require( + len(raw) == block_size * token_bytes, "native cache block extent differs" + ) + digest.update(raw[: count * token_bytes]) + remaining -= count + require(remaining == 0, "native destination prefix allocation incomplete") + return digest.hexdigest() + + +def argument_descriptor(value): + if hasattr(value, "data_ptr") and hasattr(value, "untyped_storage"): + return {"kind": "tensor", "view": cache_view(value)} + if value is None or type(value) in (str, bool, int): + return {"kind": "scalar", "value": value} + if type(value) is float and math.isfinite(value): + return {"kind": "scalar", "value": value} + if isinstance(value, (list, tuple)) and len(value) <= 4096: + return {"kind": "sequence", "value": [argument_descriptor(x) for x in value]} + raise ValueError("actual native argument type unsupported") + + +def verify_observation(value, materialization): + """Pure bounded source predicate; cannot authenticate sender or prove quality.""" + require( + value["schema"] == SCHEMA + and value["classification"] == "REAL_RUNTIME_UNQUALIFIED" + and value["formal_registration"] is False + and all( + value[k] is None + for k in ( + "GPU_qualification", + "attention_consumption", + "full_model_quality", + ) + ), + "attention source candidate scope promoted", + ) + require( + value["operator"] + in { + "torch_npu._npu_paged_attention", + "torch_npu.npu_fused_infer_attention_score", + } + and value["attn_state"] == "DecodeOnly" + and value["execution_mode"] == "eager-single-request-cached-decode", + "unsupported actual attention branch", + ) + descriptors = value["native_argument_descriptors"] + require( + value["native_argument_descriptors_sha256"] == sha(canonical(descriptors)), + "native argument descriptor bytes changed", + ) + query = descriptors["query"] + require( + query["kind"] == "tensor" + and query["view"]["device"] + == materialization["cache_views"]["K"]["after"]["device"] + and query["view"]["dtype"] + == materialization["cache_views"]["K"]["after"]["dtype"], + "native query is not actual matching-device tensor", + ) + binding = materialization["binding"] + require( + value["runtime_request_id"] == binding["runtime_request_id"] + and value["layer_name"] == binding["layer_name"] + and value["layer_index"] == binding["layer_index"] + and value["process"] == materialization["native_producer"], + "attention request/actual layer/worker does not match materialization", + ) + transaction = sha( + canonical( + { + "runtime_request_id": binding["runtime_request_id"], + "compression_plan": binding["compression_plan"], + } + ) + ) + require( + value["materialization_transaction_id"] == transaction, + "actual source compression plan transaction differs", + ) + require( + value["source_materialization_sha256"] == sha(canonical(materialization)), + "attention reference is not actual archived materialization", + ) + clocks = [ + materialization["destination_readback_completed_ns"], + value["materialization_ack_registered_monotonic_ns"], + value["metadata_bound_monotonic_ns"], + value["native_entry_readback_started_ns"], + value["native_entry_readback_completed_ns"], + value["native_call_started_ns"], + value["native_call_returned_ns"], + value["native_stream_completed_ns"], + value["monotonic_ns"], + ] + require( + all(type(x) is int and x > 0 for x in clocks) and clocks == sorted(clocks), + "actual native completion chronology differs", + ) + require( + type(value["stream"]["npu_stream"]) is int + and value["stream"]["npu_stream"] >= 0 + and value["stream_before"] == value["stream_after"] == value["stream"], + "native operation changed completion stream", + ) + rows = value["native_arguments"]["block_table_rows"] + seq = value["native_arguments"]["sequence_lengths"] + require( + isinstance(rows, list) + and len(rows) == 1 + and all(type(x) is int and x >= 0 for x in rows[0]) + and len(rows[0]) <= 4096 + and isinstance(seq, list) + and len(seq) == 1 + and type(seq[0]) is int + and seq[0] >= len(materialization["keep_indices"]), + "native block table or sequence extent is unsupported", + ) + require( + value["actual_cpu_block_ids"] + and rows[0][: len(value["actual_cpu_block_ids"])] + == value["actual_cpu_block_ids"], + "native OP block table differs from source-owned actual InputBatch row", + ) + require( + seq == value["metadata_seq_lengths_kv"] + and value["metadata_actual_seq_lengths_q"] == [1], + "actual native sequence lengths differ from bound source metadata", + ) + destinations = materialization["destination_blocks"] + require( + rows[0][: len(destinations)] == destinations, + "actual native block table does not address materialized destinations", + ) + require( + len(value["actual_cpu_block_ids"]) * materialization["block_size"] >= seq[0], + "native sequence extends beyond actual block table", + ) + require( + value["stream"]["device"] + == materialization["cache_views"]["K"]["after"]["device"], + "native completion stream is on another device", + ) + ack = value["materialization_ack"] + require( + ack["schema"] == "ecpa-native-semantic-collector-ack/v1" + and ack["formal_producer_registration"] is False + and ack["ack_sha256"] + == sha(canonical({k: v for k, v in ack.items() if k != "ack_sha256"})), + "materialization acknowledgement provenance changed", + ) + require( + value["native_selected_destination_sha256"] + == materialization["destination_readback_sha256"], + "actual selected destination bytes changed before native OP entry", + ) + for kind in ("K", "V"): + native_name = ( + ("key_cache" if kind == "K" else "value_cache") + if (value["operator"] == "torch_npu._npu_paged_attention") + else ("key" if kind == "K" else "value") + ) + require( + descriptors[native_name]["kind"] == "tensor" + and descriptors[native_name]["view"] + == value["native_arguments"]["cache_views"][kind], + "native actual argument and interpreted cache view differ", + ) + actual = value["native_arguments"]["cache_views"][kind] + ref = materialization["cache_views"][kind]["after"] + require( + set(actual) + == { + "device", + "dtype", + "shape", + "stride", + "data_ptr", + "storage_ptr", + "storage_offset", + "npu_format", + } + and all( + type(actual[k]) is int and actual[k] > 0 + for k in ("data_ptr", "storage_ptr") + ) + and type(actual["storage_offset"]) is int + and actual["storage_offset"] >= 0 + and type(actual["npu_format"]) is int + and actual["npu_format"] in _NPU_FORMAT_CODES, + "native cache view fields/types unsupported", + ) + require( + actual["device"].startswith("npu:") + and all( + actual[k] == ref[k] + for k in ( + "device", + "dtype", + "storage_ptr", + "storage_offset", + "data_ptr", + "npu_format", + ) + ), + "native OP uses another cache allocation/view/format", + ) + require( + (actual["shape"] == ref["shape"] and actual["stride"] == ref["stride"]) + or ( + value["operator"] == "torch_npu.npu_fused_infer_attention_score" + and actual["shape"] + == [ref["shape"][0], ref["shape"][1], ref["shape"][2] * ref["shape"][3]] + and actual["stride"] == [ref["stride"][0], ref["stride"][1], 1] + ), + "native OP view is not actual cache or supported exact flatten", + ) + require(len(canonical(value)) <= 49152, "bounded native operator fact required") + return { + "source_argument_completion_relation": True, + "formal_registration": False, + "attention_consumption": None, + "GPU_qualification": None, + "full_model_quality": None, + } + + +def observe_native_call(operator, native_call, arguments, *, metadata, layer_name): + """Wrap actual final native call; default/unsupported paths produce no witness.""" + if not enabled(): + return native_call(**arguments) + metadata_key = (id(metadata), layer_name) + if metadata_key not in _metadata: + # Native groups share one metadata object across many layers. Each + # layer must retain its own source-owned runtime/materialization edge. + # A known object with an unbound layer (or vice versa) is not an + # unsupported branch that can silently bypass the source checks. + require( + not any( + bound["object"] is metadata or bound["layer_name"] == layer_name + for bound in _metadata.values() + ), + "native metadata/layer source binding changed", + ) + return native_call(**arguments) + bound = _metadata[metadata_key] + require( + bound["object"] is metadata and bound["layer_name"] == layer_name, + "native metadata/layer source binding changed", + ) + key_id = (bound["runtime_request_id"], layer_name) + if key_id in _observed: + return native_call(**arguments) + registered = _materialized[key_id] + fact = registered["fact"] + require( + registered["fact_sha256"] == sha(canonical(fact)), + "registered materialization changed", + ) + if operator == "torch_npu._npu_paged_attention": + key, val = arguments["key_cache"], arguments["value_cache"] + seq = arguments["context_lens"].detach().cpu().tolist() + else: + key, val = arguments["key"], arguments["value"] + seq = list(arguments["actual_seq_lengths_kv"]) + table = arguments["block_table"].detach().cpu().tolist() + entry_readback_started = time.monotonic_ns() + entry_digests = { + "K": _read_prefix(key, fact, "K"), + "V": _read_prefix(val, fact, "V"), + } + entry_readback_completed = time.monotonic_ns() + descriptors = { + name: argument_descriptor(value) for name, value in arguments.items() + } + stream = _current_stream(key.device) + stream_id = {"npu_stream": stream.npu_stream, "device": str(stream.device)} + observed = { + "schema": SCHEMA, + "classification": "REAL_RUNTIME_UNQUALIFIED", + "formal_registration": False, + **_scope(), + "operator": operator, + "execution_mode": "eager-single-request-cached-decode", + "attn_state": "DecodeOnly", + "runtime_request_id": bound["runtime_request_id"], + "layer_name": layer_name, + "layer_index": fact["binding"]["layer_index"], + "metadata_row": 0, + "actual_cpu_block_ids": bound["actual_cpu_block_ids"], + "metadata_seq_lengths_kv": bound["metadata_seq_lengths_kv"], + "metadata_actual_seq_lengths_q": bound["metadata_actual_seq_lengths_q"], + "source_materialization_sha256": registered["fact_sha256"], + "materialization_transaction_id": sha( + canonical( + { + "runtime_request_id": bound["runtime_request_id"], + "compression_plan": fact["binding"]["compression_plan"], + } + ) + ), + "materialization_ack": registered["ack"], + "materialization_ack_registered_monotonic_ns": registered[ + "registered_monotonic_ns" + ], + "metadata_bound_monotonic_ns": bound["bound_monotonic_ns"], + "native_selected_destination_sha256": entry_digests, + "native_entry_readback_started_ns": entry_readback_started, + "native_entry_readback_completed_ns": entry_readback_completed, + "native_argument_descriptors": descriptors, + "native_argument_descriptors_sha256": sha(canonical(descriptors)), + "native_arguments": { + "cache_views": {"K": cache_view(key), "V": cache_view(val)}, + "block_table_rows": table, + "sequence_lengths": seq, + }, + "stream": stream_id, + "stream_before": stream_id, + "native_call_started_ns": time.monotonic_ns(), + "GPU_qualification": None, + "attention_consumption": None, + "full_model_quality": None, + } + result = native_call(**arguments) + observed["native_call_returned_ns"] = time.monotonic_ns() + stream.synchronize() + observed["native_stream_completed_ns"] = time.monotonic_ns() + after = _current_stream(key.device) + observed["stream_after"] = { + "npu_stream": after.npu_stream, + "device": str(after.device), + } + observed["monotonic_ns"] = time.monotonic_ns() + try: + verify_observation(observed, fact) + except ValueError as exc: + raise SourceObservationRejected( + str(exc), + _cache_rejection_fields(observed, fact), + stage="after-native-completion-validation", + ) from exc + _publisher(observed) # Failure propagates; no return is an authority ACK claim. + _observed.add(key_id) + return result diff --git a/vllm/plugins/native_block_table_observer.py b/vllm/plugins/native_block_table_observer.py new file mode 100644 index 000000000000..e6ed11fa218e --- /dev/null +++ b/vllm/plugins/native_block_table_observer.py @@ -0,0 +1,127 @@ +"""Candidate Host-native facts. Opt-in; upload observation is not KV equivalence.""" + +from __future__ import annotations + +import hashlib +import json +import os +import time +from pathlib import Path + +_observer = None + + +def register_observer(observer): + """Register one source-pinned trusted observer in this native process.""" + global _observer + if _observer is not None: + raise RuntimeError("native observer already registered") + if not callable(observer): + raise TypeError("observer must be callable") + from .native_phase_drain import callback_with_drain + _observer = callback_with_drain(observer) + + +def _scope(role): + from vllm.plugins.evidence import _identity + + identity = _identity() + if identity["assignment_source"] != "host" or identity["role"] != role: + raise RuntimeError("native Host-assigned observer role required") + launch = os.environ.get("VLLM_ECPA_LAUNCH_ID") + plan = os.environ.get("VLLM_ECPA_PLAN_ID") + if not launch or not plan: + raise RuntimeError("native plan/launch binding missing") + return { + "process": { + **identity, + "session": os.getsid(0), + "start_ticks": int(identity["start_identity"].rsplit(":", 1)[1]), + "boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip(), + }, + "launch_id": launch, + "plan_id": plan, + "monotonic_ns": time.monotonic_ns(), + } + + +def observe_request_id_mapping(request): + if _observer is None: + from .native_semantic_bootstrap import ensure_registered + ensure_registered(register_observer) + if _observer is None: + return + if not request.external_req_id or not request.request_id: + raise RuntimeError("native external/internal request IDs missing") + _observer( + { + "schema": "ecpa-native-request-id-mapping/v1", + **_scope("api-server"), + "external_request_id": request.external_req_id, + "runtime_request_id": request.request_id, + } + ) + + +def observe_block_table_upload(runner, scheduler_output): + if _observer is None: + from .native_semantic_bootstrap import ensure_registered + ensure_registered(register_observer) + if _observer is None: + return + # Imported only for an opted-in observation. No global device sync, dummy + # graph interception or model/KV-tensor correctness assertion. + import torch + + batch = runner.input_batch + tables = batch.block_table.block_tables + if not tables: + raise RuntimeError("native block-table groups missing") + devices = {str(table.block_table.gpu.device) for table in tables} + if len(devices) != 1 or tables[0].block_table.gpu.device.type != "npu": + raise RuntimeError("real same-device NPU block tables required") + scope = _scope("worker") + # commit_block_table queues nonblocking H2D copies on this stream. This + # deliberately costly qualification barrier is not a production fast path. + torch.npu.current_stream(tables[0].block_table.gpu.device).synchronize() + scheduled = scheduler_output.num_scheduled_tokens + for row, request_id in enumerate(batch.req_ids[: batch.num_reqs]): + if scheduled.get(request_id, 0) <= 0: + continue + observed = [] + for group_id, table in enumerate(tables): + width = int(table.num_blocks_per_row[row]) + if width <= 0 or width > table.block_table.gpu.shape[1]: + raise RuntimeError("invalid active native block-table prefix") + cpu = table.block_table.np[row, :width].tolist() + device = table.block_table.gpu[row, :width] + gpu = device.to(device="cpu", non_blocking=False).tolist() + if cpu != gpu: + raise RuntimeError("uploaded NPU block-table prefix differs") + observed.append({"group": group_id, "row": row, "blocks": gpu}) + payload = json.dumps(observed, sort_keys=True, separators=(",", ":")) + _observer( + { + "schema": "ecpa-native-block-table-upload/v1", + **scope, + "observed_after_barrier_monotonic_ns": time.monotonic_ns(), + "runtime_request_id": request_id, + "device": next(iter(devices)), + "uploaded_rows": observed, + "uploaded_rows_sha256": hashlib.sha256(payload.encode()).hexdigest(), + "scope": "NPU-block-table-prefix-readback-only", + "kv_tensor_equivalence": None, + "attention_consumed_observed_tensor": None, + "full_generation_quality": None, + } + ) + + +def emit_native_fact(factory): + """Lazy source-owned extension fact; authentication/ACK failures propagate.""" + if _observer is None: + from .native_semantic_bootstrap import ensure_registered + + ensure_registered(register_observer) + if _observer is not None: + return _observer(factory()) diff --git a/vllm/plugins/native_kv_materialized_projection_observer.py b/vllm/plugins/native_kv_materialized_projection_observer.py new file mode 100644 index 000000000000..d38eca46977d --- /dev/null +++ b/vllm/plugins/native_kv_materialized_projection_observer.py @@ -0,0 +1,120 @@ +"""Read the actual materialized Host row after its native scheduler update.""" + +from __future__ import annotations + +import hashlib +import json +import time +from dataclasses import dataclass + +from .native_kv_projection_observer import OPERATION, SCOPE + +SCHEMA = "ecpa-native-kv-blocktable-projection/v2" + + +@dataclass(frozen=True) +class ProjectionMarker: + pending: object + appended_block_ids: tuple[int, ...] + metadata_effect_monotonic_ns: int + + +def capture_marker(provider, scheduler_output, request_id, pending): + """Called only after actual Request/active mutation, before native append.""" + cached = scheduler_output.scheduled_cached_reqs + slots = [i for i, value in enumerate(cached.req_ids) if value == request_id] + if len(slots) != 1 or request_id not in scheduler_output.num_scheduled_tokens: + raise ValueError("applied projection lacks exact native cached request") + group = provider.attention_group_index + new = cached.new_block_ids[slots[0]] + appended = tuple(new[group]) if new is not None else () + request = provider.runner.requests[request_id] + active = provider.active[request_id] + if ( + list(request.block_ids[group]) != list(pending.block_ids) + or active.semantic_anchor != pending.semantic_anchor + or active.physical_anchor != pending.physical_anchor + or any(type(x) is not int or x < 0 for x in appended) + ): + raise ValueError("marker does not follow actual applied metadata") + return ProjectionMarker(pending, appended, time.monotonic_ns()) + + +def projection_fact(provider, scheduler_output, request_id, marker, native_scope): + if request_id not in scheduler_output.num_scheduled_tokens: + raise ValueError("materialized projection request is not scheduled") + runner = provider.runner + row = runner.input_batch.req_id_to_index.get(request_id) + if row is None: + raise ValueError("actual Host CPU row is not materialized") + pending = marker.pending + group = provider.attention_group_index + table = runner.input_batch.block_table.block_tables[group] + width = int(table.num_blocks_per_row[row]) + observed = list(map(int, table.get_numpy_array()[row, :width])) + destination = list(pending.block_ids) + appended = list(marker.appended_block_ids) + expected_full = destination + appended + request_full = list(runner.requests[request_id].block_ids[group]) + active = provider.active[request_id] + block_size = table.block_size + threshold = provider.runtime_spec.compression_threshold_tokens + if ( + not destination + or width != len(expected_full) + or observed != expected_full + or request_full != observed + or len(set(expected_full)) != len(expected_full) + or any(type(x) is not int or x < 0 for x in expected_full) + or active.semantic_anchor != pending.semantic_anchor + or active.physical_anchor != pending.physical_anchor + or type(block_size) is not int + or block_size <= 0 + or pending.physical_anchor != len(destination) * block_size + or type(threshold) is not int + or not 0 < pending.physical_anchor < threshold <= pending.semantic_anchor + or not marker.metadata_effect_monotonic_ns <= native_scope["monotonic_ns"] + ): + raise ValueError( + "materialized Host row differs from actual destination/append/anchors" + ) + transaction = { + "runtime_request_id": request_id, + "attention_group": group, + "semantic_anchor": pending.semantic_anchor, + "physical_anchor": pending.physical_anchor, + "destination_block_ids": destination, + "host_cpu_block_ids": observed, + "allocation_block_size": block_size, + "compression_threshold_tokens": threshold, + "scheduled_appended_block_ids": appended, + "metadata_effect_monotonic_ns": marker.metadata_effect_monotonic_ns, + } + transaction_id = hashlib.sha256( + json.dumps( + transaction, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode() + ).hexdigest() + return { + "schema": SCHEMA, + **native_scope, + **transaction, + "metadata_row": row, + "transaction_id": transaction_id, + "operation": OPERATION, + "scope": SCOPE, + "host_mapping": "identity-host-block-id-projection-with-scheduled-append", + "kv_tensor_equivalence": None, + "attention_consumed_observed_tensor": None, + "full_generation_quality": None, + } + + +def observe_materialized_projection(provider, scheduler_output, request_id, marker): + from . import native_block_table_observer as native + + native.emit_native_fact( + lambda: projection_fact( + provider, scheduler_output, request_id, marker, native._scope("worker") + ) + ) diff --git a/vllm/plugins/native_kv_projection_observer.py b/vllm/plugins/native_kv_projection_observer.py new file mode 100644 index 000000000000..8b58f6cdb2d5 --- /dev/null +++ b/vllm/plugins/native_kv_projection_observer.py @@ -0,0 +1,86 @@ +"""Source candidate for actual KVCompress Host metadata, not tensor equivalence.""" + +from __future__ import annotations + +import hashlib +import json + +SCHEMA = "ecpa-native-kv-blocktable-projection/v1" +OPERATION = "KVCompress-worker-host-blocktable-projection-applied" +SCOPE = "compression-host-metadata-and-uploaded-destination-binding-only" + + +def projection_fact(provider, scheduler_output, request_id, pending, native_scope): + """Read after the acknowledged barrier and actual metadata mutations. + + The pinned Host table uses identity block IDs; an expanded/unknown mapping + is unsupported. No successful-effect flag or caller-supplied observed row. + """ + if request_id not in scheduler_output.num_scheduled_tokens: + raise ValueError("projection request lacks acknowledged scheduled barrier") + runner = provider.runner + request = runner.requests[request_id] + group = provider.attention_group_index + row = runner.input_batch.req_id_to_index[request_id] + table = runner.input_batch.block_table.block_tables[group] + destination = list(pending.block_ids) + width = int(table.num_blocks_per_row[row]) + observed = list(map(int, table.get_numpy_array()[row, :width])) + active = provider.active[request_id] + block_size = table.block_size + threshold = provider.runtime_spec.compression_threshold_tokens + if ( + not destination + or width != len(destination) + or list(request.block_ids[group]) != destination + or observed != destination + or active.semantic_anchor != pending.semantic_anchor + or active.physical_anchor != pending.physical_anchor + or type(block_size) is not int + or block_size <= 0 + or pending.physical_anchor != len(destination) * block_size + or type(threshold) is not int + or not 0 < pending.physical_anchor < threshold <= pending.semantic_anchor + ): + raise ValueError("actual Host projection/destination/anchors are inconsistent") + transaction = { + "runtime_request_id": request_id, + "attention_group": group, + "semantic_anchor": pending.semantic_anchor, + "physical_anchor": pending.physical_anchor, + "destination_block_ids": destination, + "host_cpu_block_ids": observed, + "allocation_block_size": block_size, + "compression_threshold_tokens": threshold, + } + transaction_id = hashlib.sha256( + json.dumps( + transaction, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode() + ).hexdigest() + return { + "schema": SCHEMA, + **native_scope, + **transaction, + "metadata_row": row, + "transaction_id": transaction_id, + "operation": OPERATION, + "scope": SCOPE, + "host_mapping": "identity-host-block-id-projection", + "kv_tensor_equivalence": None, + "attention_consumed_observed_tensor": None, + "full_generation_quality": None, + } + + +def observe_kv_projection(provider, scheduler_output, request_id, pending): + """New clean Host candidate hook; absent native observer emits no proof.""" + from . import native_block_table_observer as native + + # The public interface installs a source-owned bootstrap and throws on a + # rejected/authentication/ACK failure. Never return a success boolean. + native.emit_native_fact( + lambda: projection_fact( + provider, scheduler_output, request_id, pending, native._scope("worker") + ) + ) diff --git a/vllm/plugins/native_phase_drain.py b/vllm/plugins/native_phase_drain.py new file mode 100644 index 000000000000..a67eca3fa29c --- /dev/null +++ b/vllm/plugins/native_phase_drain.py @@ -0,0 +1,121 @@ +"""Prospective source-owned worker drain; candidate only, never formal admission.""" + +from __future__ import annotations + +import hashlib +import json +import os +import threading +import time + +SCHEMA = "ecpa-native-worker-request-drained/v1" +ENV = "ECPA_NATIVE_PHASE_DRAIN" +_records = {} +_closed = set() +_lock = threading.RLock() + + +def canonical(value): + return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() + + +def digest(value): + return hashlib.sha256(canonical(value)).hexdigest() + + +def record_ack(runtime_request_id, ack): + """Record an actual synchronous intake ACK, never an emitted Boolean.""" + if os.environ.get(ENV) != "1": + return + if runtime_request_id in _closed: + raise ValueError("native fact after source-owned request drain") + fields = ("sequence", "event_sha256", "raw_sha256") + if not isinstance(ack, dict) or any(k not in ack for k in fields): + raise ValueError("native drain requires actual authenticated intake ACK") + item = {k: ack[k] for k in fields} + if type(item["sequence"]) is not int or item["sequence"] < 0: + raise ValueError("native drain ACK sequence is unsupported") + if any(type(item[k]) is not str or len(item[k]) != 64 for k in fields[1:]): + raise ValueError("native drain ACK digests are unsupported") + records = _records.setdefault(runtime_request_id, []) + if records and item["sequence"] <= records[-1]["sequence"]: + raise ValueError("native drain ACK order/replay differs") + records.append(item) + + +def acknowledged_publication(runtime_request_id, publish): + """Serialize actual pending transports with source closure, not a timer.""" + if os.environ.get(ENV) != "1": + return publish() + with _lock: + if runtime_request_id in _closed: + raise ValueError("native publication after source-owned request drain") + ack = publish() + record_ack(runtime_request_id, ack) + return ack + + +def callback_with_drain(callback): + def wrapped(fact): + if ( + fact.get("process", {}).get("role") == "worker" + and fact.get("schema") != SCHEMA + ): + return acknowledged_publication( + fact["runtime_request_id"], lambda: callback(fact) + ) + return callback(fact) + + return wrapped + + +def _after_finished_request_removal(runner, scheduler_output, *, scope, publish): + """Called after pinned Host removal, before the zero-token early return. + + Worker execution and all three receipt transports are synchronous. No quiet + interval, owner-provided count, response-derived token fence or guessed row. + Resubmitted IDs cannot drain. Removal may follow finish or abort; this + marker proves transport closure, while strict HTTP/task success is separate. + """ + if os.environ.get(ENV) != "1": + return + for runtime in sorted(scheduler_output.finished_req_ids): + if runtime not in _records: + continue # warmup / requests without a candidate source registration + if runtime in _closed: + raise ValueError("native request drain is replayed") + if ( + runtime in runner.requests + or runtime in runner.input_batch.req_id_to_index + or runtime in scheduler_output.num_scheduled_tokens + or any(x.req_id == runtime for x in scheduler_output.scheduled_new_reqs) + ): + raise ValueError( + "actual native request has not been removed or was resubmitted" + ) + records = _records[runtime] + fact = { + "schema": SCHEMA, + **scope("worker"), + "runtime_request_id": runtime, + "source_ack_count": len(records), + "source_ack_chain_sha256": digest(records), + "source_last_ack": records[-1], + "native_requests_contains": False, + "native_input_batch_contains": False, + "scheduled_finished": True, + "source_drain_completed_monotonic_ns": time.monotonic_ns(), + "formal_producer_registration": False, + } + # The publication callback must synchronously verify the collector ACK. + ack = publish(fact) + if not isinstance(ack, dict) or ack.get("raw_sha256") != digest(fact): + raise ValueError("native source drain publication lacks actual raw ACK") + _closed.add(runtime) + + +def after_finished_request_removal(runner, scheduler_output, *, scope, publish): + with _lock: + return _after_finished_request_removal( + runner, scheduler_output, scope=scope, publish=publish + ) diff --git a/vllm/plugins/native_semantic_bootstrap.py b/vllm/plugins/native_semantic_bootstrap.py new file mode 100644 index 000000000000..74f0ccad24f4 --- /dev/null +++ b/vllm/plugins/native_semantic_bootstrap.py @@ -0,0 +1,135 @@ +"""Source-owned opt-in callback bootstrap for the native candidate observer. + +A context may be published after native process handoff but before HTTP probes. +Absent context is non-producing warmup, never qualification. Registered source +facts remain unchanged; only their transport and authenticated ACK are handled. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from .native_semantic_collector import ( + canonical, + read_private, + sha, + source_digests, + submit_fact, +) + +BOOTSTRAP = "ecpa-native-semantic-bootstrap/v1" +ENV = "ECPA_NATIVE_SEMANTIC_BOOTSTRAP" + + +def callback_from_context(path: Path): + import json + + raw, metadata = read_private(path, 0o600) + value = json.loads(raw) + required = { + "schema", + "classification", + "formal_producer_registration", + "socket_path", + "socket_identity", + "collector", + "collector_source_sha256", + "archive_identity", + "spec_path", + "spec_sha256", + "source_files", + } + if ( + set(value) != required + or value["schema"] != BOOTSTRAP + or value["classification"] != "CPU_REVIEW_CANDIDATE_ONLY" + or value["formal_producer_registration"] is not False + ): + raise ValueError("native bootstrap context is unsupported or promoted") + spec_raw, _ = read_private(Path(value["spec_path"]), 0o600) + if sha(spec_raw) != value["spec_sha256"]: + raise ValueError("native bootstrap spec differs from freeze") + spec = json.loads(spec_raw) + peers = {x["process"]["pid"]: x for x in spec["senders"]} + if os.getpid() not in peers: + raise ValueError("native bootstrap current process is not registered") + own_sender = peers[os.getpid()] + source_digests(value["source_files"]) + source_digests(own_sender["source_files"]) + frozen_identity = (metadata.st_dev, metadata.st_ino) + frozen_sha = sha(raw) + + def callback(fact): + current, info = read_private(path, 0o600) + if (info.st_dev, info.st_ino) != frozen_identity or sha(current) != frozen_sha: + raise ValueError("native bootstrap context changed after registration") + source_digests(value["source_files"]) + if fact.get("process") != own_sender["process"]: + raise ValueError("native fact process differs from own source registration") + # The fact is actual native output. Do not add nonce, role or IDs here. + ack = submit_fact( + Path(value["socket_path"]), + fact, + expected_collector=value["collector"], + expected_socket_identity=tuple(value["socket_identity"]), + expected_collector_source_sha256=value["collector_source_sha256"], + expected_archive_identity=value["archive_identity"], + ) + if ack["phase"] not in spec["phases"]: + raise ValueError("native ACK is not a frozen challenge phase") + return ack + + return callback + + +def ensure_registered(register_observer) -> bool: + path = os.environ.get(ENV) + if not path: + return False + context = Path(path) + if not context.is_absolute(): + raise ValueError("native bootstrap context path is not absolute") + # Native warmup precedes authority peer registration. No context means no + # receipts; the qualification checker must reject any later missing proof. + if not context.exists(): + return False + register_observer(callback_from_context(context)) + return True + + +def write_context( + path: Path, *, spec_path: Path, ready: dict, source_files: list +) -> str: + """Launcher helper; readiness must already be independently peer/source checked.""" + value = { + "schema": BOOTSTRAP, + "classification": "CPU_REVIEW_CANDIDATE_ONLY", + "formal_producer_registration": False, + "socket_path": ready["socket_path"], + "socket_identity": ready["socket_identity"], + "collector": ready["collector"], + "collector_source_sha256": ready["collector_source_sha256"], + "archive_identity": ready["archive_identity"], + "spec_path": str(spec_path), + "spec_sha256": ready["spec_sha256"], + "source_files": source_files, + } + raw = canonical(value) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + try: + data = memoryview(raw) + while data: + n = os.write(fd, data) + if n <= 0: + raise OSError("native bootstrap write failed") + data = data[n:] + os.fsync(fd) + finally: + os.close(fd) + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + return sha(raw) diff --git a/vllm/plugins/native_semantic_collector.py b/vllm/plugins/native_semantic_collector.py new file mode 100644 index 000000000000..a005ce4aa1b8 --- /dev/null +++ b/vllm/plugins/native_semantic_collector.py @@ -0,0 +1,1045 @@ +"""Private Unix authority-owned raw collector; qualification candidate only. + +Kernel peer identity plus frozen source and challenge bindings precede admission +of raw facts. Native fact authenticity still requires the newly built source +candidate and real qualification; this module never opens a formal traffic gate. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import json +import os +import re +import socket +import socketserver +import stat +import struct +import threading +import time +from pathlib import Path +from typing import Any + +MAX_FRAME = 65536 +MAPPING = "ecpa-native-request-id-mapping/v1" +UPLOAD = "ecpa-native-block-table-upload/v1" +SPEC = "ecpa-native-semantic-collector-prereg/v1" +ACK = "ecpa-native-semantic-collector-ack/v1" +KV_SPEC = "ecpa-native-kv-semantic-collector-prereg/v1" +PROJECTION = "ecpa-native-kv-blocktable-projection/v1" +PROJECTION_V2 = "ecpa-native-kv-blocktable-projection/v2" +KV_SPEC_V2 = "ecpa-native-kv-semantic-collector-prereg/v2" +PROJECTION_OPERATION = "KVCompress-worker-host-blocktable-projection-applied" +PROJECTION_SCOPE = "compression-host-metadata-and-uploaded-destination-binding-only" + + +def projection_predicate(fact, requirement): + transaction_keys = { + "runtime_request_id", + "attention_group", + "semantic_anchor", + "physical_anchor", + "destination_block_ids", + "host_cpu_block_ids", + "allocation_block_size", + "compression_threshold_tokens", + } + materialized = fact.get("schema") == PROJECTION_V2 + if materialized: + transaction_keys |= { + "scheduled_appended_block_ids", + "metadata_effect_monotonic_ns", + } + exact = transaction_keys | { + "schema", + "process", + "launch_id", + "plan_id", + "monotonic_ns", + "metadata_row", + "transaction_id", + "operation", + "scope", + "host_mapping", + "kv_tensor_equivalence", + "attention_consumed_observed_tensor", + "full_generation_quality", + } + if set(fact) != exact: + raise ValueError("KV projection fields are unsupported") + blocks = fact["destination_block_ids"] + appended = fact.get("scheduled_appended_block_ids", []) + if materialized and ( + not isinstance(appended, list) + or any(type(x) is not int or x < 0 for x in appended) + or type(fact["metadata_effect_monotonic_ns"]) is not int + or not 0 < fact["metadata_effect_monotonic_ns"] <= fact["monotonic_ns"] + ): + raise ValueError("materialized projection lacks actual append/effect clock") + integer_keys = [ + "attention_group", + "semantic_anchor", + "physical_anchor", + "allocation_block_size", + "compression_threshold_tokens", + "metadata_row", + ] + if ( + any(type(fact[k]) is not int or fact[k] < 0 for k in integer_keys) + or not isinstance(blocks, list) + or not blocks + or any(type(x) is not int or x < 0 for x in blocks) + or len(set(blocks)) != len(blocks) + or fact["host_cpu_block_ids"] != blocks + appended + or len(set(blocks + appended)) != len(blocks + appended) + or not 0 + < fact["physical_anchor"] + < fact["compression_threshold_tokens"] + <= fact["semantic_anchor"] + or fact["allocation_block_size"] <= 0 + or fact["physical_anchor"] != len(blocks) * fact["allocation_block_size"] + or any( + fact[k] != requirement[k] + for k in [ + "attention_group", + "semantic_anchor", + "physical_anchor", + "allocation_block_size", + "compression_threshold_tokens", + ] + ) + or len(blocks) != requirement["destination_block_count"] + or fact["operation"] != PROJECTION_OPERATION + or fact["scope"] != PROJECTION_SCOPE + or fact["host_mapping"] + != ( + "identity-host-block-id-projection-with-scheduled-append" + if materialized + else "identity-host-block-id-projection" + ) + or any( + fact[k] is not None + for k in [ + "kv_tensor_equivalence", + "attention_consumed_observed_tensor", + "full_generation_quality", + ] + ) + or fact["transaction_id"] + != sha(canonical({k: fact[k] for k in transaction_keys})) + ): + raise ValueError("KV projection lacks frozen actual destination/anchor binding") + + +def projection_upload_match(projection, upload): + """Only later same-worker destination-prefix binding; no tensor equivalence.""" + if ( + projection["process"] != upload["process"] + or any( + projection[k] != upload[k] + for k in ["runtime_request_id", "launch_id", "plan_id"] + ) + or upload["monotonic_ns"] < projection["monotonic_ns"] + or upload["observed_after_barrier_monotonic_ns"] < projection["monotonic_ns"] + ): + return False + destination = projection["host_cpu_block_ids"] + return any( + row["group"] == projection["attention_group"] + and row["blocks"][: len(destination)] == destination + for row in upload["uploaded_rows"] + ) + + +def canonical(value: Any) -> bytes: + return json.dumps( + value, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode() + + +def sha(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest() + + +def identity(pid: int) -> dict: + raw = Path(f"/proc/{pid}/stat").read_bytes() + fields = raw[raw.rfind(b")") + 2 :].split() + ticks = int(fields[19]) + return { + "pid": pid, + "start_ticks": ticks, + "start_identity": f"pid:{pid}:start_ticks:{ticks}", + "boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip(), + "session": os.getsid(pid), + } + + +def read_private(path: Path, mode: int) -> tuple[bytes, os.stat_result]: + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + try: + metadata = os.fstat(fd) + if ( + not stat.S_ISREG(metadata.st_mode) + or metadata.st_uid != os.getuid() + or stat.S_IMODE(metadata.st_mode) != mode + or metadata.st_size > MAX_FRAME + ): + raise ValueError("frozen input is not a bounded private regular file") + return os.read(fd, MAX_FRAME + 1), metadata + finally: + os.close(fd) + + +def source_digests(files: list[dict]) -> dict: + if not files: + raise ValueError("sender source bytes are unavailable") + result = {} + for binding in files: + path = Path(binding["path"]) + if not path.is_absolute() or path.is_symlink() or str(path) in result: + raise ValueError("sender source path is not immutable and unique") + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + try: + metadata = os.fstat(fd) + if not stat.S_ISREG(metadata.st_mode): + raise ValueError("sender source is not a regular file") + chunks = [] + while chunk := os.read(fd, MAX_FRAME): + chunks.append(chunk) + actual = sha(b"".join(chunks)) + finally: + os.close(fd) + if actual != binding["sha256"]: + raise ValueError("sender source bytes changed") + result[str(path)] = actual + return result + + +def native_fact_predicate( + raw: bytes, + sender: dict, + spec: dict, + *, + accepted_monotonic_ns: int | None, + mappings: dict, + mapping_stamps: dict, + upload_stamps: dict, + upload_counts: dict, +) -> tuple: + """Same pure source/field/predicate/chronology gate for live and raw replay. + + None accept time retains legacy static validation, never full chronology. + This function performs no filesystem, PID, clock or network reads. + """ + if accepted_monotonic_ns is not None and ( + type(accepted_monotonic_ns) is not int or accepted_monotonic_ns <= 0 + ): + raise ValueError("native collector accept clock is invalid") + phases = {x["external_request_id"]: x for x in spec["phases"]} + fact = json.loads(raw) + if not isinstance(fact, dict): + raise ValueError("native raw fact is not an object") + common_fields = { + "schema", + "process", + "launch_id", + "plan_id", + "monotonic_ns", + } + mapping_fields = common_fields | { + "external_request_id", + "runtime_request_id", + } + upload_fields = common_fields | { + "observed_after_barrier_monotonic_ns", + "runtime_request_id", + "device", + "uploaded_rows", + "uploaded_rows_sha256", + "scope", + "kv_tensor_equivalence", + "attention_consumed_observed_tensor", + "full_generation_quality", + } + if set(fact) != ( + mapping_fields + if fact.get("schema") == MAPPING + else ( + set(fact) + if fact.get("schema") in {PROJECTION, PROJECTION_V2} + else upload_fields + ) + ): + raise ValueError("native fact fields do not match a supported source schema") + if set(fact.get("process", {})) != set(sender["process"]): + raise ValueError("native process fields differ from frozen Host source") + if canonical(fact) != raw: + raise ValueError("native raw bytes are noncanonical or replayed") + if any(fact.get("process", {}).get(k) != v for k, v in sender["process"].items()): + raise ValueError("native sender role/process differs from kernel registration") + if ( + fact.get("plan_id") != spec["native_plan_id"] + or fact.get("launch_id") != spec["native_launch_id"] + ): + raise ValueError("native plan/launch differs from frozen scope") + now = accepted_monotonic_ns + stamp = fact.get("monotonic_ns") + if ( + type(stamp) is not int + or stamp < spec["frozen_monotonic_ns"] + or (now is not None and (stamp > now or now - stamp > spec["max_event_age_ns"])) + ): + raise ValueError("native event is stale or from another monotonic epoch") + runtime = fact.get("runtime_request_id") + if not isinstance(runtime, str) or not runtime: + raise ValueError("native runtime request identity is absent") + if fact.get("schema") == MAPPING: + if sender["process"]["role"] != "api-server": + raise ValueError("request mapping sender is not native API") + external = fact.get("external_request_id") + phase = phases.get(external) + if ( + phase is None + or runtime in mappings + or any(x["external_request_id"] == external for x in mappings.values()) + ): + raise ValueError("actual native API mapping is unknown or repeated") + update = ("mapping", runtime) + elif fact.get("schema") in {PROJECTION, PROJECTION_V2}: + if ( + spec.get("schema") + != (KV_SPEC_V2 if fact["schema"] == PROJECTION_V2 else KV_SPEC) + or sender["process"]["role"] != "worker" + or runtime not in mappings + ): + raise ValueError( + "KV projection lacks opt-in scope/native API mapping/worker source" + ) + phase = mappings[runtime] + if stamp < mapping_stamps[runtime]: + raise ValueError("KV projection predates actual native API mapping") + projection_predicate(fact, spec["kv_requirements"][phase["phase_id"]]) + if ( + fact["schema"] == PROJECTION_V2 + and not mapping_stamps[runtime] + <= fact.get("metadata_effect_monotonic_ns", 0) + <= stamp + ): + raise ValueError("materialized projection effect/row chronology differs") + if upload_counts.get(("projection", runtime, sender["process"]["ordinal"]), 0): + raise ValueError( + "KV phase permits one actual projection per registered rank" + ) + update = ("projection", sender["process"]["ordinal"]) + elif fact.get("schema") == UPLOAD: + if sender["process"]["role"] != "worker" or runtime not in mappings: + raise ValueError("GPU upload lacks native API mapping or worker source") + phase = mappings[runtime] + if stamp < mapping_stamps[runtime]: + raise ValueError("upload predates actual native API request mapping") + rank = sender["process"]["ordinal"] + if upload_counts.get((runtime, rank), 0) >= spec["max_uploads_per_rank"]: + raise ValueError("native upload exceeds frozen finite per-rank bound") + after = fact.get("observed_after_barrier_monotonic_ns") + if type(after) is int and after <= upload_stamps.get((runtime, rank), 0): + raise ValueError("native upload barrier timestamp is replayed") + rows = fact.get("uploaded_rows") + if ( + type(after) is not int + or not stamp <= after + or (now is not None and after > now) + or not isinstance(rows, list) + or not rows + or fact.get("uploaded_rows_sha256") != sha(canonical(rows)) + or not isinstance(fact.get("device"), str) + or re.fullmatch(r"npu:[0-9]+", fact["device"]) is None + or fact.get("scope") != "NPU-block-table-prefix-readback-only" + or any( + k not in fact or fact[k] is not None + for k in ( + "kv_tensor_equivalence", + "attention_consumed_observed_tensor", + "full_generation_quality", + ) + ) + ): + raise ValueError("native uploaded-row evidence is unsupported or promoted") + slots = set() + for row in rows: + if ( + set(row) != {"group", "row", "blocks"} + or type(row["group"]) is not int + or row["group"] < 0 + or type(row["row"]) is not int + or row["row"] < 0 + or not isinstance(row["blocks"], list) + or not row["blocks"] + or any(type(x) is not int or x < 0 for x in row["blocks"]) + or (row["group"], row["row"]) in slots + ): + raise ValueError( + "native uploaded row has no exact physical block binding" + ) + slots.add((row["group"], row["row"])) + update = ("upload", rank) + else: + raise ValueError("native schema is unsupported") + return fact, phase, update + + +class Collector: + def __init__(self, spec_path: Path, journal_path: Path): + raw, _ = read_private(spec_path, 0o600) + self.spec = json.loads(raw) + fields = { + "schema", + "classification", + "formal_producer_registration", + "native_plan_id", + "native_launch_id", + "frozen_monotonic_ns", + "max_event_age_ns", + "max_uploads_per_rank", + "senders", + "phases", + } + if self.spec.get("schema") in {KV_SPEC, KV_SPEC_V2}: + fields.add("kv_requirements") + if set(self.spec) != fields: + raise ValueError("collector prereg fields are unsupported") + self.spec_sha256 = sha(raw) + self.source_sha256 = sha(Path(__file__).read_bytes()) + self.failed = False + if ( + self.spec.get("schema") not in {SPEC, KV_SPEC, KV_SPEC_V2} + or self.spec.get("classification") != "CPU_REVIEW_CANDIDATE_ONLY" + or self.spec.get("formal_producer_registration") is not False + or not self.spec.get("native_plan_id") + or not self.spec.get("native_launch_id") + or type(self.spec.get("frozen_monotonic_ns")) is not int + or type(self.spec.get("max_event_age_ns")) is not int + or self.spec["max_event_age_ns"] <= 0 + or type(self.spec.get("max_uploads_per_rank")) is not int + or not 0 < self.spec["max_uploads_per_rank"] <= 4096 + ): + raise ValueError("collector scope is incomplete or promoted") + if self.spec["schema"] in {KV_SPEC, KV_SPEC_V2}: + requirements = self.spec["kv_requirements"] + if set(requirements) != {p["phase_id"] for p in self.spec["phases"]}: + raise ValueError("KV projection phase preregistration is incomplete") + for requirement in requirements.values(): + if set(requirement) != { + "attention_group", + "semantic_anchor", + "physical_anchor", + "allocation_block_size", + "compression_threshold_tokens", + "destination_block_count", + } or any(type(v) is not int or v < 0 for v in requirement.values()): + raise ValueError("KV preregistration predicate is malformed") + if ( + not 0 + < requirement["physical_anchor"] + < requirement["compression_threshold_tokens"] + <= requirement["semantic_anchor"] + or requirement["allocation_block_size"] <= 0 + or requirement["destination_block_count"] <= 0 + or requirement["physical_anchor"] + != requirement["destination_block_count"] + * requirement["allocation_block_size"] + ): + raise ValueError("KV preregistration anchors are inconsistent") + self.senders = {x["process"]["pid"]: x for x in self.spec["senders"]} + if len(self.senders) != len(self.spec["senders"]): + raise ValueError("sender PID is repeated") + slots = set() + for sender in self.senders.values(): + if set(sender) != { + "process", + "peer_uid", + "peer_gid", + "cmdline_sha256", + "source_files", + }: + raise ValueError("sender prereg fields are unsupported") + process = sender["process"] + if set(process) != { + "host", + "role", + "ordinal", + "pid", + "start_identity", + "process_epoch", + "assignment_source", + "session", + "start_ticks", + "boot_id", + }: + raise ValueError("Host sender process fields are unsupported") + if ( + process.get("role") not in {"api-server", "worker"} + or type(process.get("ordinal")) is not int + or type(process.get("process_epoch")) is not int + or process.get("assignment_source") != "host" + ): + raise ValueError("sender role/epoch is unsupported") + slot = (process["role"], process["ordinal"]) + if slot in slots: + raise ValueError("sender role slot is repeated") + slots.add(slot) + if any(process.get(k) != v for k, v in identity(process["pid"]).items()): + raise ValueError("sender process identity changed before freeze") + if ( + type(sender["peer_uid"]) is not int + or type(sender["peer_gid"]) is not int + or sender["peer_uid"] != os.getuid() + or sender["peer_gid"] != os.getgid() + ): + raise ValueError("sender UID differs from deployment") + if sender["cmdline_sha256"] != sha( + Path(f"/proc/{process['pid']}/cmdline").read_bytes() + ): + raise ValueError("sender argv differs from freeze") + source_digests(sender["source_files"]) + if slots != {("api-server", 0), ("worker", 0), ("worker", 1)}: + raise ValueError("exact API and two worker sender set is unavailable") + self.phases = {} + client_ids = set() + nonces = set() + phase_ids = set() + for phase in self.spec["phases"]: + if set(phase) != { + "phase_id", + "client_request_id", + "challenge_nonce", + "external_request_id", + "expected_ranks", + }: + raise ValueError("challenge prereg fields are unsupported") + if ( + not all( + isinstance(phase.get(k), str) and phase[k] + for k in ( + "phase_id", + "client_request_id", + "challenge_nonce", + "external_request_id", + ) + ) + or phase.get("expected_ranks") != [0, 1] + or phase["external_request_id"] in self.phases + or phase["client_request_id"] in client_ids + or phase["challenge_nonce"] in nonces + or phase["phase_id"] in phase_ids + ): + raise ValueError( + "challenge phase/nonce/request binding is absent or repeated" + ) + self.phases[phase["external_request_id"]] = phase + client_ids.add(phase["client_request_id"]) + nonces.add(phase["challenge_nonce"]) + phase_ids.add(phase["phase_id"]) + if not self.phases: + raise ValueError("challenge set is empty") + parent = journal_path.parent.stat() + if parent.st_uid != os.getuid() or stat.S_IMODE(parent.st_mode) != 0o700: + raise ValueError("collector directory is not private") + self.fd = os.open( + journal_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600 + ) + self.archive_identity = { + "directory_device": parent.st_dev, + "directory_inode": parent.st_ino, + "journal_device": os.fstat(self.fd).st_dev, + "journal_inode": os.fstat(self.fd).st_ino, + } + self.lock = threading.Lock() + self.sequence = 0 + self.previous = "0" * 64 + self.maps = {} + self.mapping_stamps = {} + self.ranks = {} + self.upload_stamps = {} + self.upload_counts = {} + self.seen_raw = set() + self.collector_identity = identity(os.getpid()) + self._append( + { + "kind": "collector.start", + "started_monotonic_ns": time.monotonic_ns(), + "intake_predicate_schema": ( + "ecpa-native-semantic-intake/v3" + if self.spec["schema"] == KV_SPEC_V2 + else "ecpa-native-semantic-intake/v2" + ), + "spec_sha256": sha(raw), + "collector": self.collector_identity, + "collector_source_sha256": sha(Path(__file__).read_bytes()), + "archive_identity": self.archive_identity, + } + ) + directory = os.open(journal_path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + + def _append(self, value: dict) -> dict: + event = value | {"sequence": self.sequence, "previous_sha256": self.previous} + event["event_sha256"] = sha(canonical(event)) + data = memoryview(canonical(event) + b"\n") + try: + while data: + n = os.write(self.fd, data) + if n <= 0: + raise OSError("collector journal write failed") + data = data[n:] + os.fsync(self.fd) + except OSError: + self.failed = True + raise + self.sequence += 1 + self.previous = event["event_sha256"] + return event + + def accept(self, peer: tuple[int, int, int], raw: bytes) -> dict: + with self.lock: + if self.failed: + raise ValueError( + "collector persistence failed; no further ACK is possible" + ) + pid, uid, gid = peer + sender = self.senders.get(pid) + if ( + sender is None + or sender["peer_uid"] != uid + or uid != os.getuid() + or sender["peer_gid"] != gid + ): + raise ValueError("kernel peer is not a registered source sender") + actual = identity(pid) + if any(sender["process"].get(k) != v for k, v in actual.items()): + raise ValueError("kernel peer epoch/boot/session changed") + if ( + sha(Path(f"/proc/{pid}/cmdline").read_bytes()) + != sender["cmdline_sha256"] + ): + raise ValueError("kernel peer argv changed") + sources = source_digests(sender["source_files"]) + if sha(raw) in self.seen_raw: + raise ValueError("native raw bytes are replayed") + accepted_ns = time.monotonic_ns() + fact, phase, update = native_fact_predicate( + raw, + sender, + self.spec, + accepted_monotonic_ns=accepted_ns, + mappings=self.maps, + mapping_stamps=self.mapping_stamps, + upload_stamps=self.upload_stamps, + upload_counts=self.upload_counts, + ) + runtime = fact["runtime_request_id"] + stamp = fact["monotonic_ns"] + event = self._append( + { + "kind": "native.accepted", + "accepted_monotonic_ns": accepted_ns, + "raw_base64": base64.b64encode(raw).decode(), + "raw_sha256": sha(raw), + "peer": {"pid": pid, "uid": uid, "gid": gid}, + "source_sha256": sources, + "process": actual, + "phase": phase, + "runtime_request_id": runtime, + "archive_identity": self.archive_identity, + } + ) + self.seen_raw.add(sha(raw)) + if update[0] == "mapping": + self.maps[runtime] = phase + self.mapping_stamps[runtime] = stamp + elif update[0] == "projection": + self.upload_counts[("projection", runtime, update[1])] = 1 + elif update[0] == "upload": + self.ranks.setdefault(runtime, set()).add(update[1]) + self.upload_stamps[(runtime, update[1])] = fact[ + "observed_after_barrier_monotonic_ns" + ] + self.upload_counts[(runtime, update[1])] = ( + self.upload_counts.get((runtime, update[1]), 0) + 1 + ) + ack = { + "schema": ACK, + "raw_sha256": sha(raw), + "event_sha256": event["event_sha256"], + "sequence": event["sequence"], + "phase": phase, + "collector": self.collector_identity, + "collector_source_sha256": self.source_sha256, + "archive_identity": self.archive_identity, + "formal_producer_registration": False, + "classification": "CPU_REVIEW_CANDIDATE_ONLY", + } + ack["ack_sha256"] = sha(canonical(ack)) + return ack + + +class Server(socketserver.ThreadingUnixStreamServer): + daemon_threads = True + + def __init__(self, path: Path, collector: Collector): + self.collector = collector + metadata = path.parent.stat() + if ( + metadata.st_uid != os.getuid() + or stat.S_IMODE(metadata.st_mode) != 0o700 + or (metadata.st_dev, metadata.st_ino) + != ( + collector.archive_identity["directory_device"], + collector.archive_identity["directory_inode"], + ) + ): + raise ValueError("collector socket directory differs from private archive") + super().__init__(str(path), Handler) + os.chmod(path, 0o600) + + +class Handler(socketserver.StreamRequestHandler): + def handle(self): + peer = struct.unpack( + "3i", self.request.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12) + ) + try: + frame = self.rfile.readline(MAX_FRAME + 1) + if not frame.endswith(b"\n") or len(frame) > MAX_FRAME: + raise ValueError("native frame is incomplete or oversized") + request = json.loads(frame) + if request == {"op": "describe"}: + collector = self.server.collector + metadata = Path(self.server.server_address).stat() + with collector.lock: + os.fsync(collector.fd) + issued_ns = time.monotonic_ns() + response = { + "schema": "ecpa-native-semantic-collector-status/v1", + "issued_monotonic_ns": issued_ns, + "spec_sha256": collector.spec_sha256, + "collector_source_sha256": collector.source_sha256, + "collector": collector.collector_identity, + "socket_identity": [metadata.st_dev, metadata.st_ino], + "socket_path": str(self.server.server_address), + "archive_identity": collector.archive_identity, + "classification": "CPU_REVIEW_CANDIDATE_ONLY", + "formal_producer_registration": False, + "failed": collector.failed, + "acknowledged_prefix": { + "next_sequence": collector.sequence, + "last_event_sha256": collector.previous, + }, + } + self.wfile.write(canonical(response) + b"\n") + return + if set(request) != {"raw_base64"}: + raise ValueError("native frame carries unsupported fields") + raw = base64.b64decode(request["raw_base64"], validate=True) + response = self.server.collector.accept(peer, raw) + except (OSError, ValueError, KeyError, TypeError) as exc: + response = { + "schema": "ecpa-native-semantic-collector-reject/v1", + "reason": str(exc), + "formal_producer_registration": False, + } + self.wfile.write(canonical(response) + b"\n") + + +def submit_fact( + path: Path, + fact: dict, + *, + expected_collector: dict, + expected_socket_identity: tuple[int, int], + expected_collector_source_sha256: str, + expected_archive_identity: dict, +) -> dict: + metadata = path.stat() + if ( + metadata.st_dev, + metadata.st_ino, + ) != expected_socket_identity or not stat.S_ISSOCK(metadata.st_mode): + raise ValueError("native collector socket changed") + raw = canonical(fact) + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel: + channel.settimeout(5) + channel.connect(str(path)) + pid, uid, _ = struct.unpack( + "3i", channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12) + ) + if uid != os.getuid() or identity(pid) != expected_collector: + raise ValueError("ACK kernel peer is not the frozen authority process") + channel.sendall( + canonical({"raw_base64": base64.b64encode(raw).decode()}) + b"\n" + ) + response = b"" + while not response.endswith(b"\n"): + chunk = channel.recv(MAX_FRAME + 1) + if not chunk or len(response) + len(chunk) > MAX_FRAME: + raise ValueError("native ACK is incomplete or oversized") + response += chunk + ack = json.loads(response) + if ( + ack.get("schema") != ACK + or ack.get("raw_sha256") != sha(raw) + or ack.get("collector") != expected_collector + or ack.get("collector_source_sha256") != expected_collector_source_sha256 + or ack.get("archive_identity") != expected_archive_identity + or ack.get("formal_producer_registration") is not False + or ack.get("ack_sha256") + != sha(canonical({k: v for k, v in ack.items() if k != "ack_sha256"})) + ): + raise ValueError( + "native collector rejected fact or ACK digest differs: " + + str(ack.get("reason")) + ) + return ack + + +def probe_collector( + path: Path, + *, + expected_collector: dict, + expected_socket_identity: tuple[int, int], + expected_collector_source_sha256: str, + expected_spec_sha256: str, + expected_archive_identity: dict, +) -> dict: + """Read-only live owner seal; caller freezes identities from owned launch.""" + meta = path.stat() + if (meta.st_dev, meta.st_ino) != expected_socket_identity or not stat.S_ISSOCK( + meta.st_mode + ): + raise ValueError("collector status socket identity changed") + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel: + channel.settimeout(5) + channel.connect(str(path)) + pid, uid, _ = struct.unpack( + "3i", channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12) + ) + if uid != os.getuid() or identity(pid) != expected_collector: + raise ValueError("collector status peer is not the frozen owner") + channel.sendall(canonical({"op": "describe"}) + b"\n") + data = b"" + while not data.endswith(b"\n"): + chunk = channel.recv(MAX_FRAME + 1) + if not chunk or len(data) + len(chunk) > MAX_FRAME: + raise ValueError("collector status is incomplete or oversized") + data += chunk + status = json.loads(data) + if ( + status.get("schema") != "ecpa-native-semantic-collector-status/v1" + or status.get("collector") != expected_collector + or status.get("socket_identity") != list(expected_socket_identity) + or status.get("collector_source_sha256") != expected_collector_source_sha256 + or status.get("spec_sha256") != expected_spec_sha256 + or status.get("archive_identity") != expected_archive_identity + or status.get("failed") is not False + or status.get("formal_producer_registration") is not False + ): + raise ValueError("collector status failed or differs from frozen owner/source") + return status + + +def verify_journal( + spec_path: Path, journal_path: Path, *, acknowledged_prefix: dict | None = None +) -> dict: + """Read authority-owned raw chain, never a semantic/formal success summary.""" + spec_raw, _ = read_private(spec_path, 0o600) + spec = json.loads(spec_raw) + expected_senders = {x["process"]["pid"]: x for x in spec["senders"]} + expected_phases = {x["external_request_id"]: x for x in spec["phases"]} + events = [json.loads(line) for line in journal_path.read_bytes().splitlines()] + if not events or events[0].get("kind") != "collector.start": + raise ValueError("authority raw chain start is missing") + start = events[0] + if ( + start["spec_sha256"] != sha(spec_raw) + or start["collector_source_sha256"] + not in { + sha(Path(__file__).read_bytes()), + "135a3a9f06f9b2e7afe12a377ae3947a7b96801924fc09b853c836c12d65bf77", + } + or spec.get("classification") != "CPU_REVIEW_CANDIDATE_ONLY" + or spec.get("formal_producer_registration") is not False + ): + raise ValueError("authority raw chain source or frozen spec differs") + previous = "0" * 64 + seen_raw = set() + projections = {} + kv_ranks = {} + mappings = {} + mapping_stamps = {} + chronology_complete = ( + start.get("intake_predicate_schema") + == ( + "ecpa-native-semantic-intake/v3" + if spec["schema"] == KV_SPEC_V2 + else "ecpa-native-semantic-intake/v2" + ) + and type(start.get("started_monotonic_ns")) is int + ) + last_accepted_ns = start.get("started_monotonic_ns") + ranks = {} + upload_stamps = {} + counts = {} + for sequence, event in enumerate(events): + if ( + event.get("sequence") != sequence + or event.get("previous_sha256") != previous + or event.get("event_sha256") + != sha(canonical({k: v for k, v in event.items() if k != "event_sha256"})) + ): + raise ValueError("authority raw chain is broken") + previous = event["event_sha256"] + if sequence == 0: + continue + if event.get("kind") != "native.accepted": + raise ValueError("authority raw chain kind is unsupported") + raw = base64.b64decode(event["raw_base64"], validate=True) + fact = json.loads(raw) + sender = expected_senders.get(event["peer"]["pid"]) + if ( + sender is None + or event["peer"]["uid"] != sender["peer_uid"] + or event["peer"]["gid"] != sender["peer_gid"] + or fact.get("process") != sender["process"] + or any(sender["process"].get(k) != v for k, v in event["process"].items()) + or event["source_sha256"] + != {x["path"]: x["sha256"] for x in sender["source_files"]} + or event["archive_identity"] != start["archive_identity"] + or canonical(fact) != raw + or sha(raw) != event["raw_sha256"] + or sha(raw) in seen_raw + or fact.get("plan_id") != spec["native_plan_id"] + or fact.get("launch_id") != spec["native_launch_id"] + ): + raise ValueError("authority raw source/peer/request scope differs") + seen_raw.add(sha(raw)) + accepted_ns = event.get("accepted_monotonic_ns") + if accepted_ns is not None: + start_ns = start.get("started_monotonic_ns") + if ( + type(start_ns) is not int + or type(accepted_ns) is not int + or accepted_ns < start_ns + or (last_accepted_ns is not None and accepted_ns < last_accepted_ns) + ): + raise ValueError("native accept clock violates collector sequence") + last_accepted_ns = accepted_ns + else: + chronology_complete = False + fact, phase, update = native_fact_predicate( + raw, + sender, + spec, + accepted_monotonic_ns=accepted_ns, + mappings=mappings, + mapping_stamps=mapping_stamps, + upload_stamps=upload_stamps, + upload_counts=counts, + ) + runtime = fact["runtime_request_id"] + if update[0] == "mapping": + mappings[runtime] = phase + mapping_stamps[runtime] = fact["monotonic_ns"] + elif update[0] == "projection": + projections.setdefault((runtime, update[1]), []).append(fact) + counts[("projection", runtime, update[1])] = 1 + else: + rank = update[1] + if any( + projection_upload_match(p, fact) + for p in projections.get((runtime, rank), []) + ): + kv_ranks.setdefault(runtime, set()).add(rank) + ranks.setdefault(runtime, set()).add(rank) + upload_stamps[(runtime, rank)] = fact["observed_after_barrier_monotonic_ns"] + counts[(runtime, rank)] = counts.get((runtime, rank), 0) + 1 + if phase != event["phase"] or runtime != event["runtime_request_id"]: + raise ValueError("authority raw challenge join differs") + complete = sum(ranks.get(runtime) == {0, 1} for runtime in mappings) + kv_complete = sum(kv_ranks.get(runtime) == {0, 1} for runtime in mappings) + if acknowledged_prefix is not None and acknowledged_prefix != { + "next_sequence": len(events), + "last_event_sha256": previous, + }: + raise ValueError( + "authority raw tail is not the authenticated durable ACK prefix" + ) + return { + "classification": "CPU_REVIEW_CANDIDATE_ONLY", + "source_authenticated_native_envelopes": len(events) - 1, + "native_API_mappings": len(mappings), + "observed_two_rank_raw_phases": complete, + "complete_two_rank_transport_phases": complete + if acknowledged_prefix is not None and chronology_complete + else None, + "native_chronology_status": "verified" + if chronology_complete + else "unverified-missing-accept-clock", + "runtime_collector_source_sha256": start["collector_source_sha256"], + "analysis_source_sha256": sha(Path(__file__).read_bytes()), + "durable_ACK_status": "verified-authenticated-prefix" + if acknowledged_prefix is not None + else "unverified", + "expected_phase_count": len(expected_phases), + "formal_producer_registration": False, + "formal_result": False, + "complete_two_rank_KV_projection_upload_phases": kv_complete + if spec.get("schema") in {KV_SPEC, KV_SPEC_V2} + and acknowledged_prefix is not None + and chronology_complete + else None, + "KV_projection_scope": PROJECTION_SCOPE + if spec.get("schema") in {KV_SPEC, KV_SPEC_V2} + else None, + "GPU_semantic_qualification": None, + "full_objective_satisfied": None, + "journal_sha256": sha(journal_path.read_bytes()), + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--spec", required=True, type=Path) + parser.add_argument("--socket", type=Path) + parser.add_argument("--verify", action="store_true") + parser.add_argument("--journal", required=True, type=Path) + args = parser.parse_args() + if args.verify: + print(json.dumps(verify_journal(args.spec, args.journal), sort_keys=True)) + return + if args.socket is None: + parser.error("--socket is required for collector service") + collector = Collector(args.spec, args.journal) + with Server(args.socket, collector) as server: + meta = args.socket.stat() + print( + json.dumps( + { + "collector": collector.collector_identity, + "socket_identity": [meta.st_dev, meta.st_ino], + "socket_path": str(args.socket), + "collector_source_sha256": sha(Path(__file__).read_bytes()), + "spec_sha256": collector.spec_sha256, + "archive_identity": collector.archive_identity, + } + ), + flush=True, + ) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/vllm/plugins/native_tensor_artifact_bootstrap.py b/vllm/plugins/native_tensor_artifact_bootstrap.py new file mode 100644 index 000000000000..9070ad101280 --- /dev/null +++ b/vllm/plugins/native_tensor_artifact_bootstrap.py @@ -0,0 +1,192 @@ +"""Opt-in source-owned private tensor artifact bootstrap; absent means no proof.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +if __package__ and __package__.startswith("vllm."): + from . import native_semantic_collector as native + from . import native_tensor_artifact_transport as transport +else: + from experiments.exposure_gate import native_semantic_collector as native + from experiments.native_kv_tensor_candidate import artifact_transport as transport + +SCHEMA = "ecpa-selected-KV-private-artifact-bootstrap/v1" +RUNTIME_SCHEMA = "ecpa-selected-KV-private-artifact-runtime-bootstrap/v1" +ENV = "ECPA_NATIVE_TENSOR_ARTIFACT_BOOTSTRAP" + + +def callback_from_context(path, *, actual_scope=None): + path = Path(path) + raw, info = native.read_private(path, 0o600) + value = json.loads(raw) + transport.require( + set(value) + == { + "schema", + "classification", + "formal_registration", + "spec_path", + "spec_sha256", + "policy_path", + "policy_sha256", + "ready", + "source_files", + } + and value["schema"] in {SCHEMA, RUNTIME_SCHEMA} + and value["classification"] + == ( + transport.runtime_contract.CLASSIFICATION + if value["schema"] == RUNTIME_SCHEMA + else "CPU_SOURCE_CANDIDATE_ONLY" + ) + and value["formal_registration"] is False, + "tensor bootstrap scope is unsupported", + ) + spec_raw, _ = native.read_private(Path(value["spec_path"]), 0o600) + policy_raw, _ = native.read_private(Path(value["policy_path"]), 0o600) + transport.require( + native.sha(spec_raw) == value["spec_sha256"] + and native.sha(policy_raw) == value["policy_sha256"], + "tensor bootstrap frozen inputs changed", + ) + spec, policy = json.loads(spec_raw), json.loads(policy_raw) + is_runtime = policy["schema"] == transport.RUNTIME_POLICY + transport.require( + (value["schema"] == RUNTIME_SCHEMA) == is_runtime + and value["classification"] == policy["classification"], + "runtime bootstrap/policy contract mismatch", + ) + senders = {s["process"]["pid"]: s for s in spec["senders"]} + sender = senders.get(os.getpid()) + transport.require( + sender is not None + and sender["process"]["role"] == "worker" + and all( + sender["process"][k] == v for k, v in native.identity(os.getpid()).items() + ), + "tensor bootstrap current kernel worker is unregistered", + ) + own = policy["worker_directories"][str(os.getpid())] + fd, _ = transport.directory(own["path"], own["identity"]) + os.close(fd) + native.source_digests(value["source_files"]) + native.source_digests(sender["source_files"]) + if actual_scope is None: + from vllm.plugins.native_block_table_observer import _scope + + def actual_scope(): + return _scope("worker") + + if os.environ.get('ECPA_NATIVE_ATTENTION_OPERATOR_OBSERVER') == '1': + import torch + import torch_npu + from .native_block_table_observer import emit_native_fact, _scope + from .native_attention_operator_observer import register_source_candidate + register_source_candidate(lambda fact: emit_native_fact(lambda: fact), + scope=lambda: _scope('worker'), current_stream=torch.npu.current_stream, + get_format=torch_npu.get_npu_format) + inode, digest = (info.st_dev, info.st_ino), native.sha(raw) + ready = value["ready"] + + def callback(fact): + current, meta = native.read_private(path, 0o600) + transport.require( + (meta.st_dev, meta.st_ino) == inode and native.sha(current) == digest, + "tensor bootstrap context changed", + ) + native.source_digests(value["source_files"]) + native.source_digests(sender["source_files"]) + transport.require( + native.read_private(Path(value["spec_path"]), 0o600)[0] == spec_raw + and native.read_private(Path(value["policy_path"]), 0o600)[0] == policy_raw, + "tensor bootstrap preregistration changed", + ) + + def bound_scope(): + scope = actual_scope() + transport.require( + scope["process"] == sender["process"], + "actual native tensor worker role changed", + ) + return scope + + published = ( + transport.runtime_contract.from_actual_native_readback( + fact, bound_scope()["process"] + ) + if is_runtime + else fact + ) + from .native_phase_drain import acknowledged_publication + ack = acknowledged_publication( + published["binding"]["runtime_request_id"], + lambda: transport.publish_candidate( + published, + directory_path=own["path"], + directory_identity=own["identity"], + policy_sha256=value["policy_sha256"], + scope=bound_scope, + socket_path=ready["socket_path"], + runtime=is_runtime, + receiver={ + "expected_collector": ready["collector"], + "expected_socket_identity": tuple(ready["socket_identity"]), + "expected_collector_source_sha256": ready["collector_source_sha256"], + "expected_archive_identity": ready["archive_identity"], + }, + ) + ) + if is_runtime: + from .native_attention_operator_observer import remember_acknowledged_materialization + remember_acknowledged_materialization(published, ack) + return ack + + return callback + + +def ensure_registered(register): + raw = os.environ.get(ENV) + if not raw: + return False + path = Path(raw) + transport.require(path.is_absolute(), "tensor bootstrap path must be absolute") + if not path.exists(): + return False + register(callback_from_context(path)) + return True + + +def write_context(path, *, spec_path, policy_path, ready, source_files): + """Launcher prepares input only; readiness must be authenticated independently.""" + policy = json.loads(Path(policy_path).read_bytes()) + is_runtime = policy["schema"] == transport.RUNTIME_POLICY + value = { + "schema": RUNTIME_SCHEMA if is_runtime else SCHEMA, + "classification": policy["classification"], + "formal_registration": False, + "spec_path": str(spec_path), + "spec_sha256": native.sha(Path(spec_path).read_bytes()), + "policy_path": str(policy_path), + "policy_sha256": native.sha(Path(policy_path).read_bytes()), + "ready": ready, + "source_files": source_files, + } + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + try: + raw = memoryview(native.canonical(value)) + while raw: + count = os.write(fd, raw) + if count <= 0: + raise OSError("tensor bootstrap short write") + raw = raw[count:] + os.fsync(fd) + finally: + os.close(fd) + parent, _ = transport.directory(Path(path).parent) + try: + os.fsync(parent) + finally: + os.close(parent) diff --git a/vllm/plugins/native_tensor_artifact_transport.py b/vllm/plugins/native_tensor_artifact_transport.py new file mode 100644 index 000000000000..21f9d8bc379d --- /dev/null +++ b/vllm/plugins/native_tensor_artifact_transport.py @@ -0,0 +1,819 @@ +"""Private selected-KV artifact transport candidate; no formal/GPU promotion.""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import re +import stat +import time +from pathlib import Path + +if __package__ and __package__.startswith("vllm."): + from . import native_semantic_collector as native + from . import native_tensor_runtime_contract as runtime_contract + from . import native_tensor_witness as witness +else: + from experiments.exposure_gate import native_semantic_collector as native + from experiments.native_kv_tensor_candidate import runtime_contract, witness + +POLICY = "ecpa-selected-KV-private-artifact-policy/v1" +RUNTIME_POLICY = "ecpa-selected-KV-private-artifact-runtime-policy/v1" +RUNTIME_ENVELOPE = "ecpa-selected-KV-private-artifact-runtime-envelope/v1" +ENVELOPE = "ecpa-selected-KV-private-artifact-envelope/v1" +MAX_ARTIFACT = 64 * 1024 * 1024 +NAME = re.compile(r"^[a-f0-9]{64}\.json$") +PLAN_FIELDS = { + "semantic_anchor", + "physical_anchor", + "source_block_ids", + "destination_block_ids", +} + + +def require(value, reason): + if not value: + raise ValueError(reason) + + +def directory(path, expected=None): + """Traverse without symlinks; pin an owned private directory inode.""" + path = Path(path) + require(path.is_absolute(), "artifact directory must be absolute") + fd = os.open("/", os.O_RDONLY | os.O_DIRECTORY) + try: + for part in path.parts[1:]: + new = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) + os.close(fd) + fd = new + info = os.fstat(fd) + require( + info.st_uid == os.getuid() and stat.S_IMODE(info.st_mode) == 0o700, + "artifact directory is not owned and private", + ) + actual = [info.st_dev, info.st_ino] + require( + expected is None or actual == expected, "artifact directory inode changed" + ) + return fd, actual + except BaseException: + os.close(fd) + raise + + +def write_new(dir_fd, name, raw): + require( + NAME.fullmatch(name) and 0 < len(raw) <= MAX_ARTIFACT, + "artifact basename/size is unsupported", + ) + fd = os.open( + name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=dir_fd + ) + try: + data = memoryview(raw) + while data: + count = os.write(fd, data) + if count <= 0: + raise OSError("artifact short write") + data = data[count:] + os.fsync(fd) + info = os.fstat(fd) + finally: + os.close(fd) + os.fsync(dir_fd) + return { + "basename": name, + "device": info.st_dev, + "inode": info.st_ino, + "size": info.st_size, + "sha256": native.sha(raw), + "mtime_ns": info.st_mtime_ns, + "ctime_ns": info.st_ctime_ns, + } + + +def descriptor_predicate(artifact): + require( + set(artifact) + == {"basename", "device", "inode", "size", "sha256", "mtime_ns", "ctime_ns"}, + "artifact descriptor fields differ", + ) + require( + isinstance(artifact["basename"], str) + and NAME.fullmatch(artifact["basename"]) + and all( + type(artifact[k]) is int and artifact[k] >= 0 + for k in ["device", "inode", "size", "mtime_ns", "ctime_ns"] + ) + and 0 < artifact["size"] <= MAX_ARTIFACT, + "artifact descriptor is invalid", + ) + + +def read_blob(dir_fd, artifact, *, with_descriptor=False): + descriptor_predicate(artifact) + fd = os.open(artifact["basename"], os.O_RDONLY | os.O_NOFOLLOW, dir_fd=dir_fd) + try: + before = os.fstat(fd) + require( + stat.S_ISREG(before.st_mode) + and stat.S_IMODE(before.st_mode) == 0o600 + and before.st_uid == os.getuid() + and before.st_nlink == 1, + "artifact is not an owned private unaliased regular file", + ) + require( + all( + getattr(before, attr) == artifact[key] + for attr, key in [ + ("st_dev", "device"), + ("st_ino", "inode"), + ("st_size", "size"), + ("st_mtime_ns", "mtime_ns"), + ("st_ctime_ns", "ctime_ns"), + ] + ), + "artifact inode/clock/size changed", + ) + chunks, count = [], 0 + while chunk := os.read(fd, min(65536, MAX_ARTIFACT + 1 - count)): + count += len(chunk) + require(count <= MAX_ARTIFACT, "artifact grew beyond bound") + chunks.append(chunk) + raw = b"".join(chunks) + after = os.fstat(fd) + require( + all( + getattr(before, key) == getattr(after, key) + for key in [ + "st_dev", + "st_ino", + "st_mode", + "st_uid", + "st_gid", + "st_nlink", + "st_size", + "st_mtime_ns", + "st_ctime_ns", + ] + ) + and len(raw) == artifact["size"] + and native.sha(raw) == artifact["sha256"], + "artifact bytes changed", + ) + observed = { + "basename": artifact["basename"], + "device": after.st_dev, + "inode": after.st_ino, + "size": after.st_size, + "sha256": native.sha(raw), + "mtime_ns": after.st_mtime_ns, + "ctime_ns": after.st_ctime_ns, + } + return (raw, observed) if with_descriptor else raw + finally: + os.close(fd) + + +def transaction(fact): + """Read actual source plan; no owner-created phase or synthetic transaction ID.""" + binding = fact["binding"] + plan = binding.get("compression_plan") + require( + isinstance(plan, dict) and set(plan) == PLAN_FIELDS, + "actual CompressionRequest.plan is missing", + ) + require( + all( + type(plan[k]) is int and plan[k] > 0 + for k in ["semantic_anchor", "physical_anchor"] + ) + and plan["semantic_anchor"] == binding["semantic_num_tokens"] + and plan["physical_anchor"] == len(fact["keep_indices"]) + and plan["physical_anchor"] < plan["semantic_anchor"], + "actual plan semantic/physical anchors differ", + ) + factor = binding.get("cache_blocks_per_scheduler_block") + require( + type(factor) is int and factor > 0, "actual scheduler/cache expansion missing" + ) + for side in ["source", "destination"]: + blocks = plan[side + "_block_ids"] + require( + isinstance(blocks, list) + and blocks + and all(type(x) is int and x >= 0 for x in blocks) + and len(set(blocks)) == len(blocks), + "actual plan blocks are aliased", + ) + expanded = [b * factor + i for b in blocks for i in range(factor)] + require( + expanded == fact[side + "_blocks"], "actual plan/device allocation differs" + ) + return native.sha( + native.canonical( + { + "runtime_request_id": binding["runtime_request_id"], + "compression_plan": plan, + } + ) + ) + + +def verify_cpu_source(fact): + witness.verify(fact) + require( + all( + view[stage]["device"] == "cpu" and view[stage]["npu_format"] is None + for view in fact["cache_views"].values() + for stage in ["before", "after"] + ), + "CPU contract cannot publish actual NPU readback under CPU classification", + ) + + +def tensor_predicate(fact, envelope, policy, sender, maps, mapping_stamps, now, spec): + is_runtime = policy["schema"] == RUNTIME_POLICY + if is_runtime: + runtime_contract.verify(fact, sender["process"]) + else: + verify_cpu_source(fact) + binding = fact["binding"] + descriptor_predicate(envelope["artifact"]) + require( + set(envelope) + == { + "schema", + "process", + "plan_id", + "launch_id", + "monotonic_ns", + "runtime_request_id", + "transaction_id", + "artifact", + "transport_policy_sha256", + } + and envelope["schema"] == (RUNTIME_ENVELOPE if is_runtime else ENVELOPE), + "tensor envelope fields differ", + ) + runtime = binding["runtime_request_id"] + require( + sender["process"]["role"] == "worker" + and type(sender["process"]["ordinal"]) is int + and type(sender["process"]["process_epoch"]) is int + and envelope["process"] == sender["process"] + and binding["rank"] == sender["process"]["ordinal"] + and binding["tensor_parallel_size"] == 2 + and runtime == envelope["runtime_request_id"] + and runtime in maps, + "tensor artifact lacks actual worker/request/API mapping", + ) + phase = maps[runtime] + requirement = policy["requirements"][phase["phase_id"]] + require( + binding["expected_layer_indices"] == requirement["layer_indices"] + and binding["layer_index"] in requirement["layer_indices"] + and binding["compression_plan"]["semantic_anchor"] + == requirement["semantic_anchor"] + and binding["compression_plan"]["physical_anchor"] + == requirement["physical_anchor"] + and envelope["transaction_id"] == transaction(fact), + "tensor layer/actual plan transaction differs from frozen requirement", + ) + stamp = envelope["monotonic_ns"] + require( + type(stamp) is int + and mapping_stamps[runtime] + <= fact["source_before_gather_ns"] + <= fact["destination_readback_completed_ns"] + <= stamp + <= now + and stamp >= spec["frozen_monotonic_ns"] + and now - stamp <= spec["max_event_age_ns"] + and envelope["plan_id"] == spec["native_plan_id"] + and envelope["launch_id"] == spec["native_launch_id"], + "tensor actual mapping/materialization/accept chronology differs", + ) + return phase + + +def policy_predicate(policy, spec, spec_sha256): + require( + set(policy) + == { + "schema", + "classification", + "formal_registration", + "native_spec_sha256", + "worker_directories", + "requirements", + } + and policy["schema"] in {POLICY, RUNTIME_POLICY} + and policy["classification"] + == ( + runtime_contract.CLASSIFICATION + if policy["schema"] == RUNTIME_POLICY + else "CPU_SOURCE_CANDIDATE_ONLY" + ) + and policy["formal_registration"] is False + and policy["native_spec_sha256"] == spec_sha256 + and spec["classification"] == "CPU_REVIEW_CANDIDATE_ONLY" + and spec["formal_producer_registration"] is False, + "artifact policy scope/native registration differs", + ) + workers = { + str(s["process"]["pid"]): s + for s in spec["senders"] + if s["process"]["role"] == "worker" + } + require( + len(workers) == 2 + and {s["process"]["ordinal"] for s in workers.values()} == {0, 1} + and set(policy["worker_directories"]) == set(workers), + "artifact directory registration must name actual two workers", + ) + required = { + native.sha(Path(native.__file__).read_bytes()), + native.sha(Path(__file__).read_bytes()), + native.sha(Path(witness.__file__).read_bytes()), + } + if policy["schema"] == RUNTIME_POLICY: + required.add(native.sha(Path(runtime_contract.__file__).read_bytes())) + for sender in workers.values(): + require( + required <= {s["sha256"] for s in sender["source_files"]}, + "tensor worker collector/transport/predicate source closure missing", + ) + for item in policy["worker_directories"].values(): + require( + set(item) == {"path", "identity"} + and isinstance(item["path"], str) + and Path(item["path"]).is_absolute() + and isinstance(item["identity"], list) + and len(item["identity"]) == 2 + and all(type(x) is int and x >= 0 for x in item["identity"]), + "artifact directory registration fields differ", + ) + require( + set(policy["requirements"]) == {p["phase_id"] for p in spec["phases"]}, + "tensor phase preregistration missing", + ) + for req in policy["requirements"].values(): + require( + set(req) == {"layer_indices", "semantic_anchor", "physical_anchor"} + and isinstance(req["layer_indices"], list) + and req["layer_indices"] + and all(type(x) is int and x >= 0 for x in req["layer_indices"]) + and len(set(req["layer_indices"])) == len(req["layer_indices"]) + and type(req["physical_anchor"]) is int + and type(req["semantic_anchor"]) is int + and 0 < req["physical_anchor"] < req["semantic_anchor"], + "tensor requirement inventory/anchors unsupported", + ) + + +class TensorArtifactCollector(native.Collector): + """Opt-in wrapper. Existing generic collector/source/raw formats are unchanged.""" + + def __init__(self, spec_path, journal_path, policy_path): + raw, _ = native.read_private(Path(policy_path), 0o600) + self.policy = json.loads(raw) + spec_raw, _ = native.read_private(Path(spec_path), 0o600) + policy_predicate(self.policy, json.loads(spec_raw), native.sha(spec_raw)) + super().__init__(Path(spec_path), Path(journal_path)) + self.policy_sha256 = native.sha(raw) + self.artifact_dirs = {} + for pid, item in self.policy["worker_directories"].items(): + self.artifact_dirs[pid] = directory(item["path"], item["identity"])[0] + self.archive_dirfd, _ = directory(Path(journal_path).parent) + self.tensor_seen = set() + self.source_sha256 = native.sha(Path(__file__).read_bytes()) + self._append( + { + "kind": "tensor.transport.start", + "policy_sha256": self.policy_sha256, + "transport_source_sha256": self.source_sha256, + "witness_source_sha256": native.sha( + Path(witness.__file__).read_bytes() + ), + "formal_registration": False, + **( + { + "runtime_contract_source_sha256": native.sha( + Path(runtime_contract.__file__).read_bytes() + ) + } + if self.policy["schema"] == RUNTIME_POLICY + else {} + ), + } + ) + os.fsync(self.archive_dirfd) + + def accept(self, peer, raw): + envelope = json.loads(raw) + require(isinstance(envelope, dict), "native envelope is not an object") + if envelope.get("schema") not in {ENVELOPE, RUNTIME_ENVELOPE}: + return super().accept(peer, raw) + with self.lock: + require(not self.failed, "collector persistence failed") + pid, uid, gid = peer + sender = self.senders.get(pid) + require( + sender is not None + and sender["process"]["role"] == "worker" + and sender["peer_uid"] == uid == os.getuid() + and sender["peer_gid"] == gid, + "kernel peer is not an artifact worker source", + ) + require( + all( + sender["process"].get(k) == v + for k, v in native.identity(pid).items() + ) + and native.sha(Path(f"/proc/{pid}/cmdline").read_bytes()) + == sender["cmdline_sha256"], + "artifact kernel peer epoch/argv differs", + ) + sources = native.source_digests(sender["source_files"]) + require( + native.canonical(envelope) == raw + and len(raw) <= native.MAX_FRAME + and envelope["transport_policy_sha256"] == self.policy_sha256, + "artifact envelope bytes/policy differ", + ) + current_fd, _ = directory( + **{ + "path": self.policy["worker_directories"][str(pid)]["path"], + "expected": self.policy["worker_directories"][str(pid)]["identity"], + } + ) + os.close(current_fd) + payload, source_artifact = read_blob( + self.artifact_dirs[str(pid)], envelope["artifact"], with_descriptor=True + ) + fact = json.loads(payload) + require( + native.canonical(fact) == payload, "artifact payload is noncanonical" + ) + accepted = time.monotonic_ns() + phase = tensor_predicate( + fact, + envelope, + self.policy, + sender, + self.maps, + self.mapping_stamps, + accepted, + self.spec, + ) + key = ( + envelope["runtime_request_id"], + envelope["transaction_id"], + fact["binding"]["rank"], + fact["binding"]["layer_index"], + ) + require( + key not in self.tensor_seen, "tensor transaction/rank/layer is replayed" + ) + try: + archived = write_new( + self.archive_dirfd, + envelope["artifact"]["sha256"] + ".json", + payload, + ) + # Re-read archived bytes independently before publishing the ACK event. + require( + read_blob(self.archive_dirfd, archived) == payload, + "collector artifact readback differs", + ) + # Late producer mutations cannot be silently accepted during intake. + require( + read_blob(self.artifact_dirs[str(pid)], envelope["artifact"]) + == payload, + "producer artifact changed before ACK", + ) + event = self._append( + { + "kind": "tensor.artifact.accepted", + "accepted_monotonic_ns": accepted, + "raw_base64": base64.b64encode(raw).decode(), + "raw_sha256": native.sha(raw), + "artifact": archived, + "source_artifact": source_artifact, + "peer": {"pid": pid, "uid": uid, "gid": gid}, + "source_sha256": sources, + "phase": phase, + "process": native.identity(pid), + "runtime_request_id": envelope["runtime_request_id"], + "archive_identity": self.archive_identity, + } + ) + except OSError: + self.failed = True + raise + self.tensor_seen.add(key) + ack = { + "schema": native.ACK, + "raw_sha256": native.sha(raw), + "event_sha256": event["event_sha256"], + "sequence": event["sequence"], + "phase": phase, + "collector": self.collector_identity, + "collector_source_sha256": self.source_sha256, + "archive_identity": self.archive_identity, + "formal_producer_registration": False, + "classification": self.policy["classification"], + } + ack["ack_sha256"] = native.sha(native.canonical(ack)) + return ack + + +def publish_candidate( + fact, + *, + directory_path, + directory_identity, + policy_sha256, + scope, + socket_path, + receiver, + runtime=False, +): + actual_scope = scope() + if runtime: + runtime_contract.verify(fact, actual_scope["process"]) + else: + verify_cpu_source(fact) + require( + actual_scope["process"]["pid"] == os.getpid() + and actual_scope["process"]["role"] == "worker", + "artifact publisher must be the actual source worker", + ) + raw = native.canonical(fact) + fd, _ = directory(directory_path, directory_identity) + try: + artifact = write_new(fd, native.sha(raw) + ".json", raw) + finally: + os.close(fd) + envelope = { + "schema": RUNTIME_ENVELOPE if runtime else ENVELOPE, + **actual_scope, + "monotonic_ns": time.monotonic_ns(), + "runtime_request_id": fact["binding"]["runtime_request_id"], + "transaction_id": transaction(fact), + "artifact": artifact, + "transport_policy_sha256": policy_sha256, + } + # Only this bounded envelope enters the existing framed Unix socket. + require( + len( + native.canonical( + {"raw_base64": base64.b64encode(native.canonical(envelope)).decode()} + ) + ) + + 1 + <= native.MAX_FRAME, + "tensor envelope exceeds native socket bound", + ) + return native.submit_fact(Path(socket_path), envelope, **receiver) + + +def verify_archive(spec_path, policy_path, journal_path, *, acknowledged_prefix=None): + """Portable bytes oracle. An ACK prefix must come from authenticated live status. + + This function reads no current native PIDs or mutable sender paths. Archived + file inodes are not portable; the original identities remain recorded, while + the archived payload's exact bytes/digest are independently checked. + """ + spec_raw = Path(spec_path).read_bytes() + policy_raw = Path(policy_path).read_bytes() + spec, policy = json.loads(spec_raw), json.loads(policy_raw) + require( + policy["schema"] in {POLICY, RUNTIME_POLICY} + and policy["classification"] + in {"CPU_SOURCE_CANDIDATE_ONLY", runtime_contract.CLASSIFICATION} + and policy["formal_registration"] is False + and policy["native_spec_sha256"] == native.sha(spec_raw), + "portable artifact policy scope/source differs", + ) + policy_predicate(policy, spec, native.sha(spec_raw)) + senders = {s["process"]["pid"]: s for s in spec["senders"]} + events = [json.loads(line) for line in Path(journal_path).read_bytes().splitlines()] + require( + len(events) >= 2 + and events[0]["kind"] == "collector.start" + and events[0]["spec_sha256"] == native.sha(spec_raw) + and events[0]["collector_source_sha256"] + == native.sha(Path(native.__file__).read_bytes()) + and events[1]["kind"] == "tensor.transport.start" + and events[1]["policy_sha256"] == native.sha(policy_raw) + and events[1]["transport_source_sha256"] + == native.sha(Path(__file__).read_bytes()) + and events[1]["witness_source_sha256"] + == native.sha(Path(witness.__file__).read_bytes()), + "portable collector/transport/source start differs", + ) + if policy["schema"] == RUNTIME_POLICY: + require( + events[1].get("runtime_contract_source_sha256") + == native.sha(Path(runtime_contract.__file__).read_bytes()), + "runtime contract source start differs", + ) + maps, mapping_stamps, upload_stamps, upload_counts = {}, {}, {}, {} + seen, tensor_seen, completed = set(), set(), {} + previous = "0" * 64 + last_clock = events[0]["started_monotonic_ns"] + for sequence, event in enumerate(events): + require( + event["sequence"] == sequence + and event["previous_sha256"] == previous + and event["event_sha256"] + == native.sha( + native.canonical( + {k: v for k, v in event.items() if k != "event_sha256"} + ) + ), + "portable authority chain prefix differs", + ) + previous = event["event_sha256"] + if sequence < 2: + continue + require( + event["kind"] in {"native.accepted", "tensor.artifact.accepted"}, + "portable authority event kind unsupported", + ) + raw = base64.b64decode(event["raw_base64"], validate=True) + digest = native.sha(raw) + require( + digest == event["raw_sha256"] and digest not in seen, + "portable native bytes are changed or replayed", + ) + seen.add(digest) + sender = senders.get(event["peer"]["pid"]) + require( + sender is not None + and event["peer"]["uid"] == sender["peer_uid"] + and event["peer"]["gid"] == sender["peer_gid"] + and event["source_sha256"] + == {s["path"]: s["sha256"] for s in sender["source_files"]} + and event["process"] + == { + k: sender["process"][k] + for k in ["pid", "start_ticks", "start_identity", "boot_id", "session"] + } + and event["archive_identity"] == events[0]["archive_identity"], + "portable sender source/kernel binding differs", + ) + accepted = event["accepted_monotonic_ns"] + require( + type(accepted) is int and accepted >= last_clock, + "portable collector accept chronology differs", + ) + last_clock = accepted + if event["kind"] == "native.accepted": + fact, phase, update = native.native_fact_predicate( + raw, + sender, + spec, + accepted_monotonic_ns=accepted, + mappings=maps, + mapping_stamps=mapping_stamps, + upload_stamps=upload_stamps, + upload_counts=upload_counts, + ) + runtime = fact["runtime_request_id"] + if update[0] == "mapping": + maps[runtime], mapping_stamps[runtime] = phase, fact["monotonic_ns"] + elif update[0] == "upload": + upload_stamps[(runtime, update[1])] = fact[ + "observed_after_barrier_monotonic_ns" + ] + upload_counts[(runtime, update[1])] = ( + upload_counts.get((runtime, update[1]), 0) + 1 + ) + else: + upload_counts[("projection", runtime, update[1])] = 1 + else: + envelope = json.loads(raw) + require( + native.canonical(envelope) == raw + and envelope["transport_policy_sha256"] == native.sha(policy_raw), + "portable tensor envelope policy differs", + ) + artifact = event["artifact"] + descriptor_predicate(artifact) + require( + event["source_artifact"] == envelope["artifact"], + "portable observed source artifact inode differs", + ) + require( + NAME.fullmatch(artifact["basename"]) + and artifact["sha256"] == envelope["artifact"]["sha256"], + "portable artifact digest descriptor differs", + ) + path = Path(journal_path).parent / artifact["basename"] + require( + not path.is_symlink() and path.is_file(), + "portable artifact path is not regular", + ) + payload = path.read_bytes() + require( + 0 < len(payload) <= MAX_ARTIFACT + and len(payload) == artifact["size"] == envelope["artifact"]["size"] + and native.sha(payload) == artifact["sha256"], + "portable archived blob changed", + ) + fact = json.loads(payload) + require( + native.canonical(fact) == payload, + "portable tensor payload is noncanonical", + ) + phase = tensor_predicate( + fact, envelope, policy, sender, maps, mapping_stamps, accepted, spec + ) + key = ( + fact["binding"]["runtime_request_id"], + envelope["transaction_id"], + fact["binding"]["rank"], + fact["binding"]["layer_index"], + ) + require(key not in tensor_seen, "portable tensor rank/layer replay") + tensor_seen.add(key) + completed.setdefault(key[:2], set()).add(key[2:]) + require( + event["phase"] == phase + and event["runtime_request_id"] == fact["runtime_request_id"] + if event["kind"] == "native.accepted" + else event["phase"] == phase + and event["runtime_request_id"] == fact["binding"]["runtime_request_id"], + "portable phase/runtime join differs", + ) + if acknowledged_prefix is not None: + require( + acknowledged_prefix + == {"next_sequence": len(events), "last_event_sha256": previous}, + "authenticated ACK prefix differs from archived chain", + ) + complete = [] + for (runtime, tx), observed in completed.items(): + phase = maps[runtime] + expected = { + (rank, layer) + for rank in phase["expected_ranks"] + for layer in policy["requirements"][phase["phase_id"]]["layer_indices"] + } + if observed == expected and acknowledged_prefix is not None: + complete.append( + { + "phase_id": phase["phase_id"], + "runtime_request_id": runtime, + "transaction_id": tx, + } + ) + return { + "schema": "ecpa-selected-KV-artifact-portable-analysis/v1", + "classification": policy["classification"], + "complete_transactions": complete, + "durable_artifact_join_complete": bool(complete), + "acknowledged_prefix_status": "verified" + if acknowledged_prefix is not None + else "unknown", + "GPU_qualification": None, + "attention_consumption": None, + "full_model_quality": None, + "formal_registration": False, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--spec", type=Path, required=True) + parser.add_argument("--policy", type=Path, required=True) + parser.add_argument("--journal", type=Path, required=True) + parser.add_argument("--socket", type=Path, required=True) + args = parser.parse_args() + collector = TensorArtifactCollector(args.spec, args.journal, args.policy) + with native.Server(args.socket, collector) as server: + metadata = args.socket.stat() + print( + json.dumps( + { + "collector": collector.collector_identity, + "socket_identity": [metadata.st_dev, metadata.st_ino], + "socket_path": str(args.socket), + "collector_source_sha256": collector.source_sha256, + "spec_sha256": collector.spec_sha256, + "policy_sha256": collector.policy_sha256, + "archive_identity": collector.archive_identity, + } + ), + flush=True, + ) + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/vllm/plugins/native_tensor_runtime_contract.py b/vllm/plugins/native_tensor_runtime_contract.py new file mode 100644 index 000000000000..6939224e462a --- /dev/null +++ b/vllm/plugins/native_tensor_runtime_contract.py @@ -0,0 +1,72 @@ +"""Explicit real-runtime-unqualified contract; source bytes are not qualification.""" + +from __future__ import annotations + +import re + +if __package__ and __package__.startswith("vllm."): + from . import native_tensor_witness as witness +else: + from experiments.native_kv_tensor_candidate import witness + +SCHEMA = "ecpa-selected-KV-materialization-runtime-unqualified/v1" +CLASSIFICATION = "REAL_RUNTIME_UNQUALIFIED" + + +def verify(fact, process): + witness.check( + fact.get("schema") == SCHEMA + and fact.get("classification") == CLASSIFICATION + and fact.get("native_producer") == process, + "real runtime tensor schema/source role missing", + ) + witness.check( + process.get("role") == "worker" + and process.get("assignment_source") == "host" + and type(process.get("pid")) is int + and process["pid"] > 0 + and type(process.get("ordinal")) is int + and process["ordinal"] == fact["binding"]["rank"] + and type(process.get("start_ticks")) is int + and process["start_ticks"] > 0 + and type(process.get("session")) is int + and process["session"] > 0 + and isinstance(process.get("boot_id"), str) + and process["boot_id"], + "actual native runtime worker identity missing", + ) + devices = set() + for view in fact["cache_views"].values(): + for stage in ["before", "after"]: + actual = view[stage] + witness.check( + isinstance(actual["device"], str) + and re.fullmatch(r"npu:[0-9]+", actual["device"]) + and type(actual["npu_format"]) is int + and actual["npu_format"] >= 0, + "real runtime contract rejects CPU or unobserved NPU cache format", + ) + devices.add(actual["device"]) + witness.check(len(devices) == 1, "actual K/V readbacks span different devices") + # Explicitly reuse only the original byte/layout/slot predicate. This local + # adapter is not an old CPU archive, an admission, or a hardware result. + selected = {k: v for k, v in fact.items() if k != "native_producer"} + selected["schema"] = witness.SCHEMA + selected["classification"] = "CPU_SOURCE_CANDIDATE_ONLY" + result = witness.verify(selected) + return result | { + "classification": CLASSIFICATION, + "actual_device": next(iter(devices)), + } + + +def from_actual_native_readback(fact, process): + """Source callback encodes fresh actual NPU readback; CPU cannot be upgraded.""" + witness.verify(fact) + encoded = fact | { + "schema": SCHEMA, + "classification": CLASSIFICATION, + "native_producer": dict(process), + } + verify(encoded, process) + return encoded diff --git a/vllm/plugins/native_tensor_witness.py b/vllm/plugins/native_tensor_witness.py new file mode 100644 index 000000000000..48d6104ab5ee --- /dev/null +++ b/vllm/plugins/native_tensor_witness.py @@ -0,0 +1,379 @@ +"""Selected-source byte witness candidate. No source registration or GPU result.""" + +from __future__ import annotations + +import base64 +import hashlib +import time + +SCHEMA = "ecpa-selected-KV-materialization-source-candidate/v1" +_sink = None + + +def register_candidate_sink(callback): + """CPU/source candidate only: callback return is never a durable/formal ACK.""" + global _sink + if _sink is not None or not callable(callback): + raise ValueError("candidate sink must be a single callable") + _sink = callback + + +def enabled(): + return _sink is not None + + +def check(value, reason): + if not value: + raise ValueError(reason) + + +def digest(raw): + return hashlib.sha256(raw).hexdigest() + + +def slots(blocks, indices, block_size): + check(type(block_size) is int and block_size > 0, "invalid block size") + check( + blocks + and all(type(x) is int and x >= 0 for x in blocks) + and len(set(blocks)) == len(blocks), + "invalid or aliased blocks", + ) + check( + indices + and all(type(x) is int and 0 <= x < len(blocks) * block_size for x in indices), + "invalid token indices", + ) + return [blocks[x // block_size] * block_size + x % block_size for x in indices] + + +def verify(fact): + check( + fact["schema"] == SCHEMA + and fact["classification"] == "CPU_SOURCE_CANDIDATE_ONLY" + and fact["formal_registration"] is False + and fact["GPU_qualification"] is None + and fact["attention_consumption"] is None + and fact["full_model_quality"] is None, + "candidate scope cannot become GPU/attention/formal evidence", + ) + binding = fact["binding"] + check( + isinstance(binding["runtime_request_id"], str) + and binding["runtime_request_id"] + and isinstance(binding["layer_name"], str) + and binding["layer_name"] + and type(binding["layer_index"]) is int + and binding["layer_index"] >= 0 + and type(binding["tensor_parallel_size"]) is int + and binding["tensor_parallel_size"] > 0 + and type(binding["rank"]) is int + and 0 <= binding["rank"] < binding["tensor_parallel_size"] + and isinstance(binding["expected_layer_indices"], list) + and binding["expected_layer_indices"] + and all(type(x) is int and x >= 0 for x in binding["expected_layer_indices"]) + and binding["layer_index"] in binding["expected_layer_indices"] + and len(set(binding["expected_layer_indices"])) + == len(binding["expected_layer_indices"]), + "actual request/layer binding missing", + ) + keep = fact["keep_indices"] + check( + type(binding["physical_num_tokens"]) is int + and binding["physical_num_tokens"] > 0 + and type(binding["semantic_num_tokens"]) is int + and binding["semantic_num_tokens"] >= binding["physical_num_tokens"], + "actual physical/semantic context missing", + ) + check( + keep + and all( + type(x) is int and 0 <= x < binding["physical_num_tokens"] for x in keep + ) + and keep == sorted(set(keep)) + and fact["source_slots"] + == slots(fact["source_blocks"], keep, fact["block_size"]) + and fact["destination_slots"] + == slots(fact["destination_blocks"], list(range(len(keep))), fact["block_size"]) + and not set(fact["source_blocks"]) & set(fact["destination_blocks"]), + "actual keep/source/private-destination slot mapping differs", + ) + clocks = [ + fact[k] + for k in [ + "source_before_gather_ns", + "reference_after_gather_ns", + "write_completed_barrier_ns", + "destination_readback_completed_ns", + ] + ] + check( + all(type(x) is int and x > 0 for x in clocks) and clocks == sorted(clocks), + "actual byte/barrier chronology differs", + ) + for kind in ["K", "V"]: + view = fact["cache_views"][kind] + before, after = view["before"], view["after"] + check( + before == after + and before["dtype"] in {"torch.float16", "torch.bfloat16"} + and len(before["shape"]) == len(before["stride"]) == 4 + and before["shape"][1] == fact["block_size"] + and before["stride"][1:] + == [before["shape"][2] * before["shape"][3], before["shape"][3], 1] + and before["stride"][0] >= fact["block_size"] * before["stride"][1] + and all( + type(before[k]) is int and before[k] > 0 + for k in ["data_ptr", "storage_ptr"] + ) + and type(before["storage_offset"]) is int + and before["storage_offset"] >= 0, + "actual cache dtype/layout/pointer binding differs", + ) + size = len(keep) * before["shape"][2] * before["shape"][3] * 2 + check(0 < size <= 4_194_304, "bounded selected byte payload required") + check( + all(type(x) is int and x > 0 for x in before["shape"] + before["stride"]) + and max(fact["source_blocks"] + fact["destination_blocks"]) + < before["shape"][0] + and ( + (before["device"] == "cpu" and before["npu_format"] is None) + or ( + before["device"].startswith("npu:") + and type(before["npu_format"]) is int + and before["npu_format"] >= 0 + ) + ), + "actual cache extent/device/format binding differs", + ) + token_bytes = before["shape"][2] * before["shape"][3] * 2 + block_bytes = { + int(block): base64.b64decode(value, validate=True) + for block, value in fact["exported_source_block_bytes"][kind].items() + } + needed = {fact["source_blocks"][i // fact["block_size"]] for i in keep} + source_after = { + int(block): base64.b64decode(value, validate=True) + for block, value in fact["source_after_write_block_bytes"][kind].items() + } + check( + source_after == block_bytes, + "actual source cache changed during private-destination write", + ) + check( + set(block_bytes) == needed + and all( + len(value) == fact["block_size"] * token_bytes + for value in block_bytes.values() + ), + "bounded actual source block exports differ", + ) + independent = b"".join( + block_bytes[fact["source_blocks"][i // fact["block_size"]]][ + (i % fact["block_size"]) * token_bytes : (i % fact["block_size"] + 1) + * token_bytes + ] + for i in keep + ) + source = base64.b64decode(fact["selected_source_bytes"][kind], validate=True) + destination = base64.b64decode( + fact["destination_readback_bytes"][kind], validate=True + ) + check( + len(source) == len(destination) == size + and source == destination == independent + and source + == base64.b64decode(fact["gathered_workspace_bytes"][kind], validate=True) + and digest(source) == fact["selected_source_sha256"][kind] + and digest(destination) == fact["destination_readback_sha256"][kind], + "actual selected-source/destination byte predicate differs", + ) + return { + "selected_KV_bytes_match": True, + "formal_result": False, + "GPU_qualification": None, + "attention_consumption": None, + "full_model_quality": None, + } + + +def _sync(tensor): + import torch + + if tensor.device.type == "npu": + torch.npu.current_stream(tensor.device).synchronize() + elif tensor.device.type != "cpu": + raise ValueError("candidate supports only explicit CPU/NPU stream") + return time.monotonic_ns() + + +def _raw(tensor): + import torch + + # Reinterpret, never round through a float/JSON value. Preserves bf16/f16 bits. + return bytes( + tensor.detach().cpu().contiguous().view(torch.uint8).reshape(-1).tolist() + ) + + +def _view(tensor): + native_format = None + if tensor.device.type == "npu": + import torch_npu + + native_format = int(torch_npu.get_npu_format(tensor)) + return { + "npu_format": native_format, + "device": str(tensor.device), + "dtype": str(tensor.dtype), + "shape": list(tensor.shape), + "stride": list(tensor.stride()), + "data_ptr": tensor.data_ptr(), + "storage_ptr": tensor.untyped_storage().data_ptr(), + "storage_offset": tensor.storage_offset(), + } + + +def capture_before_actual_gather( + k_cache, v_cache, binding, source_slots, destination_slots +): + if not enabled(): + return None + _sync(k_cache) + keep = binding["keep_indices"].detach().cpu().tolist() + source_blocks = list(binding["source_blocks"]) + block_size = k_cache.shape[1] + expected_slots = slots(source_blocks, keep, block_size) + check( + source_slots.detach().cpu().tolist() == expected_slots, + "actual device source slots differ from source blocks/keep indices", + ) + check( + keep[-1] < binding["physical_num_tokens"], + "selected position exceeds actual physical context", + ) + selected = {} + exports = {} + for kind, cache in [("K", k_cache), ("V", v_cache)]: + token_bytes = cache.shape[2] * cache.shape[3] * 2 + # Export only needed actual physical blocks, before workspace gathers. + needed = {source_blocks[index // block_size] for index in keep} + blocks = {block: _raw(cache[block]) for block in needed} + exports[kind] = blocks + selected[kind] = b"".join( + blocks[source_blocks[index // block_size]][ + (index % block_size) * token_bytes : (index % block_size + 1) + * token_bytes + ] + for index in keep + ) + return { + "binding": {k: v for k, v in binding.items() if k != "keep_indices"}, + "keep_indices": keep, + "source_slots": expected_slots, + "destination_slots": destination_slots.detach().cpu().tolist(), + "cache_before": {"K": _view(k_cache), "V": _view(v_cache)}, + "selected_source_raw": selected, + "source_block_exports": exports, + "source_before_gather_ns": time.monotonic_ns(), + } + + +def observe_actual_workspace(reference, k_workspace, v_workspace): + if reference is None: + return + _sync(k_workspace) + observed = {"K": _raw(k_workspace), "V": _raw(v_workspace)} + check( + observed == reference["selected_source_raw"], + "actual gather workspace differs from independent source-block reference", + ) + reference["workspace_raw"] = observed + reference["reference_after_gather_ns"] = time.monotonic_ns() + + +def observe_after_actual_write(reference, k_cache, v_cache, destination_slots, gather): + if reference is None: + return + barrier = _sync(k_cache) + # Real destination reads after real writes; independent source fixed before gather. + del gather # Never reuse the plugin gather primitive as the destination oracle. + actual_slots = destination_slots.detach().cpu().tolist() + raw = {} + source_after = {} + for kind, cache in [("K", k_cache), ("V", v_cache)]: + source_after[kind] = { + block: _raw(cache[block]) + for block in reference["source_block_exports"][kind] + } + block_size = cache.shape[1] + token_bytes = cache.shape[2] * cache.shape[3] * 2 + actual_blocks = {slot // block_size for slot in actual_slots} + blocks = {block: _raw(cache[block]) for block in actual_blocks} + raw[kind] = b"".join( + blocks[slot // block_size][ + (slot % block_size) * token_bytes : (slot % block_size + 1) + * token_bytes + ] + for slot in actual_slots + ) + _sync(k_cache) + binding = dict(reference["binding"]) + fact = { + "schema": SCHEMA, + "classification": "CPU_SOURCE_CANDIDATE_ONLY", + "formal_registration": False, + "GPU_qualification": None, + "attention_consumption": None, + "full_model_quality": None, + "binding": binding, + "keep_indices": reference["keep_indices"], + "source_blocks": binding.pop("source_blocks"), + "destination_blocks": binding.pop("destination_blocks"), + "block_size": binding.pop("block_size"), + "source_slots": reference["source_slots"], + "destination_slots": reference["destination_slots"], + "source_before_gather_ns": reference["source_before_gather_ns"], + "reference_after_gather_ns": reference["reference_after_gather_ns"], + "write_completed_barrier_ns": barrier, + "destination_readback_completed_ns": time.monotonic_ns(), + "cache_views": { + "K": {"before": reference["cache_before"]["K"], "after": _view(k_cache)}, + "V": {"before": reference["cache_before"]["V"], "after": _view(v_cache)}, + }, + "source_after_write_block_bytes": { + kind: { + str(block): base64.b64encode(value).decode() + for block, value in blocks.items() + } + for kind, blocks in source_after.items() + }, + "exported_source_block_bytes": { + kind: { + str(block): base64.b64encode(value).decode() + for block, value in blocks.items() + } + for kind, blocks in reference["source_block_exports"].items() + }, + "selected_source_bytes": { + kind: base64.b64encode(value).decode() + for kind, value in reference["selected_source_raw"].items() + }, + "gathered_workspace_bytes": { + kind: base64.b64encode(value).decode() + for kind, value in reference["workspace_raw"].items() + }, + "destination_readback_bytes": { + kind: base64.b64encode(value).decode() for kind, value in raw.items() + }, + "selected_source_sha256": { + kind: digest(value) + for kind, value in reference["selected_source_raw"].items() + }, + "destination_readback_sha256": { + kind: digest(value) for kind, value in raw.items() + }, + } + verify(fact) + # No callback Boolean or local PublicationAck becomes an authenticated ACK. + _sink(fact) diff --git a/vllm/v1/engine/input_processor.py b/vllm/v1/engine/input_processor.py index 17e8af5caeef..896812c056aa 100644 --- a/vllm/v1/engine/input_processor.py +++ b/vllm/v1/engine/input_processor.py @@ -370,6 +370,10 @@ def assign_request_id(request: EngineCoreRequest): else: request.request_id = f"{request.external_req_id}-{random_uuid():.8}" + from vllm.plugins.native_block_table_observer import observe_request_id_mapping + + observe_request_id_mapping(request) + def process_inputs( self, request_id: str, diff --git a/vllm/v1/worker/gpu_model_runner.py b/vllm/v1/worker/gpu_model_runner.py index cf67d45df857..2d9111a43fe3 100644 --- a/vllm/v1/worker/gpu_model_runner.py +++ b/vllm/v1/worker/gpu_model_runner.py @@ -1206,6 +1206,18 @@ def _update_states(self, scheduler_output: "SchedulerOutput") -> Callable | None for req_id in scheduler_output.finished_req_ids: self.input_batch.remove_request(req_id) + # Opt-in source-owned drain follows actual native removal. + from vllm.plugins.native_phase_drain import ( + after_finished_request_removal, + ) + from vllm.plugins.native_block_table_observer import ( + _scope, emit_native_fact, + ) + after_finished_request_removal( + self, scheduler_output, scope=_scope, + publish=lambda fact: emit_native_fact(lambda: fact), + ) + # Zero GPU memory for freshly allocated cache blocks to prevent # stale NaN/data from corrupting attention or SSM computation. if scheduler_output.new_block_ids_to_zero: