Skip to content

Pluggable Subsystems: Canonical Examples

On-demand reference. The rule in CLAUDE.md is: new cross-cutting subsystems follow a protocol + strategy + factory + config discriminator pattern, with safe defaults so the behaviour is opt-in. This page catalogues the canonical implementations.

Pattern recap

  • Define a Protocol interface.
  • Ship concrete strategies that implement it.
  • Register them in a factory keyed by a config discriminator.
  • Plumb the active selection through frozen config.
  • Ship safe defaults so nothing ever silently regresses.

Registries

Three registry classes replace the hand-rolled if config.type == "...": ... elif ... chains every factory used to carry. Each is immutable after construction (MappingProxyType-backed) and emits structured registry.* events for built / lookup / failure paths.

  • synthorg.core.registry.StrategyRegistry[T]: generic strategy dispatch keyed by a config.type discriminator. Used across the codebase by factories for pruning, propagation, identity store, evolution triggers, evolution proposers, execution loops, procedural capture, notification sinks, secret backends, sandbox backends, per-op rate-limit stores, per-op inflight stores, memory-consolidation strategies (selector + op composite, ADR-0005), risk-tier classifiers (timeout policy seam), and autonomy change strategies (promotion plugin).
  • synthorg.persistence.registry.PersistenceBackendRegistry: domain-specific dispatch keyed by PersistenceConfig.backend; preserves the lazy import of the optional postgres extra.
  • synthorg.memory.registry.MemoryBackendRegistry: domain-specific dispatch keyed by CompanyMemoryConfig.backend; the composite-backend child loop reuses a separate "leaf" registry to keep the wiring acyclic.

Each subsystem still owns its config discriminator; the registries replace only the dispatch step that translates the discriminator into a constructor call.

Canonical examples

Classification pipeline

  • engine/classification/protocol.py: Detector, ScopedContextLoader, ClassificationSink.
  • budget/coordination_config.py: dispatcher.

Verification graders

  • engine/quality/decomposer_protocol.py: CriteriaDecomposer.
  • engine/quality/grader_protocol.py: RubricGrader.
  • engine/quality/verification_factory.py + engine/quality/verification_config.py.

Chief of Staff

  • meta/chief_of_staff/protocol.py: OutcomeStore, ConfidenceAdjuster, OrgInflectionSink, AlertSink.
  • meta/chief_of_staff/config.py: discriminator.
  • meta/factory.py::build_confidence_adjuster().

Analytics / telemetry

  • meta/telemetry/protocol.py: AnalyticsEmitter, AnalyticsCollector, RecommendationProvider.
  • meta/telemetry/config.py: discriminator.
  • meta/telemetry/factory.py::build_analytics_emitter().

Rollout strategies

  • meta/rollout/roster.py: OrgRoster.
  • meta/rollout/group_aggregator.py: GroupSignalAggregator.
  • meta/rollout/inverse_dispatch.py: RollbackHandler + 6 mutator protocols.
  • meta/factory.py::build_rollout_strategies() + build_rollback_executor().
  • All plumbed through frozen SelfImprovementConfig, with safe defaults (SystemClock from synthorg.core.clock, NoOpOrgRoster, null aggregator) so the behaviour is opt-in.

Rollback mutators

