Syntropy.TaskScheduler (syntropy v0.2.0)

Task orchestration boundary for the Syntropy runtime.

The GenServer owns run lifecycle state: the task counter, active task map, completed task history, and durable run envelopes. The surrounding pipeline lives in focused submodules — Selection (agent routing), Runner (run execution), SourceInquiry (evidence expansion), AgentDispatch (local/remote thinks), Synthesis (result merging), and RunEnvelope (run projections).

Summary

Types

active_task_export()

agent_attempt()

agent_ref()

citation_audit()

execution_quality()

provider_profile()

requested_strategy()

resolved_mode()

run_envelope()

run_error()

selection_trace_entry()

source_review()

task_result()

thought_result()

Functions

active_agent_ids()

active_tasks_snapshot()

approve_recommendation(recommendation_id)

cancel_run(run_id)

Best-effort cancellation of an in-flight run.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get(task_id)

get_run(run_id)

recent(limit \\ 100)

recent_runs(limit \\ 100)

reject_recommendation(recommendation_id)

reset()

start_link(opts \\ [])

start_run(prompt, opts \\ [])

submit(prompt, opts \\ [])

Types

active_task_export()

@type active_task_export() :: %{
  task_id: String.t(),
  coordinator_node_id: String.t(),
  selected_agents: [agent_ref()],
  started_at: DateTime.t()
}

agent_attempt()

@type agent_attempt() :: %{
  agent_id: String.t(),
  runtime_id: String.t(),
  node_id: String.t(),
  agent_name: String.t(),
  perspective: String.t(),
  status: String.t(),
  contributed: boolean(),
  started_at: DateTime.t() | nil,
  completed_at: DateTime.t() | nil,
  failed_at: DateTime.t() | nil,
  duration_ms: non_neg_integer() | nil,
  error: map() | nil
}

agent_ref()

@type agent_ref() :: %{id: String.t(), runtime_id: String.t(), node_id: String.t()}

citation_audit()

@type citation_audit() :: map()

execution_quality()

@type execution_quality() :: %{
  status: String.t(),
  selected_count: non_neg_integer(),
  contributing_count: non_neg_integer(),
  failed_count: non_neg_integer(),
  warnings: [String.t()]
}

provider_profile()

@type provider_profile() :: %{
  provider: String.t(),
  profile: String.t(),
  model: String.t() | nil,
  timeout_ms: pos_integer(),
  retry_count: non_neg_integer(),
  max_tokens: pos_integer() | nil,
  max_concurrency: pos_integer() | nil,
  temperature: float() | nil,
  top_p: float() | nil,
  repeat_penalty: float() | nil,
  disable_thinking: boolean(),
  assistant_prefill_configured: boolean()
}

requested_strategy()

@type requested_strategy() :: :single | :parallel | :all | :auto

resolved_mode()

@type resolved_mode() :: :single | :parallel | :all | :explicit

run_envelope()

@type run_envelope() :: %{
  id: String.t(),
  status: String.t(),
  kind: String.t(),
  node_id: String.t(),
  coordinator_node_id: String.t(),
  prompt: String.t(),
  strategy: String.t(),
  resolved_mode: String.t() | nil,
  selected_agent_ids: [String.t()],
  selected_agents: [agent_ref()],
  contributing_agent_ids: [String.t()],
  contributing_agents: [agent_ref()],
  thoughts: [thought_result()],
  thought_count: non_neg_integer(),
  agent_attempts: [agent_attempt()],
  execution_quality: execution_quality(),
  provider_profile: provider_profile() | nil,
  llm_usage: map() | nil,
  synthesis: String.t(),
  structured_answer: map() | nil,
  output_quality: map() | nil,
  selection_trace: [selection_trace_entry()],
  recommendation_ids: [String.t()],
  snapshot_ids: [String.t()],
  source_review: source_review() | nil,
  citation_audit: citation_audit() | nil,
  inquiry_rounds: [map()],
  tool_attempts: [map()],
  expanded_evidence: [map()],
  inquiry_quality: map() | nil,
  error: run_error() | nil,
  started_at: DateTime.t() | nil,
  updated_at: DateTime.t() | nil,
  completed_at: DateTime.t() | nil,
  failed_at: DateTime.t() | nil
}

run_error()

@type run_error() :: %{
  optional(:reason) => String.t(),
  code: String.t(),
  message: String.t()
}

selection_trace_entry()

@type selection_trace_entry() :: %{
  agent_id: String.t(),
  runtime_id: String.t(),
  node_id: String.t(),
  agent_name: String.t(),
  rank: pos_integer(),
  selected: boolean(),
  relevance: float(),
  position: float(),
  capacity: float(),
  composite: float()
}

source_review()

@type source_review() :: map()

task_result()

@type task_result() :: %{
  task_id: String.t(),
  node_id: String.t(),
  coordinator_node_id: String.t(),
  prompt: String.t(),
  strategy: String.t(),
  resolved_mode: String.t(),
  used_temporary_join: boolean(),
  selected_agent_ids: [String.t()],
  selected_agents: [agent_ref()],
  contributing_agent_ids: [String.t()],
  contributing_agents: [agent_ref()],
  thoughts: [thought_result()],
  thought_count: non_neg_integer(),
  agent_attempts: [agent_attempt()],
  execution_quality: execution_quality(),
  provider_profile: provider_profile(),
  synthesis: String.t(),
  structured_answer: map(),
  output_quality: map(),
  selection_trace: [selection_trace_entry()],
  recommendation_ids: [String.t()]
}

thought_result()

@type thought_result() :: %{
  agent_id: String.t(),
  runtime_id: String.t(),
  node_id: String.t(),
  agent: String.t(),
  perspective: String.t(),
  thought: String.t()
}

Functions

active_agent_ids()

@spec active_agent_ids() :: [String.t()]

active_tasks_snapshot()

@spec active_tasks_snapshot() :: [active_task_export()]

approve_recommendation(recommendation_id)

@spec approve_recommendation(String.t()) ::
  {:ok, Syntropy.RecommendationStore.recommendation()} | {:error, term()}

cancel_run(run_id)

@spec cancel_run(String.t()) ::
  {:ok, run_envelope()} | {:error, :not_found | :not_cancellable}

Best-effort cancellation of an in-flight run.

Kills the executing task tree, retains a cancelled run envelope, and replies {:error, :cancelled} to any blocked submitter. Runs that already reached a terminal state return {:error, :not_cancellable}.

When the run is unknown locally, cancellation falls back to :rpc.call against every connected node so the run's coordinator can cancel it — the cockpit and REST API can cancel any run in the cluster.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get(task_id)

@spec get(String.t()) :: task_result() | nil

get_run(run_id)

@spec get_run(String.t()) :: run_envelope() | nil

recent(limit \\ 100)

@spec recent(non_neg_integer()) :: [task_result()]

recent_runs(limit \\ 100)

@spec recent_runs(non_neg_integer()) :: [run_envelope()]

reject_recommendation(recommendation_id)

@spec reject_recommendation(String.t()) ::
  {:ok, Syntropy.RecommendationStore.recommendation()} | {:error, term()}

reset()

@spec reset() :: :ok

start_link(opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()

start_run(prompt, opts \\ [])

@spec start_run(
  String.t(),
  keyword()
) :: {:ok, run_envelope()} | {:error, term()}

submit(prompt, opts \\ [])

@spec submit(
  String.t(),
  keyword()
) :: {:ok, task_result()} | {:error, term()}