From 2b645691387e1535cc63d6ebf2ba4e996387c8b2 Mon Sep 17 00:00:00 2001 From: Multica Eve Date: Wed, 21 Oct 2026 19:01:56 +0800 Subject: [PATCH] MUL-8126: feat(agent): steer running Kiro turns via _session/steer (#9396) * feat(agent): support steering running Kiro turns via _session/steer Kiro CLI 2.33.0 exposes an undocumented ACP extension, _session/steer, that queues a message and injects it into the in-flight session/prompt at the next tool or message boundary. Wire it up as the kiro backend's Session.Supplement (active only while the prompt is in flight, and only when the daemon negotiated task supplements), and gate kiro >= 2.33.0 in SupportsTaskSupplement. MUL-8126 Co-authored-by: multica-agent * fix(supplement): allow kiro in server-side steering whitelists The capability handshake (StartAgentTaskWithSupplement), direct steering (CreateTaskSupplement) and composer reply binding (BindCommentTaskSupplement) each whitelist providers, and none included kiro, so a negotiated Kiro run never got a capability row and steering stayed unreachable. Add kiro to all three, regenerate sqlc, and cover kiro in the negotiation and both delivery path tests. Also widen the kiro supplement test's readiness wait, which was flaky under full-package parallel load. MUL-8126 Co-authored-by: multica-agent * fix(supplement): allow grok in composer reply binding BindCommentTaskSupplement was the only steering whitelist without grok, so a reply from the normal composer to a running Grok turn fell back to the next run even though direct steering worked. Add grok and cover it in the composer binding test. MUL-8126 Co-authored-by: multica-agent --------- Co-authored-by: multica-agent --- README.md | 2 +- README.zh.md | 2 +- server/internal/handler/comment_steer_test.go | 2 +- .../internal/handler/task_supplement_test.go | 5 +- server/pkg/agent/kiro.go | 92 ++++++++- server/pkg/agent/kiro_supplement_test.go | 177 ++++++++++++++++++ server/pkg/agent/version.go | 5 + server/pkg/agent/version_test.go | 1 + server/pkg/db/generated/supplement.sql.go | 6 +- server/pkg/db/queries/supplement.sql | 6 +- 10 files changed, 285 insertions(+), 13 deletions(-) create mode 100644 server/pkg/agent/kiro_supplement_test.go diff --git a/README.md b/README.md index 4fd378aa9..cbd001aea 100644 --- a/README.md +++ b/README.md @@ -87,7 +87,7 @@ issue — so nobody reconstructs context, and nothing ships without a human sayi *Which agent touched this? What did it run? What did it cost? Open the run.* - **[Execution log](https://multica.ai/docs/tasks) →** Replay every tool call, command, and error, timestamped. -- **Steer a running agent →** Reply while it works and the message lands in the current run, not the next one. Claude Code, Codex, and Grok today. +- **Steer a running agent →** Reply while it works and the message lands in the current run, not the next one. Claude Code, Codex, Grok, and Kiro today. - **Usage analytics →** See what each run cost, per agent and per issue. - **[Review gates](https://multica.ai/docs/issues) →** Work lands in review, not in main. You decide what ships. - **[Inbox](https://multica.ai/docs/inbox) →** Get pinged when an agent needs a call, not for every step. diff --git a/README.zh.md b/README.zh.md index 5317472c3..e2ebcc7fd 100644 --- a/README.zh.md +++ b/README.zh.md @@ -84,7 +84,7 @@ diff，全都挂在同一个任务下——没人需要重新捋一遍上下文 *这活哪个智能体动过？它到底跑了什么？花了多少？点开那次运行。* - **[执行日志](https://multica.ai/docs/zh/tasks) →** 每次工具调用、命令和报错都带时间戳，可以完整回放。 -- **运行中插话 →** 智能体干活时直接回复，消息会进到当前这次运行，不用等下一次。目前支持 Claude Code、Codex 和 Grok。 +- **运行中插话 →** 智能体干活时直接回复，消息会进到当前这次运行，不用等下一次。目前支持 Claude Code、Codex、Grok 和 Kiro。 - **用量统计 →** 每次运行花了多少，按智能体、按任务都看得到。 - **[人来验收](https://multica.ai/docs/zh/issues) →** 活先进入审核中，不直接进 main。上不上线你说了算。 - **[收件箱](https://multica.ai/docs/zh/inbox) →** 只在智能体需要你拍板时提醒你，而不是每一步都来烦你。 diff --git a/server/internal/handler/comment_steer_test.go b/server/internal/handler/comment_steer_test.go index 40f28a7cc..1e6ea1063 100644 --- a/server/internal/handler/comment_steer_test.go +++ b/server/internal/handler/comment_steer_test.go @@ -58,7 +58,7 @@ func outcomeFor(outcomes []CommentTriggerOutcome, agentID string) (CommentTrigge } func TestCreateCommentSteersTheThreadAgentsRunningTurn(t *testing.T) { - for _, provider := range []string{"codex", "cursor"} { + for _, provider := range []string{"codex", "grok", "cursor", "kiro"} { t.Run(provider, func(t *testing.T) { f := newSupplementFixture(t, provider, "running", true) dbfx.Exec(t, `UPDATE comment SET content = $2 WHERE id = $1`, f.triggerID, diff --git a/server/internal/handler/task_supplement_test.go b/server/internal/handler/task_supplement_test.go index f26aabcf0..f27cf0c34 100644 --- a/server/internal/handler/task_supplement_test.go +++ b/server/internal/handler/task_supplement_test.go @@ -181,6 +181,7 @@ func TestTaskSupplementNegotiationFailsClosed(t *testing.T) { {name: "claude negotiated", provider: "claude", daemonAdvertises: true, wantCapabilityRow: true}, {name: "grok negotiated", provider: "grok", daemonAdvertises: true, wantCapabilityRow: true}, {name: "cursor negotiated", provider: "cursor", daemonAdvertises: true, wantCapabilityRow: true}, + {name: "kiro negotiated", provider: "kiro", daemonAdvertises: true, wantCapabilityRow: true}, {name: "claude old daemon", provider: "claude", daemonAdvertises: false, wantCapabilityRow: false}, {name: "old daemon", provider: "codex", daemonAdvertises: false, wantCapabilityRow: false}, {name: "unsupported runtime", provider: "kimi", daemonAdvertises: true, wantCapabilityRow: false}, @@ -213,7 +214,7 @@ func TestTaskSupplementCapabilityDoesNotBreakNonIssueStarts(t *testing.T) { if testHandler == nil { t.Skip("database not available") } - for _, provider := range []string{"codex", "claude", "grok", "cursor"} { + for _, provider := range []string{"codex", "claude", "grok", "cursor", "kiro"} { t.Run(provider, func(t *testing.T) { runtimeID := dbfx.Runtime(t, "supplement-no-issue", testutil.Cols{"provider": provider}) agentID := dbfx.Agent(t, "Supplement no issue", runtimeID) @@ -230,7 +231,7 @@ func TestTaskSupplementCapabilityDoesNotBreakNonIssueStarts(t *testing.T) { } func TestTaskSupplementCompletionDoesNotReplayBoundComments(t *testing.T) { - for _, provider := range []string{"codex", "claude", "grok", "cursor"} { + for _, provider := range []string{"codex", "claude", "grok", "cursor", "kiro"} { for _, status := range []string{"pending", "delivering", "delivered", "failed"} { for _, ordinary := range []string{"none", "unhandled", "queued"} { t.Run(provider+"/"+status+"/"+ordinary, func(t *testing.T) { diff --git a/server/pkg/agent/kiro.go b/server/pkg/agent/kiro.go index c148988c7..1520858f4 100644 --- a/server/pkg/agent/kiro.go +++ b/server/pkg/agent/kiro.go @@ -2,6 +2,7 @@ package agent import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -40,6 +41,74 @@ type kiroBackend struct { cfg Config } +// kiroSupplementTimeout bounds the wait for a `_session/steer` acknowledgement. +// A var so tests can shorten it. +var kiroSupplementTimeout = 8 * time.Second + +// kiroSupplementSession exposes Kiro CLI's `_session/steer` ACP extension only +// while the matching session/prompt request is in flight. Kiro does not +// advertise the extension in initialize, so the daemon gates the feature by +// CLI version (see SupportsTaskSupplement). +// +// Observed contract (kiro-cli 2.33.0): `_session/steer {sessionId, message}` +// returns `{"queued": true}` immediately. Kiro then injects the message into +// the running turn at the next boundary: after the in-flight tool call +// finishes, or after the model's current message if no tool is running. The +// same session/prompt keeps going and answers it, so "queued" already means +// the current turn will see the message. +type kiroSupplementSession struct { + client *hermesClient + mu sync.RWMutex + sessionID string + active atomic.Bool +} + +func (s *kiroSupplementSession) stop() { + s.active.Store(false) +} + +func (s *kiroSupplementSession) ready() bool { + if !s.active.Load() { + return false + } + s.mu.RLock() + defer s.mu.RUnlock() + return s.sessionID != "" +} + +func (s *kiroSupplementSession) send(ctx context.Context, text string) error { + if !s.active.Load() { + return fmt.Errorf("kiro turn is no longer active") + } + s.mu.RLock() + sessionID := s.sessionID + s.mu.RUnlock() + if sessionID == "" { + return fmt.Errorf("kiro session is not ready for steering") + } + rpcCtx, cancel := context.WithTimeout(ctx, kiroSupplementTimeout) + defer cancel() + response, err := s.client.request(rpcCtx, "_session/steer", map[string]any{ + "sessionId": sessionID, + "message": text, + }) + if err != nil { + return fmt.Errorf("kiro _session/steer failed: %w", err) + } + // A successful JSON-RPC frame alone is not proof of delivery: only + // queued=true means Kiro accepted the message for the running turn. + var result struct { + Queued bool `json:"queued"` + } + if err := json.Unmarshal(response, &result); err != nil { + return fmt.Errorf("kiro _session/steer returned an invalid result: %w", err) + } + if !result.Queued { + return fmt.Errorf("kiro _session/steer did not queue the message: %s", strings.TrimSpace(string(response))) + } + return nil +} + func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptions) (*Session, error) { execPath := b.cfg.ExecutablePath if execPath == "" { @@ -139,6 +208,7 @@ func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptio promptDone := make(chan hermesPromptResult, 1) activity := make(chan struct{}, 1) + supplements := &kiroSupplementSession{} c := &hermesClient{ cfg: b.cfg, @@ -201,6 +271,8 @@ func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptio }, } + supplements.client = c + readerDone := make(chan struct{}) go func() { defer close(readerDone) @@ -321,6 +393,9 @@ func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptio } c.sessionID = sessionID + supplements.mu.Lock() + supplements.sessionID = sessionID + supplements.mu.Unlock() b.cfg.Logger.Info("kiro session created", "session_id", sessionID) if opts.Model != "" { @@ -370,11 +445,19 @@ func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptio // shape can drive the turn. // TODO: drop one field once Kiro lands on a single canonical payload. streamingCurrentTurn.Store(true) - _, err = c.request(runCtx, "session/prompt", map[string]any{ + // Steering opens only once the prompt is on the wire and closes as + // soon as it answers, so a supplement can never target setup or a + // finished turn. + _, err = c.requestAndNotifySent(runCtx, "session/prompt", map[string]any{ "sessionId": sessionID, "content": promptBlocks, "prompt": promptBlocks, + }, func() { + if opts.EnableTaskSupplement { + supplements.active.Store(true) + } }) + supplements.stop() if err != nil { if runCtx.Err() == context.DeadlineExceeded { finalStatus = "timeout" @@ -490,7 +573,12 @@ func (b *kiroBackend) Execute(ctx context.Context, prompt string, opts ExecOptio } }() - return &Session{Messages: msgCh, Result: resCh}, nil + session := &Session{Messages: msgCh, Result: resCh} + if opts.EnableTaskSupplement { + session.Supplement = supplements.send + session.SupplementReady = supplements.ready + } + return session, nil } func isKiroGoalCompleteCloseError(err error) bool { diff --git a/server/pkg/agent/kiro_supplement_test.go b/server/pkg/agent/kiro_supplement_test.go new file mode 100644 index 000000000..36b6cf37f --- /dev/null +++ b/server/pkg/agent/kiro_supplement_test.go @@ -0,0 +1,177 @@ +package agent + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// fakeKiroSteerACPScript holds session/prompt open until a `_session/steer` +// request arrives, mirroring kiro-cli 2.33.0: the steer is acknowledged with +// {"queued":true}, then the same prompt answers it and ends the turn. +// KIRO_STEER_RESULT overrides the acknowledgement payload; KIRO_STEER_ERROR +// makes the extension fail with a JSON-RPC error instead. +func fakeKiroSteerACPScript() string { + return `#!/bin/sh +prompt_id="" +while IFS= read -r line; do + if [ -n "$KIRO_REQUESTS_FILE" ]; then + printf '%s\n' "$line" >> "$KIRO_REQUESTS_FILE" + fi + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p') + case "$line" in + *'"method":"initialize"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true}}}\n' "$id" + ;; + *'"method":"session/new"'*) + printf '{"jsonrpc":"2.0","id":%s,"result":{"sessionId":"ses_new"}}\n' "$id" + ;; + *'"method":"session/prompt"'*) + prompt_id="$id" + if [ -z "$KIRO_WAIT_FOR_STEER" ]; then + printf '{"jsonrpc":"2.0","method":"session/notification","params":{"sessionId":"ses_new","update":{"type":"AgentMessageChunk","content":{"type":"text","text":"plain"}}}}\n' + printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$prompt_id" + exit 0 + fi + ;; + *'"method":"_session/steer"'*) + if [ -n "$KIRO_STEER_ERROR" ]; then + printf '{"jsonrpc":"2.0","id":%s,"error":{"code":-32603,"message":"Internal error","data":"no active turn"}}\n' "$id" + continue + fi + steer_result='{"queued":true}' + if [ -n "$KIRO_STEER_RESULT" ]; then + steer_result="$KIRO_STEER_RESULT" + fi + printf '{"jsonrpc":"2.0","id":%s,"result":%s}\n' "$id" "$steer_result" + case "$steer_result" in + *'"queued":true'*) ;; + *) continue ;; + esac + printf '{"jsonrpc":"2.0","method":"_kiro.dev/session/update","params":{"sessionId":"ses_new","update":{"sessionUpdate":"AgentExecutionSteeringInjected","messageId":"steer-1"}}}\n' + printf '{"jsonrpc":"2.0","method":"session/notification","params":{"sessionId":"ses_new","update":{"type":"AgentMessageChunk","content":{"type":"text","text":"PINEAPPLE"}}}}\n' + printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$prompt_id" + exit 0 + ;; + esac +done +` +} + +func startKiroSteerSession(t *testing.T, env map[string]string, negotiated bool) (*Session, context.Context) { + t.Helper() + fakePath := filepath.Join(t.TempDir(), "kiro-cli") + writeTestExecutable(t, fakePath, []byte(fakeKiroSteerACPScript())) + backend, err := New("kiro", Config{ExecutablePath: fakePath, Env: env}) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + t.Cleanup(cancel) + session, err := backend.Execute(ctx, "do the task", ExecOptions{Timeout: 20 * time.Second, EnableTaskSupplement: negotiated}) + if err != nil { + t.Fatal(err) + } + return session, ctx +} + +func waitKiroSupplementReady(t *testing.T, session *Session) { + t.Helper() + // Generous: the fake shell is slow to start when the package runs in parallel. + deadline := time.Now().Add(15 * time.Second) + for !session.SupplementReady() && time.Now().Before(deadline) { + time.Sleep(5 * time.Millisecond) + } + if !session.SupplementReady() { + t.Fatal("supplement never became ready during the active prompt") + } +} + +func drainKiroMessages(session *Session) <-chan []Message { + out := make(chan []Message, 1) + go func() { + var messages []Message + for msg := range session.Messages { + messages = append(messages, msg) + } + out <- messages + }() + return out +} + +func TestKiroSupplementSteersActivePrompt(t *testing.T) { + t.Parallel() + requestsPath := filepath.Join(t.TempDir(), "requests.jsonl") + session, ctx := startKiroSteerSession(t, map[string]string{ + "KIRO_WAIT_FOR_STEER": "1", + "KIRO_REQUESTS_FILE": requestsPath, + }, true) + if session.Supplement == nil || session.SupplementReady == nil { + t.Fatal("negotiated Kiro session did not expose supplement callbacks") + } + waitKiroSupplementReady(t, session) + messages := drainKiroMessages(session) + + if err := session.Supplement(ctx, "The word is PINEAPPLE."); err != nil { + t.Fatalf("steer: %v", err) + } + result := <-session.Result + <-messages + if result.Status != "completed" { + t.Fatalf("turn status=%q error=%q", result.Status, result.Error) + } + if result.Output != "PINEAPPLE" { + t.Fatalf("turn output=%q, want the steered answer in the same turn", result.Output) + } + if session.SupplementReady() { + t.Fatal("supplement stayed ready after session/prompt completed") + } + if err := session.Supplement(ctx, "too late"); err == nil { + t.Fatal("supplement after the turn ended must fail, not report delivery") + } + requests, err := os.ReadFile(requestsPath) + if err != nil { + t.Fatal(err) + } + for _, want := range []string{`"method":"_session/steer"`, `"sessionId":"ses_new"`, `"message":"The word is PINEAPPLE."`} { + if !strings.Contains(string(requests), want) { + t.Errorf("ACP requests missing %s:\n%s", want, requests) + } + } +} + +func TestKiroSupplementRequiresQueuedAcknowledgement(t *testing.T) { + t.Parallel() + for name, env := range map[string]map[string]string{ + "not queued": {"KIRO_STEER_RESULT": `{"queued":false}`}, + "rpc error": {"KIRO_STEER_ERROR": "1"}, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + env["KIRO_WAIT_FOR_STEER"] = "1" + session, ctx := startKiroSteerSession(t, env, true) + waitKiroSupplementReady(t, session) + if err := session.Supplement(ctx, "hello"); err == nil { + t.Fatal("unacknowledged steer must not be reported as delivered") + } + // The turn is still live; t.Cleanup cancels ctx and reaps the fake. + }) + } +} + +func TestKiroSupplementNotExposedWithoutNegotiation(t *testing.T) { + t.Parallel() + session, _ := startKiroSteerSession(t, nil, false) + if session.Supplement != nil || session.SupplementReady != nil { + t.Fatal("un-negotiated Kiro session must not expose supplement callbacks") + } + messages := drainKiroMessages(session) + result := <-session.Result + <-messages + if result.Status != "completed" || result.Output != "plain" { + t.Fatalf("result=%+v, want a normal completed turn", result) + } +} diff --git a/server/pkg/agent/version.go b/server/pkg/agent/version.go index 89affd99d..3d892a5ed 100644 --- a/server/pkg/agent/version.go +++ b/server/pkg/agent/version.go @@ -142,6 +142,11 @@ func SupportsTaskSupplement(provider, version string) bool { // Grok Build 1.0.14 supports atomic delivery of in-turn interjections // through its x.ai/interject ACP extension. minimum = "1.0.14" + case "kiro": + // 2.33.0 is the oldest kiro-cli verified to fold `_session/steer` + // into the running ACP turn (queued, then injected at the next + // tool/message boundary of the same session/prompt). + minimum = "2.33.0" case "cursor": // 2026.10.14 is the oldest build verified to forward postToolUse // additional_context into the model turn in --print mode. diff --git a/server/pkg/agent/version_test.go b/server/pkg/agent/version_test.go index afcfa12f3..53321baf0 100644 --- a/server/pkg/agent/version_test.go +++ b/server/pkg/agent/version_test.go @@ -14,6 +14,7 @@ func TestTaskSupplementVersionGate(t *testing.T) { {"claude", "2.1.110 (Claude Code)", true}, {"claude", "2.1.109", false}, {"grok", "1.0.14", true}, {"grok", "1.0.13", false}, {"grok", "0.2.120", false}, {"cursor", "2026.10.16-e3cb207", true}, {"cursor", "2026.10.14", true}, {"cursor", "2026.04.17-787b533", false}, + {"kiro", "kiro-cli 2.33.0", true}, {"kiro", "2.34.1", true}, {"kiro", "kiro-cli 2.32.9", false}, {"codex", "", false}, {"claude", "dev", false}, {"kimi", "9.0.0", false}, } { if got := SupportsTaskSupplement(tc.provider, tc.version); got != tc.want { diff --git a/server/pkg/db/generated/supplement.sql.go b/server/pkg/db/generated/supplement.sql.go index f62d36adc..f37d30e01 100644 --- a/server/pkg/db/generated/supplement.sql.go +++ b/server/pkg/db/generated/supplement.sql.go @@ -104,7 +104,7 @@ WITH locked_task AS MATERIALIZED ( AND r.workspace_id = $4 AND t.status = 'running' AND cap.capability = 'task-supplement-v1' - AND r.provider IN ('codex', 'claude', 'cursor') + AND r.provider IN ('codex', 'claude', 'grok', 'cursor', 'kiro') FOR UPDATE OF t ), inserted AS ( INSERT INTO task_supplement ( @@ -261,7 +261,7 @@ WITH locked_task AS MATERIALIZED ( AND r.workspace_id = $3 AND t.status = 'running' AND cap.capability = 'task-supplement-v1' - AND r.provider IN ('codex', 'claude', 'grok', 'cursor') + AND r.provider IN ('codex', 'claude', 'grok', 'cursor', 'kiro') FOR UPDATE OF t ), touched_issue AS ( UPDATE issue i SET @@ -736,7 +736,7 @@ WITH candidate AS MATERIALIZED ( SELECT id, workspace_id, issue_id, 'task-supplement-v1' FROM candidate WHERE $2::boolean - AND provider IN ('codex', 'claude', 'grok', 'cursor') + AND provider IN ('codex', 'claude', 'grok', 'cursor', 'kiro') AND issue_id IS NOT NULL ON CONFLICT DO NOTHING RETURNING task_id diff --git a/server/pkg/db/queries/supplement.sql b/server/pkg/db/queries/supplement.sql index 4c3e57715..a6315493b 100644 --- a/server/pkg/db/queries/supplement.sql +++ b/server/pkg/db/queries/supplement.sql @@ -14,7 +14,7 @@ WITH candidate AS MATERIALIZED ( SELECT id, workspace_id, issue_id, 'task-supplement-v1' FROM candidate WHERE @enable_task_supplement::boolean - AND provider IN ('codex', 'claude', 'grok', 'cursor') + AND provider IN ('codex', 'claude', 'grok', 'cursor', 'kiro') AND issue_id IS NOT NULL ON CONFLICT DO NOTHING RETURNING task_id @@ -45,7 +45,7 @@ WITH locked_task AS MATERIALIZED ( AND r.workspace_id = @workspace_id AND t.status = 'running' AND cap.capability = 'task-supplement-v1' - AND r.provider IN ('codex', 'claude', 'grok', 'cursor') + AND r.provider IN ('codex', 'claude', 'grok', 'cursor', 'kiro') FOR UPDATE OF t ), touched_issue AS ( UPDATE issue i SET @@ -109,7 +109,7 @@ WITH locked_task AS MATERIALIZED ( AND r.workspace_id = @workspace_id AND t.status = 'running' AND cap.capability = 'task-supplement-v1' - AND r.provider IN ('codex', 'claude', 'cursor') + AND r.provider IN ('codex', 'claude', 'grok', 'cursor', 'kiro') FOR UPDATE OF t ), inserted AS ( INSERT INTO task_supplement (