const std = @import("std");

const std_compat = @import("compat");

const builtin = @import("builtin");

const Config = @import("config.zig").Config;

const gateway = @import("gateway.zig");

const ConversationContext = @import("agent/prompt.zig").ConversationContext;

const buildConversationContext = @import("agent/prompt.zig").buildConversationContext;

const streaming = @import("streaming.zig");

const agent_mod = @import("agent/root.zig");

const log = std.log.scoped(.a2a);

const MAX_TASKS: usize = 1000;

const A2A_PROTOCOL_VERSION = "0.3.0";

pub const TaskState = enum {

submitted,

working,

completed,

failed,

canceled,

input_required,

rejected,

auth_required,

unknown,

pub fn jsonName(self: TaskState) []const u8 {

return switch (self) {

.submitted => "submitted",

.working => "working",

.completed => "completed",

.failed => "failed",

.canceled => "canceled",

.input_required => "input-required",

.rejected => "rejected",

.auth_required => "auth-required",

.unknown => "unknown",

};

}

};

pub const TaskRecord = struct {

id: []u8,

context_id: []u8,

session_key: []u8,

state: TaskState,

created_at: i64,

updated_at: i64,

user_text: []u8,

agent_text: []u8,

};

pub const TaskSnapshot = struct {

id: []u8,

context_id: []u8,

session_key: []u8,

state: TaskState,

created_at: i64,

updated_at: i64,

user_text: []u8,

agent_text: []u8,

pub fn deinit(self: *TaskSnapshot, allocator: std.mem.Allocator) void {

allocator.free(self.id);

allocator.free(self.context_id);

allocator.free(self.session_key);

allocator.free(self.user_text);

allocator.free(self.agent_text);

}

};

fn deinitTaskSnapshots(allocator: std.mem.Allocator, tasks: []TaskSnapshot) void {

for (tasks) |*task| task.deinit(allocator);

allocator.free(tasks);

}

fn isTerminalState(state: TaskState) bool {

return switch (state) {

.completed, .failed, .canceled, .rejected => true,

else => false,

};

}

fn taskOrdinal(task_id: []const u8) u64 {

const prefix = "task-";

if (!std.mem.startsWith(u8, task_id, prefix)) return 0;

return std.fmt.parseInt(u64, task_id[prefix.len..], 10) catch 0;

}

fn isTaskSnapshotMoreRecent(current: TaskSnapshot, previous: TaskSnapshot) bool {

if (current.updated_at != previous.updated_at) {

return current.updated_at > previous.updated_at;

}

return taskOrdinal(current.id) > taskOrdinal(previous.id);

}

fn sortTaskSnapshotsByRecency(tasks: []TaskSnapshot) void {

var i: usize = 1;

while (i < tasks.len) : (i += 1) {

const current = tasks[i];

var j = i;

while (j > 0) : (j -= 1) {

const prev = tasks[j - 1];

const more_recent = isTaskSnapshotMoreRecent(current, prev);

if (!more_recent) break;

tasks[j] = prev;

}

tasks[j] = current;

}

}

