feat: work pipeline spine (#1960)

scripts/_ghost_wiring_manifest.txt

@@ -37,3 +37,4 @@ ENFORCED build_distributed_backend_services #1966 -- called in api/auto_wire._re
 ENFORCED DeadLetterConsumer #1966 -- constructed by workers.backend_services.build_distributed_backend_services; drains the dead subject and fails exhausted tasks (no-loss closure)
 ENFORCED SeenClaimsPruner #1966 -- constructed by workers.backend_services.build_distributed_backend_services; bounds the seen_claims dedup table
 ENFORCED WorkerHeartbeatSubscriber #1966 -- constructed by workers.backend_services.build_distributed_backend_services; surfaces worker liveness in the log pipeline
+ENFORCED build_work_pipeline #1960 -- called by workers.runtime_builder._build_runtime_work_pipeline behind the provider-present switch; composes the work spine (intake -> projects -> solo/team -> coordination metrics)

src/synthorg/api/app.py

@@ -91,6 +91,7 @@
 from synthorg.core.clock import SystemClock
 from synthorg.core.error_taxonomy import set_error_docs_base_url
 from synthorg.engine.coordination.service import MultiAgentCoordinator  # noqa: TC001
+from synthorg.engine.pipeline.protocol import WorkPipeline  # noqa: TC001
 from synthorg.engine.review_gate import ReviewGateService
 from synthorg.engine.task_engine import TaskEngine  # noqa: TC001
 from synthorg.hr.performance.tracker import PerformanceTracker  # noqa: TC001
@@ -253,6 +254,7 @@ def create_app(  # noqa: C901, PLR0912, PLR0913, PLR0915
     auth_service: AuthService | None = None,
     task_engine: TaskEngine | None = None,
     coordinator: MultiAgentCoordinator | None = None,
+    work_pipeline: WorkPipeline | None = None,
     agent_registry: AgentRegistryService | None = None,
     meeting_orchestrator: MeetingOrchestrator | None = None,
     meeting_scheduler: MeetingScheduler | None = None,
@@ -287,6 +289,8 @@ def create_app(  # noqa: C901, PLR0912, PLR0913, PLR0915
         auth_service: Pre-built auth service (for testing).
         task_engine: Centralized task state engine.
         coordinator: Multi-agent coordinator.
+        work_pipeline: Work pipeline spine (injected double wins over
+            the boot-autowired one).
         agent_registry: Agent registry service.
         meeting_orchestrator: Meeting orchestrator.
         meeting_scheduler: Meeting scheduler.
@@ -546,6 +550,7 @@ def create_app(  # noqa: C901, PLR0912, PLR0913, PLR0915
         auth_service=auth_service,
         task_engine=task_engine,
         coordinator=coordinator,
+        work_pipeline=work_pipeline,
         agent_registry=agent_registry,
         meeting_orchestrator=meeting_orchestrator,
         meeting_scheduler=meeting_scheduler,
@@ -1062,6 +1067,11 @@ async def _install_runtime_services() -> None:
         # coordinator is kept and the built one is a logged no-op then.
         if services.coordinator is not None:
             app_state.set_coordinator_if_absent(services.coordinator)
+        # Same injection-over-autowire rule for the work pipeline spine:
+        # an injected ``create_app(work_pipeline=)`` is kept, the built
+        # one is a logged no-op then.
+        if services.work_pipeline is not None:
+            app_state.set_work_pipeline_if_absent(services.work_pipeline)
         _runtime_services_installed = True
 
     startup = [*startup, _install_runtime_services]

src/synthorg/api/controllers/setup/agent_helpers.py

@@ -145,12 +145,13 @@ async def post_setup_reinit(app_state: AppState) -> None:
 
 
 async def _rebuild_runtime_services(app_state: AppState) -> None:
-    """Rebuild and hot-swap both runtime services (worker execution + coordinator).
+    """Rebuild and hot-swap the runtime services.
 
     Invoked after provider configuration to bring the full agent runtime
-    online without a process restart. Swaps the worker execution service
-    and the multi-agent coordinator so ``/coordinate`` stops returning
-    503 and the worker-callable execute endpoint uses the new provider.
+    online without a process restart. Swaps the worker execution
+    service, the multi-agent coordinator, and the work pipeline spine so
+    ``/coordinate`` stops returning 503, the worker-callable execute
+    endpoint uses the new provider, and work routing comes online.
 
     Raises on failure (either a typed ``RuntimeServicesBuildError`` or a
     wrapped exception) so :func:`post_setup_reinit` can keep the setup flag
@@ -175,6 +176,8 @@ async def _rebuild_runtime_services(app_state: AppState) -> None:
         )
         if services.coordinator is not None:
             app_state.swap_coordinator(services.coordinator)
+        if services.work_pipeline is not None:
+            app_state.swap_work_pipeline(services.work_pipeline)
     except MemoryError, RecursionError:
         raise
     except RuntimeServicesBuildError:

src/synthorg/api/state.py

@@ -63,6 +63,7 @@
 from synthorg.core.domain_errors import ServiceUnavailableError
 from synthorg.engine.approval_gate import ApprovalGate  # noqa: TC001
 from synthorg.engine.coordination.service import MultiAgentCoordinator  # noqa: TC001
+from synthorg.engine.pipeline.protocol import WorkPipeline  # noqa: TC001
 from synthorg.engine.review_gate import ReviewGateService  # noqa: TC001
 from synthorg.engine.task_engine import TaskEngine  # noqa: TC001
 from synthorg.engine.workflow.ceremony_scheduler import CeremonyScheduler  # noqa: TC001
@@ -305,6 +306,7 @@ class AppState(AppStateServicesMixin):
         "_webhook_event_bridge",
         "_webhook_replay_protector",
         "_webhook_service",
+        "_work_pipeline",
         "_worker_execution_service",
         "_workers_bridge_config",
         "_workers_bridge_config_lock",
@@ -332,6 +334,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         task_engine: TaskEngine | None = None,
         approval_gate: ApprovalGate | None = None,
         coordinator: MultiAgentCoordinator | None = None,
+        work_pipeline: WorkPipeline | None = None,
         agent_registry: AgentRegistryService | None = None,
         performance_tracker: PerformanceTracker | None = None,
         meeting_orchestrator: MeetingOrchestrator | None = None,
@@ -388,6 +391,7 @@ def __init__(  # noqa: PLR0913, PLR0915
         self._distributed_task_queue: JetStreamTaskQueue | None = None
         self._distributed_backend_services: DistributedBackendServices | None = None
         self._coordinator = coordinator
+        self._work_pipeline = work_pipeline
         self._agent_registry = agent_registry
         self._performance_tracker = performance_tracker
         self._trust_service = trust_service
@@ -1291,6 +1295,94 @@ def swap_coordinator(self, coordinator: MultiAgentCoordinator) -> None:
                 transition=transition,
             )
 
+    @property
+    def work_pipeline(self) -> WorkPipeline:
+        """Return the work pipeline spine or raise 503."""
+        return self._require_service(self._work_pipeline, "work_pipeline")
+
+    @property
+    def has_work_pipeline(self) -> bool:
+        """Check whether the work pipeline spine is configured.
+
+        Unsynchronised by design, identical to :meth:`has_coordinator`:
+        a single reference read is atomic under CPython and
+        ``swap_work_pipeline`` only reassigns one already-set pipeline
+        for another, so a concurrent reader sees a consistent
+        old-or-new instance. The only ``None -> set`` flip happens once
+        at boot before HTTP traffic.
+        """
+        return self._work_pipeline is not None
+
+    def set_work_pipeline(self, work_pipeline: WorkPipeline) -> None:
+        """Attach the work pipeline spine (once-only, boot only).
+
+        Once-only: a second set raises, matching the ``coordinator``
+        seam. The boot runtime-services hook uses
+        :meth:`set_work_pipeline_if_absent` so an explicitly injected
+        pipeline wins; hot-reload after setup uses
+        :meth:`swap_work_pipeline`.
+        """
+        self._set_once("_work_pipeline", work_pipeline, "Work pipeline")
+
+    def set_work_pipeline_if_absent(
+        self,
+        work_pipeline: WorkPipeline,
+    ) -> bool:
+        """Attach the work pipeline only if none is configured (atomic).
+
+        The boot runtime-services hook calls this unconditionally
+        behind the provider-present switch so work routing comes online
+        once a provider and intake are configured. An explicitly
+        injected pipeline (constructor ``work_pipeline=``) is already
+        set and wins: this is a logged no-op then. The check-and-set is
+        atomic under ``_lazy_service_lock`` so the boot install cannot
+        race a concurrent ``swap_work_pipeline`` or property read.
+
+        Returns:
+            ``True`` if this call installed the pipeline, ``False`` if
+            one was already configured (injected) and kept.
+        """
+        with self._lazy_service_lock:
+            if self._work_pipeline is not None:
+                logger.info(
+                    API_APP_STARTUP,
+                    service="work_pipeline",
+                    transition="skipped_injected",
+                )
+                return False
+            self._work_pipeline = work_pipeline
+            logger.info(
+                API_APP_STARTUP,
+                service="work_pipeline",
+                transition="attached",
+            )
+            return True
+
+    def swap_work_pipeline(self, work_pipeline: WorkPipeline) -> None:
+        """Replace the work pipeline spine (hot-reload).
+
+        Distinct from :meth:`set_work_pipeline`, which is once-only:
+        this replaces an already-wired pipeline so a provider
+        configured against an empty-company start brings the work spine
+        online without a restart (``post_setup_reinit``). Holds
+        ``_lazy_service_lock`` so the write is synchronised against
+        concurrent property reads, mirroring :meth:`swap_coordinator`.
+        """
+        with self._lazy_service_lock:
+            previous = self._work_pipeline
+            if previous is work_pipeline:
+                transition = "noop"
+            elif previous is None:
+                transition = "attached"
+            else:
+                transition = "replaced"
+            self._work_pipeline = work_pipeline
+            logger.info(
+                API_APP_STARTUP,
+                service="work_pipeline",
+                transition=transition,
+            )
+
     @property
     def performance_tracker(self) -> PerformanceTracker:
         """Return performance tracker or raise 503."""

src/synthorg/engine/coordination/factory.py

@@ -173,6 +173,7 @@ def build_coordinator(  # noqa: PLR0913
     performance_tracker: PerformanceTracker | None = None,
     routing_scorer_config: RoutingScorerConfig | None = None,
     coordination_metrics_collector: CoordinationMetricsCollector | None = None,
+    scorer: AgentTaskScorer | None = None,
 ) -> MultiAgentCoordinator:
     """Build a fully wired :class:`MultiAgentCoordinator`.
 
@@ -216,6 +217,11 @@ def build_coordinator(  # noqa: PLR0913
             invokes post-completion to compute and record the
             multi-agent metrics. ``None`` disables collection (the
             ``/coordination/metrics`` API stays empty).
+        scorer: Pre-built agent-task scorer to share with the work
+            pipeline's solo-path selection so both routing surfaces
+            use one instance. ``None`` builds one from
+            *routing_scorer_config* / *task_assignment_config* as
+            before.
 
     Returns:
         A fully constructed ``MultiAgentCoordinator``.
@@ -224,10 +230,11 @@ def build_coordinator(  # noqa: PLR0913
     strategy = _build_decomposition_strategy(provider, decomposition_model)
     decomposition_service = DecompositionService(strategy, classifier)
 
-    if routing_scorer_config is None:
-        scorer = AgentTaskScorer(min_score=task_assignment_config.min_score)
-    else:
-        scorer = AgentTaskScorer(config=routing_scorer_config)
+    if scorer is None:
+        if routing_scorer_config is None:
+            scorer = AgentTaskScorer(min_score=task_assignment_config.min_score)
+        else:
+            scorer = AgentTaskScorer(config=routing_scorer_config)
     topology_selector = TopologySelector(config.auto_topology_rules)
     routing_service = TaskRoutingService(scorer, topology_selector)
 

src/synthorg/engine/pipeline/__init__.py

@@ -0,0 +1,32 @@
+"""Work pipeline spine.
+
+The single coherent path from "work enters" to "agents execute it":
+intake -> projects -> decompose (solo-vs-team verdict) -> solo or
+team execution -> coordination metrics. The spine is the one
+integration point every entry adapter feeds via a typed
+:class:`WorkItem`.
+"""
+
+from synthorg.engine.pipeline.factory import build_work_pipeline
+from synthorg.engine.pipeline.models import (
+    ExecutionPath,
+    RoutingVerdict,
+    WorkItem,
+    WorkPhaseResult,
+    WorkPipelineResult,
+    WorkSource,
+)
+from synthorg.engine.pipeline.protocol import WorkPipeline
+from synthorg.engine.pipeline.service import DefaultWorkPipeline
+
+__all__ = [
+    "DefaultWorkPipeline",
+    "ExecutionPath",
+    "RoutingVerdict",
+    "WorkItem",
+    "WorkPhaseResult",
+    "WorkPipeline",
+    "WorkPipelineResult",
+    "WorkSource",
+    "build_work_pipeline",
+]

src/synthorg/engine/pipeline/errors.py

@@ -0,0 +1,72 @@
+"""Work pipeline domain errors.
+
+All inherit from :class:`WorkPipelineError`, itself a
+:class:`DomainError`, so the existing
+``(DomainError, handle_domain_error)`` registration in
+:mod:`synthorg.api.exception_handlers` dispatches every subclass via
+MRO; no bespoke handler is required.
+"""
+
+from typing import ClassVar
+
+from synthorg.core.domain_errors import DomainError
+from synthorg.core.error_taxonomy import ErrorCategory, ErrorCode
+
+
+class WorkPipelineError(DomainError):
+    """Base for every work pipeline failure (500 unless overridden)."""
+
+    default_message: ClassVar[str] = "Work pipeline failure"
+    error_category: ClassVar[ErrorCategory] = ErrorCategory.INTERNAL
+    error_code: ClassVar[ErrorCode] = ErrorCode.INTERNAL_ERROR
+    status_code: ClassVar[int] = 500
+
+
+class WorkIntakeRejectedError(WorkPipelineError):
+    """Raised when the intake strategy rejects the submitted work (422)."""
+
+    default_message: ClassVar[str] = "Work was rejected at intake"
+    error_category: ClassVar[ErrorCategory] = ErrorCategory.VALIDATION
+    error_code: ClassVar[ErrorCode] = ErrorCode.VALIDATION_ERROR
+    status_code: ClassVar[int] = 422
+
+
+class WorkProjectNotFoundError(WorkPipelineError):
+    """Raised when the work item's project cannot be resolved (404)."""
+
+    default_message: ClassVar[str] = "Work item project not found"
+    error_category: ClassVar[ErrorCategory] = ErrorCategory.NOT_FOUND
+    error_code: ClassVar[ErrorCode] = ErrorCode.PROJECT_NOT_FOUND
+    status_code: ClassVar[int] = 404
+
+
+class WorkRoutingUndecidableError(WorkPipelineError):
+    """Raised when the spine cannot route the work to an executor (500).
+
+    Covers an unknown routing-policy discriminator, an empty active
+    agent pool, and no agent scoring above the routing threshold for
+    the solo (leaf) path.
+    """
+
+    default_message: ClassVar[str] = "Work could not be routed to an executor"
+    error_category: ClassVar[ErrorCategory] = ErrorCategory.INTERNAL
+    error_code: ClassVar[ErrorCode] = ErrorCode.INTERNAL_ERROR
+    status_code: ClassVar[int] = 500
+
+
+class WorkPipelineTeamPathUnavailableError(WorkPipelineError):
+    """Raised when splittable work needs the coordinator but it is absent.
+
+    The empty-company (no-provider) boot leaves the coordinator
+    unconfigured; routing a ``SPLITTABLE`` verdict then honestly 503s
+    rather than silently degrading to a single agent.
+    """
+
+    default_message: ClassVar[str] = (
+        "Multi-agent coordinator is not configured; "
+        "configure a provider to run splittable work"
+    )
+    error_category: ClassVar[ErrorCategory] = ErrorCategory.INTERNAL
+    error_code: ClassVar[ErrorCode] = ErrorCode.SERVICE_UNAVAILABLE
+    retryable: ClassVar[bool] = True
+    status_code: ClassVar[int] = 503