From e985e1d5f0cf90e2804bfd91bc21a4889d8f5965 Mon Sep 17 00:00:00 2001 From: Nitsu45 Date: Tue, 13 Oct 2026 15:21:11 +0200 Subject: [PATCH] Add persistent lifecycle checkpoints and resumable inbox processing --- README.md | 5 +- docs/lifecycle-v1.md | 166 ++++++++++ src/viano_bug_analysis/lifecycle/__init__.py | 10 + .../lifecycle/checkpoints.py | 290 ++++++++++++++++++ src/viano_bug_analysis/lifecycle/errors.py | 11 + src/viano_bug_analysis/lifecycle/job.py | 25 ++ src/viano_bug_analysis/lifecycle/locking.py | 49 +++ src/viano_bug_analysis/lifecycle/queue.py | 161 ++++++++++ src/viano_bug_analysis/lifecycle/state.py | 40 +++ src/viano_bug_analysis/lifecycle/storage.py | 58 ++++ .../schemas/checkpoint-v1.schema.json | 24 ++ tests/test_lifecycle_checkpoints.py | 227 ++++++++++++++ tests/test_lifecycle_queue.py | 248 +++++++++++++++ tests/test_lifecycle_state.py | 29 ++ tests/test_lifecycle_storage.py | 59 ++++ 15 files changed, 1401 insertions(+), 1 deletion(-) create mode 100644 docs/lifecycle-v1.md create mode 100644 src/viano_bug_analysis/lifecycle/__init__.py create mode 100644 src/viano_bug_analysis/lifecycle/checkpoints.py create mode 100644 src/viano_bug_analysis/lifecycle/errors.py create mode 100644 src/viano_bug_analysis/lifecycle/job.py create mode 100644 src/viano_bug_analysis/lifecycle/locking.py create mode 100644 src/viano_bug_analysis/lifecycle/queue.py create mode 100644 src/viano_bug_analysis/lifecycle/state.py create mode 100644 src/viano_bug_analysis/lifecycle/storage.py create mode 100644 src/viano_bug_analysis/schemas/checkpoint-v1.schema.json create mode 100644 tests/test_lifecycle_checkpoints.py create mode 100644 tests/test_lifecycle_queue.py create mode 100644 tests/test_lifecycle_state.py create mode 100644 tests/test_lifecycle_storage.py diff --git a/README.md b/README.md index 11b08f1..ce24131 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,8 @@ Project-independent foundations for reading, validating and extending ZIP case packages. The authoritative design is [docs/architecture.md](docs/architecture.md). The implemented v1 contract is documented in [docs/package-format-v1.md](docs/package-format-v1.md). +Job state, checkpoints and inbox admission are documented in +[docs/lifecycle-v1.md](docs/lifecycle-v1.md). Requires Python 3.11 or newer. @@ -13,5 +15,6 @@ python -m venv .venv .\.venv\Scripts\python.exe -m pytest ``` -This milestone includes no lifecycle worker, CaseBuilder, LLM, +The lifecycle foundation supports one-pass inbox processing and resumable package +checkpoints. It includes no analysis worker, CaseBuilder, LLM, case interpretation, hypothesis generation or investigation tools. diff --git a/docs/lifecycle-v1.md b/docs/lifecycle-v1.md new file mode 100644 index 0000000..e0c00a0 --- /dev/null +++ b/docs/lifecycle-v1.md @@ -0,0 +1,166 @@ +# Lifecycle foundation v1 + +`architecture.md` remains the authoritative design. This milestone provides only +deterministic job state, checkpoint storage, locking and inbox admission. It does +not execute CaseBuilder, interpretation, hypotheses, diagnosis, tools or LLMs. +State APIs allow future phase implementations to confirm their work; recording +a state is not evidence that this foundation executed that phase. + +## Package state and service state + +The package manifest is authoritative for identity, generation, revision, state +and routing. Detailed progress is stored as immutable JSON artifacts at +`history/checkpoints/.json`. The manifest's +`artifacts.lifecycle_checkpoint` points to the latest artifact. Each checkpoint +links to its predecessor, with format/version, identities, output generation, +analysis revision, previous/current state, routing snapshot, optional resume state, step identity, +request hash, source generation and content fingerprint, creation time and opaque +structured `details`. `checkpoint-v1.schema.json` defines the JSON contract. +Readers additionally check paths, cycles, state values, generation/revision order, +identity and agreement with the manifest for a matching generation. Checkpoint +JSON reads use the configurable manifest-byte limit as their security cap. + +Existing `analysis/current.json` is not repurposed as worker state. The checkpoint +namespace and manifest checkpoint pointer are reserved to the lifecycle service. +Checkpoints are outside frozen analysis revision directories. Every successful +checkpoint uses PackageWriter to create and validate one new package generation; +all old artifacts remain. No checkpoint exists only in an external database. + +Service layout: + +```text +/ + inbox/ incoming files of arbitrary names/extensions + quarantine// package.zip and validation_error.json + cache/.zip exact immutable input archives + cases/// + g00000001.zip immutable confirmed generations + current.json reconstructible local lookup index + outbox/ reserved for later analyzed-package delivery + review/ reserved for later human-review delivery + runtime/ + current_job.json coarse worker summary, or null when idle + locks/worker.lock persistent lock inode/file + locks/case-.lock +``` + +Opaque IDs are hashed before use in filesystem paths, including for locking. +Stored-generation filenames are internal lookup conventions and are always +checked against the manifest; incoming filenames never supply identity or state. +Runtime includes identities, input generation, analysis revision, target stage, +start time and last confirmed checkpoint time. It contains no detailed progress. +Runtime and local index JSON are written via temporary file, flush/fsync and +atomic replacement. They can be stale after a crash and are never authoritative. +Current generation is recovered from fully validated managed packages. Corrupt +managed storage is reported rather than silently rolling back to an older file. + +## Transitions and retry + +Normal transitions follow the architecture's linear sequence from NEW through +HANDED_OFF. REOPENED proceeds to BUILDING_CASE. Active phases may checkpoint +progress in the same state, become BLOCKED, or become FAILED_TO_ANALYZE. +HANDED_OFF and FAILED_TO_ANALYZE can become REOPENED. Skips, regressions and +unrelated BLOCKED resumes are rejected. + +The initial VALIDATED to BUILDING_CASE transition starts revision 1 if it is +still zero. REOPENED increments analysis revision exactly once; subsequently +entering BUILDING_CASE preserves it. The revision summary counter is maintained +without altering other history counters. Package generation changes independently. +These are explicit controller operations, not inferred from filenames. + +BLOCKED requires a nonblank reason and timezone-aware retry timestamp. The +checkpoint records the interrupted active state; routing becomes inbox with an +incremented retry count. Resume returns only to that state, only after the retry +time, and clears the reason/timestamp while retaining retry count. The retry +clock is UTC-aware. FAILED_TO_ANALYZE routes logically to review; OUTPUT_READY +and HANDED_OFF route logically to outbox; active/reopened work routes to cases. +The state and routing fields remain separate, and intake preserves a returning +package's routing. Only fresh NEW packages are advanced to VALIDATED on admission. + +This milestone does not create diagnoses, failure reports or Change Requests and +does not deliver packages to coding agents. Outbox/review are initialized but +physical output delivery belongs to the future phase controller. Detailed +checkpoint payloads and artifact pointer updates are caller-supplied contracts; +they receive structural validation, not diagnostic interpretation. + +## Idempotence, locking and recovery + +The content fingerprint hashes sorted internal paths and uncompressed bytes, +with canonical JSON for the manifest. Archive compression, order and filename +do not affect identity. Re-delivery of a known input generation and fingerprint +does not create a new generation. Each checkpoint carries its source receipt, +so this remains true when the package moves to a service without old local files. +A seen generation with different content is quarantined as a conflict. Newer +returning packages must preserve current protected files, checkpoint lineage and +analysis revision; they cannot replace prior content or erase history. The input +manifest remains authoritative for externally performed work on newer generations. + +`advance` requires a stable nonblank `step_id`. Repeating it with the same target, +details, artifact content and retry request returns the latest confirmed package +without rerunning a write. Reusing the ID for a different request is an error. +Artifact additions and the checkpoint are committed in the same package. External +side effects are outside this transaction and need their own idempotence policy. + +Nonblocking native OS locks use msvcrt on Windows and flock on POSIX. The coarse +runtime file models one worker, guarded by a worker lock. Case locks additionally +use case identity, independent of package identity. There is no timeout-based lock +stealing and no waiting on blocked cases. The OS releases locks when a process +exits, including a crash. Lock files remain in place to avoid inode replacement +races; do not delete them while workers are active. Shared network filesystems +require a separate distributed locking strategy and are not supported here. + +Commit sequence: + +1. Validate input, checkpoint lineage and requested transition under locks. +2. Atomically write the coarse runtime summary. +3. Publish the fully validated next generation via PackageWriter. +4. Update the reconstructible local index, then clear runtime to null. +5. For inbox admission, acknowledge the original only after publication, checking + that its raw bytes still match the cached snapshot. + +Before step 3, retry repeats only the unconfirmed operation. After step 3, the +published package and embedded receipt prove completion even if steps 4 or 5 +never happened. A retry repairs local metadata and does not increment generation. +File fsync and atomic visibility are provided, without claiming portable +directory-metadata durability through power loss. Crashes may leave unpublished +temporary files or incomplete quarantine directories; those are not treated as +confirmed generations. Garbage collection is outside this milestone. + +## Public API + +```python +from viano_bug_analysis.lifecycle import ( + Workspace, InboxProcessor, CheckpointStore, State, LifecycleError, +) + +workspace = Workspace("service-data") +processor = InboxProcessor(workspace) +results = processor.poll(max_jobs=100) + +store = CheckpointStore(workspace) +try: + package = store.advance( + "case-A", "stable-package", State.BUILDING_CASE, + step_id="revision-1:start-case-building", + details={"inputs_collected": []}, + ) +except LifecycleError as error: + print(error.code, str(error)) +``` + +`poll` is a single pass, not a daemon. It validates bounded cached snapshots, +admits valid cases, preserves invalid originals in quarantine with a structured +error report, and continues to the next input after per-input errors. Oversized +archives are moved directly to quarantine without creating an oversized cache +copy. Due BLOCKED local cases are resumed once; future retries remain deferred. +Deferred or busy inputs do not consume the `max_jobs` admission limit, so they +cannot indefinitely crowd fresh work out of a poll. Such a pass may return more +than `max_jobs` results and inspect all pending files. No diagnostic work runs. + +Results are ACCEPTED, DUPLICATE, QUARANTINED, DEFERRED, BUSY, RESUMED or ERROR, +with paths and optional error code/message. Conflicting/corrupt incoming records +are quarantined; infrastructure failures leave inbox input pending. A changed +producer file is not acknowledged. Workspace storage must be service-owned on a +local filesystem supporting hard links, and producers must publish complete +files atomically without modifying them during processing. Symlink inputs are +rejected. This is not a hardened multi-tenant filesystem boundary. diff --git a/src/viano_bug_analysis/lifecycle/__init__.py b/src/viano_bug_analysis/lifecycle/__init__.py new file mode 100644 index 0000000..1e306a4 --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/__init__.py @@ -0,0 +1,10 @@ +from .checkpoints import CheckpointStore +from .errors import LifecycleError, LockBusy +from .job import Job +from .locking import FileLock +from .queue import InboxProcessor, IntakeResult +from .state import State +from .storage import Workspace + +__all__ = ["CheckpointStore", "LifecycleError", "LockBusy", "Job", "FileLock", + "InboxProcessor", "IntakeResult", "State", "Workspace"] diff --git a/src/viano_bug_analysis/lifecycle/checkpoints.py b/src/viano_bug_analysis/lifecycle/checkpoints.py new file mode 100644 index 0000000..f4f98b7 --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/checkpoints.py @@ -0,0 +1,290 @@ +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): + self.workspace = workspace + self.writer = PackageWriter(limits) + self.limits = self.writer.validator.limits + + @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: + 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 + 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) + return destination + + 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): + 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"}, + 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)) + + 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) -> 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") + 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 + resume = chain[0][1]["resume_state"] if manifest["state"] == State.BLOCKED else None + 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 diff --git a/src/viano_bug_analysis/lifecycle/errors.py b/src/viano_bug_analysis/lifecycle/errors.py new file mode 100644 index 0000000..47b6c33 --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/errors.py @@ -0,0 +1,11 @@ +class LifecycleError(Exception): + """Controlled storage, checkpoint or lifecycle rejection.""" + + def __init__(self, code: str, message: str): + self.code = code + super().__init__(message) + + +class LockBusy(LifecycleError): + def __init__(self): + super().__init__("LOCK_BUSY", "Another worker holds this lock") diff --git a/src/viano_bug_analysis/lifecycle/job.py b/src/viano_bug_analysis/lifecycle/job.py new file mode 100644 index 0000000..7f07c08 --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/job.py @@ -0,0 +1,25 @@ +from dataclasses import asdict, dataclass +from datetime import datetime, timezone + + +def timestamp() -> str: + return datetime.now(timezone.utc).isoformat() + + +@dataclass(frozen=True) +class Job: + case_id: str + package_id: str + package_generation: int + analysis_revision: int + stage: str + started_at: str + last_checkpoint_at: str | None = None + + @classmethod + def from_manifest(cls, manifest: dict, stage: str, last_checkpoint_at: str | None = None): + return cls(manifest["case_id"], manifest["package_id"], int(manifest["package_generation"]), + int(manifest["analysis_revision"]), stage, timestamp(), last_checkpoint_at) + + def to_dict(self) -> dict: + return asdict(self) diff --git a/src/viano_bug_analysis/lifecycle/locking.py b/src/viano_bug_analysis/lifecycle/locking.py new file mode 100644 index 0000000..a9dd7fd --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/locking.py @@ -0,0 +1,49 @@ +import os +from pathlib import Path + +from .errors import LifecycleError, LockBusy + + +class FileLock: + """Nonblocking OS lock; automatically released when the process exits. + + Lock files persist and must not be removed while workers are running. + """ + + def __init__(self, path: Path): + self.path = path + self._stream = None + + def __enter__(self): + if self._stream is not None: + raise RuntimeError("Lock already held") + try: + if self.path.is_symlink(): + raise LifecycleError("UNSAFE_LOCK", "Lock path is a symlink") + self._stream = open(self.path, "a+b") + if os.fstat(self._stream.fileno()).st_size == 0: + self._stream.write(b"\0") + self._stream.flush() + self._stream.seek(0) + try: + if os.name == "nt": + import msvcrt + msvcrt.locking(self._stream.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(self._stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as error: + raise LockBusy() from error + return self + except (OSError, LifecycleError) as error: + if self._stream is not None: + self._stream.close() + self._stream = None + if isinstance(error, LifecycleError): + raise + raise LifecycleError("LOCK_ERROR", str(error)) from error + + def __exit__(self, *args): + if self._stream is not None: + self._stream.close() + self._stream = None diff --git a/src/viano_bug_analysis/lifecycle/queue.py b/src/viano_bug_analysis/lifecycle/queue.py new file mode 100644 index 0000000..0716a41 --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/queue.py @@ -0,0 +1,161 @@ +import hashlib +import os +import tempfile +import uuid +from dataclasses import asdict, dataclass +from pathlib import Path + +from viano_bug_analysis.package import PackageReader, PackageReadError, PackageValidator +from viano_bug_analysis.package.validator import CHUNK_SIZE + +from .checkpoints import CheckpointStore, checkpoint_chain +from .errors import LifecycleError, LockBusy +from .state import State +from .storage import Workspace, atomic_json + + +@dataclass(frozen=True) +class IntakeResult: + status: str + source: Path + package: Path | None = None + code: str | None = None + message: str | None = None + + +class InboxProcessor: + """One inbox pass, no analysis execution or waits for external conditions.""" + + def __init__(self, workspace: Workspace, limits=None): + self.workspace = workspace + self.store = CheckpointStore(workspace, limits) + self.validator = PackageValidator(self.store.limits) + + def _snapshot(self, source: Path) -> tuple[Path, str]: + temporary = None + try: + digest = hashlib.sha256() + total = 0 + with tempfile.NamedTemporaryFile(dir=self.workspace.cache, prefix=".snapshot-", delete=False) as copied: + temporary = Path(copied.name) + with source.open("rb") as original: + while chunk := original.read(CHUNK_SIZE): + total += len(chunk) + if total > self.store.limits.max_archive_bytes: + raise LifecycleError("ARCHIVE_LIMIT", "Input exceeds configured archive byte limit") + copied.write(chunk) + digest.update(chunk) + copied.flush() + os.fsync(copied.fileno()) + raw_hash = digest.hexdigest() + destination = self.workspace.cache / f"{raw_hash}.zip" + if destination.is_symlink(): + raise LifecycleError("CACHE_CONFLICT", "Cached original is a symlink") + try: + os.link(temporary, destination) + except FileExistsError: + if self._raw_hash(destination) != raw_hash: + raise LifecycleError("CACHE_CONFLICT", "Cached original no longer matches its hash") + return destination, raw_hash + except OSError as error: + raise LifecycleError("STORAGE_ERROR", str(error)) from error + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + @staticmethod + def _raw_hash(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(CHUNK_SIZE): + digest.update(chunk) + return digest.hexdigest() + + def _acknowledge(self, source: Path, raw_hash: str) -> None: + if source.is_symlink() or self._raw_hash(source) != raw_hash: + raise LifecycleError("SOURCE_CHANGED", "Producer changed inbox input during processing") + source.unlink() + + def _quarantine(self, source: Path, issues: list[dict], raw_hash: str | None = None) -> IntakeResult: + directory = self.workspace.quarantine / uuid.uuid4().hex + directory.mkdir() + atomic_json(directory / "validation_error.json", { + "format": "viano-validation-error", "format_version": 1, + "source_name": source.name, "issues": issues, + }) + # Same workspace filesystem: preserve the exact archive without extraction. + if source.is_symlink() or (raw_hash is not None and self._raw_hash(source) != raw_hash): + raise LifecycleError("SOURCE_CHANGED", "Input changed before quarantine") + os.replace(source, directory / "package.zip") + return IntakeResult("QUARANTINED", source, directory / "package.zip", issues[0]["code"], issues[0]["message"]) + + def process(self, source: Path) -> IntakeResult: + source = Path(source) + try: + if source.parent.resolve() != self.workspace.inbox or source.is_symlink() or not source.is_file(): + raise LifecycleError("UNSAFE_INPUT", "Expected a regular file directly in the inbox") + try: + snapshot, raw_hash = self._snapshot(source) + except LifecycleError as error: + if error.code == "ARCHIVE_LIMIT": + return self._quarantine(source, [{"code": error.code, "message": str(error), "path": None}]) + raise + result = self.validator.validate(snapshot) + if not result.valid: + return self._quarantine(source, [asdict(issue) for issue in result.issues], raw_hash) + try: + status, managed = self.store.intake(snapshot) + except LockBusy: + return IntakeResult("BUSY", source, code="LOCK_BUSY") + except LifecycleError as error: + if error.code in {"GENERATION_CONFLICT", "REVISION_REGRESSION", "CHECKPOINT_HISTORY_LOST", "CHECKPOINT_INVALID", "PRESERVATION_FAILED", "INVALID_RETRY"}: + return self._quarantine(source, [{"code": error.code, "message": str(error), "path": None}], raw_hash) + raise + if status != "DEFERRED": + self._acknowledge(source, raw_hash) + return IntakeResult(status, source, managed) + except (LifecycleError, OSError, PackageReadError) as error: + return IntakeResult("ERROR", source, code=getattr(error, "code", "STORAGE_ERROR"), message=str(error)) + + def poll(self, *, max_jobs: int = 100) -> list[IntakeResult]: + if isinstance(max_jobs, bool) or not isinstance(max_jobs, int) or max_jobs <= 0: + raise ValueError("max_jobs must be a positive integer") + outcomes = [] + completed = 0 + for source in sorted(self.workspace.inbox.iterdir()): + if completed >= max_jobs: + break + if source.is_file() or source.is_symlink(): + result = self.process(source) + outcomes.append(result) + completed += result.status not in {"DEFERRED", "BUSY"} + # Waiting cases do not occupy a worker. Resume only due local checkpoints. + identities = set() + for path in self.workspace.cases.glob("*/*/g*.zip"): + try: + with PackageReader(path, self.store.limits) as reader: + manifest = reader.manifest + identities.add((manifest["case_id"], manifest["package_id"])) + except (PackageReadError, OSError) as error: + outcomes.append(IntakeResult("ERROR", path, code="PACKAGE_ERROR", message=str(error))) + for case_id, package_id in sorted(identities): + if completed >= max_jobs: + break + path = self.workspace.cases + try: + path = self.store.latest(case_id, package_id) + with PackageReader(path, self.store.limits) as reader: + manifest = reader.manifest + if manifest["state"] != State.BLOCKED or not self.store.retry_due(manifest): + continue + resume = checkpoint_chain(reader)[0][1]["resume_state"] + resumed = self.store.advance(case_id, package_id, resume, + step_id=f"retry:{int(manifest['package_generation'])}") + outcomes.append(IntakeResult("RESUMED", path, resumed)) + completed += 1 + except LockBusy: + outcomes.append(IntakeResult("BUSY", path, code="LOCK_BUSY")) + except (LifecycleError, PackageReadError, OSError) as error: + outcomes.append(IntakeResult("ERROR", path, code=getattr(error, "code", "STORAGE_ERROR"), message=str(error))) + completed += 1 + return outcomes diff --git a/src/viano_bug_analysis/lifecycle/state.py b/src/viano_bug_analysis/lifecycle/state.py new file mode 100644 index 0000000..9b44d1f --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/state.py @@ -0,0 +1,40 @@ +from enum import StrEnum + +from .errors import LifecycleError + + +class State(StrEnum): + NEW = "NEW" + VALIDATED = "VALIDATED" + BUILDING_CASE = "BUILDING_CASE" + INTERPRETING_CASE = "INTERPRETING_CASE" + GENERATING_HYPOTHESES = "GENERATING_HYPOTHESES" + INVESTIGATING_HYPOTHESES = "INVESTIGATING_HYPOTHESES" + BUILDING_DIAGNOSIS = "BUILDING_DIAGNOSIS" + OUTPUT_READY = "OUTPUT_READY" + HANDED_OFF = "HANDED_OFF" + BLOCKED = "BLOCKED" + FAILED_TO_ANALYZE = "FAILED_TO_ANALYZE" + REOPENED = "REOPENED" + + +ORDER = [State.NEW, State.VALIDATED, State.BUILDING_CASE, State.INTERPRETING_CASE, + State.GENERATING_HYPOTHESES, State.INVESTIGATING_HYPOTHESES, + State.BUILDING_DIAGNOSIS, State.OUTPUT_READY, State.HANDED_OFF] +ACTIVE = frozenset(ORDER[1:-2]) | {State.REOPENED} + + +def check_transition(previous: str, target: str, resume_state: str | None = None) -> None: + try: + old, new = State(previous), State(target) + except ValueError as error: + raise LifecycleError("INVALID_STATE", str(error)) from error + permitted = old == new and old in ACTIVE + if old in ORDER[:-1]: + permitted |= new == ORDER[ORDER.index(old) + 1] + permitted |= old in ACTIVE and new in {State.BLOCKED, State.FAILED_TO_ANALYZE} + permitted |= old == State.REOPENED and new == State.BUILDING_CASE + permitted |= old in {State.HANDED_OFF, State.FAILED_TO_ANALYZE} and new == State.REOPENED + permitted |= old == State.BLOCKED and new == resume_state and new in ACTIVE + if not permitted: + raise LifecycleError("INVALID_TRANSITION", f"Cannot transition {old} to {new}") diff --git a/src/viano_bug_analysis/lifecycle/storage.py b/src/viano_bug_analysis/lifecycle/storage.py new file mode 100644 index 0000000..35cf7c9 --- /dev/null +++ b/src/viano_bug_analysis/lifecycle/storage.py @@ -0,0 +1,58 @@ +import hashlib +import json +import os +import tempfile +from pathlib import Path + +from .errors import LifecycleError + + +def key(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() + + +def json_bytes(value: object) -> bytes: + return json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False).encode("utf-8") + + +def atomic_json(path: Path, value: object) -> None: + temporary = None + try: + with tempfile.NamedTemporaryFile(dir=path.parent, prefix=".json-", delete=False) as stream: + temporary = Path(stream.name) + stream.write(json_bytes(value)) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + except (OSError, ValueError, TypeError, RecursionError) as error: + raise LifecycleError("STORAGE_ERROR", str(error)) from error + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + + +class Workspace: + """Service-owned directories. Case identifiers never become path segments.""" + + def __init__(self, root: str | os.PathLike[str]): + self.root = Path(root).resolve() + try: + for name in ("inbox", "quarantine", "cache", "cases", "outbox", "review", "runtime"): + path = self.root / name + if path.is_symlink(): + raise LifecycleError("UNSAFE_WORKSPACE", "Workspace directories cannot be symlinks") + path.mkdir(parents=True, exist_ok=True) + setattr(self, name, path) + lock_directory = self.runtime / "locks" + if lock_directory.is_symlink(): + raise LifecycleError("UNSAFE_WORKSPACE", "Lock directory cannot be a symlink") + lock_directory.mkdir(exist_ok=True) + except OSError as error: + raise LifecycleError("STORAGE_ERROR", str(error)) from error + + def case_directory(self, case_id: str, package_id: str) -> Path: + path = self.cases / key(case_id) / key(package_id) + if path.parent.is_symlink() or path.is_symlink(): + raise LifecycleError("UNSAFE_WORKSPACE", "Case storage cannot be symlinked") + path.mkdir(parents=True, exist_ok=True) + return path diff --git a/src/viano_bug_analysis/schemas/checkpoint-v1.schema.json b/src/viano_bug_analysis/schemas/checkpoint-v1.schema.json new file mode 100644 index 0000000..696835c --- /dev/null +++ b/src/viano_bug_analysis/schemas/checkpoint-v1.schema.json @@ -0,0 +1,24 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "type": "object", + "required": ["format", "format_version", "case_id", "package_id", "package_generation", "analysis_revision", "step_id", "request_hash", "source_generation", "source_fingerprint", "previous_state", "state", "routing", "resume_state", "parent", "created_at", "details"], + "properties": { + "format": {"const": "viano-lifecycle-checkpoint"}, + "format_version": {"type": "integer", "const": 1}, + "case_id": {"type": "string", "minLength": 1}, + "package_id": {"type": "string", "minLength": 1}, + "package_generation": {"type": "integer", "minimum": 1}, + "analysis_revision": {"type": "integer", "minimum": 0}, + "step_id": {"type": "string", "minLength": 1, "pattern": "\\S"}, + "request_hash": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "source_generation": {"type": "integer", "minimum": 0}, + "source_fingerprint": {"type": "string", "pattern": "^[0-9a-f]{64}$"}, + "previous_state": {"type": "string"}, + "state": {"type": "string"}, + "routing": {"type": "object", "required": ["destination", "reason", "retry_count", "retry_not_before"]}, + "resume_state": {"type": ["string", "null"]}, + "parent": {"type": ["string", "null"]}, + "created_at": {"type": "string", "format": "date-time"}, + "details": {"type": "object"} + } +} diff --git a/tests/test_lifecycle_checkpoints.py b/tests/test_lifecycle_checkpoints.py new file mode 100644 index 0000000..2f546e6 --- /dev/null +++ b/tests/test_lifecycle_checkpoints.py @@ -0,0 +1,227 @@ +import json +from datetime import datetime, timedelta, timezone + +import pytest + +from viano_bug_analysis.lifecycle import CheckpointStore, LifecycleError, State, Workspace +from viano_bug_analysis.lifecycle.checkpoints import CHECKPOINT_POINTER, checkpoint_chain +from viano_bug_analysis.package import PackageReader, PackageValidator, PackageWriter + + +@pytest.fixture +def admitted(make_package, tmp_path): + workspace = Workspace(tmp_path / "service") + store = CheckpointStore(workspace) + status, path = store.intake(make_package(entries={"unknown": b"keep"})) + assert status == "ACCEPTED" + return workspace, store, path + + +def test_checkpoint_inside_package_and_runtime_is_coarse(admitted): + workspace, store, path = admitted + with PackageReader(path) as reader: + manifest = reader.manifest + assert manifest["state"] == State.VALIDATED + assert manifest["package_generation"] == 1 + assert manifest["analysis_revision"] == 0 + chain = checkpoint_chain(reader) + assert chain[0][1]["previous_state"] == State.NEW + assert chain[0][1]["source_generation"] == 0 + assert chain[0][1]["details"] == {"operation": "intake"} + assert reader.read_bytes("unknown") == b"keep" + assert json.loads((workspace.runtime / "current_job.json").read_text()) is None + + +def test_progress_idempotence_and_conflicting_step(admitted): + _, store, _ = admitted + path = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") + with PackageReader(path) as reader: + assert reader.manifest["package_generation"] == 2 + assert reader.manifest["analysis_revision"] == 1 + assert store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") == path + with pytest.raises(LifecycleError) as error: + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"different": True}) + assert error.value.code == "STEP_CONFLICT" + next_path = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-progress", details={"collected": ["input-A"]}, additions={"case/inputs/a": b"generic"}) + assert store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") == next_path + with PackageReader(next_path) as reader: + assert len(checkpoint_chain(reader)) == 3 + assert reader.manifest["analysis_revision"] == 1 + assert reader.read_bytes("case/inputs/a") == b"generic" + + +def test_full_lifecycle_and_reopen(admitted): + _, store, _ = admitted + states = [State.BUILDING_CASE, State.INTERPRETING_CASE, State.GENERATING_HYPOTHESES, State.INVESTIGATING_HYPOTHESES, State.BUILDING_DIAGNOSIS, State.OUTPUT_READY, State.HANDED_OFF, State.REOPENED, State.BUILDING_CASE] + for index, state in enumerate(states): + path = store.advance("case-A", "stable-package", state, step_id=f"step-{index}") + with PackageReader(path) as reader: + assert reader.manifest["analysis_revision"] == 2 + assert reader.manifest["history"]["analysis_revisions"] == 2 + assert reader.manifest["package_generation"] == 10 + assert len(checkpoint_chain(reader)) == 10 + assert PackageValidator().validate(path).valid + + +def test_failure_routes_review_then_reopen(admitted): + _, store, _ = admitted + path = store.advance("case-A", "stable-package", State.FAILED_TO_ANALYZE, step_id="failure", reason="No available analysis") + with PackageReader(path) as reader: + assert reader.manifest["routing"]["destination"] == "review" + path = store.advance("case-A", "stable-package", State.REOPENED, step_id="review-return") + with PackageReader(path) as reader: + assert reader.manifest["analysis_revision"] == 1 + + +def test_block_retry_and_resume(admitted): + _, store, _ = admitted + future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat() + path = store.advance("case-A", "stable-package", State.BLOCKED, step_id="blocked", reason="External dependency", retry_not_before=future) + with PackageReader(path) as reader: + assert reader.manifest["routing"]["retry_count"] == 1 + assert checkpoint_chain(reader)[0][1]["resume_state"] == State.VALIDATED + with pytest.raises(LifecycleError) as error: + store.advance("case-A", "stable-package", State.VALIDATED, step_id="resume") + assert error.value.code == "RETRY_NOT_DUE" + with pytest.raises(LifecycleError): + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="wrong-resume") + + +@pytest.mark.parametrize("retry", [None, "not-a-date", "2026-01-01T00:00:00", "2026-02-30T00:00:00Z"]) +def test_invalid_retry_rejected_before_publication(retry, admitted): + _, store, path = admitted + with pytest.raises(LifecycleError): + store.advance("case-A", "stable-package", State.BLOCKED, step_id="bad", reason="External dependency", retry_not_before=retry) + assert store.latest("case-A", "stable-package") == path + + +def test_restart_uses_package_not_current_index(admitted): + workspace, _, path = admitted + (path.parent / "current.json").write_text("broken index") + (workspace.runtime / "current_job.json").write_text("broken runtime summary") + restarted = CheckpointStore(workspace) + assert restarted.latest("case-A", "stable-package") == path + assert restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start").exists() + + +def test_crash_after_package_publication_before_index(admitted, monkeypatch): + workspace, store, original = admitted + def crash(*args): + raise LifecycleError("SIMULATED_CRASH", "after publication") + monkeypatch.setattr(store, "_index", crash) + with pytest.raises(LifecycleError): + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") + summary = json.loads((workspace.runtime / "current_job.json").read_text()) + assert set(summary) == {"case_id", "package_id", "package_generation", "analysis_revision", "stage", "started_at", "last_checkpoint_at"} + restarted = CheckpointStore(workspace) + path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") + assert path != original + assert len(list(path.parent.glob("g*.zip"))) == 2 + assert json.loads((workspace.runtime / "current_job.json").read_text()) is None + + +def test_crash_before_publication_repeats_only_pending_step(admitted, monkeypatch): + workspace, store, original = admitted + def crash(*args, **kwargs): + raise LifecycleError("SIMULATED_CRASH", "before publication") + monkeypatch.setattr(store.writer, "write", crash) + with pytest.raises(LifecycleError): + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") + restarted = CheckpointStore(workspace) + assert restarted.latest("case-A", "stable-package") == original + path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") + with PackageReader(path) as reader: + assert reader.manifest["package_generation"] == 2 + + +def test_import_portable_package_without_service_memory(admitted, tmp_path): + _, store, _ = admitted + original = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"progress": "retained"}) + other = CheckpointStore(Workspace(tmp_path / "other-service")) + status, imported = other.intake(original) + assert status == "ACCEPTED" + with PackageReader(imported) as reader: + assert checkpoint_chain(reader)[1][1]["details"] == {"progress": "retained"} + # Previous work remains idempotent even without the previous service's files. + assert other.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start", details={"progress": "retained"}) == imported + + +def test_newer_return_preserves_local_history(admitted, tmp_path): + _, store, original = admitted + returned = tmp_path / "returned" + PackageWriter().write(original, returned, additions={"feedback/new": b"new feedback"}) + status, path = store.intake(returned) + assert status == "ACCEPTED" + with PackageReader(path) as reader: + assert reader.manifest["package_generation"] == 3 + assert reader.read_bytes("feedback/new") == b"new feedback" + + +def test_invalid_details_not_published(admitted): + _, store, original = admitted + with pytest.raises(LifecycleError): + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="invalid", details=["wrong"]) + assert store.latest("case-A", "stable-package") == original + + +def test_corrupt_managed_package_fails_without_rollback(admitted): + _, store, path = admitted + newer = store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="build-start") + newer.write_bytes(b"broken") + with pytest.raises(LifecycleError) as error: + store.latest("case-A", "stable-package") + assert error.value.code == "STORAGE_PACKAGE_INVALID" + assert path.exists() + + +def test_real_process_crash_after_publication(admitted): + import os + import subprocess + import sys + from pathlib import Path + workspace, _, _ = admitted + script = "from viano_bug_analysis.lifecycle import Workspace,CheckpointStore,State; import os,sys; s=CheckpointStore(Workspace(sys.argv[1])); s._index=lambda *args: os._exit(23); s.advance('case-A','stable-package',State.BUILDING_CASE,step_id='process-build')" + env = dict(os.environ, PYTHONPATH=str(Path(__file__).resolve().parents[1] / "src")) + process = subprocess.run([sys.executable, "-c", script, str(workspace.root)], env=env, capture_output=True, timeout=15) + assert process.returncode == 23, process.stderr.decode() + restarted = CheckpointStore(workspace) + path = restarted.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="process-build") + with PackageReader(path) as reader: + assert reader.manifest["package_generation"] == 2 + assert len(checkpoint_chain(reader)) == 2 + assert len(list(path.parent.glob("g*.zip"))) == 2 + + +def test_reserved_checkpoint_artifacts_rejected(admitted): + _, store, original = admitted + with pytest.raises(LifecycleError) as error: + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="forbidden", artifacts={CHECKPOINT_POINTER: None}) + assert error.value.code == "RESERVED_ARTIFACT" + with pytest.raises(LifecycleError): + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="forbidden", additions={"history/checkpoints/user.json": b"{}"}) + assert store.latest("case-A", "stable-package") == original + + +def test_invalid_block_request_types(admitted): + _, store, _ = admitted + for reason, retry in [(3, "2026-01-01T00:00:00Z"), ("reason", 4)]: + with pytest.raises(LifecycleError) as error: + store.advance("case-A", "stable-package", State.BLOCKED, step_id="bad", reason=reason, retry_not_before=retry) + assert error.value.code == "INVALID_RETRY" + + +def test_oversized_checkpoint_not_published(make_package, tmp_path): + from viano_bug_analysis.package import PackageLimits + store = CheckpointStore(Workspace(tmp_path / "service"), PackageLimits(max_manifest_bytes=4000)) + _, original = store.intake(make_package()) + with pytest.raises(LifecycleError) as error: + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="large", details={"data": "x" * 5000}) + assert error.value.code == "CHECKPOINT_LIMIT" + assert store.latest("case-A", "stable-package") == original + + +def test_empty_array_details_rejected(admitted): + _, store, original = admitted + with pytest.raises(LifecycleError): + store.advance("case-A", "stable-package", State.BUILDING_CASE, step_id="bad", details=[]) + assert store.latest("case-A", "stable-package") == original diff --git a/tests/test_lifecycle_queue.py b/tests/test_lifecycle_queue.py new file mode 100644 index 0000000..e853a97 --- /dev/null +++ b/tests/test_lifecycle_queue.py @@ -0,0 +1,248 @@ +import json +import shutil +from datetime import datetime, timedelta, timezone + +import pytest + +from viano_bug_analysis.lifecycle import FileLock, InboxProcessor, LifecycleError, State, Workspace +from viano_bug_analysis.lifecycle.storage import key +from viano_bug_analysis.package import PackageLimits, PackageReader + + +@pytest.fixture +def queue(tmp_path): + return InboxProcessor(Workspace(tmp_path / "service")) + + +def deliver(queue, source, name="anything.bin"): + path = queue.workspace.inbox / name + shutil.copyfile(source, path) + return path + + +def test_intake_and_renamed_duplicate(queue, make_package): + source = make_package(entries={"unknown": b"retained"}) + before = source.read_bytes() + inbound = deliver(queue, source) + result = queue.process(inbound) + assert result.status == "ACCEPTED" + assert not inbound.exists() + assert next(queue.workspace.cache.glob("*.zip")).read_bytes() == before + duplicate = deliver(queue, source, "renamed.zip") + again = queue.process(duplicate) + assert again.status == "DUPLICATE" + assert again.package == result.package + assert not duplicate.exists() + assert len(list(result.package.parent.glob("g*.zip"))) == 1 + + +def test_repacked_duplicate(queue, make_package): + import zipfile + first = make_package(entries={"unknown": b"retained"}, name="first") + original = queue.process(deliver(queue, first)).package + repacked = make_package(entries={"unknown": b"retained"}, name="second", compression=zipfile.ZIP_DEFLATED) + result = queue.process(deliver(queue, repacked)) + assert result.status == "DUPLICATE" + assert result.package == original + + +def test_generation_conflict_quarantined(queue, make_package): + queue.process(deliver(queue, make_package(entries={"unknown": b"old"}))) + conflict = deliver(queue, make_package(entries={"unknown": b"different"})) + result = queue.process(conflict) + assert result.status == "QUARANTINED" + assert result.code == "GENERATION_CONFLICT" + assert json.loads((result.package.parent / "validation_error.json").read_text())["issues"][0]["code"] == "GENERATION_CONFLICT" + + +@pytest.mark.parametrize("raw", [b"broken", b"{}", b"[]"]) +def test_invalid_packages_preserved_in_quarantine(raw, queue, make_package): + source = make_package(raw=raw) + before = source.read_bytes() + result = queue.process(deliver(queue, source)) + assert result.status == "QUARANTINED" + assert result.package.read_bytes() == before + assert (result.package.parent / "validation_error.json").exists() + + +def test_oversized_archive_quarantined_without_caching(tmp_path): + queue = InboxProcessor(Workspace(tmp_path / "service"), PackageLimits(max_archive_bytes=20)) + source = queue.workspace.inbox / "huge" + source.write_bytes(b"x" * 100) + result = queue.process(source) + assert result.status == "QUARANTINED" + assert result.package.read_bytes() == b"x" * 100 + assert not list(queue.workspace.cache.glob("*.zip")) + + +def test_case_lock_contention_leaves_input_pending(queue, make_package): + source = deliver(queue, make_package()) + with FileLock(queue.workspace.runtime / "locks" / f"case-{key('case-A')}.lock"): + result = queue.process(source) + assert result.status == "BUSY" + assert source.exists() + assert queue.process(source).status == "ACCEPTED" + + +def test_worker_lock_contention(queue, make_package): + source = deliver(queue, make_package()) + with FileLock(queue.workspace.runtime / "locks" / "worker.lock"): + assert queue.process(source).status == "BUSY" + assert queue.process(source).status == "ACCEPTED" + + +def test_future_blocked_case_does_not_hold_worker(queue, make_package, manifest): + first = queue.process(deliver(queue, make_package())) + future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat() + blocked = queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="external-unavailable", reason="External dependency", retry_not_before=future) + # A returned blocked case keeps its retry time and does not occupy the worker. + other_service = InboxProcessor(Workspace(queue.workspace.root.parent / "other")) + deferred = other_service.process(deliver(other_service, blocked)) + assert deferred.status == "DEFERRED" + manifest.update(case_id="case-B", package_id="other-package") + fresh = deliver(other_service, make_package(name="new-case"), "second.bin") + outcomes = other_service.poll() + assert any(outcome.status == "ACCEPTED" and outcome.source == fresh for outcome in outcomes) + + +def test_due_blocked_case_resumes_once(queue, make_package): + queue.process(deliver(queue, make_package())) + past = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat() + queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="external-unavailable", reason="External dependency", retry_not_before=past) + outcomes = queue.poll() + assert [item.status for item in outcomes] == ["RESUMED"] + with PackageReader(outcomes[0].package) as reader: + assert reader.manifest["state"] == State.VALIDATED + assert reader.manifest["routing"]["retry_not_before"] is None + assert reader.manifest["routing"]["retry_count"] == 1 + from viano_bug_analysis.lifecycle.checkpoints import checkpoint_chain + blocked_checkpoint = checkpoint_chain(reader)[1][1] + assert blocked_checkpoint["routing"]["reason"] == "External dependency" + assert blocked_checkpoint["routing"]["retry_not_before"] == past + assert queue.poll() == [] + + +def test_crash_before_acknowledgement_does_not_generate_again(queue, make_package, monkeypatch): + source = deliver(queue, make_package()) + def crash(*args): + raise LifecycleError("SIMULATED_CRASH", "before acknowledgement") + monkeypatch.setattr(queue, "_acknowledge", crash) + assert queue.process(source).status == "ERROR" + assert source.exists() + restarted = InboxProcessor(queue.workspace) + outcome = restarted.process(source) + assert outcome.status == "DUPLICATE" + assert len(list(outcome.package.parent.glob("g*.zip"))) == 1 + + +def test_changed_inbox_file_is_not_deleted(queue, make_package, monkeypatch): + source = deliver(queue, make_package()) + original = queue.store.intake + def change(snapshot): + result = original(snapshot) + source.write_bytes(b"new producer content") + return result + monkeypatch.setattr(queue.store, "intake", change) + outcome = queue.process(source) + assert outcome.status == "ERROR" + assert outcome.code == "SOURCE_CHANGED" + assert source.read_bytes() == b"new producer content" + + +def test_poll_limit_and_continues_after_bad_archive(queue, make_package): + (queue.workspace.inbox / "a-invalid").write_bytes(b"not zip") + deliver(queue, make_package(), "b-valid") + assert len(queue.poll(max_jobs=1)) == 1 + assert queue.poll()[0].status == "ACCEPTED" + with pytest.raises(ValueError): + queue.poll(max_jobs=0) + + +def test_unsafe_source_rejected(queue, make_package): + source = make_package() + assert queue.process(source).code == "UNSAFE_INPUT" + assert source.exists() + + +@pytest.mark.parametrize("change", ["cycle", "identity", "version", "generation", "state", "parent_path"]) +def test_invalid_checkpoint_chain_quarantined(queue, make_package, change, tmp_path): + import hashlib + import zipfile + from viano_bug_analysis.lifecycle.checkpoints import CHECKPOINT_POINTER + admitted = queue.process(deliver(queue, make_package())).package + with zipfile.ZipFile(admitted) as archive: + entries = {name: archive.read(name) for name in archive.namelist()} + manifest = json.loads(entries["manifest.json"]) + ref = manifest["artifacts"][CHECKPOINT_POINTER] + checkpoint = json.loads(entries[ref]) + if change == "cycle": + checkpoint["parent"] = ref + elif change == "identity": + checkpoint["case_id"] = "other" + elif change == "version": + checkpoint["format_version"] = 2 + elif change == "generation": + checkpoint["source_generation"] = 42 + elif change == "state": + checkpoint["state"] = "BUILDING_CASE" + else: + checkpoint["parent"] = "../escape" + entries[ref] = json.dumps(checkpoint).encode() + manifest["integrity"][ref].update(size=len(entries[ref]), sha256=hashlib.sha256(entries[ref]).hexdigest()) + entries["manifest.json"] = json.dumps(manifest).encode() + bad = tmp_path / "bad-checkpoint" + with zipfile.ZipFile(bad, "w") as archive: + for name, data in entries.items(): + archive.writestr(name, data) + other = InboxProcessor(Workspace(tmp_path / "other-service")) + result = other.process(deliver(other, bad)) + assert result.status == "QUARANTINED" + assert result.code == "CHECKPOINT_INVALID" + + +def test_return_without_old_files_is_quarantined(queue, make_package, manifest): + queue.process(deliver(queue, make_package(entries={"unknown": b"required"}))) + manifest["package_generation"] = 2 + manifest["state"] = "VALIDATED" + result = queue.process(deliver(queue, make_package())) + assert result.status == "QUARANTINED" + assert result.code == "CHECKPOINT_HISTORY_LOST" + + +def test_managed_corruption_does_not_quarantine_valid_inbound(queue, make_package): + source = make_package() + admitted = queue.process(deliver(queue, source)).package + admitted.write_bytes(b"corrupt managed package") + inbound = deliver(queue, source) + result = queue.process(inbound) + assert result.status == "ERROR" + assert result.code == "STORAGE_PACKAGE_INVALID" + assert inbound.exists() + + +def test_runtime_stale_on_crash_does_not_block_other_cases(queue, make_package, manifest, monkeypatch): + original_index = queue.store._index + def crash(*args): + raise LifecycleError("SIMULATED_CRASH", "after publication") + monkeypatch.setattr(queue.store, "_index", crash) + source = deliver(queue, make_package()) + assert queue.process(source).status == "ERROR" + monkeypatch.setattr(queue.store, "_index", original_index) + manifest.update(case_id="second", package_id="second-package") + second = deliver(queue, make_package(name="second"), "second-case") + assert queue.process(second).status == "ACCEPTED" + assert queue.process(source).status == "DUPLICATE" + + +def test_deferred_jobs_do_not_consume_admission_limit(queue, make_package, manifest, tmp_path): + queue.process(deliver(queue, make_package())) + future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat() + blocked = queue.store.advance("case-A", "stable-package", State.BLOCKED, step_id="block", reason="Dependency", retry_not_before=future) + other = InboxProcessor(Workspace(tmp_path / "another-service")) + for index in range(3): + deliver(other, blocked, f"a-blocked-{index}") + manifest.update(case_id="new-case", package_id="new-package") + fresh = deliver(other, make_package(name="new"), "z-fresh") + outcomes = other.poll(max_jobs=1) + assert [item.status for item in outcomes] == ["DEFERRED", "DEFERRED", "DEFERRED", "ACCEPTED"] + assert not fresh.exists() diff --git a/tests/test_lifecycle_state.py b/tests/test_lifecycle_state.py new file mode 100644 index 0000000..c5b8939 --- /dev/null +++ b/tests/test_lifecycle_state.py @@ -0,0 +1,29 @@ +import pytest + +from viano_bug_analysis.lifecycle import LifecycleError, State +from viano_bug_analysis.lifecycle.state import ACTIVE, ORDER, check_transition + + +@pytest.mark.parametrize("before,after", list(zip(ORDER, ORDER[1:]))) +def test_forward_transitions(before, after): + check_transition(before, after) + + +@pytest.mark.parametrize("state", list(ACTIVE)) +def test_block_failure_and_progress_checkpoint(state): + check_transition(state, State.BLOCKED) + check_transition(state, State.FAILED_TO_ANALYZE) + check_transition(state, state) + check_transition(State.BLOCKED, state, state) + + +@pytest.mark.parametrize("before,after", [(State.NEW, State.BUILDING_CASE), (State.HANDED_OFF, State.VALIDATED), (State.BLOCKED, State.INTERPRETING_CASE), (State.FAILED_TO_ANALYZE, State.BUILDING_CASE), (State.OUTPUT_READY, State.NEW)]) +def test_forbidden_transitions(before, after): + with pytest.raises(LifecycleError, match="Cannot transition"): + check_transition(before, after) + + +def test_reopening(): + check_transition(State.HANDED_OFF, State.REOPENED) + check_transition(State.FAILED_TO_ANALYZE, State.REOPENED) + check_transition(State.REOPENED, State.BUILDING_CASE) diff --git a/tests/test_lifecycle_storage.py b/tests/test_lifecycle_storage.py new file mode 100644 index 0000000..3d69422 --- /dev/null +++ b/tests/test_lifecycle_storage.py @@ -0,0 +1,59 @@ +import json +import os +import subprocess +import sys + +import pytest + +from viano_bug_analysis.lifecycle import FileLock, LifecycleError, LockBusy, Workspace +from viano_bug_analysis.lifecycle.storage import atomic_json + + +def test_atomic_json_failure_keeps_old_file(tmp_path, monkeypatch): + path = tmp_path / "index.json" + atomic_json(path, {"old": True}) + def fail(*args): + raise OSError("replace failure") + monkeypatch.setattr(os, "replace", fail) + with pytest.raises(LifecycleError): + atomic_json(path, {"new": True}) + assert json.loads(path.read_text()) == {"old": True} + assert not list(tmp_path.glob(".json-*")) + + +def test_case_ids_do_not_become_paths(tmp_path): + workspace = Workspace(tmp_path / "service") + path = workspace.case_directory("../../case", "C:/package") + assert path.is_relative_to(workspace.cases) + assert len(path.name) == 64 + assert len(path.parent.name) == 64 + + +def test_competing_lock_and_release(tmp_path): + path = tmp_path / "lock" + with FileLock(path): + with pytest.raises(LockBusy): + with FileLock(path): + pass + with FileLock(path): + pass + + +def test_lock_released_on_process_exit(tmp_path): + path = tmp_path / "process.lock" + script = "from pathlib import Path; from viano_bug_analysis.lifecycle import FileLock; import os,sys; lock=FileLock(Path(sys.argv[1])); lock.__enter__(); print('locked',flush=True); sys.stdin.readline(); os._exit(17)" + env = dict(os.environ, PYTHONPATH=str(__import__('pathlib').Path(__file__).resolve().parents[1] / "src")) + process = subprocess.Popen([sys.executable, "-c", script, str(path)], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env) + try: + assert process.stdout.readline().strip() == "locked" + with pytest.raises(LockBusy): + with FileLock(path): + pass + process.communicate("exit\n", timeout=10) + assert process.returncode == 17 + with FileLock(path): + pass + finally: + if process.poll() is None: + process.kill() + process.wait()