pub const TaskRegistry = struct {

allocator: std.mem.Allocator,

mutex: std_compat.sync.Mutex = .{},

tasks: std.StringHashMapUnmanaged(*TaskRecord) = .empty,

next_id: u64 = 1,

pub fn init(allocator: std.mem.Allocator) TaskRegistry {

return .{ .allocator = allocator };

}

pub fn deinit(self: *TaskRegistry) void {

self.mutex.lock();

defer self.mutex.unlock();

var iter = self.tasks.iterator();

while (iter.next()) |entry| {

self.freeTask(entry.value_ptr.*);

}

self.tasks.deinit(self.allocator);

}

pub fn createTask(self: *TaskRegistry, allocator: std.mem.Allocator, user_text: []const u8, context_id: ?[]const u8) !TaskSnapshot {

self.mutex.lock();

defer self.mutex.unlock();

// Evict the least-recently-updated terminal task. If all tasks are active,

// reject the new task to keep the registry strictly capped.

if (self.tasks.count() >= MAX_TASKS) {

if (!self.evictOldestTerminalLocked()) {

return error.TaskRegistryFull;

}

}

const id_num = self.next_id;

self.next_id += 1;

const task_id = try std.fmt.allocPrint(self.allocator, "task-{d}", .{id_num});

errdefer self.allocator.free(task_id);

const owned_context_id = if (context_id) |value|

try self.allocator.dupe(u8, value)

else

try std.fmt.allocPrint(self.allocator, "ctx-{d}", .{id_num});

errdefer self.allocator.free(owned_context_id);

const session_key = try std.fmt.allocPrint(self.allocator, "a2a:{s}", .{owned_context_id});

errdefer self.allocator.free(session_key);

const owned_text = try self.allocator.dupe(u8, user_text);

errdefer self.allocator.free(owned_text);

const empty_agent = try self.allocator.dupe(u8, "");

errdefer self.allocator.free(empty_agent);

const now = std_compat.time.timestamp();

const task = try self.allocator.create(TaskRecord);

errdefer self.allocator.destroy(task);

task.* = .{

.id = task_id,

.context_id = owned_context_id,

.session_key = session_key,

.state = .submitted,

.created_at = now,

.updated_at = now,

.user_text = owned_text,

.agent_text = empty_agent,

};

try self.tasks.put(self.allocator, task_id, task);

errdefer {

if (self.tasks.fetchRemove(task_id)) |removed| {

self.freeTask(removed.value);

}

}

return self.snapshotLocked(allocator, task);

}

pub fn getTaskSnapshot(self: *TaskRegistry, allocator: std.mem.Allocator, task_id: []const u8) !?TaskSnapshot {

self.mutex.lock();

defer self.mutex.unlock();

const task = self.tasks.get(task_id) orelse return null;

return try self.snapshotLocked(allocator, task);

}

pub fn taskCount(self: *TaskRegistry) usize {

self.mutex.lock();

defer self.mutex.unlock();

return self.tasks.count();

}

pub fn setTaskState(self: *TaskRegistry, task_id: []const u8, new_state: TaskState) bool {

self.mutex.lock();

defer self.mutex.unlock();

const task = self.tasks.get(task_id) orelse return false;

if (isTerminalState(task.state) and task.state != new_state) return false;

task.state = new_state;

task.updated_at = std_compat.time.timestamp();

return true;

}

pub fn finalizeTask(

self: *TaskRegistry,

allocator: std.mem.Allocator,

task_id: []const u8,

final_state: TaskState,

agent_text: ?[]const u8,

) !?TaskSnapshot {

self.mutex.lock();

defer self.mutex.unlock();

const task = self.tasks.get(task_id) orelse return null;

if (task.state == .canceled and final_state != .canceled) {

return try self.snapshotLocked(allocator, task);

}

if (agent_text) |text| {

const new_agent_text = try self.allocator.dupe(u8, text);

errdefer self.allocator.free(new_agent_text);

self.allocator.free(task.agent_text);

task.agent_text = new_agent_text;

}

task.state = final_state;

task.updated_at = std_compat.time.timestamp();

return try self.snapshotLocked(allocator, task);

}

pub fn cancelTask(self: *TaskRegistry, allocator: std.mem.Allocator, task_id: []const u8) !?TaskSnapshot {

self.mutex.lock();

defer self.mutex.unlock();

const task = self.tasks.get(task_id) orelse return null;

if (!isTerminalState(task.state)) {

task.state = .canceled;

task.updated_at = std_compat.time.timestamp();

}

return try self.snapshotLocked(allocator, task);

}

/// List tasks with optional filtering. Returns owned task snapshots sorted by recency.

/// Caller must free the returned slice with `deinitTaskSnapshots`.

pub fn listTasks(

self: *TaskRegistry,

allocator: std.mem.Allocator,

filter_state: ?TaskState,

filter_context_id: ?[]const u8,

max_results: usize,

) ![]TaskSnapshot {

self.mutex.lock();

defer self.mutex.unlock();

var result: std.ArrayListUnmanaged(TaskSnapshot) = .empty;

errdefer {

for (result.items) |*task| task.deinit(allocator);

result.deinit(allocator);

}

var iter = self.tasks.iterator();

while (iter.next()) |entry| {

const task = entry.value_ptr.*;

if (filter_state) |s| {

if (task.state != s) continue;

}

if (filter_context_id) |ctx| {

if (!std.mem.eql(u8, task.context_id, ctx)) continue;

}

try result.append(allocator, try self.snapshotLocked(allocator, task));

}

sortTaskSnapshotsByRecency(result.items);

if (result.items.len > max_results) {

var i = max_results;

while (i < result.items.len) : (i += 1) {

result.items[i].deinit(allocator);

}

result.items.len = max_results;

}

return result.toOwnedSlice(allocator);

}

/// Evict the least-recently-updated terminal task. Must be called with mutex held.

fn evictOldestTerminalLocked(self: *TaskRegistry) bool {

var oldest_key: ?[]const u8 = null;

var oldest_time: i64 = std.math.maxInt(i64);

var oldest_ordinal: u64 = std.math.maxInt(u64);

var iter = self.tasks.iterator();

while (iter.next()) |entry| {

const task = entry.value_ptr.*;

const ordinal = taskOrdinal(task.id);

if (isTerminalState(task.state) and

(task.updated_at < oldest_time or

(task.updated_at == oldest_time and ordinal < oldest_ordinal)))

{

oldest_time = task.updated_at;

oldest_ordinal = ordinal;

oldest_key = entry.key_ptr.*;

}

}

if (oldest_key) |key| {

if (self.tasks.fetchRemove(key)) |kv| {

self.freeTask(kv.value);

return true;

}

}

return false;

}

fn freeTask(self: *TaskRegistry, task: *TaskRecord) void {

self.allocator.free(task.id);

self.allocator.free(task.context_id);

self.allocator.free(task.session_key);

self.allocator.free(task.user_text);

self.allocator.free(task.agent_text);

self.allocator.destroy(task);

}

fn snapshotLocked(self: *TaskRegistry, allocator: std.mem.Allocator, task: *const TaskRecord) !TaskSnapshot {

_ = self;

const id = try allocator.dupe(u8, task.id);

errdefer allocator.free(id);

const context_id = try allocator.dupe(u8, task.context_id);

errdefer allocator.free(context_id);

const session_key = try allocator.dupe(u8, task.session_key);

errdefer allocator.free(session_key);

const user_text = try allocator.dupe(u8, task.user_text);

errdefer allocator.free(user_text);

const agent_text = try allocator.dupe(u8, task.agent_text);

errdefer allocator.free(agent_text);

return .{

.id = id,

.context_id = context_id,

.session_key = session_key,

.state = task.state,

.created_at = task.created_at,

.updated_at = task.updated_at,

.user_text = user_text,

.agent_text = agent_text,

};

}

};

