doctl agents proxy w/codex - M0-M2

args.go

@@ -951,4 +951,16 @@ const (
 
 	// ArgAgentArchive treats the workspace payload as a tar archive.
 	ArgAgentArchive = "archive"
+
+	// ArgAgentProxyType selects which coding-agent protocol `start-proxy` impersonates.
+	ArgAgentProxyType = "type"
+
+	// ArgAgentProxySession is the session ID or name the proxy bridges to.
+	ArgAgentProxySession = "session"
+
+	// ArgAgentProxyPort is the local port `start-proxy` listens on.
+	ArgAgentProxyPort = "port"
+
+	// ArgAgentProxyReplay replays the session's event history into the first thread on connect.
+	ArgAgentProxyReplay = "replay"
 )

commands/agents.go

@@ -29,6 +29,7 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"os/signal"
 	"path/filepath"
 	"regexp"
 	"strings"
@@ -41,6 +42,8 @@ import (
 	"github.com/digitalocean/doctl/commands/charm"
 	"github.com/digitalocean/doctl/commands/displayers"
 	"github.com/digitalocean/doctl/do"
+	"github.com/digitalocean/doctl/internal/agentproxy"
+	"github.com/digitalocean/doctl/internal/agentproxy/codex"
 	"github.com/digitalocean/godo"
 	"github.com/muesli/termenv"
 	"github.com/spf13/cobra"
@@ -231,6 +234,22 @@ Use `+"`"+`--name`+"`"+` to name the session (this sets the manifest's `+"`"+`me
 	AddStringFlag(cmdStart, doctl.ArgAgentName, "", "", "Name for the new session (sets the manifest's metadata.name). If omitted, the server auto-generates a name. Must be unique among your team's active sessions.")
 	cmdStart.Example = `doctl agents start --spec agent-spec.yaml --name my-session`
 
+	cmdStartProxy := CmdBuilder(cmd, RunAgentsStartProxy, "start-proxy",
+		"Run a local facade that lets a coding-agent CLI drive a hosted session",
+		`Starts a local WebSocket server that impersonates a coding-agent's own app-server protocol, so the unmodified CLI can attach to a hosted session as if it were a local backend.
+
+`+"`"+`--type`+"`"+` selects which protocol to impersonate (v1: `+"`"+`codex`+"`"+` only; future agents get their own facade behind this same command, not a new command). Once `+"`"+`start-proxy`+"`"+` is listening, connect the real CLI, e.g. `+"`"+`codex --remote ws://127.0.0.1:1144`+"`"+`.
+
+Tested against `+"`"+`codex-cli `+codex.TestedVersion+"`"+`. Codex's WS/app-server transport is officially experimental and can change without notice — re-verify the protocol capture on every codex upgrade before trusting this against a newer CLI.
+
+The harness allows one connected device per session — a concurrent `+"`"+`doctl agents attach`+"`"+` on the same session conflicts with the proxy's own stream. Close one before opening the other.`,
+		Writer)
+	AddStringFlag(cmdStartProxy, doctl.ArgAgentProxyType, "", "codex", "Coding-agent protocol to impersonate (v1: codex)")
+	AddStringFlag(cmdStartProxy, doctl.ArgAgentProxySession, "", "", "Session ID or name to bridge to", requiredOpt())
+	AddIntFlag(cmdStartProxy, doctl.ArgAgentProxyPort, "", 1144, "Local port to listen on")
+	AddBoolFlag(cmdStartProxy, doctl.ArgAgentProxyReplay, "", false, "Replay the session's event history into the first thread on connect (not yet implemented)")
+	cmdStartProxy.Example = `doctl agents start-proxy --type codex --session my-session --port 1144`
+
 	cmdAttach := CmdBuilder(cmd, RunAgentsAttach, "attach <session>",
 		"Attach to an agent session",
 		`Opens an interactive line-mode TUI on an existing session. Streams events from the server and accepts typed input. If the SSE connection drops, doctl shows Reconnecting... and retries automatically (5 attempts with backoff). If reconnection fails, it prints an error and stops the stream.
@@ -345,6 +364,54 @@ func RunAgentsStart(c *CmdConfig) error {
 	return c.Display(&displayers.HostedAgentSession{Sessions: []do.HostedAgentSession{*sess}})
 }
 
+// RunAgentsStartProxy runs a local WebSocket facade that impersonates a
+// coding-agent's own app-server protocol, bridging an unmodified agent CLI to
+// a hosted session.
+func RunAgentsStartProxy(c *CmdConfig) error {
+	proxyType, err := c.Doit.GetString(c.NS, doctl.ArgAgentProxyType)
+	if err != nil {
+		return err
+	}
+	if proxyType != "codex" {
+		return fmt.Errorf("unsupported --type %q; v1 supports only \"codex\"", proxyType)
+	}
+
+	sessionRef, err := c.Doit.GetString(c.NS, doctl.ArgAgentProxySession)
+	if err != nil {
+		return err
+	}
+	port, err := c.Doit.GetInt(c.NS, doctl.ArgAgentProxyPort)
+	if err != nil {
+		return err
+	}
+	// --replay is accepted now so the CLI surface doesn't change again in M1,
+	// but session-history replay isn't implemented until the harness bridge
+	// lands; read it only so an explicit --replay=true doesn't silently no-op
+	// without the flag help already having said so.
+	if _, err := c.Doit.GetBool(c.NS, doctl.ArgAgentProxyReplay); err != nil {
+		return err
+	}
+
+	svc := c.HostedAgents()
+	sessionID, err := resolveSessionRef(svc, sessionRef)
+	if err != nil {
+		return err
+	}
+	if _, err := svc.GetSession(sessionID); err != nil {
+		return fmt.Errorf("session %q not found: %w", sessionRef, err)
+	}
+
+	fmt.Fprintf(c.Out, "Proxying session %s as a codex app-server on ws://127.0.0.1:%d\n", sessionID, port)
+	fmt.Fprintf(c.Out, "Connect with: codex --remote ws://127.0.0.1:%d\n", port)
+
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer cancel()
+
+	return agentproxy.Serve(ctx, port, func() agentproxy.Facade {
+		return &codex.Facade{SessionID: sessionID, Sessions: svc}
+	})
+}
+
 // readManifest returns the spec file as raw bytes. path "-" reads from stdin.
 // The only client-side validation is "non-empty after trim" so a stray
 // `--spec /dev/null` fails fast instead of hitting the server.

internal/agentproxy/agentproxytest/harness.go

@@ -0,0 +1,172 @@
+// Package agentproxytest fakes the DigitalOcean Hosted Agents session HTTP
+// API (see github.com/digitalocean/godo's hosted_agents.go) behind an
+// httptest.Server, so agentproxy tests can drive a real
+// do.HostedAgentsService end to end without a live harness backend.
+//
+// Nothing outside this package's own tests (or other packages' _test.go
+// files) should ever import it: it's test scaffolding, not production code,
+// even though it lives in an ordinary buildable package rather than a
+// _test.go file — that's what lets other packages' tests (e.g.
+// internal/agentproxy/codex's) import it across package boundaries.
+package agentproxytest
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+)
+
+// Event is one canned SSE event the harness streams back from
+// GET /v2/agents/sessions/{id}/stream, matching the event-specific part of
+// godo.HostedAgentEvent's wire shape (see HostedAgentEventKind's doc comment
+// for the canonical type strings).
+type Event struct {
+	// Type is the wire "type" discriminator, e.g. "run.started",
+	// "run.token_delta", "run.completed", "run.failed".
+	Type string
+	// Data is the event-specific payload. A nil Data marshals as "{}".
+	Data json.RawMessage
+}
+
+// eventWire is the on-the-wire SPI canonical event envelope this harness
+// emits. It mirrors godo's unexported hostedAgentEventWire field-for-field
+// (confirmed against vendor/github.com/digitalocean/godo/hosted_agents.go)
+// rather than importing it, since that type isn't exported.
+type eventWire struct {
+	EventID   string          `json:"event_id"`
+	RunID     string          `json:"run_id"`
+	TenantID  string          `json:"tenant_id"`
+	SessionID string          `json:"session_id"`
+	Timestamp string          `json:"timestamp"`
+	Seq       int             `json:"seq"`
+	Type      string          `json:"type"`
+	Data      json.RawMessage `json:"data"`
+}
+
+// Harness fakes the four Hosted Agents session routes an agentproxy test
+// needs: session lookup, SSE stream, send-input, and HITL resolve. Point a
+// *godo.Client at it with godo.SetBaseURL(h.Server.URL+"/") — do's
+// NewHostedAgentsService wraps that client directly, so a Facade built
+// against it talks to this harness exactly as it would a real backend.
+type Harness struct {
+	Server *httptest.Server
+
+	mu        sync.Mutex
+	sessionID string
+	runID     string  // returned by the next POST .../input call
+	events    []Event // streamed, in order, by the next GET .../stream call
+}
+
+// New starts the fake harness and registers its shutdown via t.Cleanup.
+// sessionID is the session id every route responds under — a test's
+// Facade.SessionID should match it.
+func New(t *testing.T, sessionID string) *Harness {
+	t.Helper()
+	h := &Harness{sessionID: sessionID}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /v2/agents/sessions/{id}", h.handleGetSession)
+	mux.HandleFunc("GET /v2/agents/sessions/{id}/stream", h.handleStream)
+	mux.HandleFunc("POST /v2/agents/sessions/{id}/input", h.handleInput)
+	mux.HandleFunc("POST /v2/agents/sessions/{id}/hitl/{requestID}", h.handleHITL)
+
+	h.Server = httptest.NewServer(mux)
+	t.Cleanup(h.Server.Close)
+	return h
+}
+
+// QueueRun arranges for the next POST .../input call to return runID, and
+// for GET .../stream to then emit events (in order, tagged with runID),
+// flushing after each one so a concurrent reader observes them incrementally
+// rather than all at once at EOF.
+//
+// One (runID, events) pair is enough for a single-turn sequence test; queue
+// again before the next turn/start if a future test drives more than one
+// turn over the same harness.
+func (h *Harness) QueueRun(runID string, events ...Event) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.runID = runID
+	h.events = events
+}
+
+func (h *Harness) handleGetSession(w http.ResponseWriter, _ *http.Request) {
+	h.mu.Lock()
+	sessionID := h.sessionID
+	h.mu.Unlock()
+
+	w.Header().Set("Content-Type", "application/json")
+	fmt.Fprintf(w, `{"session":{"session_id":%q,"team_id":15726539,`+
+		`"agent_kind":"AGENT_KIND_CODEX_CLI","status":"SESSION_STATUS_READY",`+
+		`"created_at":"2026-01-01T00:00:00Z","last_event_at":"2026-01-01T00:00:00Z"}}`,
+		sessionID)
+}
+
+func (h *Harness) handleInput(w http.ResponseWriter, r *http.Request) {
+	h.mu.Lock()
+	runID := h.runID
+	h.mu.Unlock()
+
+	if runID == "" {
+		http.Error(w, "agentproxytest: no run id queued; call Harness.QueueRun first", http.StatusInternalServerError)
+		return
+	}
+
+	// Best-effort decode only — the harness doesn't assert on the input
+	// text; callers that care can inspect the request in a custom handler.
+	var body struct {
+		Text string `json:"text"`
+	}
+	_ = json.NewDecoder(r.Body).Decode(&body)
+
+	w.Header().Set("Content-Type", "application/json")
+	_ = json.NewEncoder(w).Encode(map[string]string{"run_id": runID})
+}
+
+func (h *Harness) handleStream(w http.ResponseWriter, _ *http.Request) {
+	h.mu.Lock()
+	runID := h.runID
+	events := h.events
+	sessionID := h.sessionID
+	h.mu.Unlock()
+
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.WriteHeader(http.StatusOK)
+	flusher, canFlush := w.(http.Flusher)
+
+	for i, ev := range events {
+		data := ev.Data
+		if data == nil {
+			data = json.RawMessage("{}")
+		}
+		wire := eventWire{
+			EventID:   fmt.Sprintf("evt-%d", i),
+			RunID:     runID,
+			TenantID:  "15726539",
+			SessionID: sessionID,
+			Timestamp: "2026-01-01T00:00:00Z",
+			Seq:       i,
+			Type:      ev.Type,
+			Data:      data,
+		}
+		encoded, err := json.Marshal(wire)
+		if err != nil {
+			// A test handed the harness a payload that can't marshal — a
+			// programmer error in the test, not a runtime condition.
+			panic(fmt.Sprintf("agentproxytest: event does not marshal to JSON: %v", err))
+		}
+
+		fmt.Fprintf(w, "event: %s\ndata: %s\n\n", ev.Type, encoded)
+		if canFlush {
+			flusher.Flush()
+		}
+	}
+}
+
+func (h *Harness) handleHITL(w http.ResponseWriter, _ *http.Request) {
+	w.WriteHeader(http.StatusNoContent)
+}