Handoffs, streams and choreography

• Handoffs pass a task from one agent to the next with only the context you put in the payload.
• Streaming channels push data from a producer to subscribers as it is produced.
• Choreography lets agents react to each other's events.

Handoffs

A Handoff sends a task from one agent to another. The receiver gets only the HandoffContext, not the sender's history or reasoning. That suits a bounded sub-task, a call across a trust boundary, or a one-shot delegation. For work where a later step must revisit earlier decisions, use the recursive solver; to check another agent's output, use a judge.

HandoffContext is a record with taskId (required), payload, metadata and parentHandoffId. Null maps become empty maps. HandoffContext.of(taskId) and HandoffContext.of(taskId, payload) build the common cases, and withPayload(key, value) returns a copy with one more payload entry.

DefaultHandoff is the shipped strategy. It delivers a HANDOFF_REQUEST message with the payload and task id to the receiver, then calls the receiver's chat with Process handoff task: <taskId> | payload: <payload>. The reply becomes the result's output. If anything throws, the result is a failure with a null output.

HandoffResult is a record with success, fromAgentId, toAgentId, durationMs and output. HandoffListener has onBefore, onAfter and onFailed, all no-ops by default; add one with DefaultHandoff.addListener.

Chains and conditional routing

HandoffChain runs a fixed sequence. The first agent is only the source and never runs. Each later agent receives a handoff from the previous one, and the next context's payload is just {"previousOutput": <output>}; the task id and metadata carry over. A step added with a predicate is skipped when the predicate returns false, and the chain stops at the first failed step. execute returns one result per step that ran, or an empty list for fewer than two agents.

ConditionalHandoff routes one task to the first agent whose predicate matches, or to the otherwise agent. With no match and no fallback it returns a failure whose toAgentId is none.

package com.example.tnsai.docs;

import com.tnsai.agents.Agent;
import com.tnsai.coordination.handoff.ConditionalHandoff;
import com.tnsai.coordination.handoff.DefaultHandoff;
import com.tnsai.coordination.handoff.HandoffChain;
import com.tnsai.coordination.handoff.HandoffContext;
import com.tnsai.coordination.handoff.HandoffListener;
import com.tnsai.coordination.handoff.HandoffResult;

import java.util.List;
import java.util.Map;

/** Compilation contract for multi-agent/handoffs.md handoff chains. */
public final class HandoffExample {

    private HandoffExample() {
    }

    static void run(Agent intake, Agent extractor, Agent translator, Agent writer,
                    Agent billing, Agent support) {
        DefaultHandoff handoff = new DefaultHandoff();
        handoff.addListener(new HandoffListener() {
            @Override
            public void onAfter(HandoffResult result) {
                System.out.println(result.fromAgentId() + " -> " + result.toAgentId()
                    + " in " + result.durationMs() + "ms");
            }
        });

        // intake is the source only; extractor, translator and writer run in turn.
        HandoffChain chain = HandoffChain.builder()
            .add(intake)
            .add(extractor)
            .add(translator, ctx -> "de".equals(ctx.metadata().get("language")))
            .add(writer)
            .build();

        HandoffContext start = new HandoffContext(
            "invoice-42",
            Map.of("document", "Invoice text ..."),
            Map.of("language", "de"),
            null);

        List<HandoffResult> results = chain.execute(handoff, start);
        HandoffResult last = results.get(results.size() - 1);
        System.out.println(last.success() ? last.output() : "Failed at " + last.toAgentId());

        // Route one task to the first matching agent.
        HandoffResult routed = new ConditionalHandoff()
            .when(ctx -> ctx.payload().containsKey("invoiceId"), billing)
            .otherwise(support)
            .execute(handoff, intake, HandoffContext.of("ticket-7", Map.of("invoiceId", "42")));
        System.out.println(routed.toAgentId() + ": " + routed.output());
    }
}

Because each hop keeps only the previous output, a chain suits a known pipeline of specialised steps. Information from the first step is gone by the third.

Streaming channels

StreamingChannel<T> is a typed buffer with a dispatcher thread. Producers call emit; every subscriber receives every item in order. Items emitted before the first subscriber wait in the buffer.

package com.example.tnsai.docs;

import com.tnsai.coordination.streaming.BackpressureStrategy;
import com.tnsai.coordination.streaming.StreamingChannel;

import java.time.Duration;
import java.util.List;

/** Compilation contract for multi-agent/handoffs.md streaming channels. */
public final class StreamingChannelExample {

    private StreamingChannelExample() {
    }

    static void stream() {
        StreamingChannel<String> channel = StreamingChannel.<String>builder()
            .id("analysis-stream")
            .bufferSize(100)
            .backpressure(BackpressureStrategy.BLOCK)
            .offerTimeout(Duration.ofSeconds(5))
            .build();

        // Consumer side
        StreamingChannel<Integer> lengths = channel
            .filter(chunk -> !chunk.isBlank())
            .map(String::length);
        lengths.subscribe(length -> System.out.println("Chunk of " + length + " chars"));
        channel.onError(error -> System.err.println("Subscriber failed: " + error.getMessage()));
        channel.onComplete(() -> System.out.println("Stream finished"));

        // Producer side
        channel.emit("first chunk");
        channel.emitAll(List.of("second chunk", "third chunk"));
        channel.complete();

        StreamingChannel.ChannelMetrics metrics = channel.metrics();
        System.out.println("Emitted " + metrics.emitted() + ", dropped " + metrics.dropped()
            + " (" + metrics.dropRate() + "%)");
    }
}

