CAS and new job-running

[johnjosephhorton]

Plan: Streaming CAS Writes During Job Execution

Context

Currently, the runner builds a complete Results object after all interviews finish (build_edsl_results), then serializes it to CAS in one shot via to_jsonl_rows() → CASRepository.save(). This means:

1. All results must be held in memory simultaneously
2. If a run crashes partway through, no partial results are persisted in CAS
3. The CAS write is a single atomic operation at the very end

The idea is to stream results into the CAS directory incrementally — writing new blobs and updating the tree/commit as each interview completes, moving HEAD forward progressively. This gives crash recovery, lower peak memory, and progressive persistence.

Current Architecture (relevant pieces)

CAS Structure (per object UUID)

<uuid>/
  HEAD                  → "main\n"
  refs/main             → commit_hash
  blobs/<sha256>.json   → one JSONL row each
  trees/<sha256>.json   → {"blobs": [hash1, hash2, ...]}
  commits/<sha256>.json → {tree, parent, timestamp, message}
  current.jsonl         → snapshot

Results JSONL Format

Line 1: header   {"__header__": true, "edsl_class_name": "Results", "n_results": N, ...}
Line 2: manifest {"created_columns": [...], "n_survey_lines": S, "n_cache_lines": C, ...}
Lines 3..S+2:    Survey JSONL rows
Lines S+3..S+C+2: Cache JSONL rows
Lines S+C+3..:   Result rows (one per completed interview)

Runner Flow

JobService.submit_job()       → creates interviews + tasks
Runner._execute_job_async()   → main loop (render → enqueue → execute → on_task_completed)
  on_task_completed()         → stores answer, propagates deps, finalizes interview
  on interview complete       → marks interview done in job
After loop:
  build_edsl_results()        → batch-fetches all data, builds Result objects, returns Results

Design: Streaming CAS Writer

Core Idea

Introduce a StreamingCASWriter that:

1. On job start: Creates a UUID, writes the "preamble" (header + manifest + survey + cache rows) as blobs, creates an initial commit
2. On interview complete: Serializes the Result to a JSONL row, writes it as a new blob, builds a new tree (preamble blobs + all result blobs so far), creates a new commit pointing to previous commit, advances HEAD
3. On job finish: Updates the manifest's n_results count (or we handle this differently — see options below)

The n_results / Manifest Problem

The JSONL format's Line 2 (manifest) declares n_results and n_cache_lines upfront. With streaming, we don't know the final count until the job finishes. Options:

Option A: Placeholder manifest, rewrite at end

• Write manifest with n_results: 0 initially
• On job finish, rewrite the manifest blob and create a final commit
• Pro: Simple, compatible with existing from_jsonl
• Con: The intermediate commits have a "wrong" manifest, so loading mid-run requires special handling

Option B: Don't use n_results for parsing — use sentinel/structure