pub const A2aResponse = struct {

status: []const u8 = "200 OK",

body: []const u8,

content_type: []const u8 = "application/json",

allocated: bool = true,

};

pub fn handleAgentCard(allocator: std.mem.Allocator, cfg: *const Config, vision_capable: ?bool) A2aResponse {

var buf: std.ArrayListUnmanaged(u8) = .empty;

errdefer buf.deinit(allocator);

const endpoint_url = buildEndpointUrl(allocator, cfg.a2a.url) catch return errorResponse();

defer allocator.free(endpoint_url);

var buf_writer: std.Io.Writer.Allocating = .fromArrayList(allocator, &buf);

const w = &buf_writer.writer;

w.writeAll("{\"name\":\"") catch return errorResponse();

gateway.jsonEscapeInto(w, cfg.a2a.name) catch return errorResponse();

w.writeAll("\",\"description\":\"") catch return errorResponse();

gateway.jsonEscapeInto(w, cfg.a2a.description) catch return errorResponse();

w.writeAll("\",\"protocolVersion\":\"") catch return errorResponse();

w.writeAll(A2A_PROTOCOL_VERSION) catch return errorResponse();

w.writeAll("\",\"version\":\"") catch return errorResponse();

gateway.jsonEscapeInto(w, cfg.a2a.version) catch return errorResponse();

// url field for backward compatibility with older A2A clients.

w.writeAll("\",\"url\":\"") catch return errorResponse();

gateway.jsonEscapeInto(w, endpoint_url) catch return errorResponse();

// supported_interfaces per latest spec (required).

w.writeAll("\",\"supportedInterfaces\":[{\"url\":\"") catch return errorResponse();

gateway.jsonEscapeInto(w, endpoint_url) catch return errorResponse();

w.writeAll("\",\"protocolBinding\":\"JSONRPC\",\"protocolVersion\":\"") catch return errorResponse();

w.writeAll(A2A_PROTOCOL_VERSION) catch return errorResponse();

w.writeAll("\"}],\"preferredTransport\":\"JSONRPC\"") catch return errorResponse();

w.writeAll(",\"provider\":{\"organization\":\"") catch return errorResponse();

gateway.jsonEscapeInto(w, cfg.a2a.name) catch return errorResponse();

w.writeAll("\",\"url\":\"") catch return errorResponse();

if (cfg.a2a.url.len > 0) {

gateway.jsonEscapeInto(w, cfg.a2a.url) catch return errorResponse();

} else {

w.writeAll("https://github.com/nullclaw/nullclaw") catch return errorResponse();

}

w.writeAll("\"}") catch return errorResponse();

const effective_multi_modal = cfg.a2a.multi_modal or vision_capable == true;

if (effective_multi_modal) {

w.writeAll(",\"capabilities\":{\"streaming\":true,\"multi_modal\":true}") catch return errorResponse();

} else {

w.writeAll(",\"capabilities\":{\"streaming\":true}") catch return errorResponse();

}

w.writeAll(",\"securitySchemes\":{\"bearerAuth\":{\"type\":\"http\",\"scheme\":\"bearer\",\"description\":\"Use a pairing token from /pair as the bearer token.\"}}") catch return errorResponse();

w.writeAll(",\"security\":[{\"bearerAuth\":[]}]") catch return errorResponse();

w.writeAll(",\"defaultInputModes\":[\"text/plain\"],\"defaultOutputModes\":[\"text/plain\"]") catch return errorResponse();

w.writeAll(",\"skills\":[{\"id\":\"chat\",\"name\":\"General Chat\",\"description\":\"General-purpose AI assistant\",\"tags\":[\"chat\",\"general\"]}]") catch return errorResponse();

w.writeAll("}") catch return errorResponse();

buf = buf_writer.toArrayList();

const body = buf.toOwnedSlice(allocator) catch return errorResponse();

return .{ .body = body };

}

