Version: v0.1.1 (2025-11-29)

Synapse is a headless, declarative multi‑agent orchestration framework. It provides a domain‑agnostic signal bus, workflow engine with Postgres persistence, and a configurable agent runtime for building domain-specific multi‑agent systems. A code review domain ships as a reference implementation; register it or build your own domains using the signal registry.

Highlights

• Domain‑agnostic signal registry with runtime topic registration
• Declarative orchestrator runtime (no GenServer boilerplate)
• Signal roles for orchestrator agents and configurable actions
• Workflow engine with persistence and audit trail (workflow_executions)
• Jido plan compiler/runner for turning plans into workflow specs
• LineageIR spans/artifacts plus RunIndex and Work job emissions
• LLM gateway powered by Altar.AI (optional) with ReqLLM fallback
• Telemetry throughout (router, workflows, LLM requests)

1. Install dependencies

   mix setup

2. Create and migrate the database

   By default, dev connects to postgres://postgres:postgres@localhost:5432/synapse_dev. Override via POSTGRES_* env vars (see config/dev.exs).

   mix ecto.create
   mix ecto.migrate

3. Run the Stage 2 demo (optional sanity check)

   mix run examples/stage2_demo.exs

   This boots the runtime, publishes a review request, and prints the resulting summary so you can see the declarative orchestrator in action.

4. Run the Plan compiler demo (optional)

   mix run examples/plan_compiler_demo.exs

   This compiles a Jido plan into a workflow spec and executes it through the engine.

5. Start the runtime for development

   iex -S mix

   This boots Synapse.Runtime, the signal router, the orchestrator runtime (reading priv/orchestrator_agents.exs), and the workflow engine with Postgres persistence. The application is OTP‑only — no Phoenix endpoint is required.

6. Run the Altar.AI integration example (optional)

   mix run examples/altar_ai_integration.exs

With Synapse.Domains.CodeReview registered, publish a :review_request signal from CI, a script, or an iex session:

{:ok, _signal} =
  Synapse.SignalRouter.publish(
    Synapse.SignalRouter,
    :review_request,
    %{
      review_id: "PR-12345",
      diff: git_diff,
      files_changed: 12,
      labels: ["security"],
      intent: "feature",
      metadata: %{repo: "org/app", author: "alice", files: touched_paths}
    },
    source: "/ci/github"
  )

The coordinator workflow classifies the request, spawns security/performance specialists as needed, and persists every step to workflow_executions.

Synapse is domain-agnostic. Define your own signals in config:

# config/config.exs
config :synapse, Synapse.Signal.Registry,
  topics: [
    ticket_created: [
      type: "support.ticket.created",
      schema: [
        ticket_id: [type: :string, required: true],
        customer_id: [type: :string, required: true],
        subject: [type: :string, required: true],
        priority: [type: {:in, [:low, :medium, :high]}, default: :medium]
      ]
    ]
  ]

Or register at runtime:

Synapse.Signal.register_topic(:my_event,
  type: "my.domain.event",
  schema: [id: [type: :string, required: true]]
)

See docs/guides/custom-domains.md for a full walkthrough.

To enable the built-in code review domain (security/performance specialists, review signals), register it via config:

# config/config.exs
config :synapse, :domains, [Synapse.Domains.CodeReview]

You can also call Synapse.Domains.CodeReview.register/0 during application startup if you prefer manual control.

Orchestrator agents can specify signal roles for custom domains:

%{
  id: :my_coordinator,
  type: :orchestrator,
  signals: %{
    subscribes: [:ticket_created, :ticket_analyzed],
    emits: [:ticket_resolved],
    roles: %{
      request: :ticket_created,
      result: :ticket_analyzed,
      summary: :ticket_resolved
    }
  },
  orchestration: %{
    classify_fn: &MyApp.classify/1,
    spawn_specialists: [:analyzer, :responder],
    aggregation_fn: &MyApp.aggregate/2
  }
}

Subscribe to summaries if you want push-style notifications:

{:ok, _sub_id} = Synapse.SignalRouter.subscribe(Synapse.SignalRouter, :review_summary)

receive do
  {:signal, %{type: "review.summary", data: summary}} ->
    IO.inspect(summary, label: "Review complete")
end

Or query Postgres for historical/auditable data:

Synapse.Workflow.Execution
|> where(review_id: "PR-12345")
|> Synapse.Repo.one!()

Each execution record includes the workflow name, step-by-step audit trail, accumulated results, and the final status, so you can drive dashboards or rerun failed work.

Synapse can delegate LLM operations to altar_ai via the compatibility layer:

alias Altar.AI.Integrations.Synapse, as: LLM