Concrete implementations of the six mutator protocols live under meta/rollout/mutators/:

  • SettingsServiceConfigMutator (mutators/config_mutator.py): backs ConfigMutator with SettingsService.set. Dotted target ("<namespace>.<key>"); compose_set settings surface as RollbackMutationDeniedError.
  • PrincipleOverridePromptMutator (mutators/prompt_mutator.py): backs PromptMutator with PrincipleOverrideRepository. Persists override rows that the prompt-build path overlays onto matching principles by id via the cached PrincipleOverrideProvider (engine/strategy/principles.py::load_and_merge, wired at boot in meta_apply_wiring.py). Schema: the principle_overrides table in persistence/sqlite/schema.sql (and Postgres twin).
  • RoutedArchitectureMutator (mutators/architecture_mutator.py): backs ArchitectureMutator with a per-target-type adapter registry. Target format "<type>:<id>" (or "<type>:<id>:<sub_id>"); operators register adapters per type (role, department, workflow, etc.) without touching the executor. Unknown prefixes raise UnknownArchitectureTargetError. mutators/architecture_adapters.py::build_architecture_adapters(role_repo=..., department_service=..., workflow_service=..., clock=...) assembles the boot role / department / workflow adapters that route a revert_architecture operation's apply-time-captured previous_value back to the matching durable store.
  • WorkspaceCodeMutator (mutators/code_mutator.py): backs CodeMutator with atomic filesystem writes (tempfile.mkstemp + Path.replace) inside a workspace bounded by PathValidator so revert_code cannot escape via traversal.
  • ActivePrincipleRemovalMutator (mutators/principle_removal_mutator.py): backs PrincipleRemovalMutator with ActivePrincipleRepository.delete plus an on_principle_removed provider-refresh hook. Reverses a prompt-ADD apply by deleting the created active-principle row (the correct inverse of an add, which an override overlay cannot express).
  • BranchRevertMutator (mutators/branch_mutator.py): backs BranchMutator with HttpGitHubClient.delete_branch. Reverses a code apply by deleting the generated feature branch; only wired when the code-modification GitHub token + repo are configured.

Domain errors live at meta/errors.py::RollbackMutationDeniedError (409) and UnknownArchitectureTargetError. Wire-up assembles via meta/factory.py::build_rollback_executor(config_mutator=..., prompt_mutator=..., architecture_mutator=..., code_mutator=..., principle_removal_mutator=..., branch_mutator=...). The executor is boot-wired (off by default with the self-improvement feature): api/lifecycle_helpers/meta_apply_wiring.py::_wire builds every mutator and threads the executor into SelfImprovementService, which dispatches the applier-materialised inverse operations automatically on a post-rollout regression.

API rate limits

  • api/rate_limits/protocol.py: SlidingWindowStore.
  • api/rate_limits/in_memory.py.
  • api/rate_limits/config.py::PerOpRateLimitConfig: discriminator.
  • api/rate_limits/factory.py::build_sliding_window_store().

API per-op concurrency

  • api/rate_limits/inflight_protocol.py: InflightStore.
  • api/rate_limits/in_memory_inflight.py.
  • api/rate_limits/inflight_config.py::PerOpConcurrencyConfig: discriminator.
  • api/rate_limits/inflight_factory.py::build_inflight_store().
  • api/rate_limits/inflight_middleware.py::PerOpConcurrencyMiddleware (Litestar middleware that reads opt[per_op_concurrency] from each route handler).

