From a2e9390e660e4236cff324ea5a4f8cb8da62d5a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Fri, 20 Feb 2026 21:56:21 +0000 Subject: [PATCH] feat: improve claude code --- apps/backend/internal/agent/agents/agent.go | 11 +- .../internal/agent/agents/claude_code.go | 32 +- apps/backend/internal/agent/agents/helpers.go | 11 + .../internal/agent/agents/helpers_test.go | 45 ++ .../server/adapter/transport/shared/config.go | 21 + .../adapter/transport/shared/config_test.go | 30 + .../server/adapter/transport/shared/debug.go | 3 + .../adapter/transport/shared/tracing.go | 35 ++ .../adapter/transport/streamjson/adapter.go | 21 + .../transport/streamjson/adapter_test.go | 589 ++++++++++++++++++ .../adapter/transport/streamjson/normalize.go | 38 +- .../transport/streamjson/streamjson_init.go | 60 +- .../streamjson/streamjson_messages.go | 189 +++++- .../streamjson/streamjson_permissions.go | 157 ++++- .../transport/streamjson/streamjson_prompt.go | 12 +- .../transport/streamjson/streamjson_result.go | 23 +- .../streamjson/streamjson_tracing.go | 56 ++ .../internal/agentctl/types/streams/agent.go | 11 + .../agentctl/types/streams/tool_payload.go | 9 + .../internal/orchestrator/event_handlers.go | 12 +- .../orchestrator/event_handlers_streaming.go | 15 +- apps/backend/internal/task/models/models.go | 1 + .../internal/task/repository/sqlite/base.go | 10 + .../task/repository/sqlite/executor.go | 12 +- apps/backend/pkg/claudecode/client.go | 31 +- apps/backend/pkg/claudecode/client_test.go | 79 +++ apps/backend/pkg/claudecode/types.go | 102 ++- apps/backend/pkg/claudecode/types_test.go | 248 +++++++- docs/agents/claude_code.md | 587 +++++++++++++++++ 29 files changed, 2343 insertions(+), 107 deletions(-) create mode 100644 apps/backend/internal/agent/agents/helpers_test.go create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/shared/config_test.go create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_tracing.go create mode 100644 docs/agents/claude_code.md diff --git a/apps/backend/internal/agent/agents/agent.go b/apps/backend/internal/agent/agents/agent.go index ab9112d79d..b88803b263 100644 --- a/apps/backend/internal/agent/agents/agent.go +++ b/apps/backend/internal/agent/agents/agent.go @@ -100,10 +100,13 @@ type DiscoveryCapabilities struct { // CommandOptions are passed to BuildCommand. type CommandOptions struct { - Model string - SessionID string // for --resume flag - AutoApprove bool - PermissionValues map[string]bool // e.g. {"allow_indexing": true} + Model string + SessionID string // for --resume flag + ResumeAtMessageUUID string // for --resume-session-at flag (truncate conversation) + AutoApprove bool + PermissionPolicy string // "autonomous", "supervised", "plan" + PermissionValues map[string]bool // e.g. {"allow_indexing": true} + AgentType string // for --agent flag (e.g. "task" for subagent) } // PassthroughOptions are passed to BuildPassthroughCommand. diff --git a/apps/backend/internal/agent/agents/claude_code.go b/apps/backend/internal/agent/agents/claude_code.go index 00b601428d..ab101a6aca 100644 --- a/apps/backend/internal/agent/agents/claude_code.go +++ b/apps/backend/internal/agent/agents/claude_code.go @@ -78,23 +78,39 @@ func (a *ClaudeCode) ListModels(ctx context.Context) (*ModelList, error) { } func (a *ClaudeCode) BuildCommand(opts CommandOptions) Command { - return Cmd("npx", "-y", "@anthropic-ai/claude-code@2.1.29", + b := Cmd("npx", "-y", "@anthropic-ai/claude-code@2.1.50", "-p", "--output-format=stream-json", "--input-format=stream-json", "--permission-prompt-tool=stdio", "--disallowedTools=AskUserQuestion", - "--setting-sources=user,project", "--verbose"). + "--setting-sources=user,project", "--verbose", + "--include-partial-messages", "--replay-user-messages"). Model(NewParam("--model", "{model}"), opts.Model). Resume(NewParam("--resume"), opts.SessionID, false). - Settings(claudeCodePermSettings, opts.PermissionValues). - Build() + ResumeAt(NewParam("--resume-session-at"), opts.ResumeAtMessageUUID) + + // When using supervised/plan mode with hooks, bypass the built-in permission + // system so hooks handle access control instead + switch opts.PermissionPolicy { + case "supervised", "plan": + b = b.Flag("--permission-mode", "bypassPermissions") + default: + b = b.Settings(claudeCodePermSettings, opts.PermissionValues) + } + + if opts.AgentType != "" { + b = b.Flag("--agent", opts.AgentType) + } + + return b.Build() } func (a *ClaudeCode) Runtime() *RuntimeConfig { canRecover := true return &RuntimeConfig{ - Cmd: Cmd("npx", "-y", "@anthropic-ai/claude-code@2.1.29", + Cmd: Cmd("npx", "-y", "@anthropic-ai/claude-code@2.1.50", "-p", "--output-format=stream-json", "--input-format=stream-json", "--permission-prompt-tool=stdio", "--disallowedTools=AskUserQuestion", - "--setting-sources=user,project", "--verbose").Build(), + "--setting-sources=user,project", "--verbose", + "--include-partial-messages", "--replay-user-messages").Build(), WorkingDir: "{workspace}", RequiredEnv: []string{"ANTHROPIC_API_KEY"}, Env: map[string]string{}, @@ -122,6 +138,10 @@ var claudeCodePermSettings = map[string]PermissionSetting{ Supported: true, Default: true, Label: "Skip Permissions", Description: "Bypass all permission checks (dangerous but fast for trusted tasks)", ApplyMethod: "cli_flag", CLIFlag: "--dangerously-skip-permissions", }, + "permission_policy": { + Supported: true, Default: false, Label: "Permission Policy", Description: "Control permission mode: autonomous (default), supervised (approve writes), plan (approve plan exit)", + ApplyMethod: "custom", + }, } func claudeCodeStaticModels() []Model { diff --git a/apps/backend/internal/agent/agents/helpers.go b/apps/backend/internal/agent/agents/helpers.go index 04ef99c9b4..71521133d1 100644 --- a/apps/backend/internal/agent/agents/helpers.go +++ b/apps/backend/internal/agent/agents/helpers.go @@ -42,6 +42,17 @@ func (b *CmdBuilder) Resume(flag Param, sessionID string, nativeResume bool) *Cm return b } +// ResumeAt appends a --resume-session-at flag with the message UUID. +// Skipped when uuid is empty or flag is empty. +func (b *CmdBuilder) ResumeAt(flag Param, uuid string) *CmdBuilder { + if uuid == "" || flag.IsEmpty() { + return b + } + b.args = append(b.args, flag.args...) + b.args = append(b.args, uuid) + return b +} + // Permissions appends per-tool permission flags when not auto-approving. // Each tool gets: flag tool:ask-user (e.g. --permission launch-process:ask-user). func (b *CmdBuilder) Permissions(flag string, tools []string, opts CommandOptions) *CmdBuilder { diff --git a/apps/backend/internal/agent/agents/helpers_test.go b/apps/backend/internal/agent/agents/helpers_test.go new file mode 100644 index 0000000000..50a863c721 --- /dev/null +++ b/apps/backend/internal/agent/agents/helpers_test.go @@ -0,0 +1,45 @@ +package agents + +import ( + "testing" +) + +func TestResumeAt_Empty(t *testing.T) { + cmd := Cmd("agent").ResumeAt(NewParam("--resume-session-at"), "").Build() + args := cmd.Args() + for _, arg := range args { + if arg == "--resume-session-at" { + t.Error("ResumeAt should not append flag when uuid is empty") + } + } +} + +func TestResumeAt_WithUUID(t *testing.T) { + cmd := Cmd("agent").ResumeAt(NewParam("--resume-session-at"), "msg-uuid-123").Build() + args := cmd.Args() + + foundFlag := false + for i, arg := range args { + if arg == "--resume-session-at" { + foundFlag = true + if i+1 >= len(args) { + t.Fatal("--resume-session-at flag has no value") + } + if args[i+1] != "msg-uuid-123" { + t.Errorf("--resume-session-at value = %q, want %q", args[i+1], "msg-uuid-123") + } + break + } + } + if !foundFlag { + t.Error("--resume-session-at flag not found in args") + } +} + +func TestResumeAt_EmptyFlag(t *testing.T) { + cmd := Cmd("agent").ResumeAt(Param{}, "msg-uuid-123").Build() + args := cmd.Args() + if len(args) != 1 || args[0] != "agent" { + t.Errorf("ResumeAt with empty flag should not modify args, got %v", args) + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/shared/config.go b/apps/backend/internal/agentctl/server/adapter/transport/shared/config.go index 11b94498e6..cfbd2e46c9 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/shared/config.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/shared/config.go @@ -1,5 +1,10 @@ package shared +import "time" + +// DefaultPermissionTimeout is the default timeout for permission requests (5 minutes). +const DefaultPermissionTimeout = 5 * time.Minute + // Config holds configuration for creating transport adapters. // This is passed to transport adapters by the factory. type Config struct { @@ -14,6 +19,14 @@ type Config struct { // Defaults to "on-request" if empty. ApprovalPolicy string + // PermissionPolicy controls the permission mode: "autonomous", "supervised", "plan". + // Used to determine hook registration and --permission-mode flag. + PermissionPolicy string + + // PermissionTimeout is the maximum time to wait for a permission response. + // After timeout, the request is auto-denied with interrupt. Defaults to DefaultPermissionTimeout. + PermissionTimeout time.Duration + // McpServers is a list of MCP servers to configure for the agent McpServers []McpServerConfig @@ -35,6 +48,14 @@ type Config struct { Extra map[string]string } +// GetPermissionTimeout returns the configured permission timeout or the default. +func (c *Config) GetPermissionTimeout() time.Duration { + if c.PermissionTimeout > 0 { + return c.PermissionTimeout + } + return DefaultPermissionTimeout +} + // McpServerConfig holds configuration for an MCP server. type McpServerConfig struct { // Name is the human-readable name of the MCP server diff --git a/apps/backend/internal/agentctl/server/adapter/transport/shared/config_test.go b/apps/backend/internal/agentctl/server/adapter/transport/shared/config_test.go new file mode 100644 index 0000000000..fc915c115f --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/shared/config_test.go @@ -0,0 +1,30 @@ +package shared + +import ( + "testing" + "time" +) + +func TestGetPermissionTimeout_Default(t *testing.T) { + cfg := &Config{} + got := cfg.GetPermissionTimeout() + if got != DefaultPermissionTimeout { + t.Errorf("GetPermissionTimeout() = %v, want %v", got, DefaultPermissionTimeout) + } +} + +func TestGetPermissionTimeout_Custom(t *testing.T) { + cfg := &Config{PermissionTimeout: 30 * time.Second} + got := cfg.GetPermissionTimeout() + if got != 30*time.Second { + t.Errorf("GetPermissionTimeout() = %v, want %v", got, 30*time.Second) + } +} + +func TestGetPermissionTimeout_Zero(t *testing.T) { + cfg := &Config{PermissionTimeout: 0} + got := cfg.GetPermissionTimeout() + if got != DefaultPermissionTimeout { + t.Errorf("GetPermissionTimeout() = %v, want %v (default)", got, DefaultPermissionTimeout) + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go b/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go index 505d910c7b..b9091d1675 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/shared/debug.go @@ -38,6 +38,9 @@ const ( ProtocolCopilot = "copilot" ) +// DebugEnabled reports whether agent message debug mode is active. +func DebugEnabled() bool { return debugMode } + func resolveDebugLogDir() string { if dir := os.Getenv("KANDEV_DEBUG_LOG_DIR"); dir != "" { return dir diff --git a/apps/backend/internal/agentctl/server/adapter/transport/shared/tracing.go b/apps/backend/internal/agentctl/server/adapter/transport/shared/tracing.go index c88d12549f..38be906d51 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/shared/tracing.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/shared/tracing.go @@ -100,6 +100,41 @@ func TraceProtocolRequest( return ctx, span } +// TraceControlMessage creates a span for a control-plane protocol message. +// Used for control_request, control_response, and control_cancel_request messages +// that bypass the regular sendUpdate() tracing path. +func TraceControlMessage( + ctx context.Context, + protocol, agentID string, + direction string, + eventType string, + rawData json.RawMessage, +) { + tracer := Tracer() + spanName := protocol + ".control." + eventType + + kind := trace.SpanKindInternal + if direction == "send" { + kind = trace.SpanKindClient + } + + _, span := tracer.Start(ctx, spanName, trace.WithSpanKind(kind)) + defer span.End() + + span.SetAttributes( + attribute.String("protocol", protocol), + attribute.String("agent_id", agentID), + attribute.String("direction", direction), + attribute.String("event_type", eventType), + ) + + if len(rawData) > 0 { + span.AddEvent("payload", trace.WithAttributes( + attribute.String("data", truncate(string(rawData), maxAttrValueLen)), + )) + } +} + func truncate(s string, maxLen int) string { if len(s) <= maxLen { return s diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go index c547416e91..aef527fe7d 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter.go @@ -97,6 +97,10 @@ type Adapter struct { // and reset to false at the start of each prompt. streamingTextSentThisTurn bool + // Message UUID tracking for --resume-session-at support + lastMessageUUID string // Latest committed message UUID (safe to resume from) + pendingAssistantUUID string // UUID from assistant message, committed on result + // lastRawData holds the raw JSON of the current message being processed. // Set in handleMessage before dispatch; used by sendUpdate for OTel tracing. lastRawData json.RawMessage @@ -263,6 +267,20 @@ func (a *Adapter) handleMessage(msg *claudecode.CLIMessage) { shared.LogRawEvent(shared.ProtocolStreamJSON, a.agentID, msg.Type, a.lastRawData) } + // Update session ID from any message type. + // Claude Code can change session_id mid-conversation (e.g. on compact). + if msg.SessionID != "" { + a.mu.Lock() + if a.sessionID != msg.SessionID { + a.logger.Info("session ID updated", + zap.String("old", a.sessionID), + zap.String("new", msg.SessionID), + zap.String("message_type", msg.Type)) + a.sessionID = msg.SessionID + } + a.mu.Unlock() + } + a.mu.RLock() sessionID := a.sessionID operationID := a.operationID @@ -281,6 +299,9 @@ func (a *Adapter) handleMessage(msg *claudecode.CLIMessage) { case claudecode.MessageTypeResult: a.handleResultMessage(msg, sessionID, operationID) + case claudecode.MessageTypeRateLimit: + a.handleRateLimitMessage(msg, sessionID, operationID) + default: a.logger.Debug("unhandled message type", zap.String("type", msg.Type)) } diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter_test.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter_test.go index 4e27b8f511..8106d21bf6 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/adapter_test.go @@ -6,6 +6,7 @@ import ( "github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/pkg/claudecode" ) func newTestLogger(t *testing.T) *logger.Logger { @@ -145,3 +146,591 @@ func TestPrepareCommandArgs_MultipleServers(t *testing.T) { t.Error("missing 'other' server") } } + +func TestBuildHooks_Autonomous(t *testing.T) { + tests := []struct { + name string + policy string + }{ + {"empty policy", ""}, + {"explicit autonomous", "autonomous"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + a := NewAdapter(&shared.Config{ + AgentID: "claude-code", + PermissionPolicy: tt.policy, + }, newTestLogger(t)) + + hooks := a.buildHooks() + if hooks != nil { + t.Errorf("buildHooks() = %v, want nil for autonomous mode", hooks) + } + }) + } +} + +func TestBuildHooks_Supervised(t *testing.T) { + a := NewAdapter(&shared.Config{ + AgentID: "claude-code", + PermissionPolicy: "supervised", + }, newTestLogger(t)) + + hooks := a.buildHooks() + if hooks == nil { + t.Fatal("buildHooks() returned nil, want hooks") + } + + // Should have PreToolUse + preToolUse, ok := hooks["PreToolUse"] + if !ok { + t.Fatal("buildHooks() missing PreToolUse key") + } + entries, ok := preToolUse.([]claudecode.HookEntry) + if !ok { + t.Fatalf("PreToolUse is %T, want []claudecode.HookEntry", preToolUse) + } + if len(entries) != 1 { + t.Fatalf("PreToolUse has %d entries, want 1", len(entries)) + } + if entries[0].HookCallbackIDs[0] != "tool_approval" { + t.Errorf("PreToolUse[0].HookCallbackIDs[0] = %q, want %q", entries[0].HookCallbackIDs[0], "tool_approval") + } + + // Should have Stop + stop, ok := hooks["Stop"] + if !ok { + t.Fatal("buildHooks() missing Stop key") + } + stopEntries, ok := stop.([]claudecode.HookEntry) + if !ok { + t.Fatalf("Stop is %T, want []claudecode.HookEntry", stop) + } + if len(stopEntries) != 1 { + t.Fatalf("Stop has %d entries, want 1", len(stopEntries)) + } + if stopEntries[0].HookCallbackIDs[0] != "stop_git_check" { + t.Errorf("Stop[0].HookCallbackIDs[0] = %q, want %q", stopEntries[0].HookCallbackIDs[0], "stop_git_check") + } +} + +func TestBuildHooks_Plan(t *testing.T) { + a := NewAdapter(&shared.Config{ + AgentID: "claude-code", + PermissionPolicy: "plan", + }, newTestLogger(t)) + + hooks := a.buildHooks() + if hooks == nil { + t.Fatal("buildHooks() returned nil, want hooks") + } + + // Plan mode should have 2 PreToolUse entries: ExitPlanMode approval + auto-approve rest + preToolUse, ok := hooks["PreToolUse"] + if !ok { + t.Fatal("buildHooks() missing PreToolUse key") + } + entries, ok := preToolUse.([]claudecode.HookEntry) + if !ok { + t.Fatalf("PreToolUse is %T, want []claudecode.HookEntry", preToolUse) + } + if len(entries) != 2 { + t.Fatalf("PreToolUse has %d entries, want 2", len(entries)) + } + + // First entry: ExitPlanMode requires approval + if entries[0].Matcher != `^ExitPlanMode$` { + t.Errorf("entries[0].Matcher = %q, want %q", entries[0].Matcher, `^ExitPlanMode$`) + } + if entries[0].HookCallbackIDs[0] != "tool_approval" { + t.Errorf("entries[0].HookCallbackIDs[0] = %q, want %q", entries[0].HookCallbackIDs[0], "tool_approval") + } + + // Second entry: everything else auto-approved + if entries[1].HookCallbackIDs[0] != "auto_approve" { + t.Errorf("entries[1].HookCallbackIDs[0] = %q, want %q", entries[1].HookCallbackIDs[0], "auto_approve") + } + + // Should have Stop + if _, ok := hooks["Stop"]; !ok { + t.Error("buildHooks() missing Stop key") + } +} + +func TestSendControlResult_SendsAndTraces(t *testing.T) { + // Create adapter with a mock stdin to capture output + var buf syncBuf + a := NewAdapter(&shared.Config{AgentID: "test-agent"}, newTestLogger(t)) + a.stdin = &buf + + // Wire up a client that writes to our buffer + a.client = claudecode.NewClient(&buf, &emptyReader{}, newTestLogger(t)) + + a.sendControlResult("req-123", &claudecode.PermissionResult{ + Behavior: claudecode.BehaviorAllow, + }, "test_trace_event") + + // Verify something was written to stdin + if buf.Len() == 0 { + t.Error("sendControlResult did not write to stdin") + } + + // Parse the sent message + var resp claudecode.ControlResponseMessage + if err := json.Unmarshal(buf.Bytes(), &resp); err != nil { + t.Fatalf("failed to parse sent response: %v", err) + } + if resp.RequestID != "req-123" { + t.Errorf("RequestID = %q, want %q", resp.RequestID, "req-123") + } + if resp.Response == nil { + t.Fatal("Response is nil") + } + if resp.Response.Subtype != "success" { + t.Errorf("Response.Subtype = %q, want %q", resp.Response.Subtype, "success") + } +} + +// --- handleUserMessage tests --- + +// drainEvents collects all pending events from the adapter's update channel. +func drainEvents(a *Adapter) []AgentEvent { + var events []AgentEvent + for { + select { + case ev := <-a.updatesCh: + events = append(events, ev) + default: + return events + } + } +} + +func TestHandleUserMessage_ReplaySkipped(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + a.sessionID = "sess-1" + a.operationID = "op-1" + + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeUser, + UUID: "msg-uuid-1", + IsReplay: true, + Message: &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(`"hello"`), + }, + } + + a.handleUserMessage(msg, "sess-1", "op-1") + + events := drainEvents(a) + if len(events) != 0 { + t.Errorf("expected 0 events for replay message, got %d", len(events)) + } + + // UUID should still be tracked for --resume-session-at + a.mu.RLock() + lastUUID := a.lastMessageUUID + a.mu.RUnlock() + if lastUUID != "msg-uuid-1" { + t.Errorf("lastMessageUUID = %q, want %q", lastUUID, "msg-uuid-1") + } +} + +func TestHandleUserMessage_EchoedPromptSkipped(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeUser, + UUID: "msg-uuid-2", + Message: &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(`"hello world"`), + }, + } + + a.handleUserMessage(msg, "sess-1", "op-1") + + events := drainEvents(a) + if len(events) != 0 { + t.Errorf("expected 0 events for echoed prompt, got %d", len(events)) + } +} + +func TestHandleUserMessage_SlashCommandOutputSkipped(t *testing.T) { + // Slash command output arrives as string content in a user message. + // It is always isReplay=true in practice, but even without that flag, + // string-content user messages are skipped — the output is delivered + // via the result message instead. + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeUser, + UUID: "msg-uuid-3", + Message: &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(`"compact summary here"`), + }, + } + + a.handleUserMessage(msg, "sess-1", "op-1") + + events := drainEvents(a) + if len(events) != 0 { + t.Errorf("expected 0 events for slash command output (delivered via result), got %d", len(events)) + } +} + +func TestHandleUserMessage_ToolResultStillProcessed(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + // Pre-register a pending tool call + a.pendingToolCalls["tool-123"] = nil + + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeUser, + UUID: "msg-uuid-4", + Message: &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(`[{"type":"tool_result","tool_use_id":"tool-123","content":"output"}]`), + }, + } + + a.handleUserMessage(msg, "sess-1", "op-1") + + events := drainEvents(a) + if len(events) != 1 { + t.Fatalf("expected 1 event for tool result, got %d", len(events)) + } + if events[0].Type != "tool_update" { + t.Errorf("event type = %q, want %q", events[0].Type, "tool_update") + } + if events[0].ToolCallID != "tool-123" { + t.Errorf("tool_call_id = %q, want %q", events[0].ToolCallID, "tool-123") + } +} + +func TestHandleUserMessage_SystemPromptEchoSkipped(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + // Simulate the real bug: Claude Code echoes back the full prompt including kandev-system tags + content := `"MCP TOOLS info\n\nhello"` + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeUser, + UUID: "msg-uuid-5", + Message: &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(content), + }, + } + + a.handleUserMessage(msg, "sess-1", "op-1") + + events := drainEvents(a) + if len(events) != 0 { + t.Errorf("expected 0 events for system prompt echo, got %d (text: %q)", len(events), events[0].Text) + } +} + +// --- handleMessage session_id tracking tests --- + +func TestHandleMessage_SessionIDTracking(t *testing.T) { + tests := []struct { + name string + initialSession string + msgType string + msgSessionID string + wantSessionID string + wantSessionSame bool + }{ + { + name: "system message sets session ID", + msgType: claudecode.MessageTypeSystem, + msgSessionID: "new-session-1", + wantSessionID: "new-session-1", + }, + { + name: "user message updates session ID", + initialSession: "old-session", + msgType: claudecode.MessageTypeUser, + msgSessionID: "new-session-2", + wantSessionID: "new-session-2", + }, + { + name: "assistant message updates session ID", + initialSession: "old-session", + msgType: claudecode.MessageTypeAssistant, + msgSessionID: "compacted-session", + wantSessionID: "compacted-session", + }, + { + name: "result message updates session ID", + initialSession: "old-session", + msgType: claudecode.MessageTypeResult, + msgSessionID: "result-session", + wantSessionID: "result-session", + }, + { + name: "empty session ID does not overwrite", + initialSession: "keep-this", + msgType: claudecode.MessageTypeUser, + msgSessionID: "", + wantSessionID: "keep-this", + }, + { + name: "same session ID is a no-op", + initialSession: "same-session", + msgType: claudecode.MessageTypeAssistant, + msgSessionID: "same-session", + wantSessionID: "same-session", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + if tt.initialSession != "" { + a.sessionID = tt.initialSession + } + + msg := &claudecode.CLIMessage{ + Type: tt.msgType, + SessionID: tt.msgSessionID, + } + + // handleMessage needs a message body for user/assistant types to avoid nil panics + if tt.msgType == claudecode.MessageTypeUser || tt.msgType == claudecode.MessageTypeAssistant { + msg.Message = &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(`"test"`), + } + } + // Result messages need a result field + if tt.msgType == claudecode.MessageTypeResult { + msg.Result = json.RawMessage(`"done"`) + } + + a.handleMessage(msg) + drainEvents(a) // discard any events + + a.mu.RLock() + gotSessionID := a.sessionID + a.mu.RUnlock() + + if gotSessionID != tt.wantSessionID { + t.Errorf("sessionID = %q, want %q", gotSessionID, tt.wantSessionID) + } + }) + } +} + +// syncBuf is a thread-safe buffer for tests. +type syncBuf struct { + data []byte +} + +func (b *syncBuf) Write(p []byte) (int, error) { + b.data = append(b.data, p...) + return len(p), nil +} + +func (b *syncBuf) Len() int { return len(b.data) } +func (b *syncBuf) Bytes() []byte { return b.data } + +// emptyReader always returns EOF. +type emptyReader struct{} + +func (e *emptyReader) Read(p []byte) (int, error) { return 0, nil } + +// --- TodoWrite normalizer tests --- + +func TestNormalizeTodoWrite_ExtractsTodos(t *testing.T) { + n := NewNormalizer() + + // TodoWrite uses "todos" array with "content" (not "items" with "description") + args := map[string]any{ + "todos": []any{ + map[string]any{ + "content": "Count lines in config.yaml", + "status": "pending", + "activeForm": "Counting lines in config.yaml", + }, + map[string]any{ + "content": "Count lines in Makefile", + "status": "completed", + "activeForm": "Counting lines in Makefile", + }, + }, + } + + payload := n.NormalizeToolCall("TodoWrite", args) + if payload.Kind() != "manage_todos" { + t.Fatalf("kind = %q, want %q", payload.Kind(), "manage_todos") + } + todos := payload.ManageTodos() + if todos.Operation != "write" { + t.Errorf("operation = %q, want %q", todos.Operation, "write") + } + if len(todos.Items) != 2 { + t.Fatalf("items count = %d, want 2", len(todos.Items)) + } + if todos.Items[0].Description != "Count lines in config.yaml" { + t.Errorf("items[0].Description = %q, want %q", todos.Items[0].Description, "Count lines in config.yaml") + } + if todos.Items[0].Status != "pending" { + t.Errorf("items[0].Status = %q, want %q", todos.Items[0].Status, "pending") + } + if todos.Items[0].ActiveForm != "Counting lines in config.yaml" { + t.Errorf("items[0].ActiveForm = %q, want %q", todos.Items[0].ActiveForm, "Counting lines in config.yaml") + } + if todos.Items[1].Status != "completed" { + t.Errorf("items[1].Status = %q, want %q", todos.Items[1].Status, "completed") + } +} + +// --- tool_use_result enrichment tests --- + +func TestHandleUserMessage_SubagentToolUseResult(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + // Pre-register a pending Task tool call + n := NewNormalizer() + taskPayload := n.NormalizeToolCall("Task", map[string]any{ + "description": "Count lines", + "prompt": "Count lines in file", + "subagent_type": "Bash", + }) + a.pendingToolCalls["task-1"] = taskPayload + + // Sub-agent result with array content and tool_use_result metadata + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeUser, + UUID: "msg-1", + Message: &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(`[{"type":"tool_result","tool_use_id":"task-1","content":[{"type":"text","text":"61 config.yaml"},{"type":"text","text":"agentId: abc123"}]}]`), + }, + ToolUseResult: json.RawMessage(`{"status":"completed","agentId":"abc123","totalDurationMs":1609,"totalTokens":4540,"totalToolUseCount":1}`), + } + + a.handleUserMessage(msg, "sess-1", "op-1") + + events := drainEvents(a) + if len(events) != 1 { + t.Fatalf("expected 1 event, got %d", len(events)) + } + + payload := events[0].NormalizedPayload + if payload == nil { + t.Fatal("payload is nil") + } + st := payload.SubagentTask() + if st == nil { + t.Fatal("SubagentTask is nil") + } + if st.Status != "completed" { + t.Errorf("status = %q, want %q", st.Status, "completed") + } + if st.AgentID != "abc123" { + t.Errorf("agentID = %q, want %q", st.AgentID, "abc123") + } + if st.DurationMs != 1609 { + t.Errorf("durationMs = %d, want %d", st.DurationMs, 1609) + } + if st.TotalTokens != 4540 { + t.Errorf("totalTokens = %d, want %d", st.TotalTokens, 4540) + } + if st.ToolUseCount != 1 { + t.Errorf("toolUseCount = %d, want %d", st.ToolUseCount, 1) + } +} + +func TestHandleUserMessage_TodoWriteToolUseResult(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + // Pre-register a pending TodoWrite tool call + n := NewNormalizer() + todoPayload := n.NormalizeToolCall("TodoWrite", map[string]any{ + "todos": []any{ + map[string]any{"content": "Task A", "status": "pending"}, + }, + }) + a.pendingToolCalls["todo-1"] = todoPayload + + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeUser, + UUID: "msg-2", + Message: &claudecode.AssistantMessage{ + Role: "user", + Content: json.RawMessage(`[{"type":"tool_result","tool_use_id":"todo-1","content":"Todos modified"}]`), + }, + ToolUseResult: json.RawMessage(`{"oldTodos":[],"newTodos":[{"content":"Task A","status":"completed","activeForm":"Working on Task A"}]}`), + } + + a.handleUserMessage(msg, "sess-1", "op-1") + + events := drainEvents(a) + if len(events) != 1 { + t.Fatalf("expected 1 event, got %d", len(events)) + } + + payload := events[0].NormalizedPayload + if payload == nil || payload.ManageTodos() == nil { + t.Fatal("payload or ManageTodos is nil") + } + items := payload.ManageTodos().Items + if len(items) != 1 { + t.Fatalf("items count = %d, want 1", len(items)) + } + if items[0].Description != "Task A" { + t.Errorf("items[0].Description = %q, want %q", items[0].Description, "Task A") + } + if items[0].Status != "completed" { + t.Errorf("items[0].Status = %q, want %q", items[0].Status, "completed") + } + if items[0].ActiveForm != "Working on Task A" { + t.Errorf("items[0].ActiveForm = %q, want %q", items[0].ActiveForm, "Working on Task A") + } +} + +// --- system task_started subtype test --- + +func TestHandleSystemMessage_TaskStartedSkipped(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + // Simulate the system task_started message that Claude Code sends for sub-agents + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeSystem, + Subtype: "task_started", + SessionID: "sess-1", + } + + a.handleSystemMessage(msg) + + events := drainEvents(a) + if len(events) != 0 { + t.Errorf("expected 0 events for task_started system message, got %d", len(events)) + } +} + +func TestHandleSystemMessage_InitStillWorks(t *testing.T) { + a := NewAdapter(&shared.Config{AgentID: "test"}, newTestLogger(t)) + + // Normal init system message (subtype empty) + msg := &claudecode.CLIMessage{ + Type: claudecode.MessageTypeSystem, + SessionID: "sess-1", + SessionStatus: "new", + } + + a.handleSystemMessage(msg) + + events := drainEvents(a) + if len(events) != 1 { + t.Fatalf("expected 1 event for init system message, got %d", len(events)) + } + if events[0].Type != "session_status" { + t.Errorf("event type = %q, want %q", events[0].Type, "session_status") + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/normalize.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/normalize.go index 81468ba6a8..cdb6e2e2d9 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/normalize.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/normalize.go @@ -227,6 +227,8 @@ func (n *Normalizer) normalizeCreateTask(args map[string]any) *streams.Normalize } // normalizeManageTodos converts stream-json TaskUpdate/TaskList/TodoWrite tool data. +// TodoWrite uses {"todos": [{content, status, activeForm}]} while +// TaskUpdate uses {"items": [{id, description, status}]}. func (n *Normalizer) normalizeManageTodos(toolName string, args map[string]any) *streams.NormalizedPayload { operation := "update" switch toolName { @@ -236,21 +238,31 @@ func (n *Normalizer) normalizeManageTodos(toolName string, args map[string]any) operation = "write" } - // Extract items if present - var items []streams.TodoItem - if rawItems, ok := args["items"].([]any); ok { - for _, item := range rawItems { - if itemMap, ok := item.(map[string]any); ok { - items = append(items, streams.TodoItem{ - ID: shared.GetString(itemMap, "id"), - Description: shared.GetString(itemMap, "description"), - Status: shared.GetString(itemMap, "status"), - }) - } - } + // Try "todos" first (TodoWrite format), then "items" (TaskUpdate format) + rawItems, ok := args["todos"].([]any) + if !ok { + rawItems, _ = args["items"].([]any) + } + + var items []streams.TodoItem + for _, item := range rawItems { + itemMap, ok := item.(map[string]any) + if !ok { + continue + } + // TodoWrite uses "content" for the description text + desc := shared.GetString(itemMap, "description") + if desc == "" { + desc = shared.GetString(itemMap, "content") + } + items = append(items, streams.TodoItem{ + ID: shared.GetString(itemMap, "id"), + Description: desc, + Status: shared.GetString(itemMap, "status"), + ActiveForm: shared.GetString(itemMap, "activeForm"), + }) } - // Use factory function return streams.NewManageTodos(operation, items) } diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_init.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_init.go index 785840c800..7cc29d56da 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_init.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_init.go @@ -5,6 +5,7 @@ import ( "encoding/json" "time" + "github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared" "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/pkg/claudecode" "go.uber.org/zap" @@ -81,6 +82,7 @@ func (a *Adapter) Initialize(ctx context.Context) error { // Create Claude Code client a.client = claudecode.NewClient(a.stdin, a.stdout, a.logger) a.client.SetRequestHandler(a.handleControlRequest) + a.client.SetCancelHandler(a.handleControlCancel) a.client.SetMessageHandler(a.handleMessage) // Start reading from stdout with the adapter's context @@ -101,9 +103,19 @@ func (a *Adapter) Initialize(ctx context.Context) error { Version: "unknown", } + // Build hooks based on permission policy + hooks := a.buildHooks() + // Send initialize control request to get slash commands // This is required for streaming mode (input-format=stream-json) - initResp, err := a.client.Initialize(ctx, 60*time.Second) + initCtx, initSpan := shared.TraceProtocolRequest(ctx, shared.ProtocolStreamJSON, a.agentID, "initialize") + a.traceOutgoingSendCtx(initCtx, "initialize", map[string]any{ + "type": claudecode.MessageTypeControlRequest, + "request": map[string]any{"subtype": claudecode.SubtypeInitialize, "hooks": hooks}, + }) + initResp, err := a.client.Initialize(ctx, 60*time.Second, hooks) + initSpan.End() + if err != nil { a.logger.Warn("failed to initialize (continuing anyway)", zap.Error(err)) } else if initResp != nil && len(initResp.Commands) > 0 { @@ -128,6 +140,52 @@ func (a *Adapter) Initialize(ctx context.Context) error { return nil } +// buildHooks constructs hook configuration based on the permission policy. +// Returns nil for autonomous mode (no hooks). +func (a *Adapter) buildHooks() map[string]any { + policy := a.cfg.PermissionPolicy + if policy == "" || policy == "autonomous" { + return nil + } + + cfg := &claudecode.HookConfig{} + + switch policy { + case "supervised": + // Match write/dangerous tools — everything except read-only tools. + // Non-matching tools (Glob, Grep, Read, Task, TodoWrite, etc.) are + // implicitly auto-approved by Claude Code when no hook matches. + cfg.PreToolUse = []claudecode.HookEntry{ + { + Matcher: `^(?!(Glob|Grep|NotebookRead|Read|Task|TodoWrite)$).*`, + HookCallbackIDs: []string{"tool_approval"}, + }, + } + case "plan": + // Only ExitPlanMode requires approval — all other tools run freely during planning. + // The auto_approve callback explicitly approves so Claude doesn't fall through. + cfg.PreToolUse = []claudecode.HookEntry{ + { + Matcher: `^ExitPlanMode$`, + HookCallbackIDs: []string{"tool_approval"}, + }, + { + Matcher: `^(?!ExitPlanMode$).*`, + HookCallbackIDs: []string{"auto_approve"}, + }, + } + } + + // Always register a Stop hook to check for uncommitted changes + cfg.Stop = []claudecode.HookEntry{ + { + HookCallbackIDs: []string{"stop_git_check"}, + }, + } + + return cfg.ToMap() +} + // GetAgentInfo returns information about the connected agent. func (a *Adapter) GetAgentInfo() *AgentInfo { return a.agentInfo diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_messages.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_messages.go index e416634d12..16f9cf20a9 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_messages.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_messages.go @@ -1,28 +1,37 @@ package streamjson import ( + "encoding/json" "fmt" - "strings" + "github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared" "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/pkg/claudecode" "go.uber.org/zap" ) -// handleSystemMessage processes system init messages. -// Note: System messages are session initialization, NOT turn completion. -// Turn completion is signaled by result messages. +// handleSystemMessage processes system messages. +// Subtypes: +// - "init" (or empty): Session initialization with slash commands and status +// - "task_started": Sub-agent task lifecycle event — logged and skipped +// +// Note: Turn completion is signaled by result messages, not system messages. func (a *Adapter) handleSystemMessage(msg *claudecode.CLIMessage) { + // Skip non-init system messages (e.g. task_started for sub-agent lifecycle) + if msg.Subtype != "" && msg.Subtype != "init" { + a.logger.Debug("skipping system message with subtype", + zap.String("subtype", msg.Subtype), + zap.String("session_id", msg.SessionID)) + return + } + a.logger.Info("received system message", zap.String("session_id", msg.SessionID), zap.String("status", msg.SessionStatus), zap.Int("slash_commands_count", len(msg.SlashCommands))) - // Update session ID if provided + // Note: session ID is updated centrally in handleMessage() for all message types. a.mu.Lock() - if msg.SessionID != "" { - a.sessionID = msg.SessionID - } alreadySent := a.sessionStatusSent a.sessionStatusSent = true a.mu.Unlock() @@ -159,6 +168,13 @@ func (a *Adapter) handleAssistantMessage(msg *claudecode.CLIMessage, sessionID, return } + // Track assistant message UUID (committed on result) + if msg.UUID != "" { + a.mu.Lock() + a.pendingAssistantUUID = msg.UUID + a.mu.Unlock() + } + // Extract parent tool use ID for subagent nesting parentToolUseID := msg.ParentToolUseID @@ -215,35 +231,64 @@ func (a *Adapter) handleAssistantMessage(msg *claudecode.CLIMessage, sessionID, } } -// handleUserMessage processes user messages containing tool results or slash command output. +// handleRateLimitMessage processes rate limit notifications from the Anthropic API. +func (a *Adapter) handleRateLimitMessage(msg *claudecode.CLIMessage, sessionID, operationID string) { + message := "Rate limited by API" + if len(msg.RateLimitInfo) > 0 { + message = string(msg.RateLimitInfo) + } + + a.logger.Warn("rate limit event received", + zap.String("session_id", sessionID), + zap.String("message", message)) + + a.sendUpdate(AgentEvent{ + Type: streams.EventTypeRateLimit, + SessionID: sessionID, + OperationID: operationID, + RateLimitMessage: message, + }) +} + +// handleUserMessage processes user messages containing tool results. // Claude Code sends tool results back as user messages with tool_result content blocks. -// For slash commands, content may be a plain string wrapped in tags. +// +// User message variants (from Claude Code CLI): +// - isReplay=true: Historical context echoed on resume — drop entirely +// - String content: Echoed user prompt or slash command output — skip +// (slash command output is delivered via the result message instead) +// - Content blocks with tool_result: Tool results — process normally func (a *Adapter) handleUserMessage(msg *claudecode.CLIMessage, sessionID, operationID string) { if msg.Message == nil { return } - // Check if content is a string (slash command output) + // Clear pending assistant UUID — a new user message means any previous + // assistant message's UUID should not be committed on the next result. + // User messages are immediately safe to reference for resume. + a.mu.Lock() + a.pendingAssistantUUID = "" + if msg.UUID != "" { + a.lastMessageUUID = msg.UUID + } + a.mu.Unlock() + + // Skip replay messages — historical context echoed on resume. + // UUID tracking above still fires for --resume-session-at support. + if msg.IsReplay { + a.logger.Debug("skipping replay user message", + zap.String("session_id", sessionID), + zap.String("uuid", msg.UUID)) + return + } + + // Skip string-content user messages (echoed prompts). + // Slash command output also arrives here but is always isReplay=true (handled above). + // The output is delivered via the result message instead. if contentStr := msg.Message.GetContentString(); contentStr != "" { - // Extract text from tags if present - text := contentStr - if strings.HasPrefix(text, "") && strings.HasSuffix(text, "") { - text = strings.TrimPrefix(text, "") - text = strings.TrimSuffix(text, "") - } - - if text != "" { - a.logger.Info("received user message with string content (slash command output)", - zap.String("session_id", sessionID), - zap.Int("content_length", len(text))) - - a.sendUpdate(AgentEvent{ - Type: streams.EventTypeMessageChunk, - SessionID: sessionID, - OperationID: operationID, - Text: text, - }) - } + a.logger.Debug("skipping echoed user message", + zap.String("session_id", sessionID), + zap.Int("content_length", len(contentStr))) return } @@ -261,8 +306,14 @@ func (a *Adapter) handleUserMessage(msg *claudecode.CLIMessage, sessionID, opera a.mu.Unlock() // Enrich payload with result content - if payload != nil && block.Content != "" { - a.normalizer.NormalizeToolResult(payload, block.Content) + contentStr := block.GetContentString() + if payload != nil && contentStr != "" { + a.normalizer.NormalizeToolResult(payload, contentStr) + } + + // Enrich payload with structured tool_use_result metadata + if payload != nil && len(msg.ToolUseResult) > 0 { + a.enrichFromToolUseResult(payload, msg.ToolUseResult) } // If there's an error, set the error flag on the payload @@ -289,3 +340,77 @@ func (a *Adapter) handleUserMessage(msg *claudecode.CLIMessage, sessionID, opera }) } } + +// enrichFromToolUseResult populates payload-specific result fields from the +// top-level tool_use_result JSON on user messages. +// +// Supported formats: +// - TodoWrite: {oldTodos, newTodos} → populates ManageTodos items from newTodos +// - Task (sub-agent): {status, agentId, totalDurationMs, totalTokens, totalToolUseCount} +func (a *Adapter) enrichFromToolUseResult(payload *streams.NormalizedPayload, raw json.RawMessage) { + var data map[string]any + if err := json.Unmarshal(raw, &data); err != nil { + return + } + + switch payload.Kind() { + case streams.ToolKindManageTodos: + a.enrichTodoResult(payload, data) + case streams.ToolKindSubagentTask: + a.enrichSubagentResult(payload, data) + } +} + +// enrichTodoResult populates ManageTodos items from the newTodos array in tool_use_result. +func (a *Adapter) enrichTodoResult(payload *streams.NormalizedPayload, data map[string]any) { + if payload.ManageTodos() == nil { + return + } + newTodos, ok := data["newTodos"].([]any) + if !ok || len(newTodos) == 0 { + return + } + items := make([]streams.TodoItem, 0, len(newTodos)) + for _, item := range newTodos { + itemMap, ok := item.(map[string]any) + if !ok { + continue + } + desc := shared.GetString(itemMap, "content") + if desc == "" { + desc = shared.GetString(itemMap, "description") + } + items = append(items, streams.TodoItem{ + ID: shared.GetString(itemMap, "id"), + Description: desc, + Status: shared.GetString(itemMap, "status"), + ActiveForm: shared.GetString(itemMap, "activeForm"), + }) + } + if len(items) > 0 { + payload.ManageTodos().Items = items + } +} + +// enrichSubagentResult populates SubagentTask result fields from tool_use_result. +func (a *Adapter) enrichSubagentResult(payload *streams.NormalizedPayload, data map[string]any) { + if payload.SubagentTask() == nil { + return + } + st := payload.SubagentTask() + if v := shared.GetString(data, "status"); v != "" { + st.Status = v + } + if v := shared.GetString(data, "agentId"); v != "" { + st.AgentID = v + } + if v, ok := data["totalDurationMs"].(float64); ok { + st.DurationMs = int64(v) + } + if v, ok := data["totalTokens"].(float64); ok { + st.TotalTokens = int64(v) + } + if v, ok := data["totalToolUseCount"].(float64); ok { + st.ToolUseCount = int(v) + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_permissions.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_permissions.go index b68b7afad4..f671766e16 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_permissions.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_permissions.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/kandev/kandev/internal/agentctl/types" + "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/pkg/claudecode" "go.uber.org/zap" ) @@ -16,6 +17,13 @@ func (a *Adapter) handleControlRequest(requestID string, req *claudecode.Control zap.String("subtype", req.Subtype), zap.String("tool_name", req.ToolName)) + // Trace incoming control request (bypasses handleMessage path) + a.traceIncomingControl("control_request."+req.Subtype, map[string]any{ + "type": "control_request", + "request_id": requestID, + "request": req, + }) + switch req.Subtype { case claudecode.SubtypeCanUseTool: a.handleToolPermission(requestID, req) @@ -81,13 +89,22 @@ func (a *Adapter) handleToolPermission(requestID string, req *claudecode.Control // We pass PendingID so the handler uses Claude Code's requestID // instead of generating a new one - this ensures the frontend and backend // use the same ID for response lookup. + // Build action details, including blocked paths if present + actionDetails := req.Input + if req.BlockedPaths != "" { + if actionDetails == nil { + actionDetails = make(map[string]any) + } + actionDetails["blocked_paths"] = req.BlockedPaths + } + permReq := &PermissionRequest{ SessionID: sessionID, ToolCallID: req.ToolUseID, Title: title, Options: options, ActionType: actionType, - ActionDetails: req.Input, + ActionDetails: actionDetails, PendingID: requestID, // Use Claude Code's requestID so response lookup works } @@ -99,10 +116,36 @@ func (a *Adapter) handleToolPermission(requestID string, req *claudecode.Control return } - // Call permission handler (blocking) - it will send the notification - ctx := context.Background() + // Call permission handler (blocking) with timeout to prevent indefinite hangs + timeout := a.cfg.GetPermissionTimeout() + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + resp, err := handler(ctx, permReq) if err != nil { + // Check if it was a timeout + if ctx.Err() == context.DeadlineExceeded { + a.logger.Warn("permission request timed out, auto-denying with interrupt", + zap.String("request_id", requestID), + zap.String("tool", req.ToolName), + zap.Duration("timeout", timeout)) + + // Emit cancellation event so frontend closes the dialog + a.sendUpdate(AgentEvent{ + Type: streams.EventTypePermissionCancelled, + PendingID: requestID, + }) + + // Deny with interrupt to stop the current operation + interruptFlag := true + a.sendControlResult(requestID, &claudecode.PermissionResult{ + Behavior: claudecode.BehaviorDeny, + Interrupt: &interruptFlag, + Message: "Permission request timed out", + }, "permission_response.timeout") + return + } + a.logger.Error("permission handler error", zap.Error(err)) a.sendPermissionResponse(requestID, claudecode.BehaviorDeny) return @@ -124,38 +167,112 @@ func (a *Adapter) handleToolPermission(requestID string, req *claudecode.Control a.sendPermissionResponse(requestID, behavior) } -// sendPermissionResponse sends a permission response to the agent. -func (a *Adapter) sendPermissionResponse(requestID string, behavior string) { +// sendControlResult sends a success control response with the given result payload and traces it. +// result can be a *PermissionResult, map[string]any (for hooks), or any other JSON-serializable value. +func (a *Adapter) sendControlResult(requestID string, result any, traceEvent string) { resp := &claudecode.ControlResponseMessage{ Type: claudecode.MessageTypeControlResponse, RequestID: requestID, Response: &claudecode.ControlResponse{ Subtype: "success", - Result: &claudecode.PermissionResult{ - Behavior: behavior, - }, + Result: result, }, } - if err := a.client.SendControlResponse(resp); err != nil { - a.logger.Warn("failed to send permission response", zap.Error(err)) + a.logger.Warn("failed to send control response", + zap.Error(err), zap.String("event", traceEvent)) } + a.traceOutgoingControl(traceEvent, resp) } -// handleHookCallback processes hook callback requests. +// sendPermissionResponse sends a permission response to the agent. +func (a *Adapter) sendPermissionResponse(requestID string, behavior string) { + a.sendControlResult(requestID, &claudecode.PermissionResult{ + Behavior: behavior, + }, "permission_response") +} + +// handleControlCancel processes control_cancel_request messages. +// This cancels a pending permission request, closing the dialog in the UI. +func (a *Adapter) handleControlCancel(requestID string) { + a.logger.Info("received control_cancel_request", + zap.String("request_id", requestID)) + + a.traceIncomingControl("control_cancel_request", map[string]any{ + "type": "control_cancel_request", + "request_id": requestID, + }) + + a.sendUpdate(AgentEvent{ + Type: streams.EventTypePermissionCancelled, + PendingID: requestID, + }) +} + +// handleHookCallback processes hook callback requests, dispatching based on callback ID. +// Hook responses use different payload formats than can_use_tool responses: +// - PreToolUse hooks: hookSpecificOutput with permissionDecision +// - Stop hooks: decision + optional reason func (a *Adapter) handleHookCallback(requestID string, req *claudecode.ControlRequest) { a.logger.Info("received hook callback", zap.String("request_id", requestID), + zap.String("callback_id", req.CallbackID), zap.String("hook_name", req.HookName)) - // For now, acknowledge hook callbacks with success - if err := a.client.SendControlResponse(&claudecode.ControlResponseMessage{ - Type: claudecode.MessageTypeControlResponse, - RequestID: requestID, - Response: &claudecode.ControlResponse{ - Subtype: "success", - }, - }); err != nil { - a.logger.Warn("failed to send hook callback response", zap.Error(err)) + switch req.CallbackID { + case "tool_approval": + // Respond with "ask" to trigger a regular can_use_tool permission request. + // Claude Code will then send a separate can_use_tool control_request. + a.sendPreToolUseHookResponse(requestID, "ask", "Requires user approval") + + case "auto_approve": + // Auto-approve via hookSpecificOutput + a.sendPreToolUseHookResponse(requestID, "allow", "Auto-approved by SDK") + + case "stop_git_check": + // Check for uncommitted changes before stopping + a.handleStopGitCheck(requestID, req) + + default: + a.logger.Warn("unknown hook callback ID, auto-approving", + zap.String("callback_id", req.CallbackID)) + a.sendPreToolUseHookResponse(requestID, "allow", "Auto-approved (unknown callback)") } } + +// sendPreToolUseHookResponse sends a PreToolUse hook callback response. +// permissionDecision can be "allow", "deny", or "ask" (triggers can_use_tool flow). +func (a *Adapter) sendPreToolUseHookResponse(requestID, permissionDecision, reason string) { + a.sendControlResult(requestID, map[string]any{ + "hookSpecificOutput": map[string]any{ + "hookEventName": "PreToolUse", + "permissionDecision": permissionDecision, + "permissionDecisionReason": reason, + }, + }, "hook_response.pre_tool_use") +} + +// handleStopGitCheck processes the stop_git_check hook callback. +// Checks if the stop hook is active and approves/blocks accordingly. +func (a *Adapter) handleStopGitCheck(requestID string, req *claudecode.ControlRequest) { + // If stop_hook_active is set, just approve (hook is being deactivated) + if active, ok := req.HookInput["stop_hook_active"].(bool); ok && active { + a.sendStopHookResponse(requestID, "approve", "") + return + } + + // For now, approve the stop — the workspace tracker integration + // can be added later to check for uncommitted changes + a.sendStopHookResponse(requestID, "approve", "") +} + +// sendStopHookResponse sends a Stop hook callback response with decision format. +func (a *Adapter) sendStopHookResponse(requestID, decision, reason string) { + result := map[string]any{ + "decision": decision, + } + if reason != "" { + result["reason"] = reason + } + a.sendControlResult(requestID, result, "hook_response.stop") +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_prompt.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_prompt.go index 1d442c05c3..30d9dd7971 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_prompt.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_prompt.go @@ -58,6 +58,12 @@ func (a *Adapter) Prompt(ctx context.Context, message string, attachments []v1.M promptSpan.End() }() + // Trace the outgoing user message payload + a.traceOutgoingSend("prompt", &claudecode.UserMessage{ + Type: claudecode.MessageTypeUser, + Message: claudecode.UserMessageBody{Role: "user", Content: finalMessage}, + }) + a.logger.Info("sending prompt", zap.String("session_id", sessionID), zap.String("operation_id", operationID), @@ -191,11 +197,13 @@ func (a *Adapter) Cancel(ctx context.Context) error { a.logger.Info("cancelling operation") // Send interrupt control request - return client.SendControlRequest(&claudecode.SDKControlRequest{ + req := &claudecode.SDKControlRequest{ Type: claudecode.MessageTypeControlRequest, RequestID: uuid.New().String(), Request: claudecode.SDKControlRequestBody{ Subtype: claudecode.SubtypeInterrupt, }, - }) + } + a.traceOutgoingControl("interrupt", req) + return client.SendControlRequest(req) } diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_result.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_result.go index 87f001b85e..1bb13c118c 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_result.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_result.go @@ -143,6 +143,15 @@ func (a *Adapter) handleResultMessage(msg *claudecode.CLIMessage, sessionID, ope a.streamingTextSentThisTurn = false a.mu.Unlock() + // Commit pending assistant UUID as last message UUID + a.mu.Lock() + if a.pendingAssistantUUID != "" { + a.lastMessageUUID = a.pendingAssistantUUID + a.pendingAssistantUUID = "" + } + lastMsgUUID := a.lastMessageUUID + a.mu.Unlock() + // Build error message from errors array if available errorMsg := a.extractResultErrorMsg(msg) @@ -158,12 +167,16 @@ func (a *Adapter) handleResultMessage(msg *claudecode.CLIMessage, sessionID, ope if len(msg.Errors) > 0 { completeData["errors"] = msg.Errors } + if lastMsgUUID != "" { + completeData["last_message_uuid"] = lastMsgUUID + } a.sendUpdate(AgentEvent{ - Type: streams.EventTypeComplete, - SessionID: sessionID, - OperationID: operationID, - Error: errorMsg, - Data: completeData, + Type: streams.EventTypeComplete, + SessionID: sessionID, + OperationID: operationID, + Error: errorMsg, + Data: completeData, + LastMessageUUID: lastMsgUUID, }) // Signal completion diff --git a/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_tracing.go b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_tracing.go new file mode 100644 index 0000000000..39ac764cac --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/streamjson/streamjson_tracing.go @@ -0,0 +1,56 @@ +package streamjson + +import ( + "context" + "encoding/json" + + "github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared" +) + +// traceIncomingControl traces and logs an incoming control-plane message. +// Used for control_request messages that bypass the regular handleMessage path. +func (a *Adapter) traceIncomingControl(eventType string, msg any) { + if !shared.DebugEnabled() { + return + } + rawData, _ := json.Marshal(msg) + shared.LogRawEvent(shared.ProtocolStreamJSON, a.agentID, eventType, rawData) + shared.TraceControlMessage(a.getPromptTraceCtx(), shared.ProtocolStreamJSON, a.agentID, + "recv", eventType, rawData) +} + +// traceOutgoingControl traces and logs an outgoing control-plane message. +// Used for control_response messages sent back to the Claude CLI. +func (a *Adapter) traceOutgoingControl(eventType string, msg any) { + if !shared.DebugEnabled() { + return + } + rawData, _ := json.Marshal(msg) + shared.LogRawEvent(shared.ProtocolStreamJSON, a.agentID, eventType, rawData) + shared.TraceControlMessage(a.getPromptTraceCtx(), shared.ProtocolStreamJSON, a.agentID, + "send", eventType, rawData) +} + +// traceOutgoingSend traces and logs an outgoing data-plane message to Claude Code. +// Uses the current prompt trace context as parent span. +func (a *Adapter) traceOutgoingSend(eventType string, msg any) { + if !shared.DebugEnabled() { + return + } + rawData, _ := json.Marshal(msg) + shared.LogRawEvent(shared.ProtocolStreamJSON, a.agentID, "send."+eventType, rawData) + shared.TraceControlMessage(a.getPromptTraceCtx(), shared.ProtocolStreamJSON, a.agentID, + "send", eventType, rawData) +} + +// traceOutgoingSendCtx is like traceOutgoingSend but uses an explicit context as parent span. +// Used for messages sent outside of a prompt lifecycle (e.g., initialize). +func (a *Adapter) traceOutgoingSendCtx(ctx context.Context, eventType string, msg any) { + if !shared.DebugEnabled() { + return + } + rawData, _ := json.Marshal(msg) + shared.LogRawEvent(shared.ProtocolStreamJSON, a.agentID, "send."+eventType, rawData) + shared.TraceControlMessage(ctx, shared.ProtocolStreamJSON, a.agentID, + "send", eventType, rawData) +} diff --git a/apps/backend/internal/agentctl/types/streams/agent.go b/apps/backend/internal/agentctl/types/streams/agent.go index 608feaa204..8cd95b500a 100644 --- a/apps/backend/internal/agentctl/types/streams/agent.go +++ b/apps/backend/internal/agentctl/types/streams/agent.go @@ -42,6 +42,9 @@ const ( // EventTypeSessionMode indicates the agent's session mode has changed. EventTypeSessionMode = "session_mode" + + // EventTypeRateLimit indicates the agent is being rate-limited by the API. + EventTypeRateLimit = "rate_limit" ) // Session status constants for EventTypeSessionStatus events. @@ -180,6 +183,14 @@ type AgentEvent struct { // ToolCallContents contains rich content produced by a tool call (diffs, text, terminals). ToolCallContents []ToolCallContentItem `json:"tool_call_contents,omitempty"` + // --- Rate limit fields (for "rate_limit" type) --- + + // RateLimitMessage contains a human-readable rate limit message from the API. + RateLimitMessage string `json:"rate_limit_message,omitempty"` + + // LastMessageUUID is the UUID of the last committed message, used for --resume-session-at. + LastMessageUUID string `json:"last_message_uuid,omitempty"` + // --- Session mode fields (for "session_mode" type) --- // CurrentModeID is the active session mode identifier (e.g., "code", "ask", "architect"). diff --git a/apps/backend/internal/agentctl/types/streams/tool_payload.go b/apps/backend/internal/agentctl/types/streams/tool_payload.go index 6fe9a95bcb..15a994faf0 100644 --- a/apps/backend/internal/agentctl/types/streams/tool_payload.go +++ b/apps/backend/internal/agentctl/types/streams/tool_payload.go @@ -245,6 +245,13 @@ type SubagentTaskPayload struct { Description string `json:"description"` Prompt string `json:"prompt"` SubagentType string `json:"subagent_type"` + + // Result fields (populated from tool_use_result on completion) + Status string `json:"status,omitempty"` + AgentID string `json:"agent_id,omitempty"` + DurationMs int64 `json:"duration_ms,omitempty"` + TotalTokens int64 `json:"total_tokens,omitempty"` + ToolUseCount int `json:"tool_use_count,omitempty"` } // ShowPlanPayload contains normalized data for plan display operations. @@ -260,10 +267,12 @@ type ManageTodosPayload struct { } // TodoItem represents a single todo item. +// Claude Code's TodoWrite uses "content" for the description and includes "activeForm". type TodoItem struct { ID string `json:"id,omitempty"` Description string `json:"description"` Status string `json:"status,omitempty"` + ActiveForm string `json:"active_form,omitempty"` } // MiscPayload is for miscellaneous operations that don't fit other categories. diff --git a/apps/backend/internal/orchestrator/event_handlers.go b/apps/backend/internal/orchestrator/event_handlers.go index 7f4d60d3be..147d3fd6fb 100644 --- a/apps/backend/internal/orchestrator/event_handlers.go +++ b/apps/backend/internal/orchestrator/event_handlers.go @@ -31,13 +31,13 @@ func (s *Service) handleACPSessionCreated(ctx context.Context, data watcher.ACPS if data.SessionID == "" || data.ACPSessionID == "" { return } - s.storeResumeToken(ctx, data.TaskID, data.SessionID, data.ACPSessionID) + s.storeResumeToken(ctx, data.TaskID, data.SessionID, data.ACPSessionID, "") } // storeResumeToken stores an agent's session ID as the resume token for session recovery. -// This is called both from handleACPSessionCreated (for ACP-based agents) and from -// handleAgentStreamEvent (for stream-based agents like Claude Code). -func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSessionID string) { +// This is called from handleACPSessionCreated, handleSessionStatusEvent, and handleCompleteStreamEvent. +// The optional lastMessageUUID is persisted alongside the token for --resume-session-at support. +func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSessionID, lastMessageUUID string) { session, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil { s.logger.Warn("failed to load task session for resume token storage", @@ -62,6 +62,7 @@ func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSe Status: "ready", Resumable: resumable, ResumeToken: acpSessionID, + LastMessageUUID: lastMessageUUID, AgentExecutionID: session.AgentExecutionID, ContainerID: session.ContainerID, } @@ -82,5 +83,6 @@ func (s *Service) storeResumeToken(ctx context.Context, taskID, sessionID, acpSe s.logger.Debug("stored resume token for session", zap.String("task_id", taskID), zap.String("session_id", sessionID), - zap.String("resume_token", acpSessionID)) + zap.String("resume_token", acpSessionID), + zap.String("last_message_uuid", lastMessageUUID)) } diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 299f547fec..e29406cf71 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -102,7 +102,7 @@ func (s *Service) handleSessionStatusEvent(ctx context.Context, payload *lifecyc taskID := payload.TaskID sessionID := payload.SessionID if sessionID != "" && payload.Data.ACPSessionID != "" { - s.storeResumeToken(ctx, taskID, sessionID, payload.Data.ACPSessionID) + s.storeResumeToken(ctx, taskID, sessionID, payload.Data.ACPSessionID, "") } if sessionID == "" || s.messageCreator == nil { return @@ -404,6 +404,19 @@ func (s *Service) handleCompleteStreamEvent(ctx context.Context, payload *lifecy zap.String("session_id", payload.SessionID), zap.Int("text_length", len(payload.Data.Text)), zap.Bool("has_text", payload.Data.Text != "")) + + // Update resume token with latest ACP session ID and message UUID on every turn. + // This ensures the token stays current after session compaction. + if payload.SessionID != "" && payload.Data.ACPSessionID != "" { + var lastMsgUUID string + if data, ok := payload.Data.Data.(map[string]interface{}); ok { + if uuid, ok := data["last_message_uuid"].(string); ok { + lastMsgUUID = uuid + } + } + s.storeResumeToken(ctx, payload.TaskID, payload.SessionID, payload.Data.ACPSessionID, lastMsgUUID) + } + s.saveAgentTextIfPresent(ctx, payload) s.completeTurnForSession(ctx, payload.SessionID) s.setSessionWaitingForInput(ctx, payload.TaskID, payload.SessionID) diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 7a7de9b002..d01316c5b7 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -363,6 +363,7 @@ type ExecutorRunning struct { Status string `json:"status"` Resumable bool `json:"resumable"` ResumeToken string `json:"resume_token,omitempty"` + LastMessageUUID string `json:"last_message_uuid,omitempty"` AgentExecutionID string `json:"agent_execution_id,omitempty"` ContainerID string `json:"container_id,omitempty"` AgentctlURL string `json:"agentctl_url,omitempty"` diff --git a/apps/backend/internal/task/repository/sqlite/base.go b/apps/backend/internal/task/repository/sqlite/base.go index 5db808a450..83a9ff7359 100644 --- a/apps/backend/internal/task/repository/sqlite/base.go +++ b/apps/backend/internal/task/repository/sqlite/base.go @@ -80,9 +80,19 @@ func (r *Repository) initSchema() error { if err := r.ensureDefaultExecutorsAndEnvironments(); err != nil { return err } + if err := r.runMigrations(); err != nil { + return err + } return r.ensureWorkspaceIndexes() } +// runMigrations applies idempotent ALTER TABLE migrations for schema evolution. +func (r *Repository) runMigrations() error { + // Add last_message_uuid column to executors_running (ignore error if already exists) + _, _ = r.db.Exec(`ALTER TABLE executors_running ADD COLUMN last_message_uuid TEXT DEFAULT ''`) + return nil +} + func (r *Repository) initCoreSchema() error { if err := r.initInfraSchema(); err != nil { return err diff --git a/apps/backend/internal/task/repository/sqlite/executor.go b/apps/backend/internal/task/repository/sqlite/executor.go index 0bda8c6f48..8747231fb5 100644 --- a/apps/backend/internal/task/repository/sqlite/executor.go +++ b/apps/backend/internal/task/repository/sqlite/executor.go @@ -155,10 +155,10 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. _, err := r.db.ExecContext(ctx, r.db.Rebind(` INSERT INTO executors_running ( id, session_id, task_id, executor_id, runtime, status, resumable, resume_token, - agent_execution_id, container_id, agentctl_url, agentctl_port, pid, + last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(session_id) DO UPDATE SET id = excluded.id, task_id = excluded.task_id, @@ -167,6 +167,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. status = excluded.status, resumable = excluded.resumable, resume_token = excluded.resume_token, + last_message_uuid = excluded.last_message_uuid, agent_execution_id = excluded.agent_execution_id, container_id = excluded.container_id, agentctl_url = excluded.agentctl_url, @@ -187,6 +188,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. running.Status, dialect.BoolToInt(running.Resumable), running.ResumeToken, + running.LastMessageUUID, running.AgentExecutionID, running.ContainerID, running.AgentctlURL, @@ -206,7 +208,7 @@ func (r *Repository) UpsertExecutorRunning(ctx context.Context, running *models. func (r *Repository) ListExecutorsRunning(ctx context.Context) ([]*models.ExecutorRunning, error) { rows, err := r.ro.QueryContext(ctx, ` SELECT id, session_id, task_id, executor_id, runtime, status, resumable, resume_token, - agent_execution_id, container_id, agentctl_url, agentctl_port, + last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, worktree_id, worktree_path, worktree_branch, created_at, updated_at FROM executors_running ORDER BY updated_at DESC @@ -228,6 +230,7 @@ func (r *Repository) ListExecutorsRunning(ctx context.Context) ([]*models.Execut &running.Status, &running.Resumable, &running.ResumeToken, + &running.LastMessageUUID, &running.AgentExecutionID, &running.ContainerID, &running.AgentctlURL, @@ -258,7 +261,7 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI err := r.ro.QueryRowContext(ctx, r.ro.Rebind(` SELECT id, session_id, task_id, executor_id, runtime, status, resumable, resume_token, - agent_execution_id, container_id, agentctl_url, agentctl_port, pid, + last_message_uuid, agent_execution_id, container_id, agentctl_url, agentctl_port, pid, worktree_id, worktree_path, worktree_branch, last_seen_at, error_message, created_at, updated_at FROM executors_running @@ -272,6 +275,7 @@ func (r *Repository) GetExecutorRunningBySessionID(ctx context.Context, sessionI &running.Status, &resumable, &running.ResumeToken, + &running.LastMessageUUID, &running.AgentExecutionID, &running.ContainerID, &running.AgentctlURL, diff --git a/apps/backend/pkg/claudecode/client.go b/apps/backend/pkg/claudecode/client.go index 0aac177482..5a5c0bc96a 100644 --- a/apps/backend/pkg/claudecode/client.go +++ b/apps/backend/pkg/claudecode/client.go @@ -18,6 +18,9 @@ import ( // It receives the request ID and control request, and should call SendControlResponse. type RequestHandler func(requestID string, req *ControlRequest) +// CancelHandler handles control_cancel_request messages that cancel a pending request. +type CancelHandler func(requestID string) + // MessageHandler handles streaming messages from Claude Code CLI. type MessageHandler func(msg *CLIMessage) @@ -35,6 +38,7 @@ type Client struct { // Handlers for incoming messages requestHandler RequestHandler + cancelHandler CancelHandler messageHandler MessageHandler // Pending control requests (requests we sent, waiting for responses) @@ -64,6 +68,13 @@ func (c *Client) SetRequestHandler(handler RequestHandler) { c.requestHandler = handler } +// SetCancelHandler sets the handler for control_cancel_request messages. +func (c *Client) SetCancelHandler(handler CancelHandler) { + c.mu.Lock() + defer c.mu.Unlock() + c.cancelHandler = handler +} + // SetMessageHandler sets the handler for streaming messages. func (c *Client) SetMessageHandler(handler MessageHandler) { c.mu.Lock() @@ -94,7 +105,8 @@ func (c *Client) Stop() { // Initialize sends the initialize control request to Claude Code CLI and waits for the response. // This must be called in streaming mode (input-format=stream-json) to get slash commands. -func (c *Client) Initialize(ctx context.Context, timeout time.Duration) (*InitializeResponseData, error) { +// hooks is optional — pass nil for no hooks. +func (c *Client) Initialize(ctx context.Context, timeout time.Duration, hooks map[string]any) (*InitializeResponseData, error) { requestID := uuid.New().String() // Create pending request channel @@ -118,7 +130,7 @@ func (c *Client) Initialize(ctx context.Context, timeout time.Duration) (*Initia RequestID: requestID, Request: SDKControlRequestBody{ Subtype: SubtypeInitialize, - Hooks: nil, // We don't use SDK hooks + Hooks: hooks, }, } @@ -239,6 +251,21 @@ func (c *Client) handleLine(line []byte) { return } + // Handle control_cancel_request (cancels a pending permission/hook request) + if msg.Type == MessageTypeControlCancelRequest && msg.RequestID != "" { + c.mu.RLock() + handler := c.cancelHandler + c.mu.RUnlock() + + if handler != nil { + handler(msg.RequestID) + } else { + c.logger.Warn("received control_cancel_request but no cancel handler registered", + zap.String("request_id", msg.RequestID)) + } + return + } + // Handle control responses (from Claude back to us, e.g., initialize response) // Note: request_id is inside the response object, not at the message level if msg.Type == MessageTypeControlResponse && msg.Response != nil { diff --git a/apps/backend/pkg/claudecode/client_test.go b/apps/backend/pkg/claudecode/client_test.go index 6f8436250e..574f241ed0 100644 --- a/apps/backend/pkg/claudecode/client_test.go +++ b/apps/backend/pkg/claudecode/client_test.go @@ -363,3 +363,82 @@ func TestClient_HandleControlResponse_UnknownRequest(t *testing.T) { // Test passes if no panic occurs - the unknown response is just logged and ignored } + +func TestClient_HandleControlCancelRequest(t *testing.T) { + input := `{"type":"control_cancel_request","request_id":"cancel-req-123"}` + "\n" + + var buf bytes.Buffer + client := NewClient(&buf, strings.NewReader(input), newTestLogger()) + + var receivedID string + var mu sync.Mutex + + client.SetCancelHandler(func(requestID string) { + mu.Lock() + receivedID = requestID + mu.Unlock() + }) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + client.Start(ctx) + time.Sleep(50 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + + if receivedID != "cancel-req-123" { + t.Errorf("cancel handler received ID = %q, want %q", receivedID, "cancel-req-123") + } +} + +func TestClient_HandleControlCancelRequest_NoHandler(t *testing.T) { + // Without a cancel handler, the cancel request should be logged and ignored + input := `{"type":"control_cancel_request","request_id":"cancel-req-456"}` + "\n" + + var buf bytes.Buffer + client := NewClient(&buf, strings.NewReader(input), newTestLogger()) + + // Don't register a cancel handler + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + client.Start(ctx) + time.Sleep(50 * time.Millisecond) + + // Test passes if no panic occurs +} + +func TestClient_HandleControlCancelRequest_NotForwardedAsMessage(t *testing.T) { + // control_cancel_request should be handled by the cancel handler, + // not forwarded to the message handler + input := `{"type":"control_cancel_request","request_id":"cancel-req-789"}` + "\n" + + var buf bytes.Buffer + client := NewClient(&buf, strings.NewReader(input), newTestLogger()) + + var messageCount int + var mu sync.Mutex + + client.SetCancelHandler(func(requestID string) {}) + client.SetMessageHandler(func(msg *CLIMessage) { + mu.Lock() + messageCount++ + mu.Unlock() + }) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + client.Start(ctx) + time.Sleep(50 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + + if messageCount != 0 { + t.Errorf("message handler received %d messages, want 0", messageCount) + } +} diff --git a/apps/backend/pkg/claudecode/types.go b/apps/backend/pkg/claudecode/types.go index 9405687b7b..dc69bc2022 100644 --- a/apps/backend/pkg/claudecode/types.go +++ b/apps/backend/pkg/claudecode/types.go @@ -16,8 +16,13 @@ const ( MessageTypeControlRequest = "control_request" // MessageTypeControlResponse is a response to a control request MessageTypeControlResponse = "control_response" + // MessageTypeControlCancelRequest cancels a pending control request (e.g., permission dialog) + MessageTypeControlCancelRequest = "control_cancel_request" // MessageTypeUser is a user message (prompt) MessageTypeUser = "user" + // MessageTypeRateLimit is emitted when the Anthropic API rate-limits the agent. + // Claude Code uses "rate_limit_event" as the message type. + MessageTypeRateLimit = "rate_limit_event" ) // Control request subtypes @@ -48,6 +53,9 @@ type CLIMessage struct { // Type is the message type (system, assistant, result, control_request, control_response, etc.) Type string `json:"type"` + // UUID is a unique identifier for this message, used for --resume-session-at + UUID string `json:"uuid,omitempty"` + // For control_request messages (from Claude to us) RequestID string `json:"request_id,omitempty"` Request *ControlRequest `json:"request,omitempty"` @@ -70,7 +78,7 @@ type CLIMessage struct { // Result can be either a string (error message) or an object (ResultData) Result json.RawMessage `json:"result,omitempty"` Subtype string `json:"subtype,omitempty"` - CostUSD float64 `json:"cost_usd,omitempty"` + CostUSD float64 `json:"total_cost_usd,omitempty"` DurationMS int64 `json:"duration_ms,omitempty"` DurationAPIMS int64 `json:"duration_api_ms,omitempty"` IsError bool `json:"is_error,omitempty"` @@ -80,6 +88,20 @@ type CLIMessage struct { TotalOutputTokens int64 `json:"total_output_tokens,omitempty"` ModelUsage map[string]ModelUsageStats `json:"model_usage,omitempty"` + // For rate_limit messages + RateLimitInfo json.RawMessage `json:"rate_limit_info,omitempty"` + + // For user messages: replay/synthetic flags. + // IsReplay is true when the message is historical context echoed on resume. + // IsSynthetic is true for Claude-generated internal messages (checkpoint, injected context). + IsReplay bool `json:"isReplay,omitempty"` + IsSynthetic bool `json:"isSynthetic,omitempty"` + + // For user messages: rich metadata about tool execution results. + // Contains structured data like {stdout, stderr} for Bash, {oldTodos, newTodos} for TodoWrite, + // or {status, agentId, totalDurationMs, totalTokens, totalToolUseCount} for Task (sub-agent). + ToolUseResult json.RawMessage `json:"tool_use_result,omitempty"` + // Raw message for parsing content blocks RawContent json.RawMessage `json:"-"` } @@ -136,10 +158,44 @@ type ContentBlock struct { Name string `json:"name,omitempty"` Input map[string]any `json:"input,omitempty"` - // For tool_result blocks - ToolUseID string `json:"tool_use_id,omitempty"` - Content string `json:"content,omitempty"` - IsError bool `json:"is_error,omitempty"` + // For tool_result blocks. + // Content can be a string or an array of text blocks (sub-agent results use arrays). + // Use GetContentString() to get the text regardless of format. + ToolUseID string `json:"tool_use_id,omitempty"` + RawContent json.RawMessage `json:"content,omitempty"` + IsError bool `json:"is_error,omitempty"` +} + +// GetContentString returns tool_result content as a string. +// If content is a plain string, returns it directly. +// If content is an array of text blocks [{type, text}], concatenates all text values. +func (b *ContentBlock) GetContentString() string { + if len(b.RawContent) == 0 { + return "" + } + + // Try as plain string first + var s string + if err := json.Unmarshal(b.RawContent, &s); err == nil { + return s + } + + // Try as array of text blocks + var blocks []struct { + Text string `json:"text"` + } + if err := json.Unmarshal(b.RawContent, &blocks); err == nil { + var result string + for i, block := range blocks { + if i > 0 { + result += "\n" + } + result += block.Text + } + return result + } + + return "" } // Usage contains token usage information. @@ -196,9 +252,10 @@ type ControlRequest struct { Subtype string `json:"subtype"` // For can_use_tool requests - ToolName string `json:"tool_name,omitempty"` - Input map[string]any `json:"input,omitempty"` - ToolUseID string `json:"tool_use_id,omitempty"` + ToolName string `json:"tool_name,omitempty"` + Input map[string]any `json:"input,omitempty"` + ToolUseID string `json:"tool_use_id,omitempty"` + BlockedPaths string `json:"blocked_paths,omitempty"` // For hook_callback requests CallbackID string `json:"callback_id,omitempty"` @@ -260,8 +317,9 @@ type ControlResponse struct { // Subtype is the response type (success, error) Subtype string `json:"subtype"` - // For success responses - Result *PermissionResult `json:"result,omitempty"` + // For success responses — holds a *PermissionResult for can_use_tool, + // or a map[string]any for hook callbacks (hookSpecificOutput, decision/reason). + Result any `json:"result,omitempty"` // For error responses Error string `json:"error,omitempty"` @@ -305,6 +363,30 @@ type SDKControlRequestBody struct { Mode string `json:"mode,omitempty"` } +// HookConfig defines hook configuration passed during initialize. +type HookConfig struct { + PreToolUse []HookEntry `json:"PreToolUse,omitempty"` + Stop []HookEntry `json:"Stop,omitempty"` +} + +// HookEntry defines a single hook entry with a matcher pattern and callback IDs. +type HookEntry struct { + Matcher string `json:"matcher,omitempty"` + HookCallbackIDs []string `json:"hookCallbackIds"` +} + +// ToMap converts HookConfig to map[string]any for use in SDKControlRequestBody.Hooks. +func (h *HookConfig) ToMap() map[string]any { + result := make(map[string]any) + if len(h.PreToolUse) > 0 { + result["PreToolUse"] = h.PreToolUse + } + if len(h.Stop) > 0 { + result["Stop"] = h.Stop + } + return result +} + // UserMessage is sent to provide a prompt to Claude Code. type UserMessage struct { Type string `json:"type"` // "user" diff --git a/apps/backend/pkg/claudecode/types_test.go b/apps/backend/pkg/claudecode/types_test.go index 753a9cbc2d..f79e5796ee 100644 --- a/apps/backend/pkg/claudecode/types_test.go +++ b/apps/backend/pkg/claudecode/types_test.go @@ -246,8 +246,8 @@ func TestContentBlock_Types(t *testing.T) { if block.ToolUseID != "tool123" { t.Errorf("ToolUseID = %q, want %q", block.ToolUseID, "tool123") } - if block.Content != "output" { - t.Errorf("Content = %q, want %q", block.Content, "output") + if block.GetContentString() != "output" { + t.Errorf("Content = %q, want %q", block.GetContentString(), "output") } }, }, @@ -438,3 +438,247 @@ func TestIncomingControlResponse_JSONParsing(t *testing.T) { t.Errorf("Error = %q, want %q", errorResp.Error, "Something went wrong") } } + +func TestCLIMessage_IsReplay_IsSynthetic(t *testing.T) { + tests := []struct { + name string + json string + wantReplay bool + wantSynthetic bool + }{ + { + name: "replay user message", + json: `{"type":"user","uuid":"abc","session_id":"sess-1","isReplay":true,"message":{"role":"user","content":"hello"}}`, + wantReplay: true, + wantSynthetic: false, + }, + { + name: "synthetic user message", + json: `{"type":"user","uuid":"abc","session_id":"sess-1","isSynthetic":true,"message":{"role":"user","content":"checkpoint"}}`, + wantReplay: false, + wantSynthetic: true, + }, + { + name: "replay and synthetic", + json: `{"type":"user","uuid":"abc","isReplay":true,"isSynthetic":true,"message":{"role":"user","content":"old"}}`, + wantReplay: true, + wantSynthetic: true, + }, + { + name: "neither replay nor synthetic", + json: `{"type":"user","uuid":"abc","message":{"role":"user","content":"hello"}}`, + wantReplay: false, + wantSynthetic: false, + }, + { + name: "assistant message has no replay fields", + json: `{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"hi"}]}}`, + wantReplay: false, + wantSynthetic: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var msg CLIMessage + if err := json.Unmarshal([]byte(tt.json), &msg); err != nil { + t.Fatalf("failed to parse: %v", err) + } + if msg.IsReplay != tt.wantReplay { + t.Errorf("IsReplay = %v, want %v", msg.IsReplay, tt.wantReplay) + } + if msg.IsSynthetic != tt.wantSynthetic { + t.Errorf("IsSynthetic = %v, want %v", msg.IsSynthetic, tt.wantSynthetic) + } + }) + } +} + +func TestHookConfig_ToMap(t *testing.T) { + tests := []struct { + name string + config HookConfig + wantKeys []string + absentKeys []string + }{ + { + name: "empty config", + config: HookConfig{}, + wantKeys: nil, + absentKeys: []string{"PreToolUse", "Stop"}, + }, + { + name: "PreToolUse only", + config: HookConfig{ + PreToolUse: []HookEntry{ + {Matcher: `^Bash$`, HookCallbackIDs: []string{"tool_approval"}}, + }, + }, + wantKeys: []string{"PreToolUse"}, + absentKeys: []string{"Stop"}, + }, + { + name: "Stop only", + config: HookConfig{ + Stop: []HookEntry{ + {HookCallbackIDs: []string{"stop_git_check"}}, + }, + }, + wantKeys: []string{"Stop"}, + absentKeys: []string{"PreToolUse"}, + }, + { + name: "both hooks", + config: HookConfig{ + PreToolUse: []HookEntry{ + {Matcher: `^Bash$`, HookCallbackIDs: []string{"tool_approval"}}, + }, + Stop: []HookEntry{ + {HookCallbackIDs: []string{"stop_git_check"}}, + }, + }, + wantKeys: []string{"PreToolUse", "Stop"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := tt.config.ToMap() + for _, key := range tt.wantKeys { + if _, ok := result[key]; !ok { + t.Errorf("ToMap() missing key %q", key) + } + } + for _, key := range tt.absentKeys { + if _, ok := result[key]; ok { + t.Errorf("ToMap() should not have key %q", key) + } + } + }) + } +} + +func TestHookConfig_ToMap_PreservesEntries(t *testing.T) { + cfg := HookConfig{ + PreToolUse: []HookEntry{ + {Matcher: `^Bash$`, HookCallbackIDs: []string{"tool_approval"}}, + {Matcher: `^Edit$`, HookCallbackIDs: []string{"auto_approve"}}, + }, + } + result := cfg.ToMap() + entries, ok := result["PreToolUse"].([]HookEntry) + if !ok { + t.Fatal("PreToolUse is not []HookEntry") + } + if len(entries) != 2 { + t.Errorf("PreToolUse has %d entries, want 2", len(entries)) + } + if entries[0].Matcher != `^Bash$` { + t.Errorf("entries[0].Matcher = %q, want %q", entries[0].Matcher, `^Bash$`) + } + if entries[1].HookCallbackIDs[0] != "auto_approve" { + t.Errorf("entries[1].HookCallbackIDs[0] = %q, want %q", entries[1].HookCallbackIDs[0], "auto_approve") + } +} + +func TestContentBlock_GetContentString(t *testing.T) { + tests := []struct { + name string + json string + want string + }{ + { + name: "string content", + json: `{"type":"tool_result","tool_use_id":"t1","content":"hello world"}`, + want: "hello world", + }, + { + name: "array of text blocks", + json: `{"type":"tool_result","tool_use_id":"t1","content":[{"type":"text","text":"line 1"},{"type":"text","text":"line 2"}]}`, + want: "line 1\nline 2", + }, + { + name: "single text block array", + json: `{"type":"tool_result","tool_use_id":"t1","content":[{"type":"text","text":"only line"}]}`, + want: "only line", + }, + { + name: "empty content", + json: `{"type":"tool_result","tool_use_id":"t1"}`, + want: "", + }, + { + name: "empty string content", + json: `{"type":"tool_result","tool_use_id":"t1","content":""}`, + want: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var block ContentBlock + if err := json.Unmarshal([]byte(tt.json), &block); err != nil { + t.Fatalf("failed to parse: %v", err) + } + got := block.GetContentString() + if got != tt.want { + t.Errorf("GetContentString() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestCLIMessage_RateLimitEventType(t *testing.T) { + // Claude Code sends "rate_limit_event", not "rate_limit" + jsonStr := `{"type":"rate_limit_event","rate_limit_info":{"status":"allowed"},"session_id":"sess-1"}` + var msg CLIMessage + if err := json.Unmarshal([]byte(jsonStr), &msg); err != nil { + t.Fatalf("failed to parse: %v", err) + } + if msg.Type != MessageTypeRateLimit { + t.Errorf("Type = %q, want %q (MessageTypeRateLimit)", msg.Type, MessageTypeRateLimit) + } +} + +func TestCLIMessage_TotalCostUSD(t *testing.T) { + // Claude Code sends "total_cost_usd", not "cost_usd" + jsonStr := `{"type":"result","total_cost_usd":0.123,"session_id":"sess-1"}` + var msg CLIMessage + if err := json.Unmarshal([]byte(jsonStr), &msg); err != nil { + t.Fatalf("failed to parse: %v", err) + } + if msg.CostUSD != 0.123 { + t.Errorf("CostUSD = %f, want %f", msg.CostUSD, 0.123) + } +} + +func TestCLIMessage_ToolUseResult(t *testing.T) { + // Sub-agent task result with rich metadata + jsonStr := `{ + "type":"user", + "message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"t1","content":"result"}]}, + "tool_use_result":{"status":"completed","agentId":"abc","totalDurationMs":1500,"totalTokens":4000,"totalToolUseCount":2}, + "session_id":"sess-1" + }` + var msg CLIMessage + if err := json.Unmarshal([]byte(jsonStr), &msg); err != nil { + t.Fatalf("failed to parse: %v", err) + } + if len(msg.ToolUseResult) == 0 { + t.Fatal("ToolUseResult is empty") + } + + var result map[string]any + if err := json.Unmarshal(msg.ToolUseResult, &result); err != nil { + t.Fatalf("failed to parse ToolUseResult: %v", err) + } + if result["status"] != "completed" { + t.Errorf("status = %v, want %q", result["status"], "completed") + } + if result["agentId"] != "abc" { + t.Errorf("agentId = %v, want %q", result["agentId"], "abc") + } + if result["totalDurationMs"].(float64) != 1500 { + t.Errorf("totalDurationMs = %v, want %v", result["totalDurationMs"], 1500) + } +} diff --git a/docs/agents/claude_code.md b/docs/agents/claude_code.md new file mode 100644 index 0000000000..6bd4523938 --- /dev/null +++ b/docs/agents/claude_code.md @@ -0,0 +1,587 @@ +# Claude Code Integration + +Technical reference for Kandev's Claude Code CLI integration. Covers the stream-json protocol, message lifecycle, control plane, tool normalization, session resume, and permission hooks. + +## Protocol Overview + +Kandev communicates with Claude Code via the **stream-json** protocol: newline-delimited JSON over stdin/stdout. The CLI runs as a subprocess managed by agentctl. + +``` +┌─────────────┐ stdin (JSON lines) ┌──────────────────┐ +│ agentctl │ ──────────────────► │ Claude Code CLI │ +│ (adapter) │ ◄────────────────── │ (subprocess) │ +└─────────────┘ stdout (JSON lines) └──────────────────┘ +``` + +**CLI invocation:** +``` +npx -y @anthropic-ai/claude-code@ + -p # Print mode (non-interactive) + --output-format=stream-json # Streaming JSON output + --input-format=stream-json # Streaming JSON input + --permission-prompt-tool=stdio # Permission prompts via stdio control messages + --include-partial-messages # Stream in-progress assistant messages + --replay-user-messages # Replay user messages on session resume + --verbose # Extended logging + --model # Model selection + --resume # Resume existing session +``` + +--- + +## Architecture + +``` +pkg/claudecode/ +├── types.go # Protocol message types, constants, structs +└── client.go # stdin/stdout client, read loop, control request handling + +internal/agentctl/server/adapter/transport/streamjson/ +├── adapter.go # Main adapter: state, dispatch, sendUpdate +├── streamjson_init.go # Initialize, hooks, MCP config +├── streamjson_session.go # NewSession / LoadSession +├── streamjson_prompt.go # Prompt sending, image attachments, Cancel +├── streamjson_messages.go # System, assistant, user, rate-limit handlers +├── streamjson_result.go # Result handler, deduplication, completion +├── streamjson_permissions.go # Permission requests, hook callbacks, timeout +├── streamjson_tracing.go # OTel tracing helpers for control messages +└── normalize.go # Tool data → NormalizedPayload conversion +``` + +--- + +## Message Types + +Every line on stdout is a JSON object with a `type` field. The client dispatches based on this type: + +| Type | Direction | Description | +|------|-----------|-------------| +| `system` | CLI → Adapter | Session initialization (session_id, status, slash commands) | +| `assistant` | CLI → Adapter | Assistant response with content blocks (text, thinking, tool_use) | +| `user` | CLI → Adapter | Tool results (tool_result blocks) or slash command output | +| `result` | CLI → Adapter | Turn completion with stats (cost, tokens, duration, errors) | +| `rate_limit` | CLI → Adapter | API rate limit notification | +| `control_request` | CLI → Adapter | Permission request or hook callback | +| `control_cancel_request` | CLI → Adapter | Cancels a pending permission/hook request | +| `control_response` | Adapter → CLI | Response to a control_request | + +--- + +## Message Lifecycle + +### Client Layer (`pkg/claudecode/client.go`) + +The `Client` reads stdout via `bufio.Scanner` (10 MB buffer), parses each line into a `CLIMessage`, and dispatches: + +``` +stdout line + │ + ├── type=control_request → RequestHandler(requestID, *ControlRequest) + ├── type=control_cancel_request → CancelHandler(requestID) + ├── type=control_response → matches pending request by ID, unblocks goroutine + └── (all other types) → MessageHandler(*CLIMessage) +``` + +Control messages are handled immediately. All other messages are forwarded to the adapter's `handleMessage()`. + +### Adapter Layer (`streamjson/adapter.go`) + +The adapter's `handleMessage()` marshals the message for debug logging/tracing, then dispatches: + +``` +handleMessage(msg) + │ + ├── system → handleSystemMessage → session_status, available_commands + ├── assistant → handleAssistantMessage → message_chunk, reasoning, tool_call, context_window + ├── user → handleUserMessage → tool_update (complete/error), message_chunk (slash cmds) + ├── result → handleResultMessage → complete, error (+ auto-complete pending tools) + └── rate_limit → handleRateLimitMessage → rate_limit +``` + +Every dispatched event goes through `sendUpdate()`, which: +1. Logs the normalized event (JSONL debug file) +2. Creates an OTel span (child of the active prompt span) +3. Sends to the `updatesCh` channel for WebSocket delivery + +--- + +## Turn Anatomy + +A single turn (prompt → result) produces this message sequence: + +``` +Adapter Claude Code CLI + │ │ + │ ── user message (prompt) ──────────► │ + │ │ + │ ◄── system (session_id, status) ──── │ (first prompt only) + │ ◄── assistant (text blocks) ──────── │ (streaming chunks) + │ ◄── assistant (thinking blocks) ──── │ (reasoning, if enabled) + │ ◄── assistant (tool_use block) ───── │ (tool invocation) + │ │ + │ ◄── control_request (can_use_tool) ─ │ (if hooks require approval) + │ ── control_response (allow/deny) ──► │ + │ │ + │ ◄── user (tool_result block) ─────── │ (tool output) + │ ◄── assistant (text blocks) ──────── │ (more streaming) + │ ◄── result ───────────────────────── │ (turn complete) + │ │ +``` + +### Text Deduplication + +The `result` message may contain a `text` field that duplicates already-streamed assistant text. The adapter tracks `streamingTextSentThisTurn`: +- Set to `true` when any `message_chunk` event is sent from assistant text blocks +- If `true` when result arrives, `result.text` is skipped +- Reset to `false` after each result + +### Message UUIDs + +Each message has a `uuid` field used for session truncation on resume: +- **Assistant UUIDs** are held as `pendingAssistantUUID` until the result message commits them to `lastMessageUUID` +- **User UUIDs** are committed immediately (safe to resume from) +- `lastMessageUUID` is included in the `complete` event and persisted for `--resume-session-at` + +--- + +## Content Blocks + +Assistant messages contain an array of content blocks: + +### Text Block +```json +{ "type": "text", "text": "Here's the implementation..." } +``` +Emitted as `message_chunk` events for real-time streaming. + +### Thinking Block +```json +{ "type": "thinking", "thinking": "Let me analyze the code structure..." } +``` +Emitted as `reasoning` events. Only present when extended thinking is enabled. + +### Tool Use Block +```json +{ + "type": "tool_use", + "id": "toolu_abc123", + "name": "Edit", + "input": { "file_path": "/src/main.go", "old_string": "...", "new_string": "..." } +} +``` +Normalized into a `NormalizedPayload` and emitted as a `tool_call` event with status `"running"`. + +### Tool Result Block (in user messages) +```json +{ + "type": "tool_result", + "tool_use_id": "toolu_abc123", + "content": "File edited successfully", + "is_error": false +} +``` +Enriches the pending `NormalizedPayload` with output data and emits a `tool_update` event with status `"complete"` or `"error"`. + +--- + +## Control Plane + +### Initialize + +Sent once after the subprocess starts. Registers hooks and retrieves available slash commands. + +``` +Adapter → CLI: +{ + "type": "control_request", + "request_id": "", + "request": { + "subtype": "initialize", + "hooks": { + "PreToolUse": [{ "matcher": "^Bash$", "hookCallbackIds": ["tool_approval"] }], + "Stop": [{ "hookCallbackIds": ["stop_git_check"] }] + } + } +} + +CLI → Adapter: +{ + "type": "control_response", + "response": { + "subtype": "success", + "request_id": "", + "response": { + "commands": [ + { "name": "commit", "description": "Create a git commit" }, + { "name": "review-pr", "description": "Review a pull request" } + ] + } + } +} +``` + +### Permission Request (can_use_tool) + +When the CLI needs approval for a tool call: + +``` +CLI → Adapter: +{ + "type": "control_request", + "request_id": "req_123", + "request": { + "subtype": "can_use_tool", + "tool_name": "Bash", + "tool_use_id": "toolu_abc", + "input": { "command": "rm -rf /tmp/test" }, + "blocked_paths": "/etc,/usr" + } +} + +Adapter → CLI: +{ + "type": "control_response", + "request_id": "req_123", + "response": { + "subtype": "success", + "result": { + "behavior": "allow", + "updatedPermissions": [], + "message": "" + } + } +} +``` + +The `behavior` field is `"allow"` or `"deny"`. On deny, an optional `interrupt: true` stops the current operation. + +### Hook Callbacks + +Hooks are registered during initialize and fire before tool execution. The response format differs from permission responses: + +**PreToolUse hook response:** +```json +{ + "type": "control_response", + "request_id": "req_456", + "response": { + "subtype": "success", + "result": { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "allow", + "permissionDecisionReason": "Auto-approved by SDK" + } + } + } +} +``` + +`permissionDecision` values: +- `"allow"` — approve the tool call +- `"deny"` — reject the tool call +- `"ask"` — trigger a separate `can_use_tool` permission request + +**Stop hook response:** +```json +{ + "type": "control_response", + "request_id": "req_789", + "response": { + "subtype": "success", + "result": { + "decision": "approve", + "reason": "" + } + } +} +``` + +### Cancel Request + +The CLI can cancel a pending permission dialog (e.g., when the tool call is superseded): + +``` +CLI → Adapter: +{ + "type": "control_cancel_request", + "request_id": "req_123" +} +``` + +The adapter emits a `permission_cancelled` event so the frontend closes the dialog. + +### Interrupt + +Sent by the adapter to cancel the current operation: + +``` +Adapter → CLI: +{ + "type": "control_request", + "request_id": "", + "request": { "subtype": "interrupt" } +} +``` + +--- + +## Permission Modes + +The adapter supports three permission policies, configured via hooks at initialize: + +### Autonomous (default) +No hooks registered. `--dangerously-skip-permissions` flag bypasses all checks. Agent runs freely. + +### Supervised +Uses `--permission-mode=bypassPermissions` with hooks: +- **PreToolUse hook**: Matches all tools except safe read-only ones (`Glob`, `Grep`, `NotebookRead`, `Read`, `Task`, `TodoWrite`). Responds with `permissionDecision: "ask"` which triggers a `can_use_tool` flow, forwarding to the frontend for user approval. +- **Stop hook**: Checks for uncommitted changes before agent exit. +- Non-matching tools (read-only) are implicitly auto-approved by Claude Code. + +### Plan +Uses `--permission-mode=bypassPermissions` with hooks: +- **PreToolUse hook (ExitPlanMode)**: Requires approval to exit plan mode. +- **PreToolUse hook (everything else)**: Auto-approved. +- **Stop hook**: Same as supervised. + +--- + +## Result Message + +The result message signals turn completion and contains aggregate statistics: + +```json +{ + "type": "result", + "subtype": "success", + "cost_usd": 0.0234, + "duration_ms": 15234, + "duration_api_ms": 12100, + "is_error": false, + "num_turns": 3, + "total_input_tokens": 45000, + "total_output_tokens": 2300, + "model_usage": { + "claude-sonnet-4-5": { "context_window": 200000 } + }, + "result": { + "text": "I've implemented the changes...", + "session_id": "sess_abc123" + } +} +``` + +On result, the adapter: +1. Auto-completes any pending tool calls that never received a `tool_result` +2. Updates context window size from `model_usage` if available +3. Emits result text only if no streaming text was sent this turn +4. Commits the pending assistant UUID as `lastMessageUUID` +5. Emits `complete` event with cost/token/duration stats +6. Signals completion to unblock the `Prompt()` goroutine +7. Emits `error` event if `is_error: true` + +--- + +## Session Resume + +### Resume by Session ID + +``` +--resume +``` + +Resumes the conversation from the last message. The session ID is stored in the `executors_running` table as the resume token. + +### Resume at Message UUID + +``` +--resume-session-at +``` + +Truncates the conversation to the specified message UUID and resumes from there. Used for retry — when a turn fails, resume from the last known-good message instead of replaying the failed context. + +The `lastMessageUUID` is: +- Tracked from message `uuid` fields during the conversation +- Committed on result messages (assistant UUIDs are pending until result confirms success) +- Included in the `complete` event's `data.last_message_uuid` field +- Persisted in the `executors_running.last_message_uuid` column + +--- + +## Tool Normalization + +The normalizer converts Claude Code tool calls into protocol-agnostic `NormalizedPayload` structs. This allows the frontend to render tools uniformly regardless of which agent produced them. + +### Tool → Kind Mapping + +| Claude Code Tool | NormalizedPayload Kind | Frontend Renderer | +|-----------------|----------------------|-------------------| +| `Edit`, `Write`, `NotebookEdit` | `modify_file` | `tool_edit` | +| `Read` | `read_file` | `tool_read` | +| `Glob`, `Grep` | `code_search` | `tool_search` | +| `Bash` | `shell_exec` | `tool_execute` | +| `WebFetch`, `WebSearch` | `http_request` | `tool_call` | +| `Task` | `subagent_task` | `tool_call` | +| `TaskCreate` | `create_task` | `tool_call` | +| `TaskUpdate`, `TaskList`, `TodoWrite` | `manage_todos` | `todo` | +| (unknown) | `generic` | `tool_call` | + +### Normalization Flow + +``` +tool_use block (name, id, input) + │ + ├── NormalizeToolCall(name, input) → NormalizedPayload + │ stored in pendingToolCalls[id] + │ + ├── emit tool_call event (status: "running", payload attached) + │ + ... (tool executes) ... + │ + ├── tool_result block (tool_use_id, content, is_error) + │ + ├── NormalizeToolResult(payload, content) → enriches payload with output + │ + └── emit tool_update event (status: "complete" or "error", payload attached) +``` + +### NormalizedPayload Structure + +A discriminated union — exactly one kind-specific field is populated: + +``` +NormalizedPayload + ├── kind: ToolKind (read_file, modify_file, shell_exec, ...) + ├── read_file: { file_path, offset, limit, output: { content, line_count, truncated } } + ├── modify_file: { file_path, mutations: [{ type, content, old_content, new_content, diff }] } + ├── shell_exec: { command, work_dir, description, timeout, background, output: { exit_code, stdout, stderr } } + ├── code_search: { query, pattern, path, glob, output: { files, file_count, truncated } } + ├── http_request: { url, method, response, is_error } + ├── subagent_task: { description, prompt, subagent_type } + ├── create_task: { title, description } + ├── manage_todos: { operation, items: [{ id, description, status }] } + ├── generic: { name, input, output } + └── misc: { label, details } +``` + +--- + +## Agent Events + +Events emitted through the `updatesCh` channel to the frontend via WebSocket: + +| Event Type | When | Key Fields | +|-----------|------|------------| +| `session_status` | First system message | `session_id`, `data.session_status` ("new"/"resumed") | +| `available_commands` | After initialize or system message | `available_commands[]` (name, description) | +| `message_chunk` | Assistant text or slash command output | `text` | +| `reasoning` | Thinking/chain-of-thought | `reasoning_text` | +| `tool_call` | Tool invocation starts | `tool_call_id`, `tool_name`, `tool_title`, `tool_status: "running"`, `normalized` | +| `tool_update` | Tool result received | `tool_call_id`, `tool_status: "complete"/"error"`, `normalized` | +| `context_window` | Token usage update | `context_window_size`, `context_window_used`, `context_window_remaining`, `context_efficiency` | +| `complete` | Turn finished | `data` (cost_usd, duration_ms, num_turns, input/output_tokens, last_message_uuid) | +| `error` | Turn failed | `error` message | +| `rate_limit` | API rate limited | `rate_limit_message` | +| `permission_cancelled` | Permission dialog cancelled | `pending_id` | + +### Subagent Nesting + +When Claude Code uses the `Task` tool, it spawns subagents. Messages from subagents include `parent_tool_use_id`, which maps to the parent `tool_call_id`. The frontend uses this for visual nesting. + +--- + +## Context Window Tracking + +Context usage is tracked from two sources: + +1. **Assistant messages** — `usage` field provides per-message token counts: + ``` + context_used = input_tokens + output_tokens + cache_creation_input_tokens + cache_read_input_tokens + ``` + +2. **Result messages** — `model_usage` map provides the actual context window size per model: + ```json + { "claude-sonnet-4-5": { "context_window": 200000 } } + ``` + +The adapter tracks `mainModelName` (first model seen, excluding subagent models) and updates `mainModelContextWindow` from result stats. The default fallback is 200,000 tokens. + +--- + +## MCP Server Configuration + +MCP servers are passed to Claude Code via the `--mcp-config` flag: + +``` +--mcp-config '{"mcpServers":{"my-server":{"command":"node","args":["server.js"]}}}' +``` + +Supported transports: +- **stdio**: `{ "command": "...", "args": [...] }` +- **SSE/HTTP**: `{ "url": "...", "type": "sse" }` + +Configuration is built from `cfg.McpServers` in `PrepareCommandArgs()`. + +--- + +## Image Attachments + +Since stream-json doesn't support multimodal content blocks in prompts, image attachments are handled by: + +1. Saving base64-decoded images to `.kandev/temp/images/` in the workspace +2. Prepending file references to the prompt text +3. Claude Code reads the images via its `Read` tool + +Supported formats: PNG, JPEG, GIF, WebP. + +--- + +## Debug & Tracing + +Enabled via `KANDEV_DEBUG_AGENT_MESSAGES=true`. + +### Debug Logging +Writes JSONL files in the working directory: +- `raw-streamjson-.jsonl` — raw protocol messages as received +- `normalized-streamjson-.jsonl` — normalized AgentEvents + +### OTel Tracing +Requires `OTEL_EXPORTER_OTLP_ENDPOINT` in addition to the debug flag. + +**Span hierarchy:** +``` +streamjson.prompt (parent span, covers entire turn) + ├── streamjson.system (session init) + ├── streamjson.message_chunk (streaming text) + ├── streamjson.reasoning (thinking) + ├── streamjson.tool_call (tool invocation) + ├── streamjson.tool_update (tool result) + ├── streamjson.context_window (token usage) + ├── streamjson.control.control_request.can_use_tool (incoming permission request) + ├── streamjson.control.permission_response (outgoing permission response) + ├── streamjson.control.hook_response.pre_tool_use (outgoing hook response) + └── streamjson.complete (turn completion) +``` + +Each span carries `raw` and `normalized` events for side-by-side comparison in Jaeger/Tempo. + +--- + +## Key Source Files + +| File | Package | Purpose | +|------|---------|---------| +| `pkg/claudecode/types.go` | `claudecode` | Protocol message types and constants | +| `pkg/claudecode/client.go` | `claudecode` | stdin/stdout client, read loop, control requests | +| `streamjson/adapter.go` | `streamjson` | Main adapter state, message dispatch, sendUpdate | +| `streamjson/streamjson_init.go` | `streamjson` | Initialize, hook building, MCP config | +| `streamjson/streamjson_session.go` | `streamjson` | Session create/resume | +| `streamjson/streamjson_prompt.go` | `streamjson` | Prompt send, wait, cancel, image attachments | +| `streamjson/streamjson_messages.go` | `streamjson` | System, assistant, user, rate-limit handlers | +| `streamjson/streamjson_result.go` | `streamjson` | Result processing, text dedup, completion signaling | +| `streamjson/streamjson_permissions.go` | `streamjson` | Permissions, hooks, timeout, cancel | +| `streamjson/normalize.go` | `streamjson` | Tool call/result → NormalizedPayload | +| `streams/tool_payload.go` | `streams` | NormalizedPayload discriminated union | +| `streams/agent.go` | `streams` | AgentEvent types and structure | +| `agents/claude_code.go` | `agents` | Agent definition, CLI flags, models |