Continuum.Observer (continuum v0.7.1)

Data and action helpers for the optional Continuum Observer.

The Observer is mounted from a host Phoenix router with Continuum.Observer.Router.continuum_observer/2. Continuum does not start an Observer supervisor and does not provide authentication; mount it only inside an authenticated admin scope.

Query helpers in this module are Phoenix-independent and operate on the configured Continuum instance repo. Event payloads are decoded with :erlang.binary_to_term/1 because Continuum stores its own trusted journal data as bytea; the Observer is not a boundary for untrusted database writes.

Summary

Types

run_state()

Functions

cancel_run(run_id, opts \\ [])

Cancels a run through the public Continuum API using the Observer instance.

decode_signal_payload(json)

Decodes a JSON payload from the Observer signal form.

get_run(run_id, opts \\ [])

Loads one run for the Observer detail view.

health(opts \\ [])

Builds the operational health report shown by the Observer health panel.

list_activity_tasks(run_id, opts \\ [])

Lists bounded activity task state and last-heartbeat progress for a run.

list_events(run_id, opts \\ [])

Lists a bounded keyset page of decoded journal events ordered by sequence.

list_runs(opts \\ [])

Lists runs for the Observer index.

pretty(term, opts \\ [])

Pretty prints an event payload for display with a hard byte cap.

repair_health(action, subject_id, opts \\ [])

Plans or executes a fenced operational repair from the Observer.

run_topic(run_id)

Returns the per-run topic used by run detail pages.

runs_topic()

Returns the low-fidelity per-instance runs topic used by the Observer index.

send_signal(run_id, name, payload, opts \\ [])

Sends a signal through the public Continuum API using the Observer instance.

signal_contracts(run_id, opts \\ [])

Returns the signal contracts declared by the run's workflow version.

subscribe_run(run_id, opts \\ [])

Subscribes the caller to full-fidelity updates for a single run.

subscribe_runs(opts \\ [])

Subscribes the caller to coarse run-index updates for an instance.

successor_run_id(run_id, opts \\ [])

Returns the run id that this run continued into via continue_as_new, or nil.

Types

run_state()

@type run_state() :: :running | :suspended | :completed | :failed | :cancelled

Functions

cancel_run(run_id, opts \\ [])

@spec cancel_run(
  binary(),
  keyword()
) :: :ok | {:error, term()}

Cancels a run through the public Continuum API using the Observer instance.

decode_signal_payload(json)

@spec decode_signal_payload(binary()) :: {:ok, term()} | {:error, term()}

Decodes a JSON payload from the Observer signal form.

get_run(run_id, opts \\ [])

@spec get_run(
  binary(),
  keyword()
) :: {:ok, Continuum.Run.t()} | {:error, :not_found | term()}

Loads one run for the Observer detail view.

health(opts \\ [])

@spec health(keyword()) :: {:ok, map()} | {:error, term()}

Builds the operational health report shown by the Observer health panel.

list_activity_tasks(run_id, opts \\ [])

@spec list_activity_tasks(
  binary(),
  keyword()
) :: {:ok, [map()]} | {:error, term()}

Lists bounded activity task state and last-heartbeat progress for a run.

Heartbeat details pass through the same configurable Observer redactor as event payloads.

list_events(run_id, opts \\ [])

@spec list_events(
  binary(),
  keyword()
) :: {:ok, Continuum.Page.t(map())} | {:error, term()}

Lists a bounded keyset page of decoded journal events ordered by sequence.

Pass :after_seq to continue from a previous page, :limit (capped at 100), :max_payload_bytes to reject oversized encoded payloads before decoding, and :redactor as a unary function or module that exports redact/1. The configured :observer_redactor application setting is used when :redactor is omitted.

list_runs(opts \\ [])

@spec list_runs(keyword()) ::
  {:ok, Continuum.Page.t(Continuum.Run.t())} | {:error, term()}

Lists runs for the Observer index.

Options:

:instance - Continuum instance name or struct. Defaults to Continuum.
:state - atom/string run state filter.
:workflow - workflow module substring filter.
:search - run id or workflow substring filter.
:page - 1-based page number.
:per_page - page size, capped at 100.

pretty(term, opts \\ [])

@spec pretty(
  term(),
  keyword()
) :: binary()

Pretty prints an event payload for display with a hard byte cap.

repair_health(action, subject_id, opts \\ [])

@spec repair_health(atom() | binary(), binary(), keyword()) ::
  {:ok, map()} | {:error, term()}

Plans or executes a fenced operational repair from the Observer.

Repairs remain dry-run by default; pass execute: true after presenting an explicit confirmation to the operator.

run_topic(run_id)

@spec run_topic(binary()) :: binary()

Returns the per-run topic used by run detail pages.

runs_topic()

@spec runs_topic() :: binary()

Returns the low-fidelity per-instance runs topic used by the Observer index.

send_signal(run_id, name, payload, opts \\ [])

@spec send_signal(binary(), atom() | binary(), term(), keyword()) ::
  :ok | {:error, term()}

Sends a signal through the public Continuum API using the Observer instance.

signal_contracts(run_id, opts \\ [])

@spec signal_contracts(
  binary(),
  keyword()
) :: {:ok, Continuum.SignalContract.contracts()} | {:error, term()}

Returns the signal contracts declared by the run's workflow version.

subscribe_run(run_id, opts \\ [])

@spec subscribe_run(
  binary(),
  keyword()
) :: :ok | {:error, term()}

Subscribes the caller to full-fidelity updates for a single run.

subscribe_runs(opts \\ [])

@spec subscribe_runs(keyword()) :: :ok | {:error, term()}

Subscribes the caller to coarse run-index updates for an instance.

successor_run_id(run_id, opts \\ [])

@spec successor_run_id(
  binary(),
  keyword()
) :: binary() | nil

Returns the run id that this run continued into via continue_as_new, or nil.