feat(j5): agent tools follow the #519 decisions for listing, stop, send and admitted tools

apps/server/src/j5/a2a/DeliveryTransport.ts

@@ -22,6 +22,7 @@ import { HttpClient, HttpClientRequest } from "effect/http";
 import * as SqlClient from "effect/sql/SqlClient";
 
 import * as ThreadManagement from "../../orchestration-v2/ThreadManagementService.ts";
+import { DispatchModeLimit, type DispatchModes } from "../../orchestration-v2/DispatchModeLimit.ts";
 import { EffectOutboxV2 } from "../../orchestration-v2/EffectOutbox.ts";
 import { OrchestratorV2 } from "../../orchestration-v2/Orchestrator.ts";
 import {
@@ -35,6 +36,7 @@ import { PeerRegistryService } from "./PeerRegistryService.ts";
 import { peerProtocolHeaders, responseProtocolMismatch } from "./peerProtocol.ts";
 import {
   type DeliveryEnvelopeChannel,
+  type SendMode,
   LedgerProjectId,
   ExchangeId,
   isHumanParticipantId,
@@ -111,6 +113,10 @@ export interface AgentDeliveryInput {
   readonly exchangeRole: "none" | "ask" | "followup" | "reply" | "terminal_notice";
   readonly message: string;
   readonly attachments?: ReadonlyArray<ChatAttachment>;
+  /** How the sender asked for the message to land; absent means `queue`. */
+  readonly mode?: SendMode;
+  /** The broadest modes the sender could touch when it asked for a steer or a restart. */
+  readonly modeLimit?: DispatchModes;
   readonly envelopeChannel: DeliveryEnvelopeChannel;
   /** The peer server a received message came from, named in its envelope; absent for a local sender. */
   readonly senderServerName?: string;
@@ -161,6 +167,24 @@ const stablePart = (value: string) => encodeURIComponent(value);
 export const deliveryCommandId = (messageId: LedgerMessageId) =>
   CommandId.make(`command:j5:a2a:delivery:${stablePart(messageId)}`);
 
+/**
+ * The command an asked-for steer or restart is dispatched under. It is not the delivery's own:
+ * upstream records a rejected command for good, and a restart is rejected when the turn it was
+ * aimed at has just ended, so the plain delivery that follows must keep a command id that nothing
+ * has refused.
+ */
+export const askedDeliveryCommandId = (
+  messageId: LedgerMessageId,
+  mode: Exclude<SendMode, "queue">,
+) => CommandId.make(`${deliveryCommandId(messageId)}:${mode}`);
+
+/** Every command one message's delivery may have been dispatched under. */
+const deliveryCommandIds = (messageId: LedgerMessageId) => [
+  deliveryCommandId(messageId),
+  askedDeliveryCommandId(messageId, "steer"),
+  askedDeliveryCommandId(messageId, "restart"),
+];
+
 const DELIVERY_MESSAGE_ID_PREFIX = "message:j5:a2a:delivery:";
 
 export const deliveryMessageId = (messageId: LedgerMessageId) =>
@@ -231,6 +255,34 @@ export const gpt6PeerSteeringRun = (
     : undefined;
 };
 
+/**
+ * The running turn a `steer` or `restart` lands on, by upstream's own steering policy: a steer
+ * needs a provider that steers a turn directly or by interrupting and restarting it, and a restart
+ * needs the second. Undefined when the receiver has no such turn, and the message is then
+ * delivered as any other: it starts a turn on an idle agent, or waits behind a busy one.
+ */
+export const requestedSteeringRun = (
+  target: Pick<
+    OrchestrationV2ThreadProjection,
+    "thread" | "runs" | "providerTurns" | "providerThreads" | "providerSessions"
+  >,
+  channel: DeliveryEnvelopeChannel,
+  mode: Exclude<SendMode, "queue">,
+) => {
+  if (channel !== "peer" || target.thread.archivedAt !== null) return undefined;
+  const run = ThreadManagement.latestSteerableRun(target);
+  if (run === undefined) return undefined;
+  const providerThread = target.providerThreads.find(
+    (thread) => thread.id === run.providerThreadId,
+  );
+  const turns = target.providerSessions.find(
+    (candidate) => candidate.id === providerThread?.providerSessionId,
+  )?.capabilities.turns;
+  if (turns === undefined) return undefined;
+  const restarts = turns.supportsInterrupt && turns.supportsSteeringByInterruptRestart;
+  return (mode === "steer" ? turns.supportsActiveSteering || restarts : restarts) ? run : undefined;
+};
+
 export const formatAgentDeliveryEnvelope = (
   input: AgentDeliveryInput & { readonly originProjectTitle?: string | undefined },
 ): string => {
@@ -337,7 +389,11 @@ export const live: Layer.Layer<
               participantId: input.receiverId,
               state: "not an agent",
             });
-          const priorEffects = yield* outbox.listByCommandId(deliveryCommandId(input.messageId));
+          // A steer may sit under the delivery's own command (Astra) or the one its sender asked
+          // for; its outcome, not the message in the thread, says whether the provider took it.
+          const priorEffects = (yield* Effect.forEach(deliveryCommandIds(input.messageId), (id) =>
+            outbox.listByCommandId(id),
+          )).flat();
           const steer = priorEffects.find(
             (effect) => effect.request.type === "provider-turn.steer",
           );
@@ -471,6 +527,68 @@ export const live: Layer.Layer<
                   new A2ADeliveryHeldError({ participantId: input.receiverId, runId: run.id }),
                 )
               : Effect.void;
+          const mode = input.mode ?? "queue";
+          if (mode !== "queue") {
+            // The sender asked for a steer or a restart (see `askedDeliveryCommandId`).
+            const askedCommandId = askedDeliveryCommandId(input.messageId, mode);
+            const landed = Effect.fn("j5.a2a.delivery.landedAsAsked")(function* () {
+              const steer = (yield* outbox.listByCommandId(askedCommandId)).find(
+                (effect) => effect.request.type === "provider-turn.steer",
+              );
+              if (steer !== undefined) {
+                yield* awaitSteeringOutcome(steer.id);
+                return true;
+              }
+              // A restart, and a steer the provider takes by interrupting and restarting, leave
+              // no steer effect: the message in the thread is the proof.
+              const accepted = yield* orchestrator.getThreadProjection(participant.threadId);
+              if (!accepted.messages.some((message) => message.id === sendInput.messageId)) {
+                return false;
+              }
+              yield* heldRun(
+                accepted.runs.find((run) => run.userMessageId === sendInput.messageId),
+              );
+              return true;
+            });
+            if (yield* landed()) return;
+            const askedRun = requestedSteeringRun(target, input.envelopeChannel, mode);
+            if (askedRun !== undefined) {
+              yield* orchestrator
+                .dispatch({
+                  type: "message.dispatch",
+                  commandId: askedCommandId,
+                  threadId: sendInput.threadId,
+                  messageId: sendInput.messageId,
+                  text: sendInput.text,
+                  attachments: sendInput.attachments,
+                  ...(sendInput.senderThreadId === undefined
+                    ? {}
+                    : { senderThreadId: sendInput.senderThreadId }),
+                  createdBy: sendInput.createdBy,
+                  creationSource: sendInput.creationSource,
+                  modelSelection: askedRun.modelSelection,
+                  dispatchMode: {
+                    type: mode === "restart" ? "restart_active" : "steer_active",
+                    targetRunId: askedRun.id,
+                  },
+                })
+                .pipe(
+                  // The sender's limit when it sent. The receiver's user can raise its modes
+                  // before delivery, and upstream checks again under the thread's lock.
+                  Effect.provideService(DispatchModeLimit, input.modeLimit),
+                  // Refused or too late: the message is delivered as any other, below.
+                  Effect.catch((cause) =>
+                    Effect.logInfo("J5 A2A delivery could not land as its sender asked", {
+                      messageId: input.messageId,
+                      mode,
+                      cause,
+                    }),
+                  ),
+                );
+              if (yield* landed()) return;
+            }
+            return yield* heldRun((yield* threads.sendToThread(sendInput)).run);
+          }
           if (steeringRun === undefined) {
             yield* heldRun((yield* threads.sendToThread(sendInput)).run);
           } else {

apps/server/src/j5/a2a/DeliveryWorker.ts

@@ -347,6 +347,8 @@ const makeLayer = (daemon: boolean) =>
             exchangeRole: row.exchange_role,
             message: row.message_text,
             ...(payload?.attachments === undefined ? {} : { attachments: payload.attachments }),
+            ...(payload?.mode === undefined ? {} : { mode: payload.mode }),
+            ...(payload?.modeLimit === undefined ? {} : { modeLimit: payload.modeLimit }),
             envelopeChannel: row.envelope_channel,
             ...(senderServerName === undefined ? {} : { senderServerName }),
           });

