package cli import ( "bufio" "bytes" "context" "encoding/json" "errors" "fmt" "io" "net/http" "os/exec" "regexp" "strings" "sync" "time" "go.uber.org/zap" "iop/apps/node/internal/runtime" "iop/packages/go/config" ) // opencodeSSESession tracks a logical session backed by an OpenCode server. // When owned is true, the local server process is started by the adapter and // must be killed on close. When false, the adapter attaches to an external // server via --attach and never owns the process lifecycle. type opencodeSSESession struct { key sessionKey serverURL string sessionID string cmd *exec.Cmd owned bool mu sync.Mutex } // opencodeRunOpts is the subset of opencode CLI options that affect SSE mode. type opencodeRunOpts struct { AttachURL string Model string Agent string Variant string Title string Dir string DangerouslySkipPermissions bool } func parseOpencodeRunArgs(args []string) opencodeRunOpts { var opts opencodeRunOpts for i := 0; i < len(args); i++ { a := args[i] next := func() (string, bool) { if i+1 < len(args) { v := args[i+1] i++ return v, true } return "", false } switch a { case "--attach": if v, ok := next(); ok { opts.AttachURL = v } case "--model": if v, ok := next(); ok { opts.Model = v } case "--agent": if v, ok := next(); ok { opts.Agent = v } case "--variant": if v, ok := next(); ok { opts.Variant = v } case "--title": if v, ok := next(); ok { opts.Title = v } case "--dir": if v, ok := next(); ok { opts.Dir = v } case "--dangerously-skip-permissions": opts.DangerouslySkipPermissions = true } } return opts } // decodeSSEDataLine returns the JSON payload of an SSE "data:" line, or // (nil, false) if the line is not a usable data line. func decodeSSEDataLine(line []byte) ([]byte, bool) { if !bytes.HasPrefix(line, []byte("data:")) { return nil, false } data := bytes.TrimSpace(line[len("data:"):]) if len(data) == 0 { return nil, false } return data, true } func (c *CLI) executeOpencodeSSE(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, sink runtime.EventSink) error { opts := parseOpencodeRunArgs(profile.Args) sess, err := c.resolveOpencodeSession(ctx, spec, profile, opts) if err != nil { return err } sess.mu.Lock() defer sess.mu.Unlock() // Open SSE stream before sending the prompt so early events are not missed. // OpenCode 1.14.x publishes session events through /global/event as a // top-level payload wrapper; /event only reports server connection events. sseReq, err := http.NewRequestWithContext(ctx, http.MethodGet, sess.serverURL+"/global/event", nil) if err != nil { return fmt.Errorf("cli adapter: opencode sse new request: %w", err) } sseReq.Header.Set("Accept", "text/event-stream") sseResp, err := http.DefaultClient.Do(sseReq) if err != nil { _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("opencode sse connect: %v", err), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: opencode sse connect: %w", err) } defer sseResp.Body.Close() if sseResp.StatusCode != http.StatusOK { body, _ := io.ReadAll(io.LimitReader(sseResp.Body, 1024)) msg := fmt.Sprintf("opencode sse status %d: %s", sseResp.StatusCode, strings.TrimSpace(string(body))) _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: msg, Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: %s", msg) } if sess.sessionID == "" { id, err := opencodeCreateSession(ctx, sess.serverURL, opts.Title) if err != nil { _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("opencode create session: %v", err), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: opencode create session: %w", err) } sess.sessionID = id } _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeStart, Timestamp: time.Now(), }) prompt := extractPrompt(spec.Input) if err := opencodePromptAsync(ctx, sess.serverURL, sess.sessionID, prompt, opts); err != nil { _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("opencode prompt async: %v", err), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: opencode prompt async: %w", err) } return c.driveOpencodeSSE(ctx, spec, sess, opts, sseResp.Body, sink) } func (c *CLI) resolveOpencodeSession(ctx context.Context, spec runtime.ExecutionSpec, profile config.CLIProfileConf, opts opencodeRunOpts) (*opencodeSSESession, error) { target := cliTargetName(spec) key := sessionKey{target: target, sessionID: normalizeSessionID(spec.SessionID)} c.mu.Lock() if sess, ok := c.opencodeSessions[key]; ok { c.mu.Unlock() return sess, nil } if spec.SessionMode == runtime.SessionModeRequireExisting { c.mu.Unlock() return nil, fmt.Errorf("cli adapter: no persistent session for target %q session %q", target, key.sessionID) } c.mu.Unlock() serverURL := strings.TrimRight(opts.AttachURL, "/") var cmd *exec.Cmd owned := false if serverURL == "" { url, started, err := startOpencodeLocalServer(ctx, profile, c.logger) if err != nil { return nil, fmt.Errorf("cli adapter: start opencode server: %w", err) } serverURL = url cmd = started owned = true } sess := &opencodeSSESession{ key: key, serverURL: serverURL, cmd: cmd, owned: owned, } c.mu.Lock() if existing, ok := c.opencodeSessions[key]; ok { c.mu.Unlock() if owned && cmd != nil && cmd.Process != nil { _ = cmd.Process.Kill() } return existing, nil } c.opencodeSessions[key] = sess c.mu.Unlock() return sess, nil } var opencodeServerListenRE = regexp.MustCompile(`(?i)opencode server listening on (https?://\S+)`) func startOpencodeLocalServer(ctx context.Context, profile config.CLIProfileConf, logger *zap.Logger) (string, *exec.Cmd, error) { args := []string{"serve", "--hostname", "127.0.0.1", "--port", "0"} cmd := exec.Command(profile.Command, args...) if len(profile.Env) > 0 { cmd.Env = append(cmd.Environ(), profile.Env...) } stdout, err := cmd.StdoutPipe() if err != nil { return "", nil, fmt.Errorf("stdout pipe: %w", err) } stderr, err := cmd.StderrPipe() if err != nil { return "", nil, fmt.Errorf("stderr pipe: %w", err) } if err := cmd.Start(); err != nil { return "", nil, fmt.Errorf("start opencode serve: %w", err) } urlCh := make(chan string, 2) scan := func(r io.Reader, name string) { s := bufio.NewScanner(r) s.Buffer(make([]byte, 64*1024), 1024*1024) for s.Scan() { line := s.Text() if logger != nil { logger.Debug("opencode server log", zap.String("stream", name), zap.String("line", line)) } if m := opencodeServerListenRE.FindStringSubmatch(line); len(m) >= 2 { select { case urlCh <- m[1]: default: } } } } go scan(stdout, "stdout") go scan(stderr, "stderr") select { case u := <-urlCh: return strings.TrimRight(u, "/"), cmd, nil case <-time.After(30 * time.Second): _ = cmd.Process.Kill() return "", nil, errors.New("timeout waiting for opencode server listen line") case <-ctx.Done(): _ = cmd.Process.Kill() return "", nil, ctx.Err() } } type opencodeEnvelope struct { Type string `json:"type"` Properties map[string]any `json:"properties"` } type opencodeGlobalEnvelope struct { Payload *opencodeEnvelope `json:"payload"` } func opencodeEventEnvelope(data []byte) (opencodeEnvelope, bool) { var env opencodeEnvelope if err := json.Unmarshal(data, &env); err == nil && env.Type != "" && env.Type != "sync" { return env, true } var global opencodeGlobalEnvelope if err := json.Unmarshal(data, &global); err != nil || global.Payload == nil || global.Payload.Type == "" || global.Payload.Type == "sync" { return opencodeEnvelope{}, false } return *global.Payload, true } // sseEvtTrace summarises one received SSE event for timeout/cancel diagnostics. type sseEvtTrace struct { Type string `json:"t"` SID string `json:"sid,omitempty"` // "ok", "mm" (mismatch), or "" Outcome string `json:"out,omitempty"` // "delta", "idle", "error", "skip", "filtered" } func (c *CLI) driveOpencodeSSE(ctx context.Context, spec runtime.ExecutionSpec, sess *opencodeSSESession, opts opencodeRunOpts, body io.Reader, sink runtime.EventSink) error { type evt struct { ev opencodeEnvelope raw []byte } events := make(chan evt, 32) scanErrCh := make(chan error, 1) go func() { defer close(events) s := bufio.NewScanner(body) s.Buffer(make([]byte, 64*1024), 4*1024*1024) for s.Scan() { line := s.Bytes() data, ok := decodeSSEDataLine(line) if !ok { continue } env, ok := opencodeEventEnvelope(data) if !ok { continue } cp := make([]byte, len(data)) copy(cp, data) events <- evt{ev: env, raw: cp} } scanErrCh <- s.Err() }() var ( inputTokens int outputTokens int partTypes = map[string]string{} ) // Bounded ring buffer for cancel/timeout diagnostics. const traceSize = 8 var ( traceRing [traceSize]sseEvtTrace traceHead int traceSeen int ) addTrace := func(entry sseEvtTrace) { traceRing[traceHead%traceSize] = entry traceHead++ traceSeen++ } recentTrace := func() []sseEvtTrace { n := traceHead if n > traceSize { n = traceSize } out := make([]sseEvtTrace, n) start := traceHead - n for i := 0; i < n; i++ { out[i] = traceRing[(start+i)%traceSize] } return out } finalize := func(reason error) error { bg, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() if sess.sessionID != "" { _ = opencodeAbort(bg, sess.serverURL, sess.sessionID) } if c.logger != nil { c.logger.Warn("opencode sse run cancelled without completion", zap.String("run_id", spec.RunID), zap.String("reason", reason.Error()), zap.Int("total_events", traceSeen), zap.Any("recent_events", recentTrace()), ) } _ = sink.Emit(context.Background(), runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeCancelled, Message: cancelEventForContext(reason), Timestamp: time.Now(), }) return runtime.ErrRunCancelled } for { select { case <-ctx.Done(): return finalize(ctx.Err()) case e, ok := <-events: if !ok { if ctx.Err() != nil { return finalize(ctx.Err()) } if err := <-scanErrCh; err != nil && !errors.Is(err, io.EOF) { _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: fmt.Sprintf("opencode sse scan: %v", err), Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: opencode sse scan: %w", err) } return emitOpencodeComplete(ctx, sink, spec.RunID, inputTokens, outputTokens) } // Filter events that carry a sessionID mismatching ours, when present. if sid, ok := stringFromProp(e.ev.Properties, "sessionID"); ok && sid != "" && sess.sessionID != "" && sid != sess.sessionID { addTrace(sseEvtTrace{Type: e.ev.Type, SID: "mm", Outcome: "filtered"}) continue } sidStatus := "" if sid, ok := stringFromProp(e.ev.Properties, "sessionID"); ok && sid != "" { sidStatus = "ok" } switch e.ev.Type { case "session.next.text.delta": delta := opencodeDeltaText(e.ev.Properties) if delta != "" { outputTokens += len(strings.Fields(delta)) _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeDelta, Delta: delta, Timestamp: time.Now(), }) addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "delta"}) } else { addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "skip"}) } case "message.part.updated": if partID, partType, ok := opencodePartInfo(e.ev.Properties); ok { partTypes[partID] = partType } if in, out, ok := opencodeStepTokens(e.ev.Properties); ok { if in > 0 { inputTokens = in } if out > 0 { outputTokens = out } } addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "ok"}) case "message.part.delta": delta := opencodeMessagePartDeltaText(e.ev.Properties, partTypes) if delta != "" { outputTokens += len(strings.Fields(delta)) _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeDelta, Delta: delta, Timestamp: time.Now(), }) addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "delta"}) } else { addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "skip"}) } case "session.next.step.ended": if in, out, ok := opencodeStepTokens(e.ev.Properties); ok { if in > 0 { inputTokens = in } if out > 0 { outputTokens = out } } addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "ok"}) case "session.idle": addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "idle"}) return emitOpencodeComplete(ctx, sink, spec.RunID, inputTokens, outputTokens) case "session.status": if opencodeStatusIdle(e.ev.Properties) { addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "idle"}) return emitOpencodeComplete(ctx, sink, spec.RunID, inputTokens, outputTokens) } addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "skip"}) case "session.error": msg := opencodeErrorMessage(e.ev.Properties) addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "error"}) _ = sink.Emit(ctx, runtime.RuntimeEvent{ RunID: spec.RunID, Type: runtime.EventTypeError, Error: msg, Timestamp: time.Now(), }) return fmt.Errorf("cli adapter: opencode session error: %s", msg) case "permission.asked": permID, _ := stringFromProp(e.ev.Properties, "id") if permID == "" { permID, _ = stringFromProp(e.ev.Properties, "permissionID") } if permID != "" { reply := "reject" if opts.DangerouslySkipPermissions { reply = "once" } bg, cancel := context.WithTimeout(context.Background(), 5*time.Second) _ = opencodePermissionReply(bg, sess.serverURL, permID, reply) cancel() } addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "ok"}) default: addTrace(sseEvtTrace{Type: e.ev.Type, SID: sidStatus, Outcome: "skip"}) } } } } func stringFromProp(props map[string]any, key string) (string, bool) { if props == nil { return "", false } v, ok := props[key] if !ok { return "", false } s, ok := v.(string) return s, ok } func opencodeDeltaText(props map[string]any) string { if props == nil { return "" } if s, ok := props["delta"].(string); ok && s != "" { return s } if part, ok := props["part"].(map[string]any); ok { if s, ok := part["text"].(string); ok { return s } } return "" } func opencodePartInfo(props map[string]any) (string, string, bool) { if props == nil { return "", "", false } part, ok := props["part"].(map[string]any) if !ok { return "", "", false } id, _ := part["id"].(string) partType, _ := part["type"].(string) if id == "" || partType == "" { return "", "", false } return id, partType, true } func opencodeMessagePartDeltaText(props map[string]any, partTypes map[string]string) string { if props == nil { return "" } field, _ := props["field"].(string) if field != "text" { return "" } partID, _ := props["partID"].(string) // Only reject the delta when we positively know the part is not text type. // If message.part.updated was missed (e.g. second SSE connection opened after // the server already emitted it), partTypes has no entry and we allow the // delta rather than silently dropping it. if partID != "" { if pt, known := partTypes[partID]; known && pt != "text" { return "" } } delta, _ := props["delta"].(string) return delta } func opencodeStepTokens(props map[string]any) (int, int, bool) { if props == nil { return 0, 0, false } tokens, ok := props["tokens"].(map[string]any) if !ok { if part, partOK := props["part"].(map[string]any); partOK { tokens, ok = part["tokens"].(map[string]any) } } if !ok { return 0, 0, false } in := intFromNumber(tokens["input"]) out := intFromNumber(tokens["output"]) return in, out, true } func intFromNumber(v any) int { switch n := v.(type) { case float64: return int(n) case int: return n case json.Number: if i, err := n.Int64(); err == nil { return int(i) } } return 0 } func emitOpencodeComplete(ctx context.Context, sink runtime.EventSink, runID string, inputTokens, outputTokens int) error { return sink.Emit(ctx, runtime.RuntimeEvent{ RunID: runID, Type: runtime.EventTypeComplete, Message: "opencode sse execution complete", Usage: &runtime.UsageStats{ InputTokens: inputTokens, OutputTokens: outputTokens, }, Timestamp: time.Now(), }) } // opencodeModelPayload parses "provider/model" into the OpenCode prompt_async // model object {"providerID": provider, "modelID": model}. Returns an error // when either side is missing. func opencodeModelPayload(s string) (map[string]any, error) { idx := strings.Index(s, "/") if idx < 0 { return nil, fmt.Errorf("cli adapter: opencode model %q missing provider/model separator", s) } provider := s[:idx] model := s[idx+1:] if provider == "" || model == "" { return nil, fmt.Errorf("cli adapter: opencode model %q has empty provider or model", s) } return map[string]any{"providerID": provider, "modelID": model}, nil } // opencodeStatusIdle reports whether a session.status event's status payload // represents idle. OpenCode emits the status as a SessionStatus object // ({"type":"idle"}); older builds may use the bare string "idle". func opencodeStatusIdle(props map[string]any) bool { if props == nil { return false } switch v := props["status"].(type) { case string: return v == "idle" case map[string]any: t, _ := v["type"].(string) return t == "idle" } return false } func opencodeErrorMessage(props map[string]any) string { if props == nil { return "opencode error" } if errObj, ok := props["error"].(map[string]any); ok { if data, ok := errObj["data"].(map[string]any); ok { if m, ok := data["message"].(string); ok && m != "" { return m } } if name, ok := errObj["name"].(string); ok && name != "" { return name } } if m, ok := props["message"].(string); ok && m != "" { return m } return "opencode error" } // --- HTTP helpers --- func opencodeCreateSession(ctx context.Context, baseURL, title string) (string, error) { payload := map[string]any{} if title != "" { payload["title"] = title } var body bytes.Buffer if err := json.NewEncoder(&body).Encode(payload); err != nil { return "", err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/session", &body) if err != nil { return "", err } req.Header.Set("Content-Type", "application/json") resp, err := http.DefaultClient.Do(req) if err != nil { return "", err } defer resp.Body.Close() if resp.StatusCode/100 != 2 { b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) return "", fmt.Errorf("opencode create session status %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) } var raw map[string]any if err := json.NewDecoder(resp.Body).Decode(&raw); err != nil { return "", fmt.Errorf("decode session response: %w", err) } if id, ok := raw["id"].(string); ok && id != "" { return id, nil } if s, ok := raw["session"].(map[string]any); ok { if id, ok := s["id"].(string); ok && id != "" { return id, nil } } return "", errors.New("opencode create session: missing id in response") } func opencodePromptAsync(ctx context.Context, baseURL, sessionID, prompt string, opts opencodeRunOpts) error { payload := map[string]any{ "parts": []map[string]any{ {"type": "text", "text": prompt}, }, } if opts.Model != "" { model, err := opencodeModelPayload(opts.Model) if err != nil { return err } payload["model"] = model } if opts.Agent != "" { payload["agent"] = opts.Agent } if opts.Variant != "" { payload["variant"] = opts.Variant } var body bytes.Buffer if err := json.NewEncoder(&body).Encode(payload); err != nil { return err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/session/"+sessionID+"/prompt_async", &body) if err != nil { return err } req.Header.Set("Content-Type", "application/json") resp, err := http.DefaultClient.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode/100 != 2 { b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) return fmt.Errorf("status %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) } _, _ = io.Copy(io.Discard, resp.Body) return nil } func opencodePermissionReply(ctx context.Context, baseURL, permID, reply string) error { var body bytes.Buffer if err := json.NewEncoder(&body).Encode(map[string]any{"reply": reply}); err != nil { return err } req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/permission/"+permID+"/reply", &body) if err != nil { return err } req.Header.Set("Content-Type", "application/json") resp, err := http.DefaultClient.Do(req) if err != nil { return err } defer resp.Body.Close() _, _ = io.Copy(io.Discard, resp.Body) return nil } func opencodeAbort(ctx context.Context, baseURL, sessionID string) error { req, err := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/session/"+sessionID+"/abort", nil) if err != nil { return err } resp, err := http.DefaultClient.Do(req) if err != nil { return err } defer resp.Body.Close() _, _ = io.Copy(io.Discard, resp.Body) return nil } // closeOpencodeSession kills the owned local server process if any. For // attached sessions, no process is owned and no kill happens. func closeOpencodeSession(sess *opencodeSSESession) { if sess == nil { return } if sess.owned && sess.cmd != nil && sess.cmd.Process != nil { _ = sess.cmd.Process.Kill() } }