pub fn handleJsonRpc(

allocator: std.mem.Allocator,

body: []const u8,

registry: *TaskRegistry,

session_mgr: anytype,

) A2aResponse {

const method = extractJsonRpcMethod(body) orelse {

const err_body = buildJsonRpcError(allocator, "null", -32600, "Missing method") catch

return errorResponse();

return .{ .body = err_body };

};

// Extract JSON-RPC id — may be a string or number.

const request_id = extractJsonRpcId(body) orelse "null";

if (std.mem.eql(u8, method, "message/send") or

std.mem.eql(u8, method, "message/stream") or

std.mem.eql(u8, method, "SendMessage") or

std.mem.eql(u8, method, "SendStreamingMessage"))

{

return handleSendMessage(allocator, body, request_id, registry, session_mgr);

} else if (std.mem.eql(u8, method, "tasks/get") or std.mem.eql(u8, method, "GetTask")) {

return handleGetTask(allocator, body, request_id, registry);

} else if (std.mem.eql(u8, method, "tasks/cancel") or std.mem.eql(u8, method, "CancelTask")) {

return handleCancelTask(allocator, body, request_id, registry, session_mgr);

} else if (std.mem.eql(u8, method, "tasks/list") or std.mem.eql(u8, method, "ListTasks")) {

return handleListTasks(allocator, body, request_id, registry);

} else if (std.mem.startsWith(u8, method, "tasks/pushNotificationConfig/") or

std.mem.eql(u8, method, "CreateTaskPushNotificationConfig") or

std.mem.eql(u8, method, "GetTaskPushNotificationConfig") or

std.mem.eql(u8, method, "ListTaskPushNotificationConfigs") or

std.mem.eql(u8, method, "DeleteTaskPushNotificationConfig"))

{

const err_body = buildJsonRpcError(allocator, request_id, -32003, "Push notifications not supported") catch

return errorResponse();

return .{ .body = err_body };

} else if (std.mem.eql(u8, method, "agent/getAuthenticatedExtendedCard") or std.mem.eql(u8, method, "GetExtendedAgentCard")) {

const err_body = buildJsonRpcError(allocator, request_id, -32004, "Extended agent card not supported") catch

return errorResponse();

return .{ .body = err_body };

} else {

const err_body = buildJsonRpcError(allocator, request_id, -32601, "Method not found") catch

return errorResponse();

return .{ .body = err_body };

}

}

