feat: improve claude code
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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(`"<local-command-stdout>compact summary here</local-command-stdout>"`),
|
||||
},
|
||||
}
|
||||
|
||||
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 := `"<kandev-system>MCP TOOLS info</kandev-system>\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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
+59
-1
@@ -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
|
||||
|
||||
+157
-32
@@ -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 <local-command-stdout> 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 <local-command-stdout> tags if present
|
||||
text := contentStr
|
||||
if strings.HasPrefix(text, "<local-command-stdout>") && strings.HasSuffix(text, "</local-command-stdout>") {
|
||||
text = strings.TrimPrefix(text, "<local-command-stdout>")
|
||||
text = strings.TrimSuffix(text, "</local-command-stdout>")
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
+137
-20
@@ -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")
|
||||
}
|
||||
|
||||
+10
-2
@@ -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)
|
||||
}
|
||||
|
||||
+18
-5
@@ -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
|
||||
|
||||
+56
@@ -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)
|
||||
}
|
||||
@@ -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").
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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@<version>
|
||||
-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> # Model selection
|
||||
--resume <session-id> # 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": "<uuid>",
|
||||
"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": "<uuid>",
|
||||
"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": "<uuid>",
|
||||
"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 <session-id>
|
||||
```
|
||||
|
||||
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 <message-uuid>
|
||||
```
|
||||
|
||||
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-<agentId>.jsonl` — raw protocol messages as received
|
||||
- `normalized-streamjson-<agentId>.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 |
|
||||
Reference in New Issue
Block a user