|
| 1 | +package claude |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "fmt" |
| 6 | + "strings" |
| 7 | + "sync" |
| 8 | +) |
| 9 | + |
| 10 | +// Stream represents an active claude subprocess streaming session. |
| 11 | +// |
| 12 | +// Call Events() to range over the stream of events. The channel is closed when |
| 13 | +// the agent finishes, the subprocess exits, or the context is cancelled. |
| 14 | +// |
| 15 | +// Control methods (SetModel, SetPermissionMode, SetMaxThinkingTokens, Interrupt) |
| 16 | +// may be called concurrently from any goroutine while the stream is active. |
| 17 | +type Stream struct { |
| 18 | + events chan Event |
| 19 | + write func(any) error |
| 20 | + ctx context.Context |
| 21 | + interrupt func() // graceful shutdown trigger (idempotent) |
| 22 | + |
| 23 | + // pending maps request_id → response channel for blocking control requests. |
| 24 | + pending map[string]chan controlResponse |
| 25 | + pendingMu sync.Mutex |
| 26 | +} |
| 27 | + |
| 28 | +// Events returns the receive-only channel of events streamed from the subprocess. |
| 29 | +// The channel is closed when the session ends. Callers should always range until |
| 30 | +// the channel closes. |
| 31 | +func (s *Stream) Events() <-chan Event { |
| 32 | + return s.events |
| 33 | +} |
| 34 | + |
| 35 | +// SetModel asks the claude CLI to switch to a different model mid-session. |
| 36 | +// Blocks until the CLI acknowledges the change or the context is cancelled. |
| 37 | +func (s *Stream) SetModel(model string) error { |
| 38 | + return s.sendControlRequest("set_model", map[string]any{"model": model}) |
| 39 | +} |
| 40 | + |
| 41 | +// SetPermissionMode asks the claude CLI to change the permission mode mid-session. |
| 42 | +// Blocks until the CLI acknowledges the change or the context is cancelled. |
| 43 | +func (s *Stream) SetPermissionMode(mode PermissionMode) error { |
| 44 | + return s.sendControlRequest("set_permission_mode", map[string]any{ |
| 45 | + "permission_mode": string(mode), |
| 46 | + }) |
| 47 | +} |
| 48 | + |
| 49 | +// SetMaxThinkingTokens asks the claude CLI to update the max thinking token budget. |
| 50 | +// Blocks until the CLI acknowledges the change or the context is cancelled. |
| 51 | +func (s *Stream) SetMaxThinkingTokens(n int) error { |
| 52 | + return s.sendControlRequest("set_max_thinking_tokens", map[string]any{ |
| 53 | + "max_thinking_tokens": n, |
| 54 | + }) |
| 55 | +} |
| 56 | + |
| 57 | +// Interrupt initiates graceful shutdown of the session: stdin is closed and |
| 58 | +// SIGTERM is sent to the claude subprocess. If the process does not exit within |
| 59 | +// 5 seconds, SIGKILL is sent. Interrupt is idempotent. |
| 60 | +func (s *Stream) Interrupt() error { |
| 61 | + s.interrupt() |
| 62 | + return nil |
| 63 | +} |
| 64 | + |
| 65 | +// sendControlRequest writes a control_request with the given subtype and extra |
| 66 | +// fields, then blocks until a matching control_response arrives or the ctx |
| 67 | +// is cancelled. |
| 68 | +func (s *Stream) sendControlRequest(subtype string, extras map[string]any) error { |
| 69 | + reqID := newUUID() |
| 70 | + respCh := make(chan controlResponse, 1) |
| 71 | + |
| 72 | + s.pendingMu.Lock() |
| 73 | + s.pending[reqID] = respCh |
| 74 | + s.pendingMu.Unlock() |
| 75 | + |
| 76 | + req := map[string]any{"subtype": subtype} |
| 77 | + for k, v := range extras { |
| 78 | + req[k] = v |
| 79 | + } |
| 80 | + |
| 81 | + err := s.write(map[string]any{ |
| 82 | + "type": "control_request", |
| 83 | + "request_id": reqID, |
| 84 | + "request": req, |
| 85 | + }) |
| 86 | + if err != nil { |
| 87 | + s.pendingMu.Lock() |
| 88 | + delete(s.pending, reqID) |
| 89 | + s.pendingMu.Unlock() |
| 90 | + return fmt.Errorf("claude: %s: %w", subtype, err) |
| 91 | + } |
| 92 | + |
| 93 | + select { |
| 94 | + case resp := <-respCh: |
| 95 | + if !resp.Success { |
| 96 | + return fmt.Errorf("claude: %s: %s", subtype, resp.Error) |
| 97 | + } |
| 98 | + return nil |
| 99 | + case <-s.ctx.Done(): |
| 100 | + s.pendingMu.Lock() |
| 101 | + delete(s.pending, reqID) |
| 102 | + s.pendingMu.Unlock() |
| 103 | + return s.ctx.Err() |
| 104 | + } |
| 105 | +} |
| 106 | + |
| 107 | +// Query runs the claude agent with the given prompt and returns a *Stream for |
| 108 | +// real-time event processing. |
| 109 | +// |
| 110 | +// The Stream.Events() channel is closed when the agent emits a TypeResult |
| 111 | +// message, the subprocess exits, or ctx is cancelled. Callers should always |
| 112 | +// range over the channel until it is closed. |
| 113 | +// |
| 114 | +// Stream control methods (SetModel, SetPermissionMode, SetMaxThinkingTokens, |
| 115 | +// Interrupt) may be called at any time while the stream is active. |
| 116 | +// |
| 117 | +// Example — stream all events: |
| 118 | +// |
| 119 | +// stream, err := claude.Query(ctx, "What is 2+2?") |
| 120 | +// if err != nil { ... } |
| 121 | +// for event := range stream.Events() { |
| 122 | +// switch event.Type { |
| 123 | +// case claude.TypeAssistant: |
| 124 | +// fmt.Print(event.Assistant.Text()) |
| 125 | +// case claude.TypeResult: |
| 126 | +// fmt.Println("session:", event.Result.SessionID) |
| 127 | +// } |
| 128 | +// } |
| 129 | +func Query(ctx context.Context, prompt string, opts ...Option) (*Stream, error) { |
| 130 | + o := defaultOptions() |
| 131 | + for _, opt := range opts { |
| 132 | + opt(o) |
| 133 | + } |
| 134 | + return spawnAndStream(ctx, o, prompt) |
| 135 | +} |
| 136 | + |
| 137 | +// Run is a convenience wrapper around Query that blocks until the agent |
| 138 | +// finishes and returns only the final Result. |
| 139 | +// |
| 140 | +// Intermediate events (streaming deltas, system messages, rate-limit events) |
| 141 | +// are discarded. Use Query directly if you need to process them. |
| 142 | +// |
| 143 | +// Errors from the subprocess itself (bad flags, auth failures, crashes) are |
| 144 | +// surfaced as Go errors so callers always get a meaningful message. |
| 145 | +// |
| 146 | +// Example: |
| 147 | +// |
| 148 | +// result, err := claude.Run(ctx, "What is 2+2?", |
| 149 | +// claude.WithModel("claude-haiku-4-5-20251001"), |
| 150 | +// claude.WithThinking(claude.ThinkingDisabled), |
| 151 | +// ) |
| 152 | +// if err != nil { ... } |
| 153 | +// fmt.Println(result.Result) |
| 154 | +// fmt.Println("session:", result.SessionID) |
| 155 | +func Run(ctx context.Context, prompt string, opts ...Option) (*Result, error) { |
| 156 | + stream, err := Query(ctx, prompt, opts...) |
| 157 | + if err != nil { |
| 158 | + return nil, err |
| 159 | + } |
| 160 | + |
| 161 | + for event := range stream.Events() { |
| 162 | + switch event.Type { |
| 163 | + |
| 164 | + case TypeResult: |
| 165 | + r := event.Result |
| 166 | + if r.IsError { |
| 167 | + msg := r.Subtype |
| 168 | + if len(r.Errors) > 0 { |
| 169 | + msg = strings.Join(r.Errors, "; ") |
| 170 | + } |
| 171 | + return nil, fmt.Errorf("claude: agent error (%s): %s", r.Subtype, msg) |
| 172 | + } |
| 173 | + return r, nil |
| 174 | + |
| 175 | + case TypeSystem: |
| 176 | + // Surface process-level errors (bad flag, auth failure, crash) that |
| 177 | + // were synthesised by spawnAndStream because no result message arrived. |
| 178 | + if event.System != nil && event.System.Subtype == "error" { |
| 179 | + return nil, fmt.Errorf("claude: %s", event.System.Message) |
| 180 | + } |
| 181 | + } |
| 182 | + } |
| 183 | + |
| 184 | + return nil, fmt.Errorf("claude: agent finished without a result message") |
| 185 | +} |
0 commit comments