diff --git a/go.mod b/go.mod
index f466b5e..94f025c 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,7 @@ module github.com/flashcatcloud/flashduty-cli
 go 1.26.0
 
 require (
-	github.com/flashcatcloud/go-flashduty v0.15.21-0.20261027024437-bab720b88e34
+	github.com/flashcatcloud/go-flashduty v0.15.21
 	github.com/mattn/go-runewidth v0.0.30
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/pflag v1.0.10
diff --git a/go.sum b/go.sum
index 74515f4..1add092 100644
--- a/go.sum
+++ b/go.sum
@@ -1,8 +1,8 @@
 github.com/clipperhouse/uax29/v2 v2.2.0 h1:ChwIKnQN3kcZteTXMgb1wztSgaU+ZemkgWdohwgs8tY=
 github.com/clipperhouse/uax29/v2 v2.2.0/go.mod h1:EFJ2TJMRUaplDxHKj1qAEhCtQPW2tJSwu5BF98AuoVM=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
-github.com/flashcatcloud/go-flashduty v0.15.21-0.20261027024437-bab720b88e34 h1:Cje7u+AOH3qL/PyxV97Eh2tVTFnDSG3tB21Q2HZwT3o=
-github.com/flashcatcloud/go-flashduty v0.15.21-0.20261027024437-bab720b88e34/go.mod h1:YpHiTYXR5NXBI/rGRZfUy537XMkhdCkwA8NW1QoRHwk=
+github.com/flashcatcloud/go-flashduty v0.15.21 h1:GLXsWw55BKm0xodjnYrA4Q32G5nmCm6PW+qkVLZPT8M=
+github.com/flashcatcloud/go-flashduty v0.15.21/go.mod h1:YpHiTYXR5NXBI/rGRZfUy537XMkhdCkwA8NW1QoRHwk=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/mattn/go-runewidth v0.0.30 h1:+KUuiDA4fF0R1p5FeueHefjDm+GIM+kWfFnDjybOPgk=
diff --git a/internal/cli/root.go b/internal/cli/root.go
index 7e72ea0..8627444 100644
--- a/internal/cli/root.go
+++ b/internal/cli/root.go
@@ -171,6 +171,9 @@ func init() {
 	// attach their curated leaves to the generated path groups.
 	attachEnrichmentMappingDataUpload(rootCmd)
 	attachSafariSkillUpload(rootCmd)
+
+	// Curated AI SRE chat commands (message/subscribe are not generated).
+	attachSafariChat(rootCmd)
 }
 
 // Execute runs the root command as name, the command word the binary was
diff --git a/internal/cli/safari_send.go b/internal/cli/safari_send.go
new file mode 100644
index 0000000..d412590
--- /dev/null
+++ b/internal/cli/safari_send.go
@@ -0,0 +1,217 @@
+package cli
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"unicode/utf8"
+
+	"github.com/flashcatcloud/go-flashduty"
+	"github.com/spf13/cobra"
+	"golang.org/x/term"
+)
+
+// Curated chat commands for the AI SRE agent, attached to the generated
+// `safari` group: the message, subscribe and attachment endpoints are not part
+// of the generated surface.
+
+const (
+	safariAppName       = "ai-sre"
+	safariEntryKind     = "api"
+	safariSurface       = "cli"
+	safariReplyOnlyMode = "reply_only"
+	// maxMessageRunes is the server limit on a message's text.
+	maxMessageRunes = 32000
+	// maxAttachmentBytes is the server limit on one attachment. File types are
+	// not checked here: unknown extensions are accepted by the server only when
+	// the content is UTF-8 text, and its error is passed through.
+	maxAttachmentBytes = 50 << 20
+)
+
+func attachSafariChat(root *cobra.Command) {
+	g := genGroup(root, "safari", "AI SRE API")
+	genAddLeaf(g, newSafariSendCmd())
+	genAddLeaf(g, newSafariWatchCmd())
+}
+
+// readSendMessage resolves the message from the positional argument,
+// --message-file (a path, or - for stdin) or piped stdin, accepting exactly one
+// source.
+func readSendMessage(args []string, messageFile string) (string, error) {
+	sources := 0
+	if len(args) > 0 {
+		sources++
+	}
+	if messageFile != "" {
+		sources++
+	}
+	if sources > 1 {
+		return "", fmt.Errorf("give the message as an argument or --message-file, not both")
+	}
+	switch {
+	case len(args) > 0:
+		return args[0], nil
+	case messageFile == "-":
+		return readAllStdin("--message-file")
+	case messageFile != "":
+		b, err := os.ReadFile(messageFile)
+		if err != nil {
+			return "", fmt.Errorf("cannot read --message-file: %w", err)
+		}
+		return string(b), nil
+	case !term.IsTerminal(int(os.Stdin.Fd())):
+		return readAllStdin("stdin")
+	}
+	return "", nil
+}
+
+func readAllStdin(flag string) (string, error) {
+	b, err := readStdin(flag)
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}
+
+func newSafariSendCmd() *cobra.Command {
+	var (
+		sessionID, messageFile, envKind, envID string
+		files                                  []string
+		teamID                                 int64
+	)
+
+	cmd := &cobra.Command{
+		Use:   "send [message]",
+		Short: "Send a message to an AI SRE session",
+		Long: `Send one message to an AI SRE session and print the acknowledgement as a JSON
+object: {"session_id","invocation_id","root_invocation_id","delivery_mode"}.
+
+The message is the positional argument, --message-file (a path, or - for
+stdin), or piped stdin. Without --session-id a new session is started;
+--team-id binds it to a team (0 = account scope; the caller must belong to the
+team) and --env-kind/--env-id pick the execution environment. Each --file is
+uploaded as an attachment to the session and attached to the message.
+
+The message is sent on the CLI channel (reply-only delivery): the agent answers
+through its reply tool and the answers are read with 'safari watch'. If the
+server does not accept that channel, the command fails instead of falling back.
+
+Follow the answer with:
+  flashduty safari watch --session-id <session_id> --root-invocation-id <root_invocation_id>
+
+API: POST /safari/session/message, POST /safari/attachment/upload`,
+		Example: `  flashduty safari send "why is checkout latency up?"
+  flashduty safari send --session-id sess_01 --file trace.json "does this match?"
+  cat question.md | flashduty safari send --team-id 42 --env-kind byoc`,
+		Args: cobra.MaximumNArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return runCommand(cmd, args, func(ctx *RunContext) error {
+				if envID != "" && envKind == "" {
+					return fmt.Errorf("--env-id requires --env-kind")
+				}
+				text, err := readSendMessage(args, messageFile)
+				if err != nil {
+					return err
+				}
+				text = strings.TrimSpace(text)
+				if text == "" && len(files) == 0 {
+					return fmt.Errorf("a message (argument, --message-file or stdin) or at least one --file is required")
+				}
+				if n := utf8.RuneCountInString(text); n > maxMessageRunes {
+					return fmt.Errorf("message is %d characters; the limit is %d (attach longer content with --file)", n, maxMessageRunes)
+				}
+
+				cctx := cmdContext(ctx.Cmd)
+				req := &flashduty.SessionMessageRequest{
+					SessionID:       sessionID,
+					Query:           text,
+					Surface:         safariSurface,
+					DeliveryMode:    safariReplyOnlyMode,
+					TeamID:          teamID,
+					EnvironmentKind: envKind,
+					EnvironmentID:   envID,
+				}
+
+				for _, path := range files {
+					// Fail before a session is created for a file that cannot be read.
+					if path == "-" {
+						continue
+					}
+					info, err := os.Stat(path)
+					if err != nil {
+						return fmt.Errorf("--file %s: %w", path, err)
+					}
+					if info.Size() > maxAttachmentBytes {
+						return fmt.Errorf("--file %s is %d MB; the attachment limit is %d MB", path, info.Size()>>20, maxAttachmentBytes>>20)
+					}
+				}
+
+				if len(files) > 0 {
+					// Attachments are stored under an existing session, so a new
+					// conversation is created before its files go up.
+					if req.SessionID == "" {
+						created, _, err := ctx.Client.Sessions.WriteCreate(cctx, &flashduty.SessionCreateRequest{
+							AppName: safariAppName, EntryKind: safariEntryKind, TeamID: teamID,
+						})
+						if err != nil {
+							return err
+						}
+						req.SessionID = created.SessionID
+					}
+					for _, path := range files {
+						part, err := uploadSessionFile(ctx, req.SessionID, path)
+						if err != nil {
+							return fmt.Errorf("--file %s: %w", path, err)
+						}
+						req.Parts = append(req.Parts, part)
+					}
+				}
+				if len(req.Parts) > 0 && text != "" {
+					// The server ignores query when parts are present, so the
+					// text rides as the leading part (as in the web composer).
+					req.Parts = append([]flashduty.SessionMessagePart{{Type: "text", Text: text}}, req.Parts...)
+					req.Query = ""
+				}
+				if req.SessionID == "" {
+					req.AppName = safariAppName
+					req.EntryKind = safariEntryKind
+				}
+
+				ack, _, err := ctx.Client.Sessions.WriteMessage(cctx, req)
+				if err != nil {
+					return err
+				}
+				if ack.DeliveryMode != safariReplyOnlyMode {
+					return fmt.Errorf("the server did not accept the CLI reply channel (delivery_mode=%q, want %q); the message was queued as session_id=%s invocation_id=%s but its answers will not reach 'safari watch'",
+						ack.DeliveryMode, safariReplyOnlyMode, ack.SessionID, ack.InvocationID)
+				}
+				return json.NewEncoder(ctx.Writer).Encode(ack)
+			})
+		},
+	}
+	cmd.Flags().StringVar(&sessionID, "session-id", "", "Continue this session (default: start a new one)")
+	cmd.Flags().StringVar(&messageFile, "message-file", "", "Read the message from this file, or - for stdin")
+	cmd.Flags().StringArrayVar(&files, "file", nil, "Attach a file (repeatable); uploaded as a session attachment")
+	cmd.Flags().Int64Var(&teamID, "team-id", 0, "Team scope for a new session (0 = account scope)")
+	cmd.Flags().StringVar(&envKind, "env-kind", "", "Execution environment kind: cloud or byoc")
+	registerEnumFlag(cmd, "env-kind", "cloud", "byoc")
+	cmd.Flags().StringVar(&envID, "env-id", "", "Specific environment of --env-kind")
+	return cmd
+}
+
+// uploadSessionFile uploads one --file and returns its message part.
+func uploadSessionFile(ctx *RunContext, sessionID, path string) (flashduty.SessionMessagePart, error) {
+	f, name, err := openUploadFile(path, "stdin")
+	if err != nil {
+		return flashduty.SessionMessagePart{}, err
+	}
+	defer func() { _ = f.Close() }()
+	att, _, err := ctx.Client.Sessions.UploadAttachment(cmdContext(ctx.Cmd), &flashduty.SessionAttachmentUploadRequest{
+		SessionID: sessionID, Filename: name, File: f,
+	})
+	if err != nil {
+		return flashduty.SessionMessagePart{}, err
+	}
+	return att.FilePart(), nil
+}
diff --git a/internal/cli/safari_send_test.go b/internal/cli/safari_send_test.go
new file mode 100644
index 0000000..8d1f5a4
--- /dev/null
+++ b/internal/cli/safari_send_test.go
@@ -0,0 +1,238 @@
+package cli
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/flashcatcloud/go-flashduty"
+)
+
+// sendServer records the calls `safari send` makes and answers each endpoint.
+type sendServer struct {
+	mu           sync.Mutex
+	paths        []string
+	message      map[string]any
+	uploadSess   string
+	uploadName   string
+	uploadBody   string
+	ackMode      string
+	createCalled map[string]any
+}
+
+func newSendServer(t *testing.T, ackMode string) *sendServer {
+	t.Helper()
+	s := &sendServer{ackMode: ackMode}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		s.mu.Lock()
+		defer s.mu.Unlock()
+		s.paths = append(s.paths, r.URL.Path)
+		reply := func(data any) {
+			_ = json.NewEncoder(w).Encode(map[string]any{"request_id": "r", "data": data})
+		}
+		switch r.URL.Path {
+		case "/safari/session/create":
+			_ = json.NewDecoder(r.Body).Decode(&s.createCalled)
+			reply(map[string]any{"session_id": "sess-new", "app_name": "ai-sre"})
+		case "/safari/attachment/upload":
+			if err := r.ParseMultipartForm(1 << 20); err != nil {
+				t.Errorf("multipart: %v", err)
+			}
+			s.uploadSess = r.FormValue("session_id")
+			f, hdr, _ := r.FormFile("file")
+			b, _ := io.ReadAll(f)
+			s.uploadName, s.uploadBody = hdr.Filename, string(b)
+			reply(map[string]any{"object_key": "attachments/a1/" + hdr.Filename, "mime": "text/plain", "filename": hdr.Filename, "size": len(b), "attachment_id": "a1"})
+		case "/safari/session/message":
+			_ = json.NewDecoder(r.Body).Decode(&s.message)
+			sid, _ := s.message["session_id"].(string)
+			if sid == "" {
+				sid = "sess-auto"
+			}
+			w.WriteHeader(http.StatusAccepted)
+			reply(map[string]any{"session_id": sid, "invocation_id": "inv-1", "root_invocation_id": "inv-1", "delivery_mode": s.ackMode})
+		default:
+			t.Errorf("unexpected path %s", r.URL.Path)
+		}
+	}))
+	t.Cleanup(srv.Close)
+	newClientFn = func() (*flashduty.Client, error) {
+		return flashduty.NewClient("test-key", flashduty.WithBaseURL(srv.URL))
+	}
+	return s
+}
+
+func TestSafariSendNewSessionAck(t *testing.T) {
+	saveAndResetGlobals(t)
+	s := newSendServer(t, "reply_only")
+
+	out, err := execCommand("safari", "send", "why is latency up?", "--team-id", "42", "--env-kind", "byoc", "--env-id", "env-9")
+	if err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	var ack map[string]string
+	if err := json.Unmarshal([]byte(strings.TrimSpace(out)), &ack); err != nil {
+		t.Fatalf("stdout is not one JSON object: %q", out)
+	}
+	if ack["session_id"] != "sess-auto" || ack["invocation_id"] != "inv-1" || ack["root_invocation_id"] != "inv-1" || ack["delivery_mode"] != "reply_only" {
+		t.Errorf("ack = %v", ack)
+	}
+	if len(s.paths) != 1 || s.paths[0] != "/safari/session/message" {
+		t.Errorf("calls = %v, want only the message call", s.paths)
+	}
+	m := s.message
+	if m["query"] != "why is latency up?" || m["surface"] != "cli" || m["delivery_mode"] != "reply_only" ||
+		m["app_name"] != "ai-sre" || m["entry_kind"] != "api" || m["team_id"] != float64(42) ||
+		m["environment_kind"] != "byoc" || m["environment_id"] != "env-9" {
+		t.Errorf("message body = %v", m)
+	}
+	if _, ok := m["session_id"]; ok {
+		t.Errorf("new session must omit session_id: %v", m)
+	}
+}
+
+func TestSafariSendExistingSessionOmitsNewSessionFields(t *testing.T) {
+	saveAndResetGlobals(t)
+	s := newSendServer(t, "reply_only")
+	if _, err := execCommand("safari", "send", "--session-id", "sess-7", "follow up"); err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	m := s.message
+	if m["session_id"] != "sess-7" {
+		t.Errorf("session_id = %v", m["session_id"])
+	}
+	if _, ok := m["app_name"]; ok {
+		t.Errorf("app_name must be omitted for an existing session: %v", m)
+	}
+	if _, ok := m["entry_kind"]; ok {
+		t.Errorf("entry_kind must be omitted for an existing session: %v", m)
+	}
+}
+
+func TestSafariSendFilesCreateSessionThenUploadThenMessage(t *testing.T) {
+	saveAndResetGlobals(t)
+	s := newSendServer(t, "reply_only")
+	dir := t.TempDir()
+	path := filepath.Join(dir, "trace.txt")
+	if err := os.WriteFile(path, []byte("stack trace"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	out, err := execCommand("safari", "send", "--file", path, "--team-id", "5", "see attached")
+	if err != nil {
+		t.Fatalf("send: %v", err)
+	}
+	if strings.Join(s.paths, ",") != "/safari/session/create,/safari/attachment/upload,/safari/session/message" {
+		t.Errorf("call order = %v", s.paths)
+	}
+	if s.createCalled["app_name"] != "ai-sre" || s.createCalled["entry_kind"] != "api" || s.createCalled["team_id"] != float64(5) {
+		t.Errorf("create body = %v", s.createCalled)
+	}
+	if s.uploadSess != "sess-new" || s.uploadName != "trace.txt" || s.uploadBody != "stack trace" {
+		t.Errorf("upload = %q %q %q", s.uploadSess, s.uploadName, s.uploadBody)
+	}
+	m := s.message
+	if m["session_id"] != "sess-new" {
+		t.Errorf("message session_id = %v", m["session_id"])
+	}
+	if _, ok := m["app_name"]; ok {
+		t.Errorf("app_name must be omitted once the session exists: %v", m)
+	}
+	parts, _ := m["parts"].([]any)
+	if len(parts) != 2 {
+		t.Fatalf("parts = %v", m["parts"])
+	}
+	file := parts[1].(map[string]any)["file"].(map[string]any)
+	if parts[1].(map[string]any)["type"] != "file" || file["object_key"] != "attachments/a1/trace.txt" || file["mime"] != "text/plain" || file["filename"] != "trace.txt" {
+		t.Errorf("file part = %v", parts[1])
+	}
+	if _, ok := m["query"]; ok {
+		t.Errorf("query must be omitted when parts are present: %v", m["query"])
+	}
+	lead := parts[0].(map[string]any)
+	if lead["type"] != "text" || lead["text"] != "see attached" {
+		t.Errorf("leading part = %v, want the message text", lead)
+	}
+	if !strings.Contains(out, `"session_id":"sess-new"`) {
+		t.Errorf("stdout = %q", out)
+	}
+}
+
+func TestSafariSendFailsWhenReplyChannelNotAccepted(t *testing.T) {
+	saveAndResetGlobals(t)
+	newSendServer(t, "")
+	out, err := execCommand("safari", "send", "hello")
+	if err == nil {
+		t.Fatal("expected an error when the ack does not grant reply_only")
+	}
+	for _, want := range []string{"reply channel", "sess-auto", "inv-1"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Errorf("error %q lacks %q", err, want)
+		}
+	}
+	if strings.TrimSpace(out) != "" {
+		t.Errorf("stdout must stay empty on failure, got %q", out)
+	}
+}
+
+func TestSafariSendMessageSources(t *testing.T) {
+	saveAndResetGlobals(t)
+	s := newSendServer(t, "reply_only")
+	dir := t.TempDir()
+	path := filepath.Join(dir, "q.md")
+	if err := os.WriteFile(path, []byte("  from file \n"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := execCommand("safari", "send", "--message-file", path); err != nil {
+		t.Fatalf("message-file: %v", err)
+	}
+	if s.message["query"] != "from file" {
+		t.Errorf("query = %q", s.message["query"])
+	}
+
+	stdinReader = strings.NewReader("from stdin")
+	if _, err := execCommand("safari", "send", "--message-file", "-"); err != nil {
+		t.Fatalf("message-file -: %v", err)
+	}
+	if s.message["query"] != "from stdin" {
+		t.Errorf("query = %q", s.message["query"])
+	}
+}
+
+func TestSafariSendValidation(t *testing.T) {
+	saveAndResetGlobals(t)
+	s := newSendServer(t, "reply_only")
+	dir := t.TempDir()
+	path := filepath.Join(dir, "q.md")
+	_ = os.WriteFile(path, []byte("x"), 0o600)
+
+	cases := []struct {
+		name string
+		args []string
+		want string
+	}{
+		{"both sources", []string{"safari", "send", "a", "--message-file", path}, "not both"},
+		{"env-id without kind", []string{"safari", "send", "a", "--env-id", "e1"}, "--env-id requires --env-kind"},
+		{"too long", []string{"safari", "send", strings.Repeat("字", 32001)}, "limit is 32000"},
+		{"missing file", []string{"safari", "send", "a", "--file", filepath.Join(dir, "nope")}, "--file"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			_, err := execCommand(c.args...)
+			if err == nil || !strings.Contains(err.Error(), c.want) {
+				t.Fatalf("err = %v, want it to contain %q", err, c.want)
+			}
+		})
+	}
+	for _, p := range s.paths {
+		if p == "/safari/session/message" {
+			t.Errorf("a rejected command must not reach the message endpoint: %v", s.paths)
+		}
+	}
+}
diff --git a/internal/cli/safari_watch.go b/internal/cli/safari_watch.go
new file mode 100644
index 0000000..ede6b5c
--- /dev/null
+++ b/internal/cli/safari_watch.go
@@ -0,0 +1,500 @@
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/flashcatcloud/go-flashduty"
+	"github.com/spf13/cobra"
+)
+
+// Frame types on the session event stream that matter to a watcher.
+const (
+	frameTypeSettled     = "session_turn_settled"
+	frameTypeEnvFailed   = "environment.failed"
+	frameTypeError       = "error"
+	replyToolName        = "reply"
+	replyPurposeResult   = "result"
+	replyPurposeProgress = "progress"
+)
+
+// watchEvent is the one user-facing line `safari watch` prints. Only the fields
+// of its Type are set.
+type watchEvent struct {
+	Type             string `json:"type"`
+	Purpose          string `json:"purpose,omitempty"`
+	Message          string `json:"message,omitempty"`
+	InvocationID     string `json:"invocation_id,omitempty"`
+	RootInvocationID string `json:"root_invocation_id,omitempty"`
+	EventID          string `json:"event_id,omitempty"`
+	CreatedAt        int64  `json:"created_at,omitempty"`
+	Reason           string `json:"reason,omitempty"`
+	StandingTasks    int    `json:"standing_tasks,omitempty"`
+	Code             string `json:"code,omitempty"`
+}
+
+// settledPayload is the payload of a session_turn_settled frame (the subset a
+// watcher needs).
+type settledPayload struct {
+	RootInvocationID string            `json:"root_invocation_id"`
+	Reason           string            `json:"reason"`
+	StandingTasks    []json.RawMessage `json:"standing_tasks"`
+}
+
+// frameOutcome is what one frame means for the watch loop.
+type frameOutcome struct {
+	Events []watchEvent
+	// Settled is true for a session_turn_settled frame of the watched root with
+	// no standing tasks: the root has no more automatic work, so the request
+	// is a candidate for "fully done" (confirmed against is_running).
+	Settled bool
+	// Fatal is true for a frame after which the session will produce nothing
+	// more (environment provisioning failed before the turn started).
+	Fatal bool
+}
+
+// watchState turns raw stream frames into events. It owns the root filter, the
+// dedupe set and the resume cursor, none of which depend on the transport.
+type watchState struct {
+	rootID string
+	seen   map[string]struct{}
+
+	cursorAt int64
+	cursorID string
+}
+
+func newWatchState(rootID string, since int64) *watchState {
+	return &watchState{rootID: rootID, seen: map[string]struct{}{}, cursorAt: since}
+}
+
+// frameRoot is the root invocation a frame belongs to. A frame that carries no
+// root id belongs to the root whose own invocation id it carries.
+func frameRoot(f *flashduty.SessionEventFrame) string {
+	if f.RootInvocationID != "" {
+		return f.RootInvocationID
+	}
+	return f.InvocationID
+}
+
+// handle processes one frame. Persisted frames always advance the resume
+// cursor, even when filtered out, so a reconnect never replays them.
+func (w *watchState) handle(f *flashduty.SessionEventFrame) frameOutcome {
+	out := w.classify(f)
+	for i := range out.Events {
+		out.Events[i].CreatedAt = f.CreatedAt
+	}
+	return out
+}
+
+func (w *watchState) classify(f *flashduty.SessionEventFrame) frameOutcome {
+	if f.EventID != "" {
+		// The server streams persisted frames in its own cursor order (the
+		// database collation of (created_at, event_id)), so the last one
+		// received is the cursor; comparing ids here would use a different
+		// order.
+		w.cursorAt, w.cursorID = f.CreatedAt, f.EventID
+		if _, dup := w.seen[f.EventID]; dup {
+			return frameOutcome{}
+		}
+		w.seen[f.EventID] = struct{}{}
+	}
+	if w.rootID != "" && frameRoot(f) != w.rootID {
+		return frameOutcome{}
+	}
+
+	switch f.Type {
+	case frameTypeSettled:
+		var p settledPayload
+		_ = json.Unmarshal(f.Payload, &p)
+		if w.rootID != "" && p.RootInvocationID != "" && p.RootInvocationID != w.rootID {
+			return frameOutcome{}
+		}
+		root := p.RootInvocationID
+		if root == "" {
+			root = frameRoot(f)
+		}
+		ev := watchEvent{Type: "settled", RootInvocationID: root, Reason: p.Reason, EventID: f.EventID, StandingTasks: len(p.StandingTasks)}
+		return frameOutcome{Events: []watchEvent{ev}, Settled: len(p.StandingTasks) == 0}
+	case frameTypeEnvFailed:
+		var p struct {
+			ErrorMessage string `json:"error_message"`
+		}
+		_ = json.Unmarshal(f.Payload, &p)
+		msg := strings.TrimSpace(p.ErrorMessage)
+		if msg == "" {
+			msg = "environment provisioning failed"
+		}
+		return frameOutcome{Events: []watchEvent{{Type: "error", Message: msg, Code: "environment_failed"}}, Fatal: true}
+	case frameTypeError:
+		return frameOutcome{Events: []watchEvent{{Type: "error", Message: errorFrameMessage(f), Code: f.ErrorCode}}}
+	case "assistant", "model":
+		if replies := repliesFromFrame(f); len(replies) > 0 {
+			return frameOutcome{Events: replies}
+		}
+		if f.IsError {
+			return frameOutcome{Events: []watchEvent{{Type: "error", Message: errorFrameMessage(f), Code: f.ErrorCode, EventID: f.EventID}}}
+		}
+	}
+	return frameOutcome{}
+}
+
+// adkContent is the slice of an ADK content payload the reply tool result
+// lives in.
+type adkContent struct {
+	Parts []struct {
+		Text             string `json:"text"`
+		FunctionResponse *struct {
+			Name     string `json:"name"`
+			Response struct {
+				Submitted bool   `json:"submitted"`
+				Delivered bool   `json:"delivered"`
+				Message   string `json:"message"`
+				Purpose   string `json:"purpose"`
+			} `json:"response"`
+		} `json:"function_response"`
+	} `json:"parts"`
+}
+
+// repliesFromFrame extracts the messages the agent handed over through the
+// reply tool. Under the reply-only contract these tool results are the only
+// user-facing output; assistant prose, thinking and other tool calls are
+// private working text and are never returned. A result counts only when the
+// tool accepted it (submitted, or the legacy delivered flag) and the purpose is
+// one the protocol knows: an unknown purpose fails closed.
+func repliesFromFrame(f *flashduty.SessionEventFrame) []watchEvent {
+	if len(f.Payload) == 0 {
+		return nil
+	}
+	var c adkContent
+	if err := json.Unmarshal(f.Payload, &c); err != nil {
+		return nil
+	}
+	var out []watchEvent
+	for _, part := range c.Parts {
+		r := part.FunctionResponse
+		if r == nil || strings.TrimSpace(r.Name) != replyToolName {
+			continue
+		}
+		if !r.Response.Submitted && !r.Response.Delivered {
+			continue
+		}
+		purpose := strings.TrimSpace(r.Response.Purpose)
+		switch purpose {
+		case "":
+			purpose = replyPurposeResult
+		case replyPurposeResult, replyPurposeProgress:
+		default:
+			continue
+		}
+		msg := strings.TrimSpace(r.Response.Message)
+		if msg == "" {
+			continue
+		}
+		out = append(out, watchEvent{Type: "reply", Purpose: purpose, Message: msg, InvocationID: f.InvocationID, EventID: f.EventID})
+	}
+	return out
+}
+
+// errorFrameMessage pulls a human-readable message out of an error frame:
+// a message-like JSON field, else the text parts, else the error code.
+func errorFrameMessage(f *flashduty.SessionEventFrame) string {
+	var obj map[string]any
+	if json.Unmarshal(f.Payload, &obj) == nil {
+		for _, k := range []string{"message", "error", "detail", "text"} {
+			if s, ok := obj[k].(string); ok && strings.TrimSpace(s) != "" {
+				return strings.TrimSpace(s)
+			}
+		}
+	}
+	var c adkContent
+	if json.Unmarshal(f.Payload, &c) == nil {
+		var texts []string
+		for _, p := range c.Parts {
+			if t := strings.TrimSpace(p.Text); t != "" {
+				texts = append(texts, t)
+			}
+		}
+		if len(texts) > 0 {
+			return strings.Join(texts, "\n")
+		}
+	}
+	if f.ErrorCode != "" {
+		return f.ErrorCode
+	}
+	return "session error"
+}
+
+// watchRunner drives one watch: subscribe, map frames to events, reconnect on
+// stream loss, and decide when the request is fully done.
+type watchRunner struct {
+	client *flashduty.Client
+	state  *watchState
+	sessID string
+	out    io.Writer // stdout: events only
+	errw   io.Writer // stderr: diagnostics
+
+	maxFailures    int
+	backoff        func(attempt int) time.Duration
+	confirmRetries int
+	confirmGap     time.Duration
+	// healthyAfter is how long a stream must stay open before its clean close
+	// counts as healthy. The server closes a subscription after minutes without
+	// a frame; heartbeats are not frames, so a quiet stretch must not count
+	// toward maxFailures.
+	healthyAfter time.Duration
+}
+
+func newWatchRunner(client *flashduty.Client, sessionID, rootID string, since int64, out, errw io.Writer) *watchRunner {
+	return &watchRunner{
+		client: client, state: newWatchState(rootID, since), sessID: sessionID, out: out, errw: errw,
+		maxFailures:    8,
+		backoff:        func(attempt int) time.Duration { return time.Second << min(attempt, 5) },
+		confirmRetries: 5,
+		confirmGap:     2 * time.Second,
+		healthyAfter:   30 * time.Second,
+	}
+}
+
+func (r *watchRunner) emit(ev watchEvent) error {
+	b, err := json.Marshal(ev)
+	if err != nil {
+		return err
+	}
+	_, err = fmt.Fprintf(r.out, "%s\n", b)
+	return err
+}
+
+// errFatalFrame ends the watch after an event that closes the session's output.
+var errFatalFrame = errors.New("session produced a terminal error")
+
+// run blocks until the request is fully done (nil), an unrecoverable error, or
+// ctx cancellation.
+//
+// Done rule: a session_turn_settled frame for the watched root that lists no
+// standing tasks means the root has no more automatic work; the watcher then
+// asks POST /safari/session/get and exits only when is_running is false (no
+// in-flight turn, so no queued follow-up turn or notification round is about
+// to start). A settle that lists standing tasks keeps the watch open: those
+// background tasks post later rounds into the same root.
+//
+// Why the first such settle is final for the root: the server persists a
+// settle only when the whole session has no open tasks and no queued
+// notifications (logic/session/actor/settled.go, maybePersistSessionTurnSettled),
+// and stamps standing_tasks with the running bash_background/monitor tasks
+// (model/task/task.go, ListSessionRunningProcessTasks). With both empty nothing
+// can start another round under that root; a user follow-up opens a new root.
+func (r *watchRunner) run(ctx context.Context) error {
+	failures := 0
+	settledSeen := false
+	confirmTries := r.confirmRetries
+	for {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		if settledSeen {
+			done, err := r.confirmIdle(ctx, confirmTries)
+			if err != nil {
+				return err
+			}
+			if done {
+				return nil
+			}
+			// Still running: resume listening, and re-check once per later
+			// reconnect instead of waiting out the full retry budget again.
+			confirmTries = 1
+		}
+
+		started := time.Now()
+		progressed, settled, err := r.streamOnce(ctx)
+		if err == nil && time.Since(started) >= r.healthyAfter {
+			progressed = true
+		}
+		if settled {
+			settledSeen = true
+			confirmTries = r.confirmRetries
+		}
+		if err != nil {
+			if errors.Is(err, errFatalFrame) || ctx.Err() != nil || isPermanent(err) {
+				return err
+			}
+		}
+		if progressed {
+			failures = 0
+		} else {
+			failures++
+			if failures > r.maxFailures {
+				if err == nil {
+					err = errors.New("stream closed repeatedly without events")
+				}
+				return fmt.Errorf("giving up after %d failed attempts: %w", failures-1, err)
+			}
+		}
+		if settled {
+			// Go straight to the idle confirmation at the top of the loop.
+			continue
+		}
+		reason := "stream closed"
+		if err != nil {
+			reason = err.Error()
+		}
+		_, _ = fmt.Fprintf(r.errw, "watch: %s; resuming from %d\n", reason, r.state.cursorAt)
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-time.After(r.backoff(failures)):
+		}
+	}
+}
+
+// streamOnce opens one subscription and processes frames until the stream ends.
+// settled reports that a done-candidate settle frame was seen; progressed that
+// at least one frame arrived.
+func (r *watchRunner) streamOnce(ctx context.Context) (progressed, settled bool, err error) {
+	rc, _, err := r.client.Sessions.Subscribe(ctx, &flashduty.SessionSubscribeRequest{
+		SessionID:   r.sessID,
+		LastEventAt: r.state.cursorAt,
+		LastEventID: r.state.cursorID,
+	})
+	if err != nil {
+		return false, false, err
+	}
+	defer func() { _ = rc.Close() }()
+
+	fr := flashduty.NewSessionFrameReader(rc)
+	for {
+		frame, err := fr.Next()
+		if err == io.EOF {
+			return progressed, settled, nil
+		}
+		if err != nil {
+			return progressed, settled, err
+		}
+		progressed = true
+		out := r.state.handle(frame)
+		for _, ev := range out.Events {
+			if werr := r.emit(ev); werr != nil {
+				return progressed, settled, fmt.Errorf("%w: %v", errFatalFrame, werr)
+			}
+		}
+		if out.Fatal {
+			return progressed, settled, errFatalFrame
+		}
+		if out.Settled {
+			return progressed, true, nil
+		}
+	}
+}
+
+// confirmIdle reports whether the session has no in-flight turn. A running
+// session right after a settle is usually a lagging in-flight marker, so it is
+// re-checked a few times before the watch goes back to listening.
+func (r *watchRunner) confirmIdle(ctx context.Context, tries int) (bool, error) {
+	for i := 0; i < tries; i++ {
+		resp, _, err := r.client.Sessions.ReadInfo(ctx, &flashduty.SessionGetRequest{SessionID: r.sessID, Limit: 1})
+		if err != nil {
+			if ctx.Err() != nil {
+				return false, ctx.Err()
+			}
+			if isPermanent(err) {
+				return false, err
+			}
+		} else if !resp.Session.IsRunning {
+			return true, nil
+		}
+		select {
+		case <-ctx.Done():
+			return false, ctx.Err()
+		case <-time.After(r.confirmGap):
+		}
+	}
+	return false, nil
+}
+
+// isPermanent reports whether retrying cannot help: a 4xx API answer other
+// than rate limiting (bad session id, no access, bad credentials).
+func isPermanent(err error) bool {
+	var rl *flashduty.RateLimitError
+	if errors.As(err, &rl) {
+		return false
+	}
+	var apiErr *flashduty.ErrorResponse
+	if errors.As(err, &apiErr) && apiErr.Response != nil {
+		s := apiErr.Response.StatusCode
+		return s >= 400 && s < 500 && s != http.StatusRequestTimeout
+	}
+	return false
+}
+
+func newSafariWatchCmd() *cobra.Command {
+	var sessionID, rootID, sinceEventID string
+	var since int64
+
+	cmd := &cobra.Command{
+		Use:   "watch --session-id <id> [--root-invocation-id <id>] [--since <unix-ms>]",
+		Short: "Stream an AI SRE session's replies as one JSON line each",
+		Long: `Follow an AI SRE session and print what the agent sends the user, one compact
+JSON object per line on stdout (nothing else goes to stdout; diagnostics go to
+stderr). Built to run under a tool that turns every stdout line into a
+notification, such as Claude Code's Monitor.
+
+Events:
+  {"type":"reply","purpose":"progress|result","message":"...","invocation_id":"...","event_id":"...","created_at":1700000000000}
+  {"type":"settled","root_invocation_id":"...","reason":"...","event_id":"...","created_at":1700000000000}
+  {"type":"error","message":"...","created_at":1700000000000}
+
+Only messages the agent explicitly sends with its reply tool are printed; tool
+calls, thinking and ordinary assistant text are not. Every event carries created_at (Unix ms of its frame). Events are de-duplicated
+by event id and the stream is resumed from the last seen event after a drop.
+
+To re-arm a watch later, pass the last line's created_at as --since and its
+event_id as --since-event-id: replay then starts strictly after that event.
+--since alone resumes at that millisecond inclusive, so the frame at exactly
+that time (and any sibling in the same millisecond) is delivered again.
+
+With --root-invocation-id (printed by 'safari send') only that request's frames
+count. Without it, --since (Unix milliseconds) bounds the replay; the default
+is "now".
+
+The command exits 0 once the request is fully done: the root has settled with
+no background tasks left and the session reports no running turn. It exits
+non-zero on an unrecoverable error.
+
+API: POST /safari/session/subscribe, POST /safari/session/get`,
+		Example: `  flashduty safari watch --session-id sess_01 --root-invocation-id inv_01`,
+		Args:    cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return runCommand(cmd, args, func(ctx *RunContext) error {
+				if sinceEventID != "" && since == 0 {
+					return fmt.Errorf("--since-event-id requires --since (the created_at of the same event)")
+				}
+				if since == 0 && rootID == "" {
+					since = time.Now().UnixMilli()
+				}
+				r := newWatchRunner(ctx.Client, sessionID, rootID, since, ctx.Writer, ctx.Cmd.ErrOrStderr())
+				if sinceEventID != "" {
+					r.state.cursorID = sinceEventID
+					r.state.seen[sinceEventID] = struct{}{}
+				}
+				err := r.run(cmdContext(ctx.Cmd))
+				if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, errFatalFrame) {
+					// Surface the failure on stdout as well: a notification
+					// consumer reads nothing else.
+					_ = r.emit(watchEvent{Type: "error", Message: err.Error()})
+				}
+				return err
+			})
+		},
+	}
+	cmd.Flags().StringVar(&sessionID, "session-id", "", "Session to follow (required)")
+	cmd.Flags().StringVar(&rootID, "root-invocation-id", "", "Only follow this request (the root_invocation_id printed by 'safari send')")
+	cmd.Flags().Int64Var(&since, "since", 0, "Replay from this Unix-millisecond timestamp (default: now, or the start of the session with --root-invocation-id)")
+	cmd.Flags().StringVar(&sinceEventID, "since-event-id", "", "With --since: resume strictly after the event with this id (the event_id of the last line seen)")
+	_ = cmd.MarkFlagRequired("session-id")
+	return cmd
+}
diff --git a/internal/cli/safari_watch_resume_test.go b/internal/cli/safari_watch_resume_test.go
new file mode 100644
index 0000000..1ffed2d
--- /dev/null
+++ b/internal/cli/safari_watch_resume_test.go
@@ -0,0 +1,66 @@
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/flashcatcloud/go-flashduty"
+)
+
+func TestWatchEventsCarryCreatedAt(t *testing.T) {
+	w := newWatchState("", 0)
+	for _, f := range []*flashduty.SessionEventFrame{
+		replyFrame("e1", "i", "r", "result", "hi", 1700000000123),
+		settleFrame("e2", "r", 1700000000456, 0),
+		{EventID: "e3", CreatedAt: 1700000000789, Type: "assistant", IsError: true, Payload: json.RawMessage(`{"parts":[{"text":"boom"}]}`)},
+	} {
+		out := w.handle(f)
+		if len(out.Events) != 1 || out.Events[0].CreatedAt != f.CreatedAt {
+			t.Errorf("frame %s: events = %+v, want created_at %d", f.EventID, out.Events, f.CreatedAt)
+			continue
+		}
+		b, _ := json.Marshal(out.Events[0])
+		if !strings.Contains(string(b), fmt.Sprintf(`"created_at":%d`, f.CreatedAt)) {
+			t.Errorf("serialized event lacks created_at: %s", b)
+		}
+	}
+}
+
+func TestWatchSinceEventIDRequiresSince(t *testing.T) {
+	saveAndResetGlobals(t)
+	newClientFn = func() (*flashduty.Client, error) {
+		return flashduty.NewClient("k", flashduty.WithBaseURL("http://127.0.0.1:1"))
+	}
+	_, err := execCommand("safari", "watch", "--session-id", "s1", "--root-invocation-id", "r1", "--since-event-id", "e1")
+	if err == nil || !strings.Contains(err.Error(), "--since-event-id requires --since") {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestWatchSinceEventIDResumesStrictlyAfter(t *testing.T) {
+	s := &watchServer{
+		streams: []string{sse(
+			// A server that only knows the millisecond replays the boundary frame.
+			replyFrame("e1", "i", "r1", "result", "already seen", 500),
+			replyFrame("e2", "i", "r1", "result", "new", 501),
+			settleFrame("e3", "r1", 502, 0),
+		)},
+		running: []bool{false},
+	}
+	r, out, _ := newTestWatchRunner(t, s, "r1")
+	r.state = newWatchState("r1", 500)
+	r.state.cursorID = "e1"
+	r.state.seen["e1"] = struct{}{}
+	if err := r.run(context.Background()); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if !strings.Contains(s.cursors[0], `"last_event_at":500`) || !strings.Contains(s.cursors[0], `"last_event_id":"e1"`) {
+		t.Errorf("subscribe body = %s", s.cursors[0])
+	}
+	if strings.Contains(out.String(), "already seen") || !strings.Contains(out.String(), "new") {
+		t.Errorf("stdout:\n%s", out.String())
+	}
+}
diff --git a/internal/cli/safari_watch_test.go b/internal/cli/safari_watch_test.go
new file mode 100644
index 0000000..3ca38c3
--- /dev/null
+++ b/internal/cli/safari_watch_test.go
@@ -0,0 +1,473 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/flashcatcloud/go-flashduty"
+)
+
+func replyFrame(eventID, invocation, root, purpose, message string, createdAt int64) *flashduty.SessionEventFrame {
+	payload := fmt.Sprintf(`{"parts":[{"function_response":{"name":"reply","response":{"submitted":true,"purpose":%q,"message":%q}}}]}`, purpose, message)
+	return &flashduty.SessionEventFrame{
+		EventID: eventID, CreatedAt: createdAt, Type: "assistant",
+		InvocationID: invocation, RootInvocationID: root, Payload: json.RawMessage(payload),
+	}
+}
+
+func settleFrame(eventID, root string, createdAt int64, standing int) *flashduty.SessionEventFrame {
+	tasks := make([]string, standing)
+	for i := range tasks {
+		tasks[i] = fmt.Sprintf(`{"task_id":"t%d","kind":"process"}`, i)
+	}
+	payload := fmt.Sprintf(`{"root_invocation_id":%q,"reason":"completed","standing_tasks":[%s]}`, root, strings.Join(tasks, ","))
+	return &flashduty.SessionEventFrame{
+		EventID: eventID, CreatedAt: createdAt, Type: "session_turn_settled",
+		RootInvocationID: root, Payload: json.RawMessage(payload),
+	}
+}
+
+func TestWatchStateReplyMapping(t *testing.T) {
+	w := newWatchState("", 0)
+
+	out := w.handle(replyFrame("e1", "inv1", "root1", "progress", " looking at logs ", 10))
+	if len(out.Events) != 1 {
+		t.Fatalf("events = %+v", out.Events)
+	}
+	ev := out.Events[0]
+	if ev.Type != "reply" || ev.Purpose != "progress" || ev.Message != "looking at logs" || ev.InvocationID != "inv1" || ev.EventID != "e1" {
+		t.Errorf("event = %+v", ev)
+	}
+
+	// An empty purpose is a result (frames written before purpose existed).
+	out = w.handle(replyFrame("e2", "inv1", "root1", "", "done", 11))
+	if len(out.Events) != 1 || out.Events[0].Purpose != "result" {
+		t.Errorf("empty purpose events = %+v", out.Events)
+	}
+
+	// The legacy receipt flag still counts.
+	legacy := &flashduty.SessionEventFrame{EventID: "e3", Type: "assistant", Payload: json.RawMessage(
+		`{"parts":[{"function_response":{"name":"reply","response":{"delivered":true,"message":"old"}}}]}`)}
+	if out = w.handle(legacy); len(out.Events) != 1 || out.Events[0].Message != "old" {
+		t.Errorf("legacy events = %+v", out.Events)
+	}
+
+	// Two replies in one frame are both emitted, in order.
+	multi := &flashduty.SessionEventFrame{EventID: "e4", Type: "assistant", Payload: json.RawMessage(
+		`{"parts":[{"function_response":{"name":"reply","response":{"submitted":true,"message":"a"}}},{"function_response":{"name":"reply","response":{"submitted":true,"purpose":"progress","message":"b"}}}]}`)}
+	if out = w.handle(multi); len(out.Events) != 2 || out.Events[0].Message != "a" || out.Events[1].Purpose != "progress" {
+		t.Errorf("multi events = %+v", out.Events)
+	}
+}
+
+func TestWatchStateDropsEverythingButReplies(t *testing.T) {
+	cases := map[string]string{
+		"assistant prose":        `{"parts":[{"text":"private working text"}]}`,
+		"tool call":              `{"parts":[{"function_call":{"name":"reply","args":{"message":"x"}}}]}`,
+		"other tool result":      `{"parts":[{"function_response":{"name":"shell","response":{"submitted":true,"message":"x"}}}]}`,
+		"refused reply":          `{"parts":[{"function_response":{"name":"reply","response":{"submitted":false,"message":"x"}}}]}`,
+		"unknown purpose":        `{"parts":[{"function_response":{"name":"reply","response":{"submitted":true,"purpose":"weird","message":"x"}}}]}`,
+		"empty message":          `{"parts":[{"function_response":{"name":"reply","response":{"submitted":true,"message":"  "}}}]}`,
+		"thinking text":          `{"parts":[{"text":"hmm","thought":true}]}`,
+		"malformed payload":      `not json`,
+		"ephemeral token chunk":  `{"parts":[{"text":"par"}]}`,
+		"user echo of own query": `{"parts":[{"text":"question"}]}`,
+	}
+	for name, payload := range cases {
+		t.Run(name, func(t *testing.T) {
+			w := newWatchState("", 0)
+			frame := &flashduty.SessionEventFrame{EventID: "e", Type: "assistant", Payload: json.RawMessage(payload)}
+			if name == "ephemeral token chunk" {
+				frame = &flashduty.SessionEventFrame{Type: "assistant_token", Payload: json.RawMessage(payload)}
+			}
+			if name == "user echo of own query" {
+				frame = &flashduty.SessionEventFrame{EventID: "e", Type: "user", Payload: json.RawMessage(payload)}
+			}
+			if out := w.handle(frame); len(out.Events) != 0 || out.Settled || out.Fatal {
+				t.Errorf("outcome = %+v, want nothing", out)
+			}
+		})
+	}
+}
+
+func TestWatchStateDedupeAndCursor(t *testing.T) {
+	w := newWatchState("", 5)
+	f := replyFrame("e1", "i", "r", "result", "once", 100)
+	if len(w.handle(f).Events) != 1 {
+		t.Fatal("first delivery must emit")
+	}
+	if len(w.handle(f).Events) != 0 {
+		t.Error("a replayed event id must not emit twice")
+	}
+	if w.cursorAt != 100 || w.cursorID != "e1" {
+		t.Errorf("cursor = %d/%s", w.cursorAt, w.cursorID)
+	}
+	// The cursor is the last persisted frame received, in the server's order,
+	// not a client-side maximum: an id that sorts lower bytewise still wins
+	// when it arrives later.
+	w.handle(replyFrame("B1", "i", "r", "result", "same ms, collation-later", 100))
+	if w.cursorAt != 100 || w.cursorID != "B1" {
+		t.Errorf("cursor = %d/%s, want 100/B1", w.cursorAt, w.cursorID)
+	}
+}
+
+func TestWatchStateRootFilter(t *testing.T) {
+	w := newWatchState("rootA", 0)
+	if out := w.handle(replyFrame("e1", "invB", "rootB", "result", "other", 1)); len(out.Events) != 0 {
+		t.Errorf("other root emitted: %+v", out.Events)
+	}
+	// Filtered frames still advance the resume cursor.
+	if w.cursorID != "e1" {
+		t.Errorf("cursor id = %q", w.cursorID)
+	}
+	if out := w.handle(replyFrame("e2", "invA", "rootA", "result", "mine", 2)); len(out.Events) != 1 {
+		t.Errorf("own root dropped: %+v", out.Events)
+	}
+	// A frame without a root id belongs to the root whose own invocation it is.
+	bare := replyFrame("e3", "rootA", "", "result", "bare", 3)
+	if out := w.handle(bare); len(out.Events) != 1 {
+		t.Errorf("bare invocation frame dropped: %+v", out.Events)
+	}
+	if out := w.handle(settleFrame("e4", "rootB", 4, 0)); len(out.Events) != 0 || out.Settled {
+		t.Errorf("other root's settle leaked: %+v", out)
+	}
+}
+
+func TestWatchStateSettle(t *testing.T) {
+	w := newWatchState("r", 0)
+	out := w.handle(settleFrame("s1", "r", 1, 0))
+	if !out.Settled || len(out.Events) != 1 || out.Events[0].Type != "settled" || out.Events[0].Reason != "completed" || out.Events[0].RootInvocationID != "r" {
+		t.Errorf("settle outcome = %+v", out)
+	}
+	// Standing tasks keep the root open: reported, but not a done candidate.
+	out = w.handle(settleFrame("s2", "r", 2, 2))
+	if out.Settled || len(out.Events) != 1 || out.Events[0].StandingTasks != 2 {
+		t.Errorf("standing settle outcome = %+v", out)
+	}
+}
+
+func TestWatchStateErrors(t *testing.T) {
+	w := newWatchState("", 0)
+	out := w.handle(&flashduty.SessionEventFrame{Type: "environment.failed", Payload: json.RawMessage(`{"error_message":"no sandbox capacity"}`)})
+	if !out.Fatal || len(out.Events) != 1 || out.Events[0].Type != "error" || out.Events[0].Message != "no sandbox capacity" {
+		t.Errorf("env failed outcome = %+v", out)
+	}
+	out = w.handle(&flashduty.SessionEventFrame{Type: "error", Payload: json.RawMessage(`{"message":"model unavailable"}`)})
+	if out.Fatal || len(out.Events) != 1 || out.Events[0].Message != "model unavailable" {
+		t.Errorf("error frame outcome = %+v", out)
+	}
+	out = w.handle(&flashduty.SessionEventFrame{EventID: "e9", Type: "assistant", IsError: true, ErrorCode: "llm_failed", Payload: json.RawMessage(`{"parts":[{"text":"provider error"}]}`)})
+	if len(out.Events) != 1 || out.Events[0].Message != "provider error" || out.Events[0].Code != "llm_failed" {
+		t.Errorf("is_error outcome = %+v", out)
+	}
+}
+
+// watchServer scripts /subscribe (one SSE body per connection) and /get.
+type watchServer struct {
+	t         *testing.T
+	streams   []string
+	conns     atomic.Int32
+	cursors   []string
+	mu        sync.Mutex
+	running   []bool // is_running answers, consumed in order; the last repeats
+	getCalls  atomic.Int32
+	streamErr int // when >0, the first N subscribe calls answer HTTP 500
+}
+
+func (s *watchServer) handler(w http.ResponseWriter, r *http.Request) {
+	body, _ := io.ReadAll(r.Body)
+	switch r.URL.Path {
+	case "/safari/session/subscribe":
+		n := int(s.conns.Add(1)) - 1
+		s.mu.Lock()
+		s.cursors = append(s.cursors, string(body))
+		s.mu.Unlock()
+		if n < s.streamErr {
+			w.WriteHeader(http.StatusInternalServerError)
+			_, _ = io.WriteString(w, `{"error":{"code":"InternalError","message":"boom"}}`)
+			return
+		}
+		i := n - s.streamErr
+		if i >= len(s.streams) {
+			i = len(s.streams) - 1
+		}
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, s.streams[i])
+	case "/safari/session/get":
+		k := int(s.getCalls.Add(1)) - 1
+		s.mu.Lock()
+		if k >= len(s.running) {
+			k = len(s.running) - 1
+		}
+		running := s.running[k]
+		s.mu.Unlock()
+		_ = json.NewEncoder(w).Encode(map[string]any{"request_id": "r", "data": map[string]any{
+			"events": []any{}, "pending_messages": []any{}, "session": map[string]any{"session_id": "s1", "is_running": running},
+		}})
+	default:
+		s.t.Errorf("unexpected path %s", r.URL.Path)
+	}
+}
+
+func sse(frames ...*flashduty.SessionEventFrame) string {
+	var b strings.Builder
+	b.WriteString(": ping\n\n")
+	for _, f := range frames {
+		raw, _ := json.Marshal(f)
+		fmt.Fprintf(&b, "data: %s\n\n", raw)
+	}
+	return b.String()
+}
+
+func newTestWatchRunner(t *testing.T, s *watchServer, root string) (*watchRunner, *bytes.Buffer, *bytes.Buffer) {
+	t.Helper()
+	s.t = t
+	srv := httptest.NewServer(http.HandlerFunc(s.handler))
+	t.Cleanup(srv.Close)
+	c, err := flashduty.NewClient("k", flashduty.WithBaseURL(srv.URL))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var out, errw bytes.Buffer
+	r := newWatchRunner(c, "s1", root, 0, &out, &errw)
+	r.backoff = func(int) time.Duration { return time.Millisecond }
+	r.confirmGap = time.Millisecond
+	return r, &out, &errw
+}
+
+func outputLines(b *bytes.Buffer) []string {
+	s := strings.TrimSpace(b.String())
+	if s == "" {
+		return nil
+	}
+	return strings.Split(s, "\n")
+}
+
+func TestWatchRunExitsWhenSettledAndIdle(t *testing.T) {
+	s := &watchServer{
+		streams: []string{sse(
+			replyFrame("e1", "i1", "r1", "progress", "checking", 1),
+			replyFrame("e2", "i1", "r1", "result", "root cause: disk full", 2),
+			settleFrame("e3", "r1", 3, 0),
+		)},
+		running: []bool{false},
+	}
+	r, out, _ := newTestWatchRunner(t, s, "r1")
+	if err := r.run(context.Background()); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	lines := outputLines(out)
+	if len(lines) != 3 {
+		t.Fatalf("stdout lines = %q", lines)
+	}
+	for _, l := range lines {
+		var m map[string]any
+		if err := json.Unmarshal([]byte(l), &m); err != nil {
+			t.Errorf("stdout line is not a JSON object: %q", l)
+		}
+	}
+	if !strings.Contains(lines[1], `"purpose":"result"`) || !strings.Contains(lines[2], `"type":"settled"`) {
+		t.Errorf("lines = %q", lines)
+	}
+}
+
+func TestWatchRunKeepsListeningWhileSessionRunning(t *testing.T) {
+	// The first settle is followed by a still-running session (a second round is
+	// starting); the watcher must keep listening and finish after round two.
+	s := &watchServer{
+		streams: []string{
+			sse(replyFrame("e1", "i1", "r1", "result", "round one", 1), settleFrame("e2", "r1", 2, 0)),
+			sse(replyFrame("e1", "i1", "r1", "result", "round one", 1), replyFrame("e3", "i2", "r1", "result", "round two", 3), settleFrame("e4", "r1", 4, 0)),
+		},
+		running: []bool{true, true, false},
+	}
+	r, out, _ := newTestWatchRunner(t, s, "r1")
+	r.confirmRetries = 2
+	if err := r.run(context.Background()); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	text := out.String()
+	if strings.Count(text, "round one") != 1 || strings.Count(text, "round two") != 1 {
+		t.Errorf("replies not delivered exactly once:\n%s", text)
+	}
+	if strings.Count(text, `"type":"settled"`) != 2 {
+		t.Errorf("want two settled events:\n%s", text)
+	}
+}
+
+func TestWatchRunStandingTasksKeepWatchOpen(t *testing.T) {
+	s := &watchServer{
+		streams: []string{
+			sse(settleFrame("e1", "r1", 1, 1)),
+			sse(replyFrame("e2", "i2", "r1", "result", "task finished", 2), settleFrame("e3", "r1", 3, 0)),
+		},
+		running: []bool{false},
+	}
+	r, out, _ := newTestWatchRunner(t, s, "r1")
+	if err := r.run(context.Background()); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if s.conns.Load() != 2 {
+		t.Errorf("connections = %d, want 2 (a settle with standing tasks must not end the watch)", s.conns.Load())
+	}
+	if !strings.Contains(out.String(), "task finished") {
+		t.Errorf("stdout:\n%s", out.String())
+	}
+}
+
+func TestWatchRunResumesFromCursorAfterDrop(t *testing.T) {
+	// The stream drops after one reply without [DONE]; the reconnect must carry
+	// the cursor, and the replayed frame must not print twice.
+	s := &watchServer{
+		streams: []string{
+			sse(replyFrame("e1", "i1", "r1", "progress", "step one", 7)),
+			sse(replyFrame("e1", "i1", "r1", "progress", "step one", 7), settleFrame("e2", "r1", 8, 0)),
+		},
+		running: []bool{false},
+	}
+	r, out, errw := newTestWatchRunner(t, s, "r1")
+	if err := r.run(context.Background()); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if strings.Count(out.String(), "step one") != 1 {
+		t.Errorf("duplicate reply:\n%s", out.String())
+	}
+	if len(s.cursors) != 2 || !strings.Contains(s.cursors[1], `"last_event_at":7`) || !strings.Contains(s.cursors[1], `"last_event_id":"e1"`) {
+		t.Errorf("subscribe bodies = %q", s.cursors)
+	}
+	if !strings.Contains(errw.String(), "resuming") {
+		t.Errorf("diagnostics must go to stderr, got %q", errw.String())
+	}
+	if strings.Contains(out.String(), "resuming") {
+		t.Errorf("diagnostics leaked to stdout:\n%s", out.String())
+	}
+}
+
+func TestWatchRunRetriesTransientHTTPErrors(t *testing.T) {
+	s := &watchServer{
+		streamErr: 2,
+		streams:   []string{sse(settleFrame("e1", "r1", 1, 0))},
+		running:   []bool{false},
+	}
+	r, _, _ := newTestWatchRunner(t, s, "r1")
+	if err := r.run(context.Background()); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if s.conns.Load() != 3 {
+		t.Errorf("connections = %d, want 3", s.conns.Load())
+	}
+}
+
+func TestWatchRunGivesUpAfterRepeatedFailures(t *testing.T) {
+	s := &watchServer{streamErr: 100, streams: []string{""}, running: []bool{false}}
+	r, _, _ := newTestWatchRunner(t, s, "r1")
+	r.maxFailures = 2
+	if err := r.run(context.Background()); err == nil || !strings.Contains(err.Error(), "giving up") {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestWatchRunFailsFastOnClientErrors(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+		_, _ = io.WriteString(w, `{"error":{"code":"AccessDenied","message":"session not accessible"}}`)
+	}))
+	t.Cleanup(srv.Close)
+	c, _ := flashduty.NewClient("k", flashduty.WithBaseURL(srv.URL))
+	var out, errw bytes.Buffer
+	r := newWatchRunner(c, "s1", "", 0, &out, &errw)
+	r.backoff = func(int) time.Duration { return time.Millisecond }
+	err := r.run(context.Background())
+	if err == nil || !strings.Contains(err.Error(), "not accessible") {
+		t.Fatalf("err = %v", err)
+	}
+}
+
+func TestWatchRunEnvironmentFailureIsFatal(t *testing.T) {
+	s := &watchServer{
+		streams: []string{sse(&flashduty.SessionEventFrame{Type: "environment.failed", RootInvocationID: "r1", Payload: json.RawMessage(`{"error_message":"no capacity"}`)})},
+		running: []bool{false},
+	}
+	r, out, _ := newTestWatchRunner(t, s, "r1")
+	if err := r.run(context.Background()); err == nil {
+		t.Fatal("expected a non-nil error")
+	}
+	if !strings.Contains(out.String(), `"type":"error"`) || !strings.Contains(out.String(), "no capacity") {
+		t.Errorf("stdout:\n%s", out.String())
+	}
+}
+
+func TestWatchRunStopsOnContextCancel(t *testing.T) {
+	release := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = io.WriteString(w, ": ping\n\n")
+		w.(http.Flusher).Flush()
+		select {
+		case <-r.Context().Done():
+		case <-release:
+		}
+	}))
+	t.Cleanup(func() { close(release); srv.Close() })
+	c, _ := flashduty.NewClient("k", flashduty.WithBaseURL(srv.URL))
+	var out, errw bytes.Buffer
+	r := newWatchRunner(c, "s1", "r1", 0, &out, &errw)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- r.run(ctx) }()
+	time.Sleep(50 * time.Millisecond)
+	cancel()
+	select {
+	case err := <-done:
+		if err == nil {
+			t.Error("expected a cancellation error")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("run did not return after cancel")
+	}
+	if out.Len() != 0 {
+		t.Errorf("stdout must stay empty, got %q", out.String())
+	}
+}
+
+func TestWatchRunQuietCleanClosesAreNotFailures(t *testing.T) {
+	// Every stream closes cleanly with no frames (a quiet stretch). With a zero
+	// healthy threshold each counts as healthy, so more reconnects than
+	// maxFailures still reach the settle that follows.
+	quiet := sse()
+	streams := make([]string, 0, 6)
+	for i := 0; i < 5; i++ {
+		streams = append(streams, quiet)
+	}
+	streams = append(streams, sse(settleFrame("e1", "r1", 1, 0)))
+	s := &watchServer{streams: streams, running: []bool{false}}
+	r, _, _ := newTestWatchRunner(t, s, "r1")
+	r.maxFailures = 2
+	r.healthyAfter = 0
+	if err := r.run(context.Background()); err != nil {
+		t.Fatalf("run: %v", err)
+	}
+	if s.conns.Load() != 6 {
+		t.Errorf("connections = %d, want 6", s.conns.Load())
+	}
+}
+
+func TestWatchRunQuickCleanClosesStillCountAsFailures(t *testing.T) {
+	s := &watchServer{streams: []string{sse()}, running: []bool{false}}
+	r, _, _ := newTestWatchRunner(t, s, "r1")
+	r.maxFailures = 2
+	if err := r.run(context.Background()); err == nil || !strings.Contains(err.Error(), "giving up") {
+		t.Fatalf("err = %v", err)
+	}
+}
diff --git a/skills/flashduty/reference/safari.md b/skills/flashduty/reference/safari.md
index 0dd6a80..c537c5c 100644
--- a/skills/flashduty/reference/safari.md
+++ b/skills/flashduty/reference/safari.md
@@ -391,6 +391,15 @@ Update MCP server
 - body-only (`--data`): env (object); headers (object)
 - response: same shape as `mcp-server-create` above
 