pub fn isStreamingMethod(body: []const u8) bool {

const method = extractJsonRpcMethod(body) orelse return false;

return std.mem.eql(u8, method, "message/stream") or

std.mem.eql(u8, method, "tasks/resubscribe") or

std.mem.eql(u8, method, "SendStreamingMessage") or

std.mem.eql(u8, method, "SubscribeToTask");

}

pub const SseStreamCtx = struct {

stream: *std_compat.net.Stream,

allocator: std.mem.Allocator,

request_id: []const u8,

task_id: []const u8,

context_id: []const u8,

filter: streaming.TagFilter = undefined,

/// Write an SSE "data:" line with the given JSON payload.

fn writeSseEvent(self: *SseStreamCtx, json_data: []const u8) void {

self.stream.writeAll("data: ") catch return;

self.stream.writeAll(json_data) catch return;

self.stream.writeAll("\n\n") catch return;

}

/// Build and emit an SSE event with a working status and text delta.

fn emitChunkEvent(self: *SseStreamCtx, text: []const u8) void {

const data = buildArtifactUpdateEvent(self.allocator, self.request_id, self.task_id, self.context_id, text, false) catch return;

defer self.allocator.free(data);

self.writeSseEvent(data);

}

fn sseCallback(ctx: *anyopaque, event: streaming.Event) void {

const self: *SseStreamCtx = @ptrCast(@alignCast(ctx));

switch (event.stage) {

.chunk => {

if (event.text.len > 0) self.emitChunkEvent(event.text);

},

.final => {}, // Final event is handled after processMessageStreaming returns.

}

}

pub fn makeSink(self: *SseStreamCtx) streaming.Sink {

const raw_sink = streaming.Sink{

.callback = sseCallback,

.ctx = @ptrCast(self),

};

self.filter = streaming.TagFilter.init(raw_sink);

return self.filter.sink();

}

};

const StatusMessage = struct {

prefix: []const u8 = "",

text: []const u8,

};

const A2aProgressCtx = struct {

sse_ctx: *SseStreamCtx,

fn progressCallback(ctx: *anyopaque, hint: agent_mod.ProgressHint) void {

const self: *A2aProgressCtx = @ptrCast(@alignCast(ctx));

const data = buildProgressHintEvent(

self.sse_ctx.allocator,

self.sse_ctx.request_id,

self.sse_ctx.task_id,

self.sse_ctx.context_id,

std_compat.time.timestamp(),

hint.text,

) catch return;

defer self.sse_ctx.allocator.free(data);

self.sse_ctx.writeSseEvent(data);

}

pub fn makeSink(self: *A2aProgressCtx) agent_mod.ProgressSink {

return .{ .callback = progressCallback, .ctx = @ptrCast(self) };

}

};

fn buildProgressHintEvent(

allocator: std.mem.Allocator,

request_id: []const u8,

task_id: []const u8,

context_id: []const u8,

updated_at: i64,

tool_name: []const u8,

) ![]u8 {

return buildStatusUpdateEvent(

allocator,

request_id,

task_id,

context_id,

.working,

updated_at,

.{ .prefix = "Calling tool: ", .text = tool_name },

false,

);

}

