Skip to content

Commit bc58a6e

Browse files
markturanskyAmbient Code Botclaude
authored
fix(sse): flush StreamRunnerEvents per chunk; add acpctl session send -f (#1226)
## Summary - **StreamRunnerEvents buffering fix**: replaced `io.Copy` with a chunked read loop that calls `http.Flusher.Flush()` after every write, so AG-UI SSE events stream to clients in real time rather than batching until `RUN_FINISHED` closes the runner connection - **`acpctl session send -f`**: after pushing a message, immediately opens the `/events` SSE stream and assembles `TEXT_MESSAGE_CONTENT` deltas into readable text output; closes naturally when `RUN_FINISHED` arrives (synchronous send-and-wait UX) - **`acpctl session send -f --json`**: same flow but emits raw AG-UI JSON events one per line (matches existing `session events` behavior) - Spec CLI table updated with new `send -f` and `send -f --json` rows ## Test plan - [ ] `acpctl session send <id> "hello"` returns `sent (seq=N)` immediately - [ ] `acpctl session send <id> "hello" -f` streams readable text tokens as Claude responds, exits cleanly on `RUN_FINISHED` - [ ] `acpctl session send <id> "hello" -f --json` streams raw AG-UI JSON events, one per line - [ ] `acpctl session events <id>` (unchanged) still works - [ ] Ctrl+C cancels `-f` cleanly 🤖 Generated with [Claude Code](https://claude.ai/code) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit ## Release Notes * **New Features** * Added `-f/--follow` CLI option to stream session events after sending a message. * Added `--json` flag to format streamed event output as JSON. * **Documentation** * Updated command reference for the CLI message send functionality. * Added end-to-end manual test workflow demonstrating session creation, credential setup, and event monitoring. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: Ambient Code Bot <bot@ambient-code.local> Co-authored-by: Claude <noreply@anthropic.com>
1 parent fffe4ba commit bc58a6e

3 files changed

Lines changed: 127 additions & 7 deletions

File tree

‎components/ambient-api-server/plugins/sessions/handler.go‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -328,10 +328,24 @@ func (h sessionHandler) StreamRunnerEvents(w http.ResponseWriter, r *http.Reques
328328
f.Flush()
329329
}
330330

331-
if _, copyErr := io.Copy(w, resp.Body); copyErr != nil {
332-
glog.V(4).Infof("StreamRunnerEvents: stream ended for session %s: %v", id, copyErr)
333-
}
334-
if f, ok := w.(http.Flusher); ok {
335-
f.Flush()
331+
flusher, canFlush := w.(http.Flusher)
332+
buf := make([]byte, 4096)
333+
for {
334+
n, readErr := resp.Body.Read(buf)
335+
if n > 0 {
336+
if _, writeErr := w.Write(buf[:n]); writeErr != nil {
337+
glog.V(4).Infof("StreamRunnerEvents: write error for session %s: %v", id, writeErr)
338+
return
339+
}
340+
if canFlush {
341+
flusher.Flush()
342+
}
343+
}
344+
if readErr != nil {
345+
if readErr != io.EOF {
346+
glog.V(4).Infof("StreamRunnerEvents: read error for session %s: %v", id, readErr)
347+
}
348+
return
349+
}
336350
}
337351
}

‎components/ambient-cli/cmd/acpctl/session/send.go‎

Lines changed: 66 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,44 @@
11
package session
22

33
import (
4+
"bufio"
45
"context"
6+
"encoding/json"
57
"fmt"
8+
"os"
9+
"os/signal"
10+
"strings"
611

712
"github.com/ambient-code/platform/components/ambient-cli/pkg/config"
813
"github.com/ambient-code/platform/components/ambient-cli/pkg/connection"
914
"github.com/spf13/cobra"
1015
)
1116

17+
var sendFollow bool
18+
var sendFollowJSON bool
19+
1220
var sendCmd = &cobra.Command{
1321
Use: "send <session-id> <message>",
1422
Short: "Send a message to a session",
1523
Long: `Send a message to a session.
1624
25+
Without -f, prints the message sequence number and returns immediately.
26+
With -f, streams the assistant response as readable text until RUN_FINISHED.
27+
With -f --json, streams raw AG-UI JSON events instead of assembled text.
28+
1729
Examples:
1830
acpctl session send <id> "Hello! What's today's date?"
19-
acpctl session send <id> "Run the tests"`,
31+
acpctl session send <id> "Run the tests" -f
32+
acpctl session send <id> "Run the tests" -f --json`,
2033
Args: cobra.ExactArgs(2),
2134
RunE: runSend,
2235
}
2336

37+
func init() {
38+
sendCmd.Flags().BoolVarP(&sendFollow, "follow", "f", false, "stream response after sending until RUN_FINISHED")
39+
sendCmd.Flags().BoolVar(&sendFollowJSON, "json", false, "with -f: emit raw AG-UI JSON events instead of text")
40+
}
41+
2442
func runSend(cmd *cobra.Command, args []string) error {
2543
sessionID := args[0]
2644
payload := args[1]
@@ -44,5 +62,52 @@ func runSend(cmd *cobra.Command, args []string) error {
4462
}
4563

4664
fmt.Fprintf(cmd.OutOrStdout(), "sent (seq=%d)\n", msg.Seq)
65+
66+
if !sendFollow {
67+
return nil
68+
}
69+
70+
streamCtx, streamCancel := signal.NotifyContext(cmd.Context(), os.Interrupt)
71+
defer streamCancel()
72+
73+
stream, err := client.Sessions().StreamEvents(streamCtx, sessionID)
74+
if err != nil {
75+
return fmt.Errorf("stream events: %w", err)
76+
}
77+
defer stream.Close()
78+
79+
out := cmd.OutOrStdout()
80+
scanner := bufio.NewScanner(stream)
81+
for scanner.Scan() {
82+
line := scanner.Text()
83+
if !strings.HasPrefix(line, "data: ") {
84+
continue
85+
}
86+
data := line[6:]
87+
88+
if sendFollowJSON {
89+
fmt.Fprintln(out, data)
90+
continue
91+
}
92+
93+
var evt struct {
94+
Type string `json:"type"`
95+
Delta string `json:"delta"`
96+
}
97+
if err := json.Unmarshal([]byte(data), &evt); err != nil {
98+
continue
99+
}
100+
if evt.Type == "TEXT_MESSAGE_CONTENT" && evt.Delta != "" {
101+
fmt.Fprint(out, evt.Delta)
102+
}
103+
}
104+
105+
if !sendFollowJSON {
106+
fmt.Fprintln(out)
107+
}
108+
109+
if scanErr := scanner.Err(); scanErr != nil {
110+
return fmt.Errorf("stream error: %w", scanErr)
111+
}
47112
return nil
48113
}

‎docs/internal/design/ambient-model.spec.md‎

Lines changed: 42 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -314,7 +314,9 @@ The `acpctl` CLI mirrors the API 1-for-1. Every REST operation has a correspondi
314314
| `GET /sessions/{id}` | `acpctl describe session <id>` | ✅ implemented |
315315
| `DELETE /sessions/{id}` | `acpctl delete session <id>` | ✅ implemented |
316316
| `GET /sessions/{id}/messages` | `acpctl session messages <id>` | ✅ implemented |
317-
| `POST /sessions/{id}/messages` | `acpctl session send <id> --body <text>` | ✅ implemented |
317+
| `POST /sessions/{id}/messages` | `acpctl session send <id> <message>` | ✅ implemented |
318+
| `POST /sessions/{id}/messages` + `GET /sessions/{id}/events` | `acpctl session send <id> <message> -f` | ✅ implemented |
319+
| `POST /sessions/{id}/messages` + `GET /sessions/{id}/events` | `acpctl session send <id> <message> -f --json` | ✅ implemented |
318320
| `GET /sessions/{id}/events` | `acpctl session events <id>` | ✅ implemented |
319321

320322
#### Credentials
@@ -959,3 +961,42 @@ All Kinds with `labels`/`annotations` store them as JSON strings in the DB (`*st
959961
| Project/Agent/Session label subcommands | 🔲 no `acpctl label`/`acpctl annotate` | add typed label helpers to SDK first, then CLI |
960962
| `GET /roles`, `GET /role_bindings` | 🔲 list/get not exposed | add to `get` command resource switch |
961963
| `DELETE /role_bindings/{id}` | 🔲 not exposed | add to `delete` command resource switch |
964+
965+
966+
Manual Test
967+
968+
# 1. Project
969+
acpctl create project --name test-cred-1 --description "cred test"
970+
acpctl project test-cred-1
971+
972+
# 2. Agent
973+
acpctl agent create --project-id test-cred-1 --name github-agent \
974+
--prompt "You are a GitHub automation agent."
975+
976+
AGENT_ID=$(acpctl agent list --project-id test-cred-1 -o json | python3 -c "import sys,json; print(json.load(sys.stdin)['items'][0]['id'])")
977+
echo "AGENT_ID=$AGENT_ID"
978+
979+
# 3. Credential (apply from file — only working path)
980+
printf 'kind: Credential\nname: github-pat-test\nprovider: github\ntoken: %s\ndescription: test\n' \
981+
"$(cat ~/projects/secrets/github.ambient-pat.token)" > /tmp/cred.yaml
982+
acpctl apply -f /tmp/cred.yaml && rm /tmp/cred.yaml
983+
984+
CRED_ID=$(acpctl get credentials -o json | python3 -c "import sys,json; print(next(i['id'] for i in json.load(sys.stdin)['items'] if i['name']=='github-pat-test'))")
985+
echo "CRED_ID=$CRED_ID"
986+
987+
# 4. Role binding
988+
ROLE_ID=$(acpctl get roles -o json | python3 -c "import sys,json; print(next(i['id'] for i in json.load(sys.stdin)['items'] if i['name']=='credential:token-reader'))")
989+
MY_USER=$(acpctl whoami | awk '/^User:/{print $2}')
990+
echo "ROLE_ID=$ROLE_ID MY_USER=$MY_USER"
991+
992+
acpctl create role-binding --user-id "$MY_USER" --role-id "$ROLE_ID" \
993+
--scope agent --scope-id "$AGENT_ID"
994+
995+
# 5. Start session
996+
SESSION_ID=$(acpctl agent start github-agent --project-id test-cred-1 \
997+
--prompt "Fetch credential $CRED_ID token and confirm you received it." \
998+
-o json | python3 -c "import sys,json; print(json.load(sys.stdin)['id'])")
999+
echo "SESSION_ID=$SESSION_ID"
1000+
1001+
# 6. Watch events
1002+
acpctl session events "$SESSION_ID"

0 commit comments

Comments
 (0)