Skip to content
Peppy

Actions

Actions are for long-running tasks that need feedback during execution and support cancellation. A client sends a goal to an action node, which can provide periodic feedback while working and delivers a final result upon completion.

Use actions for tasks like navigation, arm movement, or any operation that runs over time and benefits from progress updates. (For a continuous bidirectional exchange with no defined end, an inner control loop rather than a discrete job, use a pairing instead.) For the full map of mechanisms, see Choosing a communication pattern.

A node can drive multiple goals concurrently for the same action. Each accepted goal yields its own GoalContext (owning that goal’s feedback stream, cancel signal, and result), so one server can route by a discriminator in the request (e.g. arm_id) and drive several independent resources in parallel. The framework routes each client’s cancel and result requests to the right goal by goal_id; deciding whether to accept a second concurrent goal is up to your goal handler.

Action lifecycle

Section titled “Action lifecycle”

An action consists of three communication channels built on top of services and topics:

  1. Goal (service): the client sends a goal request; the server accepts or rejects it.
  2. Feedback (topic): the server publishes progress updates while working on the goal.
  3. Result (service): the client requests the final result once the server finishes.

Additionally, the client can issue a cancel request at any time to abort an active goal.

Client                              Server
  │                                    │
  │──── fire_goal (request) ──────────>│
  │<─── goal ack (+ GoalResponse) ─────│
  │                                    │
  │<─── feedback ──────────────────────│  (repeated)
  │<─── feedback ──────────────────────│
  │                                    │
  │──── get_result (request) ─────────>│
  │<─── ResultResponse ────────────────│

Exposing an action

Section titled “Exposing an action”

A node that handles action goals declares its actions under interfaces.actions.exposes in its peppy.json5. Each action defines a goal_service, a feedback_topic, and a result_service:

