symfonic.infra

infra

symfonic.infra -- infrastructure abstractions for external services.

Provides protocol interfaces that the library depends on without importing concrete implementations (Celery, APScheduler, etc.).

ExecutionTrace dataclass

ExecutionTrace(
    run_id: str,
    source: str = "chat",
    start_time: float = time.monotonic(),
    events: list[TraceEvent] = list(),
)

Complete trace for a single request or background task.

add_event

add_event(
    event_type: str, label: str, **details: Any
) -> None

Append a timestamped event to this trace.

Source code in src/symfonic/infra/trace.py

def add_event(self, event_type: str, label: str, **details: Any) -> None:
    """Append a timestamped event to this trace."""
    elapsed = (time.monotonic() - self.start_time) * 1000
    self.events.append(
        TraceEvent(
            elapsed_ms=round(elapsed, 1),
            event_type=event_type,
            label=label,
            details=details,
        )
    )

to_dict

to_dict() -> dict[str, Any]

Serialise to a plain dict suitable for JSON responses.

Source code in src/symfonic/infra/trace.py

def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain dict suitable for JSON responses."""
    return {
        "run_id": self.run_id,
        "source": self.source,
        "events": [
            {
                "elapsed_ms": e.elapsed_ms,
                "event_type": e.event_type,
                "label": e.label,
                "details": e.details,
            }
            for e in self.events
        ],
    }

InMemoryTraceStore

InMemoryTraceStore(max_traces: int = MAX_TRACES)

Default trace store using an in-memory bounded deque.

Suitable for development and single-process deployments. Data is lost on restart. For production, inject a PostgresTraceStore or similar adapter that satisfies TraceStoreProtocol.

Source code in src/symfonic/infra/trace.py

def __init__(self, max_traces: int = MAX_TRACES) -> None:
    self._traces: deque[ExecutionTrace] = deque(maxlen=max_traces)
    self._current: dict[str, ExecutionTrace] = {}

finish_trace

finish_trace(run_id: str) -> None

Append a done event and move the trace to completed storage.

Source code in src/symfonic/infra/trace.py

def finish_trace(self, run_id: str) -> None:
    """Append a done event and move the trace to completed storage."""
    trace = self._current.pop(run_id, None)
    if trace:
        trace.add_event("done", "Complete")
        self._traces.append(trace)
        logger.debug("trace finished: run_id=%s", run_id)
    else:
        logger.debug("finish_trace called for unknown run_id=%s (no-op)", run_id)

get_trace

get_trace(run_id: str) -> ExecutionTrace | None

Return the active trace for run_id, or None.

Source code in src/symfonic/infra/trace.py

def get_trace(self, run_id: str) -> ExecutionTrace | None:
    """Return the active trace for run_id, or None."""
    return self._current.get(run_id)

latest

latest(limit: int = 10) -> list[dict[str, Any]]

Return up to limit most-recently completed traces.

Source code in src/symfonic/infra/trace.py

def latest(self, limit: int = 10) -> list[dict[str, Any]]:
    """Return up to `limit` most-recently completed traces."""
    return [t.to_dict() for t in list(self._traces)[-limit:]]

start_trace

start_trace(
    run_id: str, source: str = "chat"
) -> ExecutionTrace

Create a new active trace and return it.

Source code in src/symfonic/infra/trace.py

def start_trace(self, run_id: str, source: str = "chat") -> ExecutionTrace:
    """Create a new active trace and return it."""
    trace = ExecutionTrace(run_id=run_id, source=source)
    trace.add_event("start", f"Start {source}")
    self._current[run_id] = trace
    logger.debug("trace started: run_id=%s source=%s", run_id, source)
    return trace

NullScheduler

No-op scheduler that silently discards all task requests.

Used as the default when no external scheduler is configured, ensuring the library functions standalone without Celery or any other task infrastructure.

cancel_task async

cancel_task(task_id: str) -> bool

Always return False (nothing to cancel).

Source code in src/symfonic/infra/protocols.py

async def cancel_task(self, task_id: str) -> bool:
    """Always return False (nothing to cancel)."""
    logger.debug("NullScheduler: discarding cancel_task(%s)", task_id)
    return False

enqueue_task async

enqueue_task(
    task_name: str, params: dict[str, Any] | None = None
) -> str

Accept and discard the task request.

Source code in src/symfonic/infra/protocols.py

async def enqueue_task(
    self,
    task_name: str,
    params: dict[str, Any] | None = None,
) -> str:
    """Accept and discard the task request."""
    self._counter += 1
    task_id = f"null-{self._counter}"
    logger.debug(
        "NullScheduler: discarding enqueue_task(%s) -> %s",
        task_name,
        task_id,
    )
    return task_id

schedule_recurring async

schedule_recurring(
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str

Accept and discard the recurring schedule request.

Source code in src/symfonic/infra/protocols.py

async def schedule_recurring(
    self,
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str:
    """Accept and discard the recurring schedule request."""
    self._counter += 1
    task_id = f"null-recurring-{self._counter}"
    logger.debug(
        "NullScheduler: discarding schedule_recurring(%s, %ss) -> %s",
        task_name,
        interval_seconds,
        task_id,
    )
    return task_id

TaskSchedulerProtocol

Bases: Protocol

Abstract interface for background task scheduling.

All methods accept a task_name string that the adapter maps to a concrete task implementation (e.g. a Celery task, an APScheduler job, etc.).

cancel_task async

cancel_task(task_id: str) -> bool

Cancel a previously enqueued or recurring task.

Parameters:

Name	Type	Description	Default
task_id	str	The identifier returned by enqueue_task or schedule_recurring.	required

Returns:

Type	Description
bool	True if the task was found and cancelled, False otherwise.

Source code in src/symfonic/infra/protocols.py

async def cancel_task(self, task_id: str) -> bool:
    """Cancel a previously enqueued or recurring task.

    Args:
        task_id: The identifier returned by ``enqueue_task``
            or ``schedule_recurring``.

    Returns:
        True if the task was found and cancelled, False otherwise.
    """
    ...

enqueue_task async

enqueue_task(
    task_name: str, params: dict[str, Any] | None = None
) -> str

Enqueue a task for immediate background execution.

Parameters:

Name	Type	Description	Default
task_name	str	Identifier for the task to execute.	required
params	dict[str, Any] | None	Optional keyword arguments for the task.	None

Returns:

Type	Description
str	A task ID string that can be used with cancel_task.

Source code in src/symfonic/infra/protocols.py

async def enqueue_task(
    self,
    task_name: str,
    params: dict[str, Any] | None = None,
) -> str:
    """Enqueue a task for immediate background execution.

    Args:
        task_name: Identifier for the task to execute.
        params: Optional keyword arguments for the task.

    Returns:
        A task ID string that can be used with ``cancel_task``.
    """
    ...

schedule_recurring async

schedule_recurring(
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str

Schedule a task to run repeatedly at a fixed interval.

Parameters:

Name	Type	Description	Default
task_name	str	Identifier for the task.	required
interval_seconds	float	Seconds between successive executions.	required
params	dict[str, Any] | None	Optional keyword arguments for each invocation.	None

Returns:

Type	Description
str	A task ID string for cancellation.

Source code in src/symfonic/infra/protocols.py

async def schedule_recurring(
    self,
    task_name: str,
    interval_seconds: float,
    params: dict[str, Any] | None = None,
) -> str:
    """Schedule a task to run repeatedly at a fixed interval.

    Args:
        task_name: Identifier for the task.
        interval_seconds: Seconds between successive executions.
        params: Optional keyword arguments for each invocation.

    Returns:
        A task ID string for cancellation.
    """
    ...

TraceEvent dataclass

TraceEvent(
    elapsed_ms: float,
    event_type: str,
    label: str,
    details: dict[str, Any] = dict(),
)

Single event in an execution trace.

TraceStoreProtocol

Bases: Protocol

Abstract interface for execution trace storage.

All concrete adapters (in-memory, PostgreSQL, Redis, …) must satisfy this structural protocol. The library never imports a concrete class.

finish_trace

finish_trace(run_id: str) -> None

Mark a trace as complete and move it from active to storage.

A "done" event is appended before archiving. Calling with an unknown run_id is a no-op.

Parameters:

Name	Type	Description	Default
run_id	str	Identifier of the trace to finish.	required

Source code in src/symfonic/infra/trace.py

def finish_trace(self, run_id: str) -> None:
    """Mark a trace as complete and move it from active to storage.

    A "done" event is appended before archiving.  Calling with an
    unknown run_id is a no-op.

    Args:
        run_id: Identifier of the trace to finish.
    """
    ...

get_trace

get_trace(run_id: str) -> ExecutionTrace | None

Return an active (in-progress) trace, or None if not found.

Parameters:

Name	Type	Description	Default
run_id	str	Identifier to look up.	required

Source code in src/symfonic/infra/trace.py

def get_trace(self, run_id: str) -> ExecutionTrace | None:
    """Return an active (in-progress) trace, or None if not found.

    Args:
        run_id: Identifier to look up.
    """
    ...

latest

latest(limit: int = 10) -> list[dict[str, Any]]

Return the most recent completed traces as plain dicts.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of traces to return.	10

Source code in src/symfonic/infra/trace.py

def latest(self, limit: int = 10) -> list[dict[str, Any]]:
    """Return the most recent completed traces as plain dicts.

    Args:
        limit: Maximum number of traces to return.
    """
    ...

start_trace

start_trace(
    run_id: str, source: str = "chat"
) -> ExecutionTrace

Begin tracking a new execution trace.

Parameters:

Name	Type	Description	Default
run_id	str	Unique identifier for this execution run.	required
source	str	Label for the caller type (e.g. "chat", "stream", "worker").	'chat'

Returns:

Type	Description
ExecutionTrace	The newly created ExecutionTrace with an initial "start" event.

Source code in src/symfonic/infra/trace.py

def start_trace(self, run_id: str, source: str = "chat") -> ExecutionTrace:
    """Begin tracking a new execution trace.

    Args:
        run_id: Unique identifier for this execution run.
        source: Label for the caller type (e.g. "chat", "stream", "worker").

    Returns:
        The newly created ExecutionTrace with an initial "start" event.
    """
    ...