diff --git a/AgentChat.cs b/AgentChat.cs index 02099ce..62cf3c6 100644 --- a/AgentChat.cs +++ b/AgentChat.cs @@ -8,6 +8,8 @@ sealed class AgentChatEntry public string? Status { get; set; } public string? ToolKind { get; set; } public string? ToolCallId { get; set; } + public string? Source { get; set; } + public string? Kind { get; set; } public object Snapshot() => new { @@ -18,6 +20,8 @@ sealed class AgentChatEntry title = Title, status = Status, toolKind = ToolKind, - toolCallId = ToolCallId + toolCallId = ToolCallId, + source = Source, + kind = Kind }; } diff --git a/Dashboard.html b/Dashboard.html index 7f0ac04..0c4ab36 100644 --- a/Dashboard.html +++ b/Dashboard.html @@ -14,7 +14,14 @@ .chat{flex:1;min-height:0;overflow:auto;padding:15px 15px 20px;scrollbar-color:#35445d transparent;scrollbar-width:thin}.chat-empty{display:grid;place-items:center;height:100%;color:var(--subtle);text-align:center}.chat-empty .orb{display:grid;place-items:center;width:42px;height:42px;margin:0 auto 10px;border:1px solid #315078;border-radius:13px;background:#16243a;color:#8ab9f4}.message{margin:0 0 14px}.message-label{margin:0 0 5px;color:var(--muted);font-size:10px;font-weight:750;text-transform:uppercase;letter-spacing:.06em}.message-body{max-width:92%;padding:10px 12px;border:1px solid var(--line);border-radius:10px;background:#151e2d;color:#dce5f3;line-height:1.48;white-space:pre-wrap;overflow-wrap:anywhere}.message.user{display:flex;flex-direction:column;align-items:flex-end}.message.user .message-label{text-align:right}.message.user .message-body{border-color:#315989;background:#173254;color:#eef6ff}.message.assistant .message-body{border-color:#29384e;background:#131b28}.message.error .message-body{border-color:#6a3540;background:#321b24;color:#ffc6cb}.system-message{margin:11px auto;padding:5px 9px;width:max-content;max-width:90%;border:1px solid var(--line);border-radius:999px;background:#121a27;color:var(--muted);font-size:10px;text-align:center}.thought{margin:0 0 12px;border-left:2px solid #556987;padding-left:10px;color:var(--muted)}.thought summary{cursor:pointer;font-size:11px;font-weight:700;user-select:none}.thought-text{margin-top:7px;max-height:180px;overflow:auto;color:#8fa0b8;font-size:11px;line-height:1.45;white-space:pre-wrap}.tool-call{margin:0 0 9px;padding:8px 10px;border:1px solid var(--line);border-radius:9px;background:#101824}.tool-main{display:flex;align-items:center;gap:8px}.tool-icon{display:grid;place-items:center;width:22px;height:22px;border-radius:6px;background:#1d2a3e;color:#9bbbe8}.tool-title{min-width:0;flex:1;overflow:hidden;text-overflow:ellipsis;white-space:nowrap;color:#c8d5e7;font:11px ui-monospace,SFMono-Regular,Consolas,monospace}.tool-status{color:var(--muted);font-size:10px}.tool-status.in_progress{color:var(--yellow)}.tool-status.completed{color:var(--green)}.tool-status.failed{color:var(--red)}.tool-call details{margin-top:7px}.tool-call summary{color:var(--subtle);cursor:pointer;font-size:10px}.tool-detail{max-height:180px;margin:7px 0 0;padding:8px;overflow:auto;border-radius:6px;background:#0a101a;color:#8fa4c2;font:10px/1.45 ui-monospace,SFMono-Regular,Consolas,monospace;white-space:pre-wrap} .composer{padding:11px;border-top:1px solid var(--line);background:#111925}.composer-box{display:flex;align-items:flex-end;gap:8px;padding:8px 8px 8px 11px;border:1px solid var(--line2);border-radius:11px;background:#0c121d;transition:.15s}.composer-box:focus-within{border-color:#4d709e;box-shadow:0 0 0 3px #416caa18}.composer textarea{min-height:24px;max-height:110px;flex:1;resize:none;border:0;outline:0;background:transparent;color:var(--text);line-height:1.45}.composer textarea::placeholder{color:#65748a}.send-btn{display:grid;place-items:center;width:30px;height:30px;flex:none;border:0;border-radius:8px;background:#2c66a4;color:white;cursor:pointer}.send-btn:hover{background:#3979bd}.send-btn:disabled{cursor:not-allowed;opacity:.4}.composer-foot{display:flex;align-items:center;gap:6px;margin-top:7px;color:var(--subtle);font-size:9px}.scope-chip{padding:2px 5px;border-radius:4px;background:#192438;color:#8fa4c1;font:9px ui-monospace,SFMono-Regular,Consolas,monospace}.composer-foot .spacer{flex:1} dialog{width:min(520px,calc(100vw - 30px));padding:0;border:1px solid var(--line2);border-radius:15px;background:#111826;color:var(--text);box-shadow:0 35px 100px #000b}dialog::backdrop{background:#050812b8;backdrop-filter:blur(5px)}.dialog-head{display:flex;align-items:center;padding:17px 18px;border-bottom:1px solid var(--line)}.dialog-head h2{margin:0;font-size:16px}.dialog-body{display:grid;gap:13px;padding:18px}.field label{display:block;margin-bottom:6px;color:var(--muted);font-size:11px;font-weight:700}.field input,.field textarea{width:100%;padding:9px 10px;border:1px solid var(--line);border-radius:8px;outline:none;background:#0b111c;color:var(--text)}.field textarea{min-height:76px;resize:vertical}.field input:focus,.field textarea:focus{border-color:#5278a8}.field-row{display:grid;grid-template-columns:1fr 1fr;gap:12px}.dialog-foot{display:flex;justify-content:flex-end;gap:8px;padding:13px 18px;border-top:1px solid var(--line)}.toast-stack{position:fixed;right:18px;bottom:18px;z-index:50;display:grid;gap:8px}.toast{max-width:380px;padding:10px 12px;border:1px solid var(--line2);border-radius:9px;background:#172131;box-shadow:0 12px 35px #0008;color:#dce5f2;font-size:12px}.toast.error{border-color:#7b3b48;color:#ffc6cc} - @media(max-width:800px){.topbar{flex-wrap:wrap;padding:12px 14px}.brand{min-width:auto}.fleet-stats{order:3;width:100%}.toolbar{gap:6px}.search{width:180px}.fleet{padding:13px}.agent-card{max-height:none;height:calc(100vh - 150px);min-height:560px}.field-row{grid-template-columns:1fr}} + .count-badge{display:inline-block;min-width:17px;margin-left:6px;padding:1px 6px;border-radius:999px;background:#28598f;color:#fff;font-size:10px;font-weight:800;text-align:center} + .handoff-entry{margin:0 0 14px;max-width:92%;padding:9px 12px;border:1px solid #3d3658;border-left:3px solid var(--violet);border-radius:10px;background:#151a2c}.handoff-head{display:flex;align-items:center;gap:7px;margin-bottom:4px;color:var(--violet);font-size:10px;font-weight:800;text-transform:uppercase;letter-spacing:.06em}.handoff-source{color:var(--muted);font-weight:600;text-transform:none;letter-spacing:0}.handoff-title{margin:0 0 4px;color:var(--text);font-size:12.5px;font-weight:700}.handoff-text{color:#d5dcee;font-size:13px;line-height:1.48;white-space:pre-wrap;overflow-wrap:anywhere} + .segmented{display:flex;gap:4px;padding:3px;border:1px solid var(--line);border-radius:9px;background:#0b111c}.segmented button{flex:1;padding:6px 4px;border:1px solid transparent;border-radius:7px;background:transparent;color:var(--muted);cursor:pointer;font-size:11px;font-weight:700;transition:.14s}.segmented button:hover{color:var(--text)}.segmented button.active{border-color:#456fa8;background:#1d3252;color:#eaf2ff} + .recipient-grid{display:flex;flex-wrap:wrap;gap:6px}.recipient-chip{display:flex;align-items:center;gap:6px;padding:5px 10px;border:1px solid var(--line);border-radius:999px;background:#0e1522;color:#c6d2e5;font-size:11.5px;cursor:pointer;user-select:none}.recipient-chip input{margin:0;accent-color:#6ea8fe}.recipient-chip:has(input:checked){border-color:#456fa8;background:#1a2c47;color:#eaf2ff}.recipient-chip:has(input:disabled){opacity:.75;cursor:default} + .hint{color:var(--subtle);font-size:11px;line-height:1.5}.label-row{display:flex;align-items:center;justify-content:space-between;margin-bottom:6px}.label-row label{margin-bottom:0}.link-btn{padding:0;border:0;background:none;color:var(--blue);cursor:pointer;font-size:10.5px;font-weight:700}.link-btn:hover{text-decoration:underline}.field-error{min-height:15px;color:var(--red);font-size:11px} + dialog.drawer{width:min(600px,96vw);max-width:96vw;height:100%;max-height:100%;margin:0 0 0 auto;border:0;border-left:1px solid var(--line2);border-radius:0}dialog.drawer[open]{animation:drawer-in .18s ease}@keyframes drawer-in{from{transform:translateX(50px);opacity:0}}.drawer .dialog-head{position:sticky;top:0;z-index:2;background:#111826} + .drawer-list{display:grid;gap:9px;padding:16px 18px;overflow:auto}.handoff-record{padding:11px 12px;border:1px solid var(--line);border-radius:10px;background:#0e1522}.hr-top{display:flex;align-items:center;gap:8px;flex-wrap:wrap}.hr-route{min-width:0;color:#dce5f3;font-size:12px;overflow:hidden;text-overflow:ellipsis}.hr-route strong{color:var(--text)}.hr-arrow{margin:0 5px;color:var(--violet)}.kind-chip{padding:2px 7px;border:1px solid var(--line2);border-radius:999px;color:#c6d2e5;font-size:9px;font-weight:800;text-transform:uppercase;letter-spacing:.05em}.h-status{padding:2px 8px;border:1px solid;border-radius:999px;font-size:9px;font-weight:800;text-transform:uppercase;letter-spacing:.06em}.h-status.preparing,.h-status.delivering{color:var(--yellow);border-color:#6b5a2e;background:#2a2314}.h-status.delivered{color:var(--green);border-color:#2e5c46;background:#12291e}.h-status.partial{color:var(--violet);border-color:#4b3f6e;background:#1c1830}.h-status.failed{color:var(--red);border-color:#6a3540;background:#2a151a}.hr-time{margin-left:auto;color:var(--subtle);font-size:10px;white-space:nowrap}.hr-title{margin-top:5px;color:var(--text);font-size:12px;font-weight:700}.hr-details{margin-top:7px}.hr-details summary{cursor:pointer;color:var(--muted);font-size:11px;font-weight:700;user-select:none}.hr-content{margin:6px 0 0;max-height:220px;overflow:auto;padding:8px;border:1px solid var(--line);border-radius:8px;background:#0b111c;color:#c9d5e6;font:11px ui-monospace,SFMono-Regular,Consolas,monospace;white-space:pre-wrap;overflow-wrap:anywhere}.hr-error{margin-top:6px;color:var(--red);font-size:11px}.hr-deliveries{display:grid;gap:3px;margin-top:6px;color:var(--muted);font-size:11px}.hr-deliveries .dl-ok{color:var(--green)}.hr-deliveries .dl-wait{color:var(--yellow)}.hr-deliveries .dl-fail{color:var(--red)} + @media(max-width:800px){.topbar{flex-wrap:wrap;padding:12px 14px}.brand{min-width:auto}.fleet-stats{order:3;width:100%}.toolbar{gap:6px}.search{width:180px}.fleet{padding:13px}.agent-card{max-height:none;height:calc(100vh - 150px);min-height:560px}.field-row{grid-template-columns:1fr}dialog.drawer{width:100vw;max-width:100vw}.segmented{flex-wrap:wrap}.segmented button{flex:1 1 31%}} @@ -24,6 +31,7 @@
+
@@ -40,11 +48,35 @@
+ + +
+

Hand off · agent

+
+
+
+
+
+ +
+
+
+
+ + + +

Handoffs

+
+ +
diff --git a/Handoff.cs b/Handoff.cs new file mode 100644 index 0000000..bea29a9 --- /dev/null +++ b/Handoff.cs @@ -0,0 +1,44 @@ +sealed class Handoff +{ + public static readonly string[] Kinds = ["message", "context", "delegate", "review", "broadcast"]; + public const int HistoryLimit = 100; + public const int MaxInFlight = 128; + + public required string Id { get; init; } + public required DateTimeOffset Time { get; init; } + public required string From { get; init; } + public required string[] Recipients { get; init; } + public required string Kind { get; init; } + public string? Title { get; init; } + public string? Instructions { get; init; } + public string? Message { get; init; } + public string Status { get; set; } = "preparing"; + public string? Error { get; set; } + public string? Package { get; set; } + public List Deliveries { get; } = []; + + public object Snapshot() => new + { + id = Id, + createdAt = Time, + from = From, + recipients = Recipients, + kind = Kind, + title = Title, + instructions = Instructions, + message = Message, + status = Status, + error = Error, + content = Package, + deliveries = Deliveries.Select(d => d.Snapshot()).ToArray() + }; +} + +sealed class HandoffDelivery +{ + public required string Agent { get; init; } + public string State { get; set; } = "pending"; + public string? Error { get; set; } + + public object Snapshot() => new { agent = Agent, status = State, error = Error }; +} diff --git a/McpStdioServer.cs b/McpStdioServer.cs index 8b35b60..3aa5792 100644 --- a/McpStdioServer.cs +++ b/McpStdioServer.cs @@ -61,6 +61,10 @@ sealed class McpStdioServer(FleetHost fleet) "fleet_compact_context" => Compact(Required(arguments, "agent"), Optional(arguments, "instructions")), "fleet_reset_context" => await Reset(Required(arguments, "agent")), "fleet_request_review" => Review(Required(arguments, "reviewer"), Required(arguments, "author"), Optional(arguments, "context")), + "fleet_handoff_context" => fleet.BeginHandoff( + Required(arguments, "from"), RequiredArray(arguments, "recipients"), Required(arguments, "kind"), + Optional(arguments, "title"), Optional(arguments, "instructions"), Optional(arguments, "message")), + "fleet_list_handoffs" => fleet.ListHandoffs(Optional(arguments, "agent")), "fleet_stop_agent" => Stop(Required(arguments, "agent")), _ => throw new InvalidOperationException($"Unknown KimiFleet tool '{name}'.") }; @@ -119,6 +123,12 @@ sealed class McpStdioServer(FleetHost fleet) input.TryGetProperty(name, out var values) && values.ValueKind == JsonValueKind.Array ? values.EnumerateArray().Where(x => x.ValueKind == JsonValueKind.String).Select(x => x.GetString()!).ToArray() : null; + private static string[] RequiredArray(JsonElement input, string name) + { + var values = StringArray(input, name); + return values is { Length: > 0 } ? values : throw new InvalidOperationException($"'{name}' must be a non-empty array of agent names."); + } + private static async Task SendAsync(JsonElement id, object result) { var json = $"{{\"jsonrpc\":\"2.0\",\"id\":{id.GetRawText()},\"result\":{JsonSerializer.Serialize(result, Json)}}}"; @@ -136,7 +146,9 @@ sealed class McpStdioServer(FleetHost fleet) Tool("fleet_compact_context", "Compact an agent context while preserving important decisions and active work.", new { type = "object", required = new[] { "agent" }, properties = new { agent = new { type = "string" }, instructions = new { type = "string" } } }), Tool("fleet_reset_context", "Create a fresh ACP context for an idle agent without changing files or scope ownership.", new { type = "object", required = new[] { "agent" }, properties = new { agent = new { type = "string" } } }), Tool("fleet_request_review", "Ask a second agent for read-only peer review of an author's current diff.", new { type = "object", required = new[] { "reviewer", "author" }, properties = new { reviewer = new { type = "string" }, author = new { type = "string" }, context = new { type = "string" } } }), - Tool("fleet_stop_agent", "Stop an agent and release its exclusive scopes.", new { type = "object", required = new[] { "agent" }, properties = new { agent = new { type = "string" } } }) + Tool("fleet_stop_agent", "Stop an agent and release its exclusive scopes.", new { type = "object", required = new[] { "agent" }, properties = new { agent = new { type = "string" } } }), + Tool("fleet_handoff_context", "Send agent-to-agent context over the fleet bus and return the accepted handoff record. kind 'message' delivers trimmed text without source summarization. For 'context', 'delegate', 'review' and 'broadcast', the source Kimi summarizes its live session into a structured package and KimiFleet injects it into every recipient's model context, consuming their token budgets.", new { type = "object", required = new[] { "from", "recipients", "kind" }, properties = new { from = new { type = "string", description = "Name of the sending agent." }, recipients = new { type = "array", minItems = 1, maxItems = 32, items = new { type = "string" }, description = "Names of the agents that receive the handoff." }, kind = new { type = "string", @enum = new[] { "message", "context", "delegate", "review", "broadcast" }, description = "'message' = direct text; other kinds = source-generated context package injected into recipient contexts." }, title = new { type = "string", maxLength = 200, description = "Short subject; labels the handoff and generated package." }, instructions = new { type = "string", maxLength = 8000, description = "Guidance for the source summary and the recipients' next action." }, message = new { type = "string", maxLength = 48000, description = "Required text for kind 'message' (48k max); optional extra detail for generated kinds (16k max)." } } }), + Tool("fleet_list_handoffs", "List handoffs recorded on the fleet bus; pass 'agent' to keep only handoffs sent by or addressed to that agent.", new { type = "object", properties = new { agent = new { type = "string" } } }) ]; private static object Tool(string name, string description, object inputSchema) => new { name, description, inputSchema }; diff --git a/Program.cs b/Program.cs index aef8981..c91cd81 100644 --- a/Program.cs +++ b/Program.cs @@ -53,6 +53,9 @@ sealed class FleetHost : IDisposable private readonly ConcurrentDictionary> _eventSubscribers = new(); private readonly List _recentEvents = []; private readonly object _eventsLock = new(); + private readonly List _handoffs = []; + private readonly object _handoffsLock = new(); + private long _handoffId; private bool _disposed; public FleetHost(FleetOptions options) @@ -95,6 +98,10 @@ sealed class FleetHost : IDisposable await PromptAgentAsync(context, parts[1]); else if (context.Request.HttpMethod == "POST" && parts.Length == 3 && parts[0] == "agents" && parts[2] == "review") await ReviewAsync(context, parts[1]); + else if (context.Request.HttpMethod == "POST" && parts.Length == 3 && parts[0] == "agents" && parts[2] == "handoff") + await BeginHandoffAsync(context, parts[1]); + else if (context.Request.HttpMethod == "GET" && path == "handoffs") + await RespondJsonAsync(context, ListHandoffs(context.Request.QueryString["agent"])); else if (context.Request.HttpMethod == "POST" && parts.Length == 3 && parts[0] == "agents" && parts[2] == "stop") await StopAgentAsync(context, parts[1]); else if (context.Request.HttpMethod == "POST" && parts.Length == 3 && parts[0] == "agents" && parts[2] == "cancel") @@ -135,6 +142,13 @@ sealed class FleetHost : IDisposable await RespondJsonAsync(context, new { accepted = true, reviewer = reviewerName, author = request.Author }); } + private async Task BeginHandoffAsync(HttpListenerContext context, string from) + { + var request = await ReadJsonAsync(context); + var handoff = BeginHandoff(from, request.Recipients, request.Kind, request.Title, request.Instructions, request.Message); + await RespondJsonAsync(context, handoff, 202); + } + private async Task StopAgentAsync(HttpListenerContext context, string name) { RemoveAgent(name); @@ -199,6 +213,186 @@ sealed class FleetHost : IDisposable } public void Stop(string name) => RemoveAgent(name); + public object BeginHandoff(string from, string[]? recipients, string? kind, string? title, string? instructions, string? message) + { + var sender = GetAgent(from); + if (!sender.CanReceiveHandoffs) + throw new InvalidOperationException($"Source agent '{sender.Name}' has no active Kimi session."); + if (string.IsNullOrWhiteSpace(kind) || !Handoff.Kinds.Contains(kind.Trim(), StringComparer.OrdinalIgnoreCase)) + throw new InvalidOperationException($"Handoff kind must be one of: {string.Join(", ", Handoff.Kinds)}."); + var normalizedKind = kind.Trim().ToLowerInvariant(); + if (recipients is null || recipients.Length == 0) + throw new InvalidOperationException("Handoff needs at least one recipient."); + if (recipients.Length > 32) + throw new InvalidOperationException("A handoff supports at most 32 recipients."); + var names = new List(); + foreach (var recipient in recipients) + { + var name = ValidateName(recipient); + if (string.Equals(name, sender.Name, StringComparison.OrdinalIgnoreCase)) + throw new InvalidOperationException("Sender cannot be a recipient of its own handoff."); + if (names.Contains(name, StringComparer.OrdinalIgnoreCase)) + throw new InvalidOperationException($"Duplicate recipient '{name}'."); + var target = GetAgent(name); + if (!target.CanReceiveHandoffs) + throw new InvalidOperationException($"Recipient agent '{name}' has no active Kimi session."); + names.Add(name); + } + if (normalizedKind == "message" && string.IsNullOrWhiteSpace(message)) + throw new InvalidOperationException("Kind 'message' requires non-empty 'message' text."); + var normalizedTitle = NormalizeHandoffField(title, "title", 200); + var normalizedInstructions = NormalizeHandoffField(instructions, "instructions", 8_000); + var normalizedMessage = NormalizeHandoffField(message, "message", normalizedKind == "message" ? 48_000 : 16_000); + var handoff = new Handoff + { + Id = $"h-{Interlocked.Increment(ref _handoffId)}", + Time = DateTimeOffset.UtcNow, + From = sender.Name, + Recipients = names.ToArray(), + Kind = normalizedKind, + Title = normalizedTitle, + Instructions = normalizedInstructions, + Message = normalizedMessage + }; + foreach (var name in names) handoff.Deliveries.Add(new HandoffDelivery { Agent = name }); + lock (_handoffsLock) + { + if (_handoffs.Count(h => h.Status is "preparing" or "delivering") >= Handoff.MaxInFlight) + throw new InvalidOperationException($"The fleet already has {Handoff.MaxInFlight} handoffs in flight."); + _handoffs.Add(handoff); + TrimHandoffsLocked(); + } + Publish(sender.Name, "handoff", $"Handoff {handoff.Id} ({normalizedKind}) accepted for {string.Join(", ", names)}."); + _ = ProcessHandoffAsync(handoff, sender); + lock (_handoffsLock) return handoff.Snapshot(); + } + + public IReadOnlyList ListHandoffs(string? agent) + { + lock (_handoffsLock) + return _handoffs + .Where(h => string.IsNullOrWhiteSpace(agent) + || string.Equals(h.From, agent, StringComparison.OrdinalIgnoreCase) + || h.Recipients.Contains(agent, StringComparer.OrdinalIgnoreCase)) + .Reverse() + .Select(h => (object)h.Snapshot()) + .ToArray(); + } + + private async Task ProcessHandoffAsync(Handoff handoff, KimiAgent sender) + { + try + { + string package; + if (handoff.Kind == "message") + { + package = handoff.Message!; + } + else + { + var produced = await sender.RunHiddenPromptAsync(PackagePrompt(handoff), capture: true); + if (string.IsNullOrWhiteSpace(produced)) + throw new InvalidOperationException($"Source agent '{sender.Name}' produced no handoff package."); + package = produced.Trim(); + if (handoff.Message is not null) package += $"\n\nNote from {sender.Name}: {handoff.Message}"; + } + lock (_handoffsLock) { handoff.Package = package; handoff.Status = "delivering"; } + Publish(handoff.From, "handoff", $"Handoff {handoff.Id} package ready ({package.Length} chars); delivering to {handoff.Deliveries.Count} recipient(s)."); + await Task.WhenAll(handoff.Deliveries.Select(delivery => DeliverHandoffAsync(handoff, delivery, package))); + lock (_handoffsLock) + { + var failed = handoff.Deliveries.Count(d => d.State == "failed"); + handoff.Status = failed == 0 ? "delivered" : failed == handoff.Deliveries.Count ? "failed" : "partial"; + if (handoff.Status != "delivered") + handoff.Error = string.Join("; ", handoff.Deliveries.Where(d => d.Error is not null).Select(d => $"{d.Agent}: {d.Error}")); + TrimHandoffsLocked(); + } + Publish(handoff.From, "handoff", $"Handoff {handoff.Id} finished with status '{handoff.Status}'."); + } + catch (Exception ex) + { + lock (_handoffsLock) + { + handoff.Status = "failed"; + handoff.Error = ex.Message; + foreach (var delivery in handoff.Deliveries.Where(d => d.State is "pending" or "delivering")) + { + delivery.State = "failed"; + delivery.Error = ex.Message; + } + TrimHandoffsLocked(); + } + Publish(handoff.From, "handoff", $"Handoff {handoff.Id} failed: {ex.Message}"); + } + } + + private async Task DeliverHandoffAsync(Handoff handoff, HandoffDelivery delivery, string package) + { + try + { + var recipient = GetAgent(delivery.Agent); + var completed = await recipient.RunHiddenPromptAsync( + DeliveryPrompt(handoff, package), capture: false, + onTurnStart: () => + { + lock (_handoffsLock) delivery.State = "delivering"; + recipient.AddHandoffEntry(handoff, package); + Publish(delivery.Agent, "handoff", $"Handoff {handoff.Id} delivery started."); + }); + if (completed is null) throw new InvalidOperationException("Recipient turn failed; see the agent chat for details."); + lock (_handoffsLock) delivery.State = "delivered"; + Publish(delivery.Agent, "handoff", $"Handoff {handoff.Id} from '{handoff.From}' delivered."); + } + catch (Exception ex) + { + lock (_handoffsLock) { delivery.State = "failed"; delivery.Error = ex.Message; } + Publish(delivery.Agent, "handoff", $"Handoff {handoff.Id} delivery to '{delivery.Agent}' failed: {ex.Message}"); + } + } + + private static string PackagePrompt(Handoff handoff) + { + var prompt = + $"You are handing off work to teammate agents via the KimiFleet context bus (kind: {handoff.Kind}). " + + "Produce a compact handoff package (250 words max) with exactly these sections: objective, decisions, constraints, changed files, verification, risks, next steps. " + + "Base it on your actual session state and real work so far. Output only the package, no preamble or closing remarks."; + if (handoff.Title is not null) prompt += $" Handoff title: {handoff.Title}."; + if (handoff.Instructions is not null) prompt += $" Coordinator instructions: {handoff.Instructions}"; + return prompt; + } + + private static string DeliveryPrompt(Handoff handoff, string package) + { + var header = $"[KimiFleet handoff {handoff.Id} · kind={handoff.Kind} · from={handoff.From}{(handoff.Title is null ? string.Empty : $" · title={handoff.Title}")}]"; + var directive = handoff.Kind switch + { + "delegate" => "Continue this work as your own task, autonomously and within your scopes. Do not wait for further instructions.", + "review" => "Review this package and report findings by severity with concrete file/line references where possible. Review only; do not edit files.", + "message" => "Read the message and respond as appropriate.", + _ => "Absorb this context into your working knowledge and acknowledge with one short sentence. Do not start new work unless the package explicitly asks you to." + }; + return $"{header}\n{directive}\n\n{package}"; + } + + private static string? NormalizeHandoffField(string? value, string name, int maxLength) + { + if (string.IsNullOrWhiteSpace(value)) return null; + var normalized = value.Trim(); + if (normalized.Length > maxLength) + throw new InvalidOperationException($"Handoff {name} is limited to {maxLength:N0} characters."); + return normalized; + } + + private void TrimHandoffsLocked() + { + while (_handoffs.Count > Handoff.HistoryLimit) + { + var completedIndex = _handoffs.FindIndex(h => h.Status is "delivered" or "failed" or "partial"); + if (completedIndex < 0) break; + _handoffs.RemoveAt(completedIndex); + } + } + private KimiAgent GetAgent(string name) => _agents.TryGetValue(name, out var agent) ? agent : throw new InvalidOperationException($"Unknown agent '{name}'."); @@ -351,6 +545,7 @@ sealed class FleetHost : IDisposable private sealed record StartAgentRequest(string Name, string Workspace, string? Role, string[]? Scopes); private sealed record PromptRequest(string Prompt); private sealed record ReviewRequest(string Author, string? Context); + private sealed record HandoffRequest(string[]? Recipients, string? Kind, string? Title, string? Instructions, string? Message); private sealed record CompactContextRequest(string? Instructions); } @@ -362,6 +557,7 @@ sealed class KimiAgent : IDisposable private readonly ConcurrentDictionary> _pending = new(); private readonly SemaphoreSlim _stdinLock = new(1, 1); private readonly SemaphoreSlim _promptGate = new(1, 1); + private readonly SemaphoreSlim _turnStateGate = new(1, 1); private readonly List _chat = []; private readonly Dictionary _toolEntries = []; private readonly object _chatLock = new(); @@ -370,10 +566,12 @@ sealed class KimiAgent : IDisposable private long _chatId; private long _estimatedTokens; private int _queuedPrompts; - private int _cancellationRequested; + private bool _cancelSent; private string? _sessionId; private string? _assistantEntryId; private string? _thoughtEntryId; + private StringBuilder? _captureBuffer; + private bool _hideTurnTranscript; private bool _resettingContext; private bool _disposed; @@ -384,6 +582,7 @@ sealed class KimiAgent : IDisposable public string State { get; private set; } = "starting"; public string Activity { get; private set; } = "Starting ACP session"; public DateTimeOffset ActivityUpdatedAt { get; private set; } = DateTimeOffset.UtcNow; + public bool CanReceiveHandoffs => _sessionId is not null && State is not "starting" and not "stopped" && !_resettingContext && !_disposed; public KimiAgent(string name, string workspace, string role, IReadOnlyList scopes, FleetOptions options, Action publish) { @@ -429,62 +628,113 @@ sealed class KimiAgent : IDisposable _publish(Name, "initialized", initialized.GetProperty("agentInfo").GetProperty("name").GetString() ?? "Kimi Code CLI"); var session = await RequestAsync("session/new", new { cwd = Workspace, mcpServers = Array.Empty() }); _sessionId = session.GetProperty("sessionId").GetString() ?? throw new InvalidOperationException("ACP did not return sessionId."); + await SeedFleetContextAsync(); State = "ready"; SetActivity("Ready for a task"); AddChat("system", $"Kimi K3 session ready · {Role}"); _publish(Name, "ready", $"Role={Role}; scopes={string.Join(", ", Scopes)}"); } - public Task PromptAsync(string prompt) => QueuePromptAsync(prompt); + public Task PromptAsync(string prompt) => QueuePromptAsync(prompt, hidden: false, capture: false); - private Task QueuePromptAsync(string prompt) + public Task RunHiddenPromptAsync(string prompt, bool capture, Action? onTurnStart = null) => + QueuePromptAsync(prompt, hidden: true, capture, onTurnStart); + + private Task QueuePromptAsync(string prompt, bool hidden, bool capture, Action? onTurnStart = null) { if (string.IsNullOrWhiteSpace(prompt)) throw new InvalidOperationException("Prompt must not be empty."); if (_sessionId is null) throw new InvalidOperationException($"Agent '{Name}' has no ACP session."); int queuedCount; lock (_promptLifecycleLock) { + if (State is "starting" or "stopped" || _disposed) + throw new InvalidOperationException($"Agent '{Name}' is not ready for prompts."); if (_resettingContext) throw new InvalidOperationException("Cannot queue a prompt while context is being reset."); - AddChat("user", prompt); + if (!hidden) AddChat("user", prompt); AddEstimatedTokens(prompt); queuedCount = Interlocked.Increment(ref _queuedPrompts); } - return RunPromptAsync(prompt, queuedCount); + return RunPromptAsync(prompt, queuedCount, hidden, capture, onTurnStart); } - private async Task RunPromptAsync(string prompt, int queuedCount) + private async Task RunPromptAsync(string prompt, int queuedCount, bool hidden, bool capture, Action? onTurnStart) { if (_promptGate.CurrentCount == 0) _publish(Name, "queued", $"Task queued at position {queuedCount}."); await _promptGate.WaitAsync(); Interlocked.Decrement(ref _queuedPrompts); + var captureBuffer = capture ? new StringBuilder() : null; try { - State = "working"; - SetActivity("Planning assigned task"); - _publish(Name, "prompt", prompt.Length > 160 ? prompt[..160] + "…" : prompt); - await RequestAsync("session/prompt", new { sessionId = _sessionId, prompt = new[] { new { type = "text", text = prompt } } }); - State = "ready"; - SetActivity("Ready for a task"); - var cancelled = Interlocked.Exchange(ref _cancellationRequested, 0) == 1; + onTurnStart?.Invoke(); + await _turnStateGate.WaitAsync(); + try + { + _cancelSent = false; + SetActivity(hidden ? "Processing a handoff" : "Planning assigned task"); + } + finally { _turnStateGate.Release(); } + lock (_chatLock) + { + _assistantEntryId = null; + _thoughtEntryId = null; + if (captureBuffer is not null) _captureBuffer = captureBuffer; + _hideTurnTranscript = hidden && capture; + } + var promptResult = await RequestAsync( + "session/prompt", + new { sessionId = _sessionId, prompt = new[] { new { type = "text", text = prompt } } }, + onSent: async () => + { + await _turnStateGate.WaitAsync(); + try + { + State = "working"; + if (!hidden) _publish(Name, "prompt", prompt.Length > 160 ? prompt[..160] + "…" : prompt); + } + finally { _turnStateGate.Release(); } + }); + bool cancelled; + await _turnStateGate.WaitAsync(); + try + { + cancelled = promptResult.TryGetProperty("stopReason", out var stopReason) && stopReason.GetString() == "cancelled"; + _cancelSent = false; + State = "ready"; + SetActivity("Ready for a task"); + } + finally { _turnStateGate.Release(); } _publish(Name, cancelled ? "cancelled" : "completed", cancelled ? "Task cancelled." : "Prompt completed."); - return true; + return cancelled ? null : captureBuffer?.ToString() ?? string.Empty; } catch (Exception ex) { - Interlocked.Exchange(ref _cancellationRequested, 0); - State = "error"; - SetActivity("Needs attention"); + await _turnStateGate.WaitAsync(); + try + { + _cancelSent = false; + State = "error"; + SetActivity("Needs attention"); + } + finally { _turnStateGate.Release(); } AddChat("error", ex.Message); _publish(Name, "error", ex.Message); - return false; + return null; + } + finally + { + lock (_chatLock) + { + if (captureBuffer is not null) _captureBuffer = null; + _hideTurnTranscript = false; + } + _promptGate.Release(); } - finally { _promptGate.Release(); } } public async Task CompactContextAsync(string? instructions) { var command = string.IsNullOrWhiteSpace(instructions) ? "/compact" : $"/compact {instructions.Trim()}"; - if (await QueuePromptAsync(command)) + if (await QueuePromptAsync(command, hidden: false, capture: false) is not null) { Interlocked.Exchange(ref _estimatedTokens, Math.Min(Interlocked.Read(ref _estimatedTokens), _options.MaxContextTokens / 4)); AddChat("system", "Context compacted"); @@ -514,6 +764,7 @@ sealed class KimiAgent : IDisposable _thoughtEntryId = null; } Interlocked.Exchange(ref _estimatedTokens, 0); + await SeedFleetContextAsync(); State = "ready"; SetActivity("Ready for a task"); AddChat("system", "Fresh Kimi K3 context created"); @@ -527,15 +778,39 @@ sealed class KimiAgent : IDisposable public async Task CancelAsync() { if (_sessionId is null) throw new InvalidOperationException($"Agent '{Name}' has no ACP session."); - if (State != "working") throw new InvalidOperationException($"Agent '{Name}' is not working."); - await SendRawAsync(JsonSerializer.Serialize(new { jsonrpc = "2.0", method = "session/cancel", @params = new { sessionId = _sessionId } })); - Interlocked.Exchange(ref _cancellationRequested, 1); - SetActivity("Cancelling current task"); - AddChat("system", "Cancellation requested"); - _publish(Name, "cancel", "Cancellation requested."); + await _turnStateGate.WaitAsync(); + try + { + if (State != "working") throw new InvalidOperationException($"Agent '{Name}' is not working."); + if (_cancelSent) throw new InvalidOperationException($"Cancellation is already pending for agent '{Name}'."); + await SendRawAsync(JsonSerializer.Serialize(new { jsonrpc = "2.0", method = "session/cancel", @params = new { sessionId = _sessionId } })); + _cancelSent = true; + SetActivity("Cancelling current task"); + AddChat("system", "Cancellation requested"); + _publish(Name, "cancel", "Cancellation requested."); + } + finally { _turnStateGate.Release(); } } - private async Task RequestAsync(string method, object parameters) + private async Task SeedFleetContextAsync() + { + var prompt = + $"You are KimiFleet agent '{Name}' in a local multi-agent fleet. Your fleet API is http://127.0.0.1:{_options.Port}. " + + "You may autonomously coordinate when it materially helps the task: list teammates with GET /agents, and send context with " + + $"POST /agents/{Name}/handoff using JSON {{\"recipients\":[\"agent-name\"],\"kind\":\"message|context|delegate|review|broadcast\",\"title\":\"...\",\"instructions\":\"...\",\"message\":\"...\"}}. " + + "Use kind=message for direct text; other kinds ask you to summarize this live session before delivery. Use only your own name in the source URL, respect scope ownership, and avoid unnecessary broadcasts. Retain these instructions silently; do not acknowledge them."; + AddEstimatedTokens(prompt); + lock (_chatLock) + { + _assistantEntryId = null; + _thoughtEntryId = null; + _hideTurnTranscript = true; + } + try { await RequestAsync("session/prompt", new { sessionId = _sessionId, prompt = new[] { new { type = "text", text = prompt } } }); } + finally { lock (_chatLock) _hideTurnTranscript = false; } + } + + private async Task RequestAsync(string method, object parameters, Func? onSent = null) { var id = Interlocked.Increment(ref _requestId); var completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); @@ -544,6 +819,7 @@ sealed class KimiAgent : IDisposable try { await SendRawAsync(envelope); + if (onSent is not null) await onSent(); var requestTimeout = method == "session/prompt" ? TimeSpan.FromHours(2) : TimeSpan.FromMinutes(2); using var timeout = new CancellationTokenSource(requestTimeout); await using var registration = timeout.Token.Register(() => completion.TrySetException(new TimeoutException($"ACP request '{method}' timed out."))); @@ -702,18 +978,35 @@ sealed class KimiAgent : IDisposable if (type == "agent_message_chunk") { var text = ExtractText(update.TryGetProperty("content", out var content) ? content : default); - AppendChatChunk("assistant", text, ref _assistantEntryId); + bool hideTranscript; + lock (_chatLock) hideTranscript = _hideTurnTranscript; + if (!hideTranscript) AppendChatChunk("assistant", text, ref _assistantEntryId); + if (!string.IsNullOrEmpty(text)) + { + lock (_chatLock) + { + if (_captureBuffer is not null && _captureBuffer.Length < 48_000) + { + var remaining = 48_000 - _captureBuffer.Length; + _captureBuffer.Append(text.AsSpan(0, Math.Min(remaining, text.Length))); + } + } + } AddEstimatedTokens(text); } else if (type == "agent_thought_chunk") { var text = ExtractText(update.TryGetProperty("content", out var content) ? content : default); - AppendChatChunk("thought", text, ref _thoughtEntryId); + bool hideTranscript; + lock (_chatLock) hideTranscript = _hideTurnTranscript; + if (!hideTranscript) AppendChatChunk("thought", text, ref _thoughtEntryId); AddEstimatedTokens(text); } else if (type is "tool_call" or "tool_call_update") { - UpsertToolEntry(update); + bool hideTranscript; + lock (_chatLock) hideTranscript = _hideTurnTranscript; + if (!hideTranscript) UpsertToolEntry(update); _assistantEntryId = null; _thoughtEntryId = null; } @@ -777,6 +1070,20 @@ sealed class KimiAgent : IDisposable } } + public void AddHandoffEntry(Handoff handoff, string package) + { + lock (_chatLock) + { + var entry = NewChatEntry("handoff", LimitText(package, 48_000)); + entry.Title = handoff.Title ?? $"{handoff.Kind} from {handoff.From}"; + entry.Source = handoff.From; + entry.Kind = handoff.Kind; + AddChatEntryLocked(entry); + _assistantEntryId = null; + _thoughtEntryId = null; + } + } + private AgentChatEntry NewChatEntry(string type, string text) => new() { Id = $"{Name}-{Interlocked.Increment(ref _chatId)}", diff --git a/README.md b/README.md index 0491009..b424d61 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,9 @@ Start both the dashboard and an MCP stdio server: dotnet run --project KimiFleet.csproj -- --mcp --port 7373 ``` -The server implements `fleet_list_agents`, `fleet_start_agent`, `fleet_get_chat`, `fleet_assign_task`, `fleet_cancel_task`, `fleet_compact_context`, `fleet_reset_context`, `fleet_request_review`, and `fleet_stop_agent`. In MCP mode stdout is protocol-only; operational logs are written to stderr and remain visible in the dashboard. +The server implements `fleet_list_agents`, `fleet_start_agent`, `fleet_get_chat`, `fleet_assign_task`, `fleet_cancel_task`, `fleet_compact_context`, `fleet_reset_context`, `fleet_request_review`, `fleet_handoff_context`, `fleet_list_handoffs`, and `fleet_stop_agent`. In MCP mode stdout is protocol-only; operational logs are written to stderr and remain visible in the dashboard. + +`fleet_handoff_context` requires `from`, `recipients` and `kind`, and accepts optional `title`, `instructions` and `message`. With `kind: "message"` the literal `message` text is delivered unchanged; with `context`, `delegate`, `review` or `broadcast` the source Kimi generates a structured context package from its current session plus `title`/`instructions`, and KimiFleet injects it into the recipients' model contexts. `fleet_list_handoffs` takes an optional `agent` and returns the recorded handoff history. ## API @@ -64,5 +66,61 @@ Useful endpoints: - `GET /health`, `GET /agents`, `GET /agents/{name}/chat`, `GET /events` - `POST /agents`, `POST /agents/{name}/prompt`, `POST /agents/{name}/cancel`, `POST /agents/{name}/review`, `POST /agents/{name}/stop` - `POST /agents/{name}/context/compact`, `POST /agents/{name}/context/reset` +- `POST /agents/{from}/handoff`, `GET /handoffs?agent={name}` All ACP children launch as `kimi --yolo --model kimi-code/k3 acp`. Scope locks reject exact and nested overlaps in the same workspace. They coordinate ownership; yolo agents can still run shell commands, so the locks do not replace peer review or a final `git status` check before edits. + +## Agent-to-agent context bus + +Agents hand context to each other through fleet handoffs instead of pasting transcripts into prompts. Every new or reset Kimi session is silently given its own fleet URL and REST handoff contract, so an agent can initiate a transfer itself with its Bash tool; the dashboard and an external MCP orchestrator can initiate the same operation on its behalf. A handoff names a sender (`from`), one or more `recipients`, and a `kind`: + +- `message` — direct handoff. The trimmed `message` text is delivered without asking the source to generate a summary. +- `context`, `delegate`, `review`, `broadcast` — generated handoffs. The source Kimi summarizes its live session into a structured package, guided by `title` and `instructions` (plus any extra `message` detail): `context` shares working context, `delegate` assigns a bounded piece of work, `review` asks for read-only peer review, and `broadcast` delivers the same package to every recipient. + +Generated packages are injected into each recipient's model context, so they consume the recipients' token budget — keep titles and instructions focused. The same applies on the MCP side via `fleet_handoff_context`. + +Direct message between two agents: + +```bash +curl -X POST http://127.0.0.1:7373/agents/vulkan/handoff \ + -H 'Content-Type: application/json' \ + -d '{"recipients":["reviewer"],"kind":"message","title":"Render-pass refactor ready","message":"The refactor is done and tests pass; please take a look when free."}' +``` + +Generated delegate package: + +```bash +curl -X POST http://127.0.0.1:7373/agents/lead/handoff \ + -H 'Content-Type: application/json' \ + -d '{"recipients":["vulkan"],"kind":"delegate","title":"Validation layer cleanup","instructions":"Enable Vulkan validation layers in debug builds only and document the toggle."}' +``` + +Generated review package: + +```bash +curl -X POST http://127.0.0.1:7373/agents/vulkan/handoff \ + -H 'Content-Type: application/json' \ + -d '{"recipients":["reviewer"],"kind":"review","title":"Review render-pass refactor","instructions":"Review only, do not edit. Check synchronization and resource lifetime in the current diff."}' +``` + +Generated broadcast to several agents: + +```bash +curl -X POST http://127.0.0.1:7373/agents/lead/handoff \ + -H 'Content-Type: application/json' \ + -d '{"recipients":["vulkan","reviewer","docs"],"kind":"broadcast","title":"Swapchain API freeze","instructions":"The swapchain API is frozen until Friday; base all pending work on it."}' +``` + +Query handoff history, filtered to one agent or across the fleet: + +```bash +curl 'http://127.0.0.1:7373/handoffs?agent=vulkan' +curl http://127.0.0.1:7373/handoffs +``` + +Delivery and history semantics: + +- `POST /agents/{from}/handoff` validates the sender and recipients, records the handoff, and returns the accepted record (id, timestamp, sender, recipients, kind, title). +- Delivery is asynchronous: a handoff for a busy agent is queued behind its current task and delivered in order, using the same per-agent prompt queue as regular tasks, so one instance never receives overlapping prompts. +- The host retains handoff history for the lifetime of the process; `GET /handoffs` returns the full history, with the optional `agent` filter keeping only handoffs sent by or addressed to that agent. +- In the dashboard, handoffs surface in the receiving agent's chat transcript and in the live event feed (`/events`), so you can watch context move between agents as it happens.