From ccdcf9bdf7d5181422c7650112a774825a637aa0 Mon Sep 17 00:00:00 2001 From: Mark Slater Date: Tue, 21 Jul 2026 15:50:41 -0700 Subject: [PATCH] fix sync self-ingesting worker transcripts --- src/codealmanac/app.py | 1 + src/codealmanac/services/runs/service.py | 8 ++ src/codealmanac/services/runs/store.py | 11 +++ src/codealmanac/workflows/sync/evaluation.py | 21 +++++ src/codealmanac/workflows/sync/service.py | 3 + tests/test_sync_workflow.py | 92 ++++++++++++++++++-- 6 files changed, 129 insertions(+), 7 deletions(-) diff --git a/src/codealmanac/app.py b/src/codealmanac/app.py index ca69e029..c1c8fba7 100644 --- a/src/codealmanac/app.py +++ b/src/codealmanac/app.py @@ -344,6 +344,7 @@ def create_workflows( sync = SyncWorkflow( services.repositories, services.sources, + services.runs, queue, SyncStateStore(services.local_state.database_path), ) diff --git a/src/codealmanac/services/runs/service.py b/src/codealmanac/services/runs/service.py index 60c2b133..285afeb1 100644 --- a/src/codealmanac/services/runs/service.py +++ b/src/codealmanac/services/runs/service.py @@ -3,6 +3,7 @@ from datetime import UTC, datetime from codealmanac.core.errors import NotFoundError +from codealmanac.services.harnesses.models import HarnessTranscriptRef from codealmanac.services.repositories.models import Repository, RepositoryName from codealmanac.services.repositories.requests import SelectRepositoryRequest from codealmanac.services.repositories.service import RepositoriesService @@ -75,6 +76,13 @@ def list(self, request: ListRunsRequest) -> tuple[RunRecord, ...]: repository_id = None if repository is None else repository.repository_id return self.store.list(request.limit, repository_id=repository_id) + def list_harness_transcripts( + self, + repository_id: str, + ) -> tuple[HarnessTranscriptRef, ...]: + self.repositories.get(repository_id) + return self.store.list_harness_transcripts(repository_id) + def show(self, request: ShowRunRequest) -> RunRecord: record = self.store.read(request.run_id) self.require_run_matches_repository(record, request.repository_name) diff --git a/src/codealmanac/services/runs/store.py b/src/codealmanac/services/runs/store.py index ac5ad822..869dc975 100644 --- a/src/codealmanac/services/runs/store.py +++ b/src/codealmanac/services/runs/store.py @@ -111,6 +111,17 @@ def list( with self.connect() as connection: return list_run_records(connection, limit, repository_id) + def list_harness_transcripts( + self, + repository_id: str, + ) -> tuple[HarnessTranscriptRef, ...]: + records = self.list(limit=None, repository_id=repository_id) + return tuple( + record.harness_transcript + for record in records + if record.harness_transcript is not None + ) + def read(self, run_id: str) -> RunRecord: with self.connect() as connection: record = read_run_record(connection, run_id) diff --git a/src/codealmanac/workflows/sync/evaluation.py b/src/codealmanac/workflows/sync/evaluation.py index 8791a96a..00a96ee4 100644 --- a/src/codealmanac/workflows/sync/evaluation.py +++ b/src/codealmanac/workflows/sync/evaluation.py @@ -6,6 +6,7 @@ from codealmanac.services.repositories.models import Repository from codealmanac.services.repositories.requests import SelectRepositoryRequest from codealmanac.services.repositories.service import RepositoriesService +from codealmanac.services.runs.service import RunsService from codealmanac.services.sources.models import TranscriptCandidate from codealmanac.services.sources.requests import DiscoverTranscriptsRequest from codealmanac.services.sources.service import SourcesService @@ -29,10 +30,12 @@ def __init__( self, repositories: RepositoriesService, sources: SourcesService, + runs: RunsService, state_store: SyncStateStore, ): self.repositories = repositories self.sources = sources + self.runs = runs self.state_store = state_store def evaluate( @@ -62,6 +65,10 @@ def evaluate( normalize_path(repository.root_path): repository for repository in selected_repositories } + owned_sessions_by_repository_id = { + repository.repository_id: self.owned_sessions(repository) + for repository in selected_repositories + } for transcript in transcripts: repository = repositories_by_path.get(normalize_path(transcript.cwd)) if repository is None: @@ -70,6 +77,12 @@ def evaluate( if transcript.modified_at < active_since: skipped.append(skipped_transcript(transcript, "inactive")) continue + if ( + transcript.app.value, + transcript.session_id, + ) in owned_sessions_by_repository_id[repository.repository_id]: + skipped.append(skipped_transcript(transcript, "codealmanac-run")) + continue transcripts_by_repository_id[repository.repository_id].append(transcript) repository_ingests = tuple( @@ -109,6 +122,14 @@ def selected_repositories( ) return (repository,) + def owned_sessions(self, repository: Repository) -> frozenset[tuple[str, str]]: + return frozenset( + (transcript.kind.value, transcript.session_id) + for transcript in self.runs.list_harness_transcripts( + repository.repository_id + ) + ) + def transcript_sort_key(candidate: TranscriptCandidate) -> tuple[str, str, str]: return ( diff --git a/src/codealmanac/workflows/sync/service.py b/src/codealmanac/workflows/sync/service.py index 9560c433..080187dc 100644 --- a/src/codealmanac/workflows/sync/service.py +++ b/src/codealmanac/workflows/sync/service.py @@ -1,6 +1,7 @@ from datetime import UTC, datetime from codealmanac.services.repositories.service import RepositoriesService +from codealmanac.services.runs.service import RunsService from codealmanac.services.sources.service import SourcesService from codealmanac.workflows.run_queue.service import RunQueue from codealmanac.workflows.sync.evaluation import SyncEvaluator @@ -23,12 +24,14 @@ def __init__( self, repositories: RepositoriesService, sources: SourcesService, + runs: RunsService, queue: RunQueue, state_store: SyncStateStore, ): self.evaluator = SyncEvaluator( repositories=repositories, sources=sources, + runs=runs, state_store=state_store, ) self.executor = SyncIngestQueue( diff --git a/tests/test_sync_workflow.py b/tests/test_sync_workflow.py index 60bbbbe4..be230a85 100644 --- a/tests/test_sync_workflow.py +++ b/tests/test_sync_workflow.py @@ -2,12 +2,18 @@ from datetime import UTC, datetime, timedelta from pathlib import Path +import pytest from conftest import initialize_repository from codealmanac.app import create_app -from codealmanac.services.harnesses.models import HarnessKind +from codealmanac.services.harnesses.models import HarnessKind, HarnessTranscriptRef from codealmanac.services.runs.models import RunKind, RunWorkerSpawnResult -from codealmanac.services.runs.requests import ReadRunSpecRequest, SpawnRunWorkerRequest +from codealmanac.services.runs.requests import ( + ReadRunSpecRequest, + RecordRunHarnessTranscriptRequest, + SpawnRunWorkerRequest, + StartRunRequest, +) from codealmanac.services.sources.models import TranscriptApp, TranscriptCandidate from codealmanac.services.sources.requests import DiscoverTranscriptsRequest from codealmanac.settings import AppConfig @@ -16,9 +22,12 @@ class FakeTranscriptDiscoveryAdapter: - app = TranscriptApp.CODEX - - def __init__(self, candidates: tuple[TranscriptCandidate, ...]): + def __init__( + self, + candidates: tuple[TranscriptCandidate, ...], + app: TranscriptApp = TranscriptApp.CODEX, + ): + self.app = app self.candidates = candidates self.requests: list[DiscoverTranscriptsRequest] = [] @@ -104,6 +113,73 @@ def test_sync_uses_exact_registered_cwd_without_root_hopping( assert summary.skipped[0].reason == "unregistered-cwd" +@pytest.mark.parametrize( + ("transcript_app", "harness_kind"), + ( + (TranscriptApp.CLAUDE, HarnessKind.CLAUDE), + (TranscriptApp.CODEX, HarnessKind.CODEX), + ), +) +def test_sync_skips_transcripts_from_codealmanac_runs( + tmp_path: Path, + isolated_home: Path, + transcript_app: TranscriptApp, + harness_kind: HarnessKind, +): + repo = initialized_repo(tmp_path, "repo") + own_work = transcript_candidate( + repo, + tmp_path / "own-work.jsonl", + current_time(), + app=transcript_app, + session_id=f"{transcript_app.value}-codealmanac-session", + ) + user_work = transcript_candidate( + repo, + tmp_path / "user-work.jsonl", + current_time(), + app=transcript_app, + session_id=f"{transcript_app.value}-user-session", + ) + app, _, spawner = app_with_sync( + isolated_home, + candidates=(own_work, user_work), + app=transcript_app, + ) + repository = initialize_repository(app, path=repo) + run = app.runs.start( + StartRunRequest( + repository_id=repository.repository_id, + kind=RunKind.INGEST, + title="Sync own work", + ) + ) + app.runs.record_harness_transcript( + RecordRunHarnessTranscriptRequest( + run_id=run.run_id, + transcript=HarnessTranscriptRef( + kind=harness_kind, + session_id=f"{transcript_app.value}-codealmanac-session", + ), + ) + ) + + summary = app.workflows.sync.status( + SyncStatusRequest( + apps=(transcript_app,), + now=current_time(), + ) + ) + + assert summary.eligible == 1 + assert summary.ready[0].transcript_paths == (user_work.transcript_path,) + assert summary.skipped[0].reason == "codealmanac-run" + assert summary.skipped[0].session_id == ( + f"{transcript_app.value}-codealmanac-session" + ) + assert spawner.requests == [] + + def test_sync_run_queues_one_ingest_run_per_repository_and_records_scan_time( tmp_path: Path, isolated_home: Path, @@ -240,10 +316,11 @@ def app_with_sync( isolated_home: Path, *, candidates: tuple[TranscriptCandidate, ...], + app: TranscriptApp = TranscriptApp.CODEX, database_path: Path | None = None, spawner: SyncWorkerSpawner | None = None, ): - adapter = FakeTranscriptDiscoveryAdapter(candidates) + adapter = FakeTranscriptDiscoveryAdapter(candidates, app=app) selected_spawner = spawner or SyncWorkerSpawner() app = create_app( AppConfig( @@ -261,11 +338,12 @@ def transcript_candidate( path: Path, modified_at: datetime, *, + app: TranscriptApp = TranscriptApp.CODEX, session_id: str = "session-1", ) -> TranscriptCandidate: path.write_text("transcript\n", encoding="utf-8") return TranscriptCandidate( - app=TranscriptApp.CODEX, + app=app, session_id=session_id, transcript_path=path, cwd=repo,