# Handoffs, streams and choreography Three ways to move work between agents without a central coordinator, all in `tnsai-coordination`: - **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](recursive-solver.md); to check another agent's output, use a [judge](judge.md). `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: | 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": }`; 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`. ```java 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 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` 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. ```java 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 channel = StreamingChannel.builder() .id("analysis-stream") .bufferSize(100) .backpressure(BackpressureStrategy.BLOCK) .offerTimeout(Duration.ofSeconds(5)) .build(); // Consumer side StreamingChannel 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() + "%)"); } } ``` | Builder method | Default | |---|---| | `id(String)` | A random 8-character id | | `bufferSize(int)` | 256 | | `backpressure(BackpressureStrategy)` | `DROP_OLDEST` | | `offerTimeout(Duration)` | 5 seconds, used by `BLOCK` | | `BackpressureStrategy` | When the buffer is full | |---|---| | `BLOCK` | Waits up to `offerTimeout` for space, then drops the item | | `DROP_OLDEST` | Drops the oldest buffered item to make room | | `DROP_NEWEST` | Drops the new item | | `ERROR` | Throws `StreamingChannel.BufferOverflowException` | `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 `-mapped` and `-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. ```java AgentStreamBridge bridge = new AgentStreamBridge(); StreamingChannel 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. ```java 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: ` 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](choose-topology.md) - [Recursive solver](recursive-solver.md) - [Judge](judge.md) - [Messaging](messaging.md)