{:ok, response} = LLM.chat_completion(%{prompt: "Hello"}, profile: :openai)

Synapse.ReqLLM remains available but is deprecated and will be removed in a future release.

Specialists and the coordinator are declared in priv/orchestrator_agents.exs and are reconciled and run by Synapse.Orchestrator.Runtime. Update this file to add or tune agents — no GenServer code needed.

Example snippet:

%{
  id: :coordinator,
  type: :orchestrator,
  actions: [Synapse.Domains.CodeReview.Actions.ClassifyChange],
  orchestration: %{
    classify_fn: &MyStrategies.classify/1,
    spawn_specialists: [:security_specialist, :performance_specialist],
    aggregation_fn: &MyStrategies.aggregate/2,
    negotiate_fn: &MyStrategies.resolve_conflicts/2
  },
  signals: %{
    subscribes: [:review_request, :review_result],
    emits: [:review_summary],
    roles: %{
      request: :review_request,
      result: :review_result,
      summary: :review_summary
    }
  },
  state_schema: [review_count: [type: :non_neg_integer, default: 0]]
}

On boot, the orchestrator runtime validates configs, spawns any missing agents, and monitors process health. Update the file and trigger a reload to reconcile changes.

The workflow engine executes declarative specs (steps + dependencies) with retries, compensation, and telemetry. It powers specialist runs and the coordinator paths while giving you a uniform audit trail and optional Postgres persistence.

Key pieces

• Spec: Synapse.Workflow.Spec with Step and Output helpers
• Engine: Synapse.Workflow.Engine.execute/2 runs the spec and emits telemetry
• Persistence (optional): snapshots to workflow_executions via Synapse.Workflow.Persistence

Minimal example

alias Synapse.Workflow.{Spec, Engine}
alias Synapse.Workflow.Spec.Step

spec =
  Spec.new(
    name: :example_workflow,
    description: "Analyze and generate critique with retries",
    metadata: %{version: 1},
    steps: [
      Step.new(
        id: :fetch_context,
        action: MyApp.Actions.FetchContext,
        # env = %{input, results, context, step, workflow}
        params: fn env -> %{id: env.input.review_id} end
      ),
      Step.new(
        id: :analyze,
        action: Synapse.Actions.CriticReview,
        requires: [:fetch_context],
        retry: [max_attempts: 3, backoff: 200],
        on_error: :continue,
        params: fn env -> %{diff: env.input.diff, metadata: env.results.fetch_context} end
      ),
      Step.new(
        id: :generate_critique,
        action: Synapse.Actions.GenerateCritique,
        requires: [:analyze],
        params: fn env ->
          review = env.results.analyze
          %{
            prompt: "Summarize issues",
            messages: [%{role: "user", content: Enum.join(review.issues || [], ", ")}],
            profile: :openai
          }
        end
      )
    ],
    outputs: [
      Spec.output(:review, from: :analyze),
      Spec.output(:critique, from: :generate_critique, path: [:content])
    ]
  )

input = %{review_id: "PR-42", diff: "..."}
ctx = %{request_id: "req_abc123"} # required when persistence is enabled

case Engine.execute(spec, input: input, context: ctx) do
  {:ok, %{results: _results, outputs: outputs, audit_trail: audit}} ->
    IO.inspect(outputs.critique, label: "Critique")
    IO.inspect(audit.steps, label: "Audit trail")

  {:error, %{failed_step: step, error: error, audit_trail: audit}} ->
    IO.inspect({step, error}, label: "Workflow failed")
    IO.inspect(audit.steps, label: "Audit trail")
end

Parameters and env

• Step params may be a map/keyword, or a function fn env -> ... end where env includes :input, :results, :context, :step, and :workflow.
• Steps support requires: [:other_step], retry: [max_attempts:, backoff:], and on_error: :halt | :continue.
• Outputs map step results to the final payload; path lets you pick nested fields; transform is available for custom shaping.

Persistence

• Dev config enables persistence by default: see config/config.exs for config :synapse, Synapse.Workflow.Engine, persistence: {Synapse.Workflow.Persistence.Postgres, []}.
• When persistence is enabled, you must provide a :request_id in context; the engine will snapshot before/after steps to workflow_executions.
• You can override per call: Engine.execute(spec, input: ..., context: ..., persistence: nil).

Telemetry

• Emits [:synapse, :workflow, :step, :start|:stop|:exception] with metadata like :workflow, :workflow_step, :workflow_attempt.
• Example handler:

:telemetry.attach(
  "wf-logger",
  [[:synapse, :workflow, :step, :stop]],
  fn _evt, m, meta, _ ->
    require Logger
    Logger.info("step done",
      workflow: meta.workflow,
      step: meta.workflow_step,
      attempt: meta.workflow_attempt,
      duration_us: m.duration_us
    )
  end,
  nil
)

