from viano_bug_analysis.telemetry.decorators import observed import hashlib import json from contextlib import contextmanager from datetime import datetime, timezone from importlib.resources import files from pathlib import Path from jsonschema import Draft202012Validator from jsonschema.exceptions import ValidationError from viano_bug_analysis.package import PackageReader, PackageReadError, PackageWriter, PackageWriteError from viano_bug_analysis.package.manifest import FORMAT_CHECKER, parse_manifest, safe_path from viano_bug_analysis.package.validator import CHUNK_SIZE from .errors import LifecycleError from .job import Job, timestamp from .locking import FileLock from .state import ACTIVE, State, check_transition from .storage import Workspace, atomic_json, json_bytes, key CHECKPOINT_POINTER = "lifecycle_checkpoint" CHECKPOINT_PREFIX = "history/checkpoints/" def _validator(): schema = json.loads(files("viano_bug_analysis.schemas").joinpath("checkpoint-v1.schema.json").read_text()) return Draft202012Validator(schema, format_checker=FORMAT_CHECKER) def fingerprint(reader: PackageReader) -> str: """Content identity independent of archive filename, ordering and compression.""" digest = hashlib.sha256() for info in sorted(reader._archive.infolist(), key=lambda item: item.filename): encoded = info.filename.encode("utf-8") digest.update(len(encoded).to_bytes(8, "big")) digest.update(encoded) if info.filename == "manifest.json": data = json_bytes(reader.manifest) digest.update(len(data).to_bytes(8, "big")) digest.update(data) else: digest.update(info.file_size.to_bytes(8, "big")) with reader._archive.open(info) as stream: while chunk := stream.read(CHUNK_SIZE): digest.update(chunk) return digest.hexdigest() def checkpoint_chain(reader: PackageReader) -> list[tuple[str, dict]]: validator = _validator() manifest = reader.manifest ref = manifest["artifacts"].get(CHECKPOINT_POINTER) result = [] visited = set() upper_generation = int(manifest["package_generation"]) + 1 upper_revision = int(manifest["analysis_revision"]) steps = set() try: while ref is not None: if not safe_path(ref) or not ref.startswith(CHECKPOINT_PREFIX) or ref in visited: raise ValueError("Unsafe or cyclic checkpoint chain") visited.add(ref) if ref not in reader.files: raise ValueError("Missing checkpoint parent") with reader._archive.open(ref) as stream: raw = reader.validator._read_bounded(stream, min(reader.validator.limits.max_manifest_bytes, reader.validator.limits.max_file_bytes)) data = parse_manifest(raw) validator.validate(data) State(data["state"]) State(data["previous_state"]) if data["resume_state"] is not None: if State(data["resume_state"]) not in ACTIVE or data["state"] != State.BLOCKED: raise ValueError("Invalid checkpoint resume state") elif data["state"] == State.BLOCKED: raise ValueError("Blocked checkpoint has no resume state") if data["case_id"] != manifest["case_id"] or data["package_id"] != manifest["package_id"]: raise ValueError("Checkpoint identity differs from manifest") generation = data["package_generation"] if generation >= upper_generation or data["source_generation"] != generation - 1: raise ValueError("Checkpoint generations are inconsistent") if data["analysis_revision"] > upper_revision or data["step_id"] in steps: raise ValueError("Checkpoint revision or step history is inconsistent") steps.add(data["step_id"]) upper_generation = generation upper_revision = int(data["analysis_revision"]) result.append((ref, data)) ref = data["parent"] if result and result[0][1]["package_generation"] == manifest["package_generation"]: if (result[0][1]["state"] != manifest["state"] or result[0][1]["analysis_revision"] != manifest["analysis_revision"] or result[0][1]["routing"] != manifest["routing"]): raise ValueError("Latest checkpoint disagrees with manifest") return result except (ValueError, TypeError, KeyError, RecursionError, PackageReadError) as error: raise LifecycleError("CHECKPOINT_INVALID", str(error)) from error except ValidationError as error: raise LifecycleError("CHECKPOINT_INVALID", str(error)) from error class CheckpointStore: def __init__(self, workspace: Workspace, limits=None, *, telemetry=None): self.workspace = workspace self.telemetry = telemetry self.writer = PackageWriter(limits) self.limits = self.writer.validator.limits if telemetry is not None and telemetry.package_limits is None: telemetry.package_limits = self.limits self._validation_cache = {} @contextmanager def _locks(self, case_id: str): with FileLock(self.workspace.runtime / "locks" / "worker.lock"): with FileLock(self.workspace.runtime / "locks" / f"case-{key(case_id)}.lock"): yield def latest(self, case_id: str, package_id: str) -> Path | None: directory = self.workspace.case_directory(case_id, package_id) candidates = {} for path in directory.glob("g*.zip"): if path.is_symlink(): raise LifecycleError("UNSAFE_WORKSPACE", "Managed packages cannot be symlinks") try: # Cache only after complete validation, keyed by a fresh hash # of every archive byte. Never trust mtime/filename as identity. archive_hash = hashlib.sha256() total = 0 with path.open("rb") as stream: while chunk := stream.read(CHUNK_SIZE): total += len(chunk) if total > self.limits.max_archive_bytes: raise LifecycleError("STORAGE_PACKAGE_INVALID", "Managed archive exceeds configured limits") archive_hash.update(chunk) cache_key = (case_id, package_id, path, repr(self.limits)) cached = self._validation_cache.get(cache_key) if cached and cached[0] == archive_hash.hexdigest(): candidates[cached[1]] = path continue with PackageReader(path, self.limits) as reader: manifest = reader.manifest generation = int(manifest["package_generation"]) if manifest["case_id"] != case_id or manifest["package_id"] != package_id or path.name != f"g{generation:08d}.zip": raise LifecycleError("STORAGE_CONFLICT", "Managed package identity or generation is inconsistent") chain = checkpoint_chain(reader) if not chain or chain[0][1]["package_generation"] != generation: raise LifecycleError("CHECKPOINT_INVALID", "Managed generation has no matching checkpoint") candidates[generation] = path validated_hash = hashlib.sha256() reader._source.seek(0) while chunk := reader._source.read(CHUNK_SIZE): validated_hash.update(chunk) if validated_hash.hexdigest() != archive_hash.hexdigest(): raise LifecycleError("STORAGE_PACKAGE_CHANGED", "Managed archive changed during validation") self._validation_cache[cache_key] = (validated_hash.hexdigest(), generation) if len(self._validation_cache) > 4096: self._validation_cache.pop(next(iter(self._validation_cache))) except PackageReadError as error: raise LifecycleError("STORAGE_PACKAGE_INVALID", str(error)) from error except LifecycleError as error: if error.code == "CHECKPOINT_INVALID": raise LifecycleError("STORAGE_CHECKPOINT_INVALID", str(error)) from error raise return candidates[max(candidates)] if candidates else None def _index(self, path: Path, manifest: dict) -> None: atomic_json(path.parent / "current.json", { "package": path.name, "package_id": manifest["package_id"], "case_id": manifest["case_id"], "package_generation": manifest["package_generation"], }) def _commit(self, reader, step_id, request_hash, state, *, updates=None, details=None, additions=None, resume_state=None): source_manifest = reader.manifest generation = int(source_manifest["package_generation"]) + 1 revision = (updates or {}).get("analysis_revision", source_manifest["analysis_revision"]) record = { "format": "viano-lifecycle-checkpoint", "format_version": 1, "case_id": source_manifest["case_id"], "package_id": source_manifest["package_id"], "package_generation": generation, "analysis_revision": revision, "step_id": step_id, "request_hash": request_hash, "source_generation": source_manifest["package_generation"], "source_fingerprint": fingerprint(reader), "previous_state": source_manifest["state"], "state": state, "routing": source_manifest["routing"] | (updates or {}).get("routing", {}), "resume_state": resume_state, "parent": source_manifest["artifacts"].get(CHECKPOINT_POINTER), "created_at": timestamp(), "details": details or {}, } ref = CHECKPOINT_PREFIX + key(step_id) + ".json" try: _validator().validate(record) except ValidationError as error: raise LifecycleError("CHECKPOINT_INVALID", str(error)) from error artifacts = dict((updates or {}).get("artifacts", {})) artifacts[CHECKPOINT_POINTER] = ref changes = dict(additions or {}) if any(name.startswith(CHECKPOINT_PREFIX) for name in changes): raise LifecycleError("RESERVED_ARTIFACT", "Checkpoint namespace is service-managed") encoded = json_bytes(record) if len(encoded) > min(self.limits.max_manifest_bytes, self.limits.max_file_bytes): raise LifecycleError("CHECKPOINT_LIMIT", "Checkpoint exceeds configured JSON byte limit") changes[ref] = encoded destination = self.workspace.case_directory(record["case_id"], record["package_id"]) / f"g{generation:08d}.zip" chain = checkpoint_chain(reader) job = Job.from_manifest(source_manifest | {"analysis_revision": revision}, state, chain[0][1]["created_at"] if chain else None) atomic_json(self.workspace.runtime / "current_job.json", job.to_dict()) result = self.writer.write(reader.path, destination, additions=changes, manifest_updates=dict(updates or {}) | {"state": state, "artifacts": artifacts}) self._index(destination, result.manifest) atomic_json(self.workspace.runtime / "current_job.json", None) if self.telemetry is not None: self.telemetry.guard(self.telemetry.case_checkpoint, result.manifest, record["details"]) return destination @observed("intake", identity=False) def intake(self, source: Path) -> tuple[str, Path]: try: with PackageReader(source, self.limits) as inbound: manifest = inbound.manifest with self._locks(manifest["case_id"]): latest = self.latest(manifest["case_id"], manifest["package_id"]) digest = fingerprint(inbound) chain = checkpoint_chain(inbound) if latest is not None: with PackageReader(latest, self.limits) as current: old = current.manifest old_chain = checkpoint_chain(current) known = [(cp["source_generation"], cp["source_fingerprint"]) for _, cp in old_chain] known.append((old["package_generation"], fingerprint(current))) if (manifest["package_generation"], digest) in known: self._index(latest, old) atomic_json(self.workspace.runtime / "current_job.json", None) return "DUPLICATE", latest if manifest["package_generation"] <= old["package_generation"]: raise LifecycleError("GENERATION_CONFLICT", "Previously seen generation has different content") if manifest["analysis_revision"] < old["analysis_revision"]: raise LifecycleError("REVISION_REGRESSION", "Returning case lost its analysis revision") if not {ref for ref, _ in old_chain}.issubset({ref for ref, _ in chain}): raise LifecycleError("CHECKPOINT_HISTORY_LOST", "Returning case lost checkpoint lineage") self.writer._verify_preservation(current, source, {"analysis/current.json"}) if manifest["state"] == State.BLOCKED: if not chain or chain[0][1]["state"] != State.BLOCKED: raise LifecycleError("CHECKPOINT_INVALID", "Blocked input requires its resume checkpoint") if not manifest["routing"]["reason"] or manifest["routing"]["retry_not_before"] is None: raise LifecycleError("INVALID_RETRY", "Blocked input requires a reason and retry timestamp") if not self.retry_due(manifest) and manifest["routing"]["reason"] != "AWAITING_TEST_FEEDBACK": return "DEFERRED", source state = State.VALIDATED if manifest["state"] == State.NEW else manifest["state"] step_id = f"intake:{int(manifest['package_generation'])}:{digest}" request_hash = key(step_id) path = self._commit(inbound, step_id, request_hash, state, resume_state=chain[0][1]["resume_state"] if state == State.BLOCKED else None, details={"operation": "intake", **({"automatic_retry": False} if state == State.BLOCKED and chain[0][1]["details"].get("automatic_retry") is False else {})}, updates={"routing": {"destination": "cases"}} if manifest["state"] == State.NEW else {}) return "ACCEPTED", path except (PackageReadError, PackageWriteError) as error: if isinstance(error, PackageWriteError) and error.result.issues[0].code == "PRESERVATION_FAILED": raise LifecycleError("PRESERVATION_FAILED", str(error)) from error raise LifecycleError("PACKAGE_ERROR", str(error)) from error except (OSError, ValueError, TypeError, RecursionError) as error: raise LifecycleError("LIFECYCLE_ERROR", str(error)) from error @staticmethod def retry_due(manifest: dict, now: datetime | None = None) -> bool: value = manifest["routing"]["retry_not_before"] return value is None or datetime.fromisoformat(value.upper().replace("Z", "+00:00")) <= (now or datetime.now(timezone.utc)) @observed("checkpoint_advance") def advance(self, case_id: str, package_id: str, target: str, *, step_id: str, reason: str | None = None, retry_not_before: str | None = None, details: dict | None = None, additions: dict[str, bytes] | None = None, artifacts: dict[str, str | None] | None = None, expected_generation: int | None = None) -> Path: try: if not isinstance(step_id, str) or not step_id.strip(): raise LifecycleError("INVALID_STEP", "A nonblank stable step_id is required") if details is not None and not isinstance(details, dict): raise LifecycleError("CHECKPOINT_INVALID", "Checkpoint details must be a JSON object") if CHECKPOINT_POINTER in (artifacts or {}): raise LifecycleError("RESERVED_ARTIFACT", "Lifecycle checkpoint pointer is service-managed") if expected_generation is not None and (isinstance(expected_generation, bool) or not isinstance(expected_generation, int) or expected_generation < 0): raise LifecycleError("INVALID_GENERATION", "expected_generation must be a nonnegative integer") payload = {"target": target, "reason": reason, "retry_not_before": retry_not_before, "details": details or {}, "artifacts": artifacts or {}, "additions": {name: hashlib.sha256(data).hexdigest() for name, data in (additions or {}).items()}} request_hash = hashlib.sha256(json_bytes(payload)).hexdigest() with self._locks(case_id): latest = self.latest(case_id, package_id) if latest is None: raise LifecycleError("CASE_NOT_FOUND", "Case has not been admitted") with PackageReader(latest, self.limits) as reader: manifest = reader.manifest chain = checkpoint_chain(reader) for _, checkpoint in chain: if checkpoint["step_id"] == step_id: if checkpoint["request_hash"] != request_hash: raise LifecycleError("STEP_CONFLICT", "step_id already used for different work") self._index(latest, manifest) atomic_json(self.workspace.runtime / "current_job.json", None) return latest if expected_generation is not None and manifest["package_generation"] != expected_generation: raise LifecycleError("GENERATION_CHANGED", "Case changed while artifacts were being prepared") resume = chain[0][1]["resume_state"] if manifest["state"] == State.BLOCKED else None if manifest["state"] == State.BLOCKED and manifest["routing"]["reason"] == "AWAITING_TEST_FEEDBACK": raise LifecycleError("EXTERNAL_FEEDBACK_REQUIRED", "Confirmed test feedback is required to resume this case") check_transition(manifest["state"], target, resume) if manifest["state"] == State.BLOCKED and not self.retry_due(manifest): raise LifecycleError("RETRY_NOT_DUE", "Retry timestamp has not been reached") routing = {"destination": "cases", "reason": None, "retry_not_before": None} if target == State.BLOCKED: if not isinstance(reason, str) or not reason.strip() or not isinstance(retry_not_before, str) or not FORMAT_CHECKER.conforms(retry_not_before, "date-time"): raise LifecycleError("INVALID_RETRY", "BLOCKED needs a reason and timezone-aware retry timestamp") routing.update(destination="inbox", reason=reason, retry_not_before=retry_not_before, retry_count=int(manifest["routing"]["retry_count"]) + 1) resume = manifest["state"] else: resume = None if target == State.FAILED_TO_ANALYZE: routing.update(destination="review", reason=reason) elif target in {State.OUTPUT_READY, State.HANDED_OFF}: routing["destination"] = "outbox" revision = int(manifest["analysis_revision"]) if target == State.REOPENED or (manifest["state"] == State.VALIDATED and target == State.BUILDING_CASE and revision == 0): revision += 1 history = {"analysis_revisions": max(int(manifest["history"]["analysis_revisions"]), revision)} return self._commit(reader, step_id, request_hash, target, details=details, additions=additions, resume_state=resume, updates={"routing": routing, "analysis_revision": revision, "history": history, "artifacts": artifacts or {}}) except (PackageReadError, PackageWriteError) as error: raise LifecycleError("PACKAGE_ERROR", str(error)) from error except (OSError, ValueError, TypeError, RecursionError) as error: raise LifecycleError("LIFECYCLE_ERROR", str(error)) from error