From a4972cd45655bde4b87e6b83094a4fc78c78cc2b Mon Sep 17 00:00:00 2001 From: Ollie Agent Date: Sat, 10 Oct 2026 15:57:48 +0200 Subject: [PATCH] Separate chat persistence from live delivery via chat.raw Streaming partials were appended to log.raw per chunk, each carrying the full cumulative content, so one response left dozens of partial lines persisted. GUI clients parsing the snapshot re-rendered the growing block once per partial (O(N^2)) and replayed all historical partials on every reconnect, causing intermittent rendering loops. Partials are no longer persisted: AppendBlock writes only finalized blocks to log.raw; SetPartial broadcasts the in-flight block without storing it. New chat.raw StreamRaw file is the authoritative live JSONL source (finalized history replay, then live deltas); log.raw is a finalized-only one-shot snapshot. GUI streams chat.raw, never polls log.raw. --- AGENTS.md | 5 +- cmd/olliesrv/internal/agent/agent.go | 4 + cmd/olliesrv/internal/agent/chat.go | 83 ++++++++++++++++++- cmd/olliesrv/internal/agent/chatlog.go | 2 +- cmd/olliesrv/internal/fs/spec.go | 7 ++ .../internal/prompts/system_prompt.md | 3 +- doc/architecture-9p.md | 3 +- doc/evolution.md | 28 ++++++- kde/gui/ollie9pclient.cpp | 2 +- 9 files changed, 128 insertions(+), 9 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index e0d8a0b..c8fd0ee 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -121,8 +121,9 @@ This is a standing preference, not a per-task instruction. Apply it without aski 11. **Peers**: Persistent agents in the same session can be linked via `peeradd`. Links are bidirectional. Agents communicate by writing to `peer/{name}`, which delivers to the target's prompt handler. Only declared peers can be messaged — the `peer/` directory is the access control surface. 12. **9P namespace declaration**: `cmd/olliesrv/internal/fs/spec.go` declares the olliesrv namespace. The toolsrv namespace is declared by `cmd/toolsrv/p9.go` using `cmd/toolsrv/internal/server.Spec`; process state and handlers are in `cmd/toolsrv/internal/server/`. Both use the `virtfs` EDSL and `virtfs.BuildTree()`. 13. **Working directory (default + override)**: Session cwd is **required** at session creation and is the inheritance root for every agent — the sane default so many agents can work in one directory with zero per-agent config. Agent cwd is an **optional** per-agent override (empty = inherit the session cwd). Because the per-session `toolsrv` sets `cmd.Dir` per tool call, per-agent cwd needs no extra process: `toolsrv` keeps an in-memory `agentCWD` map resolved per call from the `agent=` field (`cmd/toolsrv/internal/server/proc.go`), falling back to the session-level global cwd. The override is set only over the controlled `olliesrv`→`toolsrv` ctl channel (`agentcwd `), never embedded in a tool-call payload, so the model cannot influence where its own tools run. The map is process-local, so `Agent.SyncCwdToToolServer` re-pushes the override on every (re)connect — profile switch, resume, restore — matching how env and tools are resynced. Set/clear via the agent `cfg` (`cwd=...`) or ctl (`cwd [|-]`); the GUI Agent Settings dialog shows `(inherit: )` as the placeholder. -14. **Chat log files**: Each agent exposes four views of its conversation history via the 9P namespace: - - `log.raw` — JSONL stream (one JSON object per line). Blocking; suitable for GUI live tailing. Each line is `{"role":"...","id":"...","content":"..."}` with optional `name`, `format`, and `partial` fields. Partial blocks (`"partial":true`) represent in-progress streaming; they are replaced by the final block when complete. +14. **Chat log files**: Each agent exposes views of its conversation history via the 9P namespace: + - `chat.raw` — authoritative live JSONL stream. On open it replays the finalized history (one line per block), then streams live deltas: the current in-flight partial block whenever it changes, and finalized blocks as they are appended. GUI clients consume this and must not poll `log.raw`. Partials are delivered out-of-band (never persisted) and collapse by block ID on the client, so reconnects stay O(history), not O(streaming chunks). + - `log.raw` — JSONL snapshot (one JSON object per line) of **finalized** blocks only. One-shot read: returns the full history and EOFs. Used for explicit history loads (initial populate, bookmark reload), never polled for live updates. Each line is `{"role":"...","id":"...","content":"..."}` with optional `name` and `format` fields. - `log` — Rendered text snapshot (last 64KB). Non-blocking; suitable for one-shot inspection. Filters out reasoning, call, and tool blocks; shows only user/assistant/context content. - `chat` — Rendered text stream. Blocking; suitable for TUI live tailing. Same filtering as `log`. - `block` — Rdwr lookup: write a block ID (8-char hex), read the matching block as JSON. Returns the full block object or an error if not found. diff --git a/cmd/olliesrv/internal/agent/agent.go b/cmd/olliesrv/internal/agent/agent.go index 9fc3f59..5f58622 100644 --- a/cmd/olliesrv/internal/agent/agent.go +++ b/cmd/olliesrv/internal/agent/agent.go @@ -78,6 +78,8 @@ type Agent struct { chatCond *sync.Cond // signaled on new chat data chatSignalMu sync.Mutex chatSignalCh chan struct{} // closed on new chat data; replaced with fresh channel + partialLine []byte // current in-flight (partial) block as a JSONL line; never persisted to rawLog + partialVers uint64 // bumped whenever partialLine changes, for stream change detection plan []byte // agent's plan file contents blockCounter uint64 // monotonic counter for block ID generation } @@ -319,6 +321,8 @@ func (ag *Agent) Clear() error { ag.rawStart = 0 ag.textLog = nil ag.textStart = 0 + ag.partialLine = nil + ag.partialVers++ ag.chatVers++ ag.chatMu.Unlock() ag.chatCond.Broadcast() diff --git a/cmd/olliesrv/internal/agent/chat.go b/cmd/olliesrv/internal/agent/chat.go index c75969b..f9b8516 100644 --- a/cmd/olliesrv/internal/agent/chat.go +++ b/cmd/olliesrv/internal/agent/chat.go @@ -10,6 +10,7 @@ package agent import ( "bytes" + "context" "crypto/sha256" "encoding/hex" "fmt" @@ -34,7 +35,8 @@ func (ag *Agent) NextBlockID() string { return hex.EncodeToString(h.Sum(nil))[:8] } -// AppendBlock writes a block to both logs and notifies stream readers. +// AppendBlock writes a finalized block to both logs and notifies stream readers. +// Partial (in-flight) blocks are never persisted here; see SetPartial. func (ag *Agent) AppendBlock(b format.Block) { raw := format.MarshalBlock(b) text := []byte(format.RenderBlock(b)) @@ -42,9 +44,29 @@ func (ag *Agent) AppendBlock(b format.Block) { ag.chatMu.Lock() ag.rawLog = appendWithOverflow(ag.rawLog, raw, maxLogBytes, &ag.rawStart) ag.textLog = appendWithOverflow(ag.textLog, text, maxLogBytes, &ag.textStart) + // A finalized block supersedes any in-flight partial for the same block. + ag.partialLine = nil + ag.partialVers++ ag.chatVers++ ag.chatMu.Unlock() + ag.signalChat() +} + +// SetPartial records the current in-flight (streaming) block as a JSONL line. +// It is delivered to live stream readers but never appended to rawLog, so the +// one-shot log.raw snapshot never accumulates per-chunk partial duplicates. +func (ag *Agent) SetPartial(b format.Block) { + line := format.MarshalBlock(b) + ag.chatMu.Lock() + ag.partialLine = line + ag.partialVers++ + ag.chatMu.Unlock() + ag.signalChat() +} + +// signalChat wakes blocking chat/stream readers. +func (ag *Agent) signalChat() { ag.chatCond.Broadcast() ag.chatSignalMu.Lock() close(ag.chatSignalCh) @@ -87,6 +109,65 @@ func (ag *Agent) RawLogRead(base string) ([]byte, string, error) { return ag.logRead(base, &ag.rawLog, &ag.rawStart) } +// RawLogStream is the authoritative live JSONL source for GUI clients. +// +// On a fresh open (base == "") it replays the finalized history (rawLog, one +// line per block, no partials), then streams live deltas: the current in-flight +// partial block whenever it changes, and finalized blocks as they are appended. +// rawLog is append-only and never rewritten, so finals are delivered with +// offset semantics; partials are delivered out-of-band and collapse by block ID +// on the client. This stream never replays accumulated partials, so reconnects +// stay O(history) rather than O(partial chunks). +// +// base encodes ":". +func (ag *Agent) RawLogStream(ctx context.Context, base string) ([]byte, string, error) { + var off int + var pv uint64 + if base != "" { + fmt.Sscanf(base, "%d:%d", &off, &pv) + } + for { + ag.chatMu.RLock() + if base == "" { + off = ag.rawStart + } + if off < ag.rawStart { + off = ag.rawStart + } + rawTail := ag.rawStart + len(ag.rawLog) + // 1) Deliver any un-sent finalized blocks first (offset-safe). + if off < rawTail { + localOff := off - ag.rawStart + data := make([]byte, len(ag.rawLog)-localOff) + copy(data, ag.rawLog[localOff:]) + nextPv := ag.partialVers + ag.chatMu.RUnlock() + return data, fmt.Sprintf("%d:%d", rawTail, nextPv), nil + } + // 2) Caught up to finals: deliver the current partial if it changed. + curPv := ag.partialVers + if curPv != pv && len(ag.partialLine) > 0 { + data := make([]byte, len(ag.partialLine)) + copy(data, ag.partialLine) + ag.chatMu.RUnlock() + return data, fmt.Sprintf("%d:%d", rawTail, curPv), nil + } + // 3) Nothing new; sync base and block until signaled. + pv = curPv + base = "sync" + ag.chatMu.RUnlock() + + ag.chatSignalMu.Lock() + sig := ag.chatSignalCh + ag.chatSignalMu.Unlock() + select { + case <-sig: + case <-ctx.Done(): + return nil, fmt.Sprintf("%d:%d", off, pv), nil + } + } +} + // TextLogRead returns new rendered text since the given offset. func (ag *Agent) TextLogRead(base string) ([]byte, string, error) { return ag.logRead(base, &ag.textLog, &ag.textStart) diff --git a/cmd/olliesrv/internal/agent/chatlog.go b/cmd/olliesrv/internal/agent/chatlog.go index 9e6e6e4..a8fdb0e 100644 --- a/cmd/olliesrv/internal/agent/chatlog.go +++ b/cmd/olliesrv/internal/agent/chatlog.go @@ -26,7 +26,7 @@ func (ag *Agent) initChatHandler() { if streamingID == "" { return } - ag.AppendBlock(format.Block{ + ag.SetPartial(format.Block{ Role: streamingRole, ID: streamingID, Content: streamingContent.String(), diff --git a/cmd/olliesrv/internal/fs/spec.go b/cmd/olliesrv/internal/fs/spec.go index dfd1f33..c448544 100644 --- a/cmd/olliesrv/internal/fs/spec.go +++ b/cmd/olliesrv/internal/fs/spec.go @@ -852,6 +852,13 @@ func buildAgentChildren(a *agent.Agent, s *session.Session) []virtfs.FsNodeDecl return data, nil }), ), + // chat.raw — authoritative live JSONL stream (one Block per line). + // Replays finalized history on open, then streams live partials and + // finalized blocks. GUI clients consume this; they never poll log.raw. + virtfs.FileNode("chat.raw", 0440, + virtfs.Doc("JSONL chat stream. Finalized history then live deltas. Blocks for new data."), + virtfs.StreamRaw(a.RawLogStream), + ), // chat — Rendered plain text stream for TUI (blocking) virtfs.FileNode("chat", 0440, virtfs.Doc("Rendered chat log stream (plain text). Blocks for new data."), diff --git a/cmd/olliesrv/internal/prompts/system_prompt.md b/cmd/olliesrv/internal/prompts/system_prompt.md index 5334156..b142275 100644 --- a/cmd/olliesrv/internal/prompts/system_prompt.md +++ b/cmd/olliesrv/internal/prompts/system_prompt.md @@ -206,7 +206,8 @@ Use `client_9p` for all 9P operations. Your session ID is `${OLLIE_SESSION_ID}`. | `prompt` | write | Submit a prompt to this agent | | `fifo` | r/w | Prompt queue. Write: enqueue. Read: dequeue. | | `log` | read | Conversation snapshot (filtered text, last 64KB) | -| `log.raw` | read | Full conversation as JSONL stream | +| `log.raw` | read | Full conversation as JSONL snapshot of finalized blocks (one-shot read) | +| `chat.raw` | read | Live JSONL stream: finalized history then live deltas (blocking) | | `chat` | read | Conversation stream (filtered text, blocking) | | `block` | rdwr | Lookup block by ID (write ID, read JSON) | | `state` | read | Current agent state (idle, calling, thinking, paused) | diff --git a/doc/architecture-9p.md b/doc/architecture-9p.md index efe1100..d49d40c 100644 --- a/doc/architecture-9p.md +++ b/doc/architecture-9p.md @@ -88,7 +88,8 @@ echo "name=coding cwd=$PWD" | ollie-9p write session/myproj/agent/new | `agent/{aname}/prompt` | w | Queue a user turn. | | `agent/{aname}/fifo` | r/w | Prompt queue. | | `agent/{aname}/log` | r | Filtered text conversation snapshot (last 64KB). | -| `agent/{aname}/log.raw` | r | Full JSONL streaming output. | +| `agent/{aname}/log.raw` | r | Full JSONL conversation snapshot of finalized blocks (one-shot read). | +| `agent/{aname}/chat.raw` | r | Live JSONL stream: finalized history then live deltas (blocking). | | `agent/{aname}/chat` | r | Filtered text conversation stream (blocking). | | `agent/{aname}/block` | rdwr | Lookup block by ID (write ID, read JSON). | | `agent/{aname}/state` | r | Current agent state (idle, calling, thinking, paused). | diff --git a/doc/evolution.md b/doc/evolution.md index cb64d28..7869f86 100644 --- a/doc/evolution.md +++ b/doc/evolution.md @@ -2297,7 +2297,7 @@ Each line is a complete JSON object with: **agent/chatlog.go**: Event handler emits JSONL blocks via `flushPartial` (partial=true per chunk) and `closeBlock` (partial=false final). Skips internal events: `state`, `usage`, `limitretry`. **fs/spec.go**: New 9P files: -- `log.raw` — JSONL stream (blocking, for GUI) +- `log.raw` — JSONL snapshot (one-shot read, for GUI startup) - `log` — Rendered text snapshot (non-blocking, last 64KB) - `chat` — Rendered text stream (blocking, for TUI) - `block` — Rdwr lookup: write block ID, read JSON @@ -2308,7 +2308,7 @@ Deleted: `chat.raw`, `chat.search` **chatblockmodel.cpp**: Complete rewrite. Uses `QJsonDocument` for parsing instead of regex state machine. Filters `context` role from display. Removed `m_inContext` tracking, `renderBlock` method, and regex patterns. -**ollie9pclient.cpp**: `readLogForSession` reads from `/log.raw` (JSONL) instead of `/log` (plain text). Streaming via `startActiveAgentStreams` uses `/log.raw`. +**ollie9pclient.cpp**: `readLogForSession` reads from `/log.raw` (JSONL) instead of `/log` (plain text). `log.raw` is a one-shot snapshot read, not a stream; live updates arrive via the separate `event` stream. **chatblockmodel.h**: Added `Context` to `ChatBlock::Type` enum. Removed unused members. @@ -2335,3 +2335,27 @@ ollie9pclient.cpp +5 read from log.raw ``` Net: **-680 lines** — simpler, more robust, standard format. + +## Phase 41: Authoritative Live Chat Stream (Oct 10) + +Phase 40's JSONL `log.raw` doubled as both the persistent snapshot and the GUI's live stream. Because the JSONL buffer is append-only, `flushPartial` appended a *new* line for every streaming chunk — each carrying the full cumulative content — so a single assistant response left dozens of partial lines permanently in `log.raw`. The GUI, parsing the snapshot keyed by block ID, re-rendered the growing block once per partial line (O(N²) per response), and every reconnect replayed all historical partials. This produced the intermittent GUI rendering loops. It was fundamentally a data-input problem: partials were persisted, not just streamed. + +### Fix: separate persistence from live delivery + +- **Partials are no longer persisted.** `AppendBlock` writes only finalized blocks to `rawLog`/`textLog`. A new `SetPartial` records the current in-flight block in a single `partialLine` field (bumping `partialVers`) and broadcasts it to stream readers — it never touches `rawLog`. `chatlog.go`'s `flushPartial` now calls `SetPartial`; `closeBlock` still calls `AppendBlock`, which clears the partial. +- **`log.raw` is a finalized-only one-shot snapshot.** One line per logical block, bounded, no duplication. Used for explicit history loads (initial populate, bookmark reload), never polled. +- **`chat.raw` is the authoritative live JSONL stream** (`Agent.RawLogStream`, `StreamRaw`). On open it replays `rawLog` (finals) from the start, then streams live deltas: newly finalized blocks (offset-safe, since `rawLog` is append-only and never rewritten) and the current partial whenever it changes (out-of-band, keyed on `partialVers`). The GUI collapses by block ID, so reconnects are O(history), not O(streaming chunks). Base encodes `:`. + +### GUI + +`ollie9pclient.cpp`: `startActiveAgentStreams` streams `/chat.raw` instead of `/log.raw`. The one-shot `readLogForSession` still reads `/log.raw` for `loadChat` and bookmark reloads — `log.raw` is now purely a snapshot, never a live source. + +### Source changes + +```text +agent/agent.go +4 partialLine/partialVers fields; cleared on Clear +agent/chat.go +60 SetPartial, signalChat, RawLogStream; AppendBlock finals-only +agent/chatlog.go ±1 flushPartial -> SetPartial +fs/spec.go +7 chat.raw StreamRaw file +ollie9pclient.cpp ±1 stream chat.raw, not log.raw +``` diff --git a/kde/gui/ollie9pclient.cpp b/kde/gui/ollie9pclient.cpp index 0434634..4ef8d14 100644 --- a/kde/gui/ollie9pclient.cpp +++ b/kde/gui/ollie9pclient.cpp @@ -1132,7 +1132,7 @@ void Ollie9pClient::startActiveAgentStreams() if (!m_daemonConnected) return; if (m_activeSessionId.isEmpty() || m_agentId.isEmpty()) return; - m_chat->start(agentPath() + "/log.raw"); + m_chat->start(agentPath() + "/chat.raw"); // Check for pending bypass request on this session checkPendingBypass();