diff --git a/cmd/odek/bg_wake.go b/cmd/odek/bg_wake.go index 499abcc..66d5b98 100644 --- a/cmd/odek/bg_wake.go +++ b/cmd/odek/bg_wake.go @@ -291,6 +291,18 @@ func wakeInitiated(msg wsClientMsg) bool { return msg.Type == "bg_wake" && msg.SystemInitiated } +// turnInitiatedLabel maps the type-gated wakeInitiated provenance to the +// initiated label carried by the turn_started frame. It is computed +// server-side only — a client prompt that forges SystemInitiated or a +// wake token can never claim system provenance (same rules as the +// session frame's system_initiated stamp, review P1-1). +func turnInitiatedLabel(msg wsClientMsg) string { + if wakeInitiated(msg) { + return "system" + } + return "operator" +} + // ── serve wiring ───────────────────────────────────────────────────────── // bgJobFrame builds the `bg_job` wire frame for a job transition (M2). diff --git a/cmd/odek/serve.go b/cmd/odek/serve.go index 950fed7..5e326f3 100644 --- a/cmd/odek/serve.go +++ b/cmd/odek/serve.go @@ -1128,6 +1128,67 @@ type wsClientMsg struct { WakeToken string `json:"wake_token,omitempty"` } +// ── Turn identity (turn_started wire frame) ─────────────────────────── + +// newTurnID returns a fresh turn identifier ("t_" + 128-bit random hex). +// Clients upsert streaming-card state by this id, so a collision would +// merge two distinct turns; entropy failure panics like newWakeToken — +// an empty id would silently break card reconciliation. +func newTurnID() string { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + panic("odek: crypto/rand unavailable for turn id: " + err.Error()) + } + return "t_" + hex.EncodeToString(b) +} + +// turnTaggedFrames lists the frame types that carry turn_id while a turn +// is active (R3). Lifecycle and sub-agent frames stay untouched so old +// clients see byte-identical shapes for them. +var turnTaggedFrames = map[string]bool{ + "thinking": true, + "token": true, + "tool_call": true, + "tool_result": true, + "done": true, + "error": true, +} + +// wsTurnAnnotator tags outbound frames with the active turn id (R3) so a +// client that attached mid-turn can attribute strays after a reconnect. +// One per connection: begin/end bracket each turn on the processor +// goroutine; wrap is shared with every emitter, including the agent's +// live tool-event and delta callbacks. Frames sent outside a turn and +// frame types outside turnTaggedFrames pass through unmodified. +type wsTurnAnnotator struct { + mu sync.Mutex + turnID string +} + +func (a *wsTurnAnnotator) begin(id string) { + a.mu.Lock() + a.turnID = id + a.mu.Unlock() +} + +func (a *wsTurnAnnotator) end() { + a.mu.Lock() + a.turnID = "" + a.mu.Unlock() +} + +func (a *wsTurnAnnotator) wrap(send func(map[string]any)) func(map[string]any) { + return func(m map[string]any) { + a.mu.Lock() + id := a.turnID + a.mu.Unlock() + if typ, ok := m["type"].(string); ok && id != "" && turnTaggedFrames[typ] { + m["turn_id"] = id + } + send(m) + } +} + // ── WebSocket Handler ────────────────────────────────────────────────── func handleWS(store *session.Store, resources *resource.Registry, resolved config.ResolvedConfig, system string, state *serveState, conn *golangws.Conn) { @@ -1182,7 +1243,17 @@ func handleWS(store *session.Store, resources *resource.Registry, resolved confi // Create ONE agent per WebSocket connection — provides buffer // continuity across turns within the same session. + // turnTag tags streamed frames with the active turn id (R3); the same + // annotated sender backs the agent's live callbacks (newServeAgent) + // and the processor-loop wsSend below, so every frame of a turn is + // attributed to it. + turnTag := &wsTurnAnnotator{} + wsSend := turnTag.wrap(func(m map[string]any) { writeWSJSON(conn, m) }) agent, bgRT, sandboxCleanup, mcpCleanup, guardCleanup, injectionGuard, approver, err := newServeAgent(resolved, system, connInfo.ID, func(v any) error { + if m, ok := v.(map[string]any); ok { + wsSend(m) + return nil + } writeWSJSON(conn, v) return nil }, &deltas) @@ -1448,13 +1519,12 @@ func handleWS(store *session.Store, resources *resource.Registry, resolved confi approver.Cancel() } } - wsSend := func(m map[string]any) { writeWSJSON(conn, m) } connInfo.setLive(wake.SessionID, true) func() { // Panic-safe Busy pairing (review F2): a panic unwinding // through handlePrompt must not latch Busy=true forever. defer connInfo.setLive(wake.SessionID, false) - currentSession = handlePrompt(promptCtx, wsSend, store, resources, resolved, agent, injectionGuard, currentSession, wakeMsg, &sessionInputTokens, &sessionOutputTokens, promptCancelWithApproval, &deltas, bgRT) + currentSession = handlePrompt(promptCtx, wsSend, store, resources, resolved, agent, injectionGuard, currentSession, wakeMsg, &sessionInputTokens, &sessionOutputTokens, promptCancelWithApproval, &deltas, bgRT, turnTag) }() promptCancel() continue @@ -1536,12 +1606,11 @@ func handleWS(store *session.Store, resources *resource.Registry, resolved confi } } - wsSend := func(m map[string]any) { writeWSJSON(conn, m) } connInfo.setLive(msg.SessionID, true) func() { // Panic-safe Busy pairing (review F2). defer connInfo.setLive(msg.SessionID, false) - currentSession = handlePrompt(promptCtx, wsSend, store, resources, resolved, agent, injectionGuard, currentSession, msg, &sessionInputTokens, &sessionOutputTokens, promptCancelWithApproval, &deltas, bgRT) + currentSession = handlePrompt(promptCtx, wsSend, store, resources, resolved, agent, injectionGuard, currentSession, msg, &sessionInputTokens, &sessionOutputTokens, promptCancelWithApproval, &deltas, bgRT, turnTag) }() connInfo.recordPrompt() sid := "" @@ -1643,6 +1712,7 @@ func handlePrompt( promptCancel context.CancelFunc, deltas *wsDeltaCounters, bg *bgRuntime, + turn *wsTurnAnnotator, ) *session.Session { prompt := msg.Content sessionID := msg.SessionID @@ -1832,6 +1902,26 @@ func handlePrompt( sessFrame["system_initiated"] = true // absent on operator turns } send(sessFrame) + + // turn_started (protocol R1): every turn — wake and operator alike — + // announces itself immediately after the session frame and before the + // first streamed frame, so a client that misses the session frame + // (socket raced a reconnect) can still open the streaming card. The + // initiated label is computed by the wakeInitiated type gate; client + // input cannot influence it (R5). The session frame keeps its legacy + // system_initiated stamp for old clients. + turnID := newTurnID() + send(map[string]any{ + "type": "turn_started", + "turn_id": turnID, + "session_id": sid, + "initiated": turnInitiatedLabel(msg), + "model": resolved.Model, + }) + if turn != nil { + turn.begin(turnID) + defer turn.end() // streamed frames stop carrying this id at return + } sl := activeServeLog() if sl != nil { sl.logf("turn_started session=%s model=%s", sid, resolved.Model) diff --git a/cmd/odek/serve_runs.go b/cmd/odek/serve_runs.go index 08f98fc..b4a6eba 100644 --- a/cmd/odek/serve_runs.go +++ b/cmd/odek/serve_runs.go @@ -751,6 +751,13 @@ func startServeRun( run.cancel = cancel var deltas wsDeltaCounters + // One annotator backs BOTH frame paths of the run: the agent's live + // callbacks (tool_call/tool_result/iteration frames below) and + // handlePrompt's send — otherwise the recorded tail mixes tagged and + // untagged frames for REST consumers (adversarial review finding, + // 2026-09-03). + var turnTag wsTurnAnnotator + recordSend := turnTag.wrap(func(m map[string]any) { _ = run.record(m) }) agent, bgRT, sandboxCleanup, mcpCleanup, guardCleanup, injectionGuard, approver, err := newServeAgent(resolved, system, run.ID, func(v any) error { // wsApprover sends its typed approvalRequest struct; everything // else arrives as map[string]any. @@ -758,6 +765,10 @@ func startServeRun( run.recordApprovalRequest(ar) return nil } + if m, ok := v.(map[string]any); ok { + recordSend(m) + return nil + } return run.record(v) }, &deltas) if err != nil { @@ -824,7 +835,7 @@ func startServeRun( defer cleanup() var sessionIn, sessionOut int serveLogf("run_started run_id=%s", run.ID) - sess := handlePrompt(ctx, func(m map[string]any) { _ = run.record(m) }, store, resources, resolved, agent, injectionGuard, nil, msg, &sessionIn, &sessionOut, cancelWithApproval, &deltas, bgRT) + sess := handlePrompt(ctx, recordSend, store, resources, resolved, agent, injectionGuard, nil, msg, &sessionIn, &sessionOut, cancelWithApproval, &deltas, bgRT, &turnTag) run.mu.Lock() if sess != nil { run.SessionID = sess.ID diff --git a/cmd/odek/turn_started_test.go b/cmd/odek/turn_started_test.go new file mode 100644 index 0000000..3a19603 --- /dev/null +++ b/cmd/odek/turn_started_test.go @@ -0,0 +1,314 @@ +package main + +// RED-first tests for the `turn_started` wire frame (bodek task spec, +// 2026-09-02). The spec: +// +// R1 — handlePrompt emits a `turn_started` frame after the `session` +// frame and before the first streamed event, for EVERY turn. Wake +// turns carry initiated:"system"; operator turns "operator". +// R2 — clients upsert by turn_id; the server must never emit two +// turn_started frames for the same turn, and ids must be unique +// per turn. +// R3 — streamed frames (thinking/token/tool_call/tool_result/done/ +// error) carry turn_id so a client attaching mid-turn can +// attribute strays after a reconnect. +// R5 — forged-initiated rejection: the initiated label is computed +// server-side via the wakeInitiated type gate; a client prompt +// that forges system_initiated must never be honored. +// +// These integration tests drive the real handleWS stack (buildServeMuxV2) +// against a mock LLM and assert on the actual frame stream. + +import ( + "encoding/json" + "net/http" + "sync" + "testing" + "time" + + golangws "golang.org/x/net/websocket" +) + +// collectTurnFrames reads frames until the turn finishes (done or error) +// and returns every decoded frame in arrival order. +func collectTurnFrames(t *testing.T, conn *golangws.Conn, deadline time.Duration) []map[string]any { + t.Helper() + conn.SetReadDeadline(time.Now().Add(deadline)) + var evts []map[string]any + for i := 0; i < 400; i++ { + var data []byte + if err := golangws.Message.Receive(conn, &data); err != nil { + t.Fatalf("Receive: %v (frames collected: %d)", err, len(evts)) + } + var evt map[string]any + if err := json.Unmarshal(data, &evt); err != nil { + continue + } + evts = append(evts, evt) + if evt["type"] == "done" || evt["type"] == "error" { + return evts + } + } + t.Fatal("turn did not finish before deadline") + return nil +} + +// findTurnStarted returns the turn_started frames in order. +func findTurnStarted(frames []map[string]any) []map[string]any { + var out []map[string]any + for _, f := range frames { + if f["type"] == "turn_started" { + out = append(out, f) + } + } + return out +} + +func indexOfFrame(frames []map[string]any, typ string) int { + for i, f := range frames { + if f["type"] == typ { + return i + } + } + return -1 +} + +// startTurnServer wires the full WS stack against a mock LLM whose chat +// response carries reasoning_content + content, so the turn produces the +// bulk thinking/token frames plus done. +func startTurnServer(t *testing.T) *golangws.Conn { + t.Helper() + llmSrv := mockLLM(t, func(w http.ResponseWriter, callCount int) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"choices":[{"message":{"reasoning_content":"pondering","content":"hi"}}]}`)) + }) + t.Cleanup(llmSrv.Close) + envCleanup := setTestEnv(t, llmSrv.URL) + t.Cleanup(envCleanup) + + store := newTestSessionStore(t) + ln, mux := buildServeMuxV2(t, store, nil) + t.Cleanup(func() { ln.Close() }) + go func() { _ = serveOnListener(ln, mux) }() + waitForHTTP(t, ln.Addr().String()) + + wsUpgradeLimiter.reset() + conn := dialTestWS(t, ln.Addr().String()) + t.Cleanup(func() { conn.Close() }) + readWSUntil(t, conn, 10*time.Second, func(e map[string]any) bool { return e["type"] == "server_info" }) + return conn +} + +// R1 (operator turn) + frame-order contract: session → turn_started → +// first streamed frame, and the frame's shape. +func TestWSTurnStarted_FrameOrderAndShape(t *testing.T) { + conn := startTurnServer(t) + + writeJSON(conn, map[string]any{"type": "prompt", "content": "hello"}) + frames := collectTurnFrames(t, conn, 15*time.Second) + + sessIdx := indexOfFrame(frames, "session") + if sessIdx < 0 { + t.Fatalf("no session frame in stream: %v", frameTypes(frames)) + } + started := findTurnStarted(frames) + if len(started) != 1 { + t.Fatalf("turn_started frames = %d, want exactly 1 (R2): %v", len(started), frameTypes(frames)) + } + startIdx := indexOfFrame(frames, "turn_started") + + // Order: session → turn_started with nothing in between, and no + // streamed frame before it. + if startIdx != sessIdx+1 { + t.Fatalf("turn_started at index %d, want immediately after session (index %d): %v", startIdx, sessIdx, frameTypes(frames)) + } + for _, streamed := range []string{"thinking", "token", "tool_call", "tool_result", "done"} { + if i := indexOfFrame(frames, streamed); i >= 0 && i < startIdx { + t.Fatalf("%s frame arrived before turn_started (index %d < %d)", streamed, i, startIdx) + } + } + + ts := started[0] + if id, _ := ts["turn_id"].(string); len(id) < 3 || id[:2] != "t_" { + t.Errorf("turn_started.turn_id = %v, want a %q-prefixed id", ts["turn_id"], "t_") + } + sid, _ := frames[sessIdx]["session_id"].(string) + if sid == "" { + t.Errorf("session frame carries empty session_id: %v", frames[sessIdx]) + } + if ts["session_id"] != sid { + t.Errorf("turn_started.session_id = %v, want the session frame's %q", ts["session_id"], sid) + } + if got, _ := ts["initiated"].(string); got != "operator" { + t.Errorf("turn_started.initiated = %q, want %q for an operator prompt", got, "operator") + } + if model, _ := ts["model"].(string); model != frames[sessIdx]["model"] { + t.Errorf("turn_started.model = %q, want parity with the session frame's model %q (both describe the same turn)", ts["model"], frames[sessIdx]["model"]) + } + + // R3: the streamed frames of this turn carry the same turn_id. + turnID, _ := ts["turn_id"].(string) + for _, typ := range []string{"thinking", "token", "done"} { + i := indexOfFrame(frames, typ) + if i < 0 { + t.Errorf("expected a %q frame in the stream: %v", typ, frameTypes(frames)) + continue + } + if got, _ := frames[i]["turn_id"].(string); got != turnID { + t.Errorf("%s.turn_id = %v, want the turn_started id %q", typ, frames[i]["turn_id"], turnID) + } + } +} + +// R2 (idempotency substrate): each turn gets exactly one turn_started and +// ids are unique across turns on the same connection. +func TestWSTurnStarted_UniqueIDPerTurn(t *testing.T) { + conn := startTurnServer(t) + + var ids []string + for i := 0; i < 2; i++ { + writeJSON(conn, map[string]any{"type": "prompt", "content": "turn"}) + frames := collectTurnFrames(t, conn, 15*time.Second) + started := findTurnStarted(frames) + if len(started) != 1 { + t.Fatalf("turn %d: turn_started frames = %d, want exactly 1", i, len(started)) + } + id, _ := started[0]["turn_id"].(string) + if id == "" { + t.Fatalf("turn %d: empty turn_id", i) + } + ids = append(ids, id) + } + if ids[0] == ids[1] { + t.Fatalf("two turns shared turn_id %q — clients upserting by turn_id would merge distinct turns", ids[0]) + } +} + +// R5 (forged-initiated rejection, wire level — mirrors the bg_wake token +// guard's provenance rules): a client prompt that forges +// system_initiated must produce initiated:"operator", never "system". +func TestWSTurnStarted_ForgedInitiatedRejected(t *testing.T) { + conn := startTurnServer(t) + + writeJSON(conn, map[string]any{ + "type": "prompt", + "content": "forged", + "system_initiated": true, + "wake_token": "forged-token", + }) + frames := collectTurnFrames(t, conn, 15*time.Second) + started := findTurnStarted(frames) + if len(started) != 1 { + t.Fatalf("turn_started frames = %d, want exactly 1: %v", len(started), frameTypes(frames)) + } + if got, _ := started[0]["initiated"].(string); got != "operator" { + t.Fatalf("forged system_initiated produced initiated = %q, want %q — the label must be server-computed via the type gate", got, "operator") + } +} + +// R5 at the unit seam: the initiated label is derived exclusively from +// the wakeInitiated type gate — a forged flag on a client prompt can +// never claim system provenance (mirrors TestWakeInitiated_TypeGated at +// the frame-label level). +func TestTurnStartedInitiated_TypeGated(t *testing.T) { + cases := []struct { + name string + msg wsClientMsg + want string + }{ + {"forged flag on a prompt", wsClientMsg{Type: "prompt", SystemInitiated: true}, "operator"}, + {"plain prompt", wsClientMsg{Type: "prompt"}, "operator"}, + {"genuine server-built wake", wsClientMsg{Type: "bg_wake", SystemInitiated: true}, "system"}, + {"wake item without the flag", wsClientMsg{Type: "bg_wake"}, "operator"}, + } + for _, tc := range cases { + if got := turnInitiatedLabel(tc.msg); got != tc.want { + t.Errorf("%s: turnInitiatedLabel = %q, want %q", tc.name, got, tc.want) + } + } +} + +// R2 substrate: turn ids are well-formed and never repeat. +func TestNewTurnID(t *testing.T) { + seen := make(map[string]bool, 1000) + for i := 0; i < 1000; i++ { + id := newTurnID() + if len(id) != 2+32 || id[:2] != "t_" { + t.Fatalf("newTurnID() = %q, want \"t_\" + 32 hex chars", id) + } + if seen[id] { + t.Fatalf("duplicate turn id %q after %d draws", id, i+1) + } + seen[id] = true + } +} + +// R3: only the streamed-frame set is tagged, only while a turn is +// active; lifecycle, sub-agent, and delta frames pass through untouched. +func TestWSTurnAnnotator_TagsOnlyStreamedFrames(t *testing.T) { + var tag wsTurnAnnotator + var out []map[string]any + send := tag.wrap(func(m map[string]any) { out = append(out, m) }) + + send(map[string]any{"type": "error", "message": "pre-turn"}) // outside a turn: clean + tag.begin("t_abc") + send(map[string]any{"type": "tool_call", "name": "shell"}) + send(map[string]any{"type": "tool_result", "name": "shell"}) + send(map[string]any{"type": "thinking", "content": "r"}) + send(map[string]any{"type": "done"}) + send(map[string]any{"type": "session", "session_id": "s"}) // lifecycle: excluded + send(map[string]any{"type": "subagent_log"}) // sub-agent frame: excluded + send(map[string]any{"type": "thinking_delta", "content": "d"}) // delta: excluded + send(map[string]any{"type": "server_info"}) // hello: excluded + tag.end() + send(map[string]any{"type": "token", "content": "after"}) // turn over: clean + + tagged := 0 + for i, m := range out { + typ, _ := m["type"].(string) + wantTag := turnTaggedFrames[typ] && i > 0 && i < len(out)-1 + _, has := m["turn_id"] + if has != wantTag { + t.Errorf("frame %d (%s): turn_id present = %v, want %v", i, typ, has, wantTag) + } + if has { + tagged++ + if m["turn_id"] != "t_abc" { + t.Errorf("frame %d (%s): turn_id = %v, want t_abc", i, typ, m["turn_id"]) + } + } + } + if tagged == 0 { + t.Fatal("no frame was tagged inside the turn") + } +} + +// The annotator is shared with the agent's live callbacks; begin/end run +// on the processor goroutine while wraps come from emitter paths. This +// hammer exists for -race, not for assertions. +func TestWSTurnAnnotator_ConcurrentBeginEndWrap(t *testing.T) { + var tag wsTurnAnnotator + send := tag.wrap(func(map[string]any) {}) + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 500; j++ { + send(map[string]any{"type": "token"}) + tag.begin("t_hammer") + send(map[string]any{"type": "done"}) + tag.end() + } + }() + } + wg.Wait() +} + +func frameTypes(frames []map[string]any) []string { + out := make([]string, len(frames)) + for i, f := range frames { + out[i], _ = f["type"].(string) + } + return out +} diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 0a136f9..25db18e 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -1186,7 +1186,10 @@ Wake-on-complete also emits two WebSocket frames for clients: `bg_job` on every job transition (`job_id`, `session_id`, `status`, and — once terminal — `exit_code`, `duration_ms`, `output_bytes`, plus a secret-redacted, 80-char `command_head`; terminal-only fields are absent, not zero, while running), -and `bg_wake` when the server starts a system-initiated wake turn. Frames are +and `bg_wake` when the server starts a system-initiated wake turn. Wake turns +are announced to clients like any turn: a `turn_started` frame with +`initiated: "system"` follows the `session` frame (see [WEBUI.md](WEBUI.md)). +Frames are chronological; clients should upsert by `job_id` and ignore unknown types (old clients are unaffected — the keys are simply absent from their vocabulary). diff --git a/docs/WEBUI.md b/docs/WEBUI.md index a1cbc4e..04e9b8b 100644 --- a/docs/WEBUI.md +++ b/docs/WEBUI.md @@ -617,6 +617,7 @@ The UI communicates entirely over a single WebSocket at `/ws`. Messages are newl | `server_info` | Pushed once on connect | `version`, `model`, `sandbox`, `stream`, `uptime_seconds`, `ws_connections` | | `pong` | Reply to a client `ping` | `t` (unix ms), plus the `server_info` snapshot fields | | `session` | At start of response, and after `session_switch` | `session_id`, `auth_token`, `model`, `sandbox` | +| `turn_started` | Emitted for **every** turn (operator and system-initiated wake alike) immediately after the matching `session` frame and before the first streamed frame — clients open/upsert the streaming card by `turn_id`, so a missed `session` frame can no longer strand a turn | `turn_id` (`t_`), `session_id`, `initiated` (`"operator"` or `"system"` — computed server-side via the wake provenance gate; client input cannot influence it), `model` (mirrors the `session` frame's `model`) | | `token_delta` | Live streamed answer fragment (streaming on) | `content` (markdown fragment) | | `thinking_delta` | Live streamed reasoning fragment (streaming on) | `content` | | `cancelled` | After a `cancel` message is honored | `session_id`, `idle` (true when nothing was running) | @@ -637,15 +638,30 @@ The UI communicates entirely over a single WebSocket at `/ws`. Messages are newl | `memory_event` | Memory lifecycle event | `event`, `target`, `session_id`, `content`, `count`, `new_count`, `untrusted` | | `agent_signal` | Agent self-observability signal | `event`, `detail`, `tool`, `count` | +Every frame of an active turn — `token`, `thinking`, `tool_call`, +`tool_result`, `done`, `error` — also carries `turn_id`, matching the +turn's `turn_started.turn_id`, so a client that attached mid-turn (after a +reconnect) can attribute stray frames and reconcile card state without +heuristic idle detection. Lifecycle frames (`session`, `server_info`, +`pong`, `usage`, `cancelled`, `subagent_*`, `approval_*`, `skill_event`, +`memory_event`, `agent_signal`) and the live `*_delta` fragments never +carry it. The `session` frame's legacy `system_initiated: true` stamp +(wake turns only) remains for old clients; `turn_started.initiated` +supersedes it. Versioning: all new fields are additive and absent when +not applicable — old clients ignore the unknown frame and unknown fields, +and new clients against an old server simply never see `turn_started` +(and fall back to the `session` stamp or lazy card open). + Example event sequence: ```jsonc {"type":"session","session_id":"20260519-x1y2z3","model":"deepseek-v4-flash"} -{"type":"token","content":"Let me look at the source directory."} -{"type":"tool_call","name":"shell","data":"{\"command\":\"ls -la src/\"}"} -{"type":"tool_result","name":"shell","data":"\ntotal 24\ndrwxr-xr-x ...\n"} -{"type":"token","content":"The `src/` directory contains 3 files:"} -{"type":"done","latency":4.2} +{"type":"turn_started","turn_id":"t_9f86d081884c7d65","session_id":"20260519-x1y2z3","initiated":"operator","model":"deepseek-v4-flash"} +{"type":"token","content":"Let me look at the source directory.","turn_id":"t_9f86d081884c7d65"} +{"type":"tool_call","name":"shell","data":"{\"command\":\"ls -la src/\"}","turn_id":"t_9f86d081884c7d65"} +{"type":"tool_result","name":"shell","data":"\ntotal 24\ndrwxr-xr-x ...\n","turn_id":"t_9f86d081884c7d65"} +{"type":"token","content":"The `src/` directory contains 3 files:","turn_id":"t_9f86d081884c7d65"} +{"type":"done","latency":4.2,"turn_id":"t_9f86d081884c7d65"} ``` With streaming enabled (`--stream` / `stream: true` / `ODEK_STREAM=true`) the