Error and result shapes

• Success: {:ok, %{results: map(), outputs: map(), audit_trail: map()}}
• Failure: {:error, %{failed_step: atom(), error: term(), attempts: pos_integer(), results: map(), audit_trail: map()}}

Further reading

• Cookbook: docs_new/workflows/engine.md
• ADR: docs_new/adr/0004-declarative-workflow-engine.md

Synapse can compile Jido.Plan DAGs into workflow specs. Use Synapse.PlanRunner to compile and execute in one step, or Synapse.PlanCompiler if you need to inspect/modify the spec before running.

alias Jido.Plan
alias Synapse.PlanRunner
alias Synapse.Workflow.Spec

plan =
  Plan.new(context: %{tenant_id: "acme"})
  |> Plan.add(:fetch, {MyApp.Actions.Fetch, %{value: 2}})
  |> Plan.add(:double, {MyApp.Actions.Double, %{value: 4}}, depends_on: :fetch)

{:ok, exec} =
  PlanRunner.run(plan,
    name: :demo_plan,
    outputs: [Spec.output(:result, from: :double, path: [:value])],
    input: %{},
    context: %{request_id: "req_plan_demo"}
  )

Plan compilation maps depends_on -> requires, applies instruction options (max_retries, backoff, timeout), and merges plan/instruction context into each workflow step.

Synapse uses Req for HTTP and provides a multi‑provider LLM gateway. Configure at runtime via environment:

export OPENAI_API_KEY=sk-...
# or
export GEMINI_API_KEY=ya29....

Configuration is assembled in config/runtime.exs. Profiles support timeouts, retries, and provider‑specific options.

Example usage:

{:ok, resp} =
  Synapse.ReqLLM.chat_completion(%{
    messages: [%{role: "user", content: "Summarize the following diff..."}],
    temperature: 0.2
  }, profile: :openai)

resp.content     # normalized string content
resp.metadata    # token usage, finish_reason, provider-specific details

See: lib/synapse/req_llm.ex, lib/synapse/providers/*, and docs_new/20251029/implementation/LLM_INTEGRATION.md.

• Adapter: Synapse.Workflow.Persistence.Postgres
• Schema: workflow_executions (created by migration at priv/repo/migrations/*_create_workflow_executions.exs)
• Test env disables persistence by default (config/test.exs)

Common tasks:

mix ecto.create
mix ecto.migrate
mix ecto.rollback

Database configuration for dev can be overridden with POSTGRES_HOST, POSTGRES_DB, POSTGRES_USER, POSTGRES_PASSWORD, POSTGRES_PORT, POSTGRES_POOL_SIZE.

• Signals: [:synapse, :signal_router, :publish|:deliver]
• LLM: [:synapse, :llm, :request, :start|:stop|:exception]
• Workflows/Orchestrator: see docs_new/20251028/remediation/telemetry_documentation.md

Attach your own handlers using :telemetry.attach/4.

Workflow execution emits LineageIR trace/span/artifact events, RunIndex run/step writes, and NSAI Work job lifecycle events. Configure adapters globally:

config :synapse,
  lineage_ir: true,
  run_index_adapter: Synapse.RunIndex.Adapters.Ecto,
  run_index_repo: Synapse.Repo,
  work_adapter: Synapse.WorkEmitter.Adapters.Telemetry

config :lineage_ir,
  sink_adapter: LineageIR.Sink.Adapters.Ecto,
  ecto_repo: Synapse.Repo

You can override per call via Engine.execute/2 options: lineage_ir, lineage_opts, run_index_adapter, run_index_opts, work_adapter, and work_opts.

Run the full suite with:

mix test

Dialyzer and other pre-commit checks are available via mix precommit.

• Roadmap: ROADMAP.md
• Orchestrator design and reference: docs_new/20251028/synapse_orchestrator/README.md
• Multi‑agent framework docs: docs_new/20251028/multi_agent_framework/README.md
• Workflow engine and persistence: docs_new/workflows/engine.md and ADRs in docs_new/adr/
• Post‑Phoenix direction: docs_new/20251109/README.md
• Custom domains guide: docs/guides/custom-domains.md
• Plan compiler guide: docs/guides/plan-compiler.md
• Migration guide: docs/guides/migration-0.1.1.md

See CHANGELOG.md.

elixir • otp • beam • erlang-vm • concurrency • fault-tolerance • supervision • jido • req • multi-agent • agent-orchestration • multi-agent-systems • orchestrator • workflows • llm • openai • gemini • postgres • telemetry • ai

Licensed under MIT. See LICENSE.