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();