• Change from_jsonl to not rely on n_results — just read all lines after the survey+cache section as results
• The manifest already has n_survey_lines and n_cache_lines, which are fixed at job start, so the result section boundary is known
• n_results becomes informational only
• Pro: Every intermediate commit is a valid, loadable Results object
• Con: Requires a small change to from_jsonl (but it's already close — it uses lines[result_start:])

Option B is strongly preferred — it makes every commit a valid checkpoint.

Key Integration Points

1. StreamingCASWriter class (new, in edsl/object_store/)

class StreamingCASWriter:
    def __init__(self, backend: StorageBackend, branch="main"):
        self._repo = CASRepository(directory=..., backend=backend)
        self._preamble_hashes = []   # blob hashes for header+manifest+survey+cache
        self._result_hashes = []     # blob hashes for result rows, grows incrementally
        self._manifest_hash_idx = 1  # index of manifest blob in preamble

    def write_preamble(self, header_row, manifest_row, survey_rows, cache_rows):
        """Write fixed preamble blobs. Called once at job start."""
        for row in [header_row, manifest_row, *survey_rows, *cache_rows]:
            h = self._write_blob(row)
            self._preamble_hashes.append(h)
        self._commit("Job started")

    def append_result(self, result_json_row: str, message=""):
        """Append one Result row and create a new commit."""
        h = self._write_blob(result_json_row)
        self._result_hashes.append(h)
        self._commit(message or f"Interview {len(self._result_hashes)} completed")

    def finalize(self, n_results: int):
        """Optional: update manifest with final count, create final commit."""
        # Rewrite manifest blob with correct n_results
        # Update preamble_hashes[1] with new manifest hash
        self._commit("Job completed")

    def _commit(self, message):
        """Build tree from preamble + results, create commit, advance ref."""
        all_hashes = self._preamble_hashes + self._result_hashes
        tree_obj = {"blobs": all_hashes}
        # ... hash tree, create commit, update ref (same as CASRepository.save)

2. Hook into on_task_completed → interview completion

In JobService.on_task_completed() (line 838), when an interview transitions out of RUNNING state, the streaming writer gets notified:

# After interview finalization:
if self._cas_writer and interview_state != InterviewState.RUNNING:
    result = self.build_edsl_result(job_id, interview_id, ...)
    row = json.dumps(result.to_dict(add_edsl_version=True))
    self._cas_writer.append_result(row)

3. Preamble written at job submission

In Runner.submit() or JobService.submit_job(), after the survey is known:

survey_rows = list(survey.to_jsonl_rows())
header = json.dumps({"__header__": True, "edsl_class_name": "Results", ...})
manifest = json.dumps({"n_survey_lines": len(survey_rows), "n_cache_lines": 0, ...})
cas_writer.write_preamble(header, manifest, survey_rows, [])

4. from_jsonl compatibility

Currently from_jsonl already does lines[result_start:] — it doesn't actually use n_results for slicing. So intermediate commits are already loadable as-is. We'd just want to make n_results in the header informational (or update it on finalize).

Commit History Example

commit 5 (HEAD → main): "Job completed (10/10 interviews)"
  tree: [header, manifest, survey×S, result×10]
  parent: commit 4

commit 4: "Interview 9 completed"
  tree: [header, manifest, survey×S, result×9]
  parent: commit 3

...

commit 1: "Job started"
  tree: [header, manifest, survey×S]
  parent: null

Each commit is a valid, loadable Results object with however many interviews have completed so far. git log-like history shows the run's progress.

Blob Deduplication Benefits

• Survey blobs are written once and shared across all commits
• Header/manifest blobs are shared until finalize
• If two interviews happen to produce identical Result dicts (unlikely but possible), they share a blob
• Trees are lightweight (just lists of hashes)

Performance Considerations

• Blob writes: One backend.write() per new result row (~1KB each). Negligible vs LLM call time.
• Tree writes: One new tree per interview completion. Tree is just a JSON array of hashes. For 1000 interviews, the final tree has 1000+ entries (~64KB). Still small.
• Commit writes: One per interview. ~200 bytes each.
• Ref update: One backend.write() per commit. Atomic on filesystem.
• No contention: The main loop is single-threaded for job orchestration; interview completion callbacks are sequential.

Batching Option

For high-throughput jobs (thousands of interviews), we could batch: accumulate N completed results, then commit once. The writer could have a flush_interval or batch_size parameter.

Files to Modify

File	Change
edsl/object_store/streaming_writer.py	New: StreamingCASWriter class
edsl/object_store/__init__.py	Export new class
edsl/runner/runner.py	Pass CAS writer option, wire up preamble at submit, finalize after loop
edsl/runner/service.py	On interview completion, call cas_writer.append_result()
edsl/results/results_serializer.py	Ensure from_jsonl doesn't depend on n_results for correctness (already the case)

What This Does NOT Change

• The existing to_jsonl() / from_jsonl() round-trip — fully preserved
• The existing ObjectStore.save() path — still works for after-the-fact saves
• The build_edsl_results() method — still used for the in-memory Results return value
• The runner's main loop structure — streaming is an opt-in addition

Open Questions

1. Where does the UUID come from? Should the user provide it, or auto-generate at job start? Probably auto-generate and return it on the JobHandle.
2. Should this integrate with ObjectStore metadata index? Probably yes — register the UUID in the index at job start so ObjectStore.list() shows in-progress runs.
3. Cache section: The JSONL format supports inline cache rows. Should we also stream cache entries? Or leave cache empty in streaming mode and populate at finalize?
4. Remote backends: The HttpBackend buffers writes and flushes on ref update. Each commit would trigger a flush (push). For remote streaming, we'd want to batch commits or use a local backend during the run and push at the end.

Verification

• Unit test: Create StreamingCASWriter, write preamble, append 5 results, verify each intermediate commit loads as valid Results via from_jsonl
• Integration test: Run a small job with streaming enabled, kill it mid-run, verify partial results loadable from CAS
• Performance test: Run a 100-interview job with and without streaming, compare wall time