controllers: ControllerRunner, plus native timestamps and severity on attributes

[coretl]

Closes #395

Three related pieces from ADR 0016, all of which the embedded ophyd-async connector (#399) needs and none of which are embedding-specific.

ControllerRunner

The controller lifecycle was inlined in FastCS.serve, so there was no way to run controllers without also pulling in transport-serving and the interactive shell. It moves to fastcs.controllers.ControllerRunner, and FastCS.serve becomes a caller of it.

runner = ControllerRunner(controller)
apis = await runner.setup()   # initialise, and build the ControllerAPIs
await runner.start()          # connect, run initial tasks, start scanning
await runner.stop()

Starting is in two halves because anything serving the controllers has to register its callbacks before the first values are read — the PVA transport already carries a comment explaining exactly this — so setup() builds the APIs and start() does the rest. start() runs setup() first if you have not, so an embedder that does not need the APIs in between just calls start()/stop(), which is the shape the ADR asks for. Idempotency is the caller's responsibility, as agreed.

The runner also owns reconnect, which closes a live gap. A scan task whose callback raises sets _connected = False and pauses rather than dying — but nothing in FastCS ever called Controller.reconnect(), so unless a driver wired its own recovery the controller stayed paused for good. The runner now watches for it and reconnects, so every controller recovers the same way. Controller.connected exposes the state that was previously only readable through the private _connected.

Native timestamps and severity

A value entering an attribute may now say when it was obtained and how wrong it is, through the Update that getters and setters could already return:

async def get_temperature() -> Update[float]:
    value, device_time = await protocol.read_with_timestamp()
    return Update(readback=value, timestamp=device_time, severity=Severity.NO_ALARM)

AttrR.timestamp and AttrR.severity read them back. A bare value is stamped with the time it arrived and reported as Severity.NO_ALARM, so nothing changes for a driver that does not care. Severity is a FastCS enum that happens to use the same strings as EPICS alarm severities; the value/timestamp/severity trio follows the shape of bluesky's Reading and shares no code with it, per the ADR.

Setpoint cache

Already delivered by #412, as the .setpoint property — cached by set() before the setter runs and regardless of whether it succeeds, which is exactly what ADR 0016's question 1 settled on. Nothing to do here beyond documenting it as part of the stable surface.

Stable interface

New docs/explanations/stable-interface.md writes down the narrow surface an embedder is restricted to (decision 13): the runner, ControllerAPI, and the attribute/command runtime methods — and that nothing should reach into BaseController.

Instructions to reviewer on how to test:

1. uv run pytest tests/test_controller_runner.py tests/test_attributes.py -v
2. Run the demo (python -m fastcs.demo run src/fastcs/demo/fastcs.yaml) against the sim, kill the sim so the scan tasks fail, restart it, and confirm the controller reconnects on its own rather than staying paused.

Checks for reviewer

• Would the PR title make sense to a user on a set of release notes
• Controller.connected is a new read-only property, so a controller that assigned self.connected = ... for its own bookkeeping now gets an AttributeError. Two test controllers in this repo did exactly that and are renamed; downstream drivers may too. The failure is loud rather than silent, and pre-1.0 is the window for it, but say if you would rather it were is_connected.
• Reconnect is now automatic, on a 1 s poll (RECONNECT_PERIOD). A driver that previously relied on a controller staying paused after a failure will now see it retried. This is what the ADR asks for — "the runner owns the whole lifecycle including reconnect" — but it is a behaviour change, not just a move.
• Severity defaults to NO_ALARM, not "unset". The ADR's consequences say "severity unset" for a bare value; modelling that as a fourth state (None) would push an Optional into every transport that reads it, and EPICS' own zero value already means the same thing. Say if you want the tri-state.

Notes

• Transports are untouched: whether EPICS/Tango/REST/GraphQL surface the new timestamp and severity is called out in the ADR as transport-specific follow-up, not part of this issue. The PVA transport still derives its alarm severity from the numeric limits.
• Readback callbacks keep their (value) signature; a transport that wants the timestamp or severity reads them off the attribute. Widening the callback would have touched every transport for no caller that needs it yet.
• The timestamp and severity are applied only after the value validates, so a rejected update leaves the cached value and the time it was obtained agreeing with each other.
• FastCS._scan_tasks/_initial_coros are gone — they live on the runner now. tests/test_control_system.py reached into them and is updated.
• The task-cancelling guards from FastCS._stop_scan_tasks are not carried across: Task.cancel returns whether the task was cancellable rather than raising, so the except (CancelledError, RuntimeError) and except Exception -> raise RuntimeError arms were unreachable.
• Overlaps attributes: replace the DataType family with python types and *Meta typed dicts #418 (Remove the DataType family — python types + *Meta typed dicts #413) in src/fastcs/attributes/attr_r.py: that PR changes how update() validates, this one adds the stamping either side of it. Both are independent branches off refactor, so whichever merges second needs a small conflict resolution in AttrR.update.
• Verified locally with uv run --locked tox -e pre-commit,type-checking, both green in full. For the tests env, this sandbox can't run docs (needs outbound network) or the PVA/p4p-backed tests (RuntimeError: Address family not supported by protocol), the same known limitation noted on demo: use ControllerVector for temperature ramp sub-controllers #409/demo: cut-down Eiger REST sim + introspectable controller example #410/demo: convert temperature controller to getter/setter style #411/attributes: getter/setter IO rework, remove AttributeIORef/AttributeIO #412. Excluding those, pytest src tests --ignore=tests/benchmarking passes 349/359, with only the same 10 pre-existing p4p/socket-family failures, which I confirmed are identical on refactor itself. Real CI covers docs and PVA.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Team

Run ID: 2deb4c67-ca26-4f66-8957-3253fa6e16c0

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 92.18%. Comparing base (e73453b) to head (6e31163).
⚠️ Report is 4 commits behind head on refactor.

Additional details and impacted files

@@             Coverage Diff              @@
##           refactor     #420      +/-   ##
============================================
+ Coverage     91.25%   92.18%   +0.93%     
============================================
  Files            72       69       -3     
  Lines          2892     3185     +293     
============================================
+ Hits           2639     2936     +297     
+ Misses          253      249       -4     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[shihab-dls]

Requesting changes related to tests. However, tested the temperature controller with these changes, and the functionality works.

[shihab-dls]

We need to be able to handle reconnects on controllers correctly. In the current iteration, we simply try to reconnect a controller if its scan loop fails; However, a controller may not own its connection, and thus reconnecting it will call a no-op connect() method. Therefore, controllers must be able to share a Connection object. The connect() method is then a method on the Connection object, and should be used for reconnections as well. Moreover, currently, if you reconnect a controller, and are doing your introspection during connect() (such as fastcs-pandablocks), then you may find different introspected data upon the reconnect. This should be checked and raise an exception. With that being said, we should go with this approach:

• A controller no longer has a self._connected bool, instead, it has a self.connection: Connection|None object. Something like this:

class Connection(ABC, Generic[T]):
    """A link to hardware. Owns its own health state.

    Several controllers may share one instance. The Runner watches connections, not
    controllers, and deduplicates by identity - so never give a Connection an __eq__.
    """

    def __init__(
        self,
        depends_on: "Connection | None" = None,
        reconnect_period: float | None = None,
        max_attempts: int | None = None,
    ) -> None:
        self._connected = False
        # Declared, never derived. A channel layered over an SSH session names that
        # session here; the Runner will not attempt this one while that one is down.
        self.depends_on = depends_on


    @property
    def connected(self) -> bool:
        return self._connected

    @abstractmethod
    async def connect(self) -> T:
        """Open the link, or raise. Return whatever introspection the caller needs.

        The framework marks the connection connected when this returns cleanly, and
        compares the return value against the one from startup on every reconnect.
        """

    @abstractmethod
    async def close(self) -> None:
        """Close the link. Called at shutdown and before a teardown-style reconnect."""

    @abstractmethod
    async def check(self) -> None:
        """Cheap query proving the link is alive. Raise if it is not.

        Abstract so it cannot be forgotten. A connection that is genuinely always
        alive once opened implements it as a no-op - a visible claim rather than a
        silent gap. Skipped automatically while normal IO is succeeding.
        """

    def set_disconnected(self) -> None:
        """Called by the connection's own IO when the transport fails."""
        self._connected = False

• A controller with a connection can pass it to its children for them use for their IO.
• A subcontroller can have its own Connection object if it needs to connect to a different part of hardware.
• Controllers should only set their self.connection object, and not call self.connection.connect() for themselves or their children.
• The ControllerRunner will collect the Connection objects, only holding unique connections such as:

def _collect_connections(roots) -> list[Connection]:
    """Run once at startup, level order, deduplicated by identity.

    Identity, not equality: two sockets with matching settings are two connections,
    and a Connection with an __eq__ would silently collapse them.
    """
    seen: dict[int, Connection] = {}
    for controller in _walk_level_order(roots):
        connection = getattr(controller, "connection", None)
        if connection is not None and id(connection) not in seen:
            seen[id(connection)] = connection
    return list(seen.values())

This way, a tree of subcontrollers that all share the same connection are treated as one, where any subcontroller can mark_disconnected() and this will be picked up in the reconnect loop. If a subcontroller has its own connection object, then this connection will be reconnected independently if required, only after its depends_on connection is connected, in case there is an order that must be followed.

• The ControllerRunner will then have:

def _health_line(self) -> None:
    """One pass over every connection, deciding what to attempt.

    Synchronous and side-effect-light on purpose: it only schedules work, so it can be
    tested by hand-building a list of connections and asserting on what came back.

    `ready` and `blocked` are rebuilt from scratch every sweep. Nothing moves between
    lists - a blocked connection simply lands in `ready` on the first sweep after its
    dependency comes back.
    """
    ready: list[Connection] = []
    blocked: list[Connection] = []

    for connection in self._connections:
        state = self._state[connection]

        if state.task is not None:
            if not state.task.done():
                continue        # an attempt from an earlier sweep is still running
            state.task = None

        if connection.connected or state.exhausted:
            continue

        dependency = connection.depends_on
        if dependency is not None and not dependency.connected:
            # Cannot succeed yet. Skipping it here means _attempt never runs, so the
            # retry budget freezes for free rather than being burnt against a dead
            # dependency.
            blocked.append(connection)
            continue

        ready.append(connection)

    # Own task per attempt rather than a gather, so one slow reconnect cannot hold up
    # the next sweep for the others.
    for connection in ready:
        self._state[connection].task = asyncio.create_task(self._attempt(connection))

    if ready or blocked:
        self.log_event("Reconnecting", attempting=len(ready), blocked=len(blocked))

and

async def _attempt(self, connection: Connection) -> None:
    """One reconnect attempt. Owns retry accounting and the introspection check, and
    is the only place a connection is marked back up."""
    state = self._state[connection]
    state.attempts += 1

    try:
        await connection.close()          # tolerate an already-closed link
        introspection = await connection.connect()
        await connection.check()          # verify, rather than assume, recovery
    except Exception:
        self.log_exception("Reconnect failed", connection=connection)
        if state.attempts >= self._max_attempts(connection):
            # Terminal until the process restarts. Dependents stay blocked and never
            # attempt, so they will not log a giving-up message of their own - name
            # them here so the log accounts for what this takes down with it.
            state.exhausted = True
            self.log_error("Giving up", connection=connection, blocks=state.dependents)
        return

    if introspection != state.introspection:
        # The device came back describing itself differently. build() can never run
        # again, so there is no way to accommodate the new shape.
        self._fatal_introspection_mismatch(connection, state.introspection, introspection)

    connection._connected = True
    state.attempts = 0                     # a clean connection restores the budget

So we try to reconnect a connection, we increment its attempt count using ConnectionState:

@dataclass
class ConnectionState:
    task: asyncio.Task | None = None
    attempts: int = 0
    exhausted: bool = False

Essentially giving each connection a max_attempts which lives on the ControllerRunner, alongside the reconnect_period which tells the runner how often to run the _health_line.

The code I shared references build() and that is because we should rename initialise to build , and post_initialise to setup. We are also saying that a connect from a Connection can return some build info which is the introspected data it may have received on connection, that will be passed to build for it to make its attributes and sub controllers. A connection may not return any info and build may not use any. Drivers that introspect on build will still work, but they will not benefit from the build info equivalency check if they reconnect, so they should move that logic to their Connection object connect().

• fastcs.yaml will accept this:

connections:
  reconnect_period: 1.0
  max_attempts: 10

which will be passed into the ControllerRunner to store as state.

• The ControllerRunner will use the controller api's to traverse the controller tree, calling connect() on all controllers of a level asynchronously, then passing any build info to build() and calling that asynchronously. Once a level is connected and built, the next level children is connected and built, and so on. It might be best to merge this process in with _collect_connections, such that we collect our unique connections as we are doing our start up sequence. A Connection should hold any build info it produces, or the ControllerRunner should map a Connection to the build info, such that if subcontrollers that all share the same parent connection have their connect() called, this should cause a no-op as the Connection is already connected (make it idempotent), and the cached build info that the parent would have gotten should be passed to the subcontroller build() method for them to use for their own adding of attributes and sub controllers. It may also be better to have the build info on the Connection as a cached property, and then get that property and pass it to build of the controller in the ControllerRunner, instead of returning it on connect().

[coretl]

No code change on this yet — it's a bigger design than #395 covers, so I'd rather put the trade-off to you than start guessing at it.

The diagnosis is right, and this PR makes it worse rather than better. Controller.reconnect() on a controller that doesn't own its connection is a no-op, and this PR is what turns that from "nothing ever happens" into "nothing happens once a second, per controller". A shared-connection tree of N sub-controllers now has N reconnect loops all calling no-op connect()s, none of which can fix the one link that's actually down. Watching connections rather than controllers, and deduplicating by identity, is the right subject.

The introspection point is also real and I hadn't considered it: initialise() runs once, so a driver that discovers its attribute tree on connect can come back after a reconnect describing a device it can no longer represent. Failing loudly there is better than silently serving a stale tree.

Why I've not just implemented it. What you've sketched is a good deal more than a fix to the reconnect loop:

• a new public Connection ABC — which, per decision 13, has to go in the stable-interface doc this PR adds, so it's ADR surface not just code;
• depends_on ordering, per-connection retry budgets, _health_line sweeps, fastcs.yaml connections: config;
• renaming initialise → build and post_initialise → setup, which breaks every downstream driver (fastcs-pandablocks, fastcs-thorlabs-mff, the demo controllers here);
• threading build info from connect() into build(), plus level-order startup.

ADR 0016 as written doesn't decide any of that — it says "the runner owns the whole lifecycle including reconnect" and stops. So this wants an ADR amendment ahead of the code, or it lands as a design decided in a PR thread.

Two ways to go, and I'd pick the second:

1. Hold controllers: ControllerRunner, plus native timestamps and severity on attributes #420 and fold the whole Connection design in. One coherent landing, no interim mechanism to throw away. Costs: [ophyd-async] Embedded FastCS connector (fastcs-embed extra) #399 (the embedded ophyd-async connector) is waiting on ControllerRunner existing at all, and the timestamps/severity half of this PR has nothing to do with connections and would sit behind a much larger change.
2. Land controllers: ControllerRunner, plus native timestamps and severity on attributes #420 as the lifecycle move + timestamps/severity, and open Connection as its own issue under ophyd-async / FastCS API Convergence #388 with an ADR 0016 amendment. The runner exists for [ophyd-async] Embedded FastCS connector (fastcs-embed extra) #399 to build on, and the connection rework changes _reconnect_loop — one method, ~20 lines — rather than colliding with the rest.

If (2), there's a sub-question: does the automatic reconnect stay in the interim? It genuinely recovers the owns-its-connection case (which the demo and fastcs-thorlabs-mff are), and is a no-op for the shared case, so it's not wrong so much as incomplete. But it's also a documented behaviour change I'd be shipping knowing it's about to be replaced. I'd keep it, and say so in the ADR amendment; happy to strip it from this PR instead if you'd rather not have a mechanism land twice.

Points I'd want settled in the ADR either way — mostly places the sketch leaves a decision implicit:

• connection._connected = True in _attempt writes a private attribute from outside. Suggest a set_connected() next to the existing set_disconnected(), so the flag has one owner.
• What "fatal" means for the introspection mismatch. _fatal_introspection_mismatch runs in a background task, where a raise is invisible — so it needs to be an explicit process exit, or a runner-level failure the embedder can observe. Worth deciding, because an embedded FastCS inside an ophyd-async process must not call sys.exit.
• How build info is compared. introspection != state.introspection is fine for a dataclass, but introspection results are often dicts of numpy arrays, where != is elementwise and ambiguous in a bool context. Either the ADR says build info must be hashable/comparable, or Connection gets a comparison hook.
• How check() gets skipped "while normal IO is succeeding." Nothing currently records a last-success time, so this needs a signal from the IO path — probably the mirror of set_disconnected().
• depends_on declared vs derived: agreed it should be declared, but then two connections can declare a cycle. Detect at collection time?
• max_attempts exhausting to a terminal state means a device that's down for longer than reconnect_period * max_attempts never comes back without a process restart. For a beamline that may be the wrong default — is None (retry forever) the default, with a finite budget opt-in?

Say which route you want and I'll get on with it. If (2), I'll open the issue with this design and the questions above, and link it here.

— overnight agent

---

Generated by Claude Code