refactor: clarify sync queue boundary

Author: rohans0509

src/codealmanac/cli/dispatch/sync.py

@@ -9,8 +9,8 @@
 from codealmanac.core.errors import ValidationFailed
 from codealmanac.services.sources.models import TranscriptApp
 from codealmanac.workflows.sync.requests import (
-    RunSyncRequest,
-    RunSyncStatusRequest,
+    SyncRequest,
+    SyncStatusRequest,
 )
 
 
@@ -19,7 +19,7 @@ def dispatch_sync(args: argparse.Namespace, app: CodeAlmanac) -> int:
         return dispatch_sync_status(args, app)
     cli_config = load_cli_config(app, args.wiki)
     result = app.workflows.sync.run(
-        RunSyncRequest(
+        SyncRequest(
             repository_name=args.wiki,
             apps=parse_sync_apps(args.source_apps),
             harness=resolve_harness(args.using, cli_config),
@@ -32,7 +32,7 @@ def dispatch_sync(args: argparse.Namespace, app: CodeAlmanac) -> int:
 
 def dispatch_sync_status(args: argparse.Namespace, app: CodeAlmanac) -> int:
     result = app.workflows.sync.status(
-        RunSyncStatusRequest(
+        SyncStatusRequest(
             repository_name=args.wiki,
             apps=parse_sync_apps(args.source_apps),
         )

…/codealmanac/workflows/sync/execution.py src/codealmanac/workflows/sync/queue.pysrc/codealmanac/workflows/sync/execution.py renamed to src/codealmanac/workflows/sync/queue.py src/codealmanac/workflows/sync/execution.py renamed to src/codealmanac/workflows/sync/queue.py

@@ -11,15 +11,15 @@
     SyncStarted,
     SyncWorkItem,
 )
-from codealmanac.workflows.sync.requests import RunSyncRequest
+from codealmanac.workflows.sync.requests import SyncRequest
 from codealmanac.workflows.sync.store import SyncStateStore
 from codealmanac.workflows.sync.summary import (
     skipped_transcript,
     started_repository,
 )
 
 
-class SyncRunExecutor:
+class SyncIngestQueue:
     def __init__(
         self,
         queue: RunQueue,
@@ -30,10 +30,10 @@ def __init__(
 
     def run(
         self,
-        request: RunSyncRequest,
+        request: SyncRequest,
         evaluation: SyncEvaluation,
         now: datetime,
-    ) -> "SyncRunExecutionResult":
+    ) -> "SyncQueueResult":
         started: list[SyncStarted] = []
         skipped = list(evaluation.summary.skipped)
         worker_cwd = None
@@ -66,19 +66,19 @@ def run(
                     for candidate in item.transcripts
                 )
         self.state_store.record_completed(now)
-        return SyncRunExecutionResult(
+        return SyncQueueResult(
             started=tuple(started),
             skipped=tuple(skipped),
         )
 
 
-class SyncRunExecutionResult(CodeAlmanacModel):
+class SyncQueueResult(CodeAlmanacModel):
     started: tuple[SyncStarted, ...]
     skipped: tuple[SyncSkipped, ...]
 
 
 def sync_ingest_request(
-    request: RunSyncRequest,
+    request: SyncRequest,
     item: SyncWorkItem,
 ) -> IngestRequest:
     return IngestRequest(

src/codealmanac/workflows/sync/requests.py

@@ -35,10 +35,10 @@ def positive_interval(cls, value: timedelta) -> timedelta:
         return value
 
 
-class RunSyncStatusRequest(SyncSelectionRequest):
+class SyncStatusRequest(SyncSelectionRequest):
     pass
 
 
-class RunSyncRequest(SyncSelectionRequest):
+class SyncRequest(SyncSelectionRequest):
     harness: HarnessKind
     auto_commit: bool = True

src/codealmanac/workflows/sync/service.py

@@ -4,16 +4,16 @@
 from codealmanac.services.sources.service import SourcesService
 from codealmanac.workflows.run_queue.service import RunQueue
 from codealmanac.workflows.sync.evaluation import SyncEvaluator
-from codealmanac.workflows.sync.execution import SyncRunExecutor
 from codealmanac.workflows.sync.models import (
     SyncEvaluation,
     SyncMode,
     SyncSummary,
 )
+from codealmanac.workflows.sync.queue import SyncIngestQueue
 from codealmanac.workflows.sync.requests import (
-    RunSyncRequest,
-    RunSyncStatusRequest,
+    SyncRequest,
     SyncSelectionRequest,
+    SyncStatusRequest,
 )
 from codealmanac.workflows.sync.store import SyncStateStore
 
@@ -31,15 +31,15 @@ def __init__(
             sources=sources,
             state_store=state_store,
         )
-        self.executor = SyncRunExecutor(
+        self.executor = SyncIngestQueue(
             queue=queue,
             state_store=state_store,
         )
 
-    def status(self, request: RunSyncStatusRequest) -> SyncSummary:
+    def status(self, request: SyncStatusRequest) -> SyncSummary:
         return self.evaluate(request, SyncMode.STATUS).summary
 
-    def run(self, request: RunSyncRequest) -> SyncSummary:
+    def run(self, request: SyncRequest) -> SyncSummary:
         now = request.now or datetime.now(UTC)
         evaluation = self.evaluate(request, SyncMode.SYNC, now=now)
         effects = self.executor.run(request, evaluation, now)

tests/test_architecture.py

@@ -831,7 +831,7 @@ def test_cli_dispatch_edge_is_split_by_command_domain():
     assert "dispatch_lifecycle(args, app)" in dispatch_root
     assert "dispatch_wiki(args, app)" in dispatch_root
     assert "dispatch_admin(args, app)" in dispatch_root
-    assert "RunIngestRequest" not in dispatch_root
+    assert "IngestRequest" not in dispatch_root
     assert "SearchPagesRequest" not in dispatch_root
     assert "AutomationStatusRequest" not in dispatch_root
     assert "DoctorRequest" not in dispatch_root
@@ -972,23 +972,23 @@ def test_cli_lifecycle_dispatch_stays_split_by_command_family():
     lifecycle = (dispatch_path / "lifecycle.py").read_text(encoding="utf-8")
     module_expectations = {
         "build.py": ("InitializeRepositoryRequest", "def dispatch_build("),
-        "operations.py": ("RunIngestRequest", "def dispatch_ingest("),
-        "sync.py": ("RunSyncRequest", "def dispatch_sync("),
+        "operations.py": ("IngestRequest", "def dispatch_ingest("),
+        "sync.py": ("SyncRequest", "def dispatch_sync("),
         "worker.py": ("DrainRunQueueRequest", "def dispatch_run_worker("),
     }
     forbidden_lifecycle_fragments = (
         "InitializeRepositoryRequest",
-        "RunIngestRequest",
+        "IngestRequest",
         "RunGardenRequest",
-        "RunSyncRequest",
-        "RunSyncStatusRequest",
+        "SyncRequest",
+        "SyncStatusRequest",
         "DrainRunQueueRequest",
         "TranscriptApp",
         "ValidationFailed",
         "load_cli_config",
         "resolve_harness",
         "parse_sync_apps",
-        "sync_execution",
+        "sync_queue",
         "render_",
     )
     oversized = []
@@ -1651,13 +1651,13 @@ def test_sync_workflow_policy_stays_out_of_service_orchestration():
     )
 
 
-def test_sync_execution_effects_stay_out_of_service_orchestration():
+def test_sync_queue_effects_stay_out_of_service_orchestration():
     sync_root = SRC_ROOT / "workflows/sync"
     service_text = (sync_root / "service.py").read_text(encoding="utf-8")
     evaluation_text = (sync_root / "evaluation.py").read_text(encoding="utf-8")
-    execution_text = (sync_root / "execution.py").read_text(encoding="utf-8")
+    queue_text = (sync_root / "queue.py").read_text(encoding="utf-8")
     forbidden_service_fragments = (
-        "RunIngestRequest",
+        "IngestRequest",
         "StartedIngestRequest",
         "FinishRunRequest",
         "RunStatus.FAILED",
@@ -1671,7 +1671,7 @@ def test_sync_execution_effects_stay_out_of_service_orchestration():
         "run_started(",
     )
     forbidden_evaluation_fragments = (
-        "RunIngestRequest",
+        "IngestRequest",
         "StartedIngestRequest",
         "FinishRunRequest",
         "RunStatus.FAILED",
@@ -1680,7 +1680,7 @@ def test_sync_execution_effects_stay_out_of_service_orchestration():
         "run_started(",
     )
 
-    assert (sync_root / "execution.py").is_file()
+    assert (sync_root / "queue.py").is_file()
     assert len(service_text.splitlines()) <= 120
     assert [
         fragment
@@ -1692,14 +1692,11 @@ def test_sync_execution_effects_stay_out_of_service_orchestration():
         for fragment in forbidden_evaluation_fragments
         if fragment in evaluation_text
     ] == []
-    assert "class SyncRunExecutor" in execution_text
-    assert "RunIngestRequest" in execution_text
-    assert "StartedIngestRequest" in execution_text
-    assert "FinishRunRequest" in execution_text
-    assert "pending_entry(" in execution_text
-    assert "absorbed_entry(" in execution_text
-    assert "queue_ingest(" in execution_text
-    assert "spawn_worker(" in execution_text
+    assert "class SyncIngestQueue" in queue_text
+    assert "IngestRequest" in queue_text
+    assert "queue_ingest(" in queue_text
+    assert "spawn_worker(" in queue_text
+    assert "record_completed(" in queue_text
 
 
 def test_viewer_jobs_surface_stays_read_only():

tests/test_sync_workflow.py

@@ -3,15 +3,15 @@
 from pathlib import Path
 
 from codealmanac.app import create_app
-from codealmanac.settings import AppConfig
 from codealmanac.services.harnesses.models import HarnessKind
 from codealmanac.services.repositories.requests import InitializeRepositoryRequest
 from codealmanac.services.runs.models import RunKind, RunWorkerSpawnResult
 from codealmanac.services.runs.requests import ReadRunSpecRequest, SpawnRunWorkerRequest
 from codealmanac.services.sources.models import TranscriptApp, TranscriptCandidate
 from codealmanac.services.sources.requests import DiscoverTranscriptsRequest
+from codealmanac.settings import AppConfig
 from codealmanac.workflows.sync.models import SyncMode
-from codealmanac.workflows.sync.requests import RunSyncRequest, RunSyncStatusRequest
+from codealmanac.workflows.sync.requests import SyncRequest, SyncStatusRequest
 
 
 class FakeTranscriptDiscoveryAdapter:
@@ -63,7 +63,7 @@ def test_sync_status_groups_active_transcripts_by_repository(
     app.workflows.build.initialize(InitializeRepositoryRequest(path=repo))
 
     summary = app.workflows.sync.status(
-        RunSyncStatusRequest(
+        SyncStatusRequest(
             cwd=repo,
             apps=(TranscriptApp.CODEX,),
             now=current_time(),
@@ -93,7 +93,7 @@ def test_sync_uses_exact_registered_cwd_without_root_hopping(
     app.workflows.build.initialize(InitializeRepositoryRequest(path=repo))
 
     summary = app.workflows.sync.status(
-        RunSyncStatusRequest(
+        SyncStatusRequest(
             cwd=repo,
             apps=(TranscriptApp.CODEX,),
             now=current_time(),
@@ -130,7 +130,7 @@ def test_sync_run_queues_one_ingest_run_per_repository_and_records_scan_time(
     app.workflows.build.initialize(InitializeRepositoryRequest(path=repo))
 
     summary = app.workflows.sync.run(
-        RunSyncRequest(
+        SyncRequest(
             cwd=repo,
             apps=(TranscriptApp.CODEX,),
             harness=HarnessKind.CODEX,
@@ -163,7 +163,7 @@ def test_sync_run_queues_one_ingest_run_per_repository_and_records_scan_time(
     assert sync_completed_at(database_path) == current_time().isoformat()
 
     next_status = app.workflows.sync.status(
-        RunSyncStatusRequest(
+        SyncStatusRequest(
             cwd=repo,
             apps=(TranscriptApp.CODEX,),
             now=current_time() + timedelta(minutes=1),
@@ -189,7 +189,7 @@ def test_sync_advances_scan_time_even_when_queueing_fails(
     app.workflows.build.initialize(InitializeRepositoryRequest(path=repo))
 
     summary = app.workflows.sync.run(
-        RunSyncRequest(
+        SyncRequest(
             cwd=repo,
             apps=(TranscriptApp.CODEX,),
             harness=HarnessKind.CODEX,
@@ -224,7 +224,7 @@ def test_sync_uses_last_completed_time_before_interval_fallback(
     write_sync_completed_at(database_path, previous)
 
     summary = app.workflows.sync.status(
-        RunSyncStatusRequest(
+        SyncStatusRequest(
             cwd=repo,
             apps=(TranscriptApp.CODEX,),
             now=current_time(),