diff --git a/source/cydo/domain/tasks/model.d b/source/cydo/domain/tasks/model.d index 4caacf27..a845e839 100644 --- a/source/cydo/domain/tasks/model.d +++ b/source/cydo/domain/tasks/model.d @@ -975,6 +975,25 @@ struct TaskHistoryStartMessage string type = "task_history_start"; int tid; int total; + int window_start; // first replayed seq; > 0 means older history exists unsent + int window_limit; // window applied, in messages; 0 = the whole history +} + +/// Frames a replay of the messages *older* than what the client already holds, +/// so it can prepend them instead of rebuilding its list. +struct TaskHistoryPrependStartMessage +{ + string type = "task_history_prepend_start"; + int tid; + int window_start; // first seq in this batch; 0 means the task's beginning + int before_seq; // where the batch stops, i.e. the client's current start +} + +struct TaskHistoryPrependEndMessage +{ + string type = "task_history_prepend_end"; + int tid; + int window_start; } struct TaskHistoryEndMessage @@ -1031,6 +1050,14 @@ struct WsMessage @JSONOptional Nullable!uint expected_num_turns; string correlation_id; string tool_use_id; + // request_history: 0 lets the server pick from its config for device_class, + // >0 is an explicit window, -1 asks for the whole history. the client is not + // asked to know the configured window, because it learns that from + // server_status, which arrives after the tasks list that triggers the first + // history request + @JSONOptional int limit; + @JSONOptional string device_class; // "mobile" or "desktop" + @JSONOptional int before_seq; // request_history_before: older than this } struct TaskCreatedMessage @@ -1168,6 +1195,8 @@ struct ServerStatusMessage bool auth_enabled; bool dev_mode; string build_id; + int history_window_desktop; // initial replay window in messages, 0 = full + int history_window_mobile; } struct ScanStatusMessage diff --git a/source/cydo/runtime/config/package.d b/source/cydo/runtime/config/package.d index d2c21393..fefaf9a2 100644 --- a/source/cydo/runtime/config/package.d +++ b/source/cydo/runtime/config/package.d @@ -104,6 +104,12 @@ struct WorkspaceConfig @Optional ProjectDiscoveryConfig project_discovery; } +struct HistoryWindowConfig +{ + @Optional int desktop; + @Optional int mobile; +} + struct CydoConfig { @Key("name") WorkspaceConfig[] workspaces; @@ -115,6 +121,10 @@ struct CydoConfig @Optional bool dev_mode; @Optional string log_level = "info"; @Optional string system_keyword = "SYSTEM"; + /// How much of a task's history to replay when it is opened, in messages + /// (user and assistant bubbles), per device class. Absent or zero replays + /// everything, which is the default. + @Optional HistoryWindowConfig history_window; /// Called by configy during parsing (configy/read.d:650), so a semantic /// error surfaces on the same path as a YAML syntax error. diff --git a/source/cydo/server/app.d b/source/cydo/server/app.d index c127f5ba..bfa0876f 100644 --- a/source/cydo/server/app.d +++ b/source/cydo/server/app.d @@ -1099,6 +1099,8 @@ class App authUser.length > 0 || authPass.length > 0, config.dev_mode, webDistDir, + config.history_window.desktop, + config.history_window.mobile, ).representation)); ws.send(Data(buildNoticesList(activeNotices).representation)); if (discoveryService.scanInProgress) @@ -1265,6 +1267,7 @@ class App { case "create_task": handleCreateTaskMsg(ws, json); break; case "request_history": handleRequestHistory(ws, json); break; + case "request_history_before": handleRequestHistoryBefore(ws, json); break; case "message": handleUserMessage(json); break; case "resume": handleResumeMsg(json); break; case "interrupt": handleInterruptMsg(json); break; @@ -1460,7 +1463,30 @@ class App private void handleRequestHistory(WebSocketAdapter ws, WsMessage json) { - historyPipeline.handleRequestHistory(ws, json.tid); + historyPipeline.handleRequestHistory(ws, json.tid, + resolveHistoryLimit(json.limit, json.device_class)); + } + + private void handleRequestHistoryBefore(WebSocketAdapter ws, WsMessage json) + { + historyPipeline.handleRequestHistoryBefore(ws, json.tid, json.before_seq, + resolveHistoryLimit(json.limit, json.device_class)); + } + + /// Turn a client's request into an actual window size. + /// + /// The server owns the numbers because it is the only side that always + /// knows them: a client that has not yet processed server_status would + /// otherwise ask for everything, which on a long task means replaying tens + /// of thousands of events. + private int resolveHistoryLimit(int requested, string deviceClass) + { + if (requested != 0) + return requested < 0 ? 0 : requested; // negative asks for it all + auto window = deviceClass == "mobile" + ? config.history_window.mobile + : config.history_window.desktop; + return window > 0 ? window : 0; } private void sendHistoryReplaySupplementalState(WebSocketAdapter ws, int tid) @@ -3395,6 +3421,8 @@ class App authUser.length > 0 || authPass.length > 0, config.dev_mode, webDistDir, + config.history_window.desktop, + config.history_window.mobile, )); infof("Config reloaded successfully"); discoveryService.endScan(); diff --git a/source/cydo/web/snapshots.d b/source/cydo/web/snapshots.d index ee58fadb..c81b6e4c 100644 --- a/source/cydo/web/snapshots.d +++ b/source/cydo/web/snapshots.d @@ -90,13 +90,16 @@ string readBuildId(string webDistDir) return m[1].idup; } -string buildServerStatus(bool authEnabled, bool devMode, string webDistDir) +string buildServerStatus(bool authEnabled, bool devMode, string webDistDir, + int historyWindowDesktop = 0, int historyWindowMobile = 0) { return toJson(ServerStatusMessage( "server_status", authEnabled, devMode, readBuildId(webDistDir), + historyWindowDesktop, + historyWindowMobile, )); } diff --git a/source/cydo/workflow/history/pipeline.d b/source/cydo/workflow/history/pipeline.d index 35a77206..ebd3ab18 100644 --- a/source/cydo/workflow/history/pipeline.d +++ b/source/cydo/workflow/history/pipeline.d @@ -23,6 +23,7 @@ import cydo.protocol : ContentBlock, ItemStartedEvent, TaskEventEnvelope, import cydo.runtime.config : AgentDriver; import cydo.domain.storage.persistence : LoadedHistory; import cydo.domain.tasks.model : QueueOperationProbe, TaskData, TaskHistoryEndMessage, + TaskHistoryPrependEndMessage, TaskHistoryPrependStartMessage, TaskHistoryStartMessage, Watermark, buildSyntheticUserEvent, extractEventFromEnvelope, extractTsFromEnvelope, watermarkFromPath; import cydo.workflow.history.native_history : HistoryAccess, @@ -392,7 +393,45 @@ class HistoryEventPipeline "replayed native history identity does not match its persisted boundary"); } - void handleRequestHistory(WebSocketAdapter ws, int tid) + /// First seq of a replay window holding exactly messageLimit rendered + /// message bubbles (user and assistant text), counted from the end. The + /// window may start mid-turn; events whose base item started before the + /// window are dropped by the client's reducers and appear once older + /// history is loaded. Returns 0 when the history holds fewer messages. + size_t historyWindowStart(TaskData* td, int messageLimit, size_t end = size_t.max) + { + assert(messageLimit > 0, "history window scan requires a positive limit"); + if (end > td.history.length) + end = td.history.length; + @JSONPartial static struct WindowProbe + { + string type; @JSONOptional string item_type; @JSONOptional bool is_meta; + @JSONOptional bool is_synthetic; + @JSONOptional bool is_sidechain; @JSONOptional string parent_tool_use_id; + } + size_t bubbles = 0; + foreach_reverse (i; 0 .. end) + { + WindowProbe probe; + td.history[i].enter((scope const(ubyte)[] bytes) { + auto event = extractEventFromEnvelope(bytes.as!(char[])); + if (event.length > 0) + probe = jsonParse!WindowProbe(event); + }); + if (probe.type != "item/started" || probe.is_meta || probe.is_synthetic + || probe.is_sidechain || probe.parent_tool_use_id.length > 0) + continue; + if (probe.item_type == "user_message" || probe.item_type == "text") + { + bubbles++; + if (bubbles >= cast(size_t) messageLimit) + return i; + } + } + return 0; + } + + void handleRequestHistory(WebSocketAdapter ws, int tid, int messageLimit = 0) { if (tid < 0) return; @@ -401,30 +440,20 @@ class HistoryEventPipeline if (td is null) return; + auto windowStart = messageLimit > 0 ? historyWindowStart(td, messageLimit) : 0; ws.send(Data(toJson(TaskHistoryStartMessage("task_history_start", tid, - cast(int) td.history.length)).representation)); - - foreach (i, ref msg; td.history) - { - Data outgoing; - msg.enter((scope ubyte[] bytes) { - auto envelope = bytes.as!(char[]); - auto event = extractEventFromEnvelope(envelope); - if (event.length == 0) - return; - auto normalized = host_.normalizeKnownSystemMessageMeta(event.idup, tid); - auto clientEnvelope = toJson(TaskEventSeqEnvelope( - tid, - cast(int) i, - extractTsFromEnvelope(envelope), - JSONFragment(normalized))); - outgoing = Data(clientEnvelope.representation); - }); - if (outgoing.length > 0) - ws.send(outgoing); - else - ws.send(msg); - } + cast(int) td.history.length, cast(int) windowStart, messageLimit)).representation)); + + // a window must not sever session identity: the newest session/init + // (and any newer session/metadata) from before the window still + // define the session the visible turns belong to, so they replay + // ahead of the range with their true seqs; a codex transcript carries + // metadata mid-stream, and metadata without its init is a hard error + // in the client's reducers + foreach (contextIndex; sessionContextBefore(td, windowStart)) + sendHistoryRecordAt(ws, tid, td, contextIndex); + + sendHistoryRange(ws, tid, td, windowStart, td.history.length); if (host_.resolveTaskHistory(tid).kind == TaskHistoryResolutionKind.access) host_.sendHistoryOperations(ws, tid); @@ -434,6 +463,115 @@ class HistoryEventPipeline host_.onHistorySubscribed(tid); } + /// Replay the messages immediately older than a window the client already + /// holds, so it can prepend them instead of rebuilding its list; growing + /// the window instead would replay everything the client already has. + void handleRequestHistoryBefore(WebSocketAdapter ws, int tid, int beforeSeq, int messageLimit) + { + if (tid < 0) + return; + ensureHistoryLoaded(tid); + auto td = host_.getTask(tid); + if (td is null) + return; + + size_t end = beforeSeq < 0 ? 0 : beforeSeq; + if (end > td.history.length) + end = td.history.length; + auto start = messageLimit > 0 ? historyWindowStart(td, messageLimit, end) : 0; + + ws.send(Data(toJson(TaskHistoryPrependStartMessage("task_history_prepend_start", + tid, cast(int) start, cast(int) end)).representation)); + sendHistoryRange(ws, tid, td, start, end); + ws.send(Data(toJson(TaskHistoryPrependEndMessage("task_history_prepend_end", + tid, cast(int) start)).representation)); + host_.subscribe(ws, tid); + host_.onHistorySubscribed(tid); + } + + private void sendHistoryRange(WebSocketAdapter ws, int tid, TaskData* td, + size_t start, size_t end) + { + // iterate by ref: indexing yields a const copy, which Data.enter and + // WebSocketAdapter.send both reject + foreach (i, ref msg; td.history) + { + if (i < start || i >= end) + continue; + emitHistoryRecord(ws, tid, msg, i); + } + } + + private void sendHistoryRecordAt(WebSocketAdapter ws, int tid, TaskData* td, + size_t index) + { + foreach (i, ref msg; td.history) + { + if (i != index) + continue; + emitHistoryRecord(ws, tid, msg, i); + return; + } + assert(false, "history record index out of range"); + } + + private void emitHistoryRecord(WebSocketAdapter ws, int tid, ref Data msg, size_t i) + { + Data outgoing; + msg.enter((scope ubyte[] bytes) { + auto envelope = bytes.as!(char[]); + auto event = extractEventFromEnvelope(envelope); + if (event.length == 0) + return; + auto normalized = host_.normalizeKnownSystemMessageMeta(event.idup, tid); + auto clientEnvelope = toJson(TaskEventSeqEnvelope( + tid, + cast(int) i, + extractTsFromEnvelope(envelope), + JSONFragment(normalized))); + outgoing = Data(clientEnvelope.representation); + }); + if (outgoing.length > 0) + ws.send(outgoing); + else + ws.send(msg); + } + + /// Indices of the newest session/init before `end`, followed by the + /// newest session/metadata after that init, when both exist. Empty when + /// the history before `end` establishes no session (claude transcripts + /// carry neither event). + private size_t[] sessionContextBefore(TaskData* td, size_t end) + { + if (end == 0 || end > td.history.length) + return null; + @JSONPartial static struct TypeProbe { string type; } + size_t metadataIndex = size_t.max; + size_t initIndex = size_t.max; + foreach_reverse (i; 0 .. end) + { + TypeProbe probe; + td.history[i].enter((scope const(ubyte)[] bytes) { + auto event = extractEventFromEnvelope(bytes.as!(char[])); + if (event.length > 0) + probe = jsonParse!TypeProbe(event); + }); + if (probe.type == "session/metadata" && metadataIndex == size_t.max) + metadataIndex = i; + if (probe.type == "session/init") + { + initIndex = i; + break; + } + } + if (initIndex == size_t.max) + return null; + if (metadataIndex == size_t.max) + return [initIndex]; + return [initIndex, metadataIndex]; + } + + void appendUnconfirmedUserMessage(int tid, const(ContentBlock)[] content, const(ContentBlock)[] broadcastContent = null, string cydoMeta = null, string nonce = null) @@ -1367,6 +1505,163 @@ unittest assert(replayedDiagnostic.severity == "error"); } +// Windowed replay: a message limit replays only the newest window, cut at a +// genuine (non-pending) user-message boundary so the client never reduces a +// split turn, with seqs left as true history indices. +unittest +{ + import ae.net.asockets : ConnectionState, DisconnectType, IConnection; + import ae.sys.dataset : joinData; + import ae.utils.array : as; + import ae.utils.json : jsonParse; + import cydo.domain.tasks.model : Watermark; + + enum tid = 74; + auto td = TaskData(tid, "local", "/tmp"); + td.history.reset(Watermark.none()); + class StubWebSocketAdapter : WebSocketAdapter + { + string[] sent; + + this() + { + super(new class IConnection + { + ConnectionState state_ = ConnectionState.connected; + void delegate(string, DisconnectType) disconnectHandler; + + @property ConnectionState state() { return state_; } + void send(scope Data[] data, int priority) {} + void disconnect(string reason, DisconnectType type) + { + state_ = ConnectionState.disconnected; + disconnectHandler(reason, type); + } + @property void handleConnect(void delegate() value) {} + @property void handleReadData(void delegate(Data) value) {} + @property void handleDisconnect(void delegate(string, DisconnectType) value) + { + disconnectHandler = value; + } + @property void handleBufferFlushed(void delegate() value) {} + }); + } + + override void send(scope Data[] data, int priority) + { + sent ~= cast(string) data.joinData().toGC().as!string; + } + } + HistoryEventPipelineHost host; + host.getTask = (int t) => t == tid ? &td : null; + host.resolveTaskHistory = (int t) => TaskHistoryResolution.noSession(); + host.injectAgentNameIntoSessionInit = (string translated, string agentName) => translated; + host.normalizeKnownSystemMessageMeta = (string translated, int t) => translated; + host.sendToSubscribed = (int t, Data data) {}; + host.subscribe = (WebSocketAdapter ws, int t) {}; + host.sendHistoryOperations = (WebSocketAdapter ws, int t) {}; + host.broadcastHistoryOperations = (int t) {}; + host.sendReplaySupplementalState = (WebSocketAdapter ws, int t) {}; + host.onHistorySubscribed = (int t) {}; + host.updateClaudeUsageFromEvent = (int t, string translated) => false; + host.planBroadcast = (int t, TranslatedEvent ev) => HistoryBroadcastPlan.init; + + auto pipeline = new HistoryEventPipeline(host); + // two full turns, then a pending queue record and a trailing assistant + // message: the pending user at seq 5 must not serve as a boundary + foreach (event; [ + `{"type":"item/started","item_type":"user_message","uuid":"u1"}`, // seq 0 + `{"type":"item/started","item_type":"text","item_id":"a1"}`, // seq 1 + `{"type":"item/started","item_type":"tool_use","item_id":"t1"}`, // seq 2 + `{"type":"item/started","item_type":"user_message","uuid":"u2"}`, // seq 3 + `{"type":"item/started","item_type":"text","item_id":"a2"}`, // seq 4 + `{"type":"item/started","item_type":"user_message","uuid":"q","pending":true}`, // seq 5 + `{"type":"item/started","item_type":"text","item_id":"a3"}`, // seq 6 + ]) + pipeline.appendAndBroadcastTaskEvent(tid, TranslatedEvent(event, null, AbsTime(1000))); + assert(td.history.length == 7); + + @JSONPartial static struct StartProbe + { + string type; int tid; int total; int window_start; int window_limit; + } + @JSONPartial static struct SeqProbe + { + int seq = -1; + } + + // no limit: the whole history replays, and says so + auto wsFull = new StubWebSocketAdapter(); + scope(exit) wsFull.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistory(wsFull, tid); + auto fullStart = jsonParse!StartProbe(wsFull.sent[0]); + assert(fullStart.total == 7 && fullStart.window_start == 0); + assert(fullStart.window_limit == 0, "a full replay reports no window"); + assert(wsFull.sent.length == 9, "full replay sends start, 7 events, and end"); + + // limit 2: exactly the two newest message bubbles, the text at seq 6 and + // the pending user at seq 5, regardless of whose turn they belong to + auto wsWindow = new StubWebSocketAdapter(); + scope(exit) wsWindow.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistory(wsWindow, tid, 2); + auto windowStart = jsonParse!StartProbe(wsWindow.sent[0]); + assert(windowStart.total == 7 && windowStart.window_start == 5); + assert(windowStart.window_limit == 2, "a windowed replay reports its limit"); + assert(wsWindow.sent.length == 4, "windowed replay sends start, 2 events, and end"); + assert(jsonParse!SeqProbe(wsWindow.sent[1]).seq == 5, + "windowed replay keeps true history indices"); + + // a limit beyond the history replays everything + auto wsBig = new StubWebSocketAdapter(); + scope(exit) wsBig.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistory(wsBig, tid, 100); + assert(jsonParse!StartProbe(wsBig.sent[0]).window_start == 0); + assert(wsBig.sent.length == 9); + + // tool events neither count toward the window nor shrink it: forty of + // them ride along with the three newest message bubbles + // loading older history replays only the slice before what the client holds, + // framed so it can prepend rather than rebuild + @JSONPartial static struct PrependProbe + { + string type; int tid; int window_start; int before_seq; + } + auto wsBefore = new StubWebSocketAdapter(); + scope(exit) wsBefore.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistoryBefore(wsBefore, tid, 3, 2); + auto prependStart = jsonParse!PrependProbe(wsBefore.sent[0]); + assert(prependStart.type == "task_history_prepend_start"); + assert(prependStart.before_seq == 3, "the batch stops where the client's window began"); + assert(prependStart.window_start == 0, "two messages back from seq 3 reaches the start"); + assert(jsonParse!PrependProbe(wsBefore.sent[$ - 1]).type == "task_history_prepend_end"); + assert(wsBefore.sent.length == 5, "only the older slice replays"); + assert(jsonParse!SeqProbe(wsBefore.sent[1]).seq == 0); + assert(jsonParse!SeqProbe(wsBefore.sent[3]).seq == 2); + + // asking for older history when nothing is held back replays nothing + auto wsNone = new StubWebSocketAdapter(); + scope(exit) wsNone.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistoryBefore(wsNone, tid, 0, 2); + assert(wsNone.sent.length == 2, "just the frames, no events"); + + // the event budget ends the window even when the message budget is nowhere + // near spent: a tool-heavy turn is what makes a page unusable, and a message + // count cannot see it coming + foreach (n; 0 .. 40) + pipeline.appendAndBroadcastTaskEvent(tid, TranslatedEvent( + `{"type":"item/started","item_type":"tool_use","item_id":"noise"}`, null, AbsTime(1000))); + pipeline.appendAndBroadcastTaskEvent(tid, TranslatedEvent( + `{"type":"item/started","item_type":"user_message","uuid":"u3"}`, null, AbsTime(1000))); + pipeline.appendAndBroadcastTaskEvent(tid, TranslatedEvent( + `{"type":"item/started","item_type":"text","item_id":"a4"}`, null, AbsTime(1000))); + + auto wsHeavy = new StubWebSocketAdapter(); + scope(exit) wsHeavy.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistory(wsHeavy, tid, 3); + assert(jsonParse!StartProbe(wsHeavy.sent[0]).window_start == 6, + "the third-newest message bubble starts the window, tool events ride along"); +} + unittest { import cydo.agent.contract : PersistedHistoryBoundaryKind; @@ -1541,3 +1836,101 @@ unittest assert(sawSteering, "steering confirmation missing"); assert(sawRemoved, "removed confirmation missing"); } + +unittest +{ + // windowing must not sever session identity: a codex transcript carries + // session/init once and session/metadata mid-stream, and a window that + // starts past them replays the newest of each ahead of the range with + // true seqs, so the client's init-before-metadata invariant holds + import ae.net.asockets : ConnectionState, DisconnectType, IConnection; + import ae.sys.dataset : joinData; + import ae.utils.json : jsonParse; + import std.algorithm : canFind; + import cydo.domain.tasks.model : Watermark; + + enum tid = 91; + auto td = TaskData(tid, "local", "/tmp/cydo-window-context"); + td.history.reset(Watermark.none()); + class StubWs : WebSocketAdapter + { + string[] sent; + + this() + { + super(new class IConnection + { + ConnectionState state_ = ConnectionState.connected; + void delegate(string, DisconnectType) disconnectHandler; + + @property ConnectionState state() { return state_; } + void send(scope Data[] data, int priority) {} + void disconnect(string reason, DisconnectType type) + { + state_ = ConnectionState.disconnected; + disconnectHandler(reason, type); + } + @property void handleConnect(void delegate() value) {} + @property void handleReadData(void delegate(Data) value) {} + @property void handleDisconnect(void delegate(string, DisconnectType) value) + { + disconnectHandler = value; + } + @property void handleBufferFlushed(void delegate() value) {} + }); + } + + override void send(scope Data[] data, int priority) + { + sent ~= cast(string) data.joinData().toGC().as!string; + } + } + HistoryEventPipelineHost host; + host.getTask = (int t) => t == tid ? &td : null; + host.resolveTaskHistory = (int t) => TaskHistoryResolution.noSession(); + host.injectAgentNameIntoSessionInit = (string translated, string agentName) => translated; + host.normalizeKnownSystemMessageMeta = (string translated, int t) => translated; + host.sendToSubscribed = (int t, Data data) {}; + host.subscribe = (WebSocketAdapter ws, int t) {}; + host.sendHistoryOperations = (WebSocketAdapter ws, int t) {}; + host.broadcastHistoryOperations = (int t) {}; + host.sendReplaySupplementalState = (WebSocketAdapter ws, int t) {}; + host.onHistorySubscribed = (int t) {}; + host.updateClaudeUsageFromEvent = (int t, string translated) => false; + host.planBroadcast = (int t, TranslatedEvent ev) => HistoryBroadcastPlan.init; + auto pipeline = new HistoryEventPipeline(host); + + foreach (event; [ + `{"type":"session/init","session_id":"s1"}`, // seq 0 + `{"type":"item/started","item_type":"user_message","uuid":"u1"}`, // seq 1 + `{"type":"session/metadata","model":"m1"}`, // seq 2 + `{"type":"item/started","item_type":"text","item_id":"a1"}`, // seq 3 + `{"type":"session/metadata","model":"m2"}`, // seq 4 + `{"type":"item/started","item_type":"user_message","uuid":"u2"}`, // seq 5 + `{"type":"item/started","item_type":"text","item_id":"a2"}`, // seq 6 + ]) + pipeline.appendAndBroadcastTaskEvent(tid, TranslatedEvent(event, null, AbsTime(1000))); + assert(td.history.length == 7); + + @JSONPartial static struct StartProbe { string type; int window_start; } + @JSONPartial static struct SeqProbe { int seq = -1; } + + auto ws = new StubWs(); + scope(exit) ws.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistory(ws, tid, 1); + assert(jsonParse!StartProbe(ws.sent[0]).window_start == 6); + assert(jsonParse!SeqProbe(ws.sent[1]).seq == 0 && ws.sent[1].canFind(`session/init`), + "the init replays first, with its true seq"); + assert(jsonParse!SeqProbe(ws.sent[2]).seq == 4 && ws.sent[2].canFind(`session/metadata`), + "the newest metadata follows the init"); + assert(jsonParse!SeqProbe(ws.sent[3]).seq == 6, "then the windowed range"); + assert(ws.sent[$ - 1].canFind("task_history_end")); + + // a window that starts before the init sends no separate context + auto wsAll = new StubWs(); + scope(exit) wsAll.disconnect("test complete", DisconnectType.requested); + pipeline.handleRequestHistory(wsAll, tid); + assert(jsonParse!SeqProbe(wsAll.sent[1]).seq == 0); + assert(wsAll.sent.length == 9, "full replay carries no duplicated context"); +} + diff --git a/web/src/app.test.tsx b/web/src/app.test.tsx index 5cc2e1a7..7420fc01 100644 --- a/web/src/app.test.tsx +++ b/web/src/app.test.tsx @@ -101,6 +101,8 @@ vi.mock("./useSessionManager", () => ({ getTaskHref: vi.fn(), getByTid: state.getByTid, refreshWorkspaces: vi.fn(), + historyWindowStep: 0, + loadMoreHistory: vi.fn(), scanState: "idle", }) satisfies TaskManager, })); diff --git a/web/src/app.tsx b/web/src/app.tsx index 36714e6f..3e7d81db 100644 --- a/web/src/app.tsx +++ b/web/src/app.tsx @@ -74,6 +74,8 @@ function AppContent() { serverError, dismissServerError, devMode, + historyWindowStep, + loadMoreHistory, navigateHome, navigateToProject, getProjectHref, @@ -519,6 +521,8 @@ function AppContent() { onEditRawEvent={editRawEvent} defaultAgent={effectiveDefaultAgent} agentUsage={agentUsage} + historyWindowStep={historyWindowStep} + onLoadMoreHistory={loadMoreHistory} getTaskHref={getTaskHref} /> diff --git a/web/src/components/MessageList.loadmore.test.tsx b/web/src/components/MessageList.loadmore.test.tsx new file mode 100644 index 00000000..58068158 --- /dev/null +++ b/web/src/components/MessageList.loadmore.test.tsx @@ -0,0 +1,70 @@ +/** + * @vitest-environment jsdom + * @vitest-environment-options { "pretendToBeVisual": true } + * + * Clicking a "Load [N] more" button gave no feedback at all until the slice + * landed; the buttons must swap for the loading line immediately, and come + * back (or disappear) once the window start moves. + */ +import { describe, expect, it, vi } from "vitest"; +import { render } from "preact"; +import { act } from "preact/test-utils"; +import { MessageList } from "./MessageList"; + +vi.hoisted(() => { + vi.stubGlobal("CSS", { supports: () => false }); +}); + +function mount( + container: HTMLElement, + windowStart: number, + onLoadMoreHistory: (step: number) => void, +) { + render( + {}} + historyWindowStart={windowStart} + historyWindowStep={30} + historyWindowed={true} + onLoadMoreHistory={onLoadMoreHistory} + />, + container, + ); +} + +describe("load-more loading feedback", () => { + it("swaps the buttons for the loading line until the slice lands", async () => { + const container = document.createElement("div"); + document.body.appendChild(container); + const calls: number[] = []; + await act(() => { + mount(container, 30, (step) => calls.push(step)); + }); + + expect(container.querySelectorAll(".load-more-row .btn")).toHaveLength(3); + expect(container.querySelector(".load-more-loading")).toBeNull(); + + await act(() => { + container + .querySelector(".load-more-row .btn")! + .dispatchEvent(new MouseEvent("click", { bubbles: true })); + }); + + expect(calls).toEqual([30]); + expect(container.querySelectorAll(".load-more-row .btn")).toHaveLength(0); + expect(container.querySelector(".load-more-loading")).not.toBeNull(); + + // the slice landed: the window start moved, older history remains + await act(() => { + mount(container, 10, (step) => calls.push(step)); + }); + expect(container.querySelector(".load-more-loading")).toBeNull(); + expect(container.querySelectorAll(".load-more-row .btn")).toHaveLength(3); + }); +}); diff --git a/web/src/components/MessageList.tsx b/web/src/components/MessageList.tsx index 26e21ce7..449f8a0f 100644 --- a/web/src/components/MessageList.tsx +++ b/web/src/components/MessageList.tsx @@ -42,6 +42,15 @@ interface Props { onViewFile?: (filePath: string) => void; spawnedTidsByItemId?: Map>; getTaskHref?: (id: string) => string; + /** whether the server windowed this replay; when it did, the DOM is small + * enough that the browser's own lazy rendering only costs scroll stability */ + historyWindowed?: boolean; + /** first replayed seq; > 0 shows the load-more row at the top */ + historyWindowStart?: number; + /** configured window size in messages, used for the button labels */ + historyWindowStep?: number; + /** load `step` more messages of older history; 0 = all */ + onLoadMoreHistory?: (step: number) => void; } function ResultMessageView({ message }: { message: DisplayMessage }) { @@ -851,9 +860,53 @@ export function MessageList({ onViewFile, spawnedTidsByItemId, getTaskHref, + historyWindowed, + historyWindowStart, + historyWindowStep, + onLoadMoreHistory, }: Props) { const containerRef = useRef(null); + // Loading older messages must leave the reader where they are, with the new + // content arriving above them. The list is column-reverse, so scrollTop is a + // distance from the bottom and is exactly the quantity to hold constant; + // left alone the browser anchors to the load-more row instead, which sits + // above everything and drags the viewport up to the new top. + const heldScrollRef = useRef(null); + const previousWindowStartRef = useRef(historyWindowStart); + // a click swaps the buttons for the loading line; the slice landing (the + // window start moving) brings the row back, or removes it entirely + const [loadingMore, setLoadingMore] = useState(false); + + useLayoutEffect(() => { + const el = containerRef.current; + const previous = previousWindowStartRef.current; + previousWindowStartRef.current = historyWindowStart; + if ( + el === null || + heldScrollRef.current === null || + previous === undefined || + historyWindowStart === undefined || + historyWindowStart >= previous + ) + return; + // older messages just landed in front + el.scrollTop = heldScrollRef.current; + heldScrollRef.current = null; + }, [historyWindowStart]); + + // remember where the reader is before the batch lands, and stop the browser + // from anchoring in the meantime + const loadMore = (step: number) => { + const el = containerRef.current; + if (el) { + heldScrollRef.current = el.scrollTop; + el.style.overflowAnchor = "none"; + } + onLoadMoreHistory?.(step); + setLoadingMore(true); + }; + // Subscribe to outbox so the component re-renders when entries are added/removed. const [outboxTick, setOutboxTick] = useState(0); useEffect(() => { @@ -912,6 +965,10 @@ export function MessageList({ [onEditRawEvent, taskTid], ); + useEffect(() => { + setLoadingMore(false); + }, [historyWindowStart, taskTid]); + // On session switch, scroll to bottom (scrollTop 0 = bottom in column-reverse). const prevTaskTid = useRef(taskTid); useLayoutEffect(() => { @@ -1029,9 +1086,55 @@ export function MessageList({ return ( -
+
+ {historyWindowStart != null && + historyWindowStart > 0 && + historyWindowStep != null && + historyWindowStep > 0 && + onLoadMoreHistory && ( +
+ {loadingMore ? ( +
+ +
+ ) : ( + <> + + + + + )} +
+ )} {topLevelMessages.map((msg) => { const resolvedBlocks = msg.type === "assistant" diff --git a/web/src/components/SessionView.tsx b/web/src/components/SessionView.tsx index 31bc4cd5..88dc9285 100644 --- a/web/src/components/SessionView.tsx +++ b/web/src/components/SessionView.tsx @@ -63,6 +63,10 @@ interface Props { exportMode?: boolean; getTaskHref?: (id: string) => string; agentUsage?: Record; + /** configured history window in messages for this device class */ + historyWindowStep?: number; + /** load `step` more messages of older history for a task; 0 = all */ + onLoadMoreHistory?: (tid: number, step: number) => void; } function SessionViewInner({ @@ -94,6 +98,8 @@ function SessionViewInner({ exportMode, getTaskHref, agentUsage, + historyWindowStep, + onLoadMoreHistory, }: Props) { const inputRef = useRef(null); const insertTextRef = useRef<((text: string) => void) | null>(null); @@ -415,6 +421,16 @@ function SessionViewInner({
) : ( { + onLoadMoreHistory(tid, step); + } + : undefined + } taskTid={tid} messages={task.messages} replacementEvents={task.replacementEvents} diff --git a/web/src/connection.ts b/web/src/connection.ts index 68c3c87a..e77684e0 100644 --- a/web/src/connection.ts +++ b/web/src/connection.ts @@ -110,6 +110,8 @@ export class Connection { raw.type === "task_reload" || raw.type === "title_update" || raw.type === "task_history_start" || + raw.type === "task_history_prepend_start" || + raw.type === "task_history_prepend_end" || raw.type === "task_history_end" || raw.type === "workspaces_list" || raw.type === "task_types_list" || @@ -256,8 +258,33 @@ export class Connection { ); } - requestHistory(tid: number): boolean { - return this.send(JSON.stringify({ type: "request_history", tid })); + requestHistory(tid: number, limit = 0, deviceClass = ""): boolean { + return this.send( + JSON.stringify({ + type: "request_history", + tid, + limit, + device_class: deviceClass, + }), + ); + } + + /** ask for the messages older than `beforeSeq`, to prepend to what is held */ + requestHistoryBefore( + tid: number, + beforeSeq: number, + limit: number, + deviceClass: string, + ): boolean { + return this.send( + JSON.stringify({ + type: "request_history_before", + tid, + before_seq: beforeSeq, + limit, + device_class: deviceClass, + }), + ); } forkTask(tid: number, afterUuid: string) { diff --git a/web/src/historyContinuation.test.ts b/web/src/historyContinuation.test.ts new file mode 100644 index 00000000..86f4de73 --- /dev/null +++ b/web/src/historyContinuation.test.ts @@ -0,0 +1,206 @@ +// "Load [N] more" must be a continuation of the initial load: its end state +// has to be exactly what an initial load with the larger window would have +// produced. The regression pinned here: a queued send leaves a seq-less +// pending placeholder in the stored history, healed only by its later +// delivery; reducing an older slice in isolation resurrected the placeholder +// as a phantom pending bubble at the bottom of the task. +import { describe, expect, it, vi } from "vitest"; + +vi.hoisted(() => { + vi.stubGlobal("CSS", { supports: () => false }); + vi.stubGlobal("document", { querySelector: () => null }); +}); + +import { rebuildFromFrames, type HistoryFrame } from "./historyContinuation"; +import { makeTaskState, type TaskState } from "./types"; + +const NONCE = "queued-nonce-1"; + +function userFrame( + text: string, + seq: number, + correlationId?: string, +): HistoryFrame { + return { + kind: "event", + msg: { + type: "item/started", + item_type: "user_message", + item_id: `user-${seq}`, + text, + content: [{ type: "text", text }], + ...(correlationId ? { correlation_id: correlationId } : {}), + } as HistoryFrame["msg"], + seq, + }; +} + +function assistantFrame(text: string, seq: number): HistoryFrame { + return { + kind: "event", + msg: { + type: "item/started", + item_type: "text", + item_id: `assistant-${seq}`, + text, + } as HistoryFrame["msg"], + seq, + }; +} + +/** The seq-less placeholder the server broadcasts (and re-sends on replay) + * for a message queued while the agent is mid-turn. */ +function placeholderFrame(text: string, correlationId: string): HistoryFrame { + return { + kind: "unconfirmed", + msg: { + type: "item/started", + item_type: "user_message", + item_id: "cc-user-msg", + text, + content: [{ type: "text", text }], + pending: true, + } as HistoryFrame["msg"], + correlationId, + }; +} + +function initFrame(seq: number): HistoryFrame { + return { + kind: "event", + msg: { + type: "session/init", + session_id: "session-1", + } as HistoryFrame["msg"], + seq, + }; +} + +function metadataFrame(model: string, seq: number): HistoryFrame { + return { + kind: "event", + msg: { type: "session/metadata", model } as HistoryFrame["msg"], + seq, + }; +} + +function baseTask(): TaskState { + return { ...makeTaskState(5, true), uuid: "task-5" }; +} + +/** Identity-free view of the rendered transcript. */ +function transcript(state: TaskState) { + return state.messages.map((m) => ({ + type: m.type, + pending: m.pending === true, + text: m.content.map((b) => ("text" in b ? (b.text ?? "") : "")).join(""), + })); +} + +// arrival order of a task where "queued mid-turn" was sent while the agent +// worked: its placeholder appears mid-stream, its delivery two events later +const FRAMES: HistoryFrame[] = [ + userFrame("first question", 0), + assistantFrame("first answer", 1), + placeholderFrame("queued mid-turn", NONCE), + assistantFrame("long turn continues", 2), + userFrame("queued mid-turn", 3, NONCE), + assistantFrame("answer to queued", 4), +]; +// the window boundary a windowed initial load would pick: the delivered +// non-pending user message at seq 3 +const SUFFIX = FRAMES.slice(4); +const OLDER_SLICE = FRAMES.slice(0, 4); + +describe("history continuation", () => { + it("load-more ends in exactly the state a larger initial load produces", () => { + const reference = rebuildFromFrames(baseTask(), FRAMES, 0); + + const windowed = rebuildFromFrames(baseTask(), SUFFIX, 3); + const afterLoadMore = rebuildFromFrames( + windowed, + OLDER_SLICE.concat(SUFFIX), + 0, + ); + + expect(transcript(afterLoadMore)).toEqual(transcript(reference)); + expect(afterLoadMore.historyWindowStart).toBe(0); + expect(afterLoadMore.historyLoaded).toBe(true); + }); + + it("does not resurrect a delivered message as a pending phantom", () => { + const windowed = rebuildFromFrames(baseTask(), SUFFIX, 3); + const afterLoadMore = rebuildFromFrames( + windowed, + OLDER_SLICE.concat(SUFFIX), + 0, + ); + + const copies = transcript(afterLoadMore).filter( + (m) => m.text === "queued mid-turn", + ); + expect(copies).toHaveLength(1); + expect(copies[0]?.pending).toBe(false); + }); + + it("keeps replayed session context ahead of a load-more slice", () => { + // a windowed load of a codex-style transcript: the server replays the + // pre-window session/init and newest session/metadata ahead of the range + const all: HistoryFrame[] = [ + initFrame(0), + userFrame("early question", 1), + metadataFrame("model-b", 2), + assistantFrame("early answer", 3), + userFrame("late question", 4), + assistantFrame("late answer", 5), + ]; + const reference = rebuildFromFrames(baseTask(), all, 0); + + const initialArrival = [ + initFrame(0), + metadataFrame("model-b", 2), + ...all.slice(4), + ]; + const windowed = rebuildFromFrames(baseTask(), initialArrival, 4); + expect(windowed.sessionInfo?.model).toBe("model-b"); + + // the prepend merge: context the slice covers is dropped, the slice goes + // first, everything already held follows (mirrors task_history_prepend_end) + const slice = all.slice(0, 4); + const sliceEnd = 4; + const newWindowStart = 0; + const context: HistoryFrame[] = []; + const rest: HistoryFrame[] = []; + for (const f of initialArrival) { + if (f.kind === "event" && f.seq !== undefined && f.seq < sliceEnd) { + if (f.seq < newWindowStart) context.push(f); + } else { + rest.push(f); + } + } + const afterLoadMore = rebuildFromFrames( + windowed, + context.concat(slice, rest), + newWindowStart, + ); + + expect(transcript(afterLoadMore)).toEqual(transcript(reference)); + expect(afterLoadMore.sessionInfo?.model).toBe("model-b"); + }); + + it("keeps a genuinely still-queued placeholder pending", () => { + // no delivery anywhere: the message is still waiting in the queue + const stillQueued = [ + userFrame("first question", 0), + assistantFrame("first answer", 1), + placeholderFrame("never delivered", "queued-nonce-2"), + ]; + const rebuilt = rebuildFromFrames(baseTask(), stillQueued, 0); + + const copies = transcript(rebuilt).filter( + (m) => m.text === "never delivered", + ); + expect(copies).toHaveLength(1); + expect(copies[0]?.pending).toBe(true); + }); +}); diff --git a/web/src/historyContinuation.ts b/web/src/historyContinuation.ts new file mode 100644 index 00000000..06e8f04a --- /dev/null +++ b/web/src/historyContinuation.ts @@ -0,0 +1,137 @@ +/** + * Continuation of a task's history replay. + * + * "Load [N] more" must behave as a continuation of the initial load: its end + * state has to be exactly what an initial load with the larger window would + * have produced. The only structure that guarantees that is re-running the + * same reduction over the same input, so every frame that built the current + * timeline is kept in arrival order, a load-more fetches only the older + * slice, and the whole sequence is reduced again from the same reset an + * initial replay starts from. Reducing the older slice in isolation and + * merging message lists loses cross-references: a queued send's placeholder + * healed by its later delivery, a consumption event upgrading an earlier + * bubble. That split is what resurrected already-delivered messages as + * phantom pending bubbles. + */ +import type { AgnosticEvent, AssistantContentBlock } from "./protocol"; +import { resetTaskForHistoryReplay } from "./historyReplayReset"; +import { reduceMessage, replaceHistoryBoundary } from "./sessionReducer"; +import type { CydoMeta, HistoryBoundary, TaskState } from "./types"; + +/** One frame as it arrived over the socket, in the two shapes the timeline + * is built from: seq'd task events, and seq-less unconfirmed placeholders. */ +export type HistoryFrame = + | { kind: "event"; msg: AgnosticEvent; seq?: number; ts?: number } + | { kind: "unconfirmed"; msg: AgnosticEvent; correlationId?: string }; + +export type HistoryBoundaryEvent = + | (Extract & { + history_boundary: HistoryBoundary; + }) + | (Extract & { + history_boundary: HistoryBoundary; + }); + +export function hasHistoryBoundary( + event: AgnosticEvent, +): event is HistoryBoundaryEvent { + return ( + (event.type === "item/started" || event.type === "turn/stop") && + event.history_boundary !== undefined + ); +} + +/// Extract text content from a user message event (for unconfirmed display). +export function extractTextContent(msg: AgnosticEvent): string { + if (msg.type !== "item/started" || msg.item_type !== "user_message") + return ""; + return msg.text ?? ""; +} + +/** The state transform for an unconfirmed (queued, not yet delivered) user + * message: upgrade a local placeholder with the same nonce if one exists, + * otherwise append a fresh pending bubble. Pure; side effects (outbox, + * notifications) stay with the live caller. */ +export function reduceUnconfirmedUserMessage( + prev: TaskState, + msg: AgnosticEvent, + correlationId?: string, +): TaskState { + const meta = (msg as Record).meta as CydoMeta | undefined; + const content = ((msg as Record).content as + | AssistantContentBlock[] + | undefined) ?? [{ type: "text" as const, text: extractTextContent(msg) }]; + + if (correlationId) { + const idx = prev.messages.findIndex( + (m) => m.type === "user" && m.nonce === correlationId, + ); + if (idx >= 0) { + return { + ...prev, + messages: prev.messages.map((m, i) => + i === idx + ? { + ...m, + ackState: 3 as const, + pending: true, + isProvisional: true, + } + : m, + ), + }; + } + } + + const msgIdCounter = prev.msgIdCounter + 1; + return { + ...prev, + msgIdCounter, + messages: [ + ...prev.messages, + { + id: `pending-${msgIdCounter}`, + type: "user" as const, + content, + ackState: 3 as const, + pending: true, + nonce: correlationId, + cydoMeta: meta, + isProvisional: true, + }, + ], + }; +} + +/** Reduce one frame into the timeline exactly the way the live path does. */ +export function reduceHistoryFrame( + state: TaskState, + frame: HistoryFrame, +): TaskState { + if (frame.kind === "unconfirmed") + return reduceUnconfirmedUserMessage(state, frame.msg, frame.correlationId); + let updated = reduceMessage(state, frame.msg, frame.seq, frame.ts); + if (hasHistoryBoundary(frame.msg) && frame.seq !== undefined) + updated = replaceHistoryBoundary(updated, frame.msg, frame.seq); + return updated; +} + +/** Rebuild a task's timeline from the complete frame sequence: the same + * reset an initial replay starts from, then every frame in order. Fields + * that do not derive from the timeline ride through the reset untouched. */ +export function rebuildFromFrames( + task: TaskState, + frames: readonly HistoryFrame[], + windowStart: number, +): TaskState { + let state: TaskState = resetTaskForHistoryReplay(task, frames.length); + for (const frame of frames) state = reduceHistoryFrame(state, frame); + return { + ...state, + historyLoaded: true, + everLoaded: true, + historyTotal: undefined, + historyReceived: undefined, + historyWindowStart: windowStart, + }; +} diff --git a/web/src/protocol.ts b/web/src/protocol.ts index 6fd8bae6..3892a6a3 100644 --- a/web/src/protocol.ts +++ b/web/src/protocol.ts @@ -268,6 +268,19 @@ export interface TaskHistoryStartMessage { type: "task_history_start"; tid: number; total: number; + window_start?: number; + window_limit?: number; +} +export interface TaskHistoryPrependStartMessage { + type: "task_history_prepend_start"; + tid: number; + window_start: number; + before_seq: number; +} +export interface TaskHistoryPrependEndMessage { + type: "task_history_prepend_end"; + tid: number; + window_start: number; } export interface TaskHistoryEndMessage { type: "task_history_end"; @@ -402,6 +415,8 @@ export interface ServerStatusMessage { auth_enabled: boolean; dev_mode?: boolean; build_id?: string; + history_window_desktop?: number; + history_window_mobile?: number; } export interface TaskDeletedMessage { type: "task_deleted"; @@ -441,6 +456,8 @@ export type ControlMessage = | FocusHintMessage | TitleUpdateMessage | TaskHistoryStartMessage + | TaskHistoryPrependStartMessage + | TaskHistoryPrependEndMessage | TaskHistoryEndMessage | WorkspacesListMessage | TaskTypesListMessage diff --git a/web/src/styles.css b/web/src/styles.css index cea89215..9d6c0caf 100644 --- a/web/src/styles.css +++ b/web/src/styles.css @@ -937,11 +937,61 @@ body, word-break: break-word; } +/* Windowed history: centered load-more row at the top of the list */ +.load-more-row { + display: flex; + justify-content: center; + gap: 8px; + padding: 12px 0 20px; +} + +.load-more-row .btn { + background: var(--bg-tertiary); + color: var(--text-dim); + border-radius: 6px; + font-size: 13px; + padding: 6px 14px; +} + +.load-more-row .btn:hover { + color: var(--text); +} + +/* while the older slice is on its way the buttons swap for the same band + that plays while awaiting a response, in grey */ +.load-more-loading { + display: flex; + flex-direction: column; + justify-content: center; + flex: 1; + /* the height of the buttons it replaces, so nothing shifts */ + height: 31px; +} + +.load-more-loading .status-band[data-status="requesting"] { + --status-color: var(--text-dim); +} + +/* the band mounts fresh at each click, and the stock 22s ease-in-out starts + at its slowest phase: a short fetch ends before any visible motion. start + mid-cycle at a faster tempo so the very first frame is already moving */ +.load-more-loading .sb-layer-requesting { + animation-duration: 8s; + animation-delay: -2s; +} + /* Messages */ .message { margin-bottom: 16px; padding: 12px 16px; max-width: 100%; +} + +/* Skipping offscreen messages is worth it only when the whole transcript is in + the DOM: the reserved 100px is a guess, so heights change as messages scroll + into view and the scroll region grows under a dragged scrollbar handle. A + windowed replay keeps the DOM small enough not to need it. */ +.message-list:not(.history-windowed) .message { contain: layout paint style; content-visibility: auto; contain-intrinsic-size: auto 100px; diff --git a/web/src/types.ts b/web/src/types.ts index de4800a2..df64b0fe 100644 --- a/web/src/types.ts +++ b/web/src/types.ts @@ -277,6 +277,13 @@ export interface TaskState { title?: string; /** Whether the task's JSONL history has been loaded from the backend. */ historyLoaded: boolean; + /** Whether the server windowed this replay at all. Drives the browser's own + * lazy rendering: with a window the DOM is small and `content-visibility` + * only makes the scroll height jitter, but an unwindowed replay needs it. */ + historyWindowed?: boolean; + /** First replayed seq of the current window; > 0 means older history exists + * on the server, unsent. */ + historyWindowStart?: number; /** Latched true once history has loaded at least once; never resets. * Used to keep tasks the user has visited rendered in the DOM across * task_reload cycles, so InputBox doesn't unmount mid-interaction. */ diff --git a/web/src/useExportedTaskManager.ts b/web/src/useExportedTaskManager.ts index ed73d27d..6f20bb75 100644 --- a/web/src/useExportedTaskManager.ts +++ b/web/src/useExportedTaskManager.ts @@ -255,6 +255,8 @@ export function useExportedTaskManager(): TaskManager { agentUsage: {}, serverError: null, dismissServerError: noop, + historyWindowStep: 0, + loadMoreHistory: () => {}, devMode: false, exportLoadError, navigateHome: noop, diff --git a/web/src/useSessionManager.submission.test.ts b/web/src/useSessionManager.submission.test.ts index 62165714..8b588898 100644 --- a/web/src/useSessionManager.submission.test.ts +++ b/web/src/useSessionManager.submission.test.ts @@ -529,7 +529,11 @@ describe("submission acknowledgement routing", () => { true, ); expect(testState.connection!.requestHistory).toHaveBeenCalledTimes(1); - expect(testState.connection!.requestHistory).toHaveBeenCalledWith(61); + expect(testState.connection!.requestHistory).toHaveBeenCalledWith( + 61, + 0, + "desktop", + ); }); it("does not route a submission acknowledgement after attachment changes", async () => { @@ -545,7 +549,11 @@ describe("submission acknowledgement routing", () => { expect(testState.navigate).not.toHaveBeenCalled(); expect(testState.connection!.requestHistory).toHaveBeenCalledTimes(1); - expect(testState.connection!.requestHistory).toHaveBeenCalledWith(62); + expect(testState.connection!.requestHistory).toHaveBeenCalledWith( + 62, + 0, + "desktop", + ); }); it("does not route a held submission acknowledgement across a pre-effect route switch", async () => { @@ -2114,7 +2122,11 @@ describe("submission acknowledgement routing", () => { }); expect(testState.connection!.requestHistory).toHaveBeenCalledTimes(1); - expect(testState.connection!.requestHistory).toHaveBeenCalledWith(71); + expect(testState.connection!.requestHistory).toHaveBeenCalledWith( + 71, + 0, + "desktop", + ); expect(testState.connection!.sendMessage).toHaveBeenCalledWith( 71, [{ type: "text", text: "draft to delete" }], @@ -2149,7 +2161,11 @@ describe("submission acknowledgement routing", () => { }); expect(testState.connection!.requestHistory).toHaveBeenCalledTimes(1); - expect(testState.connection!.requestHistory).toHaveBeenCalledWith(71); + expect(testState.connection!.requestHistory).toHaveBeenCalledWith( + 71, + 0, + "desktop", + ); expect(testState.connection!.sendMessage).toHaveBeenCalledWith( 71, [{ type: "text", text: "draft to delete" }], diff --git a/web/src/useSessionManager.ts b/web/src/useSessionManager.ts index a139339f..7d06e040 100644 --- a/web/src/useSessionManager.ts +++ b/web/src/useSessionManager.ts @@ -19,22 +19,26 @@ import type { AgnosticEvent, ControlMessage, ContentBlock, - HistoryBoundary, Notice, TasksListMessage, } from "./protocol"; -import type { CydoMeta, TaskState, UndoPending } from "./types"; +import type { TaskState, UndoPending } from "./types"; import { makeTaskState } from "./types"; +import { reduceAgentAck, replaceHistoryBoundary } from "./sessionReducer"; import { - reduceAgentAck, - reduceMessage, - replaceHistoryBoundary, -} from "./sessionReducer"; + hasHistoryBoundary, + rebuildFromFrames, + reduceHistoryFrame, + reduceUnconfirmedUserMessage, + type HistoryBoundaryEvent, + type HistoryFrame, +} from "./historyContinuation"; import { outbox } from "./outbox"; import { beginTaskHistoryReplay, excludeReloadDraftUuid, reconcileInputDraft, + resetTaskForHistoryReplay, resetTaskForReload, snapshotUserDrafts, } from "./historyReplayReset"; @@ -61,6 +65,30 @@ import { taskPath, } from "./routing"; +// Only the class travels: the server holds the window sizes, because the first +// history request goes out before server_status has taught this side what they +// are, and guessing zero there means replaying the whole task. +// +// Three signals rather than one, since a pointer query alone is not enough: +// some mobile browsers report a fine pointer despite being a phone. A +// touchscreen laptop reads as mobile here, which only costs it a smaller +// starting window. +function deviceClass(): "mobile" | "desktop" { + if (typeof window === "undefined") return "desktop"; + const media = (query: string) => + typeof window.matchMedia === "function" && window.matchMedia(query).matches; + const touch = window.navigator.maxTouchPoints > 0; + const mobileAgent = /Mobi|Android|iPhone|iPad|iPod/i.test( + window.navigator.userAgent, + ); + return touch || + mobileAgent || + media("(pointer: coarse)") || + media("(hover: none)") + ? "mobile" + : "desktop"; +} + export interface ImageAttachment { id: string; dataURL: string; @@ -68,23 +96,6 @@ export interface ImageAttachment { mediaType: string; } -type HistoryBoundaryEvent = - | (Extract & { - history_boundary: HistoryBoundary; - }) - | (Extract & { - history_boundary: HistoryBoundary; - }); - -function hasHistoryBoundary( - event: AgnosticEvent, -): event is HistoryBoundaryEvent { - return ( - (event.type === "item/started" || event.type === "turn/stop") && - event.history_boundary !== undefined - ); -} - export function revertFilesForUndo( canRevertFiles: boolean, revertFiles: boolean, @@ -255,6 +266,10 @@ export interface TaskManager { serverError: { message: string; tid?: number } | null; dismissServerError: () => void; devMode: boolean; + /** configured history window in messages for this device class; 0 = off */ + historyWindowStep: number; + /** load `step` more messages of older history (0 = all) */ + loadMoreHistory: (tid: number, step: number) => void; exportLoadError?: string | null; navigateHome: () => void; navigateToProject: (workspace: string, projectName: string) => void; @@ -491,13 +506,6 @@ function activityOnlyObservation( }; } -/// Extract text content from a user message event (for unconfirmed display). -function extractTextContent(msg: AgnosticEvent): string { - if (msg.type !== "item/started" || msg.item_type !== "user_message") - return ""; - return msg.text ?? ""; -} - // Mutable mirror of task states, keyed by uuid. Updated synchronously outside // Preact's render cycle so reducers and notification checks run immediately // when WebSocket messages arrive, even when Preact defers state updates in @@ -565,6 +573,24 @@ export function useTaskManager( tid?: number; } | null>(null); const [devMode, setDevMode] = useState(false); + // the configured window for this device class, used to label the load-more + // buttons; the server still decides what a request actually returns + const [historyWindowStep, setHistoryWindowStep] = useState(0); + // every frame that built each task's current timeline, in arrival order; + // load-more puts its fetched slice in front and re-reduces the whole + // sequence, so the result matches a larger initial window exactly + const historyFramesRef = useRef(new Map()); + // a "Load more" in flight: requested until the prepend_start marker + // arrives, then capturing the slice until prepend_end triggers the rebuild + const prependCaptureRef = useRef( + new Map< + string, + { phase: "requested" } | { phase: "capturing"; frames: HistoryFrame[] } + >(), + ); + // tasks whose window was grown by a load-more; the extra history lives only + // while the task is focused, and unfocusing releases it back to the window + const grownHistoryRef = useRef(new Set()); const addToastRef = useRef(addToast); addToastRef.current = addToast; const prevNoticeIdsRef = useRef>(new Set()); @@ -1154,11 +1180,36 @@ export function useTaskManager( [entryPointFor], ); + // continue the initial load with older messages: fetch only the slice the + // window held back, then re-reduce it in front of every frame already + // reduced, so nothing on screen is lost and nothing is fetched twice + const loadMoreHistory = useCallback((tid: number, step: number) => { + const task = findByTid(tid); + if (!task) throw new Error(`Loading more history requires task ${tid}`); + const beforeSeq = task.historyWindowStart ?? 0; + if (beforeSeq <= 0) return; // nothing older is being held back + if (prependCaptureRef.current.has(task.uuid)) return; // one batch at a time + + prependCaptureRef.current.set(task.uuid, { phase: "requested" }); + + // step 0 is "load all", i.e. everything before the current start + const sent = connRef.current?.requestHistoryBefore( + tid, + beforeSeq, + step === 0 ? -1 : step, + deviceClass(), + ); + if (!sent) { + prependCaptureRef.current.delete(task.uuid); + throw new Error(`History request failed for ${tid}`); + } + }, []); + const requestTaskHistory = useCallback((tid: number) => { const task = findByTid(tid); if (!task) throw new Error(`History request requires task ${tid}`); if (requestedHistoryRef.current.has(task.uuid)) return; - if (!connRef.current?.requestHistory(tid)) { + if (!connRef.current?.requestHistory(tid, 0, deviceClass())) { throw new Error(`History request failed for ${tid}`); } requestedHistoryRef.current.add(task.uuid); @@ -1199,67 +1250,24 @@ export function useTaskManager( executeDraftEffectsRef.current(effects); return; } + const capture = prependCaptureRef.current.get(uuid); + if (capture !== undefined && capture.phase === "capturing") { + // part of the older slice being fetched; joins the rebuild instead + capture.frames.push({ kind: "unconfirmed", msg, correlationId }); + executeDraftEffectsRef.current(effects); + return; + } + + if (correlationId) outbox.remove(correlationId); const prev = liveStates.get(uuid) ?? { ...makeTaskState(tid, true), uuid, }; - const meta = (msg as Record).meta as - | CydoMeta - | undefined; - const content = ((msg as Record).content as - | import("./protocol").AssistantContentBlock[] - | undefined) ?? [ - { type: "text" as const, text: extractTextContent(msg) }, - ]; - - // If a local ackState=4 placeholder with this nonce exists, upgrade it - // to ackState=3 (backend acked). Otherwise insert a fresh ackState=3 bubble. - let messages = prev.messages; - if (correlationId) { - outbox.remove(correlationId); - const idx = messages.findIndex( - (m) => m.type === "user" && m.nonce === correlationId, - ); - if (idx >= 0) { - messages = messages.map((m, i) => - i === idx - ? { - ...m, - ackState: 3 as const, - pending: true, - isProvisional: true, - } - : m, - ); - const updated = { ...prev, messages }; - liveStates.set(uuid, updated); - setTasks((map) => { - const next = new Map(map); - next.set(uuid, updated); - return next; - }); - executeDraftEffectsRef.current(effects); - return; - } - } - - const id = `pending-${++prev.msgIdCounter}`; - const updated = { - ...prev, - messages: [ - ...messages, - { - id, - type: "user" as const, - content, - ackState: 3 as const, - pending: true, - nonce: correlationId, - cydoMeta: meta, - isProvisional: true, - }, - ], - }; + const updated = reduceUnconfirmedUserMessage(prev, msg, correlationId); + const frame: HistoryFrame = { kind: "unconfirmed", msg, correlationId }; + const frames = historyFramesRef.current.get(uuid); + if (frames) frames.push(frame); + else historyFramesRef.current.set(uuid, [frame]); liveStates.set(uuid, updated); setTasks((map) => { const next = new Map(map); @@ -1287,13 +1295,24 @@ export function useTaskManager( executeDraftEffectsRef.current(effects); return; } + const capture = prependCaptureRef.current.get(uuid); + if (capture !== undefined && capture.phase === "capturing") { + // between the prepend markers every frame of this task belongs to + // the older slice; it is reduced during the rebuild, not here + capture.frames.push({ kind: "event", msg, seq, ts }); + executeDraftEffectsRef.current(effects); + return; + } + const prev = liveStates.get(uuid) ?? { ...makeTaskState(tid, true), uuid, }; - let updated = reduceMessage(prev, msg, seq, ts); - if (hasHistoryBoundary(msg) && seq !== undefined) - updated = replaceHistoryBoundary(updated, msg, seq); + const frame: HistoryFrame = { kind: "event", msg, seq, ts }; + const frames = historyFramesRef.current.get(uuid); + if (frames) frames.push(frame); + else historyFramesRef.current.set(uuid, [frame]); + let updated = reduceHistoryFrame(prev, frame); if (!updated.historyLoaded && updated.historyTotal !== undefined) { updated = { ...updated, @@ -1574,6 +1593,9 @@ export function useTaskManager( if (!t) break; outbox.removeForTask(tid); requestedHistoryRef.current.delete(t.uuid); + historyFramesRef.current.delete(t.uuid); + prependCaptureRef.current.delete(t.uuid); + grownHistoryRef.current.delete(t.uuid); const isEdit = msg.reason === "edit"; const excludedNativeUuid = @@ -1603,7 +1625,7 @@ export function useTaskManager( // won't re-fire because activeTaskId hasn't changed. let final = reset; if (String(tid) === activeTaskIdRef.current) { - if (connRef.current?.requestHistory(tid)) { + if (connRef.current?.requestHistory(tid, 0, deviceClass())) { requestedHistoryRef.current.add(t.uuid); final = { ...reset, @@ -1625,7 +1647,73 @@ export function useTaskManager( const { tid, total } = msg; const t0 = findByTid(tid); if (!t0) break; - const t = beginTaskHistoryReplay(t0, total); + // the frame cache mirrors the timeline reset in beginTaskHistoryReplay + if (t0.pendingHistoryReplies === 0) { + historyFramesRef.current.set(t0.uuid, []); + grownHistoryRef.current.delete(t0.uuid); + } + prependCaptureRef.current.delete(t0.uuid); + const windowStart = msg.window_start ?? 0; + // historyTotal drives the progress bar, so count only what will + // actually be sent + const t = { + ...beginTaskHistoryReplay(t0, total - windowStart), + historyWindowed: (msg.window_limit ?? 0) > 0, + historyWindowStart: windowStart, + }; + liveStates.set(t0.uuid, t); + setTasks((prev) => { + if (!prev.has(t0.uuid)) return prev; + const next = new Map(prev); + next.set(t0.uuid, t); + return next; + }); + break; + } + case "task_history_prepend_start": { + const t0 = findByTid(msg.tid); + if (!t0) break; + const capture = prependCaptureRef.current.get(t0.uuid); + if (capture === undefined || capture.phase !== "requested") + throw new Error(`Unrequested history prepend for task ${msg.tid}`); + // until prepend_end, every frame of this task is the older slice + prependCaptureRef.current.set(t0.uuid, { + phase: "capturing", + frames: [], + }); + break; + } + case "task_history_prepend_end": { + const { tid, window_start: windowStart } = msg; + const t0 = findByTid(tid); + if (!t0) break; + const capture = prependCaptureRef.current.get(t0.uuid); + prependCaptureRef.current.delete(t0.uuid); + if (capture === undefined || capture.phase !== "capturing") + throw new Error(`Unexpected history prepend end for task ${tid}`); + + // continue the initial load: the fetched slice goes in front of + // every frame already reduced, and the whole sequence re-reduces + // through the same path, so the result is exactly what an initial + // load of the larger window would have produced. session context + // (init/metadata replayed from before the window) stays ahead of + // the slice, and context the slice now covers is dropped since the + // slice carries the authoritative copy at those seqs + const cached = historyFramesRef.current.get(t0.uuid) ?? []; + const sliceEnd = t0.historyWindowStart ?? 0; + const context: HistoryFrame[] = []; + const rest: HistoryFrame[] = []; + for (const f of cached) { + if (f.kind === "event" && f.seq !== undefined && f.seq < sliceEnd) { + if (f.seq < windowStart) context.push(f); + } else { + rest.push(f); + } + } + const frames = context.concat(capture.frames, rest); + historyFramesRef.current.set(t0.uuid, frames); + grownHistoryRef.current.add(t0.uuid); + const t = rebuildFromFrames(t0, frames, windowStart); liveStates.set(t0.uuid, t); setTasks((prev) => { if (!prev.has(t0.uuid)) return prev; @@ -1876,6 +1964,11 @@ export function useTaskManager( } case "server_status": { setDevMode(msg.dev_mode ?? false); + setHistoryWindowStep( + deviceClass() === "mobile" + ? (msg.history_window_mobile ?? 0) + : (msg.history_window_desktop ?? 0), + ); const serverBuildId = msg.build_id ?? ""; if ( serverBuildId.length > 0 && @@ -2267,11 +2360,51 @@ export function useTaskManager( if (!t) return; if (requestedHistoryRef.current.has(t.uuid)) return; if (t.historyLoaded) return; - if (connRef.current?.requestHistory(tid)) { + if (connRef.current?.requestHistory(tid, 0, deviceClass())) { requestedHistoryRef.current.add(t.uuid); } }, [connected, activeTaskId, tasks]); + // Extra history loaded with "Load more" persists only while its task is + // focused: on unfocus the task falls back to the plain window, so grown + // timelines never pile up in memory. Refocusing replays the window fresh + // through the effect above (historyLoaded false, request guard cleared). + // the previous route id, not uuid: on the first run the tid-to-uuid map may + // not be populated yet, and this effect's only dep is the route id, so a + // uuid recorded then would stay null forever and mask every later unfocus + const prevActiveTaskIdRef = useRef(null); + useEffect(() => { + const previousId = prevActiveTaskIdRef.current; + prevActiveTaskIdRef.current = activeTaskId; + if (previousId === null || previousId === activeTaskId) return; + const prevTid = parseTaskId(previousId); + if (prevTid === null) return; + const previous = tidToUuid.get(prevTid); + if (previous === undefined) return; + if (!grownHistoryRef.current.has(previous)) return; + + grownHistoryRef.current.delete(previous); + historyFramesRef.current.delete(previous); + prependCaptureRef.current.delete(previous); + requestedHistoryRef.current.delete(previous); + const t = liveStates.get(previous); + if (!t) return; + const released: TaskState = { + ...resetTaskForHistoryReplay(t, 0), + historyTotal: undefined, + historyReceived: undefined, + historyWindowStart: undefined, + historyLoaded: false, + }; + liveStates.set(previous, released); + setTasks((prev) => { + if (!prev.has(previous)) return prev; + const next = new Map(prev); + next.set(previous, released); + return next; + }); + }, [activeTaskId]); + // Replay outbox entries after tasks_list arrives and WS is connected. // The backend deduplicates by nonce, so replaying is safe. useEffect(() => { @@ -3241,6 +3374,8 @@ export function useTaskManager( setServerError(null); }, devMode, + historyWindowStep, + loadMoreHistory, exportLoadError: null, navigateHome, navigateToProject,