emit returns false when the item was dropped, and throws IllegalStateException after complete() or close(). emitAll returns how many items were accepted.

onError handlers run when a subscriber throws; the dispatcher keeps going. complete() stops new emits; buffered items are still delivered, then onComplete handlers run. close() stops the dispatcher at once and discards the buffer.

map and filter return new channels with the same buffer size and backpressure, named <id>-mapped and <id>-filtered. They complete when the parent completes.

metrics() returns ChannelMetrics with channelId, emitted, dropped, bufferUsed, bufferCapacity, subscriberCount and completed. dropRate() is the dropped share as a percentage.

Named channels

AgentStreamBridge keeps channels by name so agents can find each other's streams. createChannel(name, type) uses a 256-item DROP_OLDEST buffer; another overload takes the buffer size and strategy. Creating a name twice throws IllegalStateException. getChannel(name, type) returns null for an unknown name, while subscribe(name, type, consumer) throws. removeChannel and close close the channels they remove.

AgentStreamBridge bridge = new AgentStreamBridge();
StreamingChannel<String> results = bridge.createChannel("analysis-results", String.class);
bridge.subscribe("analysis-results", String.class, chunk -> System.out.println(chunk));
results.emit("partial result");

Choreography

In a choreography, no agent directs the others. A ChoreographySpec lists interactions; each one names the event that triggers it, the agent that runs it, the task, and optionally the event it emits when done. ChoreographyEngine publishes the initial events and runs each triggered interaction on its own thread.

package com.example.tnsai.docs;

import com.tnsai.agents.Agent;
import com.tnsai.coordination.choreography.ChoreographyEngine;
import com.tnsai.coordination.choreography.ChoreographyEngine.ChoreographyResult;
import com.tnsai.coordination.choreography.ChoreographySpec;

import java.time.Duration;

/** Compilation contract for multi-agent/handoffs.md choreography. */
public final class ChoreographyExample {

    private ChoreographyExample() {
    }

    static ChoreographyResult publish(Agent researcher, Agent writer, Agent reviewer) {
        ChoreographySpec spec = ChoreographySpec.builder("article")
            .initialEvent("start")
            //           id        source        target        trigger      emits        task
            .interaction("research", "system", "researcher", "start", "researched",
                "Collect sources on agent coordination")
            .interaction("draft", "researcher", "writer", "researched", "drafted",
                "Write a draft from the research")
            .interaction("review", "writer", "reviewer", "drafted", null,
                "Review the draft")
            .build();

        ChoreographyEngine engine = ChoreographyEngine.builder()
            .spec(spec)
            .agent("researcher", researcher)
            .agent("writer", writer)
            .agent("reviewer", reviewer)
            .timeout(Duration.ofMinutes(10))
            .build();

        ChoreographyResult result = engine.execute();
        result.stepResults().forEach(step ->
            System.out.println(step.interactionId() + " by " + step.agentId() + ": "
                + (step.success() ? "ok" : step.error())));
        return result;
    }
}

ChoreographySpec.builder(name) needs at least one interaction and one initialEvent. An interaction's id, sourceAgent, targetAgent, triggerEvent and taskDescription are required; emitEvent may be null. Only the target agent runs: the engine sends it the task, followed by Context: <output> when the triggering event carries the previous agent's output. The source agent is used for deadlock detection only.

The engine builder takes spec, agent(id, agent) or agents(map), and timeout, which defaults to 5 minutes. An interaction whose target id has no registered agent fails with Agent not found. execute() waits until every interaction has run once or the timeout passes, and executeAsync() returns a CompletableFuture. An engine runs once: execute() closes its event bus.

ChoreographyResult has success, stepResults, totalDurationMs, completedCount and failedCount. success means no step failed. Interactions that never triggered before the timeout leave no step result, so a timed-out run can still report success; compare stepResults().size() with the number of interactions. Each StepResult has interactionId, agentId, output, durationMs, success and error.

Deadlock and livelock checks

Before running, the engine calls DeadlockDetector.detect(spec), which looks for cycles in the source-to-target graph. A cycle is logged as a warning; the run continues. Call detect yourself to reject a spec, or detectFromWaitGraph(map) for a runtime wait-for graph. The result has cycles and hasDeadlock.

LivelockDetector is a separate tool you feed yourself with recordMessage, recordStateChange and recordAction. analyze() reports REPEATED_MESSAGE, OSCILLATING_STATE and PROGRESS_STALL violations. After circuitBreakerThreshold violations in total, the circuit breaker trips and recording throws until reset(). Builder defaults: a message repeated 5 times within 10 seconds, 4 state alternations, no new action for 30 seconds, and a breaker threshold of 3. addListener takes a LivelockListener.

Related

• Choosing a topology
• Recursive solver
• Judge
• Messaging