Assignment ranking and pool filtering

  • engine/assignment/protocol.py: TaskAssignmentStrategy (the public Protocol; strategies are still selected by the strategy config string).
  • engine/assignment/pool_filter_protocol.py: CandidatePoolFilter (pre-scoring narrowing of available_agents; IdentityPoolFilter is the default, HierarchicalPoolFilter narrows to subordinates of the task's delegator).
  • engine/assignment/ranker_protocol.py: CandidateRanker (post-scoring ordering: ScoreDescendingRanker, WorkloadAscendingRanker, CostDescendingRanker, AuctionBidRanker).
  • engine/assignment/scoring_based.py::ScoringBasedAssignmentStrategy: composes (scorer, pool_filter, ranker). The five logical assignment strategies (role_based, load_balanced, cost_optimized, auction, hierarchical) are all ScoringBasedAssignmentStrategy instances with different filter/ranker pairs.
  • engine/assignment/registry.py::build_strategy_map(): the factory; preserves the public string discriminators.

Memory injection strategy

  • memory/injection.py: MemoryInjectionStrategy Protocol + InjectionStrategy discriminator (CONTEXT, TOOL_BASED, SELF_EDITING).
  • Concrete implementations: memory/context_injection.py::ContextInjectionStrategy, memory/tool_based.py::ToolBasedInjectionStrategy, memory/self_editing.py::SelfEditingMemoryStrategy.
  • memory/retrieval_config.py::MemoryRetrievalConfig.strategy: discriminator.
  • memory/injection_factory.py::build_memory_injection_strategy(): match-based dispatch with assert_never exhaustiveness.

Engine recovery strategy

  • engine/recovery.py: RecoveryStrategy Protocol + FailAndReassignStrategy.
  • engine/checkpoint/strategy.py::CheckpointRecoveryStrategy: resume-from-checkpoint sibling.
  • engine/recovery_config.py::EngineRecoveryConfig.strategy (RecoveryStrategyType): discriminator.
  • engine/recovery_factory.py::build_recovery_strategy(): match-based dispatch; RecoveryConfigError surfaces missing checkpoint_repo / checkpoint_config at boot rather than at recovery time.

Memory consolidation strategy (axis split, ADR-0005)

  • memory/consolidation/axis.py: EntrySelector + ConsolidationOp Protocols; SelectionGroup / ConsolidationContext / OpResult contracts.
  • Selector: memory/consolidation/selectors.py::HighestRelevanceSelector (shared by all three shipped strategies).
  • Ops: memory/consolidation/ops.py (ConcatenationOp, DensityRoutingOp, ExtractivePreservationOp, AbstractiveSummarizationOp) + memory/consolidation/llm_op.py::LLMSynthesisOp.
  • memory/consolidation/composite.py::CompositeConsolidationStrategy: selector + op aggregator (parallel=True for LLM cross-group TaskGroup fan-out).
  • memory/consolidation/config.py::ConsolidationStrategyType (SIMPLE / DUAL_MODE / LLM): discriminator.
  • memory/consolidation/factory.py::build_consolidation_strategy(): StrEnum-keyed StrategyRegistry dispatch; MemoryConfigError surfaces missing op-specific deps at construction.

Risk-tier classifier (timeout policy seam)

  • security/timeout/protocol.py::RiskTierClassifier Protocol (classify(action_type) -> ApprovalRiskLevel).
  • Impls: risk_tier_classifier.py::DefaultRiskTierClassifier (safe default), workload_adaptive.py::WorkloadAdaptiveRiskClassifier, operator_configurable.py::OperatorConfigurableRiskClassifier, time_based_elevation.py::TimeBasedRiskElevationClassifier.
  • risk_classifier_config.py::RiskClassifierType discriminator + frozen RiskClassifierConfig + RiskClassifierDeps (in-flight probe / Clock collaborators).
  • risk_classifier_factory.py::build_risk_tier_classifier(): StrEnum-keyed StrategyRegistry dispatch; RiskClassifierConfigError surfaces a missing required dep. Wired at timeout/factory.py::create_timeout_policy (tiered seam); SecOpsService + approval-tool consumers stay on the default pending a SecurityConfig.risk_classifier field.

Autonomy change strategy (promotion plugin)

  • security/autonomy/protocol.py::AutonomyChangeStrategy Protocol (request_promotion). Promotion is the only direction: an operator owns the autonomy level and nothing in the runtime lowers one.
  • Impls: change_strategy.py::HumanOnlyPromotionStrategy (safe default), budget_aware.py, escalation_chain.py (each holds the base via _base_delegate.py::BaseDelegatingStrategy).
  • Signal Protocols: signals.py::RiskBudgetSignalProvider (injected, never a concrete budget/ import), satisfied structurally by budget/risk_tracker.py::RiskTracker.headroom_fraction().
  • change_strategy_config.py::AutonomyStrategyType discriminator + frozen AutonomyStrategyConfig + AutonomyStrategyDeps.
  • change_strategy_factory.py::build_autonomy_change_strategy(): StrEnum-keyed StrategyRegistry dispatch; AutonomyStrategyConfigError surfaces a missing required signal provider. Wired at api/construction_phase.py, which builds the one RiskTracker both the strategy and the budget slice use, so every declared kind is selectable and satisfiable.

Ontology versioning (inverted backend dependency)

  • ontology/versioning.py: pure EntityDefinition snapshot deserializers; carries no backend imports.
  • persistence/sqlite/ontology_versioning.py::create_ontology_versioning(): SQLite-side factory.
  • persistence/postgres/ontology_versioning.py::create_postgres_ontology_versioning(): Postgres-side factory.
  • Each backend's lifecycle helper composes the matching factory at startup, so the dependency arrow points persistence -> ontology, never the reverse.

Backup handler registry (backend-pluggable)

  • backup/handlers/protocol.py: ComponentHandler Protocol.
  • backup/handlers/sqlite_persistence.py::SQLitePersistenceComponentHandler, backup/handlers/postgres_persistence.py::PostgresPersistenceComponentHandler, backup/handlers/memory.py::MemoryComponentHandler, backup/handlers/config_handler.py::ConfigComponentHandler.
  • backup/registry.py::PERSISTENCE_BACKUP_HANDLER_REGISTRY: StrategyRegistry keyed on config.persistence.backend ("sqlite" / "postgres").
  • backup/factory.py::build_backup_handlers(): dispatches per BackupComponent and uses the registry for the persistence handler.

Git backend storage strategy

  • engine/workspace/git_backend/protocol.py: GitBackend @runtime_checkable Protocol, with ProvisionResult/PushResult/FetchResult frozen result models.
  • engine/workspace/git_backend/config.py: GitBackendConfig (frozen) with kind: GitBackendType discriminator and GitBackendDeps (collaborators not safe in frozen config: workspace_base_root, connection_catalog, secret_backend, clock).
  • engine/workspace/git_backend/embedded.py::EmbeddedGitBackend (safe default: bare repo self-hosted on the persistent volume, no external dependency).
  • engine/workspace/git_backend/local_path.py::LocalPathGitBackend (bring-your-own on-disk git repository, push/fetch are no-ops because the on-disk repo is the durable store).
  • engine/workspace/git_backend/external_remote.py::ExternalRemoteGitBackend (GitHub / GitLab / Gitea / Forgejo resolved via the connection catalog; ships protocol + thin clone/push/fetch glue; deep OAuth hardening is a tracked follow-up).
  • engine/workspace/git_backend/factory.py::build_git_backend(): StrategyRegistry[GitBackend] keyed on GitBackendType. Missing required deps fail fast at construction with GitBackendConfigError. Wired at boot in api/app.py::_install_runtime_services under the has_persistence gate, alongside ProjectWorkspaceService.

Stakes assessment (model-routing input)

  • engine/stakes/protocol.py: StakesAssessor @runtime_checkable Protocol (assess_task(task) / assess_subtask(subtask) returning Stakes).
  • engine/stakes/heuristic.py::DefaultStakesAssessor (safe default: deterministic, combines complexity base mapping, high/critical keyword signals, and critical-priority elevation; unknown complexity fails safe upward to HIGH).
  • engine/stakes/config.py::StakesAssessmentConfig (frozen) with assessor: NotBlankStr discriminator, the complexity-to-stakes rules, and the keyword sets.
  • engine/stakes/factory.py::build_stakes_assessor(): StrategyRegistry[StakesAssessor] keyed on assessor ("heuristic" default). Consumed by DecompositionService (per-subtask) and the work pipeline's LEAF path (parent task).

Capability policy (deliberately NOT pluggable)

Listed here because it used to be a strategy seam and no longer is. Capability judgement is one object with one implementation, because a second answer to "may this agent take this work" is exactly the two-owner shape that let the coordination path route work dispatch then refused.

  • engine/routing_policy/capability_policy.py::CapabilityPolicy: the single owner of what rung the work demands (required_for, the operator's per-stakes floor raised one rung by substantial complexity), what rung an agent runs at (capability_of), whether it may take the work (judge(...).sanctioned), the per-stakes reasoning_effort, and the red-team threshold. One instance is built at boot and shared by selection and dispatch; set_config re-points the whole graph so a settings write is live without rewiring any consumer.
  • engine/routing_policy/capability_policy.py::AgentCapabilityReader: the one seam that remains, a @runtime_checkable Protocol reading an agent's rung. ResolvedAgentCapabilityReader reads the model catalogue (resolve_for_pair), falling back to the roster's ModelConfig.capability for a pair the catalogue does not grade.
  • core/capability_fit.py::partition_by_fit: the ladder both selection paths and gate-role staffing share, returning the first non-empty band in preference order (exact rung, else nearest higher, else nearest lower) and its label.
  • engine/routing_policy/config.py::CapabilityPolicyConfig (frozen): StakesCapabilityFloor (per-stakes required capability, validated non-decreasing), StakesReasoning (per-stakes effort, validated non-decreasing), red_team_min_stakes, and park_min_stakes (at or above it a weaker agent is refused rather than conceded). Every field has a live engine.* setting; CapabilityPolicySettingsSubscriber re-resolves and calls set_config.
  • Wired at boot by workers/_capability_policy_wiring.py::build_capability_policy, which returns None when no provider is configured (nothing grades a model, so there is no bar); the capability-dependent peer-review gate and the staffing sweep decline and say so rather than arming against nothing. The deterministic build/test gate needs no policy and stays attached (workers/_completion_oracle_runtime.py).

Per-project environment strategy

  • engine/workspace/environment/protocol.py: EnvironmentStrategy @runtime_checkable Protocol (kind / detect / scaffold / declaration_hash / managed_paths / runtime_env_vars / provision), with ProvisionedEnvironment / ScaffoldResult / CommandOutcome frozen result models and the EnvironmentCommandRunner seam (the resolved sandbox backend, adapted, so the subsystem never imports the tool layer).
  • engine/workspace/environment/config.py: EnvironmentConfig (frozen) with kind: EnvironmentType discriminator and EnvironmentDeps (collaborators not safe in frozen config: image_builder, clock).
  • engine/workspace/environment/manifest.py::ManifestEnvironmentStrategy (safe default: a committed synthorg.env.yaml of lockfiles + ordered setup commands; runs in both sandboxes and emits a stock bootstrap.sh so a fresh clone reproduces with no SynthOrg present).
  • engine/workspace/environment/devcontainer.py::DevcontainerEnvironmentStrategy (builds a sealed image from .devcontainer/devcontainer.json via image_builder; Docker backend only, raising EnvironmentBackendUnavailableError on a subprocess-backed project).
  • engine/workspace/environment/nix.py::NixEnvironmentStrategy (builds the declared flake.nix dev shell via nix develop; tool-wrapping of subsequent calls is a documented boundary).
  • engine/workspace/environment/factory.py::build_environment_strategy(): StrategyRegistry[EnvironmentStrategy] keyed on EnvironmentType; the devcontainer strategy falls back to the default AiodockerImageBuilder (the daemon over the mounted socket, streaming an in-process tar of the context) when no builder is injected.
  • engine/workspace/environment/service.py::EnvironmentService: provisions once per (project_id, declaration_hash) (persisted project_environments row is the durable cache), scaffolds + commits the declaration (GitWorkspaceCommitter), and is fail-loud. Wired at boot in api/app.py::_install_runtime_services alongside ProjectWorkspaceService. The result threads to the agent's sandbox via the ambient tools/sandbox/active_environment.py::ActiveSandboxEnvironment contextvar (image override + env additions), set per task in workers/execution_service.py.

Model-refresh strategy (cadence-mode seam)

  • providers/management/refresh_strategy.py: RefreshStrategy @runtime_checkable Protocol (reconcile(provider_name, provider) -> ProviderRefreshOutcome).
  • DetectOnlyStrategy (probe the live catalogue and flag removed models stale; never persists new models or recommends) and ReconcileRecommendStrategy (additionally persists newly-discovered models and produces in-family upgrade recommendations).
  • providers/management/refresh_strategy.py::build_refresh_strategy(): keyed on the RefreshMode discriminator; returns None for OFF / MANUAL_ONLY so the scheduler skips construction entirely (the off-by-cadence safe default).

Services are a distinct pattern (not pluggable subsystems)

A service wraps one or more repositories to keep controllers thin and centralise audit logging, and MAY orchestrate multiple repositories (e.g. WorkflowService spans workflow_definitions + workflow_versions; MemoryService spans fine-tune checkpoints + runs + settings).

The Protocol + Strategy + Factory + Config pattern applies only to genuinely cross-cutting subsystems that ship multiple interchangeable implementations selectable at runtime. Services do not need that machinery because there is exactly one service per domain.