symfonic.capabilities.memory.background_cycles

background_cycles

Owning a cycle as background work, apart from running one.

Split from :mod:.napping at the 300-line budget, along a seam the coordinator already had: :meth:ConsolidationCoordinator.run executes a cycle, and everything here decides what happens to it as a task -- who holds it, what a failure becomes, and what shutdown waits for.

A mixin rather than a helper module of functions, because every method here is about the coordinator's own cadence cursor and task registry, and passing both to a free function would be a wrapper around self with extra steps.

BackgroundCycles

How a coordinator starts a cycle, reports it, and waits for it.

pending property

pending: int

How many cycles this coordinator still has in flight.

aclose async

aclose() -> None

Wait for every cycle still running. Safe to call more than once.

Source code in src/symfonic/capabilities/memory/background_cycles.py

async def aclose(self) -> None:
    """Wait for every cycle still running. Safe to call more than once."""
    await self._tasks.aclose()

consolidate async

consolidate(scope: MemoryScope, cycle: ConsolidationCycle, *, run_id: str = '', root_run_id: str = '') -> ConsolidationState | None

Run a cycle and report it, converting a failure into a log line.

This is what background work runs. A cycle that raised must not become an unhandled task exception -- the turn already answered, and the adopter's console is not where a consolidation failure belongs. A cancellation is re-raised: the run is unwinding, and swallowing it here would make teardown report a task that finished when it was cut.

Source code in src/symfonic/capabilities/memory/background_cycles.py

async def consolidate(
    self,
    scope: MemoryScope,
    cycle: ConsolidationCycle,
    *,
    run_id: str = "",
    root_run_id: str = "",
) -> ConsolidationState | None:
    """Run a cycle and report it, converting a failure into a log line.

    This is what background work runs. A cycle that raised must not become
    an unhandled task exception -- the turn already answered, and the
    adopter's console is not where a consolidation failure belongs. A
    cancellation is re-raised: the run is unwinding, and swallowing it here
    would make teardown report a task that finished when it was cut.
    """
    try:
        state = await self.run(scope, cycle, run_id=run_id, root_run_id=root_run_id)
    except asyncio.CancelledError:
        # Not completed, so the cursor stays where it is and the next turn
        # over this scope tries again.
        self._uncomplete(scope)
        raise
    except Exception:
        await self._publish_terminal(
            terminal_record(
                cycle=cycle, run_id=run_id, root_run_id=root_run_id, failed=True
            )
        )
        logger.warning(
            "consolidation cycle=%s scope=%s failed before it could report",
            cycle.value,
            scope.path,
            exc_info=True,
        )
        return None
    await self._publish_terminal(
        terminal_record(
            cycle=cycle, run_id=run_id, root_run_id=root_run_id, state=state
        )
    )
    logger.info("consolidation %s", state.telemetry())
    return state

schedule_cycle

schedule_cycle(scope: MemoryScope, cycle: ConsolidationCycle, *, run_id: str = '', root_run_id: str = '') -> asyncio.Task[ConsolidationState | None]

Start a cycle this coordinator owns, and hand back the task.

Source code in src/symfonic/capabilities/memory/background_cycles.py

def schedule_cycle(
    self,
    scope: MemoryScope,
    cycle: ConsolidationCycle,
    *,
    run_id: str = "",
    root_run_id: str = "",
) -> asyncio.Task[ConsolidationState | None]:
    """Start a cycle this coordinator owns, and hand back the task."""
    return self._tasks.start(
        self.consolidate(scope, cycle, run_id=run_id, root_run_id=root_run_id),
        name=f"consolidation:{cycle.value}:{scope.path}",
    )