pub fn handleStreamingRpc(

allocator: std.mem.Allocator,

body: []const u8,

stream: *std_compat.net.Stream,

registry: *TaskRegistry,

session_mgr: anytype,

) void {

const request_id = extractJsonRpcId(body) orelse "null";

// Handle tasks/resubscribe: resume SSE for an existing task.

const method = extractJsonRpcMethod(body) orelse "message/stream";

if (std.mem.eql(u8, method, "tasks/resubscribe") or std.mem.eql(u8, method, "SubscribeToTask")) {

handleResubscribeStreaming(allocator, body, stream, request_id, registry);

return;

}

const text = (extractMessageContent(allocator, body) catch null) orelse {

writeSseError(allocator, stream, request_id, -32602, "Missing message text");

return;

};

defer allocator.free(text);

// v0.3.0: messageId is required on Message.

if (extractMessageMessageId(body) == null) {

writeSseError(allocator, stream, request_id, -32602, "Missing messageId");

return;

}

const context_id = extractMessageContextId(body);

var task = registry.createTask(allocator, text, context_id) catch {

writeSseError(allocator, stream, request_id, -32603, "Failed to create task");

return;

};

defer task.deinit(allocator);

// Write SSE headers.

stream.writeAll("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: keep-alive\r\n\r\n") catch return;

const initial_task_json = buildTaskJson(allocator, &task, null) catch return;

defer allocator.free(initial_task_json);

const initial_event = buildJsonRpcResult(allocator, request_id, initial_task_json) catch return;

defer allocator.free(initial_event);

stream.writeAll("data: ") catch return;

stream.writeAll(initial_event) catch return;

stream.writeAll("\n\n") catch return;

// Mark as working after emitting the initial submitted task.

if (!registry.setTaskState(task.id, .working)) {

var current_task = registry.getTaskSnapshot(allocator, task.id) catch return;

defer if (current_task) |*snapshot| snapshot.deinit(allocator);

const current_snapshot = current_task orelse return;

if (current_snapshot.state == .canceled) {

const final_event = buildStatusUpdateEvent(allocator, request_id, current_snapshot.id, current_snapshot.context_id, current_snapshot.state, current_snapshot.updated_at, null, true) catch return;

defer allocator.free(final_event);

stream.writeAll("data: ") catch return;

stream.writeAll(final_event) catch return;

stream.writeAll("\n\n") catch return;

}

return;

}

// Create SSE sink context.

var sse_ctx = SseStreamCtx{

.stream = stream,

.allocator = allocator,

.request_id = request_id,

.task_id = task.id,

.context_id = task.context_id,

};

const sink = sse_ctx.makeSink();

var progress_ctx = A2aProgressCtx{ .sse_ctx = &sse_ctx };

const progress_sink = progress_ctx.makeSink();

const context: ConversationContext = buildConversationContext(.{ .channel = "a2a" }).?;

const response = session_mgr.processInboundMessageStreaming(task.session_key, text, context, sink, progress_sink) catch |err| {

log.err("streaming turn failed task={s} err={s}", .{ task.id, @errorName(err) });

var failed_task = registry.finalizeTask(allocator, task.id, .failed, null) catch null;

defer if (failed_task) |*snapshot| snapshot.deinit(allocator);

if (failed_task) |snapshot| {

if (snapshot.state == .canceled) {

const final_event = buildStatusUpdateEvent(allocator, request_id, snapshot.id, snapshot.context_id, snapshot.state, snapshot.updated_at, null, true) catch return;

defer allocator.free(final_event);

sse_ctx.writeSseEvent(final_event);

} else {

writeSseErrorEvent(allocator, &sse_ctx, -32603, "Agent processing failed");

}

} else {

writeSseErrorEvent(allocator, &sse_ctx, -32603, "Agent processing failed");

}

return;

};

defer if (response) |r| freeSessionResponse(session_mgr, r);

var final_task = registry.finalizeTask(allocator, task.id, .completed, response) catch return;

defer if (final_task) |*snapshot| snapshot.deinit(allocator);

const final_snapshot = final_task orelse return;

const final_event = buildStatusUpdateEvent(allocator, request_id, final_snapshot.id, final_snapshot.context_id, final_snapshot.state, final_snapshot.updated_at, null, true) catch return;

defer allocator.free(final_event);

sse_ctx.writeSseEvent(final_event);

}

fn handleResubscribeStreaming(

allocator: std.mem.Allocator,

body: []const u8,

stream: *std_compat.net.Stream,

request_id: []const u8,

registry: *TaskRegistry,

) void {

const task_id = extractParamsId(body) orelse {

writeSseError(allocator, stream, request_id, -32602, "Missing task id");

return;

};

var task = registry.getTaskSnapshot(allocator, task_id) catch {

writeSseError(allocator, stream, request_id, -32603, "Internal error");

return;

};

defer if (task) |*snapshot| snapshot.deinit(allocator);

const snapshot = task orelse {

writeSseError(allocator, stream, request_id, -32001, "Task not found");

return;

};

// Write SSE headers.

stream.writeAll("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: keep-alive\r\n\r\n") catch return;

// buildResubscribeStatusEvent already returns a JSON-RPC envelope; write it directly.

const status_event = buildResubscribeStatusEvent(allocator, request_id, snapshot.id, snapshot.context_id, snapshot.state, snapshot.updated_at) catch return;

defer allocator.free(status_event);

stream.writeAll("data: ") catch return;

stream.writeAll(status_event) catch return;

stream.writeAll("\n\n") catch return;

}

