feat(acp): support claude-acp Monitor tool and fix incremental tool_call updates (#698)
This commit is contained in:
Carlos Florêncio
@@ -73,6 +73,107 @@ func (e *emitter) plan(entries []acp.PlanEntry) {
|
||||
})
|
||||
}
|
||||
|
||||
// monitorClaudeMeta builds the `_meta.claudeCode.toolName=Monitor` payload
|
||||
// claude-agent-acp tags Monitor tool_call notifications with. Used by
|
||||
// e2e:monitor_* directives to reproduce the wire format exactly so the
|
||||
// kandev ACP adapter's Monitor recognizers fire the same way they do in
|
||||
// production.
|
||||
func monitorClaudeMeta() any {
|
||||
return map[string]any{"claudeCode": map[string]any{"toolName": "Monitor"}}
|
||||
}
|
||||
|
||||
// monitorClaudeMetaWithTask is like monitorClaudeMeta but also embeds a
|
||||
// `toolResponse.taskId` field — claude-agent-acp emits this on the
|
||||
// registration update.
|
||||
func monitorClaudeMetaWithTask(taskID string) any {
|
||||
return map[string]any{
|
||||
"claudeCode": map[string]any{
|
||||
"toolName": "Monitor",
|
||||
"toolResponse": map[string]any{"taskId": taskID},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// startMonitorTool emits the initial Monitor tool_call (status=pending) and
|
||||
// then a registration tool_call_update whose rawOutput banner carries the
|
||||
// taskID. This reproduces the two-frame wire pattern the kandev adapter
|
||||
// expects: a recognizable banner is the only signal it has that an apparent
|
||||
// "completed" really means "registered".
|
||||
//
|
||||
// The acp-go-sdk does not expose `WithStartMeta` / `WithUpdateMeta` helpers
|
||||
// at the time of writing, so we set Meta via a local option closure.
|
||||
func (e *emitter) startMonitorTool(id acp.ToolCallId, taskID, command string) {
|
||||
withStartMeta := func(meta any) acp.ToolCallStartOpt {
|
||||
return func(tc *acp.SessionUpdateToolCall) { tc.Meta = toMetaMap(meta) }
|
||||
}
|
||||
withUpdateMeta := func(meta any) acp.ToolCallUpdateOpt {
|
||||
return func(tu *acp.SessionToolCallUpdate) { tu.Meta = toMetaMap(meta) }
|
||||
}
|
||||
_ = e.conn.SessionUpdate(e.ctx, acp.SessionNotification{
|
||||
SessionId: e.sid,
|
||||
Update: acp.StartToolCall(id, "Monitor",
|
||||
acp.WithStartKind(acp.ToolKindOther),
|
||||
acp.WithStartStatus(acp.ToolCallStatusPending),
|
||||
acp.WithStartRawInput(map[string]any{"command": command}),
|
||||
withStartMeta(monitorClaudeMeta()),
|
||||
),
|
||||
})
|
||||
banner := fmt.Sprintf("Monitor started (task %s, timeout 60000ms). You will be notified on each event.", taskID)
|
||||
_ = e.conn.SessionUpdate(e.ctx, acp.SessionNotification{
|
||||
SessionId: e.sid,
|
||||
Update: acp.UpdateToolCall(id,
|
||||
acp.WithUpdateStatus(acp.ToolCallStatusCompleted),
|
||||
acp.WithUpdateRawOutput(banner),
|
||||
withUpdateMeta(monitorClaudeMetaWithTask(taskID)),
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
// toMetaMap is a small adapter so the local meta builders can return `any`
|
||||
// (matching the kandev-side recognizer signature) while the SDK's Meta field
|
||||
// requires `map[string]any` specifically.
|
||||
func toMetaMap(v any) map[string]any {
|
||||
if v == nil {
|
||||
return nil
|
||||
}
|
||||
if m, ok := v.(map[string]any); ok {
|
||||
return m
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// emitMonitorEvent reproduces the model-echoed task-notification envelope
|
||||
// that fires when a real Monitor's stdout produces a line. The kandev
|
||||
// adapter parses these out of agent_message_chunks and routes them back to
|
||||
// the originating Monitor's tool_call card.
|
||||
func (e *emitter) emitMonitorEvent(taskID, body string) {
|
||||
envelope := fmt.Sprintf(
|
||||
"Human: <task-notification>\n<task-id>%s</task-id>\n<event>%s</event>\n</task-notification>",
|
||||
taskID, body,
|
||||
)
|
||||
_ = e.conn.SessionUpdate(e.ctx, acp.SessionNotification{
|
||||
SessionId: e.sid,
|
||||
Update: acp.UpdateAgentMessageText(envelope),
|
||||
})
|
||||
}
|
||||
|
||||
// endMonitorTool emits the terminal Monitor tool_call_update — the kandev
|
||||
// adapter would normally synthesize this on its own at prompt-end, but tests
|
||||
// drive it explicitly so they don't have to wait for the prompt to finish.
|
||||
func (e *emitter) endMonitorTool(id acp.ToolCallId) {
|
||||
withUpdateMeta := func(meta any) acp.ToolCallUpdateOpt {
|
||||
return func(tu *acp.SessionToolCallUpdate) { tu.Meta = toMetaMap(meta) }
|
||||
}
|
||||
_ = e.conn.SessionUpdate(e.ctx, acp.SessionNotification{
|
||||
SessionId: e.sid,
|
||||
Update: acp.UpdateToolCall(id,
|
||||
acp.WithUpdateStatus(acp.ToolCallStatusCompleted),
|
||||
acp.WithUpdateRawOutput("Monitor exited"),
|
||||
withUpdateMeta(monitorClaudeMeta()),
|
||||
),
|
||||
})
|
||||
}
|
||||
|
||||
// requestPermission asks the client for permission to proceed with a tool call.
|
||||
// Returns true if permission was granted, false otherwise.
|
||||
func (e *emitter) requestPermission(toolCallID acp.ToolCallId, title string, kind acp.ToolKind, input any) bool {
|
||||
|
||||
@@ -55,9 +55,67 @@ func executeCommand(e *emitter, fullPrompt, line string) {
|
||||
|
||||
case strings.HasPrefix(line, "e2e:tool_result("):
|
||||
executeSimulatedToolResult(e, line)
|
||||
|
||||
case strings.HasPrefix(line, "e2e:monitor_start("):
|
||||
executeMonitorStart(e, line)
|
||||
|
||||
case strings.HasPrefix(line, "e2e:monitor_event("):
|
||||
executeMonitorEvent(e, line)
|
||||
|
||||
case strings.HasPrefix(line, "e2e:monitor_end("):
|
||||
executeMonitorEnd(e, line)
|
||||
}
|
||||
}
|
||||
|
||||
// executeMonitorStart emits a Monitor registration sequence reproducing the
|
||||
// claude-agent-acp wire format. Format: e2e:monitor_start("taskId", "command")
|
||||
//
|
||||
// The taskId is what subsequent monitor_event / monitor_end directives use
|
||||
// to look the Monitor back up. Command is recorded in the rawInput so the
|
||||
// frontend can label the card (e.g. "gh pr checks --watch").
|
||||
func executeMonitorStart(e *emitter, line string) {
|
||||
inner := extractParenContent(line, "e2e:monitor_start(")
|
||||
taskID, rest := extractFirstStringArg(inner)
|
||||
if taskID == "" {
|
||||
e.text("Script error: monitor_start requires taskId")
|
||||
return
|
||||
}
|
||||
rest = strings.TrimSpace(rest)
|
||||
rest = strings.TrimPrefix(rest, ",")
|
||||
command, _ := extractFirstStringArg(rest)
|
||||
toolID := nextToolID()
|
||||
state.monitorTools[taskID] = toolID
|
||||
e.startMonitorTool(toolID, taskID, command)
|
||||
}
|
||||
|
||||
// executeMonitorEvent emits a `<task-notification>` envelope as if the
|
||||
// model echoed an injected user turn. Format: e2e:monitor_event("taskId", "event body")
|
||||
func executeMonitorEvent(e *emitter, line string) {
|
||||
inner := extractParenContent(line, "e2e:monitor_event(")
|
||||
taskID, rest := extractFirstStringArg(inner)
|
||||
if taskID == "" {
|
||||
e.text("Script error: monitor_event requires taskId")
|
||||
return
|
||||
}
|
||||
rest = strings.TrimSpace(rest)
|
||||
rest = strings.TrimPrefix(rest, ",")
|
||||
body, _ := extractFirstStringArg(rest)
|
||||
e.emitMonitorEvent(taskID, body)
|
||||
}
|
||||
|
||||
// executeMonitorEnd emits the terminal tool_call_update for a previously
|
||||
// started Monitor. Format: e2e:monitor_end("taskId")
|
||||
func executeMonitorEnd(e *emitter, line string) {
|
||||
taskID := extractStringArg(line, "e2e:monitor_end(")
|
||||
tcID, ok := state.monitorTools[taskID]
|
||||
if !ok {
|
||||
e.text("Script error: monitor_end for unknown taskId " + taskID)
|
||||
return
|
||||
}
|
||||
delete(state.monitorTools, taskID)
|
||||
e.endMonitorTool(tcID)
|
||||
}
|
||||
|
||||
// executeMCPCommand parses and executes: e2e:mcp:<server>:<tool>(<json_args>)
|
||||
func executeMCPCommand(e *emitter, fullPrompt, line string) {
|
||||
rest := strings.TrimPrefix(line, "e2e:mcp:")
|
||||
|
||||
@@ -561,3 +561,133 @@ func TestExecuteScriptToolUseBadJSON(t *testing.T) {
|
||||
t.Errorf("expected script error message, got %q", text)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExecuteScriptMonitorStart asserts e2e:monitor_start emits the two-frame
|
||||
// claude-acp wire pattern: a pending tool_call followed by a registration
|
||||
// tool_call_update whose rawOutput banner carries the taskID and whose Meta
|
||||
// is tagged with claudeCode.toolName=Monitor.
|
||||
func TestExecuteScriptMonitorStart(t *testing.T) {
|
||||
e, mock := newTestEmitter()
|
||||
resetState()
|
||||
|
||||
executeScript(e, "", `e2e:monitor_start("task-7", "tail -f /var/log/x")`)
|
||||
|
||||
updates := mock.getUpdates()
|
||||
if len(updates) != 2 {
|
||||
t.Fatalf("expected 2 updates (start + registration), got %d", len(updates))
|
||||
}
|
||||
if !isToolCallUpdate(updates[0]) {
|
||||
t.Fatal("first update should be tool_call (pending)")
|
||||
}
|
||||
tc := updates[0].notification.Update.ToolCall
|
||||
if tc.Title != "Monitor" {
|
||||
t.Errorf("title = %q, want Monitor", tc.Title)
|
||||
}
|
||||
if !hasMonitorMeta(tc.Meta) {
|
||||
t.Errorf("first update missing claudeCode.toolName=Monitor meta: %+v", tc.Meta)
|
||||
}
|
||||
|
||||
if !isToolCallCompleteUpdate(updates[1]) {
|
||||
t.Fatal("second update should be tool_call_update (registration)")
|
||||
}
|
||||
tcu := updates[1].notification.Update.ToolCallUpdate
|
||||
if tcu.Status == nil || string(*tcu.Status) != "completed" {
|
||||
t.Errorf("registration status = %v, want completed", tcu.Status)
|
||||
}
|
||||
out, _ := tcu.RawOutput.(string)
|
||||
if !strings.HasPrefix(out, "Monitor started (task task-7,") {
|
||||
t.Errorf("rawOutput = %q, want banner starting with 'Monitor started (task task-7,'", out)
|
||||
}
|
||||
|
||||
// taskID must be remembered for subsequent monitor_event / monitor_end.
|
||||
if _, ok := state.monitorTools["task-7"]; !ok {
|
||||
t.Error("monitorTools missing entry for task-7")
|
||||
}
|
||||
}
|
||||
|
||||
// TestExecuteScriptMonitorEvent asserts e2e:monitor_event emits an
|
||||
// agent_message_chunk whose text matches the `<task-notification>` envelope
|
||||
// pattern the kandev adapter parses.
|
||||
func TestExecuteScriptMonitorEvent(t *testing.T) {
|
||||
e, mock := newTestEmitter()
|
||||
resetState()
|
||||
|
||||
executeScript(e, "", `e2e:monitor_event("task-7", "first event line")`)
|
||||
|
||||
updates := mock.getUpdates()
|
||||
if len(updates) != 1 {
|
||||
t.Fatalf("expected 1 update, got %d", len(updates))
|
||||
}
|
||||
if !isTextUpdate(updates[0]) {
|
||||
t.Fatal("expected text update for monitor event envelope")
|
||||
}
|
||||
text := getTextContent(updates[0])
|
||||
if !strings.Contains(text, "<task-id>task-7</task-id>") ||
|
||||
!strings.Contains(text, "<event>first event line</event>") {
|
||||
t.Errorf("envelope text = %q, want it to contain task-id and event tags", text)
|
||||
}
|
||||
}
|
||||
|
||||
// TestExecuteScriptMonitorEndUsesStoredToolCallID asserts e2e:monitor_end
|
||||
// resolves the taskID via state.monitorTools and emits a terminal
|
||||
// tool_call_update against the same toolCallID the start created.
|
||||
func TestExecuteScriptMonitorEndUsesStoredToolCallID(t *testing.T) {
|
||||
e, mock := newTestEmitter()
|
||||
resetState()
|
||||
|
||||
executeScript(e, "",
|
||||
"e2e:monitor_start(\"task-7\", \"x\")\n"+
|
||||
"e2e:monitor_end(\"task-7\")")
|
||||
|
||||
updates := mock.getUpdates()
|
||||
if len(updates) != 3 {
|
||||
t.Fatalf("expected 3 updates (start, registration, end), got %d", len(updates))
|
||||
}
|
||||
startTCID := updates[0].notification.Update.ToolCall.ToolCallId
|
||||
endUpd := updates[2].notification.Update.ToolCallUpdate
|
||||
if endUpd == nil {
|
||||
t.Fatal("third update should be tool_call_update (terminal)")
|
||||
}
|
||||
if endUpd.ToolCallId != startTCID {
|
||||
t.Errorf("end ToolCallId = %q, want %q (same as start)", endUpd.ToolCallId, startTCID)
|
||||
}
|
||||
if endUpd.Status == nil || string(*endUpd.Status) != "completed" {
|
||||
t.Errorf("end status = %v, want completed", endUpd.Status)
|
||||
}
|
||||
if _, ok := state.monitorTools["task-7"]; ok {
|
||||
t.Error("monitorTools should drop entry after monitor_end")
|
||||
}
|
||||
}
|
||||
|
||||
// TestExecuteScriptMonitorEndUnknownTaskIDReportsError asserts e2e:monitor_end
|
||||
// for a taskID that was never started emits a "Script error" text update
|
||||
// (matches the existing handling for malformed directives).
|
||||
func TestExecuteScriptMonitorEndUnknownTaskIDReportsError(t *testing.T) {
|
||||
e, mock := newTestEmitter()
|
||||
resetState()
|
||||
|
||||
executeScript(e, "", `e2e:monitor_end("does-not-exist")`)
|
||||
|
||||
updates := mock.getUpdates()
|
||||
if len(updates) != 1 {
|
||||
t.Fatalf("expected 1 update, got %d", len(updates))
|
||||
}
|
||||
text := getTextContent(updates[0])
|
||||
if !strings.Contains(text, "Script error") || !strings.Contains(text, "does-not-exist") {
|
||||
t.Errorf("expected script error referencing unknown taskId, got %q", text)
|
||||
}
|
||||
}
|
||||
|
||||
// hasMonitorMeta returns true if the SDK Meta map carries the
|
||||
// claudeCode.toolName=Monitor marker.
|
||||
func hasMonitorMeta(meta map[string]any) bool {
|
||||
if meta == nil {
|
||||
return false
|
||||
}
|
||||
cc, ok := meta["claudeCode"].(map[string]any)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
name, _ := cc["toolName"].(string)
|
||||
return name == "Monitor"
|
||||
}
|
||||
|
||||
@@ -11,14 +11,18 @@ import (
|
||||
type scriptState struct {
|
||||
toolCallCounter int
|
||||
lastToolID string
|
||||
// monitorTools maps user-supplied Monitor taskID -> generated toolCallID,
|
||||
// so e2e:monitor_event and e2e:monitor_end can reference an earlier
|
||||
// e2e:monitor_start by name without the script needing to track IDs.
|
||||
monitorTools map[string]acp.ToolCallId
|
||||
}
|
||||
|
||||
// state is the process-wide script state. Use resetState() in tests
|
||||
// to get a clean starting point instead of touching fields directly.
|
||||
var state = &scriptState{}
|
||||
var state = &scriptState{monitorTools: map[string]acp.ToolCallId{}}
|
||||
|
||||
func resetState() {
|
||||
state = &scriptState{}
|
||||
state = &scriptState{monitorTools: map[string]acp.ToolCallId{}}
|
||||
}
|
||||
|
||||
func nextToolID() acp.ToolCallId {
|
||||
|
||||
@@ -96,6 +96,14 @@ type Adapter struct {
|
||||
// Maps toolCallId -> NormalizedPayload so we can update with results
|
||||
activeToolCalls map[string]*streams.NormalizedPayload
|
||||
|
||||
// Active Monitor tools, keyed by sessionID -> taskID -> toolCallID.
|
||||
// Claude-acp's Monitor tool runs a background script that streams events
|
||||
// back to the LLM as `<task-notification>` envelopes. We hold this map so
|
||||
// later agent_message_chunks carrying those envelopes can be routed back
|
||||
// to the originating Monitor's tool_call card. Cleared on prompt completion
|
||||
// and rebuilt during session/load replay.
|
||||
activeMonitors map[string]map[string]string
|
||||
|
||||
// OTel tracing: active prompt span context.
|
||||
// Notification spans become children of the prompt span for visual grouping.
|
||||
promptTraceCtx context.Context
|
||||
@@ -130,6 +138,7 @@ func NewAdapter(cfg *shared.Config, log *logger.Logger) *Adapter {
|
||||
normalizer: NewNormalizer(),
|
||||
updatesCh: make(chan AgentEvent, 100),
|
||||
activeToolCalls: make(map[string]*streams.NormalizedPayload),
|
||||
activeMonitors: make(map[string]map[string]string),
|
||||
attachMgr: shared.NewAttachmentManager(cfg.WorkDir, l.Zap()),
|
||||
}
|
||||
}
|
||||
@@ -475,6 +484,11 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string, mcpServers
|
||||
a.isLoadingSession = false
|
||||
a.mu.Unlock()
|
||||
|
||||
// Any Monitor still tracked at this point was running in pre-restart history
|
||||
// but has no live process to back it now — emit synthetic cancellations so
|
||||
// the frontend doesn't render a stuck "watching" card.
|
||||
a.sweepMonitorsOnReplayEnd(sessionID)
|
||||
|
||||
if replayPlan != nil {
|
||||
entries := make([]PlanEntry, len(replayPlan.Entries))
|
||||
for i, e := range replayPlan.Entries {
|
||||
@@ -615,6 +629,12 @@ func (a *Adapter) Prompt(ctx context.Context, message string, attachments []v1.M
|
||||
// tool_call without a terminal status update from the agent).
|
||||
a.cancelActiveToolCalls(sessionID)
|
||||
|
||||
// Mark any tracked Monitors as ended. They live longer than a typical tool
|
||||
// call (the script keeps running across model turns), so this sweep runs
|
||||
// after `cancelActiveToolCalls` to give the Monitor card a clean terminal
|
||||
// state when the parent prompt completes naturally.
|
||||
a.sweepMonitorsOnPromptEnd(sessionID)
|
||||
|
||||
// Emit complete event via the stream, including the StopReason from the agent.
|
||||
// This normalizes ACP behavior to match other adapters (stream-json, amp, copilot, opencode).
|
||||
a.logger.Debug("emitting complete event after prompt",
|
||||
@@ -674,13 +694,31 @@ func (a *Adapter) Cancel(ctx context.Context) error {
|
||||
|
||||
// cancelActiveToolCalls emits cancelled tool_update events for all in-flight tool calls
|
||||
// and clears the activeToolCalls map.
|
||||
//
|
||||
// Monitor tool calls are intentionally skipped here — they are tracked in
|
||||
// activeMonitors and given their own terminal sweep (sweepMonitorsOnPromptEnd
|
||||
// or sweepMonitorsOnReplayEnd) which uses the appropriate status and a
|
||||
// payload snapshot. Without this skip the Monitor would receive two
|
||||
// terminal events with conflicting states.
|
||||
func (a *Adapter) cancelActiveToolCalls(sessionID string) {
|
||||
a.mu.Lock()
|
||||
toolCalls := a.activeToolCalls
|
||||
a.activeToolCalls = make(map[string]*streams.NormalizedPayload)
|
||||
monitorToolCallIDs := make(map[string]bool)
|
||||
for _, tcID := range a.activeMonitors[sessionID] {
|
||||
monitorToolCallIDs[tcID] = true
|
||||
}
|
||||
toCancel := make(map[string]*streams.NormalizedPayload)
|
||||
preserved := make(map[string]*streams.NormalizedPayload)
|
||||
for tcID, payload := range a.activeToolCalls {
|
||||
if monitorToolCallIDs[tcID] {
|
||||
preserved[tcID] = payload
|
||||
} else {
|
||||
toCancel[tcID] = payload
|
||||
}
|
||||
}
|
||||
a.activeToolCalls = preserved
|
||||
a.mu.Unlock()
|
||||
|
||||
for toolCallID, normalized := range toolCalls {
|
||||
for toolCallID, normalized := range toCancel {
|
||||
a.logger.Debug("cancelling active tool call",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("tool_call_id", toolCallID))
|
||||
@@ -823,6 +861,13 @@ func (a *Adapter) handleACPUpdate(n acp.SessionNotification) {
|
||||
}
|
||||
a.mu.Unlock()
|
||||
|
||||
// Even though we suppress replay notifications from reaching clients,
|
||||
// reconstruct any in-progress Monitor registrations so the post-replay
|
||||
// sweep can mark them ended-on-restart. Without this, a session where
|
||||
// Monitor was running before agentctl died would keep showing a
|
||||
// "watching" card forever after resume.
|
||||
a.captureReplayMonitor(string(n.SessionId), u)
|
||||
|
||||
// Suppress conversation history events during load.
|
||||
// AvailableCommandsUpdate is intentionally NOT suppressed — it may arrive
|
||||
// after the replay completes as a "ready" signal, and the frontend treats
|
||||
@@ -986,7 +1031,23 @@ func (a *Adapter) convertMessageChunk(sessionID string, content acp.ContentBlock
|
||||
|
||||
// Text content goes directly into the Text field for backward compatibility
|
||||
if content.Text != nil {
|
||||
event.Text = content.Text.Text
|
||||
text := content.Text.Text
|
||||
// Claude-acp's Monitor tool injects each script line back to the model
|
||||
// as a `<task-notification>` user turn. The wrapper suppresses the
|
||||
// user_message_chunk so the model often "echoes" the envelope into its
|
||||
// own assistant text. Parse those out into proper Monitor events and
|
||||
// strip them from the chat text. Assistant role only — genuine user
|
||||
// messages don't carry these.
|
||||
if role == "assistant" {
|
||||
text = a.routeMonitorEvents(sessionID, text)
|
||||
if isMonitorHumanEcho(text) {
|
||||
return nil
|
||||
}
|
||||
if strings.TrimSpace(text) == "" {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
event.Text = text
|
||||
return event
|
||||
}
|
||||
|
||||
@@ -999,6 +1060,38 @@ func (a *Adapter) convertMessageChunk(sessionID string, content acp.ContentBlock
|
||||
return event
|
||||
}
|
||||
|
||||
// routeMonitorEvents extracts Monitor `<task-notification>` envelopes from an
|
||||
// agent_message_chunk text, emits a synthetic tool_call_update for each event
|
||||
// against the originating Monitor's toolCallID, and returns the cleaned text.
|
||||
// Returns the original text unchanged when no envelope is present (the common
|
||||
// case for non-Monitor sessions).
|
||||
func (a *Adapter) routeMonitorEvents(sessionID, text string) string {
|
||||
cleaned, events := extractMonitorEvents(text)
|
||||
if len(events) == 0 {
|
||||
return text
|
||||
}
|
||||
for _, ev := range events {
|
||||
toolCallID, ok := a.lookupMonitorByTaskID(sessionID, ev.TaskID)
|
||||
if !ok {
|
||||
a.logger.Debug("monitor event for unknown task, dropping envelope and event body",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", ev.TaskID))
|
||||
continue
|
||||
}
|
||||
a.mu.Lock()
|
||||
payload := a.activeToolCalls[toolCallID]
|
||||
appendMonitorEvent(payload, ev.TaskID, monitorCommandFromPayload(payload), ev.Body)
|
||||
a.mu.Unlock()
|
||||
a.sendUpdate(monitorEventEvent(sessionID, toolCallID, ev.Body, payload))
|
||||
a.logger.Debug("monitor event routed",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", ev.TaskID),
|
||||
zap.String("tool_call_id", toolCallID),
|
||||
zap.Int("body_len", len(ev.Body)))
|
||||
}
|
||||
return cleaned
|
||||
}
|
||||
|
||||
// convertAvailableCommands converts an ACP AvailableCommandsUpdate to an AgentEvent,
|
||||
// including input hints when available.
|
||||
func (a *Adapter) convertAvailableCommands(sessionID string, update *acp.SessionAvailableCommandsUpdate) *AgentEvent {
|
||||
@@ -1371,6 +1464,38 @@ func (a *Adapter) convertToolCallResultUpdate(sessionID string, tcu *acp.Session
|
||||
if status == "completed" {
|
||||
status = toolStatusComplete
|
||||
}
|
||||
// Claude-acp sends incremental updates (title, rawInput, content) with no
|
||||
// Status field — e.g. the second tool_call_update for Bash carries the actual
|
||||
// command and human-readable title. The orchestrator only persists updates
|
||||
// with a known status, so without a synthesized "in_progress" here those
|
||||
// fields are silently dropped and the message stays on the placeholder
|
||||
// "Terminal" title from the initial pending tool_call.
|
||||
if status == "" && (tcu.Title != nil || tcu.RawInput != nil || len(tcu.Content) > 0) {
|
||||
status = "in_progress"
|
||||
}
|
||||
|
||||
// Recognize Monitor registration: claude-acp sends `tool_call_update` with
|
||||
// status="completed" and a `Monitor started (task X, …)` rawOutput about a
|
||||
// second after the Monitor starts. That status is misleading — the Monitor
|
||||
// itself is just beginning. Override to "in_progress" so the card stays
|
||||
// open, and remember taskID -> toolCallID so subsequent task-notification
|
||||
// envelopes can route their events back to this card.
|
||||
monitorTaskID, isMonitorRegistration := recognizeMonitorRegistration(tcu.Meta, tcu.RawOutput)
|
||||
if isMonitorRegistration && status == toolStatusComplete {
|
||||
a.trackMonitor(sessionID, monitorTaskID, toolCallID)
|
||||
status = "in_progress"
|
||||
a.logger.Info("monitor registered",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", monitorTaskID),
|
||||
zap.String("tool_call_id", toolCallID))
|
||||
}
|
||||
|
||||
// A terminal tool_call_update for an already-tracked Monitor (the agent
|
||||
// proactively ended the watch). NormalizeToolResult would otherwise stomp
|
||||
// the `{monitor: …}` view in Generic.Output with the raw string body, so
|
||||
// we suppress the normalize call and let the closing-out logic below mark
|
||||
// the view as ended instead.
|
||||
isTrackedMonitorTerminal := !isMonitorRegistration && isMonitorMeta(tcu.Meta) && a.isTrackedMonitor(sessionID, toolCallID)
|
||||
|
||||
isTerminal := status == toolStatusComplete || status == toolStatusError || status == "cancelled"
|
||||
|
||||
@@ -1383,11 +1508,29 @@ func (a *Adapter) convertToolCallResultUpdate(sessionID string, tcu *acp.Session
|
||||
a.normalizer.UpdatePayloadInput(payload, tcu.RawInput)
|
||||
}
|
||||
|
||||
// Update stored payload with tool result output
|
||||
if tcu.RawOutput != nil && payload != nil {
|
||||
// Update stored payload with tool result output. Skip for tracked-Monitor
|
||||
// terminal updates so Generic.Output stays the structured `{monitor: …}`
|
||||
// view rather than getting clobbered by the rawOutput string.
|
||||
if tcu.RawOutput != nil && payload != nil && !isTrackedMonitorTerminal {
|
||||
a.normalizer.NormalizeToolResult(payload, tcu.RawOutput)
|
||||
}
|
||||
|
||||
// Seed the Monitor view AFTER NormalizeToolResult so we overwrite the
|
||||
// banner string the normalizer just stuffed into Generic.Output. The
|
||||
// Monitor card detects itself by `output.monitor` presence — the banner
|
||||
// would shadow it and the frontend would render this as a generic
|
||||
// tool_call instead.
|
||||
if isMonitorRegistration && payload != nil {
|
||||
seedMonitorView(payload, monitorTaskID, monitorCommandFromPayload(payload))
|
||||
}
|
||||
|
||||
// Preserve and mark-ended the Monitor view on tracked-Monitor terminal
|
||||
// updates so the card flips from "watching" to "ended" without losing
|
||||
// the accumulated event count or recent-events tail.
|
||||
if isTrackedMonitorTerminal && payload != nil {
|
||||
markMonitorEnded(payload, "exited")
|
||||
}
|
||||
|
||||
// Enrich modify_file payload from tool_call_contents.
|
||||
// Claude ACP sends path and content in tool_call_update, not in the initial tool_call.
|
||||
if payload != nil && payload.Kind() == streams.ToolKindModifyFile {
|
||||
@@ -1405,6 +1548,12 @@ func (a *Adapter) convertToolCallResultUpdate(sessionID string, tcu *acp.Session
|
||||
|
||||
if isTerminal {
|
||||
delete(a.activeToolCalls, toolCallID)
|
||||
// Also drop tracked Monitor: this terminal update is the
|
||||
// agent-emitted close, so the prompt-end sweep must not re-emit a
|
||||
// "Monitor exited" event for this same toolCallID.
|
||||
if isTrackedMonitorTerminal {
|
||||
a.dropMonitorByToolCallIDLocked(sessionID, toolCallID)
|
||||
}
|
||||
}
|
||||
a.mu.Unlock()
|
||||
|
||||
|
||||
@@ -0,0 +1,545 @@
|
||||
package acp
|
||||
|
||||
import (
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
acp "github.com/coder/acp-go-sdk"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
)
|
||||
|
||||
// monitorToolName is the literal toolName Claude-acp tags Monitor tool calls with
|
||||
// in `_meta.claudeCode.toolName`. Used to recognize Monitor across the lifecycle.
|
||||
const monitorToolName = "Monitor"
|
||||
|
||||
// monitorRegistrationOutputPrefix identifies the rawOutput banner Claude-acp
|
||||
// emits when a Monitor registers (~1s after start). The wrapper sets status to
|
||||
// "completed" on this update, but the Monitor itself is only just starting —
|
||||
// we override that status downstream.
|
||||
const monitorRegistrationOutputPrefix = "Monitor started (task "
|
||||
|
||||
// monitorRegistrationRE captures the taskId out of the registration banner.
|
||||
// Banner format: `Monitor started (task <taskId>, timeout <ms>ms)…`
|
||||
var monitorRegistrationRE = regexp.MustCompile(`^Monitor started \(task ([a-zA-Z0-9_-]+),`)
|
||||
|
||||
// monitorTaskNotifRE matches the `<task-notification>` envelope claude-agent-acp's
|
||||
// model auto-completes when a Monitor event fires. The model leaks these as
|
||||
// agent_message_chunk text because the wrapper suppresses the corresponding
|
||||
// user_message_chunk. Capture groups: 1=taskId, 2=event body.
|
||||
//
|
||||
// Pattern is intentionally permissive: the outer `Human:` prefix is optional
|
||||
// (some chunks split across multiple deltas), we match across newlines, and
|
||||
// the event body uses lazy `(.*?)` rather than `[^<]*` so script lines that
|
||||
// contain `<` characters (e.g. `<error>build failed</error>`, ANSI escape
|
||||
// fragments, shell redirection text) still match instead of leaking the
|
||||
// raw envelope into the chat.
|
||||
var monitorTaskNotifRE = regexp.MustCompile(
|
||||
`(?s)(?:Human:\s*)?<task-notification>\s*<task-id>([^<]+)</task-id>.*?<event>(.*?)</event>\s*</task-notification>`,
|
||||
)
|
||||
|
||||
// monitorHumanEchoRE matches an orphan "Human:" prefix with no closing
|
||||
// `</task-notification>` — i.e. the model started echoing an envelope but the
|
||||
// chunk got cut off. Conservative pattern: drop only when the chunk is just the
|
||||
// prefix (possibly with a partial `<…` opener), so genuine assistant messages
|
||||
// that happen to mention "Human:" survive.
|
||||
var monitorHumanEchoRE = regexp.MustCompile(`^Human:\s*(<[^>]*)?\s*$`)
|
||||
|
||||
// recognizeMonitorRegistration inspects a tool_call_update for the Monitor
|
||||
// registration banner. Returns the parsed taskID and true when the update
|
||||
// represents a registration (not a real completion). The caller is expected to
|
||||
// override the outgoing status from "completed" to "in_progress" so the
|
||||
// Monitor card stays open in the UI.
|
||||
func recognizeMonitorRegistration(meta map[string]any, rawOutput any) (string, bool) {
|
||||
if !isMonitorMeta(meta) {
|
||||
return "", false
|
||||
}
|
||||
out, ok := rawOutput.(string)
|
||||
if !ok || !strings.HasPrefix(out, monitorRegistrationOutputPrefix) {
|
||||
return "", false
|
||||
}
|
||||
matches := monitorRegistrationRE.FindStringSubmatch(out)
|
||||
if len(matches) != 2 {
|
||||
return "", false
|
||||
}
|
||||
return matches[1], true
|
||||
}
|
||||
|
||||
// isMonitorMeta returns true if the ACP `_meta.claudeCode.toolName` field
|
||||
// equals "Monitor". The meta map is shaped untyped over the wire (`{claudeCode:
|
||||
// {toolName: …}}`), so we walk it defensively.
|
||||
func isMonitorMeta(meta map[string]any) bool {
|
||||
if meta == nil {
|
||||
return false
|
||||
}
|
||||
cc, ok := meta["claudeCode"].(map[string]any)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
name, _ := cc["toolName"].(string)
|
||||
return name == monitorToolName
|
||||
}
|
||||
|
||||
// recognizeMonitorTaskCall inspects a `tool_call` notification's metadata and
|
||||
// rawInput. Returns the watched command (best-effort) when the call is a
|
||||
// Monitor invocation, plus true. Used during replay to seed activeMonitors
|
||||
// before the matching registration update arrives.
|
||||
func recognizeMonitorTaskCall(tc *acp.SessionUpdateToolCall) (string, bool) {
|
||||
if tc == nil || !isMonitorMeta(tc.Meta) {
|
||||
return "", false
|
||||
}
|
||||
cmd := ""
|
||||
if input, ok := tc.RawInput.(map[string]any); ok {
|
||||
if c, ok := input["command"].(string); ok {
|
||||
cmd = c
|
||||
}
|
||||
}
|
||||
return cmd, true
|
||||
}
|
||||
|
||||
// monitorCommandFromPayload extracts the watched command string from a
|
||||
// Generic tool payload's Input field. The ACP normalizer stores the entire
|
||||
// args map (including a `raw_input` key) under Input, so the command lives
|
||||
// at `Input.raw_input.command` for Generic tools — not at a top-level key.
|
||||
// Returns empty string when the path is missing or the wrong shape.
|
||||
func monitorCommandFromPayload(payload *streams.NormalizedPayload) string {
|
||||
if payload == nil || payload.Generic() == nil {
|
||||
return ""
|
||||
}
|
||||
args, ok := payload.Generic().Input.(map[string]any)
|
||||
if !ok {
|
||||
return ""
|
||||
}
|
||||
rawInput, ok := args["raw_input"].(map[string]any)
|
||||
if !ok {
|
||||
// Fallback: some agents pass command at the top level.
|
||||
cmd, _ := args["command"].(string)
|
||||
return cmd
|
||||
}
|
||||
cmd, _ := rawInput["command"].(string)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// isTrackedMonitor returns true if any taskID -> toolCallID mapping in the
|
||||
// session's activeMonitors entry equals the given toolCallID. Used by
|
||||
// convertToolCallResultUpdate to recognize agent-emitted terminal updates
|
||||
// for Monitors that we already know about, so the rawOutput string doesn't
|
||||
// stomp the structured Monitor view in Generic.Output.
|
||||
func (a *Adapter) isTrackedMonitor(sessionID, toolCallID string) bool {
|
||||
a.mu.RLock()
|
||||
defer a.mu.RUnlock()
|
||||
for _, tcID := range a.activeMonitors[sessionID] {
|
||||
if tcID == toolCallID {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// dropMonitorByToolCallIDLocked is the same as dropMonitorByToolCallID but
|
||||
// the caller already holds Adapter.mu. Used inside convertToolCallResultUpdate
|
||||
// where the lock is already held for the duration of the payload update
|
||||
// block.
|
||||
func (a *Adapter) dropMonitorByToolCallIDLocked(sessionID, toolCallID string) {
|
||||
monitors := a.activeMonitors[sessionID]
|
||||
if monitors == nil {
|
||||
return
|
||||
}
|
||||
for taskID, tcID := range monitors {
|
||||
if tcID == toolCallID {
|
||||
delete(monitors, taskID)
|
||||
}
|
||||
}
|
||||
if len(monitors) == 0 {
|
||||
delete(a.activeMonitors, sessionID)
|
||||
}
|
||||
}
|
||||
|
||||
// trackMonitor records a registered Monitor against a session.
|
||||
func (a *Adapter) trackMonitor(sessionID, taskID, toolCallID string) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
monitors := a.activeMonitors[sessionID]
|
||||
if monitors == nil {
|
||||
monitors = make(map[string]string)
|
||||
a.activeMonitors[sessionID] = monitors
|
||||
}
|
||||
monitors[taskID] = toolCallID
|
||||
}
|
||||
|
||||
// lookupMonitorByTaskID resolves a taskID to its toolCallID for a session.
|
||||
// Returns the toolCallID and true if the Monitor is currently tracked.
|
||||
func (a *Adapter) lookupMonitorByTaskID(sessionID, taskID string) (string, bool) {
|
||||
a.mu.RLock()
|
||||
defer a.mu.RUnlock()
|
||||
monitors := a.activeMonitors[sessionID]
|
||||
if monitors == nil {
|
||||
return "", false
|
||||
}
|
||||
tcID, ok := monitors[taskID]
|
||||
return tcID, ok
|
||||
}
|
||||
|
||||
// takeActiveMonitors atomically removes and returns the active Monitor map
|
||||
// for a session. Used by the prompt-end sweep and the post-replay restart
|
||||
// sweep, both of which need to drain the map to emit terminal updates.
|
||||
func (a *Adapter) takeActiveMonitors(sessionID string) map[string]string {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
monitors := a.activeMonitors[sessionID]
|
||||
delete(a.activeMonitors, sessionID)
|
||||
return monitors
|
||||
}
|
||||
|
||||
// extractMonitorEvents parses every `<task-notification>` envelope out of an
|
||||
// agent_message_chunk text, replaces each match with empty string, and
|
||||
// returns the cleaned text plus the parsed events in order.
|
||||
//
|
||||
// Each event carries the matched taskID (so the caller can route to the right
|
||||
// Monitor's toolCallID) and the event body (the actual line stdout produced).
|
||||
func extractMonitorEvents(text string) (cleaned string, events []monitorEvent) {
|
||||
matches := monitorTaskNotifRE.FindAllStringSubmatchIndex(text, -1)
|
||||
if len(matches) == 0 {
|
||||
return text, nil
|
||||
}
|
||||
events = make([]monitorEvent, 0, len(matches))
|
||||
var b strings.Builder
|
||||
prev := 0
|
||||
for _, m := range matches {
|
||||
// m: [matchStart, matchEnd, taskIdStart, taskIdEnd, eventStart, eventEnd]
|
||||
b.WriteString(text[prev:m[0]])
|
||||
events = append(events, monitorEvent{
|
||||
TaskID: text[m[2]:m[3]],
|
||||
Body: text[m[4]:m[5]],
|
||||
})
|
||||
prev = m[1]
|
||||
}
|
||||
b.WriteString(text[prev:])
|
||||
return b.String(), events
|
||||
}
|
||||
|
||||
// isMonitorHumanEcho returns true when text is a stripped, orphan "Human:"
|
||||
// prefix the model emitted while trying to echo a task-notification envelope.
|
||||
// Such chunks should be dropped to keep them out of the chat UI.
|
||||
func isMonitorHumanEcho(text string) bool {
|
||||
trimmed := strings.TrimSpace(text)
|
||||
if trimmed == "" {
|
||||
return false
|
||||
}
|
||||
return monitorHumanEchoRE.MatchString(trimmed)
|
||||
}
|
||||
|
||||
// monitorEvent represents one Monitor stdout line decoded from a
|
||||
// `<task-notification>` envelope.
|
||||
type monitorEvent struct {
|
||||
TaskID string
|
||||
Body string
|
||||
}
|
||||
|
||||
// monitorPayloadCap bounds how many recent event lines we attach to the
|
||||
// Monitor's NormalizedPayload. Keeping this small keeps message metadata
|
||||
// reasonable for long-running Monitors (a CI watch can fire hundreds of
|
||||
// events) while still giving the user a tail of activity in the UI.
|
||||
const monitorPayloadCap = 10
|
||||
|
||||
// monitorPayloadView is the structured shape we tuck into the Generic tool
|
||||
// payload's Output field for Monitor tools. It tells the frontend how to
|
||||
// render the card (count, recent tail, command) without needing the
|
||||
// adapter to invent a new NormalizedPayload kind.
|
||||
type monitorPayloadView struct {
|
||||
Kind string `json:"kind"`
|
||||
TaskID string `json:"task_id"`
|
||||
Command string `json:"command,omitempty"`
|
||||
EventCount int `json:"event_count"`
|
||||
RecentEvents []string `json:"recent_events,omitempty"`
|
||||
Ended bool `json:"ended,omitempty"`
|
||||
EndReason string `json:"end_reason,omitempty"`
|
||||
}
|
||||
|
||||
// seedMonitorView writes an empty Monitor view onto the Generic payload at
|
||||
// registration time so the frontend can switch the card renderer immediately
|
||||
// — before any events fire. Without this seed a CI watch with a long quiet
|
||||
// startup would render as a generic tool_call for minutes.
|
||||
func seedMonitorView(payload *streams.NormalizedPayload, taskID, command string) {
|
||||
if payload == nil {
|
||||
return
|
||||
}
|
||||
g := payload.Generic()
|
||||
if g == nil {
|
||||
return
|
||||
}
|
||||
view := readMonitorView(g)
|
||||
if view.TaskID == "" {
|
||||
view.TaskID = taskID
|
||||
}
|
||||
if view.Command == "" {
|
||||
view.Command = command
|
||||
}
|
||||
g.Output = monitorOutputWrapper(view)
|
||||
}
|
||||
|
||||
// updateMonitorPayloadView mutates the Monitor tool's NormalizedPayload to
|
||||
// reflect a new event (or a terminal state). Returns the same payload pointer
|
||||
// so callers can pass it back into the synthetic AgentEvent.
|
||||
//
|
||||
// We piggyback on the Generic payload that was created by `normalizeGeneric`
|
||||
// when the original tool_call (kind="other") arrived. The adapter's existing
|
||||
// activeToolCalls map already holds it.
|
||||
func appendMonitorEvent(payload *streams.NormalizedPayload, taskID, command, body string) *streams.NormalizedPayload {
|
||||
if payload == nil {
|
||||
return nil
|
||||
}
|
||||
g := payload.Generic()
|
||||
if g == nil {
|
||||
// If somebody upstream changed the tool_call kind we won't have a
|
||||
// Generic payload — leave the payload alone rather than reconstructing.
|
||||
return payload
|
||||
}
|
||||
view := readMonitorView(g)
|
||||
if view.TaskID == "" {
|
||||
view.TaskID = taskID
|
||||
}
|
||||
if view.Command == "" {
|
||||
view.Command = command
|
||||
}
|
||||
view.EventCount++
|
||||
view.RecentEvents = appendCapped(view.RecentEvents, body, monitorPayloadCap)
|
||||
g.Output = monitorOutputWrapper(view)
|
||||
return payload
|
||||
}
|
||||
|
||||
// markMonitorEnded sets the terminal flag and reason on the Monitor view
|
||||
// without bumping the event counter.
|
||||
func markMonitorEnded(payload *streams.NormalizedPayload, reason string) *streams.NormalizedPayload {
|
||||
if payload == nil {
|
||||
return nil
|
||||
}
|
||||
g := payload.Generic()
|
||||
if g == nil {
|
||||
return payload
|
||||
}
|
||||
view := readMonitorView(g)
|
||||
view.Ended = true
|
||||
view.EndReason = reason
|
||||
g.Output = monitorOutputWrapper(view)
|
||||
return payload
|
||||
}
|
||||
|
||||
// readMonitorView extracts the existing Monitor view from a Generic payload
|
||||
// (the wrapper persists across tool_call_updates because the adapter mutates
|
||||
// the same payload pointer). Returns a zero-valued view when no Monitor data
|
||||
// is attached yet.
|
||||
func readMonitorView(g *streams.GenericPayload) monitorPayloadView {
|
||||
if g == nil || g.Output == nil {
|
||||
return monitorPayloadView{Kind: monitorToolName}
|
||||
}
|
||||
wrapper, ok := g.Output.(map[string]any)
|
||||
if !ok {
|
||||
return monitorPayloadView{Kind: monitorToolName}
|
||||
}
|
||||
raw, ok := wrapper["monitor"].(map[string]any)
|
||||
if !ok {
|
||||
return monitorPayloadView{Kind: monitorToolName}
|
||||
}
|
||||
view := monitorPayloadView{Kind: monitorToolName}
|
||||
if s, ok := raw["task_id"].(string); ok {
|
||||
view.TaskID = s
|
||||
}
|
||||
if s, ok := raw["command"].(string); ok {
|
||||
view.Command = s
|
||||
}
|
||||
if n, ok := raw["event_count"].(float64); ok {
|
||||
view.EventCount = int(n)
|
||||
} else if n, ok := raw["event_count"].(int); ok {
|
||||
view.EventCount = n
|
||||
}
|
||||
if list, ok := raw["recent_events"].([]any); ok {
|
||||
view.RecentEvents = make([]string, 0, len(list))
|
||||
for _, item := range list {
|
||||
if s, ok := item.(string); ok {
|
||||
view.RecentEvents = append(view.RecentEvents, s)
|
||||
}
|
||||
}
|
||||
} else if list, ok := raw["recent_events"].([]string); ok {
|
||||
view.RecentEvents = append([]string{}, list...)
|
||||
}
|
||||
if b, ok := raw["ended"].(bool); ok {
|
||||
view.Ended = b
|
||||
}
|
||||
if s, ok := raw["end_reason"].(string); ok {
|
||||
view.EndReason = s
|
||||
}
|
||||
return view
|
||||
}
|
||||
|
||||
// monitorOutputWrapper boxes the typed view in the `{monitor: {...}}` shape
|
||||
// the frontend monitor card reads.
|
||||
func monitorOutputWrapper(view monitorPayloadView) map[string]any {
|
||||
return map[string]any{"monitor": map[string]any{
|
||||
"kind": view.Kind,
|
||||
"task_id": view.TaskID,
|
||||
"command": view.Command,
|
||||
"event_count": view.EventCount,
|
||||
"recent_events": view.RecentEvents,
|
||||
"ended": view.Ended,
|
||||
"end_reason": view.EndReason,
|
||||
}}
|
||||
}
|
||||
|
||||
// appendCapped appends s to xs, keeping at most cap most-recent entries.
|
||||
func appendCapped(xs []string, s string, capacity int) []string {
|
||||
xs = append(xs, s)
|
||||
if len(xs) > capacity {
|
||||
return xs[len(xs)-capacity:]
|
||||
}
|
||||
return xs
|
||||
}
|
||||
|
||||
// monitorEventEvent constructs a synthetic tool_call_update AgentEvent for a
|
||||
// Monitor event. The frontend appends the body line to the Monitor's content
|
||||
// stream and renders an updated event count.
|
||||
func monitorEventEvent(sessionID, toolCallID string, body string, payload *streams.NormalizedPayload) AgentEvent {
|
||||
return AgentEvent{
|
||||
Type: streams.EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
ToolCallID: toolCallID,
|
||||
ToolStatus: "in_progress",
|
||||
NormalizedPayload: payload,
|
||||
ToolCallContents: []streams.ToolCallContentItem{
|
||||
{Type: "content", Content: &streams.ContentBlock{Type: "text", Text: body}},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// captureReplayMonitor walks a single replayed session/update during
|
||||
// session/load history replay and registers any Monitor it finds in
|
||||
// `activeMonitors`. The post-replay sweep then knows which Monitors to mark
|
||||
// ended-on-restart. This is purely an internal capture path — the replay
|
||||
// notifications themselves are still suppressed from reaching clients.
|
||||
//
|
||||
// Two notification shapes can register a Monitor:
|
||||
// - `tool_call` with `_meta.claudeCode.toolName == "Monitor"` (records by toolCallID;
|
||||
// we don't yet know taskID until the registration update arrives)
|
||||
// - `tool_call_update` whose `rawOutput` matches the registration banner
|
||||
// (records the taskID -> toolCallID mapping)
|
||||
//
|
||||
// A `tool_call_update` whose status is terminal (`completed` with non-banner
|
||||
// rawOutput, `failed`, `cancelled`) means the Monitor already ended in
|
||||
// history — drop it from the map.
|
||||
func (a *Adapter) captureReplayMonitor(sessionID string, u acp.SessionUpdate) {
|
||||
if tc := u.ToolCall; tc != nil {
|
||||
if _, ok := recognizeMonitorTaskCall(tc); ok {
|
||||
a.trackMonitor(sessionID, replayPendingTaskID(string(tc.ToolCallId)), string(tc.ToolCallId))
|
||||
}
|
||||
}
|
||||
if tcu := u.ToolCallUpdate; tcu != nil {
|
||||
toolCallID := string(tcu.ToolCallId)
|
||||
if taskID, ok := recognizeMonitorRegistration(tcu.Meta, tcu.RawOutput); ok {
|
||||
// Replace any pending-by-toolCallID placeholder with the real taskID.
|
||||
a.replaceMonitorPendingKey(sessionID, toolCallID, taskID)
|
||||
a.trackMonitor(sessionID, taskID, toolCallID)
|
||||
return
|
||||
}
|
||||
// Terminal update for a tracked Monitor — drop it from the map so the
|
||||
// post-replay sweep doesn't double-emit a cancellation.
|
||||
if tcu.Status != nil {
|
||||
s := string(*tcu.Status)
|
||||
if s == "completed" || s == "failed" || s == "cancelled" {
|
||||
a.dropMonitorByToolCallID(sessionID, toolCallID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// replayPendingTaskID synthesizes a placeholder taskID for a Monitor we saw
|
||||
// during replay before the registration banner arrived. It's namespaced by
|
||||
// toolCallID so it can never collide with a real taskID. Replaced in place
|
||||
// once the banner reveals the real taskID via replaceMonitorPendingKey.
|
||||
func replayPendingTaskID(toolCallID string) string {
|
||||
return "_pending_" + toolCallID
|
||||
}
|
||||
|
||||
// replaceMonitorPendingKey rewrites an entry in activeMonitors that was
|
||||
// registered during replay with a placeholder taskID once the real taskID
|
||||
// arrives via the registration banner.
|
||||
func (a *Adapter) replaceMonitorPendingKey(sessionID, toolCallID, realTaskID string) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
monitors := a.activeMonitors[sessionID]
|
||||
if monitors == nil {
|
||||
return
|
||||
}
|
||||
delete(monitors, replayPendingTaskID(toolCallID))
|
||||
monitors[realTaskID] = toolCallID
|
||||
}
|
||||
|
||||
// dropMonitorByToolCallID removes any entry in activeMonitors mapped to the
|
||||
// given toolCallID. Used during replay to discard Monitors that already had
|
||||
// a terminal status in history.
|
||||
func (a *Adapter) dropMonitorByToolCallID(sessionID, toolCallID string) {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
monitors := a.activeMonitors[sessionID]
|
||||
if monitors == nil {
|
||||
return
|
||||
}
|
||||
for taskID, tcID := range monitors {
|
||||
if tcID == toolCallID {
|
||||
delete(monitors, taskID)
|
||||
}
|
||||
}
|
||||
if len(monitors) == 0 {
|
||||
delete(a.activeMonitors, sessionID)
|
||||
}
|
||||
}
|
||||
|
||||
// sweepMonitorsOnPromptEnd emits a terminal `complete` tool_call_update for
|
||||
// every tracked Monitor in a session and clears the map. Called when the
|
||||
// parent prompt naturally completes — the Monitor process exits with the
|
||||
// agent turn, so the card should flip from "watching" to "ended".
|
||||
func (a *Adapter) sweepMonitorsOnPromptEnd(sessionID string) {
|
||||
for _, toolCallID := range a.takeActiveMonitors(sessionID) {
|
||||
a.mu.Lock()
|
||||
payload := a.activeToolCalls[toolCallID]
|
||||
markMonitorEnded(payload, "exited")
|
||||
delete(a.activeToolCalls, toolCallID)
|
||||
a.mu.Unlock()
|
||||
a.sendUpdate(monitorTerminalEvent(sessionID, toolCallID, toolStatusComplete, "Monitor exited", payload))
|
||||
}
|
||||
}
|
||||
|
||||
// sweepMonitorsOnReplayEnd emits a `cancelled` tool_call_update for every
|
||||
// Monitor reconstructed from session-load history. The agent process restart
|
||||
// killed the underlying script, so any Monitor that was in-flight before the
|
||||
// restart is dead — we surface that to the UI rather than leaving the card
|
||||
// stuck in "watching".
|
||||
func (a *Adapter) sweepMonitorsOnReplayEnd(sessionID string) {
|
||||
for _, toolCallID := range a.takeActiveMonitors(sessionID) {
|
||||
a.mu.Lock()
|
||||
payload := a.activeToolCalls[toolCallID]
|
||||
markMonitorEnded(payload, "session_restart")
|
||||
delete(a.activeToolCalls, toolCallID)
|
||||
a.mu.Unlock()
|
||||
a.sendUpdate(monitorTerminalEvent(sessionID, toolCallID, "cancelled", "Monitor ended (session restart)", payload))
|
||||
}
|
||||
}
|
||||
|
||||
// monitorTerminalEvent constructs the synthetic tool_call_update emitted when
|
||||
// a tracked Monitor must be marked finished — either because the parent
|
||||
// prompt ended (status "complete") or because the agent process restarted
|
||||
// (status "cancelled" with a "session restart" note in the contents).
|
||||
func monitorTerminalEvent(sessionID, toolCallID, status, note string, payload *streams.NormalizedPayload) AgentEvent {
|
||||
var contents []streams.ToolCallContentItem
|
||||
if note != "" {
|
||||
contents = []streams.ToolCallContentItem{
|
||||
{Type: "content", Content: &streams.ContentBlock{Type: "text", Text: note}},
|
||||
}
|
||||
}
|
||||
return AgentEvent{
|
||||
Type: streams.EventTypeToolUpdate,
|
||||
SessionID: sessionID,
|
||||
ToolCallID: toolCallID,
|
||||
ToolStatus: status,
|
||||
NormalizedPayload: payload,
|
||||
ToolCallContents: contents,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,575 @@
|
||||
package acp
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
acp "github.com/coder/acp-go-sdk"
|
||||
"github.com/kandev/kandev/internal/agentctl/types/streams"
|
||||
)
|
||||
|
||||
// monitorMeta builds the `_meta.claudeCode.toolName=Monitor` payload Claude-acp
|
||||
// attaches to every Monitor tool_call notification.
|
||||
func monitorMeta() map[string]any {
|
||||
return map[string]any{"claudeCode": map[string]any{"toolName": monitorToolName}}
|
||||
}
|
||||
|
||||
// nonMonitorMeta builds a `_meta.claudeCode.toolName=Bash` payload — used to
|
||||
// confirm the Monitor recognizers don't fire for unrelated tools.
|
||||
func nonMonitorMeta() map[string]any {
|
||||
return map[string]any{"claudeCode": map[string]any{"toolName": "Bash"}}
|
||||
}
|
||||
|
||||
// seedMonitor registers a Monitor as if its registration update had arrived.
|
||||
// Returns the toolCallID for use in subsequent assertions.
|
||||
func seedMonitor(t *testing.T, a *Adapter, sessionID, taskID, toolCallID string) {
|
||||
t.Helper()
|
||||
a.trackMonitor(sessionID, taskID, toolCallID)
|
||||
if got, ok := a.lookupMonitorByTaskID(sessionID, taskID); !ok || got != toolCallID {
|
||||
t.Fatalf("seedMonitor: lookup after track failed (ok=%v, got=%q)", ok, got)
|
||||
}
|
||||
}
|
||||
|
||||
// --- recognizeMonitorRegistration ---
|
||||
|
||||
func TestRecognizeMonitorRegistration_HappyPath(t *testing.T) {
|
||||
taskID, ok := recognizeMonitorRegistration(monitorMeta(),
|
||||
"Monitor started (task abc123, timeout 60000ms). You will be notified on each event.")
|
||||
if !ok || taskID != "abc123" {
|
||||
t.Errorf("recognize = (%q, %v), want (abc123, true)", taskID, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecognizeMonitorRegistration_WrongToolName(t *testing.T) {
|
||||
if _, ok := recognizeMonitorRegistration(nonMonitorMeta(),
|
||||
"Monitor started (task abc123, timeout 60000ms)."); ok {
|
||||
t.Error("recognized non-Monitor meta as Monitor registration")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecognizeMonitorRegistration_NotABanner(t *testing.T) {
|
||||
if _, ok := recognizeMonitorRegistration(monitorMeta(), "some other output"); ok {
|
||||
t.Error("recognized arbitrary string as registration banner")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecognizeMonitorRegistration_NonStringRawOutput(t *testing.T) {
|
||||
if _, ok := recognizeMonitorRegistration(monitorMeta(),
|
||||
map[string]any{"output": "Monitor started (task abc, …)"}); ok {
|
||||
t.Error("recognized object rawOutput as banner")
|
||||
}
|
||||
}
|
||||
|
||||
// --- extractMonitorEvents ---
|
||||
|
||||
func TestExtractMonitorEvents_SingleEvent(t *testing.T) {
|
||||
in := "Human: <task-notification>\n<task-id>t1</task-id>\n<event>event-A</event>\n</task-notification>"
|
||||
cleaned, events := extractMonitorEvents(in)
|
||||
if cleaned != "" {
|
||||
t.Errorf("cleaned = %q, want empty", cleaned)
|
||||
}
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("got %d events, want 1", len(events))
|
||||
}
|
||||
if events[0].TaskID != "t1" || events[0].Body != "event-A" {
|
||||
t.Errorf("event = %+v, want {t1, event-A}", events[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractMonitorEvents_MultipleEnvelopes(t *testing.T) {
|
||||
in := "<task-notification><task-id>t1</task-id><event>a</event></task-notification>" +
|
||||
" some text " +
|
||||
"<task-notification><task-id>t1</task-id><event>b</event></task-notification>"
|
||||
cleaned, events := extractMonitorEvents(in)
|
||||
if !strings.Contains(cleaned, "some text") {
|
||||
t.Errorf("cleaned = %q, want it to contain 'some text'", cleaned)
|
||||
}
|
||||
if strings.Contains(cleaned, "task-notification") {
|
||||
t.Errorf("cleaned still contains envelope: %q", cleaned)
|
||||
}
|
||||
if len(events) != 2 || events[0].Body != "a" || events[1].Body != "b" {
|
||||
t.Errorf("events = %+v, want [{t1,a},{t1,b}]", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractMonitorEvents_NoEnvelopeReturnsInputUnchanged(t *testing.T) {
|
||||
in := "regular assistant text without envelopes"
|
||||
cleaned, events := extractMonitorEvents(in)
|
||||
if cleaned != in {
|
||||
t.Errorf("cleaned mutated text without envelopes: %q", cleaned)
|
||||
}
|
||||
if events != nil {
|
||||
t.Errorf("events = %v, want nil", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractMonitorEvents_EmptyEventBody(t *testing.T) {
|
||||
in := "<task-notification><task-id>t1</task-id><event></event></task-notification>"
|
||||
_, events := extractMonitorEvents(in)
|
||||
if len(events) != 1 || events[0].Body != "" {
|
||||
t.Errorf("events = %+v, want one with empty body", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractMonitorEvents_BodyContainingAngleBrackets(t *testing.T) {
|
||||
// Real-world scripts emit lines containing `<` (XML-ish errors, shell
|
||||
// redirection like `< /dev/null`, ANSI fragments). The regex must not
|
||||
// abort on the first `<` it sees inside the event body; otherwise the
|
||||
// whole envelope leaks to the chat as raw assistant text.
|
||||
in := "<task-notification><task-id>t1</task-id><event><error>build failed: exit < 1</error></event></task-notification>"
|
||||
_, events := extractMonitorEvents(in)
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("got %d events, want 1 (regex should accept '<' in body)", len(events))
|
||||
}
|
||||
if events[0].Body != "<error>build failed: exit < 1</error>" {
|
||||
t.Errorf("event body = %q, want literal '<error>...' carried through", events[0].Body)
|
||||
}
|
||||
}
|
||||
|
||||
// --- isMonitorHumanEcho ---
|
||||
|
||||
func TestIsMonitorHumanEcho_Variants(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
text string
|
||||
want bool
|
||||
}{
|
||||
{"bare prefix", "Human:", true},
|
||||
{"prefix with whitespace", " Human: ", true},
|
||||
{"prefix with partial open tag", "Human: <task-noti", true},
|
||||
{"genuine assistant text mentioning Human:", "Human: he said yes.", false},
|
||||
{"empty", "", false},
|
||||
{"normal assistant text", "Sure, here's the result.", false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := isMonitorHumanEcho(tc.text); got != tc.want {
|
||||
t.Errorf("isMonitorHumanEcho(%q) = %v, want %v", tc.text, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// --- routeMonitorEvents (integration with adapter state) ---
|
||||
|
||||
func TestRouteMonitorEvents_EmitsSyntheticUpdateAndStripsEnvelope(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedMonitor(t, a, "s1", "t1", "tc-monitor")
|
||||
|
||||
cleaned := a.routeMonitorEvents("s1",
|
||||
"<task-notification><task-id>t1</task-id><event>event-A</event></task-notification>")
|
||||
if strings.TrimSpace(cleaned) != "" {
|
||||
t.Errorf("cleaned = %q, want empty (envelope-only chunk)", cleaned)
|
||||
}
|
||||
|
||||
events := drainEvents(a)
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("got %d events, want 1", len(events))
|
||||
}
|
||||
ev := events[0]
|
||||
if ev.Type != streams.EventTypeToolUpdate {
|
||||
t.Errorf("Type = %q, want %q", ev.Type, streams.EventTypeToolUpdate)
|
||||
}
|
||||
if ev.ToolCallID != "tc-monitor" {
|
||||
t.Errorf("ToolCallID = %q, want tc-monitor", ev.ToolCallID)
|
||||
}
|
||||
if ev.ToolStatus != "in_progress" {
|
||||
t.Errorf("ToolStatus = %q, want in_progress", ev.ToolStatus)
|
||||
}
|
||||
if len(ev.ToolCallContents) != 1 || ev.ToolCallContents[0].Content == nil ||
|
||||
ev.ToolCallContents[0].Content.Text != "event-A" {
|
||||
t.Errorf("contents = %+v, want one text content 'event-A'", ev.ToolCallContents)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouteMonitorEvents_UnknownTaskIDLeavesTextIntact(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
// no Monitor registered
|
||||
in := "<task-notification><task-id>tX</task-id><event>orphan</event></task-notification>"
|
||||
cleaned := a.routeMonitorEvents("s1", in)
|
||||
// extractMonitorEvents still strips the matched envelope from the cleaned
|
||||
// text (the envelope is meaningless to the user even when un-routable),
|
||||
// but no synthetic update is emitted.
|
||||
if strings.Contains(cleaned, "task-notification") {
|
||||
t.Errorf("cleaned still contains envelope: %q", cleaned)
|
||||
}
|
||||
if got := drainEvents(a); len(got) != 0 {
|
||||
t.Errorf("got %d events for unknown taskID, want 0", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
func TestRouteMonitorEvents_NoEnvelopeShortCircuits(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
in := "regular assistant text"
|
||||
cleaned := a.routeMonitorEvents("s1", in)
|
||||
if cleaned != in {
|
||||
t.Errorf("cleaned = %q, want unchanged", cleaned)
|
||||
}
|
||||
if got := drainEvents(a); len(got) != 0 {
|
||||
t.Errorf("got %d events for no-envelope text, want 0", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
// --- convertToolCallResultUpdate Monitor registration override ---
|
||||
|
||||
func TestConvertToolCallResultUpdate_MonitorRegistrationOverridesCompleted(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
completed := acp.ToolCallStatus("completed")
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-monitor",
|
||||
Status: &completed,
|
||||
Meta: monitorMeta(),
|
||||
RawOutput: "Monitor started (task taskZZ, timeout 60000ms). You will be notified.",
|
||||
}
|
||||
|
||||
ev := a.convertToolCallResultUpdate("s1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event, got nil")
|
||||
}
|
||||
if ev.ToolStatus != "in_progress" {
|
||||
t.Errorf("ToolStatus = %q, want in_progress (registration banner must override)", ev.ToolStatus)
|
||||
}
|
||||
// Side effect: activeMonitors records taskID -> toolCallID
|
||||
if got, ok := a.lookupMonitorByTaskID("s1", "taskZZ"); !ok || got != "tc-monitor" {
|
||||
t.Errorf("activeMonitors lookup = (%q, %v), want (tc-monitor, true)", got, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertToolCallResultUpdate_RealCompletedStaysComplete(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
completed := acp.ToolCallStatus("completed")
|
||||
// rawOutput is NOT a registration banner, so we should not override.
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-bash",
|
||||
Status: &completed,
|
||||
Meta: monitorMeta(), // even Monitor-tagged completes can occur (e.g. final exit)
|
||||
RawOutput: "exit 0\n",
|
||||
}
|
||||
ev := a.convertToolCallResultUpdate("s1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event")
|
||||
}
|
||||
if ev.ToolStatus != "complete" {
|
||||
t.Errorf("ToolStatus = %q, want complete (non-banner output must not override)", ev.ToolStatus)
|
||||
}
|
||||
}
|
||||
|
||||
// TestConvertToolCallResultUpdate_MonitorRegistrationSurvivesNormalize is a
|
||||
// regression guard: NormalizeToolResult writes the rawOutput banner string
|
||||
// into the Generic payload's Output field, which would shadow the
|
||||
// `output.monitor` view the frontend uses to detect Monitor cards. The
|
||||
// Monitor seed must run AFTER NormalizeToolResult so the synthetic
|
||||
// `{monitor: …}` wrapper is the final value.
|
||||
func TestConvertToolCallResultUpdate_MonitorRegistrationSurvivesNormalize(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
|
||||
// First the initial pending tool_call lands in activeToolCalls so
|
||||
// convertToolCallResultUpdate has a Generic payload to mutate.
|
||||
tc := &acp.SessionUpdateToolCall{
|
||||
ToolCallId: "tc-monitor",
|
||||
Title: monitorToolName,
|
||||
Kind: acp.ToolKind("other"),
|
||||
Meta: monitorMeta(),
|
||||
RawInput: map[string]any{"command": "tail -f /var/log/x"},
|
||||
}
|
||||
if ev := a.convertToolCallUpdate("s1", tc); ev == nil {
|
||||
t.Fatalf("seed: convertToolCallUpdate returned nil")
|
||||
}
|
||||
|
||||
completed := acp.ToolCallStatus("completed")
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-monitor",
|
||||
Status: &completed,
|
||||
Meta: monitorMeta(),
|
||||
RawOutput: "Monitor started (task realTaskId, timeout 60000ms). You will be notified.",
|
||||
}
|
||||
|
||||
ev := a.convertToolCallResultUpdate("s1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event, got nil")
|
||||
}
|
||||
if ev.NormalizedPayload == nil || ev.NormalizedPayload.Generic() == nil {
|
||||
t.Fatalf("expected Generic payload, got %+v", ev.NormalizedPayload)
|
||||
}
|
||||
out, ok := ev.NormalizedPayload.Generic().Output.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("Generic.Output = %v (%T), want map (the {monitor: …} wrapper, not the raw banner string)",
|
||||
ev.NormalizedPayload.Generic().Output, ev.NormalizedPayload.Generic().Output)
|
||||
}
|
||||
monitor, ok := out["monitor"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("Generic.Output[monitor] missing or wrong type — frontend would fall back to generic tool_call rendering")
|
||||
}
|
||||
if monitor["task_id"] != "realTaskId" {
|
||||
t.Errorf("monitor.task_id = %v, want realTaskId", monitor["task_id"])
|
||||
}
|
||||
if monitor["command"] != "tail -f /var/log/x" {
|
||||
t.Errorf("monitor.command = %v, want it carried over from initial tool_call", monitor["command"])
|
||||
}
|
||||
}
|
||||
|
||||
// --- sweepMonitorsOnPromptEnd ---
|
||||
|
||||
func TestSweepMonitorsOnPromptEnd_EmitsCompleteAndClears(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedMonitor(t, a, "s1", "t1", "tc-1")
|
||||
seedMonitor(t, a, "s1", "t2", "tc-2")
|
||||
|
||||
a.sweepMonitorsOnPromptEnd("s1")
|
||||
|
||||
events := drainEvents(a)
|
||||
if len(events) != 2 {
|
||||
t.Fatalf("got %d events, want 2", len(events))
|
||||
}
|
||||
for _, ev := range events {
|
||||
if ev.ToolStatus != "complete" {
|
||||
t.Errorf("ToolStatus = %q, want complete", ev.ToolStatus)
|
||||
}
|
||||
}
|
||||
if _, ok := a.lookupMonitorByTaskID("s1", "t1"); ok {
|
||||
t.Error("activeMonitors not cleared after sweep")
|
||||
}
|
||||
}
|
||||
|
||||
// TestConvertToolCallResultUpdate_AgentEmittedMonitorEndPreservesView is a
|
||||
// regression guard for an E2E failure on the first PR push: when the agent
|
||||
// emits its own terminal `tool_call_update` for a tracked Monitor (status
|
||||
// completed, plain rawOutput like "Monitor exited"), `NormalizeToolResult`
|
||||
// would clobber `Generic.Output` with that string and the frontend's
|
||||
// MonitorMessage matcher (which checks for `output.monitor`) would no
|
||||
// longer fire. Frontend would fall back to the generic tool_call card.
|
||||
func TestConvertToolCallResultUpdate_AgentEmittedMonitorEndPreservesView(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
|
||||
// Initial pending tool_call so activeToolCalls has a Generic payload.
|
||||
tc := &acp.SessionUpdateToolCall{
|
||||
ToolCallId: "tc-monitor",
|
||||
Title: monitorToolName,
|
||||
Kind: acp.ToolKind("other"),
|
||||
Meta: monitorMeta(),
|
||||
RawInput: map[string]any{"command": "tail -f /var/log/x"},
|
||||
}
|
||||
if ev := a.convertToolCallUpdate("s1", tc); ev == nil {
|
||||
t.Fatalf("seed tool_call returned nil")
|
||||
}
|
||||
|
||||
// Registration update — establishes the Monitor view and tracks taskID.
|
||||
completed := acp.ToolCallStatus("completed")
|
||||
registerTcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-monitor",
|
||||
Status: &completed,
|
||||
Meta: monitorMeta(),
|
||||
RawOutput: "Monitor started (task taskZZ, timeout 60000ms).",
|
||||
}
|
||||
if ev := a.convertToolCallResultUpdate("s1", registerTcu); ev == nil {
|
||||
t.Fatalf("registration update returned nil")
|
||||
}
|
||||
|
||||
// Agent-emitted Monitor end — what mock-agent's e2e:monitor_end produces
|
||||
// and what real claude-acp emits when a Monitor script exits naturally.
|
||||
endTcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-monitor",
|
||||
Status: &completed,
|
||||
Meta: monitorMeta(),
|
||||
RawOutput: "Monitor exited",
|
||||
}
|
||||
ev := a.convertToolCallResultUpdate("s1", endTcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event for end update, got nil")
|
||||
}
|
||||
if ev.NormalizedPayload == nil || ev.NormalizedPayload.Generic() == nil {
|
||||
t.Fatalf("expected Generic payload to survive, got %+v", ev.NormalizedPayload)
|
||||
}
|
||||
out, ok := ev.NormalizedPayload.Generic().Output.(map[string]any)
|
||||
if !ok {
|
||||
t.Fatalf("Generic.Output = %v (%T), want the {monitor: …} wrapper preserved through agent-emitted end",
|
||||
ev.NormalizedPayload.Generic().Output, ev.NormalizedPayload.Generic().Output)
|
||||
}
|
||||
monitor, ok := out["monitor"].(map[string]any)
|
||||
if !ok {
|
||||
t.Fatal("Generic.Output[monitor] missing — agent-emitted end stomped the view")
|
||||
}
|
||||
if monitor["ended"] != true {
|
||||
t.Errorf("monitor.ended = %v, want true (terminal update should mark ended)", monitor["ended"])
|
||||
}
|
||||
// The tracked entry should be dropped so the prompt-end sweep does not
|
||||
// re-emit a terminal event for the same toolCallID.
|
||||
if a.isTrackedMonitor("s1", "tc-monitor") {
|
||||
t.Error("Monitor still tracked after agent-emitted end — sweep would double-emit")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSweepMonitorsOnPromptEnd_NotDoubleCancelledByActiveToolCalls is a
|
||||
// regression guard for a real wire-bug discovered in CI: when the parent
|
||||
// prompt naturally ends, `cancelActiveToolCalls` would blanket-cancel every
|
||||
// entry in `activeToolCalls` (including the Monitor's tool call) before
|
||||
// `sweepMonitorsOnPromptEnd` ran, leaving the frontend with two
|
||||
// conflicting terminal events for the same Monitor. Monitor entries must
|
||||
// be skipped in the cancel loop so the sweep emits the single
|
||||
// authoritative "Monitor exited" event.
|
||||
func TestSweepMonitorsOnPromptEnd_NotDoubleCancelledByActiveToolCalls(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
|
||||
// Land an initial Monitor tool_call so it sits in activeToolCalls.
|
||||
tc := &acp.SessionUpdateToolCall{
|
||||
ToolCallId: "tc-monitor",
|
||||
Title: monitorToolName,
|
||||
Kind: acp.ToolKind("other"),
|
||||
Meta: monitorMeta(),
|
||||
RawInput: map[string]any{"command": "tail -f log"},
|
||||
}
|
||||
if ev := a.convertToolCallUpdate("s1", tc); ev == nil {
|
||||
t.Fatalf("seed tool_call returned nil")
|
||||
}
|
||||
a.trackMonitor("s1", "task-X", "tc-monitor")
|
||||
drainEvents(a) // ignore the initial events
|
||||
|
||||
a.cancelActiveToolCalls("s1")
|
||||
a.sweepMonitorsOnPromptEnd("s1")
|
||||
|
||||
events := drainEvents(a)
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("got %d events, want 1 (the Monitor sweep's terminal event only)", len(events))
|
||||
}
|
||||
if events[0].ToolCallID != "tc-monitor" {
|
||||
t.Errorf("event ToolCallID = %q, want tc-monitor", events[0].ToolCallID)
|
||||
}
|
||||
if events[0].ToolStatus != "complete" {
|
||||
t.Errorf("event ToolStatus = %q, want complete (cancelActiveToolCalls must skip Monitors)", events[0].ToolStatus)
|
||||
}
|
||||
if events[0].NormalizedPayload == nil {
|
||||
t.Errorf("event NormalizedPayload is nil — sweep lost payload (activeToolCalls drained too early)")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSweepMonitorsOnReplayEnd_EmitsCancelledWithRestartNote(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedMonitor(t, a, "s1", "t1", "tc-1")
|
||||
|
||||
a.sweepMonitorsOnReplayEnd("s1")
|
||||
|
||||
events := drainEvents(a)
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("got %d events, want 1", len(events))
|
||||
}
|
||||
ev := events[0]
|
||||
if ev.ToolStatus != "cancelled" {
|
||||
t.Errorf("ToolStatus = %q, want cancelled", ev.ToolStatus)
|
||||
}
|
||||
if len(ev.ToolCallContents) == 0 || ev.ToolCallContents[0].Content == nil ||
|
||||
!strings.Contains(ev.ToolCallContents[0].Content.Text, "session restart") {
|
||||
t.Errorf("contents = %+v, want a 'session restart' note", ev.ToolCallContents)
|
||||
}
|
||||
}
|
||||
|
||||
// --- captureReplayMonitor (replay rebuild) ---
|
||||
|
||||
func TestCaptureReplayMonitor_RebuildsFromToolCallAndUpdate(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
|
||||
// Replay the initial tool_call (we don't yet know taskID).
|
||||
tc := &acp.SessionUpdateToolCall{
|
||||
ToolCallId: "tc-replay",
|
||||
Title: monitorToolName,
|
||||
Kind: acp.ToolKind("other"),
|
||||
Meta: monitorMeta(),
|
||||
RawInput: map[string]any{"command": "tail -f /var/log/x"},
|
||||
}
|
||||
a.captureReplayMonitor("s1", acp.SessionUpdate{ToolCall: tc})
|
||||
|
||||
// Replay the registration banner — taskID becomes known.
|
||||
completed := acp.ToolCallStatus("completed")
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-replay",
|
||||
Status: &completed,
|
||||
Meta: monitorMeta(),
|
||||
RawOutput: "Monitor started (task realTask99, timeout 60000ms).",
|
||||
}
|
||||
a.captureReplayMonitor("s1", acp.SessionUpdate{ToolCallUpdate: tcu})
|
||||
|
||||
if got, ok := a.lookupMonitorByTaskID("s1", "realTask99"); !ok || got != "tc-replay" {
|
||||
t.Errorf("after replay: lookup = (%q, %v), want (tc-replay, true)", got, ok)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCaptureReplayMonitor_TerminalUpdateRemovesFromMap(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedMonitor(t, a, "s1", "tX", "tc-9")
|
||||
|
||||
cancelled := acp.ToolCallStatus("cancelled")
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-9",
|
||||
Status: &cancelled,
|
||||
Meta: monitorMeta(),
|
||||
}
|
||||
a.captureReplayMonitor("s1", acp.SessionUpdate{ToolCallUpdate: tcu})
|
||||
|
||||
if _, ok := a.lookupMonitorByTaskID("s1", "tX"); ok {
|
||||
t.Error("terminal replay update did not drop Monitor from map")
|
||||
}
|
||||
}
|
||||
|
||||
// --- convertMessageChunk integration ---
|
||||
|
||||
func TestConvertMessageChunk_MonitorEnvelopeStrippedAndRouted(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedMonitor(t, a, "s1", "t1", "tc-monitor")
|
||||
|
||||
chunk := acp.TextBlock(
|
||||
"Human: <task-notification><task-id>t1</task-id><event>line-1</event></task-notification>",
|
||||
)
|
||||
ev := a.convertMessageChunk("s1", chunk, "assistant")
|
||||
if ev != nil {
|
||||
t.Errorf("expected nil (envelope-only chunk should drop), got %+v", ev)
|
||||
}
|
||||
|
||||
// One synthetic event should have been routed to the Monitor's tool card.
|
||||
events := drainEvents(a)
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("got %d routed events, want 1", len(events))
|
||||
}
|
||||
if events[0].ToolCallID != "tc-monitor" {
|
||||
t.Errorf("ToolCallID = %q, want tc-monitor", events[0].ToolCallID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertMessageChunk_PlainAssistantTextPassesThrough(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
chunk := acp.TextBlock("Sure, here's the result.")
|
||||
ev := a.convertMessageChunk("s1", chunk, "assistant")
|
||||
if ev == nil {
|
||||
t.Fatal("expected event for plain text, got nil")
|
||||
}
|
||||
if ev.Text != "Sure, here's the result." {
|
||||
t.Errorf("Text = %q, want unchanged", ev.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertMessageChunk_HumanEchoOnlyChunkDropped(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
chunk := acp.TextBlock("Human:")
|
||||
ev := a.convertMessageChunk("s1", chunk, "assistant")
|
||||
if ev != nil {
|
||||
t.Errorf("expected nil for orphan Human: prefix, got %+v", ev)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertMessageChunk_UserRoleSkipsMonitorRouting(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedMonitor(t, a, "s1", "t1", "tc-monitor")
|
||||
|
||||
chunk := acp.TextBlock(
|
||||
"<task-notification><task-id>t1</task-id><event>x</event></task-notification>",
|
||||
)
|
||||
ev := a.convertMessageChunk("s1", chunk, "user")
|
||||
if ev == nil {
|
||||
t.Fatal("expected event for user role, got nil")
|
||||
}
|
||||
// User text must reach the chat untouched — the parser should only run for assistant role.
|
||||
if !strings.Contains(ev.Text, "task-notification") {
|
||||
t.Errorf("user text was scrubbed: %q", ev.Text)
|
||||
}
|
||||
if got := drainEvents(a); len(got) != 0 {
|
||||
t.Errorf("got %d routed events for user role, want 0", len(got))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
package acp
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
acp "github.com/coder/acp-go-sdk"
|
||||
)
|
||||
|
||||
// seedExecuteToolCall registers a pending Bash tool_call (Claude-acp pattern: empty
|
||||
// rawInput, default "Terminal" title) so subsequent tool_call_update tests have an
|
||||
// activeToolCalls entry to update.
|
||||
func seedExecuteToolCall(t *testing.T, a *Adapter, toolCallID string) {
|
||||
t.Helper()
|
||||
tc := &acp.SessionUpdateToolCall{
|
||||
ToolCallId: acp.ToolCallId(toolCallID),
|
||||
Title: "Terminal",
|
||||
Status: acp.ToolCallStatus("pending"),
|
||||
Kind: acp.ToolKind("execute"),
|
||||
RawInput: map[string]any{},
|
||||
}
|
||||
if ev := a.convertToolCallUpdate("session-1", tc); ev == nil {
|
||||
t.Fatalf("seed: convertToolCallUpdate returned nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestConvertToolCallResultUpdate_StatusLessUpdateBecomesInProgress reproduces the
|
||||
// claude-acp Bash flow: an initial tool_call with empty rawInput and "Terminal"
|
||||
// placeholder title, followed by a tool_call_update that carries the actual command
|
||||
// and title but no Status field. Without this fix the orchestrator drops the update
|
||||
// (its switch only matches known statuses) and the message stays on "Terminal".
|
||||
func TestConvertToolCallResultUpdate_StatusLessUpdateBecomesInProgress(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedExecuteToolCall(t, a, "tc-1")
|
||||
|
||||
cmdTitle := "ls -la /tmp | head -5"
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-1",
|
||||
Title: &cmdTitle,
|
||||
RawInput: map[string]any{
|
||||
"command": "ls -la /tmp | head -5",
|
||||
"description": "List first 5 entries in /tmp",
|
||||
},
|
||||
}
|
||||
|
||||
ev := a.convertToolCallResultUpdate("session-1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event, got nil")
|
||||
}
|
||||
if ev.ToolStatus != "in_progress" {
|
||||
t.Errorf("ToolStatus = %q, want %q (status-less updates with content must route through orchestrator)", ev.ToolStatus, "in_progress")
|
||||
}
|
||||
if ev.ToolTitle != cmdTitle {
|
||||
t.Errorf("ToolTitle = %q, want %q", ev.ToolTitle, cmdTitle)
|
||||
}
|
||||
if ev.NormalizedPayload == nil || ev.NormalizedPayload.ShellExec() == nil {
|
||||
t.Fatalf("expected ShellExec payload, got %+v", ev.NormalizedPayload)
|
||||
}
|
||||
if got := ev.NormalizedPayload.ShellExec().Command; got != "ls -la /tmp | head -5" {
|
||||
t.Errorf("ShellExec.Command = %q, want command from rawInput", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertToolCallResultUpdate_StatusLessRawInputOnlyBecomesInProgress(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedExecuteToolCall(t, a, "tc-2")
|
||||
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-2",
|
||||
RawInput: map[string]any{"command": "pwd"},
|
||||
}
|
||||
|
||||
ev := a.convertToolCallResultUpdate("session-1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event, got nil")
|
||||
}
|
||||
if ev.ToolStatus != "in_progress" {
|
||||
t.Errorf("ToolStatus = %q, want %q", ev.ToolStatus, "in_progress")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertToolCallResultUpdate_StatusLessContentOnlyBecomesInProgress(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedExecuteToolCall(t, a, "tc-3")
|
||||
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-3",
|
||||
Content: []acp.ToolCallContent{
|
||||
{
|
||||
Content: &acp.ToolCallContentContent{
|
||||
Content: acp.TextBlock("partial output"),
|
||||
Type: "content",
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
ev := a.convertToolCallResultUpdate("session-1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event, got nil")
|
||||
}
|
||||
if ev.ToolStatus != "in_progress" {
|
||||
t.Errorf("ToolStatus = %q, want %q", ev.ToolStatus, "in_progress")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertToolCallResultUpdate_FullyEmptyUpdateKeepsEmptyStatus(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedExecuteToolCall(t, a, "tc-4")
|
||||
|
||||
// A no-op update — no status, no title, no rawInput, no content. Should not
|
||||
// be promoted to in_progress; orchestrator already ignores empty status, so
|
||||
// behaviour is unchanged from today.
|
||||
tcu := &acp.SessionToolCallUpdate{ToolCallId: "tc-4"}
|
||||
|
||||
ev := a.convertToolCallResultUpdate("session-1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event, got nil")
|
||||
}
|
||||
if ev.ToolStatus != "" {
|
||||
t.Errorf("ToolStatus = %q, want empty (no synthesized status for no-op update)", ev.ToolStatus)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConvertToolCallResultUpdate_ExplicitCompletedStatusUnchanged(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
seedExecuteToolCall(t, a, "tc-5")
|
||||
|
||||
completed := acp.ToolCallStatus("completed")
|
||||
tcu := &acp.SessionToolCallUpdate{
|
||||
ToolCallId: "tc-5",
|
||||
Status: &completed,
|
||||
RawOutput: "ok",
|
||||
}
|
||||
|
||||
ev := a.convertToolCallResultUpdate("session-1", tcu)
|
||||
if ev == nil {
|
||||
t.Fatal("expected event, got nil")
|
||||
}
|
||||
// "completed" is normalized to "complete" by the existing logic — regression guard.
|
||||
if ev.ToolStatus != "complete" {
|
||||
t.Errorf("ToolStatus = %q, want %q", ev.ToolStatus, "complete")
|
||||
}
|
||||
}
|
||||
@@ -21,6 +21,7 @@ import { TodoMessage } from "@/components/task/chat/messages/todo-message";
|
||||
import { ScriptExecutionMessage } from "@/components/task/chat/messages/script-execution-message";
|
||||
import { ClarificationRequestMessage } from "@/components/task/chat/messages/clarification-request-message";
|
||||
import { ToolSubagentMessage } from "@/components/task/chat/messages/tool-subagent-message";
|
||||
import { MonitorMessage } from "@/components/task/chat/messages/monitor-message";
|
||||
import { AgentPlanMessage } from "@/components/task/chat/messages/agent-plan-message";
|
||||
import { ActionMessage } from "@/components/task/chat/messages/action-message";
|
||||
|
||||
@@ -127,6 +128,21 @@ const adapters: MessageAdapter[] = [
|
||||
<ToolExecuteMessage comment={comment} worktreePath={ctx.worktreePath} />
|
||||
),
|
||||
},
|
||||
{
|
||||
// Claude-acp's Monitor tool — long-lived background script with streaming
|
||||
// events. Rendered with a dedicated card that shows watching state, event
|
||||
// count, and the most recent event tail. Detected via the structured
|
||||
// `monitor` view the adapter writes into the Generic payload's output
|
||||
// wrapper (presence-based rather than title-based so renames upstream
|
||||
// don't break the match). Must run BEFORE the generic tool_call adapter.
|
||||
matches: (comment) => {
|
||||
if (comment.type !== "tool_call") return false;
|
||||
const meta = comment.metadata as ToolCallMetadata | undefined;
|
||||
const out = meta?.normalized?.generic?.output as { monitor?: unknown } | undefined;
|
||||
return !!out && typeof out === "object" && !!out.monitor;
|
||||
},
|
||||
render: (comment) => <MonitorMessage comment={comment} />,
|
||||
},
|
||||
{
|
||||
// Subagent Task tool calls with nested children
|
||||
matches: (comment, ctx) => {
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { renderToStaticMarkup } from "react-dom/server";
|
||||
import { MonitorMessage } from "./monitor-message";
|
||||
import type { Message } from "@/lib/types/http";
|
||||
|
||||
// monitorMessage is a tiny test-helper that constructs the kandev Message
|
||||
// shape produced by the orchestrator for a Monitor tool_call. The fields
|
||||
// match what `service_messages.applyToolCallMessageUpdate` writes after
|
||||
// receiving an in_progress tool_call_update from the ACP adapter.
|
||||
function monitorMessage(opts: {
|
||||
status?: "pending" | "running" | "complete" | "error";
|
||||
command?: string;
|
||||
taskId?: string;
|
||||
eventCount?: number;
|
||||
recentEvents?: string[];
|
||||
ended?: boolean;
|
||||
endReason?: string;
|
||||
}): Message {
|
||||
return {
|
||||
id: "msg-1",
|
||||
session_id: "s1",
|
||||
task_id: "t1",
|
||||
author_type: "agent",
|
||||
content: "Monitor",
|
||||
type: "tool_call",
|
||||
created_at: "2026-04-25T10:00:00Z",
|
||||
metadata: {
|
||||
tool_call_id: "tc-monitor",
|
||||
title: "Monitor",
|
||||
status: opts.status ?? "running",
|
||||
normalized: {
|
||||
kind: "generic",
|
||||
generic: {
|
||||
name: "other",
|
||||
input: { command: opts.command ?? "" },
|
||||
output: {
|
||||
monitor: {
|
||||
kind: "Monitor",
|
||||
task_id: opts.taskId ?? "task-1",
|
||||
command: opts.command ?? "",
|
||||
event_count: opts.eventCount ?? 0,
|
||||
recent_events: opts.recentEvents ?? [],
|
||||
ended: opts.ended ?? false,
|
||||
end_reason: opts.endReason ?? "",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("MonitorMessage", () => {
|
||||
it("renders a 'watching' pill while the monitor is in progress", () => {
|
||||
const html = renderToStaticMarkup(
|
||||
<MonitorMessage comment={monitorMessage({ command: "gh pr checks", eventCount: 0 })} />,
|
||||
);
|
||||
expect(html).toContain("Monitor");
|
||||
expect(html).toContain("gh pr checks");
|
||||
expect(html).toContain("watching");
|
||||
expect(html).not.toContain("ended");
|
||||
});
|
||||
|
||||
it("shows the event count when events have arrived", () => {
|
||||
const html = renderToStaticMarkup(
|
||||
<MonitorMessage comment={monitorMessage({ eventCount: 3, recentEvents: ["a", "b", "c"] })} />,
|
||||
);
|
||||
expect(html).toContain("3 events");
|
||||
// recent events render under the card body when expanded; the auto-expand
|
||||
// logic kicks in while watching, so all three should be in the markup.
|
||||
expect(html).toContain(">a<");
|
||||
expect(html).toContain(">b<");
|
||||
expect(html).toContain(">c<");
|
||||
});
|
||||
|
||||
it("uses singular 'event' for a count of 1", () => {
|
||||
const html = renderToStaticMarkup(
|
||||
<MonitorMessage comment={monitorMessage({ eventCount: 1, recentEvents: ["x"] })} />,
|
||||
);
|
||||
expect(html).toContain("1 event");
|
||||
expect(html).not.toContain("1 events");
|
||||
});
|
||||
|
||||
it("flips to 'ended' when the monitor exits cleanly", () => {
|
||||
const html = renderToStaticMarkup(
|
||||
<MonitorMessage
|
||||
comment={monitorMessage({ ended: true, endReason: "exited", status: "complete" })}
|
||||
/>,
|
||||
);
|
||||
expect(html).toContain("ended");
|
||||
expect(html).not.toContain("watching");
|
||||
});
|
||||
|
||||
it("keeps the events tail expanded after the monitor ends", () => {
|
||||
// After the run completes the user still wants to see what fired.
|
||||
// Auto-expand keys off `recentEvents.length > 0`, not `!ended`.
|
||||
const html = renderToStaticMarkup(
|
||||
<MonitorMessage
|
||||
comment={monitorMessage({
|
||||
ended: true,
|
||||
endReason: "exited",
|
||||
status: "complete",
|
||||
eventCount: 3,
|
||||
recentEvents: ["queued", "running", "passed"],
|
||||
})}
|
||||
/>,
|
||||
);
|
||||
expect(html).toContain(">queued<");
|
||||
expect(html).toContain(">running<");
|
||||
expect(html).toContain(">passed<");
|
||||
});
|
||||
|
||||
it("flips to 'ended (session restart)' when the agent process restarted", () => {
|
||||
const html = renderToStaticMarkup(
|
||||
<MonitorMessage comment={monitorMessage({ ended: true, endReason: "session_restart" })} />,
|
||||
);
|
||||
expect(html).toContain("ended (session restart)");
|
||||
});
|
||||
|
||||
it("renders a baseline card even when no monitor view is attached", () => {
|
||||
// Defensive guard: if the adapter didn't seed the view (e.g. very old
|
||||
// session before this code shipped), the renderer should still produce
|
||||
// the "watching" baseline rather than crashing.
|
||||
const empty: Message = {
|
||||
id: "msg-1",
|
||||
session_id: "s1",
|
||||
task_id: "t1",
|
||||
author_type: "agent",
|
||||
content: "Monitor",
|
||||
type: "tool_call",
|
||||
created_at: "2026-04-25T10:00:00Z",
|
||||
metadata: { tool_call_id: "tc-monitor", title: "Monitor", status: "running" },
|
||||
};
|
||||
const html = renderToStaticMarkup(<MonitorMessage comment={empty} />);
|
||||
expect(html).toContain("Monitor");
|
||||
expect(html).toContain("watching");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,135 @@
|
||||
"use client";
|
||||
|
||||
import { memo } from "react";
|
||||
import { IconActivity, IconCheck, IconX } from "@tabler/icons-react";
|
||||
import { GridSpinner } from "@/components/grid-spinner";
|
||||
import { cn } from "@/lib/utils";
|
||||
import type { Message } from "@/lib/types/http";
|
||||
import type { ToolCallMetadata } from "@/components/task/chat/types";
|
||||
import { readMonitorView } from "@/components/task/chat/types";
|
||||
import { ExpandableRow } from "./expandable-row";
|
||||
import { useExpandState } from "./use-expand-state";
|
||||
|
||||
type MonitorMessageProps = {
|
||||
comment: Message;
|
||||
};
|
||||
|
||||
// Status icon mirroring the other tool cards: spinner while watching,
|
||||
// checkmark when the parent prompt completes naturally, X when the agent
|
||||
// process restarted (the adapter sets status=cancelled in that case).
|
||||
function MonitorStatusIcon({ status, ended }: { status: string | undefined; ended: boolean }) {
|
||||
if (!ended) return <GridSpinner className="text-muted-foreground" />;
|
||||
if (status === "cancelled") return <IconX className="h-3.5 w-3.5 text-amber-500" />;
|
||||
return <IconCheck className="h-3.5 w-3.5 text-green-500" />;
|
||||
}
|
||||
|
||||
function formatStatusLabel(
|
||||
status: string | undefined,
|
||||
ended: boolean,
|
||||
endReason: string | undefined,
|
||||
) {
|
||||
if (!ended) return "watching";
|
||||
if (status === "cancelled" || endReason === "session_restart") return "ended (session restart)";
|
||||
return "ended";
|
||||
}
|
||||
|
||||
type MonitorViewModel = {
|
||||
status: ToolCallMetadata["status"];
|
||||
ended: boolean;
|
||||
endReason: string;
|
||||
eventCount: number;
|
||||
recentEvents: string[];
|
||||
title: string;
|
||||
countSuffix: string;
|
||||
};
|
||||
|
||||
function pluralizeEventCount(n: number): string {
|
||||
if (n <= 0) return "";
|
||||
return ` · ${n} event${n === 1 ? "" : "s"}`;
|
||||
}
|
||||
|
||||
function buildMonitorViewModel(comment: Message): MonitorViewModel {
|
||||
const metadata = comment.metadata as ToolCallMetadata | undefined;
|
||||
const view = readMonitorView(metadata?.normalized?.generic) ?? {};
|
||||
const command = view.command ?? "";
|
||||
const eventCount = view.event_count ?? 0;
|
||||
return {
|
||||
status: metadata?.status,
|
||||
ended: view.ended === true,
|
||||
endReason: view.end_reason ?? "",
|
||||
eventCount,
|
||||
recentEvents: view.recent_events ?? [],
|
||||
title: command ? `Monitor: ${command}` : "Monitor",
|
||||
countSuffix: pluralizeEventCount(eventCount),
|
||||
};
|
||||
}
|
||||
|
||||
function MonitorHeader({ vm }: { vm: MonitorViewModel }) {
|
||||
return (
|
||||
<div className="flex items-center gap-2 text-xs" data-testid="monitor-card">
|
||||
<span className="inline-flex items-center gap-1.5">
|
||||
<span className="font-mono text-xs text-muted-foreground">{vm.title}</span>
|
||||
{vm.countSuffix && (
|
||||
<span className="text-xs text-muted-foreground/80" data-testid="monitor-event-count">
|
||||
{vm.countSuffix}
|
||||
</span>
|
||||
)}
|
||||
<span
|
||||
className={cn(
|
||||
"text-[10px] uppercase tracking-wide rounded px-1.5 py-0.5",
|
||||
vm.ended
|
||||
? "bg-muted text-muted-foreground"
|
||||
: "bg-blue-500/10 text-blue-700 dark:text-blue-400",
|
||||
)}
|
||||
data-testid="monitor-status-pill"
|
||||
>
|
||||
{formatStatusLabel(vm.status, vm.ended, vm.endReason)}
|
||||
</span>
|
||||
<MonitorStatusIcon status={vm.status} ended={vm.ended} />
|
||||
</span>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
export const MonitorMessage = memo(function MonitorMessage({ comment }: MonitorMessageProps) {
|
||||
const vm = buildMonitorViewModel(comment);
|
||||
// Auto-expand whenever there are events to surface — both during the
|
||||
// watch (so users see the stream live) and after it ends (so they can
|
||||
// review what happened). Empty Monitors collapse since there's nothing
|
||||
// to show.
|
||||
const { isExpanded, handleToggle } = useExpandState(vm.status, vm.recentEvents.length > 0);
|
||||
|
||||
return (
|
||||
<ExpandableRow
|
||||
icon={
|
||||
<IconActivity
|
||||
className={cn("h-4 w-4", vm.ended ? "text-muted-foreground" : "text-blue-500")}
|
||||
data-testid="monitor-card-icon"
|
||||
/>
|
||||
}
|
||||
header={<MonitorHeader vm={vm} />}
|
||||
hasExpandableContent={vm.recentEvents.length > 0}
|
||||
isExpanded={isExpanded}
|
||||
onToggle={handleToggle}
|
||||
>
|
||||
<MonitorEventList events={vm.recentEvents} />
|
||||
</ExpandableRow>
|
||||
);
|
||||
});
|
||||
|
||||
function MonitorEventList({ events }: { events: string[] }) {
|
||||
if (events.length === 0) return null;
|
||||
return (
|
||||
<div className="pl-4 border-l-2 border-border/30 space-y-1" data-testid="monitor-event-list">
|
||||
{events.map((line, idx) => (
|
||||
<pre
|
||||
key={`${idx}-${line.slice(0, 32)}`}
|
||||
className="text-xs bg-muted/30 rounded p-2 overflow-x-auto whitespace-pre-wrap"
|
||||
data-testid="monitor-event"
|
||||
>
|
||||
{line}
|
||||
</pre>
|
||||
))}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@@ -12,6 +12,32 @@ export type GenericPayload = {
|
||||
output?: unknown;
|
||||
};
|
||||
|
||||
// MonitorView is the structured shape the kandev ACP adapter writes into the
|
||||
// Generic tool payload's `output.monitor` field for Claude-acp's `Monitor`
|
||||
// tool. The adapter mutates this in place across tool_call_updates so the
|
||||
// UI sees a stable view of the Monitor's state (event count, recent tail,
|
||||
// terminal flag) without needing a custom NormalizedPayload kind.
|
||||
export type MonitorView = {
|
||||
kind?: string;
|
||||
task_id?: string;
|
||||
command?: string;
|
||||
event_count?: number;
|
||||
recent_events?: string[];
|
||||
ended?: boolean;
|
||||
end_reason?: string;
|
||||
};
|
||||
|
||||
// Output wrapper helper: when the adapter attaches a Monitor view, it lands
|
||||
// at `generic.output.monitor`. Frontend code should narrow via this guard
|
||||
// rather than casting on its own.
|
||||
export function readMonitorView(payload: GenericPayload | undefined): MonitorView | null {
|
||||
const out = payload?.output;
|
||||
if (!out || typeof out !== "object") return null;
|
||||
const wrapper = out as { monitor?: unknown };
|
||||
if (!wrapper.monitor || typeof wrapper.monitor !== "object") return null;
|
||||
return wrapper.monitor as MonitorView;
|
||||
}
|
||||
|
||||
export type ReadFileOutput = {
|
||||
content?: string;
|
||||
line_count?: number;
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
import { test, expect } from "../../fixtures/test-base";
|
||||
import { SessionPage } from "../../pages/session-page";
|
||||
|
||||
// All Monitor scenarios drive the kandev backend with the mock-agent's new
|
||||
// e2e:monitor_* directives, which reproduce claude-agent-acp's wire format
|
||||
// without depending on the real Claude Code SDK. The kandev ACP adapter
|
||||
// recognizes the registration banner, parses task-notification envelopes,
|
||||
// strips the model's "Human:" echoes, and tracks live monitor state — those
|
||||
// behaviours are unit-tested in the backend; this file asserts the full
|
||||
// pipeline lands the right thing in the chat UI.
|
||||
|
||||
test.describe("Claude-acp Monitor tool", () => {
|
||||
test("renders watching card, accumulates events, hides envelope text", async ({
|
||||
testPage,
|
||||
apiClient,
|
||||
seedData,
|
||||
}) => {
|
||||
test.setTimeout(60_000);
|
||||
|
||||
// Three monitor events fire over ~3s. The agent then ends the monitor and
|
||||
// produces a final assistant message so the turn completes deterministically.
|
||||
const script = [
|
||||
'e2e:monitor_start("task-1", "gh pr checks --watch")',
|
||||
"e2e:delay(200)",
|
||||
'e2e:monitor_event("task-1", "queued: lint")',
|
||||
"e2e:delay(200)",
|
||||
'e2e:monitor_event("task-1", "in_progress: lint")',
|
||||
"e2e:delay(200)",
|
||||
'e2e:monitor_event("task-1", "success: lint")',
|
||||
'e2e:monitor_end("task-1")',
|
||||
'e2e:message("watching done")',
|
||||
].join("\n");
|
||||
|
||||
const task = await apiClient.createTaskWithAgent(
|
||||
seedData.workspaceId,
|
||||
"Monitor watching",
|
||||
seedData.agentProfileId,
|
||||
{
|
||||
description: script,
|
||||
workflow_id: seedData.workflowId,
|
||||
workflow_step_id: seedData.startStepId,
|
||||
repository_ids: [seedData.repositoryId],
|
||||
},
|
||||
);
|
||||
|
||||
await testPage.goto(`/t/${task.id}`);
|
||||
const session = new SessionPage(testPage);
|
||||
await session.waitForLoad();
|
||||
await expect(session.idleInput()).toBeVisible({ timeout: 30_000 });
|
||||
|
||||
// The dedicated Monitor card renders, not the generic tool_call row.
|
||||
const monitorCard = session.chat.locator('[data-testid="monitor-card"]').first();
|
||||
await expect(monitorCard).toBeVisible();
|
||||
await expect(monitorCard).toContainText("gh pr checks --watch");
|
||||
|
||||
// Event count badge surfaces all three events. Status pill flipped to
|
||||
// "ended" because the script issued monitor_end and the parent turn
|
||||
// completed.
|
||||
await expect(monitorCard).toContainText("3 events");
|
||||
await expect(session.chat.locator('[data-testid="monitor-status-pill"]').first()).toContainText(
|
||||
/ended/,
|
||||
);
|
||||
|
||||
// Each event body landed in the recent-events tail.
|
||||
const eventList = session.chat.locator('[data-testid="monitor-event"]');
|
||||
await expect(eventList).toHaveCount(3);
|
||||
await expect(eventList.nth(0)).toContainText("queued: lint");
|
||||
await expect(eventList.nth(2)).toContainText("success: lint");
|
||||
|
||||
// Critical: the model's "Human: <task-notification>" echo must NOT appear
|
||||
// anywhere in the chat. The adapter strips matched envelopes from the
|
||||
// assistant text and drops orphan "Human:" prefixes entirely.
|
||||
await expect(session.chat).not.toContainText("<task-notification>");
|
||||
await expect(session.chat).not.toContainText("Human: <task");
|
||||
});
|
||||
|
||||
test("singular event count uses 'event' not 'events'", async ({
|
||||
testPage,
|
||||
apiClient,
|
||||
seedData,
|
||||
}) => {
|
||||
test.setTimeout(45_000);
|
||||
|
||||
const script = [
|
||||
'e2e:monitor_start("task-1", "tail -f log")',
|
||||
"e2e:delay(100)",
|
||||
'e2e:monitor_event("task-1", "first and only line")',
|
||||
'e2e:monitor_end("task-1")',
|
||||
'e2e:message("done")',
|
||||
].join("\n");
|
||||
|
||||
const task = await apiClient.createTaskWithAgent(
|
||||
seedData.workspaceId,
|
||||
"Monitor singular",
|
||||
seedData.agentProfileId,
|
||||
{
|
||||
description: script,
|
||||
workflow_id: seedData.workflowId,
|
||||
workflow_step_id: seedData.startStepId,
|
||||
repository_ids: [seedData.repositoryId],
|
||||
},
|
||||
);
|
||||
|
||||
await testPage.goto(`/t/${task.id}`);
|
||||
const session = new SessionPage(testPage);
|
||||
await session.waitForLoad();
|
||||
await expect(session.idleInput()).toBeVisible({ timeout: 30_000 });
|
||||
|
||||
const card = session.chat.locator('[data-testid="monitor-card"]').first();
|
||||
await expect(card).toContainText("1 event");
|
||||
await expect(card).not.toContainText("1 events");
|
||||
});
|
||||
|
||||
test("page reload preserves the monitor card and recent events", async ({
|
||||
testPage,
|
||||
apiClient,
|
||||
seedData,
|
||||
}) => {
|
||||
test.setTimeout(60_000);
|
||||
|
||||
const script = [
|
||||
'e2e:monitor_start("task-1", "wait for ci")',
|
||||
"e2e:delay(150)",
|
||||
'e2e:monitor_event("task-1", "step-1")',
|
||||
"e2e:delay(150)",
|
||||
'e2e:monitor_event("task-1", "step-2")',
|
||||
'e2e:monitor_end("task-1")',
|
||||
'e2e:message("done")',
|
||||
].join("\n");
|
||||
|
||||
const task = await apiClient.createTaskWithAgent(
|
||||
seedData.workspaceId,
|
||||
"Monitor reload",
|
||||
seedData.agentProfileId,
|
||||
{
|
||||
description: script,
|
||||
workflow_id: seedData.workflowId,
|
||||
workflow_step_id: seedData.startStepId,
|
||||
repository_ids: [seedData.repositoryId],
|
||||
},
|
||||
);
|
||||
|
||||
await testPage.goto(`/t/${task.id}`);
|
||||
const session = new SessionPage(testPage);
|
||||
await session.waitForLoad();
|
||||
await expect(session.idleInput()).toBeVisible({ timeout: 30_000 });
|
||||
await expect(session.chat.locator('[data-testid="monitor-card"]').first()).toContainText(
|
||||
"2 events",
|
||||
);
|
||||
|
||||
// Reload — SSR + Zustand hydration must reconstitute the card from DB.
|
||||
await testPage.reload();
|
||||
await session.waitForLoad();
|
||||
await expect(session.idleInput()).toBeVisible({ timeout: 30_000 });
|
||||
|
||||
const card = session.chat.locator('[data-testid="monitor-card"]').first();
|
||||
await expect(card).toBeVisible();
|
||||
await expect(card).toContainText("wait for ci");
|
||||
await expect(card).toContainText("2 events");
|
||||
await expect(session.chat.locator('[data-testid="monitor-event"]')).toHaveCount(2);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,143 @@
|
||||
# Monitor Tool Support in the Claude ACP Adapter
|
||||
|
||||
**Date:** 2026-04-25
|
||||
**Status:** proposed
|
||||
**PR:** TBD
|
||||
**Decision:** —
|
||||
|
||||
## Problem
|
||||
|
||||
Newer versions of the Claude Code agent SDK (used via `@agentclientprotocol/claude-agent-acp` ≥ 0.31.0) ship a `Monitor` tool that runs a long-lived background script and streams each stdout line back to the model as a synthetic user turn. Production users running long polling skills (`pr-fixup`, `loop`) report that the kandev chat UI:
|
||||
|
||||
1. Shows the `Monitor` tool card as "completed" within 1 second of starting, even though the watch lasts 30+ minutes.
|
||||
2. Shows nothing at all during the watch period — when the agent stays quiet between events, no chat output reaches the UI.
|
||||
3. "Avalanches" output when the user types a follow-up message: kandev's `messageQueue` holds the new prompt until `agent.ready` fires, which only happens when the wrapper's `prompt()` call returns at the end of the watch.
|
||||
|
||||
Investigation (captured in `acp-debug` JSONL traces, see `acp-debug/claude-acp-prompt-*.jsonl`) showed:
|
||||
|
||||
- `claude-agent-acp`'s wrapper keeps `prompt()` alive for the full Monitor lifetime — `session_state_changed=idle` is only emitted once Monitor exits. So `EventTypeComplete` is delayed for the entire watch.
|
||||
- The wrapper emits the `tool_call_update(Monitor)` with `status=completed` *immediately* after Monitor registers — its `rawOutput` is the literal banner `Monitor started (task X, timeout Yms)…`, **not** a real completion.
|
||||
- Each Monitor event is injected into the SDK as a "user" turn carrying a `<task-notification><task-id>X</task-id>...<event>Y</event></task-notification>` envelope. The wrapper *suppresses* `user_message_chunk` for plain-text user turns (`acp-agent.js:712-719`), so kandev never sees the events directly.
|
||||
- The events do leak through indirectly when the model produces an `agent_message_chunk` whose text starts with `Human: <task-notification>...` — the model auto-completing the suppressed user turn.
|
||||
|
||||
We want to fix this without modifying `claude-agent-acp`.
|
||||
|
||||
## Design
|
||||
|
||||
All work lands in **`apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go`** (the ACP transport adapter inside agentctl) plus minor frontend additions. No new database tables, no upstream patches.
|
||||
|
||||
### Wire-format quick reference
|
||||
|
||||
```
|
||||
tool_call (Monitor) _meta.claudeCode.toolName=Monitor, rawInput.command=…
|
||||
tool_call_update (Monitor) _meta.claudeCode.toolResponse.taskId=X
|
||||
status=completed
|
||||
rawOutput="Monitor started (task X, timeout Yms)…" ← REGISTRATION, not completion
|
||||
agent_message_chunk text="Human: <task-notification><task-id>X</task-id>
|
||||
…<event>Y</event></task-notification>" ← per event, parsed out
|
||||
… [model output, possibly empty for long stretches] …
|
||||
[prompt() returns, EventTypeComplete fires] ← real Monitor end
|
||||
```
|
||||
|
||||
### Adapter changes
|
||||
|
||||
1. **In-memory tracking** — add `activeMonitors map[string]map[string]string` to `Adapter` (sessionID → taskID → toolCallID), guarded by the existing `Adapter.mu`.
|
||||
|
||||
2. **Recognize Monitor registration** in `convertToolCallResultUpdate` (around line 1363):
|
||||
- When `_meta.claudeCode.toolName == "Monitor"` AND `rawOutput` matches `^Monitor started \(task (?P<task>[a-z0-9]+),`:
|
||||
- Record `activeMonitors[sessionID][taskID] = toolCallID`.
|
||||
- Override outgoing `ToolStatus` from `completed` to `in_progress`.
|
||||
- Pass through the rawOutput text unchanged so the card still shows the banner.
|
||||
|
||||
3. **Parse Monitor events** in `convertMessageChunk` (around line 976):
|
||||
- Compile once: `taskNotifRE = regexp.MustCompile((?s)Human:\s*<task-notification>\s*<task-id>([^<]+)</task-id>.*?<event>([^<]*)</event>\s*</task-notification>)`.
|
||||
- For each match:
|
||||
- Look up `activeMonitors[sessionID][taskID]` → toolCallID. If not found, leave the chunk alone (defensive — preserves text in case the parser ever misses).
|
||||
- Emit a synthetic `EventTypeToolUpdate` with the matched toolCallID, status `in_progress`, and the event body appended via the normalized payload's `content` field. Bump an event counter on the Monitor's normalized payload so the title can read e.g. "Monitor (3 events)".
|
||||
- Replace the matched block with empty string. If the remaining text is whitespace-only, return `nil` (drop the chunk). Otherwise emit a normal message chunk with the cleaned text.
|
||||
|
||||
4. **Defensive `Human:` echo filter** — same function, after parsing: if the cleaned text is `Human:` followed by a partial/unmatched `<` block (i.e., the regex didn't match but the prefix is suspicious), drop. Conservative pattern: `^Human:\s*(<[^>]*)?$`.
|
||||
|
||||
5. **Prompt-end sweep** — at the end of `Prompt()` (around line 614, alongside `cancelActiveToolCalls`):
|
||||
- For each remaining entry in `activeMonitors[sessionID]`, emit a synthetic terminal `tool_call_update(Monitor)` with `status=completed` and a final `content` line "Monitor exited" (so the card's terminal state has a clear marker).
|
||||
- Clear `activeMonitors[sessionID]`.
|
||||
|
||||
6. **Replay-time rebuild** for session restart resilience — in `LoadSession`, while `isLoadingSession` is true (line 817 onward), the existing logic *suppresses* replay notifications from going to clients. Add an inverse hook here: parse Monitor `tool_call`s and `tool_call_update`s out of the replay stream and reconstruct `activeMonitors[sessionID]`. Then, immediately after `isLoadingSession` is cleared (line 475), emit a synthetic `tool_call_update` with `status=cancelled` and content "Monitor ended (session restart)" for each entry, and clear the map. This handles the case where the agent process restarted and the actual scripts are dead.
|
||||
|
||||
### Backend orchestration changes
|
||||
|
||||
In **`apps/backend/internal/orchestrator/event_handlers_agent.go`** (or a sibling, depending on where session-recovery hooks live):
|
||||
|
||||
- After `RecoverInstances()` reconnects to a live agentctl, if any Monitor `tool_call`s in the DB are still in `in_progress`, re-broadcast their latest `tool_call_update` so frontends that connected during the gap window converge to the right state. Single SQL scan + WS publish.
|
||||
|
||||
### Frontend changes
|
||||
|
||||
In **`apps/web/components/task/chat/messages/`**:
|
||||
|
||||
- New `monitor-card.tsx` component, rendered when a tool_call has `toolName == "Monitor"`. Renders:
|
||||
- Title: `Monitor: <command-summary>` with a count badge (`N events`).
|
||||
- Status pill: `watching` | `ended` | `ended (session restart)` driven by latest `status`.
|
||||
- Inline event list, scrollable, newest-last.
|
||||
- Spinner while `status == "in_progress"`.
|
||||
- `use-processed-messages.ts` — Monitor `tool_call_update`s with `status=in_progress` are valid activity messages; ensure grouping logic at line 212 doesn't drop them.
|
||||
- `apps/web/lib/state/slices/session/` — accumulate Monitor events into the tool's normalized payload as they arrive so the card can read them straight from the store.
|
||||
|
||||
### Restart matrix
|
||||
|
||||
| Scenario | What restarts | How state recovers |
|
||||
|---|---|---|
|
||||
| Browser refresh | SPA only | SSR fetches messages from DB, `tool_call.status=in_progress` in DB drives the watching card. New events arrive over the WS once it reconnects. |
|
||||
| Backend restart | Go backend only | agentctl + adapter in-memory map survive (they live alongside the agent process). WS reconnect flushes the small buffered window. Orchestrator re-broadcasts in-progress Monitors on startup as a belt-and-braces. |
|
||||
| Agent process restart | agentctl + agent (e.g., container kill) | LoadSession replay rebuilds `activeMonitors`. Stale-sweep at end of replay marks them `cancelled (session restart)`. New prompts start fresh Monitor cards. |
|
||||
|
||||
## Test Strategy
|
||||
|
||||
### Mock agent extension
|
||||
|
||||
`apps/backend/cmd/mock-agent/handler.go` + `script.go` already supports `e2e:tool_use(...)`, `e2e:message(...)`, etc. Add:
|
||||
|
||||
- `e2e:monitor_start(taskId, command)` — emits `tool_call(Monitor)` then `tool_call_update(Monitor)` with the registration-banner rawOutput and `_meta.claudeCode.toolResponse.taskId`.
|
||||
- `e2e:monitor_event(taskId, body)` — emits an `agent_message_chunk` with the `Human: <task-notification>…</task-notification>` envelope.
|
||||
- `e2e:monitor_end(taskId)` — emits the final terminal update.
|
||||
|
||||
These three primitives let tests script the full wire contract without depending on the real Claude Code SDK.
|
||||
|
||||
### Go unit tests
|
||||
|
||||
Land in **`apps/backend/internal/agentctl/server/adapter/transport/acp/conversion_test.go`** following the existing table-driven `newTestAdapter()` + `drainEvents()` pattern. Cases:
|
||||
|
||||
1. Monitor registration update → status rewritten to `in_progress`; `activeMonitors` populated.
|
||||
2. Plain `agent_message_chunk` text passes through unchanged (regression guard).
|
||||
3. Single `<task-notification>` block in chunk → synthetic `tool_call_update` with event body; remaining text emitted only if non-empty.
|
||||
4. Multiple blocks in one chunk → one synthetic update each, in order.
|
||||
5. `<task-notification>` with unknown taskID → fall through, leave chunk text intact.
|
||||
6. Two concurrent Monitors → events route to the right toolCallID.
|
||||
7. `Human:` echo without closing tag → dropped.
|
||||
8. End of `Prompt()` with active Monitors → terminal updates emitted, map cleared.
|
||||
|
||||
In **`load_suppression_test.go`**: replay containing Monitor `tool_call`s rebuilds `activeMonitors`; post-replay sweep emits `cancelled (session restart)` updates.
|
||||
|
||||
In **`ordering_race_test.go`**: race between concurrent monitor_event and monitor_end (run with `-race`).
|
||||
|
||||
### Playwright E2E
|
||||
|
||||
New file **`apps/web/e2e/tests/chat/monitor.spec.ts`** using the `backend` + `apiClient` fixtures from `e2e/fixtures/`. Each test scripts the mock-agent via `monitor_start` / `monitor_event` / `monitor_end` directives.
|
||||
|
||||
1. **Watching state visible**: assert Monitor card with `data-testid="monitor-card"`, status text "watching", event count badge updates to 3 over time, raw `Human: <task-notification>` text never appears, terminal `monitor_end` flips status to "ended".
|
||||
2. **Page reload mid-monitor**: emit start + 2 events, `await page.reload()`, assert "watching" + 2 events still rendered, then emit 3rd event + end and assert they land. (Case A)
|
||||
3. **Backend restart mid-monitor**: same as above but `backend.restart()`. (Case B)
|
||||
4. **Agent process restart**: emit start + events, force a fresh `session/new` (mock-agent's `--resume false` path), assert original card flips to "ended (session restart)" and a new card appears for the next start. (Case C)
|
||||
5. **User message during Monitor**: type a message; assert queue indicator visible; let Monitor end; assert message is delivered as a new prompt and lands in chat history.
|
||||
|
||||
### Frontend component tests
|
||||
|
||||
Vitest test next to `monitor-card.tsx`: render with `(status, events[])` permutations, snapshot transitions. One case in `use-processed-messages.test.ts`: Monitor `in_progress` tool_call_update doesn't break activity grouping.
|
||||
|
||||
### Out of scope
|
||||
|
||||
- No integration test against the real `@agentclientprotocol/claude-agent-acp` (too slow, model-flaky). Mock-agent fully covers the wire contract.
|
||||
- No fuzz tests on the parser; grammar is small and wire format is not user-controlled.
|
||||
|
||||
## Implementation Notes
|
||||
|
||||
_To be filled after the PR lands. Capture any deviations from this plan, surprises in the wire format from real claude-agent-acp versions, and migration considerations if the Monitor envelope format changes upstream._
|
||||
@@ -7,3 +7,4 @@ Create new plans via `/record plan` or manually following the format below.
|
||||
| Plan | Status | Date | PR |
|
||||
|------|--------|------|----|
|
||||
| [Host utility agentctl](2026-04-host-utility-agentctl.md) | implemented | 2026-04-08 | TBD |
|
||||
| [Monitor tool ACP adapter support](2026-04-monitor-tool-acp-support.md) | proposed | 2026-04-25 | TBD |
|
||||
|
||||
Reference in New Issue
Block a user