diff --git a/commands/agents.go b/commands/agents.go index b42efa323..14d73ee5f 100644 --- a/commands/agents.go +++ b/commands/agents.go @@ -68,19 +68,6 @@ var ( colMuted = charm.Colors.Muted ) -// Stream-state transport frames (SSE kind "stream.state"). Defined locally so -// doctl builds against godo pins that have not yet exported HostedAgentEventKindStreamState -// / HostedAgentStreamState*. Wire values match the published godo API. -const ( - hostedAgentEventKindStreamState godo.HostedAgentEventKind = "stream.state" - hostedAgentStreamStateSuperseded = "superseded" -) - -type hostedAgentStreamState struct { - State string `json:"state"` - Cursor string `json:"cursor,omitempty"` -} - // detectStyling reports whether ANSI styling should be emitted for the current // process: stdout is a terminal and NO_COLOR is unset. func detectStyling() bool { @@ -231,7 +218,9 @@ func Agents() *Command { A session is one long-lived agent process (Claude Code, OpenCode, ...) running inside a workspace sandbox. doctl drives it: starting it from an agent spec, attaching an interactive TUI, listing existing sessions, resolving HITL approvals out of band, and tearing it down. -Commands that act on a single session accept either the session ID or its name. A name must match exactly one session; if it is ambiguous, pass the session ID instead.`, +Commands that act on a single session accept either the session ID or its name. A name must match exactly one session; if it is ambiguous, pass the session ID instead. + +These commands talk to the hosted-agents endpoint (` + "`" + `https://ohr-agent.do-ai.run` + "`" + `) rather than ` + "`" + `https://api.digitalocean.com` + "`" + `. Pass ` + "`" + `--api-url` + "`" + ` (or set ` + "`" + `DIGITALOCEAN_API_URL` + "`" + `) to point them somewhere else.`, GroupID: hostedAgentsGroup, }, } @@ -1298,7 +1287,7 @@ func RunAgentsLogs(c *CmdConfig) error { for stream.Next() { ev := stream.Current() // Connection health, not session activity — never part of the history. - if ev.Kind == hostedAgentEventKindStreamState { + if ev.Kind == godo.HostedAgentEventKindStreamState { continue } if ev.Kind == godo.HostedAgentEventKindTokenChunk { @@ -2297,9 +2286,9 @@ func drainStream(stream *godo.HostedAgentSessionStream, out io.Writer, pending * // stream.state reports the health of the connection, not session // activity, so it never renders and never moves the cursor. - if ev.Kind == hostedAgentEventKindStreamState { - var st hostedAgentStreamState - if err := json.Unmarshal(ev.Payload, &st); err == nil && st.State == hostedAgentStreamStateSuperseded { + if ev.Kind == godo.HostedAgentEventKindStreamState { + var st godo.HostedAgentStreamState + if err := json.Unmarshal(ev.Payload, &st); err == nil && st.State == godo.HostedAgentStreamStateSuperseded { thinking.stop() acc.flush(out) flushAwaitingApproval(out, &awaiting) diff --git a/commands/agents_test.go b/commands/agents_test.go index 95e19fb3a..44bc7fbd1 100644 --- a/commands/agents_test.go +++ b/commands/agents_test.go @@ -2319,7 +2319,7 @@ func TestStreamWithReconnect_supersededStopsWithoutReconnect(t *testing.T) { stubReconnectSleep(t) body := sseFrame("evt-1", string(godo.HostedAgentEventKindSessionUpdated), `{}`) + - sseFrame("", string(hostedAgentEventKindStreamState), `{"state":"superseded","cursor":""}`) + sseFrame("", string(godo.HostedAgentEventKindStreamState), `{"state":"superseded","cursor":""}`) srv := httptest.NewServer(hostedAgentSSEHandler(body, nil)) t.Cleanup(srv.Close) @@ -2378,9 +2378,9 @@ func TestDrainStream_HITLReattachShowsCommand(t *testing.T) { // frame is transport bookkeeping: it renders nothing and must not become the // reconnect cursor, or a reconnect would resume from a position no event holds. func TestDrainStream_skipsStreamStateControlFrames(t *testing.T) { - body := sseFrame("", string(hostedAgentEventKindStreamState), `{"state":"live","cursor":""}`) + + body := sseFrame("", string(godo.HostedAgentEventKindStreamState), `{"state":"live","cursor":""}`) + sseFrame("evt-7", string(godo.HostedAgentEventKindSessionUpdated), `{}`) + - sseFrame("", string(hostedAgentEventKindStreamState), `{"state":"catching_up","cursor":""}`) + sseFrame("", string(godo.HostedAgentEventKindStreamState), `{"state":"catching_up","cursor":""}`) srv := httptest.NewServer(hostedAgentSSEHandler(body, nil)) t.Cleanup(srv.Close) @@ -2495,8 +2495,9 @@ func TestStreamWithReconnect_replayCursorAfterMidStreamDrop(t *testing.T) { mu.Lock() calls++ n := calls - // Resume cursor rides as replay_from on control-plane /stream. - replayFrom := r.URL.Query().Get("replay_from") + // The live stream carries the resume cursor in the standard SSE + // Last-Event-ID header, not a replay_from query parameter. + replayFrom := r.Header.Get("Last-Event-ID") mu.Unlock() w.Header().Set("Content-Type", "text/event-stream") diff --git a/commands/command_config.go b/commands/command_config.go index 44254c4dc..3f09b6e82 100644 --- a/commands/command_config.go +++ b/commands/command_config.go @@ -17,6 +17,7 @@ import ( "fmt" "io" + "github.com/digitalocean/godo" "github.com/spf13/cobra" "github.com/spf13/viper" @@ -111,6 +112,14 @@ func NewCmdConfig(ns string, dc doctl.Config, out io.Writer, args []string, init return fmt.Errorf("Unable to initialize DigitalOcean API client: %s", err) } + // The hosted-agents surface lives on its own host rather than + // api.digitalocean.com, so it gets its own client. --api-url still + // overrides both (see doctl.HostedAgentsAPIURL). + agentsClient, err := c.Doit.GetGodoClient(Trace, true, accessToken, godo.SetBaseURL(doctl.HostedAgentsAPIURL)) + if err != nil { + return fmt.Errorf("Unable to initialize DigitalOcean Hosted Agents API client: %s", err) + } + c.Keys = func() do.KeysService { return do.NewKeysService(godoClient) } c.Sizes = func() do.SizesService { return do.NewSizesService(godoClient) } c.Regions = func() do.RegionsService { return do.NewRegionsService(godoClient) } @@ -167,9 +176,9 @@ func NewCmdConfig(ns string, dc doctl.Config, out io.Writer, args []string, init c.Nfs = func() do.NfsService { return do.NewNfsService(godoClient) } c.NfsActions = func() do.NfsActionsService { return do.NewNfsActionsService(godoClient) } c.Security = func() do.SecurityService { return do.NewSecurityService(godoClient) } - c.HostedAgents = func() do.HostedAgentsService { return do.NewHostedAgentsService(godoClient) } + c.HostedAgents = func() do.HostedAgentsService { return do.NewHostedAgentsService(agentsClient) } c.HostedAgentTriggers = func() do.HostedAgentTriggersService { - return do.NewHostedAgentTriggersService(godoClient) + return do.NewHostedAgentTriggersService(agentsClient) } c.Secrets = func() do.SecretsService { return do.NewSecretsService(godoClient) } return nil diff --git a/doit.go b/doit.go index 3df852a9f..71db6d209 100644 --- a/doit.go +++ b/doit.go @@ -47,6 +47,16 @@ import ( const ( // LatestReleaseURL is the latest release URL endpoint. LatestReleaseURL = "https://api.github.com/repos/digitalocean/doctl/releases/latest" + + // HostedAgentsAPIURL is the API endpoint the `doctl agents` commands use. + // Hosted agents are fronted by their own host, which serves both the + // session control plane (/v2/agents/...) and the data-plane event stream + // (.../events), rather than by api.digitalocean.com. + // + // An explicit --api-url (DIGITALOCEAN_API_URL) still wins, which is how a + // non-production environment is reached — the preview host, for instance, + // is https://ohr-agent.do-ai-test.run. + HostedAgentsAPIURL = "https://ohr-agent.do-ai.run/" ) // Version is the version info for doit. @@ -209,7 +219,7 @@ func (glv *GithubLatestVersioner) LatestVersion() (string, error) { // Config is an interface that represent doit's config. type Config interface { - GetGodoClient(trace, allowRetries bool, accessToken string) (*godo.Client, error) + GetGodoClient(trace, allowRetries bool, accessToken string, opts ...godo.ClientOpt) (*godo.Client, error) GetDockerEngineClient() (builder.DockerEngineClient, error) SSH(user, host, keyPath string, port int, opts ssh.Options) runner.Runner Listen(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, inCh <-chan []byte) listen.ListenerService @@ -235,8 +245,10 @@ type LiveConfig struct { var _ Config = &LiveConfig{} -// GetGodoClient returns a GodoClient. -func (c *LiveConfig) GetGodoClient(trace, allowRetries bool, accessToken string) (*godo.Client, error) { +// GetGodoClient returns a GodoClient. opts are applied before the --api-url +// override, so a caller-supplied default base URL (see HostedAgentsAPIURL) +// yields to an endpoint the user asked for explicitly. +func (c *LiveConfig) GetGodoClient(trace, allowRetries bool, accessToken string, opts ...godo.ClientOpt) (*godo.Client, error) { if accessToken == "" { return nil, fmt.Errorf("access token is required. (hint: run 'doctl auth init')") } @@ -273,6 +285,8 @@ func (c *LiveConfig) GetGodoClient(trace, allowRetries bool, accessToken string) args = append(args, godo.WithRetryAndBackoffs(retryConfig)) } + args = append(args, opts...) + apiURL := viper.GetString("api-url") if apiURL != "" { args = append(args, godo.SetBaseURL(apiURL)) @@ -535,7 +549,7 @@ func NewTestConfig() *TestConfig { // GetGodoClient mocks a GetGodoClient call. The returned godo client will // be nil. -func (c *TestConfig) GetGodoClient(trace, allowRetries bool, accessToken string) (*godo.Client, error) { +func (c *TestConfig) GetGodoClient(trace, allowRetries bool, accessToken string, opts ...godo.ClientOpt) (*godo.Client, error) { return &godo.Client{}, nil } diff --git a/doit_test.go b/doit_test.go index c4da474fc..5de702da3 100644 --- a/doit_test.go +++ b/doit_test.go @@ -17,6 +17,9 @@ import ( "os" "regexp" "testing" + + "github.com/digitalocean/godo" + "github.com/spf13/viper" ) func TestMain(m *testing.M) { @@ -128,6 +131,55 @@ func (slr stubLatestRelease) LatestVersion() (string, error) { return slr.version, nil } +// TestGetGodoClientBaseURL pins the precedence that lets the `doctl agents` +// commands default to their own host without taking the endpoint away from a +// user who named one: caller-supplied options set a default, --api-url +// overrides it. +func TestGetGodoClientBaseURL(t *testing.T) { + cases := []struct { + name string + apiURL string + opts []godo.ClientOpt + want string + }{ + { + name: "no options and no api-url leaves godo's default", + want: "https://api.digitalocean.com/", + }, + { + name: "caller-supplied base URL applies", + opts: []godo.ClientOpt{godo.SetBaseURL(HostedAgentsAPIURL)}, + want: HostedAgentsAPIURL, + }, + { + name: "api-url wins over a caller-supplied base URL", + apiURL: "https://ohr-agent.do-ai-test.run/", + opts: []godo.ClientOpt{godo.SetBaseURL(HostedAgentsAPIURL)}, + want: "https://ohr-agent.do-ai-test.run/", + }, + { + name: "api-url applies with no caller options", + apiURL: "https://example.test/", + want: "https://example.test/", + }, + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + viper.Set("api-url", c.apiURL) + t.Cleanup(func() { viper.Set("api-url", "") }) + + client, err := (&LiveConfig{}).GetGodoClient(false, false, "fake-token", c.opts...) + if err != nil { + t.Fatalf("GetGodoClient() unexpected error: %v", err) + } + if got := client.BaseURL.String(); got != c.want { + t.Errorf("BaseURL = %q; want %q", got, c.want) + } + }) + } +} + func TestCommandName(t *testing.T) { t.Run("snap name set to goland", func(t *testing.T) { const snapName = "goland" diff --git a/internal/agentproxy/agentproxytest/harness.go b/internal/agentproxy/agentproxytest/harness.go index 8982fea87..cec6b5d00 100644 --- a/internal/agentproxy/agentproxytest/harness.go +++ b/internal/agentproxy/agentproxytest/harness.go @@ -18,10 +18,12 @@ import ( "sync" "testing" "time" + + "github.com/digitalocean/godo" ) // Event is one canned SSE event the harness streams back from -// GET /v2/agents/sessions/{id}/stream, matching the event-specific part of +// GET /v2/agents/sessions/{id}/events, matching the event-specific part of // godo.HostedAgentEvent's wire shape (see HostedAgentEventKind's doc comment // for the canonical type strings). type Event struct { @@ -75,8 +77,8 @@ type Harness struct { hangAfterEvents bool sessionID string runID string // returned by the next POST .../input call - events []Event // streamed, in order, by the next GET .../stream call - replayEvents []Event // streamed instead of events when GET .../stream carries replay_only=true + events []Event // streamed, in order, by the next GET .../events call + replayEvents []Event // streamed instead of events when GET .../events carries replay_only=true streamErrorStatus int // 0 = serve normally; nonzero = return this HTTP status instead streamErrorRemaining int // >0: decrement per hit, clearing streamErrorStatus at 0; <=0 with status set: permanent streamErrorSkip int // succeed this many opens before applying streamErrorStatus (reconnect-path tests) @@ -116,13 +118,13 @@ func New(t *testing.T, sessionID string) *Harness { mux := http.NewServeMux() mux.HandleFunc("GET /v2/agents/sessions/{id}", h.handleGetSession) - // Two SSE surfaces, matching godo's StreamSession: live reads go to the - // data plane at .../events, replay-only reads to the control plane at - // .../stream?replay_only=true (see QueueReplayHistory). handleStream - // serves both, branching on the replay_only query parameter. + // One SSE surface, matching godo's StreamSession: both the live and the + // replay-only read go to the data plane at .../events, the latter carrying + // replay_only=true (see QueueReplayHistory). handleStream serves both, + // branching on that query parameter. The control plane's .../stream is + // deliberately not registered, so a request landing there fails the test + // instead of quietly passing against a route the client no longer uses. mux.HandleFunc("GET /v2/agents/sessions/{id}/events", h.handleStream) - // Temporarily on control-plane .../stream until OHP /events is on stage2. - 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) @@ -132,7 +134,7 @@ func New(t *testing.T, sessionID string) *Harness { } // QueueRun arranges for the next POST .../input call to return runID, and -// for GET .../stream to then emit events (in order, tagged with runID), +// for GET .../events 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. // @@ -146,7 +148,7 @@ func (h *Harness) QueueRun(runID string, events ...Event) { h.events = events } -// QueueReplayHistory arranges for a GET .../stream call carrying +// QueueReplayHistory arranges for a GET .../events call carrying // replay_only=true to return these events instead of whatever QueueRun set // up, then end — mirrors harness-api's own handleStreamSession, which never // continues a replay_only request into a live tail (see @@ -159,7 +161,7 @@ func (h *Harness) QueueReplayHistory(events ...Event) { h.replayEvents = events } -// SetStreamErrorStatus makes GET .../stream return status immediately +// SetStreamErrorStatus makes GET .../events return status immediately // (instead of opening an SSE stream) for the next `times` calls, then // resume normal behavior (serving whatever's queued via QueueRun) — // simulating a StreamSession failure rather than a mid-stream drop. times @@ -185,12 +187,12 @@ func (h *Harness) SetStreamErrorStatusAfter(skip, status, times int) { h.streamErrorRemaining = times } -// DropConnectionAfterEvents makes the very next GET .../stream call return +// DropConnectionAfterEvents makes the very next GET .../events call return // after sending only the first n queued events (a clean end, no error — // exactly how a genuine mid-stream drop/idle-timeout looks from the // client's side) instead of the whole list, then resets to 0 so any later // connection serves normally. One-shot: simulates a real drop-and-resume for -// a test verifying a reconnect actually resumes (via replay_from) and dedups +// a test verifying a reconnect actually resumes (via Last-Event-ID) and dedups // whatever prefix gets redelivered, rather than only ever seeing a stream // that's already run to completion. func (h *Harness) DropConnectionAfterEvents(n int) { @@ -283,20 +285,20 @@ func (h *Harness) handleStream(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) flusher, canFlush := w.(http.Flusher) - // Optional stream.state control frame (same wire value as the data-plane - // transport). Consumers must skip it — emit it so tests exercise that. - const streamStateKind = "stream.state" + // The data plane opens every stream with a stream.state control frame. It + // belongs to no run, so consumers must skip it rather than mistake it for + // session activity — emit it here so tests exercise that. streamState, err := json.Marshal(eventWire{ TenantID: "15726539", SessionID: sessionID, Timestamp: "2026-01-01T00:00:00Z", - Type: streamStateKind, + Type: string(godo.HostedAgentEventKindStreamState), Data: json.RawMessage(`{"state":"live","cursor":""}`), }) if err != nil { panic(fmt.Sprintf("agentproxytest: stream.state does not marshal to JSON: %v", err)) } - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", streamStateKind, streamState) + fmt.Fprintf(w, "event: %s\ndata: %s\n\n", godo.HostedAgentEventKindStreamState, streamState) if canFlush { flusher.Flush() } diff --git a/internal/agentproxy/codex/facade_test.go b/internal/agentproxy/codex/facade_test.go index f69bfb7a0..5c4f5c16d 100644 --- a/internal/agentproxy/codex/facade_test.go +++ b/internal/agentproxy/codex/facade_test.go @@ -1252,7 +1252,7 @@ func TestFacade_Replay_ThreadResume(t *testing.T) { // touches replay-only history, even when a session has some queued — the // flag must actually gate the behavior, not just always run it. Asserts // both "no notifications" and "replay never started": a silent fetch -// failure (e.g. the harness 404ing .../stream) would satisfy expectNone +// failure (e.g. the harness 404ing .../events) would satisfy expectNone // alone and hide a broken gate. func TestFacade_Replay_Disabled(t *testing.T) { f, h, rec := newTestFacade(t) @@ -1354,7 +1354,7 @@ func TestFacade_Replay_UnaffectedByConcurrentLiveTurnsReset(t *testing.T) { // no-op forever the way a plain sync.Once would have. // // The first attempt must fail for a reason that only hits once the -// control-plane .../stream route is actually registered (injected 500) — +// data-plane .../events route is actually registered (injected 500) — // a missing-route 404 would also unwind without marking replayDone and // make this half of the test pass for the wrong reason. func TestFacade_Replay_RetriesAfterAbortedAttempt(t *testing.T) { diff --git a/vendor/github.com/digitalocean/godo/hosted_agents.go b/vendor/github.com/digitalocean/godo/hosted_agents.go index ec5dbae73..5a316edc0 100644 --- a/vendor/github.com/digitalocean/godo/hosted_agents.go +++ b/vendor/github.com/digitalocean/godo/hosted_agents.go @@ -22,7 +22,7 @@ const ( hostedAgentsSessionsBasePath = "/v2/agents/sessions" hostedAgentSessionByIDPath = hostedAgentsSessionsBasePath + "/%s" - hostedAgentSessionStreamPath = hostedAgentSessionByIDPath + "/stream" + hostedAgentSessionEventsPath = hostedAgentSessionByIDPath + "/events" hostedAgentSessionInputPath = hostedAgentSessionByIDPath + "/input" hostedAgentSessionHITLPath = hostedAgentSessionByIDPath + "/hitl/%s" hostedAgentSessionSandboxExecPath = hostedAgentSessionByIDPath + "/sandbox/exec" @@ -52,6 +52,10 @@ const ( workspaceIsArchiveHeader = "X-Workspace-Is-Archive" workspaceSizeBytesHeader = "X-Workspace-Size-Bytes" + // sseLastEventIDHeader is the standard SSE resume cursor. The data-plane + // events endpoint takes the cursor here rather than as a query parameter. + sseLastEventIDHeader = "Last-Event-ID" + // workspaceDownloadFooter is appended by OHS after a successful download // payload so integrity survives intermediaries that strip HTTP trailers // (e.g. Cloudflare). Format: DOWSSHA1 + 64 lowercase hex + '\n' = 73 bytes. @@ -243,6 +247,39 @@ const ( HostedAgentEventKindRunSandboxReleased HostedAgentEventKind = "run.sandbox_released" HostedAgentEventKindRunCostAccrued HostedAgentEventKind = "run.cost_accrued" HostedAgentEventKindRunLog HostedAgentEventKind = "run.log" + + // HostedAgentEventKindStreamState is a transport control frame, not an agent + // event: it reports the health of the SSE connection itself. Only the + // data-plane events endpoint emits it. It arrives in the same envelope as an + // event, but only SessionID, At and Payload are meaningful — decode Payload + // with HostedAgentStreamState. Renderers should skip it rather than display + // it as session activity. + HostedAgentEventKindStreamState HostedAgentEventKind = "stream.state" +) + +// HostedAgentStreamState is the payload of a HostedAgentEventKindStreamState +// frame: the current health of the SSE connection. +type HostedAgentStreamState struct { + State HostedAgentStreamStateValue `json:"state"` + Cursor string `json:"cursor,omitempty"` +} + +// HostedAgentStreamStateValue enumerates the stream health states. +type HostedAgentStreamStateValue string + +const ( + // HostedAgentStreamStateLive means events are being delivered contiguously. + HostedAgentStreamStateLive HostedAgentStreamStateValue = "live" + // HostedAgentStreamStateCatchingUp means the connection is replaying recent + // history before it joins the live tail. + HostedAgentStreamStateCatchingUp HostedAgentStreamStateValue = "catching_up" + // HostedAgentStreamStateDegraded means delivery continues with reduced + // guarantees (the server fell back to polling). + HostedAgentStreamStateDegraded HostedAgentStreamStateValue = "degraded" + // HostedAgentStreamStateSuperseded means a newer connection from the same + // device took over. The server closes this stream; the client should stop + // rather than reconnect, or the two connections will evict each other. + HostedAgentStreamStateSuperseded HostedAgentStreamStateValue = "superseded" ) // HostedAgentSessionOriginProduct identifies the product workflow that created @@ -330,7 +367,7 @@ type HostedAgentHITLDecision struct { Reason string `json:"reason,omitempty"` } -// HostedAgentEvent is one SSE payload from GET /v2/agents/sessions/{id}/stream. +// HostedAgentEvent is one SSE payload from a session stream (see StreamSession). // // The server serializes the SPI canonical event envelope, whose JSON shape // differs from this struct's field names: the discriminator is `type` (not @@ -419,8 +456,17 @@ type HostedAgentSessionsListResponse struct { } // HostedAgentSessionStreamOptions configures the session SSE stream. +// +// ReplayOnly selects between the two modes StreamSession can open on the events +// endpoint (see StreamSession); Before and Limit page backwards through history +// within the replay-only mode. type HostedAgentSessionStreamOptions struct { + // ReplayFrom is the resume cursor: the id of the last event the caller + // already rendered. On the live stream it is sent as Last-Event-ID; on a + // ReplayOnly read it is sent as the replay_from query parameter. ReplayFrom string + // ReplayOnly reads the stored event history and ends the stream at the last + // stored event instead of holding the connection open for live events. ReplayOnly bool // Before turns the request into a single backward page of durable @@ -766,34 +812,59 @@ func (s *HostedAgentsServiceOp) ResumeSession(ctx context.Context, sessionID str } // StreamSession opens the SSE stream for a session. Callers MUST Close the stream. +// +// Both reads are served by the data plane at GET .../sessions/{id}/events; the +// two modes differ in where the stream starts and whether it ends: +// +// - Live (the default) delivers forward-only from the moment of attach, +// preceded by whatever recent history the server decides to replay, and +// holds the connection open. ReplayFrom is sent as the standard +// Last-Event-ID header. +// - ReplayOnly adds ?replay_only=true: the server writes the session's stored +// event history and then ends the stream, so the read terminates on its own. +// ReplayFrom is sent as the replay_from query parameter, since it is an +// explicit pagination cursor here rather than a resume hint. Before and +// Limit page backwards from an event id within this mode; see +// HostedAgentSessionStreamOptions.Before and HasMore. +// +// Both carry HostedAgentEventKindStreamState control frames (see +// HostedAgentStreamState). A replay-only read reports catching_up and then +// simply ends; it never reaches live. func (s *HostedAgentsServiceOp) StreamSession(ctx context.Context, sessionID string, opt *HostedAgentSessionStreamOptions) (*HostedAgentSessionStream, *Response, error) { if sessionID == "" { return nil, nil, errors.New("hosted agents: session id is required") } - path := fmt.Sprintf(hostedAgentSessionStreamPath, sessionID) + replayOnly := opt != nil && opt.ReplayOnly + cursor := "" + before := "" + limit := 0 if opt != nil { // The server answers this combination with a 400; rejecting it here // spends no round trip to learn the same thing. if opt.Before != "" && !opt.ReplayOnly { return nil, nil, errors.New("hosted agents: before requires replay only") } + cursor = opt.ReplayFrom + before = opt.Before + limit = opt.Limit + } + + path := fmt.Sprintf(hostedAgentSessionEventsPath, sessionID) + if replayOnly { q := url.Values{} - if opt.ReplayFrom != "" { - q.Set("replay_from", opt.ReplayFrom) + q.Set("replay_only", "true") + if cursor != "" { + q.Set("replay_from", cursor) } - if opt.ReplayOnly { - q.Set("replay_only", "true") + if before != "" { + q.Set("before", before) } - if opt.Before != "" { - q.Set("before", opt.Before) - } - if opt.Limit > 0 { - q.Set("limit", strconv.Itoa(opt.Limit)) - } - if encoded := q.Encode(); encoded != "" { - path += "?" + encoded + if limit > 0 { + q.Set("limit", strconv.Itoa(limit)) } + path += "?" + q.Encode() } + req, err := s.client.NewRequest(ctx, http.MethodGet, path, nil) if err != nil { return nil, nil, err @@ -801,6 +872,9 @@ func (s *HostedAgentsServiceOp) StreamSession(ctx context.Context, sessionID str req.Header.Set("Accept", "text/event-stream") req.Header.Set("Cache-Control", "no-cache") req.Header.Set("Connection", "keep-alive") + if !replayOnly && cursor != "" { + req.Header.Set(sseLastEventIDHeader, cursor) + } resp, err := s.client.DoStream(ctx, req) if err != nil {