fn writeSseErrorEvent(allocator: std.mem.Allocator, sse_ctx: *SseStreamCtx, code: i32, message: []const u8) void {

const err_json = buildJsonRpcError(allocator, sse_ctx.request_id, code, message) catch return;

defer allocator.free(err_json);

sse_ctx.writeSseEvent(err_json);

}

fn writeSseError(allocator: std.mem.Allocator, stream: *std_compat.net.Stream, request_id: []const u8, code: i32, message: []const u8) void {

stream.writeAll("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\nConnection: keep-alive\r\n\r\n") catch return;

const err_json = buildJsonRpcError(allocator, request_id, code, message) catch return;

defer allocator.free(err_json);

stream.writeAll("data: ") catch return;

stream.writeAll(err_json) catch return;

stream.writeAll("\n\n") catch return;

}

fn handleSendMessage(

allocator: std.mem.Allocator,

body: []const u8,

request_id: []const u8,

registry: *TaskRegistry,

session_mgr: anytype,

) A2aResponse {

const text = (extractMessageContent(allocator, body) catch null) orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "Missing message text") catch

return errorResponse();

return .{ .body = err_body };

};

defer allocator.free(text);

// v0.3.0: messageId is required on Message.

if (extractMessageMessageId(body) == null) {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "Missing messageId") catch

return errorResponse();

return .{ .body = err_body };

}

const context_id = extractMessageContextId(body);

// Parse optional configuration.

const config = parseSendMessageConfiguration(body);

if (config.history_length) |history_length| {

if (history_length < 0) {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "historyLength must be non-negative") catch

return errorResponse();

return .{ .body = err_body };

}

}

// Check acceptedOutputModes: if specified, must include text/plain.

if (config.has_accepted_output_modes and !config.accepts_text_plain) {

const err_body = buildJsonRpcError(allocator, request_id, -32005, "Incompatible content types") catch

return errorResponse();

return .{ .body = err_body };

}

var task = registry.createTask(allocator, text, context_id) catch {

const err_body = buildJsonRpcError(allocator, request_id, -32603, "Failed to create task") catch

return errorResponse();

return .{ .body = err_body };

};

defer task.deinit(allocator);

// Update state to working.

_ = registry.setTaskState(task.id, .working);

const context: ConversationContext = buildConversationContext(.{ .channel = "a2a" }).?;

const response = session_mgr.processInboundMessage(task.session_key, text, context) catch |err| {

log.err("turn failed task={s} err={s}", .{ task.id, @errorName(err) });

_ = registry.finalizeTask(allocator, task.id, .failed, null) catch null;

const err_body = buildJsonRpcError(allocator, request_id, -32603, "Agent processing failed") catch

return errorResponse();

return .{ .body = err_body };

};

defer if (response) |r| freeSessionResponse(session_mgr, r);

var completed_task = registry.finalizeTask(allocator, task.id, .completed, response) catch {

const err_body = buildJsonRpcError(allocator, request_id, -32603, "Out of memory") catch

return errorResponse();

return .{ .body = err_body };

};

defer if (completed_task) |*snapshot| snapshot.deinit(allocator);

const task_snapshot = completed_task orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32603, "Failed to build response") catch

return errorResponse();

return .{ .body = err_body };

};

const task_json = buildTaskJson(allocator, &task_snapshot, config.history_length) catch {

const err_body = buildJsonRpcError(allocator, request_id, -32603, "Failed to build response") catch

return errorResponse();

return .{ .body = err_body };

};

defer allocator.free(task_json);

const result = buildJsonRpcResult(allocator, request_id, task_json) catch

return errorResponse();

return .{ .body = result };

}

fn handleGetTask(

allocator: std.mem.Allocator,

body: []const u8,

request_id: []const u8,

registry: *TaskRegistry,

) A2aResponse {

const task_id = extractParamsId(body) orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "Missing task id") catch