apps/server/src/j5/a2a/EnvelopeFormatter.test.ts

@@ -52,7 +52,7 @@ it("renders the versioned peer envelope with exact reply semantics", () => {
     message: "Please verify the worker.",
   });
 
-  assert.equal(A2A_ENVELOPE_VERSION, 23);
+  assert.equal(A2A_ENVELOPE_VERSION, 24);
   assert.include(rendered, "Cross-agent message");
   assert.notMatch(rendered, /\b(?:J5|A2A)\b/);
   assert.include(rendered, "agent:sender");
@@ -159,7 +159,7 @@ it("renders the receiver backlog notice with its measured counts", () => {
 it("keeps the tool descriptions on their documented contracts", () => {
   assert.equal(
     A2A_SEND_TOOL_DESCRIPTION,
-    "Send one durable message. To another agent, three uses: a **plain send** when you don't need a reply; an **ask** — set expect_reply=true with a one-line intent, opening an exchange the receiver owes a reply to; a **reply** — include the exchange_id from the ask you are answering, which closes that exchange. To the human, only an ask: a plain send to a person is refused — if nobody needs to act, say it in your own thread instead. Set urgency only when asking the human. Use this tool only for participants already returned by j5_list_participants; when creating a Peer Agent, put any reply expectation in j5_spawn_agent's brief instead of sending a follow-up ask. Returns once the message is committed; delivery continues asynchronously — carry on with your work, and the reply arrives later as an incoming message. An agent that is busy usually handles each message as its own turn after its current one ends, so put related updates in one message rather than sending them one by one; the result's deliveryNotice says when your message will wait behind the receiver's current turn. A provider Subagent is not a participant and is refused. Reuse client_request_id to retry the same send safely.",
+    "Send one durable message. To another agent, three uses: a **plain send** when you don't need a reply; an **ask** — set expect_reply=true with a one-line intent, opening an exchange the receiver owes a reply to; a **reply** — include the exchange_id from the ask you are answering, which closes that exchange. To the human, only an ask: a plain send to a person is refused — if nobody needs to act, say it in your own thread instead. Set urgency only when asking the human. Use this tool only for participants already returned by j5_list_participants; when creating a Peer Agent, put any reply expectation in j5_spawn_agent's brief instead of sending a follow-up ask. Returns once the message is committed; delivery continues asynchronously — carry on with your work, and the reply arrives later as an incoming message. An agent that is busy usually handles each message as its own turn after its current one ends, so put related updates in one message rather than sending them one by one; the result's deliveryNotice says when your message will wait behind the receiver's current turn. To interrupt an agent that is busy, set mode: mode=steer puts your message into the turn it is running now, and mode=restart stops that turn and starts it again on your message. Without mode the message is delivered the ordinary way, whatever its text or its intent says: most agents take it after their current turn ends, and Codex GPT 6 agents take it into the turn they are running. intent only labels an ask. Leave mode out for updates and replies. A steer can cancel the tool calls the receiver has in flight, so keep it for a correction the receiver must see now, not for an update. An idle receiver starts a turn on your message whichever mode you set. Attach files for an agent on this server with attachments: pending uploads from t3_attachment_prepare_upload, or files its thread already holds. A provider Subagent is not a participant and is refused. Reuse client_request_id to retry the same send safely.",
   );
   assert.include(
     A2A_SEND_TOOL_DESCRIPTION,
@@ -170,7 +170,7 @@ it("keeps the tool descriptions on their documented contracts", () => {
   assert.include(A2A_CLEAR_OWN_ASK_TOOL_DESCRIPTION, "client_request_id");
   assert.equal(
     A2A_LIST_TOOL_DESCRIPTION,
-    "Your address book: the participants around you — agents and the human — with the display name to recognize them by, the participant_id to address them with, the project_id and project_title that place them, and what each accepts (messages, exchanges, urgency). Once this server is peered with others, each row also carries `server`: the name of the server the participant lives on, and whether it is this one (`local`), so you can choose a participant by the machine it runs on. When you're told to message someone by name or role, resolve them here first. Your own row is marked self=true and its project_title is the project you work in; it cannot receive messages or open exchanges from you — use schedule_task if you need a future trigger for yourself. Provider Subagents are not participants: they do not appear here and cannot be messaged. Archived agents are hidden by default; set include_archived=true to see them with archived=true. They cannot receive messages or open Exchanges. The roster changes — after you spawn, archive, unarchive, or delete an agent, call this again instead of reusing a stale listing.",
+    "Your address book, one project at a time: the project's agents, and the human, with the display name to recognize them by, the participant_id to address them with, the project_id and project_title that place them, and what each accepts (messages, exchanges, urgency). Leave project_id out to list your own project; pass another project's id, from t3_project_list, to list that one. People appear in every listing. Results come a page at a time: your own row first, then the people, then agents with the most recently active first. Narrow a listing with name_contains, or with settled=true or settled=false for settled or active agents. If this server is peered with others, each row also carries `server`: the name of the server the participant lives on, and whether it is this one (`local`), and the result names every server in `servers`; pass a peer server's name as server to list that server's agents instead of this one's. When you're told to message someone by name or role, resolve them here first. Your own row is marked self=true and its project_title is the project you work in; it cannot receive messages or open exchanges from you — use schedule_task if you need a future trigger for yourself. Provider Subagents are not participants: they do not appear here and cannot be messaged. Archived agents are hidden by default; set include_archived=true to see them with archived=true. They cannot receive messages or open Exchanges. The roster changes — after you spawn, archive, unarchive, or delete an agent, call this again instead of reusing a stale listing.",
   );
   for (const clause of [
     "Your own row is marked self=true",

apps/server/src/j5/a2a/SendService.ts

@@ -317,6 +317,36 @@ export type A2ASendError =
   | A2AClearOwnAskUnknownExchangeError
   | A2AClearOwnAskCommandConflictError;
 
+/**
+ * A send the service refused before it committed anything: the sender, the receiver or the
+ * request itself was not acceptable. Storage and decoding failures are not here, because after
+ * one of those nobody knows whether the message was committed. A caller that prepared something
+ * for the message, such as claimed attachments, undoes it only for these.
+ */
+export const isSendRefusal = Schema.is(
+  Schema.Union([
+    A2AMachineCannotReceiveError,
+    A2AMachineSenderNotRegisteredError,
+    A2ASenderNotJoinedError,
+    A2ASenderRetiredError,
+    A2AHomeMembershipStateError,
+    A2AParticipantNotFoundError,
+    A2AAmbiguousParticipantError,
+    A2APeersUnreadError,
+    A2AParticipantArchivedError,
+    A2AMessageTooLongError,
+    A2AIntentRequiredError,
+    A2AUrgencyRequiredError,
+    A2AUrgencyNotAcceptedError,
+    A2AUrgencyRequiresExchangeError,
+    A2AHumanAskOrReplyRequiredError,
+    A2AHumanFollowupNotAllowedError,
+    A2AExchangeNotOpenError,
+    A2AExchangeAlreadyAnsweredError,
+    A2AExchangeParticipantMismatchError,
+  ]),
+);
+
 interface MembershipRow {
   readonly project_id: string;
   readonly participant_id: string;
@@ -983,6 +1013,12 @@ const rawLayer: Layer.Layer<
               messageId,
               text: input.message,
               ...(input.attachments === undefined ? {} : { attachments: input.attachments }),
+              ...(input.mode === undefined || input.mode === "queue"
+                ? {}
+                : {
+                    mode: input.mode,
+                    ...(input.modeLimit === undefined ? {} : { modeLimit: input.modeLimit }),
+                  }),
               originProjectId: sender.projectId,
               receiverProjectId: receiver.projectId,
               ...(receiver.environmentId === null

apps/server/src/j5/a2a/contracts.ts

@@ -1,4 +1,4 @@
-import { ChatAttachment, ThreadId } from "@t3tools/contracts";
+import { ChatAttachment, ProviderInteractionMode, RuntimeMode, ThreadId } from "@t3tools/contracts";
 import { PeerSenderLabel } from "@t3tools/contracts/j5";
 import * as Schema from "effect/Schema";
 
@@ -30,6 +30,25 @@ export type LedgerMessageId = typeof LedgerMessageId.Type;
 export const Urgency = Schema.Literals(["blocking", "soon", "fyi"]);
 export type Urgency = typeof Urgency.Type;
 
+/**
+ * How a sender asks for its message to land on an agent, in upstream's words: `queue` is J5's
+ * ordinary delivery (after the receiver's turn ends, or into it for the models D4 excepts),
+ * `steer` joins the running turn, and `restart` interrupts the turn and restarts it with the
+ * message. Upstream's `auto` is left out: it steers whenever it can.
+ */
+export const SendMode = Schema.Literals(["queue", "steer", "restart"]);
+export type SendMode = typeof SendMode.Type;
+
+/**
+ * The broadest modes a sender could touch when it asked for a steer or a restart. Delivery is
+ * later than the send, and the receiver's modes can be raised in between, so the limit is kept
+ * with the message and upstream checks it again when the message is dispatched.
+ */
+export const SendModeLimit = Schema.Struct({
+  runtimeMode: RuntimeMode,
+  interactionMode: ProviderInteractionMode,
+});
+
 export const DeliveryEnvelopeChannel = Schema.Literals([
   "peer",
   "silence_notice",
@@ -264,6 +283,9 @@ export type TerminalFact = typeof TerminalFact.Type;
 
 export const MessageSentPayload = Schema.Struct({
   attachments: Schema.optional(Schema.Array(ChatAttachment)),
+  /** How the sender asked for the message to land; absent means `queue`. */
+  mode: Schema.optional(SendMode),
+  modeLimit: Schema.optional(SendModeLimit),
   /** Present on a terminal notice headed to a peer server: the fact it carries, fixed when the notice is written. */
   terminal: Schema.optional(TerminalFact),
   messageId: LedgerMessageId,
@@ -344,6 +366,8 @@ export type ClearOwnAskResult = typeof ClearOwnAskResult.Type;
 
 export const SendMessageInput = Schema.Struct({
   attachments: Schema.optional(Schema.Array(ChatAttachment)),
+  mode: Schema.optional(SendMode),
+  modeLimit: Schema.optional(SendModeLimit),
   commandId: CommCommandId,
   senderThreadId: ThreadId,
   to: ParticipantId,