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 (