{
  peppy_schema: "node/v1",
  manifest: {
    name: "brain",
    tag: "v1",
  },
  interfaces: {
    actions: {
      exposes: [
        {
          name: "move_arm",
          goal_service: {
            request_message_format: {
              arm_id: "u16",
              desired_position: {
                $type: "array",
                $items: "i32",
                $length: 3
              }
            }
          },
          feedback_topic: {
            qos_profile: "sensor_data",
            message_format: {
              new_position: {
                $type: "array",
                $items: "i32",
                $length: 3
              }
            }
          },
          result_service: {
            response_message_format: {
              success: "bool",
              error_msg: {
                $type: "string",
                $optional: true
              },
              final_position: {
                $type: "array",
                $items: "i32",
                $length: 3
              }
            }
          }
        }
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "brain"]
  },
}

The goal_service payloads and the result_service response are all optional: any of goal_service.request_message_format, goal_service.response_message_format, and result_service.response_message_format can be omitted (or set to {}) when the corresponding payload is empty. For example, a calibrate action can omit goal_service.request_message_format because it has no goal parameters, and a result_service that only signals completion can omit its response_message_format too.

The result_service never carries a user-defined request. A get_result call is correlated to its goal by the goal_id the framework assigns at fire_goal, so the only payload you declare on the result side is response_message_format, which the producer supplies through complete or complete_cancelled. Declaring a result_service.request_message_format is rejected when the config is parsed, since no generated API, wire message, or producer handler could ever read it.

The feedback_topic block is optional: actions that don’t stream progress can omit it entirely. When you do declare feedback_topic, however, its message_format is required and cannot be empty: every feedback message must carry a non-empty payload, since empty payloads are reserved as the framework’s per-goal end-of-stream signal (see Receiving feedback). Code generation will fail if feedback_topic is declared without a message_format.

Handling goals

Section titled “Handling goals”

After running peppy node sync, the code generator creates a module for each exposed action under peppygen::exposed_actions. Use ActionHandle::expose to set up the action, then loop accepting goals. handle_goal_next_request returns the next accepted goal as a GoalContext; rejected goals are answered and skipped for you, so the accept loop ends only when the goal stream closes (the node is shutting down). Spawn a worker per context so goals run concurrently; each context owns that goal’s feedback, cancel signal, and result, so nothing crosses between goals:

import asyncio


from peppygen import NodeBuilder
from peppygen.exposed_actions import move_arm




# Drive one accepted goal concurrently with any others already running. The
# context owns this goal's feedback, cancel signal, and result.
async def drive(ctx):
    # Feedback goes through this goal's context, not a shared slot.
    await ctx.publish_feedback([7, 31, 43])
    # Deliver the result for this specific goal; the client's
    # get_result(handle) is routed back here by goal_id.
    await ctx.complete(True, None, [98, 4, 26])




async def run_action(node_runner):
    action = await move_arm.ActionHandle.expose(node_runner)


    # The decider sets the concurrency policy: e.g. reject a goal for a busy
    # arm_id. Rejected goals are answered and skipped internally.
    def decide(request):
        print(
            f"goal from {request.instance_id}: arm_id={request.data.arm_id} "
            f"desired={request.data.desired_position}",
            flush=True,
        )
        return move_arm.GoalDecision.accept()


    while True:
        ctx = await action.handle_goal_next_request(decide)
        if ctx is None:
            break  # goal stream closed (node shutting down)
        asyncio.create_task(drive(ctx))




async def setup(parameters, node_runner):
    # Spawn the accept loop so setup returns and the node starts serving.
    return [asyncio.create_task(run_action(node_runner))]




def main():
    NodeBuilder().run(setup)




if __name__ == "__main__":
    main()

GoalDecision controls admission. The decision and its optional rejection reason travel in a framework-owned envelope wrapped around the goal reply, so they never appear in your schema: GoalResponse contains exactly the fields declared in goal_service.response_message_format, and the framework never adds an accepted or error field to it.

  • GoalDecision::accept() (GoalDecision.accept() in Python) admits the goal, replies to the client, and yields a GoalContext. When the action declares a response_message_format, the constructor takes the declared response instead: GoalDecision::accept(response).
  • GoalDecision::reject(reason) (GoalDecision.reject(reason) in Python) declines it: the client sees accepted == false and the reason on its handle, no context is created, and the accept loop transparently moves on to the next goal. This is where you enforce per-resource concurrency limits.
  • A reject can also carry the declared response payload: construct GoalDecision::Reject { reason, response } directly in Rust, or pass both arguments in Python with GoalDecision.reject(reason, response). The client’s data is absent when a rejection carried no payload.

The GoalRequest passed to the decider contains:

  • instance_id: the client instance that sent the goal. The producer is binding-agnostic; it doesn’t know which slot on the client this goal is heading to.
  • core_node: the core node of the caller.
  • data: the deserialized goal parameters (only present when request_message_format is defined).

The GoalContext (ctx) is the only handle you need to drive the goal:

  • ctx.request(): the decoded GoalRequest.
  • ctx.goal_id(): this goal’s correlation id.
  • ctx.publish_feedback(...): publish a feedback message on this goal’s stream.
  • ctx.cancel_signal() / ctx.is_cancelled(): observe cancellation (see below).
  • ctx.complete(...) / ctx.complete_cancelled(...): deliver the final result.

A producer exposes its action once and serves any client that fires a goal at it. Starting the producer before or after the client is equally valid.

To enforce a concurrency limit (say, one in-flight goal per arm), reject a goal whose arm is already busy, and release the arm when the goal finishes. The decider and the per-goal workers share the busy set:

# Arms currently driving a goal. Shared by the decider and the workers;
# asyncio is single-threaded, so a plain set is safe here.
busy: set[int] = set()




async def drive(ctx):
    arm_id = ctx.request().data.arm_id
    await ctx.complete(True, None, [98, 4, 26])
    # Release the arm so future goals for it are accepted again.
    busy.discard(arm_id)




def decide(request):
    arm_id = request.data.arm_id
    if arm_id in busy:
        # The reason rides back to the client in the goal ack.
        return move_arm.GoalDecision.reject(f"arm {arm_id} is already moving")
    busy.add(arm_id)
    return move_arm.GoalDecision.accept()




while True:
    ctx = await action.handle_goal_next_request(decide)
    if ctx is None:
        break
    asyncio.create_task(drive(ctx))

Handling cancellation

Section titled “Handling cancellation”

A worker reacts to a cancel for its goal via ctx.cancel_signal(), which resolves when a cancel request arrives for that goal_id. Pair it with the goal’s work and report the outcome with complete_cancelled:

async def drive(ctx):
    cancel_task = asyncio.ensure_future(ctx.cancel_signal())
    work_task = asyncio.ensure_future(run_arm(ctx.request().data.arm_id))
    done, pending = await asyncio.wait(
        [cancel_task, work_task], return_when=asyncio.FIRST_COMPLETED
    )
    for task in pending:
        task.cancel()


    if cancel_task in done:
        # A cancel arrived for this goal; wind down and report it.
        await ctx.complete_cancelled(False, "cancelled", last_known_position)
    else:
        outcome = work_task.result()
        for position in outcome.steps:
            await ctx.publish_feedback(position)
        await ctx.complete(True, None, outcome.final_position)

Cancellation is auto-acknowledged by the framework: the client’s cancel_goal returns a typed CancelState: Signalled when a goal with that goal_id is in flight (the signal was delivered), AlreadyTerminal when it had already finished, or Unknown. Signalled means delivered, not will stop; a worker is free to ignore the signal and keep running. The worker decides the goal’s fate: calling complete_cancelled (or complete) is what a subsequent get_result returns. Whichever completion runs first wins; the framework closes this goal’s feedback stream on completion, so the client’s on_next_feedback_message loop ends cleanly. The clean close is not the only way a stream ends: if the producer instance dies mid-goal, the client’s loop ends with a producer-gone error instead (see Receiving feedback). A goal’s cancel never affects other concurrent goals.

Consuming an action

Section titled “Consuming an action”

A node that sends goals declares what it consumes under interfaces.actions.consumes. Dependencies are declared in manifest.depends_on and referenced by link_id in the interface:

{
  peppy_schema: "node/v1",
  manifest: {
    name: "controller",
    tag: "v1",
    depends_on: {
      nodes: [
        { name: "brain", tag: "v1", link_id: "brain" },
      ]
    },
  },
  interfaces: {
    actions: {
      consumes: [
        {
          link_id: "brain",  // References depends_on.nodes[].link_id
          name: "move_arm",        // Action name on that node
        },
      ],
    },
  },
  execution: {
    language: "python",
    build_cmd: ["uv", "sync"],
    run_cmd: ["uv", "run", "controller"]
  },
}

Firing a goal

Section titled “Firing a goal”

The code generator creates a module for each consumed action under peppygen::consumed_actions. Use fire_goal to send a goal, then listen for feedback and request the result:

import asyncio


from peppygen import NodeBuilder, QoSProfile
from peppygen.consumed_actions.brain import move_arm




async def run(node_runner):
    # `one`: the accessor returns the slot's sole producer directly.
    arm = move_arm.bound_producer(node_runner)
    request = move_arm.GoalRequest(arm_id=7, desired_position=[10, 20, 30])
    action_handle = await move_arm.ActionHandle.fire_goal(
        node_runner,
        arm,                    # which bound instance executes the goal
        request,
        5.0,                    # timeout (seconds)
        QoSProfile.SensorData,  # QoS for the feedback topic
    )
    print(
        f"goal accepted={action_handle.accepted} reason={action_handle.reason}",
        flush=True,
    )




async def setup(parameters, node_runner):
    return [asyncio.create_task(run(node_runner))]




def main():
    NodeBuilder().run(setup)




if __name__ == "__main__":
    main()

fire_goal requires the caller to pass one explicit target for every cardinality, including the default one; obtain it from the slot’s cardinality-typed accessor (the singular, infallible bound_producer() on a one slot, bound_producers() on the multi cardinalities, never empty for one_or_more). The target must belong to the slot’s own set; anything else fails with a runtime error before the goal reaches the wire. Firing a goal at every bound producer of a multi-cardinality slot is a plain loop over bound_producers(); each returned handle drives its own feedback stream, cancel, and result, exactly as a single goal does.

fire_goal returns an ActionHandle. Its accepted flag and optional rejection reason are decoded from the framework goal ack, and, when the action declares a response_message_format, its data field holds the declared goal response (absent when a rejection carried no payload). The handle retains the selected target: goal submission, feedback, result retrieval, and cancellation all stay pinned to that same producer; no stage uses a wildcard, fallback producer, discovery, or automatic retargeting. The handle is what you use for subsequent feedback, result, and cancel calls. You can fire multiple goals concurrently, and each returns its own handle, so the server can drive them in parallel.

Receiving feedback

Section titled “Receiving feedback”

Use on_next_feedback_message on the handle to receive feedback. The idiomatic pattern is a loop that drains feedback until a terminal error ends the stream:

while True:
    try:
        feedback = await action_handle.on_next_feedback_message()
        print(f"new_position={feedback.new_position}", flush=True)
    except ConnectionError:
        break  # producer instance died; get_result yields ABANDONED
    except Exception:
        break  # server has completed (or cancelled) this goal

Each goal has its own feedback stream, addressed by goal_id, so feedback for one goal never reaches another handle, even when several goals run concurrently on the same server. Two terminal errors end the drain loop:

  • Clean close (ActionFeedbackChannelClosed; RuntimeError in Python): the server closed this goal’s stream: its worker completed the goal (complete or complete_cancelled) or abandoned it without completing it (an early return or a panic). Call get_result to learn the outcome.
  • Producer gone (ActionFeedbackProducerGone; ConnectionError in Python): the producer instance this goal is pinned to died without closing the stream (process killed, host lost). Feedback already in flight (including a clean close that did make it out) is drained first, so a graceful shutdown never surfaces as producer-gone. After this error, get_result resolves to the Abandoned outcome.

Either way, a client draining feedback always terminates rather than hanging. In Python, catch ConnectionError before the generic handler when you want to distinguish producer death from a clean close; a bare except Exception treats both as end-of-stream.

Getting the result

Section titled “Getting the result”

Use get_result on the handle to request the final result. The call is routed to this goal by goal_id and parks until the goal reaches a definitive terminal state, then returns a typed outcome:

result = await action_handle.get_result(5.0)


if result.status == move_arm.ResultStatus.COMPLETED:
    print(
        f"completed: success={result.data.success} error={result.data.error_msg} "
        f"final_position={result.data.final_position}",
        flush=True,
    )
elif result.status == move_arm.ResultStatus.CANCELLED:
    print(f"cancelled at {result.data.final_position}", flush=True)
elif result.status == move_arm.ResultStatus.ABANDONED:
    print("the worker abandoned the goal without producing a result", flush=True)
elif result.status == move_arm.ResultStatus.EXPIRED:
    print("the result expired before it was fetched", flush=True)

A poll on a still-running goal parks until the goal reaches a terminal state, then returns a typed outcome, so you can call get_result whenever you like; a forwarding/relaying node does not have to time its poll to the worker’s lifecycle. The returned outcome (a Rust ResultOutcome enum, or result.status plus optional result.data in Python) tells you exactly what happened:

  • Completed / Cancelled: the worker delivered a result via complete / complete_cancelled; the payload is in data.
  • Abandoned: the worker dropped the goal without ever delivering a result (an early return or a panic), or the producer instance itself died mid-goal. A poll on a goal whose producer is confirmed gone resolves to Abandoned instead of parking forever.
  • Expired: the goal finished, but its result was retained only for a bounded window (30 s by default) that has since elapsed, or the result was already evicted.

A terminal result stays fetchable for that retention window, and you can fetch it more than once within it, which makes relaying it reliable. If the poll outlives the caller’s own timeout, you get a normal timeout error like any other request.

Cancelling a goal

Section titled “Cancelling a goal”

Use cancel_goal on the handle to request cancellation of that specific goal. It returns a typed CancelState:

cancel_response = await action_handle.cancel_goal(5.0)


if cancel_response.state == move_arm.CancelState.SIGNALLED:
    print("cancel delivered to a live goal", flush=True)
elif cancel_response.state == move_arm.CancelState.ALREADY_TERMINAL:
    print("goal had already finished", flush=True)
elif cancel_response.state == move_arm.CancelState.UNKNOWN:
    print("no goal with that id is known", flush=True)

The CancelState reports what the cancel found:

  • Signalled: a live goal received the cancel signal. Delivered, not necessarily will stop: the worker may ignore it.
  • AlreadyTerminal: the goal had already reached a terminal state, so there was nothing to cancel (best-effort; observable only while the result is still retained).
  • Unknown: no goal with that goal_id is known (it never existed, or was evicted long ago).

The cancel targets only this goal; other concurrent goals are unaffected. To learn the goal’s final state, call get_result: if the worker reacted with complete_cancelled you get a Cancelled outcome; if it ignored the cancel and finished normally you get Completed.

Bindings and routing

Section titled “Bindings and routing”

Routing for actions uses the same consumer-side model as topics and services. A binding KEY: VALUE creates a private channel from producer instance VALUE to one of the client’s declared slots; the action server itself is binding-agnostic.

An action slot resolves through its bindings: the generated fire_goal checks the caller-selected target against the slot’s bound set and pins the goal, cancel, result, and feedback channels to that one producer’s full (core_node, instance_id) wire address. How many producers may be bound to the slot is its declared cardinality (one when omitted); because the whole goal cycle needs exactly one server, every fired goal selects exactly one member of the bound set, and a one / one_or_more slot with no binding is rejected at launch validation, before anything spawns.

In a launcher / stack config:

{
  source: { local: "./consumer" },
  instances: [{
    instance_id: "my_consumer",
    links: { brain: "left-arm-1" },
  }],
}

or, when launching a single node during development:

Terminal window
peppy node run --link brain@left-arm-1 .

Worked example: openarm01_backbone

Section titled “Worked example: openarm01_backbone”

A client that wires two depth cameras to two dedicated slots:

openarm01_backbone/peppy.json5
{
  manifest: {
    name: "openarm01_backbone",
    tag: "v1",
    depends_on: {
      contracts: [
        { name: "depth_camera", tag: "v1", link_id: "wrist_left_camera" },
        { name: "depth_camera", tag: "v1", link_id: "wrist_right_camera" },
      ],
    },
  },
  // ...
}
peppy_launcher.json5
{
  deployments: [
    { source: { name: "depth_camera:v1" }, instances: [
      { instance_id: "left_cam" },
      { instance_id: "right_cam" },
    ]},
    { source: { name: "openarm01_backbone:v1" }, instances: [
      { instance_id: "backbone_inst_1", links: {
        wrist_left_camera: "left_cam",
        wrist_right_camera: "right_cam",
      }},
    ]},
  ],
}

Three contract statements follow from this manifest:

  1. fire_goal on the nested wrist_left_camera.<action> (Python) / wrist_left_camera::<action> (Rust) module reaches left_cam.
  2. fire_goal on the nested wrist_right_camera.<action> (Python) / wrist_right_camera::<action> (Rust) module reaches right_cam.
  3. If the wrist_right_camera binding line were removed, validation would reject the launch (every declared one / one_or_more slot must have a binding): a goal cycle has no wildcard fallback.

Why an explicit single target?

Section titled “Why an explicit single target?”

Without a pinned target, a goal request would be broadcast to every matching producer and each one would execute the goal concurrently, which for state-changing actions (motor commands, file writes, payment dispatches) is a real-world hazard. Requiring every fire_goal to name one member of the slot’s validated bound set removes that hazard by construction: the selected target carries the producer’s full (core_node, instance_id) wire address, so the goal addresses exactly one producer with no probe round-trip, the goal, cancel, result, and feedback channels all stay on that producer for the whole goal cycle, and an out-of-set target (never checked by plan-time binding validation) is rejected before it reaches the wire. If the selected producer dies mid-cycle, the affected call surfaces ServiceUnreachable and the caller can retry once the producer is back; the bound set itself never shrinks or rebinds at runtime.

Validator rules

Section titled “Validator rules”

The launcher validator runs these checks before the stack starts:

  1. Every KEY must name a declared slot, and every declared slot must resolve. A binding whose KEY matches no depends_on link_id is rejected; there are no free-form keys. A declared one / one_or_more slot with no binding entry fails the launch before anything is spawned; a zero_or_more slot with no entry resolves to the empty set.
  2. The value’s shape must match the slot’s cardinality. A one slot takes a scalar, a multi slot takes an array, an empty array meets only zero_or_more, and duplicate targets within one slot are rejected. Repeated --link KEY@… flags accumulate on a multi-slot and are a hard error on a one slot.
  3. Every target must satisfy the slot, checked per bound instance. A target instance_id that deploys a different node than the slot expects (or one that does not implement the requested contract) is rejected.
  4. Stack-wide instance_id uniqueness. Every instance_id must be unique across the entire stack, not just within a (node_name, node_tag) group. The --link syntax names producers by instance_id, so a duplicate would make the binding ambiguous.
  5. Bindings are stamped with the daemon’s core_node. The wire addresses producers by the full (core_node, instance_id) pair; the validator stamps the launching daemon’s core_node into every resolved binding, preserving application declaration order, so generated calls address exactly the selected producer and never match on instance_id alone.

Concurrent processing

Section titled “Concurrent processing”

A single action server can drive many goals at once. The accept loop only waits for the next goal; each accepted goal runs in its own spawned task with its own GoalContext. The framework routes every cancel and result request to the right goal by goal_id, and each goal has its own feedback stream, so goals never interfere with one another.

This makes the “one server, many resources” pattern natural: include a discriminator in the goal request (e.g. arm_id or device_id) and route to a per-resource worker. Your goal handler is responsible for the concurrency policy: accept goals to run them in parallel, or reject a goal (with GoalDecision::reject(reason)) when its target resource is already busy. A goal that is not accepted yields no GoalContext and cannot be cancelled or completed.