diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 68fe7bc..ccf4837 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -19,7 +19,13 @@ Thanks for your interest in contributing. 1. Create a feature branch. 2. Make focused changes. 3. Add or update tests. -4. Run checks before opening a PR: +4. Before every commit, run the full test suite and demo smoke checks: + ```bash + uv run pytest tests/ -v + uv run synix demo run templates/01-chatbot-export-synthesis + uv run synix demo run templates/05-batch-build + ``` +5. Run the full release checks before opening a PR: ```bash uv run release ``` diff --git a/README.md b/README.md index f4bcd40..cc4c4ea 100644 --- a/README.md +++ b/README.md @@ -50,9 +50,17 @@ Browse, search, and validate: uvx synix list # all artifacts, grouped by layer uvx synix show final-report # render an artifact uvx synix search "hiking" # full-text search +uvx synix runs list # immutable artifact snapshots for this project +uvx synix runs list --json # machine-readable snapshot history (schema_version + runs[]) uvx synix validate # run declared validators (experimental) ``` +Successful builds record canonical immutable artifact snapshots under `.synix/`. The local `build/` directory still exists as the default compatibility materialization surface for current commands and demos, but it is no longer the source of truth for build history. Projection release state remains in that local surface until the explicit `release`/adapter slice lands. `uvx synix clean` only removes the mutable local surface; it does not delete snapshot history. + +> **Note:** The `.synix` on-disk snapshot format is new in `v0.15.x` and may evolve before `v1.0`. Objects are schema-versioned, and future changes will preserve a compatibility path rather than silently reusing incompatible state. + +> **Note:** Run refs currently use opaque, time-prefixed ids (for example `refs/runs/20260306T082007123456Z-1f2e3d4c`) and remain experimental before `v1.0`. Prefer `uvx synix runs list --json` over scraping the table output; the JSON shape is versioned as `{ "schema_version": 1, "runs": [...] }`. + ## Defining a Pipeline A pipeline is a Python file. Layers are real objects with dependencies expressed as object references. @@ -141,6 +149,7 @@ Pre-built transforms for common agent memory patterns. Import from `synix.transf | `uvx synix build` | Run the pipeline. Only rebuilds what changed | | `uvx synix plan` | Dry-run — show what would build without running transforms | | `uvx synix plan --explain-cache` | Plan with inline cache decision reasons | +| `uvx synix runs list` | List immutable build snapshots recorded under `.synix` | | `uvx synix list [layer]` | List all artifacts, optionally filtered by layer | | `uvx synix show ` | Display an artifact. Resolves by label or ID prefix. `--raw` for JSON | | `uvx synix search ` | Full-text search. `--mode hybrid` for semantic | diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 0b99f08..776b9f5 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -106,6 +106,17 @@ Synix gives you: edit the pipeline definition, run it again, and your same raw d The name "Synix" comes from **synthesis** — the core action of transforming raw information into processed understanding. In chip design, synthesis turns high-level descriptions into gate-level implementations. In software, build systems turn source files into artifacts. Synix does the same for agent memory: sources become artifacts through declared build rules, with full lineage tracking and incremental rebuilds. +### 1.5 Immutable Build History + +As of the snapshotting work in `v0.15.x`, Synix distinguishes between: + +- the **mutable local build surface** (`build/`), which current commands and demos still use as a compatibility materialization target +- the **immutable canonical history** (`.synix/`), which stores content-addressed objects, manifests, snapshots, and refs + +This split is intentional. Build history should survive `synix clean`, and future release targets should be materialized from immutable snapshot state rather than treated as the source of truth. + +The first shipped slice records **artifact snapshots only**. Projection/release state is still materialized into the local build surface until the explicit release/adapter layer lands. + --- ## Part II: Conceptual Grounding diff --git a/docs/snapshots-release-rfc.md b/docs/snapshots-release-rfc.md new file mode 100644 index 0000000..34d7ccd --- /dev/null +++ b/docs/snapshots-release-rfc.md @@ -0,0 +1,885 @@ +# RFC: Immutable Snapshots, Refs, and Projection Release + +**Issue**: [#34](https://github.com/marklubin/synix/issues/34) +**Status**: Proposed +**Baseline**: `v0.15.0` (`93b9c6b`) +**Decision target**: Design approval before implementation + +## Summary + +Synix should move from a mutable `build/` directory model to a git-like snapshot model: + +- `.synix` is the canonical store +- `HEAD` is a first-class ref +- a `manifest` is a closure over artifacts and projections +- a `snapshot` points to one manifest +- `release` materializes projections +- `revert` means releasing an older ref + +This design is the generic platform substrate for reproducibility, diffing, release promotion, rollback, and multi-variation deployment. It is required by LENS, but it is not benchmark-specific. + +LENS here refers to an external benchmark effort that motivated some of the +requirements. The snapshot/ref model in this RFC is intended to be generic +Synix platform functionality, not benchmark-specific behavior. + +The implementation should land in slices. The first mergeable slice is: + +- immutable artifact snapshots +- refs and first-class `HEAD` +- manifest and snapshot objects +- compatibility local `build/` outputs retained for existing commands + +Projection build-state capture and `synix release` remain follow-on work. Until that lands, projections are still compatibility outputs under `build/`, not part of the canonical persisted snapshot closure. + +## Motivation + +At `v0.15.0`, Synix still assumes a single mutable build root: + +- artifact payloads and `manifest.json` are rewritten under `build/` +- `provenance.json` is mutable +- `search.db` is mutable +- `.projection_cache.json` is mutable +- CLI commands default to that mutable root + +That model is sufficient for one live local build. It is not sufficient for: + +- immutable experimental history +- clean diffs between runs +- rollback to known-good states +- multi-target releases +- future projection adapters with incremental release semantics +- checkpointed memory banks for LENS + +## Design Goals + +1. Every successful build creates an immutable logical snapshot. +2. `HEAD` and refs are first-class platform concepts. +3. A manifest represents a complete closure over artifacts and projections. +4. Projections remain distinct from artifacts because they are build targets. +5. Release is separate from build. +6. Revert is release to an older snapshot, not replay of inverse operations. +7. Projection adapters own target-specific reconciliation semantics. +8. The first implementation can start with file-backed projections and full rebuild release strategies without blocking future incremental adapters. + +## Non-Goals + +- No generic platform-level per-input apply/revert op model +- No remote object storage in the first implementation +- No branching UX beyond refs in the first implementation +- No requirement that every projection support incremental release on day one +- No notebook or mutable overlay implementation in this RFC + +## Terminology + +### Object + +Any immutable stored item in `.synix/objects`. + +### Artifact + +Build data such as transcripts, episodes, rollups, summaries, or core-memory blocks. + +### Projection + +A build target that exposes artifacts through a usable output surface such as: + +- search index +- flat-file context doc +- future Postgres or vector database targets + +Projections are not “just another artifact.” They are build targets with different lifecycle semantics. + +### Manifest + +A closure over the exact artifacts and projections for a build state. + +### Snapshot + +An immutable commit-like object that points to one manifest. + +### Ref + +A named pointer to a snapshot. Examples: + +- `HEAD` +- `refs/heads/main` +- `refs/runs/2026-03-06T12-30-11Z` +- `refs/releases/prod` + +### Release + +Materialization of projection targets from a ref or snapshot. + +### Release Receipt + +The provenance record of what was actually materialized, where, and by which adapter. + +## Identity Model + +Synix should use two different identities for two different jobs: + +- `oid` + - object-store id for any stored object in `.synix/objects` +- `artifact_id` + - content identity already used inside artifact semantics + +The object store should not be keyed directly by the current `artifact_id`, because `artifact_id` is already overloaded with content semantics inside the artifact model. + +## Canonical Layout + +```text +project/ + pipeline.py + .synix/ + HEAD + refs/ + heads/ + main + runs/ + 2026-03-06T12-30-11Z + tags/ + v0.15.0 + releases/ + prod + canary + ab-a + ab-b + objects/ + aa/ + bbccddeeff... + 12/ + 34567890ab... + receipts/ + 2026-03-06T12-31-02Z.json +``` + +`HEAD` should be textual and first-class: + +```text +ref: refs/heads/main +``` + +## Object Types + +```text +blob +artifact +projection +manifest +snapshot +release_receipt +``` + +## Object Schemas + +### Blob Object + +Current v0.x implementation stores raw content bytes directly at their content-addressed oid. A separate blob metadata object may still be useful later for richer content descriptors, but artifact `content_oid` values should refer directly to raw bytes. + +```json +{ + "oid": "sha256(raw-bytes)" +} +``` + +### Artifact Object + +```json +{ + "type": "artifact", + "schema_version": 1, + "label": "ep-conv-001", + "artifact_type": "episode", + "artifact_id": "sha256:...", + "content_oid": "oid_bytes_1", + "input_ids": ["sha256:..."], + "prompt_id": "episode_summary_v1", + "model_config": { + "model": "claude-sonnet-4-20250514" + }, + "parent_labels": ["tx-conv-001"], + "metadata": { + "source_conversation_id": "conv-001" + } +} +``` + +### Projection Object + +Projection objects are build targets. + +```json +{ + "type": "projection", + "schema_version": 1, + "name": "memory-index", + "projection_type": "search_index", + "input_oids": ["oid_art_1", "oid_art_2"], + "build_state_oid": "oid_blob_or_state", + "adapter": "search_index", + "release_mode": "full", + "metadata": {} +} +``` + +Future incremental example: + +```json +{ + "type": "projection", + "schema_version": 1, + "name": "customer-db", + "projection_type": "postgres", + "input_oids": ["oid_art_10", "oid_art_11"], + "build_state_oid": "oid_delta_plan", + "adapter": "postgres", + "release_mode": "incremental", + "metadata": { + "schema": "customer_memory" + } +} +``` + +### Manifest Object + +The manifest is the exact closure over artifacts and projections for one build state. + +```json +{ + "type": "manifest", + "schema_version": 1, + "pipeline_name": "monthly-memory", + "pipeline_fingerprint": "sha256:...", + "artifacts": [ + {"label": "tx-conv-001", "oid": "oid_art_1"}, + {"label": "ep-conv-001", "oid": "oid_art_2"}, + {"label": "core", "oid": "oid_art_3"} + ], + "projections": { + "memory-index": "oid_proj_1", + "context-doc": "oid_proj_2" + } +} +``` + +### Snapshot Object + +```json +{ + "type": "snapshot", + "schema_version": 1, + "manifest_oid": "oid_manifest_1", + "parent_snapshot_oids": ["oid_snapshot_prev"], + "created_at": "2026-03-06T12:30:11Z", + "pipeline_name": "monthly-memory" +} +``` + +### Release Receipt + +```json +{ + "type": "release_receipt", + "schema_version": 1, + "ref": "HEAD", + "resolved_snapshot_oid": "oid_snapshot_1", + "manifest_oid": "oid_manifest_1", + "projection_oid": "oid_proj_1", + "adapter": "search_index", + "release_mode": "full", + "target": ".synix/releases/local/current/search.db", + "created_at": "2026-03-06T12:31:02Z" +} +``` + +## Snapshot vs Ref + +A snapshot and a ref solve different problems: + +- `snapshot` + - immutable build state +- `ref` + - movable human-meaningful name pointing to a snapshot + +The relationship is: + +```text +snapshot = what was built +ref = what Synix should mean by default right now +``` + +Git analogy: + +```text +commit ~= snapshot +branch ~= ref +HEAD ~= current ref +``` + +## Refs and Ergonomics + +Example: + +```text +HEAD -> refs/heads/customer-memory +refs/heads/customer-memory -> snapshot S42 +refs/runs/2026-03-06T12:30Z -> snapshot S42 +refs/releases/prod -> snapshot S40 +refs/releases/canary -> snapshot S42 +``` + +This means: + +- `HEAD` is the default build line +- `refs/runs/...` gives immutable run history +- release refs show what is actually materialized + +Build and release are intentionally separate: + +- `synix build` advances the active build ref +- `synix release` advances a chosen release ref + +## Projections As Build Targets + +Projections are the build targets that `synix release` materializes. + +The hierarchy is: + +```text +sources/transforms -> artifacts -> projections -> release +``` + +Artifacts are build data. Projections are the externally usable outputs. + +Examples: + +- flat file +- SQLite search index +- future Postgres target +- future Qdrant or Neo4j target + +## Projection Lifecycle + +Transforms and projections should not share the same lifecycle contract. + +Transforms: + +- produce artifacts + +Projections: + +- build projection state +- release projection state +- reconcile current release to target state +- verify release + +Recommended split: + +```text +ProjectionBuilder +- build(...) +- diff(...) + +ProjectionAdapter +- inspect_release(...) +- plan_release(...) +- apply_release(...) +- verify_release(...) +``` + +Synix may collapse those into fewer types later, but the lifecycle phases must remain explicit. + +## Build, Release, and Revert + +### Build + +`synix build` should: + +1. compute changed artifacts incrementally +2. compute projection build state +3. write immutable objects +4. write a manifest object +5. write a snapshot object +6. move a build ref to the new snapshot + +### Release + +`synix release ` should: + +1. resolve a ref to a snapshot +2. load the manifest +3. select projections +4. dispatch to projection adapters +5. materialize usable targets +6. write release receipts +7. optionally advance a release ref + +### Revert + +`synix revert ` should be a thin wrapper over release of an older snapshot. + +Revert should not mean “replay inverse ops.” + +Revert means: + +- resolve the older target snapshot +- reconcile current released state to that target + +## ASCII: Build Model + +```text +sources/transforms + | + v + artifacts + | + v +projection build state + | + v + manifest + | + v + snapshot + | + v + refs +``` + +## ASCII: Release Model + +```text + HEAD + | + v + refs/heads/main + | + v + snapshot + | + v + manifest + / \ + v v + artifacts projections + | + v + projection adapter plan + | + v + release target + | + v + release receipt +``` + +## Automated Pipeline Workflow + +### Initial Build + +```bash +synix build +``` + +Result: + +```text +artifacts built +projections built +manifest M1 written +snapshot S1 written +refs/heads/main -> S1 +refs/runs/2026-03-06T12:30Z -> S1 +HEAD -> refs/heads/main +``` + +Then: + +```bash +synix release HEAD --to refs/releases/prod +``` + +That materializes the build targets and updates the chosen release ref. + +### Incremental Rebuild + +New source data arrives: + +```bash +synix build +``` + +Incremental build behavior: + +- only changed source artifacts rebuild +- only affected downstream artifacts rebuild +- unchanged artifacts are reused +- new projection build state is computed +- a new manifest and snapshot are written + +Result: + +```text +S1 = older snapshot +S2 = new snapshot +refs/heads/main -> S2 +refs/runs/2026-03-06T12:45Z -> S2 +refs/releases/prod -> S1 +``` + +Then: + +```bash +synix diff refs/releases/prod HEAD +synix release HEAD --to refs/releases/prod +``` + +### Multi-Variation Release + +This model supports multiple release targets: + +```text +refs/heads/main -> S42 +refs/releases/prod -> S40 +refs/releases/canary -> S42 +refs/releases/ab-a -> S41 +refs/releases/ab-b -> S42 +``` + +Example commands: + +```bash +synix release HEAD --to refs/releases/canary +synix release refs/runs/2026-03-05T12:30Z --to refs/releases/ab-a +synix release HEAD --to refs/releases/ab-b +``` + +Clients then choose the release ref they want to read. + +## Incremental Build vs Incremental Release + +There are two different incremental stories: + +### Incremental Build + +- only changed artifacts recompute +- this is already broadly aligned with current Synix behavior + +### Incremental Release + +- only changed projection state is applied to the live target +- this must be owned by the projection adapter + +Example: + +- one new conversation arrives +- build computes one new summary artifact +- snapshot `S2` records the new logical state +- release adapter decides whether to: + - rebuild a whole SQLite file and atomically swap it + - append just one new index entry later + - apply upserts or migrations in a future Postgres adapter + +Synix core should not force one reconciliation strategy across all projections. + +## Adapter Reconciliation Contract + +Synix core should own: + +- refs +- snapshots +- manifests +- release orchestration +- receipts +- diffing old vs target logical state + +Projection adapters should own: + +- full rebuild vs incremental apply +- shadow swap vs upsert vs migration +- target-specific rollback and verification + +Core primitive: + +```text +reconcile(current_release_state, target_projection_state) -> plan +apply(plan) -> receipt +``` + +Not: + +```text +apply op 1 +revert op 1 +apply op 2 +revert op 2 +... +``` + +That second model leaks projection internals into the platform and should be avoided. + +## Projection Semantics By Type + +### Flat File + +- build: final bytes +- release: copy or symlink +- revert: re-copy older bytes + +### SQLite Search Index + +- build: index image or equivalent build state +- release: full atomic swap initially +- later: optional incremental reconcile + +### Future Postgres Target + +- build: logical desired state, delta plan, or migration plan +- release: adapter applies incremental change set +- revert: adapter reconciles back to older target state + +The RFC intentionally does not require incremental release for every projection in the first implementation. + +## Streaming / Near-Real-Time Compatibility + +This model still works if incremental rebuilds move closer to streaming updates. + +The model becomes: + +```text +event stream -> incremental builder -> snapshots -> release refs -> clients +``` + +For higher-rate streaming, Synix may later choose: + +1. micro-snapshots +2. streaming build refs plus periodic durable snapshots + +That is a policy layer on top of the same snapshot/ref model. The model remains valid as long as build refs, snapshot refs, and release refs remain distinct. + +## Notebook / Mutable Overlay + +This RFC does not implement notebook semantics, but the design intentionally leaves room for them. + +The intended model is: + +- immutable build lane + - artifacts + - projections + - snapshots + - release refs +- mutable runtime lane + - notebook overlay + - append-only journal + - periodic fold-back into immutable snapshots + +Notebook writes should bypass full snapshot/projection/release machinery, but not bypass provenance, audit logging, scoping, or checkpoint semantics. + +That design should be handled in a separate RFC. + +## Compatibility and Migration + +The first implementation should prioritize an incremental migration path: + +1. Add `.synix` as the canonical store and ref namespace. +2. Keep existing CLI ergonomics where possible by resolving `HEAD` by default. +3. Support file-backed release for the projections Synix already has. +4. Preserve enough compatibility that a user can adopt the new model without destructive migration. + +The implementation strategy may choose a temporary bridge layer from the old `build/` layout to the new `.synix` store, but the target mental model should be the one in this RFC. + +## Command Semantics + +### Build + +```bash +synix build +``` + +- builds artifacts incrementally +- computes projection build state +- writes manifest and snapshot objects +- advances the active build ref +- prints the new snapshot and run ref + +### Show / Search / List + +By default these should resolve `HEAD`: + +```bash +synix list +synix show ep-conv-001 +synix search "anthropic" +``` + +They should also be able to target a specific ref or snapshot: + +```bash +synix list --ref refs/runs/2026-03-06T12:30Z +synix search "anthropic" --ref refs/releases/prod +``` + +### Diff + +```bash +synix diff HEAD refs/releases/prod +``` + +The long-term model should diff refs or snapshots, not mutable build directories. + +### Release + +```bash +synix release HEAD --to refs/releases/prod +synix release refs/runs/2026-03-06T12:30Z --to refs/releases/canary +``` + +### Revert + +```bash +synix revert refs/runs/2026-03-06T12:30Z --to refs/releases/prod +``` + +This should be equivalent to release of an older target. + +## First Implementation Scope + +1. object store under `.synix/objects` +2. first-class refs and `HEAD` +3. immutable manifest and snapshot objects +4. build writing snapshots and moving build refs +5. file-backed projection release +6. release receipts +7. ref/snapshot diffing +8. adapter interface that leaves room for incremental reconcile later + +## Out of Scope For First Implementation + +- generic notebook overlay +- remote object storage +- true incremental release for every projection type +- branching UX beyond refs +- external database adapters +- projection DAG redesign beyond what is necessary for release + +## Test Strategy + +The snapshotting feature must match Synix’s existing test discipline: + +- `tmp_path` for all filesystem tests +- `CliRunner` for CLI behavior +- mocked LLMs in unit and integration tests +- no shared state +- explicit failure-mode coverage +- every functional behavior change gets e2e coverage + +### Unit Tests + +Add: + +- `tests/unit/test_object_store.py` + - object write and read roundtrip + - stable oid computation + - invalid object rejection + +- `tests/unit/test_refs.py` + - symbolic `HEAD` + - direct ref resolution + - missing ref failure + - atomic ref move + +- `tests/unit/test_manifest_snapshot.py` + - manifest closure serialization + - snapshot creation + - parent snapshot linkage + +- `tests/unit/test_release_receipts.py` + - release receipt schema + - ref and snapshot provenance fields + - target metadata persistence + +- `tests/unit/test_projection_release_contract.py` + - file-backed adapter planning + - file-backed adapter apply + - verify and receipt behavior + +- `tests/unit/test_legacy_layout_resolver.py` + - old mutable layout detection + - `HEAD` default resolution + - compatibility lookup during migration + +### Integration Tests + +Add: + +- `tests/integration/test_snapshot_build.py` + - build writes objects, manifest, snapshot, and refs + +- `tests/integration/test_incremental_snapshot_rebuild.py` + - only affected artifacts rebuild when one source changes + - unchanged artifacts are reused + - a new snapshot is created + +- `tests/integration/test_projection_release_local.py` + - release file-backed projections from `HEAD` + +- `tests/integration/test_revert_release.py` + - release newer snapshot, then revert to an older snapshot + +- `tests/integration/test_multi_release_refs.py` + - `prod`, `canary`, and A/B release refs can diverge safely + +### End-to-End Tests + +Add: + +- `tests/e2e/test_snapshot_flow.py` + - build -> build again -> diff -> release -> revert + +Update: + +- `tests/e2e/test_demo_flow.py` + - stop assuming mutable root-level `manifest.json`, `provenance.json`, and `search.db` + - assert ref and snapshot semantics instead + +- other affected demo e2e tests that assume direct mutable build-root layout + +### Must-Have Failure Tests + +- failed build does not advance the build ref +- failed release does not advance the release ref +- older snapshots remain readable after later builds +- revert to an older ref works after a newer release +- multiple release refs do not interfere with each other +- release receipts are still written or rolled back consistently on partial failures + +## Documentation Deliverables + +Before closing the implementation: + +- update `docs/entity-model.md` +- update CLI docs and examples +- update pipeline API docs if build/release commands change usage semantics +- record demo/template follow-ons for snapshot/release behavior + +## Open Questions + +These should be answered during implementation design, but do not block approval of the model: + +1. Should build automatically materialize file-backed projections into cached build state, or should release handle all final target materialization? +2. How much compatibility should be preserved for legacy `build/` root lookup during the transition? +3. Should `synix diff` accept refs first and legacy build dirs second, or should both remain long term? + +## Decision Summary + +This RFC recommends: + +- first-class `HEAD` and refs +- immutable snapshots +- manifests as closures over artifacts and projections +- projections as build targets with distinct lifecycle semantics +- separate build and release phases +- adapter-owned reconciliation logic +- platform-owned release and revert semantics + +That is the cleanest long-term substrate for Synix and the right platform model for LENS. diff --git a/src/synix/build/artifacts.py b/src/synix/build/artifacts.py index b6df121..9729287 100644 --- a/src/synix/build/artifacts.py +++ b/src/synix/build/artifacts.py @@ -66,8 +66,12 @@ def _save_manifest(self) -> None: def save_artifact(self, artifact: Artifact, layer_name: str, layer_level: int) -> None: """Save an artifact to the build directory.""" + if not isinstance(artifact.content, str): + msg = f"artifact {artifact.label!r} content must be a string, got {type(artifact.content).__name__}" + raise TypeError(msg) + # Ensure artifact ID (content hash) is computed - if not artifact.artifact_id and artifact.content: + if not artifact.artifact_id: artifact.artifact_id = f"sha256:{hashlib.sha256(artifact.content.encode()).hexdigest()}" # Create layer directory @@ -139,6 +143,10 @@ def get_artifact_id(self, label: str) -> str | None: return None return entry["artifact_id"] + def iter_entries(self) -> dict[str, dict]: + """Return a shallow copy of the manifest entries keyed by label.""" + return dict(self._manifest) + def resolve_prefix(self, prefix: str) -> str | None: """Resolve a prefix to a full label (git-like semantics). diff --git a/src/synix/build/object_store.py b/src/synix/build/object_store.py new file mode 100644 index 0000000..25c44fb --- /dev/null +++ b/src/synix/build/object_store.py @@ -0,0 +1,333 @@ +"""Immutable object storage for snapshots and related metadata.""" + +from __future__ import annotations + +import codecs +import hashlib +import json +import os +import re +import shutil +import tempfile +from pathlib import Path +from typing import Any + +SCHEMA_VERSION = 1 +_OID_RE = re.compile(r"^[0-9a-f]{64}$") + +_REQUIRED_FIELDS: dict[str, set[str]] = { + "artifact": { + "type", + "schema_version", + "label", + "artifact_type", + "artifact_id", + "content_oid", + "input_ids", + "metadata", + }, + "projection": { + "type", + "schema_version", + "name", + "projection_type", + "input_oids", + "build_state_oid", + "adapter", + "release_mode", + }, + "manifest": {"type", "schema_version", "pipeline_name", "pipeline_fingerprint", "artifacts", "projections"}, + "snapshot": { + "type", + "schema_version", + "manifest_oid", + "parent_snapshot_oids", + "created_at", + "pipeline_name", + "run_id", + }, + "release_receipt": { + "type", + "schema_version", + "resolved_snapshot_oid", + "manifest_oid", + "projection_oid", + "adapter", + "release_mode", + "target", + "created_at", + }, +} + +_FIELD_TYPES: dict[str, dict[str, type | tuple[type, ...]]] = { + "artifact": { + "label": str, + "artifact_type": str, + "artifact_id": str, + "content_oid": str, + "input_ids": list, + "metadata": dict, + }, + "projection": { + "name": str, + "projection_type": str, + "input_oids": list, + "build_state_oid": str, + "adapter": str, + "release_mode": str, + }, + "manifest": { + "pipeline_name": str, + "pipeline_fingerprint": str, + "artifacts": list, + "projections": dict, + }, + "snapshot": { + "manifest_oid": str, + "parent_snapshot_oids": list, + "created_at": str, + "pipeline_name": str, + "run_id": str, + }, +} + +_OPTIONAL_FIELD_TYPES: dict[str, dict[str, type | tuple[type, ...]]] = { + "artifact": { + "prompt_id": (str, type(None)), + "model_config": (dict, type(None)), + "created_at": str, + "parent_labels": list, + }, +} + + +def _atomic_write_bytes(path: Path, data: bytes) -> None: + fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp") + try: + view = memoryview(data) + while view: + written = os.write(fd, view) + view = view[written:] + os.fsync(fd) + os.close(fd) + os.replace(tmp, str(path)) + except BaseException: + os.close(fd) + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def _canonical_json_bytes(payload: dict[str, Any]) -> bytes: + return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + + +def _validate_object_payload(payload: dict[str, Any], *, allow_older_schema: bool = False) -> None: + object_type = payload.get("type") + if not isinstance(object_type, str) or not object_type: + msg = "object payload must include a non-empty string 'type'" + raise ValueError(msg) + + schema_version = payload.get("schema_version") + if not isinstance(schema_version, int): + msg = "object payload must include integer schema_version" + raise ValueError(msg) + if allow_older_schema: + if schema_version > SCHEMA_VERSION: + msg = f"object payload schema_version={schema_version} is newer than supported={SCHEMA_VERSION}" + raise ValueError(msg) + elif schema_version != SCHEMA_VERSION: + msg = f"object payload must include schema_version={SCHEMA_VERSION}" + raise ValueError(msg) + + required = _REQUIRED_FIELDS.get(object_type) + if required is None: + msg = f"unsupported object type: {object_type!r}" + raise ValueError(msg) + + missing = sorted(required.difference(payload)) + if missing: + msg = f"object payload for type {object_type!r} is missing required fields: {', '.join(missing)}" + raise ValueError(msg) + + for field_name, expected_type in _FIELD_TYPES.get(object_type, {}).items(): + value = payload.get(field_name) + if not isinstance(value, expected_type): + msg = ( + f"object payload for type {object_type!r} field {field_name!r} " + f"must be of type {expected_type}, got {type(value)}" + ) + raise ValueError(msg) + + for field_name, expected_type in _OPTIONAL_FIELD_TYPES.get(object_type, {}).items(): + if field_name not in payload: + continue + value = payload[field_name] + if not isinstance(value, expected_type): + msg = ( + f"object payload for type {object_type!r} field {field_name!r} " + f"must be of type {expected_type}, got {type(value)}" + ) + raise ValueError(msg) + + if object_type == "manifest": + artifacts = payload["artifacts"] + for idx, entry in enumerate(artifacts): + if not isinstance(entry, dict): + msg = f"manifest artifacts[{idx}] must be an object, got {type(entry)}" + raise ValueError(msg) + label = entry.get("label") + oid = entry.get("oid") + if not isinstance(label, str) or not label: + msg = f"manifest artifacts[{idx}] must include non-empty string label" + raise ValueError(msg) + if not isinstance(oid, str) or not _OID_RE.fullmatch(oid): + msg = f"manifest artifacts[{idx}] must include valid oid" + raise ValueError(msg) + + +class ObjectStore: + """Content-addressed storage rooted at .synix/objects.""" + + def __init__(self, synix_dir: str | Path): + self.synix_dir = Path(synix_dir) + self.objects_dir = self.synix_dir / "objects" + self.objects_dir.mkdir(parents=True, exist_ok=True) + + def _path_for_oid(self, oid: str) -> Path: + prefix = oid[:2] + rest = oid[2:] + return self.objects_dir / prefix / rest + + def put_bytes(self, data: bytes) -> str: + """Store raw bytes and return the content-addressed oid.""" + oid = hashlib.sha256(data).hexdigest() + path = self._path_for_oid(oid) + if path.exists(): + return oid + path.parent.mkdir(parents=True, exist_ok=True) + _atomic_write_bytes(path, data) + return oid + + def put_file(self, path: str | Path) -> tuple[str, int]: + """Store a file's bytes by content hash without reading the whole file into memory.""" + source_path = Path(path) + digest = hashlib.sha256() + size_bytes = 0 + with source_path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + size_bytes += len(chunk) + digest.update(chunk) + + oid = digest.hexdigest() + target_path = self._path_for_oid(oid) + if target_path.exists(): + return oid, size_bytes + + target_path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(dir=target_path.parent, suffix=".tmp") + try: + with os.fdopen(fd, "wb") as tmp_handle, source_path.open("rb") as source_handle: + shutil.copyfileobj(source_handle, tmp_handle, length=1024 * 1024) + tmp_handle.flush() + os.fsync(tmp_handle.fileno()) + os.replace(tmp, str(target_path)) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + return oid, size_bytes + + def put_text(self, text: str, *, encoding: str = "utf-8") -> tuple[str, int]: + """Store text as encoded bytes in one pass. + + Text is incrementally encoded while writing to a temporary file so the + digest and persisted bytes are computed from the same byte stream. + """ + if not isinstance(text, str): + msg = f"text must be a string, got {type(text).__name__}" + raise TypeError(msg) + + digest = hashlib.sha256() + size_bytes = 0 + fd, tmp = tempfile.mkstemp(dir=self.objects_dir, suffix=".tmp") + try: + encoder = codecs.getincrementalencoder(encoding)() + with os.fdopen(fd, "wb") as handle: + for start in range(0, len(text), 64 * 1024): + chunk = text[start : start + 64 * 1024] + encoded = encoder.encode(chunk) + if encoded: + size_bytes += len(encoded) + digest.update(encoded) + handle.write(encoded) + + final_bytes = encoder.encode("", final=True) + if final_bytes: + size_bytes += len(final_bytes) + digest.update(final_bytes) + handle.write(final_bytes) + + handle.flush() + os.fsync(handle.fileno()) + + oid = digest.hexdigest() + path = self._path_for_oid(oid) + if path.exists(): + os.unlink(tmp) + return oid, size_bytes + + path.parent.mkdir(parents=True, exist_ok=True) + os.replace(tmp, str(path)) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + return oid, size_bytes + + def get_bytes(self, oid: str) -> bytes: + """Load raw bytes by oid.""" + path = self._path_for_oid(oid) + try: + return path.read_bytes() + except OSError as exc: + msg = f"failed to read object bytes for oid {oid} at {path}: {exc}" + raise OSError(msg) from exc + + def put_json(self, payload: dict[str, Any]) -> str: + """Store canonical JSON and return the content-addressed oid.""" + _validate_object_payload(payload, allow_older_schema=False) + encoded = _canonical_json_bytes(payload) + oid = hashlib.sha256(encoded).hexdigest() + path = self._path_for_oid(oid) + if path.exists(): + return oid + path.parent.mkdir(parents=True, exist_ok=True) + _atomic_write_bytes(path, encoded) + return oid + + def get_json(self, oid: str) -> dict[str, Any]: + """Load a structured JSON object by oid.""" + path = self._path_for_oid(oid) + try: + raw_text = path.read_text(encoding="utf-8") + except OSError as exc: + msg = f"failed to read object json for oid {oid} at {path}: {exc}" + raise OSError(msg) from exc + try: + payload = json.loads(raw_text) + except json.JSONDecodeError as exc: + msg = f"object {oid} at {path} is not valid JSON: {exc}" + raise ValueError(msg) from exc + if not isinstance(payload, dict): + msg = f"object {oid} is not a JSON object" + raise ValueError(msg) + _validate_object_payload(payload, allow_older_schema=True) + return payload diff --git a/src/synix/build/pipeline.py b/src/synix/build/pipeline.py index 3bd64b4..b893c09 100644 --- a/src/synix/build/pipeline.py +++ b/src/synix/build/pipeline.py @@ -7,6 +7,7 @@ from pathlib import Path from synix.build.dag import compute_levels, resolve_build_order +from synix.build.refs import synix_dir_for_build_dir from synix.core.models import Pipeline, Source @@ -41,6 +42,11 @@ def load_pipeline(path: str) -> Pipeline: raise ValueError(f"Pipeline module {path} must define a 'pipeline' variable") if not isinstance(pipeline, Pipeline): raise TypeError(f"'pipeline' variable must be a Pipeline instance, got {type(pipeline)}") + if pipeline.synix_dir is None: + build_dir = Path(pipeline.build_dir) + if not build_dir.is_absolute(): + build_dir = (filepath.parent / build_dir).resolve() + pipeline.synix_dir = str(synix_dir_for_build_dir(build_dir)) validate_pipeline(pipeline) return pipeline diff --git a/src/synix/build/refs.py b/src/synix/build/refs.py new file mode 100644 index 0000000..1e89097 --- /dev/null +++ b/src/synix/build/refs.py @@ -0,0 +1,151 @@ +"""Git-like refs and HEAD management for Synix snapshots.""" + +from __future__ import annotations + +import re +from pathlib import Path + +from synix.core.errors import atomic_write + +HEAD_FILENAME = "HEAD" +DEFAULT_HEAD_REF = "refs/heads/main" +MAX_REF_DEPTH = 16 + +_REF_RE = re.compile(r"^refs(?:/[A-Za-z0-9._-]+)+$") +_OID_RE = re.compile(r"^[0-9a-f]{64}$") + + +def synix_dir_for_build_dir(build_dir: str | Path, *, configured_synix_dir: str | Path | None = None) -> Path: + """Resolve the canonical .synix directory for a build. + + Write path precedence: + - explicit configured_synix_dir + - persistent sibling store (`/.synix`) + + Read-side helpers also honor an existing nested `build/.synix` store for + compatibility with experiments, but new callers should default to the + sibling store so `synix clean` does not delete snapshot history. + """ + if configured_synix_dir is not None: + return Path(configured_synix_dir).resolve() + + build_path = Path(build_dir).resolve() + legacy = build_path.parent / ".synix" + nested = build_path / ".synix" + if legacy.exists() and nested.exists(): + msg = ( + f"ambiguous snapshot store resolution for {build_path}: " + f"both {legacy} and {nested} exist; pass an explicit synix_dir" + ) + raise ValueError(msg) + if legacy.exists(): + return legacy + if nested.exists(): + return nested + return legacy + + +def _validate_ref_name(ref_name: str) -> None: + if ref_name == "HEAD": + return + if not _REF_RE.fullmatch(ref_name): + msg = f"invalid ref name: {ref_name!r}" + raise ValueError(msg) + + +def _validate_oid(oid: str) -> None: + if not _OID_RE.fullmatch(oid): + msg = f"invalid oid: {oid!r}" + raise ValueError(msg) + + +class RefStore: + """Manage refs and HEAD under .synix.""" + + def __init__(self, synix_dir: str | Path): + self.synix_dir = Path(synix_dir) + self.refs_dir = self.synix_dir / "refs" + self.head_path = self.synix_dir / HEAD_FILENAME + self.refs_dir.mkdir(parents=True, exist_ok=True) + + def ensure_head(self, default_ref: str = DEFAULT_HEAD_REF) -> str: + """Create HEAD if needed and return its target ref.""" + _validate_ref_name(default_ref) + if not self.head_path.exists(): + self.head_path.parent.mkdir(parents=True, exist_ok=True) + atomic_write(self.head_path, f"ref: {default_ref}\n") + return self.read_head_target() + + def read_head_target(self) -> str: + """Return the ref target that HEAD points to.""" + raw = self.head_path.read_text(encoding="utf-8").strip() + if not raw.startswith("ref: "): + msg = f"HEAD has invalid contents: {raw!r}" + raise ValueError(msg) + target = raw[5:] + _validate_ref_name(target) + return target + + def write_head(self, target_ref: str) -> None: + """Update HEAD to a symbolic ref target.""" + _validate_ref_name(target_ref) + atomic_write(self.head_path, f"ref: {target_ref}\n") + + def _ref_path(self, ref_name: str) -> Path: + if ref_name == "HEAD": + return self.head_path + _validate_ref_name(ref_name) + return self.synix_dir / ref_name + + def write_ref(self, ref_name: str, oid: str) -> None: + """Update a direct ref to an oid.""" + _validate_oid(oid) + path = self._ref_path(ref_name) + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write(path, f"{oid}\n") + + def read_ref(self, ref_name: str) -> str | None: + """Read a direct or symbolic ref and return the resolved oid.""" + return self._read_ref(ref_name, seen=[]) + + def _read_ref(self, ref_name: str, *, seen: list[str]) -> str | None: + if len(seen) >= MAX_REF_DEPTH: + msg = f"ref resolution exceeded max depth: {' -> '.join(seen + [ref_name])}" + raise ValueError(msg) + if ref_name in seen: + msg = f"ref cycle detected: {' -> '.join(seen + [ref_name])}" + raise ValueError(msg) + + if ref_name == "HEAD": + if not self.head_path.exists(): + return None + return self._read_ref(self.read_head_target(), seen=seen + [ref_name]) + + path = self._ref_path(ref_name) + if not path.exists(): + return None + + value = path.read_text(encoding="utf-8").strip() + if value.startswith("ref: "): + target = value[5:] + _validate_ref_name(target) + return self._read_ref(target, seen=seen + [ref_name]) + + _validate_oid(value) + return value or None + + def iter_refs(self, prefix: str) -> list[tuple[str, str]]: + """List refs under a prefix along with their resolved oid values.""" + _validate_ref_name(prefix) + root = self.synix_dir / prefix + if not root.exists(): + return [] + refs: list[tuple[str, str]] = [] + for path in sorted(root.rglob("*")): + if path.is_dir(): + continue + rel = path.relative_to(self.synix_dir).as_posix() + resolved = self.read_ref(rel) + if resolved is not None: + refs.append((rel, resolved)) + return refs diff --git a/src/synix/build/runner.py b/src/synix/build/runner.py index 950be33..f255043 100644 --- a/src/synix/build/runner.py +++ b/src/synix/build/runner.py @@ -4,6 +4,7 @@ import copy import hashlib +import inspect import json import logging import time @@ -16,6 +17,7 @@ from synix.build.fingerprint import Fingerprint, compute_build_fingerprint from synix.build.projections import FlatFileProjection, get_projection from synix.build.provenance import ProvenanceTracker +from synix.build.snapshots import BuildTransaction, commit_build_snapshot, start_build_transaction from synix.core.logging import SynixLogger, Verbosity from synix.core.models import ( Artifact, @@ -62,6 +64,11 @@ class RunResult: projection_stats: list[ProjectionStats] = field(default_factory=list) run_log: dict = field(default_factory=dict) validation: object | None = None # ValidationResult when validators are declared + snapshot_oid: str | None = None + manifest_oid: str | None = None + head_ref: str | None = None + run_ref: str | None = None + synix_dir: str | None = None def run( @@ -103,6 +110,7 @@ def run( build_dir=build_dir, progress=progress, ) + snapshot_txn = start_build_transaction(pipeline, build_dir, slogger.run_log.run_id) # Resolve build order build_order = resolve_build_order(pipeline) @@ -134,6 +142,13 @@ def run( artifact.metadata["layer_level"] = layer._level store.save_artifact(artifact, layer.name, layer._level) provenance.record(artifact.label, parent_labels=[], prompt_id=None, model_config=None) + _record_snapshot_artifact( + snapshot_txn, + artifact, + layer_name=layer.name, + layer_level=layer._level, + parent_labels=[], + ) stats.built += 1 slogger.artifact_built(layer.name, artifact.label) @@ -156,6 +171,13 @@ def run( art.metadata["layer_name"] = layer.name art.metadata["layer_level"] = layer._level layer_built.append(art) + _record_snapshot_artifact( + snapshot_txn, + art, + layer_name=layer.name, + layer_level=layer._level, + parent_labels=_snapshot_parent_labels(art, inputs, provenance), + ) stats.cached += 1 slogger.artifact_cached(layer.name, art.label) else: @@ -164,8 +186,14 @@ def run( transform_config["_layer_name"] = layer.name def _save_artifact( - artifact: Artifact, *, _layer=layer, _transform_fp=transform_fp, _inputs=inputs + artifact: Artifact, + *, + parent_inputs: list[Artifact] | None = None, + _layer=layer, + _transform_fp=transform_fp, + _inputs=inputs, ) -> None: + effective_inputs = parent_inputs if parent_inputs is not None else _inputs # Compute per-artifact build fingerprint build_fp = compute_build_fingerprint(_transform_fp, artifact.input_ids) @@ -181,13 +209,20 @@ def _save_artifact( artifact.metadata["build_fingerprint"] = build_fp.to_dict() artifact.metadata["transform_fingerprint"] = _transform_fp.to_dict() store.save_artifact(artifact, _layer.name, _layer._level) - parent_labels = _get_parent_labels(artifact, _inputs) + parent_labels = _provenance_parent_labels(artifact, effective_inputs, provenance) provenance.record( artifact.label, parent_labels=parent_labels, prompt_id=artifact.prompt_id, model_config=artifact.model_config, ) + _record_snapshot_artifact( + snapshot_txn, + artifact, + layer_name=_layer.name, + layer_level=_layer._level, + parent_labels=parent_labels, + ) layer_built.append(artifact) stats.built += 1 slogger.artifact_built(_layer.name, artifact.label) @@ -197,14 +232,28 @@ def _save_artifact( cached.metadata["layer_name"] = _layer.name cached.metadata["layer_level"] = _layer._level layer_built.append(cached) + _record_snapshot_artifact( + snapshot_txn, + cached, + layer_name=_layer.name, + layer_level=_layer._level, + parent_labels=_snapshot_parent_labels(cached, effective_inputs, provenance), + ) else: layer_built.append(artifact) + _record_snapshot_artifact( + snapshot_txn, + artifact, + layer_name=_layer.name, + layer_level=_layer._level, + parent_labels=_snapshot_parent_labels(artifact, effective_inputs, provenance), + ) stats.cached += 1 slogger.artifact_cached(_layer.name, artifact.label) - def _on_batch_complete(artifacts: list[Artifact]) -> None: + def _on_batch_complete(artifacts: list[Artifact], unit_inputs: list[Artifact]) -> None: for artifact in artifacts: - _save_artifact(artifact) + _save_artifact(artifact, parent_inputs=unit_inputs) # Build lookup of cached artifacts by sorted input_ids for per-unit cache checks existing_artifacts = store.list_artifacts(layer.name) @@ -217,11 +266,18 @@ def _on_batch_complete(artifacts: list[Artifact]) -> None: key = tuple(sorted(art.input_ids)) cached_by_inputs.setdefault(key, []).append(art) - def _on_cached(cached_arts: list[Artifact]) -> None: + def _on_cached(cached_arts: list[Artifact], unit_inputs: list[Artifact]) -> None: for cached_art in cached_arts: cached_art.metadata["layer_name"] = layer.name cached_art.metadata["layer_level"] = layer._level layer_built.append(cached_art) + _record_snapshot_artifact( + snapshot_txn, + cached_art, + layer_name=layer.name, + layer_level=layer._level, + parent_labels=_snapshot_parent_labels(cached_art, unit_inputs, provenance), + ) stats.cached += 1 slogger.artifact_cached(layer.name, cached_art.label) @@ -245,12 +301,12 @@ def _on_cached(cached_arts: list[Artifact]) -> None: unit_input_ids = tuple(sorted(a.artifact_id for a in unit_inputs if a.artifact_id)) cached_arts = cached_by_inputs.get(unit_input_ids) if cached_arts: - _on_cached(cached_arts) + _on_cached(cached_arts, unit_inputs) continue merged_config = {**transform_config, **config_extras} new_artifacts = layer.execute(unit_inputs, merged_config) for artifact in new_artifacts: - _save_artifact(artifact) + _save_artifact(artifact, parent_inputs=unit_inputs) layer_artifacts[layer.name] = layer_built @@ -273,6 +329,17 @@ def _on_cached(cached_arts: list[Artifact]) -> None: result.validation = run_validators(pipeline, store, provenance) + # Non-validating builds still record a snapshot; validating builds only + # advance snapshot refs when all validators pass. + if result.validation is None or result.validation.passed: + snapshot_txn.assert_complete(layer_artifacts) + snapshot_info = commit_build_snapshot(snapshot_txn) + result.snapshot_oid = snapshot_info["snapshot_oid"] + result.manifest_oid = snapshot_info["manifest_oid"] + result.head_ref = snapshot_info["head_ref"] + result.run_ref = snapshot_info["run_ref"] + result.synix_dir = snapshot_info["synix_dir"] + result.total_time = time.time() - start_time slogger.run_finish(result.total_time) result.run_log = slogger.run_log.to_dict() @@ -287,6 +354,64 @@ def _build_source_config(pipeline: Pipeline, source: Source, src_dir: str) -> di return config +def _record_snapshot_artifact( + snapshot_txn: BuildTransaction, + artifact: Artifact, + *, + layer_name: str, + layer_level: int, + parent_labels: list[str], +) -> None: + """Record the canonical artifact state used by this run into the snapshot transaction.""" + snapshot_txn.record_artifact( + artifact, + layer_name=layer_name, + layer_level=layer_level, + parent_labels=parent_labels, + ) + + +def _snapshot_parent_labels( + artifact: Artifact, + inputs: list[Artifact], + provenance: ProvenanceTracker, +) -> list[str]: + """Parent labels for immutable snapshots. + + Snapshots should not invent broad parent sets when an artifact explicitly + declared inputs but they could not be resolved unambiguously. + """ + derived = _get_parent_labels(artifact, inputs) + if derived: + return derived + stored = provenance.get_parents(artifact.label) + if stored: + return stored + if artifact.input_ids: + return [] + return [inp.label for inp in inputs] + + +def _provenance_parent_labels( + artifact: Artifact, + inputs: list[Artifact], + provenance: ProvenanceTracker, +) -> list[str]: + """Parent labels for the legacy provenance surface. + + Provenance records keep the pre-snapshot best-effort behavior so existing + lineage commands remain informative even when transforms omit explicit + input_ids for aggregate artifacts. + """ + derived = _get_parent_labels(artifact, inputs) + if derived: + return derived + stored = provenance.get_parents(artifact.label) + if stored: + return stored + return [inp.label for inp in inputs] + + def _build_transform_config(pipeline: Pipeline, layer: Transform, src_dir: str, build_dir: Path) -> dict: """Build config dict for a Transform layer.""" base_llm_config = dict(pipeline.llm_config) if pipeline.llm_config else {} @@ -359,15 +484,18 @@ def _layer_fully_cached( def _get_parent_labels(artifact: Artifact, inputs: list[Artifact]) -> list[str]: """Determine parent labels based on input IDs (hashes).""" - hash_to_label: dict[str, str] = {} + hash_to_labels: dict[str, list[str]] = {} for inp in inputs: if inp.artifact_id: - hash_to_label[inp.artifact_id] = inp.label + hash_to_labels.setdefault(inp.artifact_id, []).append(inp.label) parents = [] for h in artifact.input_ids: - if h in hash_to_label: - parents.append(hash_to_label[h]) + labels = hash_to_labels.get(h, []) + if len(labels) == 1: + parents.append(labels[0]) + elif len(labels) > 1: + return [] if not parents and inputs: parents = [inp.label for inp in inputs] @@ -394,6 +522,33 @@ def _execute_transform_concurrent( Units whose input_ids match ``cached_by_inputs`` are skipped (reported via ``on_cached``) and never submitted to the thread pool. """ + + def _invoke_callback(callback, artifacts: list[Artifact], unit_inputs: list[Artifact]) -> None: + if callback is None: + return + try: + signature = inspect.signature(callback) + except (TypeError, ValueError): + callback(artifacts, unit_inputs) + return + + params = list(signature.parameters.values()) + if any(param.kind == inspect.Parameter.VAR_POSITIONAL for param in params): + callback(artifacts, unit_inputs) + return + + positional = [ + param + for param in params + if param.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD) + ] + if len(positional) >= 2: + callback(artifacts, unit_inputs) + elif len(positional) == 1: + callback(artifacts) + else: + callback() + # Filter out cached units before submitting to pool units_to_run: list[tuple[int, list[Artifact], dict]] = [] for i, (unit_inputs, config_extras) in enumerate(units): @@ -401,8 +556,7 @@ def _execute_transform_concurrent( unit_input_ids = tuple(sorted(a.artifact_id for a in unit_inputs if a.artifact_id)) cached_arts = cached_by_inputs.get(unit_input_ids) if cached_arts: - if on_cached: - on_cached(cached_arts) + _invoke_callback(on_cached, cached_arts, unit_inputs) continue units_to_run.append((i, unit_inputs, config_extras)) @@ -454,8 +608,8 @@ def _run_one(index: int, unit_inputs: list[Artifact], config_extras: dict) -> tu try: _, artifacts = future.result() results[idx] = artifacts - if on_complete: - on_complete(artifacts) + _orig_idx, unit_inputs, _config_extras = units_to_run[idx] + _invoke_callback(on_complete, artifacts, unit_inputs) except Exception as exc: results[idx] = exc if first_error is None: diff --git a/src/synix/build/snapshots.py b/src/synix/build/snapshots.py new file mode 100644 index 0000000..a9c5ab5 --- /dev/null +++ b/src/synix/build/snapshots.py @@ -0,0 +1,417 @@ +"""Snapshot creation and lookup for Synix builds.""" + +from __future__ import annotations + +import hashlib +import inspect +import json +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import UTC, datetime +from pathlib import Path +from threading import Lock +from typing import Any + +from synix.build.object_store import SCHEMA_VERSION, ObjectStore +from synix.build.refs import DEFAULT_HEAD_REF, RefStore, synix_dir_for_build_dir +from synix.core.errors import atomic_write +from synix.core.models import Artifact, Pipeline + + +def _object(object_type: str, **fields: Any) -> dict[str, Any]: + payload = { + "type": object_type, + "schema_version": SCHEMA_VERSION, + } + payload.update(fields) + return payload + + +def _sanitize_llm_config(config: dict[str, Any]) -> dict[str, Any]: + redacted: dict[str, Any] = {} + secret_keys = { + "api_key", + "apikey", + "access_token", + "auth_token", + "refresh_token", + "secret", + "secret_key", + "password", + "token", + } + for key, value in config.items(): + lower = key.lower() + if lower in secret_keys or lower.endswith(("_api_key", "_secret", "_password")): + continue + redacted[key] = _normalize_fingerprint_value(value) + return redacted + + +def _normalize_fingerprint_value(value: Any) -> Any: + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, Path): + return str(value) + if isinstance(value, dict): + return { + str(key): _normalize_fingerprint_value(item) + for key, item in sorted(value.items(), key=lambda pair: str(pair[0])) + } + if isinstance(value, set): + normalized = [_normalize_fingerprint_value(item) for item in value] + return sorted(normalized, key=lambda item: json.dumps(item, sort_keys=True, ensure_ascii=True)) + if isinstance(value, (list, tuple)): + return [_normalize_fingerprint_value(item) for item in value] + if callable(value): + qualname = getattr(value, "__qualname__", getattr(value, "__name__", type(value).__name__)) + normalized = { + "callable": f"{getattr(value, '__module__', 'builtins')}.{qualname}", + } + try: + source = inspect.getsource(value) + except (OSError, TypeError): + source = None + if source is not None: + normalized["source_sha256"] = hashlib.sha256(source.encode("utf-8")).hexdigest() + return normalized + isoformat = getattr(value, "isoformat", None) + if callable(isoformat): + try: + return isoformat() + except TypeError: + pass + state = getattr(value, "__dict__", None) + object_type = f"{type(value).__module__}.{type(value).__qualname__}" + if isinstance(state, dict) and state: + return { + "object_type": object_type, + "state": _normalize_fingerprint_value(state), + } + return {"object_type": object_type} + + +def _pipeline_fingerprint(pipeline: Pipeline) -> str: + payload = { + "name": pipeline.name, + "llm_config": _sanitize_llm_config(pipeline.llm_config), + "layers": [ + { + "name": layer.name, + "class": type(layer).__name__, + "depends_on": [dep.name for dep in layer.depends_on], + "config": _normalize_fingerprint_value(layer.config), + } + for layer in pipeline.layers + ], + "projections": [ + { + "name": proj.name, + "class": type(proj).__name__, + "sources": [src.name for src in proj.sources], + "config": _normalize_fingerprint_value(proj.config), + } + for proj in pipeline.projections + ], + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + return f"sha256:{hashlib.sha256(encoded).hexdigest()}" + + +def _store_content_text(object_store: ObjectStore, text: str) -> str: + content_oid, _size_bytes = object_store.put_text(text) + return content_oid + + +def _artifact_text_content(artifact: Artifact) -> str: + if not isinstance(artifact.content, str): + msg = f"artifact {artifact.label!r} content must be a string, got {type(artifact.content).__name__}" + raise TypeError(msg) + return artifact.content + + +@dataclass +class BuildTransaction: + """Canonical build state accumulated during a single pipeline run. + + Artifact state is captured as it is produced or reused during the build so + final snapshot commit does not need to scrape the mutable compatibility + release surface under ``build/``. + + Projection state capture is intentionally deferred to the explicit + build/release adapter slice. Projections still materialize into ``build/`` + today as compatibility outputs, but they are not yet part of the canonical + snapshot closure. + """ + + pipeline: Pipeline + build_dir: Path + synix_dir: Path + run_id: str + object_store: ObjectStore + head_ref: str + parent_snapshot_oid: str | None + artifact_oids: dict[str, str] = field(default_factory=dict) + layer_artifact_oids: dict[str, list[str]] = field(default_factory=dict) + projection_oids: dict[str, str] = field(default_factory=dict) + _lock: Lock = field(default_factory=Lock, init=False, repr=False) + + @classmethod + def start(cls, pipeline: Pipeline, build_dir: str | Path, run_id: str) -> BuildTransaction: + build_path = Path(build_dir).resolve() + synix_dir = synix_dir_for_build_dir(build_path, configured_synix_dir=pipeline.synix_dir) + synix_dir.mkdir(parents=True, exist_ok=True) + + ref_store = RefStore(synix_dir) + head_ref = ref_store.ensure_head(DEFAULT_HEAD_REF) + + return cls( + pipeline=pipeline, + build_dir=build_path, + synix_dir=synix_dir, + run_id=run_id, + object_store=ObjectStore(synix_dir), + head_ref=head_ref, + parent_snapshot_oid=ref_store.read_ref("HEAD"), + ) + + def record_artifact( + self, + artifact: Artifact, + *, + layer_name: str, + layer_level: int, + parent_labels: list[str], + ) -> str: + with self._lock: + content = _artifact_text_content(artifact) + snapshot_metadata = dict(artifact.metadata) + snapshot_metadata.setdefault("layer_name", layer_name) + snapshot_metadata.setdefault("layer_level", layer_level) + content_hash = f"sha256:{hashlib.sha256(content.encode('utf-8')).hexdigest()}" + if artifact.artifact_id and artifact.artifact_id != content_hash: + msg = ( + f"artifact {artifact.label!r} has artifact_id {artifact.artifact_id!r} " + f"that does not match its content hash {content_hash!r}" + ) + raise ValueError(msg) + if not artifact.artifact_id: + artifact.artifact_id = content_hash + content_oid = _store_content_text(self.object_store, content) + artifact_payload = _object( + "artifact", + label=artifact.label, + artifact_type=artifact.artifact_type, + artifact_id=content_hash, + content_oid=content_oid, + input_ids=list(artifact.input_ids), + prompt_id=artifact.prompt_id, + model_config=artifact.model_config, + metadata=snapshot_metadata, + parent_labels=parent_labels, + ) + artifact_oid = self.object_store.put_json(artifact_payload) + + previous_oid = self.artifact_oids.get(artifact.label) + if previous_oid is not None and previous_oid != artifact_oid: + layer_oids = self.layer_artifact_oids.get(layer_name, []) + if previous_oid in layer_oids: + layer_oids.remove(previous_oid) + + self.artifact_oids[artifact.label] = artifact_oid + layer_oids = self.layer_artifact_oids.setdefault(layer_name, []) + if artifact_oid not in layer_oids: + layer_oids.append(artifact_oid) + + return artifact_oid + + def assert_complete(self, layer_artifacts: dict[str, list[Artifact]]) -> None: + """Fail closed if the transaction missed artifacts present in the current build state.""" + expected_labels = { + artifact.label + for artifacts in layer_artifacts.values() + for artifact in artifacts + } + recorded_labels = set(self.artifact_oids) + + missing = sorted(expected_labels.difference(recorded_labels)) + unexpected = sorted(recorded_labels.difference(expected_labels)) + if not missing and not unexpected: + return + + details: list[str] = [] + if missing: + details.append(f"missing={missing[:5]}") + if unexpected: + details.append(f"unexpected={unexpected[:5]}") + msg = "snapshot transaction closure mismatch: " + ", ".join(details) + raise RuntimeError(msg) + + +def start_build_transaction(pipeline: Pipeline, build_dir: str | Path, run_id: str) -> BuildTransaction: + """Create a build transaction that accumulates canonical snapshot state.""" + return BuildTransaction.start(pipeline, build_dir, run_id) + + +def _write_ref_update_journal(synix_dir: Path, journal_id: str, updates: dict[str, str]) -> Path: + journal_dir = synix_dir / "ref_journal" + journal_dir.mkdir(parents=True, exist_ok=True) + journal_path = journal_dir / f"{journal_id}.json" + atomic_write( + journal_path, + json.dumps( + { + "schema_version": SCHEMA_VERSION, + "type": "ref_update", + "updates": updates, + }, + sort_keys=True, + indent=2, + ), + ) + return journal_path + + +def recover_pending_ref_updates(synix_dir: str | Path, *, ref_store: RefStore | None = None) -> None: + """Apply any pending ref update journals left behind by interrupted commits.""" + resolved_synix_dir = Path(synix_dir) + journal_dir = resolved_synix_dir / "ref_journal" + if not journal_dir.exists(): + return + + store = ref_store or RefStore(resolved_synix_dir) + for journal_path in sorted(journal_dir.glob("*.json")): + payload = json.loads(journal_path.read_text(encoding="utf-8")) + updates = payload.get("updates") + if not isinstance(updates, dict): + msg = f"ref update journal {journal_path} is missing an 'updates' mapping" + raise ValueError(msg) + for ref_name, oid in updates.items(): + store.write_ref(ref_name, oid) + journal_path.unlink() + + +@contextmanager +def _snapshot_lock(synix_dir: Path): + lock_path = synix_dir / ".lock" + lock_path.parent.mkdir(parents=True, exist_ok=True) + + try: + import fcntl + except ImportError: + fcntl = None + try: + import msvcrt + except ImportError: + msvcrt = None + + with lock_path.open("a+b") as handle: + if fcntl is not None: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + elif msvcrt is not None: + handle.seek(0) + if handle.read(1) == b"": + handle.write(b"0") + handle.flush() + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1) + try: + yield + finally: + if fcntl is not None: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + elif msvcrt is not None: + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + + +def commit_build_snapshot(transaction: BuildTransaction) -> dict[str, str]: + """Commit a build transaction as an immutable snapshot and advance refs.""" + synix_dir = transaction.synix_dir + synix_dir.mkdir(parents=True, exist_ok=True) + + with _snapshot_lock(synix_dir): + ref_store = RefStore(synix_dir) + recover_pending_ref_updates(synix_dir, ref_store=ref_store) + current_head_ref = ref_store.ensure_head(DEFAULT_HEAD_REF) + if current_head_ref != transaction.head_ref: + msg = ( + "HEAD target changed during build " + f"({transaction.head_ref!r} -> {current_head_ref!r}); rerun against the latest ref state" + ) + raise RuntimeError(msg) + + current_head_oid = ref_store.read_ref("HEAD") + if current_head_oid != transaction.parent_snapshot_oid: + msg = "HEAD advanced during build; rerun against the latest snapshot" + raise RuntimeError(msg) + + manifest_payload = _object( + "manifest", + pipeline_name=transaction.pipeline.name, + pipeline_fingerprint=_pipeline_fingerprint(transaction.pipeline), + artifacts=[ + {"label": label, "oid": oid} + for label, oid in sorted(transaction.artifact_oids.items()) + ], + projections=transaction.projection_oids, + ) + manifest_oid = transaction.object_store.put_json(manifest_payload) + + snapshot_payload = _object( + "snapshot", + manifest_oid=manifest_oid, + parent_snapshot_oids=[transaction.parent_snapshot_oid] if transaction.parent_snapshot_oid else [], + created_at=datetime.now(UTC).isoformat(), + pipeline_name=transaction.pipeline.name, + run_id=transaction.run_id, + ) + snapshot_oid = transaction.object_store.put_json(snapshot_payload) + + run_ref = f"refs/runs/{transaction.run_id}" + pending_updates = { + run_ref: snapshot_oid, + transaction.head_ref: snapshot_oid, + } + journal_path = _write_ref_update_journal(synix_dir, transaction.run_id, pending_updates) + refs_applied = False + try: + for ref_name, oid in pending_updates.items(): + ref_store.write_ref(ref_name, oid) + refs_applied = True + finally: + if refs_applied and journal_path.exists(): + journal_path.unlink() + + return { + "snapshot_oid": snapshot_oid, + "manifest_oid": manifest_oid, + "head_ref": transaction.head_ref, + "run_ref": run_ref, + "synix_dir": str(synix_dir), + } + + +def list_runs(build_dir: str | Path, *, synix_dir: str | Path | None = None) -> list[dict[str, str]]: + """List recorded run refs for a build dir.""" + build_path = Path(build_dir).resolve() + resolved_synix_dir = synix_dir_for_build_dir(build_path, configured_synix_dir=synix_dir) + if not resolved_synix_dir.exists(): + return [] + + object_store = ObjectStore(resolved_synix_dir) + ref_store = RefStore(resolved_synix_dir) + recover_pending_ref_updates(resolved_synix_dir, ref_store=ref_store) + runs: list[dict[str, str]] = [] + for ref_name, oid in ref_store.iter_refs("refs/runs"): + snapshot = object_store.get_json(oid) + runs.append( + { + "ref": ref_name, + "snapshot_oid": oid, + "run_id": snapshot.get("run_id", ""), + "created_at": snapshot.get("created_at", ""), + "pipeline_name": snapshot.get("pipeline_name", ""), + } + ) + return runs diff --git a/src/synix/cli/build_commands.py b/src/synix/cli/build_commands.py index 9ca943c..8fd6eb8 100644 --- a/src/synix/cli/build_commands.py +++ b/src/synix/cli/build_commands.py @@ -224,6 +224,11 @@ def build( console.print(f"\n[bold]Total:[/bold] {result.built} built, {result.cached} cached, {result.skipped} skipped") if not is_demo_mode(): console.print(f"[bold]Time:[/bold] {elapsed:.1f}s") + if not is_demo_mode() and result.snapshot_oid and result.run_ref: + run_id = result.run_ref.rsplit("/", 1)[-1] + console.print(f"[bold]Artifact Snapshot:[/bold] {result.snapshot_oid[:12]}") + console.print(f"[bold]Run ID:[/bold] {run_id}") + console.print(f"[bold]Run Ref:[/bold] {result.run_ref}") # Show run log summary when verbose run_log = result.run_log diff --git a/src/synix/cli/main.py b/src/synix/cli/main.py index 6c3446d..ab28ad7 100644 --- a/src/synix/cli/main.py +++ b/src/synix/cli/main.py @@ -85,6 +85,7 @@ def cli(): from synix.cli.init_commands import init # noqa: E402, F401 from synix.cli.llms_commands import llms # noqa: E402, F401 from synix.cli.mesh_commands import mesh # noqa: E402, F401 +from synix.cli.runs_commands import runs_group # noqa: E402, F401 from synix.cli.search_commands import search # noqa: E402, F401 from synix.cli.validate_commands import validate # noqa: E402, F401 from synix.cli.verify_commands import diff, lineage, status, verify # noqa: E402, F401 @@ -109,3 +110,4 @@ def cli(): main.add_command(llms) main.add_command(batch_build, name="batch-build") main.add_command(mesh) +main.add_command(runs_group, name="runs") diff --git a/src/synix/cli/runs_commands.py b/src/synix/cli/runs_commands.py new file mode 100644 index 0000000..6c7ffa7 --- /dev/null +++ b/src/synix/cli/runs_commands.py @@ -0,0 +1,67 @@ +"""Run and snapshot inspection commands.""" + +from __future__ import annotations + +import json +from datetime import datetime + +import click +from rich import box +from rich.table import Table + +from synix.build.refs import synix_dir_for_build_dir +from synix.build.snapshots import list_runs +from synix.cli.main import console + +RUNS_LIST_JSON_SCHEMA_VERSION = 1 + + +@click.group("runs") +def runs_group(): + """Inspect immutable Synix artifact snapshots.""" + + +@runs_group.command("list") +@click.option("--build-dir", default="./build", help="Build directory") +@click.option("--synix-dir", default=None, help="Explicit .synix directory") +@click.option("--json", "json_output", is_flag=True, help="Emit machine-readable JSON instead of a table") +def list_runs_command(build_dir: str, synix_dir: str | None, json_output: bool): + """List recorded run refs and artifact snapshot ids.""" + resolved_synix_dir = synix_dir_for_build_dir(build_dir, configured_synix_dir=synix_dir) + if not resolved_synix_dir.exists(): + console.print("[red]No snapshot store found.[/red] Run [bold]synix build[/bold] first.") + raise SystemExit(1) + + runs = list_runs(build_dir, synix_dir=resolved_synix_dir) + if json_output: + console.print_json( + json.dumps( + { + "schema_version": RUNS_LIST_JSON_SCHEMA_VERSION, + "runs": runs, + } + ) + ) + return + if not runs: + console.print("[dim]No run snapshots found.[/dim]") + return + + table = Table(title="Run Artifact Snapshots", box=box.ROUNDED) + table.add_column("Run ID", style="bold") + table.add_column("Snapshot", no_wrap=True) + table.add_column("Created", no_wrap=True) + table.add_column("Pipeline") + table.add_column("Ref") + + for run in runs: + created_at = run["created_at"] + if created_at: + try: + created_at = datetime.fromisoformat(created_at).strftime("%Y-%m-%d %H:%M:%S") + except ValueError: + pass + table.add_row(run["run_id"], run["snapshot_oid"][:12], created_at, run["pipeline_name"], run["ref"]) + + console.print() + console.print(table) diff --git a/src/synix/core/logging.py b/src/synix/core/logging.py index d6e59ed..929c584 100644 --- a/src/synix/core/logging.py +++ b/src/synix/core/logging.py @@ -4,6 +4,7 @@ import datetime as _dt import json +import secrets import time from dataclasses import dataclass, field from datetime import datetime @@ -119,6 +120,16 @@ class RunSummary: cached: int = 0 +def _generate_run_id() -> str: + """Create a time-prefixed, collision-resistant run id. + + The UTC timestamp prefix keeps runs roughly sortable for humans, while the + random suffix avoids ref/log collisions across concurrent processes. + """ + prefix = datetime.now(_dt.UTC).strftime("%Y%m%dT%H%M%S%fZ") + return f"{prefix}-{secrets.token_hex(4)}" + + class SynixLogger: """Structured logger for Synix pipeline runs. @@ -136,7 +147,7 @@ def __init__( self.build_dir = build_dir self.progress = progress self.run_log = RunLog( - run_id=datetime.now(_dt.UTC).strftime("%Y%m%dT%H%M%SZ"), + run_id=_generate_run_id(), ) self._lock = Lock() self._log_file = None diff --git a/src/synix/core/models.py b/src/synix/core/models.py index 9e251d2..3269529 100644 --- a/src/synix/core/models.py +++ b/src/synix/core/models.py @@ -34,7 +34,7 @@ class Artifact: metadata: dict = field(default_factory=dict) def __post_init__(self): - if not self.artifact_id and self.content: + if not self.artifact_id and isinstance(self.content, str): self.artifact_id = f"sha256:{hashlib.sha256(self.content.encode()).hexdigest()}" @@ -283,12 +283,14 @@ def __init__( *, source_dir: str = "./sources", build_dir: str = "./build", + synix_dir: str | None = None, llm_config: dict | None = None, concurrency: int = 5, ): self.name = name self.source_dir = source_dir self.build_dir = build_dir + self.synix_dir = synix_dir self.llm_config: dict = llm_config or {} self.concurrency = concurrency self.layers: list[Layer] = [] # Source + Transform diff --git a/tests/e2e/test_snapshot_flow.py b/tests/e2e/test_snapshot_flow.py new file mode 100644 index 0000000..d1eba81 --- /dev/null +++ b/tests/e2e/test_snapshot_flow.py @@ -0,0 +1,142 @@ +"""E2E tests for the snapshot-aware CLI workflow.""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from click.testing import CliRunner + +from synix.build.snapshots import list_runs +from synix.cli import main + +FIXTURES_DIR = Path(__file__).parent.parent / "synix" / "fixtures" + + +@pytest.fixture +def runner(): + return CliRunner() + + +@pytest.fixture +def workspace(tmp_path): + source_dir = tmp_path / "exports" + source_dir.mkdir() + build_dir = tmp_path / "build" + + shutil.copy(FIXTURES_DIR / "chatgpt_export.json", source_dir / "chatgpt_export.json") + shutil.copy(FIXTURES_DIR / "claude_export.json", source_dir / "claude_export.json") + + return {"root": tmp_path, "source_dir": source_dir, "build_dir": build_dir} + + +@pytest.fixture +def pipeline_file(workspace): + path = workspace["root"] / "pipeline.py" + path.write_text(f""" +from synix import Pipeline, Source, SearchIndex, FlatFile +from synix.transforms import EpisodeSummary, MonthlyRollup, CoreSynthesis + +pipeline = Pipeline("snapshot-cli") +pipeline.source_dir = "{workspace["source_dir"]}" +pipeline.build_dir = "{workspace["build_dir"]}" +pipeline.llm_config = {{"model": "claude-sonnet-4-20250514", "temperature": 0.3, "max_tokens": 1024}} + +transcripts = Source("transcripts") +episodes = EpisodeSummary("episodes", depends_on=[transcripts]) +monthly = MonthlyRollup("monthly", depends_on=[episodes]) +core = CoreSynthesis("core", depends_on=[monthly], context_budget=10000) + +pipeline.add(transcripts, episodes, monthly, core) +pipeline.add(SearchIndex("memory-index", sources=[episodes, monthly, core], search=["fulltext"])) +pipeline.add(FlatFile("context-doc", sources=[core], output_path="{workspace["build_dir"] / "context.md"}")) +""") + return path + + +@pytest.fixture(autouse=True) +def mock_anthropic(monkeypatch): + def mock_create(**kwargs): + messages = kwargs.get("messages", []) + content = messages[0].get("content", "") if messages else "" + + if "summarizing a conversation" in content.lower(): + return _mock_response("Conversation summary about technical topics.") + if "monthly" in content.lower(): + return _mock_response("Monthly rollup about technical learning.") + if "core memory" in content.lower(): + return _mock_response("## Identity\nEngineer.\n\n## Current Focus\nSnapshot testing.") + return _mock_response("Mock response.") + + mock_client = MagicMock() + mock_client.messages.create = mock_create + monkeypatch.setattr("anthropic.Anthropic", lambda **kwargs: mock_client) + + +def _mock_response(text: str): + resp = MagicMock() + resp.content = [MagicMock(text=text)] + resp.model = "claude-sonnet-4-20250514" + resp.usage = MagicMock(input_tokens=100, output_tokens=50) + return resp + + +class TestSnapshotFlow: + def test_build_outputs_snapshot_and_runs_list_shows_history(self, runner, workspace, pipeline_file): + build_dir = str(workspace["build_dir"]) + + first = runner.invoke(main, ["run", str(pipeline_file), "--plain"]) + assert first.exit_code == 0, first.output + assert "Artifact Snapshot:" in first.output + assert "Run Ref:" in first.output + + first_run_ref = next(line.split("Run Ref:", 1)[1].strip() for line in first.output.splitlines() if "Run Ref:" in line) + + second = runner.invoke(main, ["run", str(pipeline_file), "--plain"]) + assert second.exit_code == 0, second.output + assert "Artifact Snapshot:" in second.output + assert "Run Ref:" in second.output + + second_run_ref = next( + line.split("Run Ref:", 1)[1].strip() for line in second.output.splitlines() if "Run Ref:" in line + ) + assert second_run_ref != first_run_ref + + runs_list = runner.invoke(main, ["runs", "list", "--build-dir", build_dir], terminal_width=160) + assert runs_list.exit_code == 0, runs_list.output + assert "Run Artifact Snapshots" in runs_list.output + assert "Run ID" in runs_list.output + assert "Ref" in runs_list.output + + runs_json = runner.invoke(main, ["runs", "list", "--build-dir", build_dir, "--json"]) + assert runs_json.exit_code == 0, runs_json.output + payload = json.loads(runs_json.output) + assert payload["schema_version"] == 1 + assert {run_info["ref"] for run_info in payload["runs"]} == {first_run_ref, second_run_ref} + + recorded_runs = list_runs(build_dir) + assert {run_info["ref"] for run_info in recorded_runs} == {first_run_ref, second_run_ref} + + def test_clean_removes_build_surface_but_preserves_snapshot_history(self, runner, workspace, pipeline_file): + build_dir = str(workspace["build_dir"]) + synix_dir = workspace["root"] / ".synix" + + built = runner.invoke(main, ["run", str(pipeline_file), "--plain"]) + assert built.exit_code == 0, built.output + assert synix_dir.exists() + assert workspace["build_dir"].exists() + + cleaned = runner.invoke(main, ["clean", build_dir, "--yes"]) + assert cleaned.exit_code == 0, cleaned.output + assert not workspace["build_dir"].exists() + assert synix_dir.exists() + + runs_json = runner.invoke(main, ["runs", "list", "--build-dir", build_dir, "--json"]) + assert runs_json.exit_code == 0, runs_json.output + payload = json.loads(runs_json.output) + assert payload["schema_version"] == 1 + assert len(payload["runs"]) == 1 + assert payload["runs"][0]["pipeline_name"] == "snapshot-cli" diff --git a/tests/integration/test_snapshots.py b/tests/integration/test_snapshots.py new file mode 100644 index 0000000..bb573d0 --- /dev/null +++ b/tests/integration/test_snapshots.py @@ -0,0 +1,521 @@ +"""Integration tests for immutable Synix snapshots.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from synix import FlatFile, Pipeline, SearchIndex, Source +from synix.build.artifacts import ArtifactStore +from synix.build.object_store import SCHEMA_VERSION, ObjectStore +from synix.build.refs import RefStore +from synix.build.runner import run +from synix.build.snapshots import _pipeline_fingerprint, commit_build_snapshot, list_runs, start_build_transaction +from synix.build.validators import RequiredField +from synix.core.models import Artifact +from synix.transforms import CoreSynthesis, EpisodeSummary, MonthlyRollup + + +def _manifest_artifact_map(manifest: dict) -> dict[str, str]: + return {entry["label"]: entry["oid"] for entry in manifest["artifacts"]} + + +def _build_pipeline(build_dir: Path, source_dir: Path, *, synix_dir: Path | None = None) -> Pipeline: + pipeline = Pipeline("snapshot-pipeline") + pipeline.build_dir = str(build_dir) + pipeline.source_dir = str(source_dir) + pipeline.synix_dir = str(synix_dir) if synix_dir is not None else str(build_dir.parent / ".synix") + pipeline.llm_config = { + "model": "claude-sonnet-4-20250514", + "temperature": 0.3, + "max_tokens": 1024, + } + + transcripts = Source("transcripts") + episodes = EpisodeSummary("episodes", depends_on=[transcripts]) + monthly = MonthlyRollup("monthly", depends_on=[episodes]) + core = CoreSynthesis("core", depends_on=[monthly], context_budget=10000) + + pipeline.add(transcripts, episodes, monthly, core) + pipeline.add(SearchIndex("memory-index", sources=[episodes, monthly, core], search=["fulltext"])) + pipeline.add(FlatFile("context-doc", sources=[core], output_path=str(build_dir / "context.md"))) + return pipeline + + +class TestSnapshots: + def test_commit_uses_transaction_state_not_mutable_build_files(self, tmp_path): + """Snapshot commit should not reread manifest/provenance from the mutable build directory.""" + build_dir = tmp_path / "build" + pipeline = Pipeline( + "transaction-only", + build_dir=str(build_dir), + synix_dir=str(tmp_path / ".synix"), + ) + transcripts = Source("transcripts") + pipeline.add(transcripts) + + txn = start_build_transaction(pipeline, build_dir, run_id="20260306T120000000000Z") + txn.record_artifact( + Artifact(label="ep-1", artifact_type="episode", content="Captured before build-dir mutation."), + layer_name="transcripts", + layer_level=0, + parent_labels=[], + ) + + build_dir.mkdir(parents=True, exist_ok=True) + (build_dir / "manifest.json").write_text("{not valid json", encoding="utf-8") + (build_dir / "provenance.json").write_text("{not valid json", encoding="utf-8") + + snapshot_info = commit_build_snapshot(txn) + object_store = ObjectStore(tmp_path / ".synix") + manifest = object_store.get_json(snapshot_info["manifest_oid"]) + + assert _manifest_artifact_map(manifest) == {"ep-1": txn.artifact_oids["ep-1"]} + stored_artifact = object_store.get_json(_manifest_artifact_map(manifest)["ep-1"]) + assert stored_artifact["label"] == "ep-1" + + def test_commit_rejects_when_head_advances_during_build(self, tmp_path): + """Snapshot commit should fail closed if another writer advances HEAD first.""" + build_dir = tmp_path / "build" + pipeline = Pipeline( + "concurrent-build", + build_dir=str(build_dir), + synix_dir=str(tmp_path / ".synix"), + ) + transcripts = Source("transcripts") + pipeline.add(transcripts) + + txn = start_build_transaction(pipeline, build_dir, run_id="20260306T120000000001Z") + txn.record_artifact( + Artifact(label="ep-1", artifact_type="episode", content="Concurrent head test."), + layer_name="transcripts", + layer_level=0, + parent_labels=[], + ) + + ref_store = RefStore(tmp_path / ".synix") + ref_store.write_ref(txn.head_ref, "1" * 64) + + with pytest.raises(RuntimeError, match="HEAD advanced during build"): + commit_build_snapshot(txn) + + def test_commit_rejects_artifact_id_content_mismatch(self, tmp_path): + """Snapshot commit should not make corrupted cached artifacts canonical.""" + build_dir = tmp_path / "build" + pipeline = Pipeline( + "artifact-integrity", + build_dir=str(build_dir), + synix_dir=str(tmp_path / ".synix"), + ) + transcripts = Source("transcripts") + pipeline.add(transcripts) + + txn = start_build_transaction(pipeline, build_dir, run_id="20260306T120000000002Z") + + with pytest.raises(ValueError, match="does not match its content hash"): + txn.record_artifact( + Artifact( + label="ep-1", + artifact_type="episode", + content="Actual content.", + artifact_id="sha256:" + "0" * 64, + ), + layer_name="transcripts", + layer_level=0, + parent_labels=[], + ) + + def test_empty_content_artifacts_are_hashed_and_snapshotted(self, tmp_path): + """Empty-string content is still valid content and must hash consistently.""" + build_dir = tmp_path / "build" + pipeline = Pipeline( + "empty-content", + build_dir=str(build_dir), + synix_dir=str(tmp_path / ".synix"), + ) + transcripts = Source("transcripts") + pipeline.add(transcripts) + + txn = start_build_transaction(pipeline, build_dir, run_id="20260306T120000000003Z") + artifact = Artifact(label="empty", artifact_type="note", content="") + txn.record_artifact( + artifact, + layer_name="transcripts", + layer_level=0, + parent_labels=[], + ) + + snapshot_info = commit_build_snapshot(txn) + manifest = ObjectStore(tmp_path / ".synix").get_json(snapshot_info["manifest_oid"]) + stored_artifact = ObjectStore(tmp_path / ".synix").get_json(_manifest_artifact_map(manifest)["empty"]) + + assert artifact.artifact_id == "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + assert stored_artifact["artifact_id"] == artifact.artifact_id + assert ObjectStore(tmp_path / ".synix").get_bytes(stored_artifact["content_oid"]) == b"" + + def test_transaction_requires_complete_artifact_closure(self, tmp_path): + """Snapshot commit must fail closed if the run's artifact closure was not fully recorded.""" + build_dir = tmp_path / "build" + pipeline = Pipeline( + "closure-check", + build_dir=str(build_dir), + synix_dir=str(tmp_path / ".synix"), + ) + transcripts = Source("transcripts") + pipeline.add(transcripts) + + txn = start_build_transaction(pipeline, build_dir, run_id="20260306T120000000004Z") + present = Artifact(label="present", artifact_type="note", content="Recorded") + missing = Artifact(label="missing", artifact_type="note", content="Not recorded") + txn.record_artifact( + present, + layer_name="transcripts", + layer_level=0, + parent_labels=[], + ) + + with pytest.raises(RuntimeError, match="snapshot transaction closure mismatch"): + txn.assert_complete({"transcripts": [present, missing]}) + + def test_runs_list_recovers_pending_ref_updates(self, tmp_path): + """Interrupted ref updates should be replayed before run history is listed.""" + build_dir = tmp_path / "build" + synix_dir = tmp_path / ".synix" + object_store = ObjectStore(synix_dir) + + manifest_oid = object_store.put_json( + { + "type": "manifest", + "schema_version": SCHEMA_VERSION, + "pipeline_name": "recovery-test", + "pipeline_fingerprint": "sha256:test", + "artifacts": [], + "projections": {}, + } + ) + snapshot_oid = object_store.put_json( + { + "type": "snapshot", + "schema_version": SCHEMA_VERSION, + "manifest_oid": manifest_oid, + "parent_snapshot_oids": [], + "created_at": "2026-03-06T08:20:07Z", + "pipeline_name": "recovery-test", + "run_id": "20260306T082007123456Z", + } + ) + + journal_dir = synix_dir / "ref_journal" + journal_dir.mkdir(parents=True, exist_ok=True) + (journal_dir / "pending.json").write_text( + json.dumps( + { + "schema_version": SCHEMA_VERSION, + "type": "ref_update", + "updates": { + "refs/runs/20260306T082007123456Z": snapshot_oid, + "refs/heads/main": snapshot_oid, + }, + }, + indent=2, + ), + encoding="utf-8", + ) + + runs = list_runs(build_dir, synix_dir=synix_dir) + ref_store = RefStore(synix_dir) + + assert ref_store.read_ref("refs/heads/main") == snapshot_oid + assert ref_store.read_ref("refs/runs/20260306T082007123456Z") == snapshot_oid + assert [run_info["ref"] for run_info in runs] == ["refs/runs/20260306T082007123456Z"] + assert not any(journal_dir.iterdir()) + + def test_pipeline_fingerprint_ignores_machine_local_paths_and_secrets(self, tmp_path): + """Fingerprint should reflect logical build config, not local directories or API keys.""" + pipeline_a = Pipeline( + "fingerprint-test", + source_dir=str(tmp_path / "sources-a"), + build_dir=str(tmp_path / "build-a"), + synix_dir=str(tmp_path / ".synix-a"), + llm_config={ + "model": "claude-sonnet-4-20250514", + "temperature": 0.3, + "api_key": "secret-a", + }, + ) + pipeline_b = Pipeline( + "fingerprint-test", + source_dir=str(tmp_path / "sources-b"), + build_dir=str(tmp_path / "build-b"), + synix_dir=str(tmp_path / ".synix-b"), + llm_config={ + "model": "claude-sonnet-4-20250514", + "temperature": 0.3, + "api_key": "secret-b", + }, + ) + + transcripts_a = Source("transcripts") + episodes_a = EpisodeSummary("episodes", depends_on=[transcripts_a]) + pipeline_a.add(transcripts_a, episodes_a) + pipeline_a.add(SearchIndex("memory-index", sources=[episodes_a], search=["fulltext"])) + + transcripts_b = Source("transcripts") + episodes_b = EpisodeSummary("episodes", depends_on=[transcripts_b]) + pipeline_b.add(transcripts_b, episodes_b) + pipeline_b.add(SearchIndex("memory-index", sources=[episodes_b], search=["fulltext"])) + + assert _pipeline_fingerprint(pipeline_a) == _pipeline_fingerprint(pipeline_b) + + def test_pipeline_fingerprint_keeps_non_secret_llm_fields(self, tmp_path): + """Normal llm config like max_tokens must remain part of the fingerprint.""" + pipeline_a = Pipeline( + "fingerprint-llm", + source_dir=str(tmp_path / "sources-a"), + build_dir=str(tmp_path / "build-a"), + synix_dir=str(tmp_path / ".synix-a"), + llm_config={ + "model": "claude-sonnet-4-20250514", + "max_tokens": 1024, + }, + ) + pipeline_b = Pipeline( + "fingerprint-llm", + source_dir=str(tmp_path / "sources-b"), + build_dir=str(tmp_path / "build-b"), + synix_dir=str(tmp_path / ".synix-b"), + llm_config={ + "model": "claude-sonnet-4-20250514", + "max_tokens": 2048, + }, + ) + + transcripts_a = Source("transcripts") + transcripts_b = Source("transcripts") + pipeline_a.add(transcripts_a) + pipeline_b.add(transcripts_b) + + assert _pipeline_fingerprint(pipeline_a) != _pipeline_fingerprint(pipeline_b) + + def test_pipeline_fingerprint_includes_callable_source_when_available(self, tmp_path): + """Callable config should contribute source identity, not just module + qualname.""" + + def strategy_a(text: str) -> str: + return text.strip() + + def strategy_b(text: str) -> str: + return text.rstrip("!") + + pipeline_a = Pipeline( + "fingerprint-callable", + build_dir=str(tmp_path / "build-a"), + synix_dir=str(tmp_path / ".synix-a"), + ) + pipeline_b = Pipeline( + "fingerprint-callable", + build_dir=str(tmp_path / "build-b"), + synix_dir=str(tmp_path / ".synix-b"), + ) + + pipeline_a.add(Source("transcripts", config={"strategy": strategy_a})) + pipeline_b.add(Source("transcripts", config={"strategy": strategy_b})) + + assert _pipeline_fingerprint(pipeline_a) != _pipeline_fingerprint(pipeline_b) + + def test_snapshot_transaction_requires_text_content(self, tmp_path): + """Snapshotting should reject non-text artifact content with a clear boundary error.""" + build_dir = tmp_path / "build" + pipeline = Pipeline( + "non-text-content", + build_dir=str(build_dir), + synix_dir=str(tmp_path / ".synix"), + ) + transcripts = Source("transcripts") + pipeline.add(transcripts) + + txn = start_build_transaction(pipeline, build_dir, run_id="20260306T120000000005Z") + artifact = Artifact(label="bad", artifact_type="note", content="") + artifact.content = None # type: ignore[assignment] + + with pytest.raises(TypeError, match="content must be a string"): + txn.record_artifact( + artifact, + layer_name="transcripts", + layer_level=0, + parent_labels=[], + ) + + def test_build_commits_manifest_and_snapshot(self, tmp_path, source_dir_with_fixtures, mock_llm): + """A successful build records immutable objects and moves HEAD.""" + build_dir = tmp_path / "build" + pipeline = _build_pipeline(build_dir, source_dir_with_fixtures) + + result = run(pipeline, source_dir=str(source_dir_with_fixtures)) + + assert result.snapshot_oid is not None + assert result.manifest_oid is not None + assert result.run_ref is not None + assert result.head_ref == "refs/heads/main" + assert result.synix_dir == str(tmp_path / ".synix") + + object_store = ObjectStore(tmp_path / ".synix") + ref_store = RefStore(tmp_path / ".synix") + + snapshot = object_store.get_json(result.snapshot_oid) + manifest = object_store.get_json(result.manifest_oid) + + assert snapshot["type"] == "snapshot" + assert snapshot["manifest_oid"] == result.manifest_oid + assert snapshot["run_id"] + assert manifest["type"] == "manifest" + assert manifest["pipeline_name"] == "snapshot-pipeline" + assert len(manifest["artifacts"]) > 0 + assert manifest["projections"] == {} + assert (build_dir / "search.db").exists() + assert (build_dir / "context.md").exists() + + build_store = ArtifactStore(build_dir) + first_label, first_artifact_oid = next(iter(_manifest_artifact_map(manifest).items())) + first_artifact = object_store.get_json(first_artifact_oid) + assert object_store.get_bytes(first_artifact["content_oid"]).decode("utf-8") == build_store.load_artifact(first_label).content + + assert ref_store.read_head_target() == "refs/heads/main" + assert ref_store.read_ref("HEAD") == result.snapshot_oid + assert ref_store.read_ref(result.run_ref) == result.snapshot_oid + + def test_snapshot_scope_is_artifacts_only_for_now(self, tmp_path, source_dir_with_fixtures, mock_llm): + """Canonical snapshots currently capture artifacts only; projections stay in the compatibility surface.""" + build_dir = tmp_path / "build" + pipeline = _build_pipeline(build_dir, source_dir_with_fixtures) + + result = run(pipeline, source_dir=str(source_dir_with_fixtures)) + assert result.manifest_oid is not None + + object_store = ObjectStore(tmp_path / ".synix") + manifest = object_store.get_json(result.manifest_oid) + + assert manifest["projections"] == {} + assert (build_dir / "search.db").exists() + assert (build_dir / "context.md").exists() + + def test_successive_builds_preserve_old_run_ref(self, tmp_path, source_dir_with_fixtures, mock_llm): + """Each successful build gets a new snapshot while older run refs remain resolvable.""" + build_dir = tmp_path / "build" + pipeline = _build_pipeline(build_dir, source_dir_with_fixtures) + + first = run(pipeline, source_dir=str(source_dir_with_fixtures)) + second = run(pipeline, source_dir=str(source_dir_with_fixtures)) + + assert first.snapshot_oid is not None + assert second.snapshot_oid is not None + assert first.snapshot_oid != second.snapshot_oid + assert first.run_ref != second.run_ref + + object_store = ObjectStore(tmp_path / ".synix") + ref_store = RefStore(tmp_path / ".synix") + first_snapshot = object_store.get_json(first.snapshot_oid) + second_snapshot = object_store.get_json(second.snapshot_oid) + + assert first_snapshot["manifest_oid"] == first.manifest_oid + assert second_snapshot["manifest_oid"] == second.manifest_oid + assert first.manifest_oid == second.manifest_oid + assert second_snapshot["parent_snapshot_oids"] == [first.snapshot_oid] + assert ref_store.read_ref("HEAD") == second.snapshot_oid + assert ref_store.read_ref(first.run_ref) == first.snapshot_oid + assert ref_store.read_ref(second.run_ref) == second.snapshot_oid + + runs = list_runs(build_dir) + assert {run_info["ref"] for run_info in runs} == {first.run_ref, second.run_ref} + + def test_source_change_creates_new_manifest(self, tmp_path, source_dir_with_fixtures, mock_llm): + """A changed source export produces a new manifest while keeping old snapshots intact.""" + build_dir = tmp_path / "build" + pipeline = _build_pipeline(build_dir, source_dir_with_fixtures) + + first = run(pipeline, source_dir=str(source_dir_with_fixtures)) + + claude_path = source_dir_with_fixtures / "claude_export.json" + data = json.loads(claude_path.read_text()) + data["conversations"].append( + { + "uuid": "conv-new-snapshot-test", + "title": "Snapshot test conversation", + "created_at": "2024-03-25T10:00:00Z", + "chat_messages": [ + { + "uuid": "msg-new-1", + "sender": "human", + "text": "Tell me about snapshotting.", + "created_at": "2024-03-25T10:00:00Z", + }, + { + "uuid": "msg-new-2", + "sender": "assistant", + "text": "Snapshotting captures immutable build state.", + "created_at": "2024-03-25T10:01:00Z", + }, + ], + } + ) + claude_path.write_text(json.dumps(data)) + + second = run(pipeline, source_dir=str(source_dir_with_fixtures)) + + assert first.snapshot_oid is not None + assert second.snapshot_oid is not None + assert first.manifest_oid is not None + assert second.manifest_oid is not None + assert first.manifest_oid != second.manifest_oid + + object_store = ObjectStore(tmp_path / ".synix") + first_manifest = object_store.get_json(first.manifest_oid) + second_manifest = object_store.get_json(second.manifest_oid) + + assert len(second_manifest["artifacts"]) > len(first_manifest["artifacts"]) + + def test_validation_failure_does_not_advance_snapshot_refs(self, tmp_path, source_dir_with_fixtures, mock_llm): + """A failing validation result should not record or advance snapshot refs.""" + build_dir = tmp_path / "build" + synix_dir = tmp_path / ".synix" + pipeline = _build_pipeline(build_dir, source_dir_with_fixtures, synix_dir=synix_dir) + + core = next(layer for layer in pipeline.layers if layer.name == "core") + pipeline.add_validator(RequiredField(field="missing_required_field", layers=[core])) + + result = run(pipeline, source_dir=str(source_dir_with_fixtures), validate=True) + + assert result.validation is not None + assert not result.validation.passed + assert result.snapshot_oid is None + assert result.run_ref is None + assert result.manifest_oid is None + assert list_runs(build_dir, synix_dir=synix_dir) == [] + + def test_flatfile_projection_outside_build_dir_is_allowed_but_not_snapshotted( + self, + tmp_path, + source_dir_with_fixtures, + mock_llm, + ): + """Projection outputs can still target arbitrary paths until release adapters own projection state.""" + build_dir = tmp_path / "build" + pipeline = _build_pipeline(build_dir, source_dir_with_fixtures) + outside_path = tmp_path / "outside.md" + pipeline.projections = [ + proj for proj in pipeline.projections if not isinstance(proj, FlatFile) + ] + [ + FlatFile( + "external-doc", + sources=[next(layer for layer in pipeline.layers if layer.name == "core")], + output_path=str(outside_path), + ) + ] + + result = run(pipeline, source_dir=str(source_dir_with_fixtures)) + assert result.manifest_oid is not None + manifest = ObjectStore(tmp_path / ".synix").get_json(result.manifest_oid) + + assert manifest["projections"] == {} + assert outside_path.exists() diff --git a/tests/unit/test_concurrent_runner.py b/tests/unit/test_concurrent_runner.py index 8bf5138..7ba21e1 100644 --- a/tests/unit/test_concurrent_runner.py +++ b/tests/unit/test_concurrent_runner.py @@ -169,6 +169,71 @@ def test_uses_multiple_threads(self): threads_used = {entry["thread"] for entry in transform.call_log} assert len(threads_used) > 1, f"Expected multiple threads, got: {threads_used}" + def test_on_complete_receives_matching_unit_inputs_when_some_units_are_cached(self): + """Callback input mapping stays correct even when cached units are skipped.""" + inputs = [_make_transcript(f"t-{i}") for i in range(4)] + transform = MockEpisodeTransform() + config = {"llm_config": {"model": "test"}} + units = [([inp], {}) for inp in inputs] + cached_by_inputs = {tuple(sorted([inputs[1].artifact_id])): [Artifact( + label="ep-t-1", + artifact_type="episode", + content="cached", + input_ids=[inputs[1].artifact_id], + )]} + seen: list[tuple[str, str]] = [] + + def on_complete(artifacts: list[Artifact], unit_inputs: list[Artifact]) -> None: + seen.append((artifacts[0].label, unit_inputs[0].label)) + + _execute_transform_concurrent( + transform, + units, + config, + concurrency=2, + cached_by_inputs=cached_by_inputs, + on_complete=on_complete, + ) + + assert sorted(seen) == [("ep-t-0", "t-0"), ("ep-t-2", "t-2"), ("ep-t-3", "t-3")] + + def test_legacy_single_argument_callbacks_still_work(self): + """Backward compatibility: callbacks that only accept artifacts still work.""" + inputs = [_make_transcript(f"t-{i}") for i in range(3)] + transform = MockEpisodeTransform() + units = [([inp], {}) for inp in inputs] + cached_by_inputs = { + tuple(sorted([inputs[0].artifact_id])): [ + Artifact( + label="ep-t-0", + artifact_type="episode", + content="cached", + input_ids=[inputs[0].artifact_id], + ) + ] + } + cached_calls: list[str] = [] + complete_calls: list[str] = [] + + def on_cached(artifacts: list[Artifact]) -> None: + cached_calls.append(artifacts[0].label) + + def on_complete(artifacts: list[Artifact]) -> None: + complete_calls.append(artifacts[0].label) + + _execute_transform_concurrent( + transform, + units, + {"llm_config": {"model": "test"}}, + concurrency=2, + cached_by_inputs=cached_by_inputs, + on_cached=on_cached, + on_complete=on_complete, + ) + + assert cached_calls == ["ep-t-0"] + assert sorted(complete_calls) == ["ep-t-1", "ep-t-2"] + class TestConcurrentBuildSameResults: """Verify that concurrent builds produce identical results to sequential builds.""" diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 468b8d9..3414422 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re from pathlib import Path import pytest @@ -12,6 +13,7 @@ StepLog, SynixLogger, Verbosity, + _generate_run_id, ) @@ -161,6 +163,13 @@ def test_to_dict_works_with_assertion_helpers(self): class TestSynixLogger: + def test_generate_run_id_is_collision_resistant(self): + """Generated run ids stay time-prefixed but include a random suffix.""" + run_ids = {_generate_run_id() for _ in range(32)} + + assert len(run_ids) == 32 + assert all(re.fullmatch(r"\d{8}T\d{6}\d{6}Z-[0-9a-f]{8}", run_id) for run_id in run_ids) + def test_logger_creation_no_build_dir(self): """Logger works without a build_dir (no file logging).""" logger = SynixLogger(verbosity=Verbosity.DEFAULT) diff --git a/tests/unit/test_object_store.py b/tests/unit/test_object_store.py new file mode 100644 index 0000000..79f87ce --- /dev/null +++ b/tests/unit/test_object_store.py @@ -0,0 +1,69 @@ +"""Unit tests for the immutable Synix object store.""" + +from __future__ import annotations + +import hashlib + +import pytest + +from synix.build.object_store import SCHEMA_VERSION, ObjectStore + + +class TestObjectStore: + def test_put_bytes_round_trips(self, tmp_path): + """Raw bytes can be stored and loaded by oid.""" + store = ObjectStore(tmp_path / ".synix") + + oid = store.put_bytes(b"hello, snapshots") + + assert store.get_bytes(oid) == b"hello, snapshots" + + def test_put_json_is_canonical(self, tmp_path): + """Equivalent JSON payloads produce the same oid.""" + store = ObjectStore(tmp_path / ".synix") + + payload1 = { + "type": "manifest", + "schema_version": SCHEMA_VERSION, + "pipeline_name": "test", + "pipeline_fingerprint": "sha256:test", + "artifacts": [{"label": "a", "oid": "0" * 64}], + "projections": {}, + } + payload2 = { + "projections": {}, + "pipeline_fingerprint": "sha256:test", + "pipeline_name": "test", + "artifacts": [{"label": "a", "oid": "0" * 64}], + "schema_version": SCHEMA_VERSION, + "type": "manifest", + } + + oid1 = store.put_json(payload1) + oid2 = store.put_json(payload2) + + assert oid1 == oid2 + assert store.get_json(oid1) == payload1 + + def test_put_json_rejects_missing_schema_version(self, tmp_path): + """Snapshot objects must declare a schema version.""" + store = ObjectStore(tmp_path / ".synix") + + with pytest.raises(ValueError, match="schema_version"): + store.put_json({"type": "manifest", "pipeline_name": "test"}) + + def test_put_bytes_oid_matches_sha256(self, tmp_path): + """Raw byte object ids are pinned to sha256(content).""" + store = ObjectStore(tmp_path / ".synix") + payload = b"synix-blob-contract" + + oid = store.put_bytes(payload) + + assert oid == hashlib.sha256(payload).hexdigest() + + def test_put_text_rejects_non_string(self, tmp_path): + """Text objects must be explicitly textual.""" + store = ObjectStore(tmp_path / ".synix") + + with pytest.raises(TypeError, match="text must be a string"): + store.put_text(b"not-text") # type: ignore[arg-type] diff --git a/tests/unit/test_refs.py b/tests/unit/test_refs.py new file mode 100644 index 0000000..320f3b9 --- /dev/null +++ b/tests/unit/test_refs.py @@ -0,0 +1,70 @@ +"""Unit tests for Synix refs and HEAD resolution.""" + +from __future__ import annotations + +import pytest + +from synix.build.refs import DEFAULT_HEAD_REF, RefStore, synix_dir_for_build_dir + +OID1 = "1" * 64 +OID2 = "2" * 64 + + +class TestRefStore: + def test_ensure_head_creates_default_symbolic_ref(self, tmp_path): + """HEAD is created as a symbolic ref to the default build branch.""" + store = RefStore(tmp_path / ".synix") + + head_target = store.ensure_head() + + assert head_target == DEFAULT_HEAD_REF + assert store.read_head_target() == DEFAULT_HEAD_REF + assert (tmp_path / ".synix" / "HEAD").read_text() == f"ref: {DEFAULT_HEAD_REF}\n" + + def test_read_ref_resolves_head_symbolically(self, tmp_path): + """Reading HEAD resolves through the symbolic ref chain.""" + store = RefStore(tmp_path / ".synix") + store.ensure_head() + store.write_ref(DEFAULT_HEAD_REF, OID1) + + assert store.read_ref("HEAD") == OID1 + assert store.read_ref(DEFAULT_HEAD_REF) == OID1 + + def test_iter_refs_lists_run_refs(self, tmp_path): + """Run refs are discoverable under refs/runs.""" + store = RefStore(tmp_path / ".synix") + store.write_ref("refs/runs/run-1", OID1) + store.write_ref("refs/runs/run-2", OID2) + + assert store.iter_refs("refs/runs") == [ + ("refs/runs/run-1", OID1), + ("refs/runs/run-2", OID2), + ] + + def test_read_ref_rejects_symbolic_cycles(self, tmp_path): + """Corrupt symbolic refs fail loudly instead of looping forever.""" + store = RefStore(tmp_path / ".synix") + store.ensure_head() + store.write_head("refs/heads/loop") + (tmp_path / ".synix" / "refs" / "heads").mkdir(parents=True, exist_ok=True) + (tmp_path / ".synix" / "refs" / "heads" / "loop").write_text("ref: refs/heads/loop\n", encoding="utf-8") + + with pytest.raises(ValueError, match="cycle"): + store.read_ref("HEAD") + + def test_synix_dir_prefers_persistent_sibling_store(self, tmp_path): + """Default store placement stays outside build/ so clean does not wipe history.""" + build_dir = tmp_path / "build" + build_dir.mkdir() + + assert synix_dir_for_build_dir(build_dir) == tmp_path / ".synix" + + def test_synix_dir_rejects_ambiguous_legacy_and_nested_store(self, tmp_path): + """Ambiguous store discovery should fail loudly instead of silently forking history.""" + build_dir = tmp_path / "build" + build_dir.mkdir() + (tmp_path / ".synix").mkdir() + (build_dir / ".synix").mkdir() + + with pytest.raises(ValueError, match="ambiguous snapshot store resolution"): + synix_dir_for_build_dir(build_dir)