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 <[email protected]>

* 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 <[email protected]>

* 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 <[email protected]>

---------

Co-authored-by: multica-agent <[email protected]>
This commit is contained in:
Multica Eve
2026-10-21 19:01:56 +08:00
committed by GitHub
co-authored by multica-agent
parent 32a5a77a7a
commit 2b64569138
10 changed files with 285 additions and 13 deletions
+1 -1
View File
@@ -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.
+1 -1
View File
@@ -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) →** 只在智能体需要你拍板时提醒你,而不是每一步都来烦你。
@@ -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,
@@ -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) {
+90 -2
View File
@@ -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 {
+177
View File
@@ -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)
}
}
+5
View File
@@ -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.
+1
View File
@@ -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 {
+3 -3
View File
@@ -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
+3 -3
View File
@@ -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 (