def build_window_cache(
*,
requests_jsonl: Path | str | bytes,
cache_dir: Path | str,
evidence_dir: Path | str,
encoder: object,
encoder_id: str,
batch_size: int,
rows_per_shard: int,
created_at_ns: int,
hardware: str,
resolved_config: Mapping[str, object],
encoder_runtime_identity: Mapping[str, object],
input_artifacts: Mapping[str, Path | str | bytes] | None = None,
logger: GenoLeWMLogger | None = None,
) -> CacheBuildReport:
"""Build or resume the cache for one exact request JSONL artifact.
``encoder`` must expose the raw-state ``CarbonStateEncoder`` contract:
``pooling_identity``, ``encode_batch``, an exact 32-byte ``encoder_hash``,
and ``normalize is False``. Existing planned shards are decoded, hashed,
compared row-for-row with the plan, and re-indexed before any missing shard
is encoded.
"""
started = time.perf_counter()
batch_size = _positive_int("batch_size", batch_size)
rows_per_shard = _positive_int("rows_per_shard", rows_per_shard)
created_at_ns = _non_negative_int("created_at_ns", created_at_ns)
if created_at_ns == 0:
raise InputError("created_at_ns must be fixed to a positive UTC nanosecond value")
if type(encoder_id) is not str or not encoder_id:
raise InputError("encoder_id must be non-empty text")
hardware = _text(hardware, field="hardware")
resolved_config_payload = _json_object_copy(resolved_config, field="resolved_config")
resolved_config_bytes = _pretty_json_bytes(resolved_config_payload)
contract = _encoder_contract(encoder)
encoder_device = _text(getattr(encoder, "device", None), field="encoder.device")
runtime_identity_payload = _encoder_runtime_identity_payload(
encoder_runtime_identity,
contract=contract,
encoder_id=encoder_id,
resolved_config=resolved_config_payload,
)
runtime_identity_bytes = _pretty_json_bytes(runtime_identity_payload)
request_bytes = (
bytes(requests_jsonl)
if isinstance(requests_jsonl, bytes)
else _read_regular_bytes(Path(requests_jsonl), label="cache build requests")
)
requests = _parse_requests(request_bytes)
request_identity = {
"sha256": sha256_bytes(request_bytes),
"size_bytes": len(request_bytes),
}
cache_root = Path(cache_dir).absolute()
evidence_root = Path(evidence_dir).absolute()
evidence = _EvidenceStore(evidence_root)
staged_inputs = _read_input_artifacts(input_artifacts or {})
input_identities = tuple(item.identity for item in staged_inputs)
report_path = evidence_root / CACHE_BUILD_REPORT_NAME
checksums_path = evidence_root / _CHECKSUMS_NAME
resolved_config_identity = {
"path": _RESOLVED_CONFIG_NAME,
"sha256": sha256_bytes(resolved_config_bytes),
"size_bytes": len(resolved_config_bytes),
}
runtime_identity = {
"path": _RUNTIME_IDENTITY_NAME,
"sha256": sha256_bytes(runtime_identity_bytes),
"size_bytes": len(runtime_identity_bytes),
}
expected_evidence_names = _expected_evidence_names(input_identities)
_assert_evidence_inventory(
evidence,
expected_names=expected_evidence_names,
require_complete=False,
)
expected_plan = _create_plan(
requests=requests,
request_identity=request_identity,
cache_root=cache_root,
encoder=encoder,
encoder_id=encoder_id,
contract=contract,
batch_size=batch_size,
rows_per_shard=rows_per_shard,
created_at_ns=created_at_ns,
hardware=hardware,
encoder_device=encoder_device,
resolved_config=resolved_config_identity,
encoder_runtime_identity=runtime_identity,
input_artifacts=input_identities,
)
if evidence.exists(_PLAN_NAME):
plan_payload = _read_json_object(evidence, _PLAN_NAME, label="cache build plan")
plan = _load_plan(
plan_payload,
expected=expected_plan,
)
else:
plan = expected_plan
_write_once(
evidence, _PLAN_NAME, _pretty_json_bytes(plan.payload), label="cache build plan"
)
# The immutable plan is validated or installed before any caller-provided
# artifact is staged. A failed invocation therefore cannot seed files that
# a later, differently configured invocation would accidentally close.
_write_once(evidence, _REQUEST_COPY_NAME, request_bytes, label="cache build request copy")
_write_once(
evidence,
_RESOLVED_CONFIG_NAME,
resolved_config_bytes,
label="cache build resolved config",
)
_write_once(
evidence,
_RUNTIME_IDENTITY_NAME,
runtime_identity_bytes,
label="cache build encoder runtime identity",
)
_stage_input_artifacts(evidence, staged_inputs)
plan_sha256 = evidence.capture(_PLAN_NAME, label="cache build plan").sha256
state = _load_or_initialize_state(evidence, plan_sha256=plan_sha256)
completed = _completed_by_id(state, plan=plan)
adopted_or_changed = False
# Resume preflight: verify every evidence-owned shard and repair its index
# rows before resolving shared-cache hits or calling encode_batch.
for shard in plan.shards:
absolute_path = cache_root / shard.relative_path
prior = completed.get(shard.shard_id)
if prior is None and not os.path.lexists(absolute_path):
continue
if prior is not None and not os.path.lexists(absolute_path):
raise CacheCorruptError(
"cache build state names a completed shard that is missing",
details={"shard_id": shard.shard_id, "path": shard.relative_path},
)
inspection = inspect_cache_shard(cache_root, shard.relative_path)
execution_shard = _execution_shard_from_records(shard, inspection.records)
_assert_inspection_matches_plan(
inspection,
shard=execution_shard,
created_at_ns=created_at_ns,
)
if prior is not None:
_assert_state_identity(
prior,
inspection,
shard=execution_shard,
plan_shard=shard,
)
else:
completed[shard.shard_id] = _completed_entry(
execution_shard,
inspection,
plan_shard_id=shard.shard_id,
origin="adopted_after_interruption",
encoded_rows=0,
encode_batch_calls=0,
encode_batch_seconds=None,
)
adopted_or_changed = True
# Re-publishing observed rows is a no-op on the immutable shard and
# makes a missing/stale SQLite index converge without model work.
write_shard(
cache_root,
encoder_id=plan.namespace,
contig=shard.contig,
stride_block=shard.stride_block,
records=inspection.records,
)
if adopted_or_changed or not evidence.exists(_STATE_NAME):
state = _state_payload(plan_sha256=plan_sha256, completed=completed)
_atomic_write(evidence, _STATE_NAME, _pretty_json_bytes(state))
resolved_before = _resolve_plan_cache(cache_root, plan, require_all=False)
evidence_owned_keys = _completed_row_keys(completed)
resumed_rows = sum(key in evidence_owned_keys for key in resolved_before.rows)
if evidence.exists(_CHECKSUMS_NAME):
return _verify_completed_bundle(
evidence=evidence,
evidence_root=evidence_root,
cache_root=cache_root,
plan=plan,
plan_sha256=plan_sha256,
expected_evidence_names=expected_evidence_names,
)
if logger is not None:
logger.info(
"data.cache.build.start",
request_count=len(requests),
unique_rows=sum(len(shard.rows) for shard in plan.shards),
planned_shards=len(plan.shards),
requests_sha256=request_identity["sha256"],
)
encoded_rows = 0
encoded_shards = 0
for shard in plan.shards:
missing_rows = tuple(row for row in shard.rows if row.key not in resolved_before.rows)
if not missing_rows:
_emit_progress(
logger,
shard=shard,
completed_shards=_resolved_plan_shard_count(plan, resolved_before.rows),
total_shards=len(plan.shards),
encoded_rows=encoded_rows,
resumed_rows=resumed_rows,
throughput_per_s=None,
status="resumed",
)
continue
execution_shard = _execution_shard(shard, rows=missing_rows)
encoded = _encode_shard(
execution_shard,
encoder=encoder,
contract=contract,
batch_size=batch_size,
created_at_ns=created_at_ns,
)
try:
path = write_shard(
cache_root,
encoder_id=plan.namespace,
contig=execution_shard.contig,
stride_block=execution_shard.stride_block,
records=encoded.records,
)
except CacheKeyAlreadyIndexedError as race_exc:
# A concurrent builder may have published the same logical misses
# after our preflight but before the serialized write reservation.
# Accept only the precise key-reservation race. Any path that now
# exists in this evidence namespace must already be owned by valid
# durable state; other cache-corruption failures remain fatal.
_assert_race_planned_path_is_evidence_owned(
cache_root,
execution_shard,
plan_shard=shard,
completed=completed,
created_at_ns=created_at_ns,
)
_assert_encoded_rows_match_cache(cache_root, encoded.records)
verified_after_race = _resolve_plan_cache(cache_root, plan, require_all=False)
if any(record.key not in verified_after_race.rows for record in encoded.records):
raise CacheCorruptError(
"concurrent cache winner disappeared during immediate verification"
) from race_exc
continue
inspection = inspect_cache_shard(cache_root, shard.relative_path)
_assert_inspection_matches_plan(
inspection,
shard=execution_shard,
created_at_ns=created_at_ns,
)
if tuple(record.embedding for record in inspection.records) != tuple(
_fp32_vector(record.embedding) for record in encoded.records
):
raise CacheCorruptError(
"published cache shard embeddings do not match encoded rows",
details={"shard_id": shard.shard_id, "path": shard.relative_path},
)
completed[shard.shard_id] = _completed_entry(
execution_shard,
inspection,
plan_shard_id=shard.shard_id,
origin="encoded",
encoded_rows=len(execution_shard.rows),
encode_batch_calls=encoded.encode_batch_calls,
encode_batch_seconds=encoded.encode_batch_seconds,
)
encoded_rows += len(execution_shard.rows)
encoded_shards += 1
state = _state_payload(plan_sha256=plan_sha256, completed=completed)
_atomic_write(evidence, _STATE_NAME, _pretty_json_bytes(state))
if logger is not None:
logger.info(
"data.shard.write",
shard_id=shard.shard_id,
path=path.relative_to(cache_root).as_posix(),
n_rows=len(execution_shard.rows),
size_bytes=inspection.size_bytes,
)
rate = (
None
if encoded.encode_batch_seconds <= 0
else len(execution_shard.rows) / encoded.encode_batch_seconds
)
_emit_progress(
logger,
shard=shard,
completed_shards=_resolved_plan_shard_count(plan, resolved_before.rows)
+ encoded_shards,
total_shards=len(plan.shards),
encoded_rows=encoded_rows,
resumed_rows=resumed_rows,
throughput_per_s=rate,
status="encoded",
)
resolved = _resolve_plan_cache(cache_root, plan, require_all=True)
reused_rows = len(resolved.rows) - resumed_rows - encoded_rows
if reused_rows < 0:
raise CacheCorruptError("cache build row provenance accounting is inconsistent")
elapsed_seconds = round(max(time.perf_counter() - started, 0.0), 6)
completion = _completion_payload(
encoded_rows=encoded_rows,
encoded_shards=encoded_shards,
resumed_rows=resumed_rows,
reused_rows=reused_rows,
resolved_unique_rows=len(resolved.rows),
planned_shards=len(plan.shards),
invocation_elapsed_seconds=elapsed_seconds,
run_id=None if logger is None else logger.run_id,
)
state = _state_payload(
plan_sha256=plan_sha256,
completed=completed,
completion=completion,
)
_atomic_write(evidence, _STATE_NAME, _pretty_json_bytes(state))
report_payload = _report_payload(
evidence=evidence,
plan=plan,
request_identity=request_identity,
request_count=len(requests),
cache_root=cache_root,
resolved=resolved,
completed=completed,
encoded_rows=encoded_rows,
encoded_shards=encoded_shards,
resumed_rows=resumed_rows,
reused_rows=reused_rows,
elapsed_seconds=elapsed_seconds,
batch_size=batch_size,
rows_per_shard=rows_per_shard,
created_at_ns=created_at_ns,
hardware=hardware,
encoder_device=encoder_device,
resolved_config=resolved_config_identity,
encoder_runtime_identity=runtime_identity,
run_id=None if logger is None else logger.run_id,
input_artifacts=input_identities,
)
if logger is not None:
throughput = cast(Mapping[str, object], report_payload["throughput"])
logger.info(
"data.cache.build.end",
completed_shards=len(plan.shards),
encoded_rows=encoded_rows,
resumed_rows=resumed_rows,
elapsed_s=cast(float, throughput["invocation_elapsed_seconds"]),
throughput_per_s=throughput["measured_encoded_rows_per_second"],
evidence_report=report_path.name,
)
# Logging is complete before the checksum closure. No builder-owned write
# occurs after SHA256SUMS is installed and verified.
_atomic_write(evidence, CACHE_BUILD_REPORT_NAME, _pretty_json_bytes(report_payload))
_write_checksums(evidence, expected_names=expected_evidence_names)
_assert_evidence_inventory(
evidence,
expected_names=expected_evidence_names,
require_complete=True,
)
_verify_checksums(evidence, expected_names=expected_evidence_names)
return CacheBuildReport(
report_path=report_path,
checksums_path=checksums_path,
payload=report_payload,
)