+### send [message]
+Send a message to an AI SRE session
+- `--env-id` string
+- `--env-kind` string
+- `--file` stringArray
+- `--message-file` string
+- `--session-id` string
+- `--team-id` int64
+
 ### session-delete <session-id>
 Delete session
 - `<session-id>` (positional, required) string — Target session ID, from the list returned by 'POST /safari/session/list'. (≥1 chars)
@@ -468,6 +477,13 @@ Upload a skill archive
 - `--skill-id` string
 - `--team-id` int64
 
+### watch --session-id <id> [--root-invocation-id <id>] [--since <unix-ms>]
+Stream an AI SRE session's replies as one JSON line each
+- `--root-invocation-id` string
+- `--session-id` string
+- `--since` int64
+- `--since-event-id` string
+
 <!-- GENERATED:safari END -->
 
 ## Key concepts
@@ -476,6 +492,7 @@ Upload a skill archive
 - **Scope (`team_id`).** `0` = account-wide (every team sees it); `>0` = that team only. Same field on every safari verb.
 - **Auth mode (`auth_mode`).** `shared` (default) = one credential for everyone, stored on the server. `per_user_secret` = each user supplies a secret matching `secret_schema` (which must carry a `header_name`). `per_user_oauth` = each user authorizes the server via OAuth.
 - **`per_user_oauth` needs no OAuth config up front.** Create it with an **empty `oauth_metadata`** — that is the normal, complete state, not a missing prerequisite. The runtime **auto-discovers the OAuth server and dynamically registers a client (DCR)** the first time a user authorizes; you do **not** collect `authorization_url` / `client_id` / `client_secret` / `scopes`. Only pass `oauth_metadata` as a rare fallback, when the endpoint advertises no discovery document.
+- **Chat with the agent: `send` then `watch`.** `send` queues one message and prints `{session_id, invocation_id, root_invocation_id, delivery_mode}`; `watch --session-id <id> --root-invocation-id <root>` prints one JSON line per agent reply (`reply` with `purpose` `progress` or `result`), `settled`, or `error`, and exits 0 when the request is fully done. Run `watch` under a tool that turns each stdout line into a notification (e.g. Claude Code's Monitor). Files go up with `send --file` as attachments, not inlined text.
 
 ## Gotchas