return errorResponse();

return .{ .body = err_body };

};

// Parse optional historyLength from params.

const params_section = extractParamsObject(body) orelse "{}";

const history_length = extractObjectIntField(params_section, "historyLength");

if (history_length) |value| {

if (value < 0) {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "historyLength must be non-negative") catch

return errorResponse();

return .{ .body = err_body };

}

}

var task = registry.getTaskSnapshot(allocator, task_id) catch return errorResponse();

defer if (task) |*snapshot| snapshot.deinit(allocator);

const task_snapshot = task orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32001, "Task not found") catch

return errorResponse();

return .{ .body = err_body };

};

const task_json = buildTaskJson(allocator, &task_snapshot, history_length) catch {

const err_body = buildJsonRpcError(allocator, request_id, -32603, "Failed to build response") catch

return errorResponse();

return .{ .body = err_body };

};

defer allocator.free(task_json);

const result = buildJsonRpcResult(allocator, request_id, task_json) catch

return errorResponse();

return .{ .body = result };

}

fn handleCancelTask(

allocator: std.mem.Allocator,

body: []const u8,

request_id: []const u8,

registry: *TaskRegistry,

session_mgr: anytype,

) A2aResponse {

const task_id = extractParamsId(body) orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "Missing task id") catch

return errorResponse();

return .{ .body = err_body };

};

var current_task = registry.getTaskSnapshot(allocator, task_id) catch return errorResponse();

defer if (current_task) |*snapshot| snapshot.deinit(allocator);

const task = current_task orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32001, "Task not found") catch

return errorResponse();

return .{ .body = err_body };

};

if (isTerminalState(task.state)) {

const err_body = buildJsonRpcError(allocator, request_id, -32002, "Task already in terminal state") catch

return errorResponse();

return .{ .body = err_body };

}

// Request interruption if working.

if (task.state == .working) {

var result = session_mgr.requestTurnInterrupt(task.session_key);

freeInterruptRequestResult(session_mgr, &result);

}

var canceled_task = registry.cancelTask(allocator, task_id) catch return errorResponse();

defer if (canceled_task) |*snapshot| snapshot.deinit(allocator);

const task_snapshot = canceled_task orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32001, "Task not found") catch

return errorResponse();

return .{ .body = err_body };

};

if (task_snapshot.state != .canceled) {

const err_body = buildJsonRpcError(allocator, request_id, -32002, "Task already in terminal state") catch

return errorResponse();

return .{ .body = err_body };

}

const task_json = buildTaskJson(allocator, &task_snapshot, null) catch {

const err_body = buildJsonRpcError(allocator, request_id, -32603, "Failed to build response") catch

return errorResponse();

return .{ .body = err_body };

};

defer allocator.free(task_json);

const result = buildJsonRpcResult(allocator, request_id, task_json) catch

return errorResponse();

return .{ .body = result };

}

fn handleListTasks(

allocator: std.mem.Allocator,

body: []const u8,

request_id: []const u8,

registry: *TaskRegistry,

) A2aResponse {

// Parse optional filters from params.

const params_section = extractParamsObject(body) orelse "{}";

const filter_state: ?TaskState = blk: {

const has_status_filter = extractObjectFieldRaw(params_section, "status") != null or

extractObjectFieldRaw(params_section, "state") != null;

if (!has_status_filter) break :blk null;

const state_str = extractObjectStringField(params_section, "status") orelse

extractObjectStringField(params_section, "state") orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "Invalid task status") catch

return errorResponse();

return .{ .body = err_body };

};

const parsed_state = parseTaskState(state_str) orelse {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "Invalid task status") catch

return errorResponse();

return .{ .body = err_body };

};

break :blk parsed_state;

};

// Optional context_id filter.

const filter_context_id = extractObjectStringField(params_section, "contextId");

const history_length = extractObjectIntField(params_section, "historyLength");

if (history_length) |value| {

if (value < 0) {

const err_body = buildJsonRpcError(allocator, request_id, -32602, "historyLength must be non-negative") catch

return errorResponse();

return .{ .body = err_body };

}

}

// Page size (default 50, max 100).

const page_size: usize = blk: {

const val = extractObjectIntField(params_section, "pageSize") orelse break :blk 50;

if (val < 1) break :blk 1;