perf(chat): eliminate redundant re-renders and markdown re-parsing (#1357)

This commit is contained in:
Carlos Florêncio
2026-06-13 19:26:54 +01:00
committed by GitHub
parent 224383c557
commit f444fba455
52 changed files with 2037 additions and 418 deletions
+8
View File
@@ -204,6 +204,14 @@ Every long-running goroutine must have a single owner with explicit start and st
- After all repos complete `initSchema`, `cmd/kandev/storage.go:recordSchemaVersion` writes the current binary version into `kandev_meta` (non-fatal; a failure just means the next boot will take a fresh snapshot).
- Migration logging: `db.MigrateLogger.Apply(name, stmt)` — success logs Info, "already exists" / "duplicate column name" is silently swallowed, anything else logs Warn but never returns an error (preserving the existing swallow-error contract).
## Schema & migrations (SQLite repository)
`initSchema()` in `internal/task/repository/sqlite/base_schema.go` runs the `init*Schema` (CREATE TABLE) steps **before** `runMigrations()`. The table-creation DDL uses `CREATE TABLE IF NOT EXISTS`, so on an **existing** database it is a no-op and never adds columns to a table that is already present.
**Rule:** when you add a column to an existing table, add it **only** via an idempotent `ADD COLUMN` migration in `runMigrations()` (`base_migrations.go`), never by editing the table's `CREATE TABLE` alone. Anything that *references* that new column — an index, a backfill `UPDATE`, a partial-index predicate — must live in `runMigrations()` **after** the `ADD COLUMN`, not in the `init*Schema` DDL. Putting a `CREATE INDEX ... (new_col)` in the schema-init block crashes existing DBs with `no such column: new_col`, because schema init runs before the migration that adds the column.
You may still list the column in the `CREATE TABLE` so fresh DBs get it inline, but the migration is the source of truth for evolution and must stand alone. New columns also need: the struct field in `models/`, the DTO field + `ToAPI` in `pkg/api/v1/`, and every `CreateX`/`UpdateX`/bulk write in the repo that should set it.
## Code-quality limits
Enforced by `apps/backend/.golangci.yml` (errors on new code only):
@@ -1,9 +1,17 @@
package mcp
import (
"bufio"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"testing/synctest"
"time"
"github.com/gin-gonic/gin"
ws "github.com/kandev/kandev/pkg/websocket"
mcplib "github.com/mark3labs/mcp-go/mcp"
"github.com/stretchr/testify/assert"
@@ -244,3 +252,149 @@ func TestAskUserQuestion_RejectionPath(t *testing.T) {
assert.Contains(t, textBlock.Text, "rejected")
assert.Contains(t, textBlock.Text, "User skipped")
}
// TestEmitKeepAlivePings_TicksUntilStop verifies the keepalive loop fires send on
// each interval and exits once the stop channel closes.
func TestEmitKeepAlivePings_TicksUntilStop(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
stop := make(chan struct{})
calls := 0
send := func() {
calls++
if calls == 3 {
close(stop)
}
}
emitKeepAlivePings(context.Background(), stop, 20*time.Second, send)
assert.Equal(t, 3, calls)
})
}
// TestEmitKeepAlivePings_StopsOnContextCancel verifies a cancelled context tears
// the loop down even if stop never closes.
func TestEmitKeepAlivePings_StopsOnContextCancel(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
stop := make(chan struct{})
defer close(stop)
calls := 0
send := func() {
calls++
if calls == 2 {
cancel()
}
}
emitKeepAlivePings(ctx, stop, 20*time.Second, send)
assert.Equal(t, 2, calls)
})
}
// TestEmitKeepAlivePings_Guards ensures the loop returns immediately for a
// non-positive interval or a nil send without spinning or panicking.
func TestEmitKeepAlivePings_Guards(t *testing.T) {
emitKeepAlivePings(context.Background(), make(chan struct{}), 0, func() {})
emitKeepAlivePings(context.Background(), make(chan struct{}), time.Second, nil)
}
// blockingAskBackend blocks the ask_user_question round-trip until release is
// closed, simulating a user who takes a long time to answer.
type blockingAskBackend struct {
release <-chan struct{}
response map[string]interface{}
}
func (b *blockingAskBackend) RequestPayload(ctx context.Context, action string, _, result interface{}) error {
if action == ws.ActionMCPAskUserQuestion {
select {
case <-b.release:
case <-ctx.Done():
return ctx.Err()
}
}
if b.response != nil && result != nil {
data, _ := json.Marshal(b.response)
return json.Unmarshal(data, result)
}
return nil
}
// TestAskUserQuestion_StreamsKeepAliveDuringWait drives a real Streamable HTTP
// tool call whose backend blocks until the test releases it, and asserts that
// progress notifications stream on the in-flight POST response before the final
// result. This is the regression guard for the agent's MCP client aborting the
// call with "fetch failed" after its idle timeout.
func TestAskUserQuestion_StreamsKeepAliveDuringWait(t *testing.T) {
prev := askQuestionKeepAliveInterval
askQuestionKeepAliveInterval = 5 * time.Millisecond
t.Cleanup(func() { askQuestionKeepAliveInterval = prev })
release := make(chan struct{})
backend := &blockingAskBackend{
release: release,
response: map[string]interface{}{
"answers": []interface{}{
map[string]interface{}{
"question_id": "q1",
"selected_options": []interface{}{"q1_opt1"},
},
},
},
}
log := newTestLogger(t)
s := New(backend, "sess-keepalive", "task-keepalive", 0, log, "", false, ModeTask)
gin.SetMode(gin.TestMode)
router := gin.New()
s.RegisterRoutes(router)
ts := httptest.NewServer(router)
defer ts.Close()
initReq := `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}`
initResp := postJSONRPC(t, ts.URL+"/mcp", initReq, "")
require.Equal(t, http.StatusOK, initResp.statusCode, "init: %s", initResp.body)
callBody := `{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"ask_user_question_kandev","arguments":{"questions":[{"id":"q1","prompt":"Which?","options":[{"label":"A","description":"a"},{"label":"B","description":"b"}]}]}}}`
req, err := http.NewRequest(http.MethodPost, ts.URL+"/mcp", strings.NewReader(callBody))
require.NoError(t, err)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json, text/event-stream")
req.Header.Set("Mcp-Session-Id", initResp.sessionID)
resp, err := http.DefaultClient.Do(req)
require.NoError(t, err)
defer func() { _ = resp.Body.Close() }()
require.Equal(t, http.StatusOK, resp.StatusCode)
scanner := bufio.NewScanner(resp.Body)
released := false
progressSeen := 0
finalSeen := false
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if !strings.HasPrefix(line, "data:") {
continue
}
payload := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
switch {
case strings.Contains(payload, "notifications/progress"):
progressSeen++
if !released {
close(release)
released = true
}
case strings.Contains(payload, `"id":2`):
finalSeen = true
assert.Contains(t, payload, "q1_opt1", "final result should carry the answer")
}
if finalSeen {
break
}
}
require.NoError(t, scanner.Err())
if !released {
close(release)
}
assert.GreaterOrEqual(t, progressSeen, 1, "expected at least one keepalive progress notification")
assert.True(t, finalSeen, "expected the final tool result to be delivered")
}
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"fmt"
"time"
ws "github.com/kandev/kandev/pkg/websocket"
"github.com/mark3labs/mcp-go/mcp"
@@ -11,6 +12,15 @@ import (
"go.uber.org/zap"
)
// askQuestionKeepAliveInterval is how often ask_user_question streams a progress
// notification to the agent while waiting for the user's answer. The agent's MCP
// client (auggie runs on Node, whose fetch/undici applies a 300s idle timeout to
// the in-flight tool-call request) aborts the call with "fetch failed" if no bytes
// arrive for that long. Emitting a progress notification well inside that window
// keeps the streamed POST/SSE response alive so the call survives until the user
// responds. Declared as a var so tests can shorten it.
var askQuestionKeepAliveInterval = 20 * time.Second
// Argument-name constants used across the ask_user_question_kandev handler.
// Pulled out so goconst stays happy and renames stay safe.
const (
@@ -26,6 +36,7 @@ const (
answeredFieldKey = "answered"
rejectedFieldKey = "rejected"
documentArg = "document"
messageArg = "message"
)
func (s *Server) listWorkspacesHandler() server.ToolHandlerFunc {
@@ -320,6 +331,15 @@ func (s *Server) askUserQuestionHandler() server.ToolHandlerFunc {
"context": questionCtx,
}
// Waiting on a human answer routinely outlasts the agent MCP client's
// idle timeout on the in-flight tool call. Stream periodic progress
// notifications until the backend responds; mcp-go flushes them onto the
// POST/SSE response, resetting the client's idle timer so the call is not
// aborted mid-question.
stop := make(chan struct{})
defer close(stop)
go emitKeepAlivePings(ctx, stop, askQuestionKeepAliveInterval, s.clarificationKeepAlive(ctx, req))
// Use the MCP request context from the agent. This ensures that if the agent's
// MCP client times out, we'll detect it and not update the session state.
var result map[string]interface{}
@@ -337,6 +357,54 @@ func (s *Server) askUserQuestionHandler() server.ToolHandlerFunc {
}
}
// emitKeepAlivePings invokes send on every interval tick until stop is closed or
// ctx is cancelled. It is the transport-agnostic core of the ask_user_question
// keepalive, split out so the timing loop is unit-testable without a live MCP
// session.
func emitKeepAlivePings(ctx context.Context, stop <-chan struct{}, interval time.Duration, send func()) {
if interval <= 0 || send == nil {
return
}
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-stop:
return
case <-ctx.Done():
return
case <-ticker.C:
send()
}
}
}
// clarificationKeepAlive builds the keepalive callback that streams a
// notifications/progress message to the agent. The progress token mirrors the one
// the client attached to the tool call when present, so spec-compliant clients
// associate the updates with the in-flight request; clients that omitted a token
// ignore the unknown one. Returns a no-op when no MCP server is bound to the
// context (e.g. direct unit-test invocation of the handler).
func (s *Server) clarificationKeepAlive(ctx context.Context, req mcp.CallToolRequest) func() {
srv := server.ServerFromContext(ctx)
if srv == nil {
return func() {}
}
var token mcp.ProgressToken = fmt.Sprintf("ask_user_question:%s", s.sessionID)
if req.Params.Meta != nil && req.Params.Meta.ProgressToken != nil {
token = req.Params.Meta.ProgressToken
}
var progress float64
return func() {
progress++
_ = srv.SendNotificationToClient(ctx, "notifications/progress", map[string]any{
"progressToken": token,
"progress": progress,
messageArg: "Waiting for your response in Kandev",
})
}
}
// parseQuestions extracts and validates the "questions" array from the request.
// Returns the normalized question payloads (with auto-assigned ids) on success
// or a tool-result error describing the first validation failure.
@@ -445,6 +445,7 @@ type Message struct {
Metadata map[string]interface{} `json:"metadata,omitempty"`
RequestsInput bool `json:"requests_input"` // True if agent is requesting user input
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"` // Authoritative per-message change signal
}
// ToAPI converts internal Message to API type.
@@ -478,6 +479,7 @@ func (m *Message) ToAPI() *v1.Message {
Metadata: meta,
RequestsInput: m.RequestsInput,
CreatedAt: m.CreatedAt,
UpdatedAt: m.UpdatedAt,
}
if hasHidden {
result.RawContent = m.Content
@@ -86,6 +86,15 @@ func (r *Repository) runMigrations() error {
r.migrate.Apply("task_sessions.workspace_path", `ALTER TABLE task_sessions ADD COLUMN workspace_path TEXT DEFAULT ''`)
r.migrate.Apply("repositories.copy_files", `ALTER TABLE repositories ADD COLUMN copy_files TEXT DEFAULT ''`)
// Authoritative per-message change signal (chat render-perf). SQLite forbids a
// non-constant default on ADD COLUMN, so the column is added nullable and
// existing rows are backfilled to created_at; new inserts/updates set it
// explicitly in CreateMessage/UpdateMessage. The backfill UPDATE is idempotent
// (WHERE updated_at IS NULL).
r.migrate.Apply("task_session_messages.updated_at", `ALTER TABLE task_session_messages ADD COLUMN updated_at TIMESTAMP`)
r.migrate.Apply("task_session_messages.updated_at.backfill", `UPDATE task_session_messages SET updated_at = created_at WHERE updated_at IS NULL`)
r.migrate.Apply("idx_messages_session_updated", `CREATE INDEX IF NOT EXISTS idx_messages_session_updated ON task_session_messages(task_session_id, updated_at)`)
// Backfill the per-session cost/token columns. Runs after the gated
// task_sessions rebuilds above so it repairs legacy DBs whose schema can no
// longer trigger a rebuild (see migrateSessionsAddCostColumns).
@@ -465,6 +465,7 @@ func (r *Repository) initMessageTurnSchema() error {
type TEXT NOT NULL DEFAULT 'message',
metadata TEXT DEFAULT '{}',
created_at TIMESTAMP NOT NULL,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (task_session_id) REFERENCES task_sessions(id) ON DELETE CASCADE,
FOREIGN KEY (turn_id) REFERENCES task_session_turns(id) ON DELETE CASCADE
);
@@ -473,6 +474,10 @@ func (r *Repository) initMessageTurnSchema() error {
CREATE INDEX IF NOT EXISTS idx_messages_created_at ON task_session_messages(created_at);
CREATE INDEX IF NOT EXISTS idx_messages_session_created ON task_session_messages(task_session_id, created_at);
CREATE INDEX IF NOT EXISTS idx_messages_turn_id ON task_session_messages(turn_id);
-- idx_messages_session_updated is created in runMigrations() after the
-- updated_at ADD COLUMN + backfill. Creating it here would fail on existing
-- DBs where CREATE TABLE IF NOT EXISTS is a no-op and the column does not
-- yet exist (schema init runs before migrations).
CREATE TABLE IF NOT EXISTS task_session_turns (
id TEXT PRIMARY KEY,
@@ -24,6 +24,9 @@ func (r *Repository) CreateMessage(ctx context.Context, message *models.Message)
if message.CreatedAt.IsZero() {
message.CreatedAt = time.Now().UTC()
}
if message.UpdatedAt.IsZero() {
message.UpdatedAt = message.CreatedAt
}
if message.AuthorType == "" {
message.AuthorType = models.MessageAuthorUser
}
@@ -48,9 +51,9 @@ func (r *Repository) CreateMessage(ctx context.Context, message *models.Message)
}
_, err := r.db.ExecContext(ctx, r.db.Rebind(`
INSERT INTO task_session_messages (id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`), message.ID, message.TaskSessionID, message.TaskID, message.TurnID, message.AuthorType, message.AuthorID, message.Content, requestsInput, messageType, metadataJSON, message.CreatedAt)
INSERT INTO task_session_messages (id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`), message.ID, message.TaskSessionID, message.TaskID, message.TurnID, message.AuthorType, message.AuthorID, message.Content, requestsInput, messageType, metadataJSON, message.CreatedAt, message.UpdatedAt)
return err
}
@@ -62,9 +65,9 @@ func (r *Repository) GetMessage(ctx context.Context, id string) (*models.Message
var messageType string
var metadataJSON string
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages WHERE id = ?
`), id).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
`), id).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt)
if err != nil {
return nil, err
}
@@ -83,7 +86,7 @@ func (r *Repository) GetMessage(ctx context.Context, id string) (*models.Message
// ListMessages returns all messages for a session ordered by creation time.
func (r *Repository) ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) {
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages WHERE task_session_id = ? ORDER BY created_at ASC
`), sessionID)
if err != nil {
@@ -99,7 +102,7 @@ func (r *Repository) ListMessages(ctx context.Context, sessionID string) ([]*mod
// own rows rather than the whole session's history.
func (r *Repository) ListMessagesByTurnID(ctx context.Context, turnID string) ([]*models.Message, error) {
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages WHERE turn_id = ? ORDER BY created_at ASC
`), turnID)
if err != nil {
@@ -162,7 +165,7 @@ func (r *Repository) resolveMessageCursor(ctx context.Context, sessionID string,
func buildListMessagesQuery(sessionID string, opts models.ListMessagesOptions, cursor *models.Message, sortDir string, limit int) (string, []interface{}) {
query := `
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages WHERE task_session_id = ?`
args := []interface{}{sessionID}
if cursor != nil {
@@ -192,7 +195,7 @@ func scanMessageRows(rows interface {
var requestsInput int
var messageType string
var metadataJSON string
if err := rows.Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt); err != nil {
if err := rows.Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt); err != nil {
return nil, false, err
}
message.RequestsInput = requestsInput == 1
@@ -233,7 +236,7 @@ func (r *Repository) SearchMessages(ctx context.Context, sessionID string, opts
escaper := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`)
pattern := "%" + escaper.Replace(query) + "%"
sql := fmt.Sprintf(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages
WHERE task_session_id = ? AND content %s ? ESCAPE '\'
ORDER BY created_at DESC, id DESC
@@ -269,12 +272,12 @@ func (r *Repository) getMessageByMetadataField(ctx context.Context, sessionID, f
var metadataJSON string
drv := r.ro.DriverName()
query := fmt.Sprintf(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages WHERE task_session_id = ? AND %s = ?
%s
`, dialect.JSONExtract(drv, "metadata", fieldName), orderSuffix)
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(query), sessionID, fieldValue).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID,
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt)
if err != nil {
return nil, err
}
@@ -315,12 +318,12 @@ func (r *Repository) FindMessageByPendingID(ctx context.Context, pendingID strin
var metadataJSON string
drv := r.ro.DriverName()
query := fmt.Sprintf(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages WHERE %s = ?
ORDER BY created_at DESC LIMIT 1
`, dialect.JSONExtract(drv, "metadata", "pending_id"))
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(query), pendingID).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID,
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt)
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt)
if err != nil {
return nil, err
}
@@ -340,7 +343,7 @@ func (r *Repository) FindMessageByPendingID(ctx context.Context, pendingID strin
func (r *Repository) FindMessagesByPendingID(ctx context.Context, pendingID string) ([]*models.Message, error) {
drv := r.ro.DriverName()
query := fmt.Sprintf(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages WHERE %s = ?
ORDER BY created_at ASC
`, dialect.JSONExtract(drv, "metadata", "pending_id"))
@@ -360,7 +363,7 @@ func (r *Repository) FindMessagesByPendingID(ctx context.Context, pendingID stri
func (r *Repository) FindPendingClarificationMessagesBySessionID(ctx context.Context, sessionID string) ([]*models.Message, error) {
drv := r.ro.DriverName()
query := fmt.Sprintf(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages
WHERE task_session_id = ? AND type = 'clarification_request' AND %s = 'pending'
ORDER BY created_at ASC
@@ -382,7 +385,7 @@ func (r *Repository) FindPendingClarificationMessagesBySessionID(ctx context.Con
func (r *Repository) FindMessageByPendingIDAndQuestion(ctx context.Context, sessionID, pendingID, questionID string) (*models.Message, error) {
drv := r.ro.DriverName()
query := fmt.Sprintf(`
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at
SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at
FROM task_session_messages
WHERE task_session_id = ? AND %s = ? AND %s = ?
ORDER BY created_at ASC LIMIT 1
@@ -396,7 +399,7 @@ func (r *Repository) FindMessageByPendingIDAndQuestion(ctx context.Context, sess
var metadataJSON string
err := r.ro.QueryRowContext(ctx, r.ro.Rebind(query), sessionID, pendingID, questionID).Scan(
&message.ID, &message.TaskSessionID, &message.TaskID, &message.TurnID, &message.AuthorType, &message.AuthorID,
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt,
&message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt, &message.UpdatedAt,
)
if err != nil {
return nil, err
@@ -420,7 +423,7 @@ func (r *Repository) CompletePendingToolCallsForTurn(ctx context.Context, turnID
drv := r.db.DriverName()
query := fmt.Sprintf(`
UPDATE task_session_messages
SET metadata = %s
SET metadata = %s, updated_at = CURRENT_TIMESTAMP
WHERE turn_id = ?
AND type != 'permission_request'
AND %s NOT IN ('complete', 'error')
@@ -449,10 +452,11 @@ func (r *Repository) UpdateMessage(ctx context.Context, message *models.Message)
requestsInput = 1
}
message.UpdatedAt = time.Now().UTC()
result, err := r.db.ExecContext(ctx, r.db.Rebind(`
UPDATE task_session_messages SET content = ?, requests_input = ?, type = ?, metadata = ?
UPDATE task_session_messages SET content = ?, requests_input = ?, type = ?, metadata = ?, updated_at = ?
WHERE id = ?
`), message.Content, requestsInput, string(message.Type), string(metadataJSON), message.ID)
`), message.Content, requestsInput, string(message.Type), string(metadataJSON), message.UpdatedAt, message.ID)
if err != nil {
return err
}
@@ -4,6 +4,8 @@ import (
"context"
"testing"
"time"
"github.com/kandev/kandev/internal/task/models"
)
// insertMsgWithType inserts a message row with a configurable type column,
@@ -58,6 +60,48 @@ func TestListMessagesByTurnID(t *testing.T) {
}
}
func TestUpdateMessageBumpsUpdatedAt(t *testing.T) {
repo := newRepoForSessionTests(t)
ctx := context.Background()
seedForMsgTest(t, repo, "task-U", "sess-U", "turn-U")
created := time.Now().UTC().Add(-time.Hour)
msg := &models.Message{
ID: "m-u",
TaskSessionID: "sess-U",
TurnID: "turn-U",
AuthorType: models.MessageAuthorAgent,
Content: "hello",
Type: models.MessageTypeMessage,
CreatedAt: created,
}
if err := repo.CreateMessage(ctx, msg); err != nil {
t.Fatalf("CreateMessage: %v", err)
}
// On insert, updated_at defaults to created_at.
got, err := repo.GetMessage(ctx, "m-u")
if err != nil {
t.Fatalf("GetMessage: %v", err)
}
if !got.UpdatedAt.Equal(got.CreatedAt) {
t.Errorf("after create, updated_at = %v, want == created_at %v", got.UpdatedAt, got.CreatedAt)
}
// Update advances updated_at past created_at.
msg.Content = "hello world"
if err := repo.UpdateMessage(ctx, msg); err != nil {
t.Fatalf("UpdateMessage: %v", err)
}
got, err = repo.GetMessage(ctx, "m-u")
if err != nil {
t.Fatalf("GetMessage after update: %v", err)
}
if !got.UpdatedAt.After(got.CreatedAt) {
t.Errorf("after update, updated_at = %v, want after created_at %v", got.UpdatedAt, got.CreatedAt)
}
}
func TestCountToolCallMessagesBySession_Empty(t *testing.T) {
repo := newRepoForSessionTests(t)
got, err := repo.CountToolCallMessagesBySession(context.Background(), nil)
@@ -324,7 +324,11 @@ func (s *Service) publishMessageEvent(ctx context.Context, eventType string, mes
"content": sysprompt.StripSystemContent(message.Content),
"type": messageType,
"requests_input": message.RequestsInput,
"created_at": message.CreatedAt.Format(time.RFC3339),
// RFC3339Nano keeps sub-second precision so rapid updates within the same
// second produce distinct timestamps; the REST/DTO path serializes these
// fields with nanosecond precision too, so both delivery channels agree.
"created_at": message.CreatedAt.Format(time.RFC3339Nano),
"updated_at": message.UpdatedAt.Format(time.RFC3339Nano),
}
if hasHidden {
+1
View File
@@ -164,6 +164,7 @@ type Message struct {
RequestsInput bool `json:"requests_input"` // True if agent is requesting user input
Metadata map[string]interface{} `json:"metadata,omitempty"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at,omitempty"` // Authoritative per-message change signal
}
// CreateMessageRequest for adding a message to a task session
+63
View File
@@ -0,0 +1,63 @@
import { afterAll, describe, expect, it } from "vitest";
import { __resetBrokenPipeGuard, ignoreBrokenPipe } from "./process";
function makeError(code: string, message: string): NodeJS.ErrnoException {
return Object.assign(new Error(message), { code });
}
function lastErrorListener(stream: NodeJS.WriteStream): (err: NodeJS.ErrnoException) => void {
const listeners = stream.listeners("error");
return listeners[listeners.length - 1] as (err: NodeJS.ErrnoException) => void;
}
describe("ignoreBrokenPipe", () => {
// Capture the handler we install so we can both exercise and clean it up.
let installed: ((err: NodeJS.ErrnoException) => void) | undefined;
afterAll(() => {
if (installed) {
process.stdout.removeListener("error", installed);
process.stderr.removeListener("error", installed);
}
// Reset the module guard alongside listener removal so a later
// ignoreBrokenPipe() reattaches instead of hitting a no-op with no
// listeners present (order-independent for other tests in this worker).
__resetBrokenPipeGuard();
});
it("attaches an error handler to stdout and stderr", () => {
const stdoutBefore = process.stdout.listeners("error").length;
const stderrBefore = process.stderr.listeners("error").length;
ignoreBrokenPipe();
expect(process.stdout.listeners("error").length).toBe(stdoutBefore + 1);
expect(process.stderr.listeners("error").length).toBe(stderrBefore + 1);
// stdout and stderr share the same handler reference.
installed = lastErrorListener(process.stdout);
expect(lastErrorListener(process.stderr)).toBe(installed);
});
it("swallows EPIPE errors without throwing", () => {
const handler = lastErrorListener(process.stdout);
expect(() => handler(makeError("EPIPE", "write EPIPE"))).not.toThrow();
});
it("rethrows non-EPIPE errors", () => {
const handler = lastErrorListener(process.stdout);
expect(() => handler(makeError("EACCES", "boom"))).toThrow("boom");
});
it("is idempotent across repeated calls", () => {
const stdoutCount = process.stdout.listeners("error").length;
const stderrCount = process.stderr.listeners("error").length;
ignoreBrokenPipe();
ignoreBrokenPipe();
expect(process.stdout.listeners("error").length).toBe(stdoutCount);
expect(process.stderr.listeners("error").length).toBe(stderrCount);
});
});
+32
View File
@@ -36,6 +36,12 @@ export function createProcessSupervisor(): {
};
const attachSignalHandlers = () => {
// Ctrl-C sends SIGINT to the whole process group, so the parent (make/shell)
// exits and closes the pipe our stdout/stderr write to. Any console.log during
// shutdown then triggers an EPIPE 'error' event with no listener, which Node
// treats as fatal and crashes us before children are gracefully terminated.
// Swallow EPIPE so shutdown can finish.
ignoreBrokenPipe();
process.on("SIGINT", onSignal);
// SIGTERM is not available on Windows — only attach where supported
if (process.platform !== "win32") {
@@ -46,6 +52,32 @@ export function createProcessSupervisor(): {
return { children, shutdown, attachSignalHandlers };
}
let brokenPipeGuarded = false;
/**
* Install error handlers on stdout/stderr that swallow EPIPE (broken pipe).
* Idempotent: only attaches once per process.
*/
export function ignoreBrokenPipe(): void {
if (brokenPipeGuarded) return;
brokenPipeGuarded = true;
const onPipeError = (err: NodeJS.ErrnoException) => {
if (err.code === "EPIPE") return;
throw err;
};
process.stdout.on("error", onPipeError);
process.stderr.on("error", onPipeError);
}
/**
* Test-only: resets the broken-pipe guard so a later `ignoreBrokenPipe()` call
* reattaches listeners. Keeps the module guard consistent with suites that
* remove the listeners they installed.
*/
export function __resetBrokenPipeGuard(): void {
brokenPipeGuarded = false;
}
/**
* Terminate a process and wait for it to exit.
* On Unix: sends SIGTERM, falls back to SIGKILL after timeout.
+80 -68
View File
@@ -360,10 +360,10 @@
@keyframes border-flash-glow {
0%,
100% {
box-shadow: inset 0 0 0 1px color-mix(in oklab, var(--primary) 20%, transparent);
opacity: 0.35;
}
50% {
box-shadow: inset 0 0 0 1px color-mix(in oklab, var(--primary) 60%, transparent);
opacity: 1;
}
}
@@ -410,71 +410,72 @@
border-radius: 4px;
}
@keyframes chat-input-running {
0%,
100% {
border-color: color-mix(in oklab, var(--primary) 40%, var(--border));
box-shadow:
0 0 6px 1px color-mix(in oklab, var(--primary) 12%, transparent),
0 0 15px 3px color-mix(in oklab, var(--primary) 6%, transparent);
}
50% {
border-color: color-mix(in oklab, var(--primary) 70%, var(--border));
box-shadow:
0 0 20px 4px color-mix(in oklab, var(--primary) 20%, transparent),
0 0 50px 8px color-mix(in oklab, var(--primary) 10%, transparent);
}
}
.chat-input-running {
animation: chat-input-running 3s ease-in-out infinite;
}
@keyframes chat-input-running-plan {
0%,
100% {
border-color: color-mix(in oklab, var(--primary) 40%, var(--border));
box-shadow:
0 0 6px 1px color-mix(in oklab, var(--primary) 12%, transparent),
0 0 15px 3px color-mix(in oklab, var(--primary) 6%, transparent);
}
50% {
border-color: color-mix(in oklab, var(--primary) 70%, var(--border));
box-shadow:
0 0 20px 4px color-mix(in oklab, var(--primary) 20%, transparent),
0 0 50px 8px color-mix(in oklab, var(--primary) 10%, transparent);
}
}
/* Chat-input "running" affordance.
*
* The colored border is a static border-color (painted once). The pulsing glow
* lives on a pseudo-element that animates only `opacity`, which the compositor
* handles on the GPU without a per-frame main-thread paint. The glow is driven
* from the OUTER `.relative` wrapper (`.chat-input-glow*`) because the inner
* input box has `overflow-hidden`, which would clip a child's outer box-shadow. */
.chat-input-running,
.chat-input-running-plan {
animation: chat-input-running-plan 3s ease-in-out infinite;
}
@keyframes chat-input-starting {
0%,
100% {
border-color: color-mix(in oklab, var(--primary) 40%, var(--border));
box-shadow:
0 0 4px 1px color-mix(in oklab, var(--primary) 10%, transparent),
0 0 10px 2px color-mix(in oklab, var(--primary) 5%, transparent);
}
50% {
border-color: color-mix(in oklab, var(--primary) 60%, var(--border));
box-shadow:
0 0 12px 3px color-mix(in oklab, var(--primary) 15%, transparent),
0 0 30px 6px color-mix(in oklab, var(--primary) 8%, transparent);
}
border-color: color-mix(in oklab, var(--primary) 55%, var(--border));
}
.chat-input-starting {
animation: chat-input-starting 2s ease-in-out infinite;
border-color: color-mix(in oklab, var(--primary) 50%, var(--border));
}
/* The glow pseudo-element sits at `z-index: -1` so the input content paints on
* top of it. `isolation: isolate` makes the wrapper its own stacking context so
* that negative z-index stays local instead of dropping behind ancestor
* backgrounds (which would hide the glow entirely). */
.chat-input-glow-running,
.chat-input-glow-starting {
position: relative;
isolation: isolate;
}
@keyframes chat-input-glow-pulse {
0%,
100% {
opacity: 0.4;
}
50% {
opacity: 1;
}
}
.chat-input-glow-running::after,
.chat-input-glow-starting::after {
content: "";
position: absolute;
inset: 0;
border-radius: 0.25rem;
pointer-events: none;
z-index: -1;
animation: chat-input-glow-pulse 3s ease-in-out infinite;
}
.chat-input-glow-running::after {
box-shadow:
0 0 20px 4px color-mix(in oklab, var(--primary) 20%, transparent),
0 0 50px 8px color-mix(in oklab, var(--primary) 10%, transparent);
}
.chat-input-glow-starting::after {
box-shadow:
0 0 12px 3px color-mix(in oklab, var(--primary) 15%, transparent),
0 0 30px 6px color-mix(in oklab, var(--primary) 8%, transparent);
animation-duration: 2s;
}
@media (prefers-reduced-motion: reduce) {
.animate-changes-flash,
.chat-input-running,
.chat-input-running-plan,
.chat-input-starting,
.chat-input-glow-running::after,
.chat-input-glow-starting::after,
.border-flash-animation::after,
.node-border-running::after,
.search-flash,
.animate-queue-open,
.animate-queue-close {
@@ -494,6 +495,7 @@
border-radius: inherit;
pointer-events: none;
z-index: 10;
box-shadow: inset 0 0 0 1px color-mix(in oklab, var(--primary) 60%, transparent);
animation: border-flash-glow 3s ease-in-out infinite;
}
}
@@ -1392,24 +1394,34 @@
border-radius: 8px;
}
/* Pipeline node rotating border for running tasks */
@keyframes node-border-pulse {
/* Pipeline node glow for running tasks — static ring on the element, the pulse
* is opacity-only on a pseudo-element so it composites on the GPU rather than
* repainting the box-shadow every frame. */
@keyframes node-border-glow {
0%,
100% {
box-shadow:
0 0 0 1px color-mix(in oklab, var(--primary) 42%, var(--border)),
0 0 0 0 color-mix(in oklab, var(--primary) 0%, transparent);
opacity: 0.5;
}
50% {
box-shadow:
0 0 0 1px color-mix(in oklab, var(--primary) 68%, var(--border)),
0 0 0 3px color-mix(in oklab, var(--primary) 12%, transparent);
opacity: 1;
}
}
.node-border-running {
position: relative;
animation: node-border-pulse 2.8s ease-in-out infinite;
box-shadow: 0 0 0 1px color-mix(in oklab, var(--primary) 42%, var(--border));
}
.node-border-running::after {
content: "";
position: absolute;
inset: 0;
border-radius: inherit;
pointer-events: none;
box-shadow:
0 0 0 1px color-mix(in oklab, var(--primary) 68%, var(--border)),
0 0 0 3px color-mix(in oklab, var(--primary) 12%, transparent);
animation: node-border-glow 2.8s ease-in-out infinite;
}
.chat-message-list {
@@ -12,7 +12,7 @@ import { useAppStore } from "@/components/state-provider";
import { getWebSocketClient } from "@/lib/ws/connection";
import { buildStartRequest } from "@/lib/services/session-launch-helpers";
import { MessageRenderer } from "@/components/task/chat/message-renderer";
import type { Message, TaskSessionState } from "@/lib/types/http";
import type { Message } from "@/lib/types/http";
type AdvancedChatPanelProps = {
taskId: string;
@@ -58,14 +58,12 @@ function StartSessionPrompt({
function MessageList({
messages,
isLoading,
sessionState,
taskId,
sessionId,
scrollRef,
}: {
messages: Message[];
isLoading: boolean;
sessionState: TaskSessionState | null;
taskId: string;
sessionId: string | null;
scrollRef: React.RefObject<HTMLDivElement | null>;
@@ -87,7 +85,6 @@ function MessageList({
key={msg.id}
comment={msg}
isTaskDescription={idx === 0 && msg.author_type === "user"}
sessionState={sessionState ?? undefined}
taskId={taskId}
sessionId={sessionId ?? undefined}
/>
@@ -198,7 +195,6 @@ export function AdvancedChatPanel({ taskId, sessionId, hideInput }: AdvancedChat
<MessageList
messages={messages}
isLoading={isLoading}
sessionState={sessionState}
taskId={taskId}
sessionId={sessionId}
scrollRef={scrollRef}
@@ -95,7 +95,6 @@ export const QuickChatContent = memo(function QuickChatContent({
messagesLoading={panelState.messagesLoading}
isWorking={panelState.isWorking}
sessionState={panelState.session?.state}
taskState={panelState.task?.state}
worktreePath={panelState.session?.worktree_path}
onOpenFile={undefined}
/>
@@ -13,69 +13,10 @@ import { isMermaidContent } from "@/components/editors/tiptap/tiptap-mermaid-ext
// eslint-disable-next-line @typescript-eslint/no-explicit-any
export const remarkPlugins: any[] = [remarkGfm, remarkBreaks, remarkGemoji];
const FENCE_OPEN_RE = /^ {0,3}(`{3,})/;
const TRAILING_FENCE_RE = /(`{3,})\s*$/;
function pureCloseLength(line: string, openCount: number): number | null {
const match = /^ {0,3}(`{3,})\s*$/.exec(line);
if (!match || match[1].length < openCount) return null;
return match[1].length;
}
function gluedCloseLength(line: string, openCount: number): number | null {
const match = TRAILING_FENCE_RE.exec(line);
if (!match || match[1].length < openCount) return null;
// Reject pure-fence lines (already handled by pureCloseLength) and lines
// where everything before the trailing run is whitespace only.
const head = line.slice(0, line.length - match[0].length).trimEnd();
if (head.length === 0) return null;
return match[1].length;
}
/**
* Pre-process a markdown string to repair fenced code blocks that have their
* closing fence glued to the last code line (`...}\`\`\`\n`prose`). Without
* this, CommonMark/GFM treats the glued backticks as code content, so the
* fence never closes and following prose gets swallowed into one huge code
* node. We split such lines into `<content>\n<backticks>` only when we're
* inside an open fence whose opener run length is ≤ the trailing run length.
*
* Pure string preprocessing, intentionally not a remark plugin.
*/
export function normalizeMarkdown(input: string): string {
if (!input || input.length === 0) return input;
const hadTrailingNewline = input.endsWith("\n");
const lines = input.split("\n");
const out: string[] = [];
let openCount: number | null = null;
for (const line of lines) {
if (openCount === null) {
const opener = FENCE_OPEN_RE.exec(line);
if (opener) openCount = opener[1].length;
out.push(line);
continue;
}
if (pureCloseLength(line, openCount) !== null) {
openCount = null;
out.push(line);
continue;
}
const glued = gluedCloseLength(line, openCount);
if (glued !== null) {
const trailingMatch = TRAILING_FENCE_RE.exec(line)!;
const head = line.slice(0, line.length - trailingMatch[0].length);
out.push(head);
out.push("`".repeat(glued));
openCount = null;
continue;
}
out.push(line);
}
const result = out.join("\n");
return hadTrailingNewline && !result.endsWith("\n") ? result + "\n" : result;
}
// `normalizeMarkdown` (pure string transform) and its cached variant live in
// the React-free markdown cache module. Re-exported here so existing importers
// keep working.
export { normalizeMarkdown } from "@/lib/markdown/normalize-cache";
/**
* Recursively extracts text content from React children.
@@ -0,0 +1,53 @@
import { useState } from "react";
import { fireEvent, render } from "@testing-library/react";
import { afterEach, describe, expect, it, vi } from "vitest";
const markdownSpy = vi.fn();
vi.mock("react-markdown", () => ({
default: ({ children }: { children: string }) => {
markdownSpy(children);
return <div data-testid="md">{children}</div>;
},
}));
// Stub the component overrides so the test stays light (no shiki / mermaid).
vi.mock("@/components/shared/markdown-components", () => ({
markdownComponents: {},
remarkPlugins: [],
}));
import { MemoizedMarkdown } from "./memoized-markdown";
import { __resetMarkdownCounters } from "@/lib/markdown/normalize-cache";
function Parent({ content }: { content: string }) {
const [tick, setTick] = useState(0);
return (
<div data-tick={tick}>
<button onClick={() => setTick((t) => t + 1)}>tick</button>
<MemoizedMarkdown content={content} />
</div>
);
}
describe("MemoizedMarkdown", () => {
afterEach(() => {
markdownSpy.mockClear();
__resetMarkdownCounters();
});
it("does not re-render markdown when the parent re-renders with same content", () => {
const { getByText } = render(<Parent content="hello world" />);
expect(markdownSpy).toHaveBeenCalledTimes(1);
fireEvent.click(getByText("tick"));
fireEvent.click(getByText("tick"));
expect(markdownSpy).toHaveBeenCalledTimes(1);
});
it("re-renders markdown when content changes", () => {
const { rerender } = render(<MemoizedMarkdown content="first" />);
expect(markdownSpy).toHaveBeenCalledTimes(1);
rerender(<MemoizedMarkdown content="second" />);
expect(markdownSpy).toHaveBeenCalledTimes(2);
});
});
@@ -0,0 +1,23 @@
"use client";
import { memo } from "react";
import ReactMarkdown from "react-markdown";
import { markdownComponents, remarkPlugins } from "@/components/shared/markdown-components";
import { normalizeCached } from "@/lib/markdown/normalize-cache";
/**
* Markdown renderer behind a `memo` boundary keyed on the `content` string.
*
* `content` is a primitive, so React compares it by value. Object-ref churn
* from a message refetch can therefore never trigger a re-parse: an identical
* string bails out of the memo and re-uses the previously parsed element tree.
* The normalized string itself is also cached (LRU) so two messages with the
* same content share a single normalize pass.
*/
export const MemoizedMarkdown = memo(function MemoizedMarkdown({ content }: { content: string }) {
return (
<ReactMarkdown remarkPlugins={remarkPlugins} components={markdownComponents}>
{normalizeCached(content)}
</ReactMarkdown>
);
});
@@ -0,0 +1,123 @@
/**
* Render-cost eval harness — the objective guard against the markdown
* re-render/re-parse "storm".
*
* It mounts a list of memoized markdown rows and asserts how many real
* normalize passes (cache misses) and per-row re-renders happen under each
* update pattern. Wave 1 owns the parse-count assertions; Wave 2/3 extend the
* row-render assertions once callbacks + store identity are stabilized.
*/
import { memo, useState } from "react";
import { act, fireEvent, render } from "@testing-library/react";
import { beforeEach, describe, expect, it, vi } from "vitest";
vi.mock("react-markdown", () => ({
default: ({ children }: { children: string }) => <div>{children}</div>,
}));
vi.mock("@/components/shared/markdown-components", () => ({
markdownComponents: {},
remarkPlugins: [],
}));
import { MemoizedMarkdown } from "@/components/shared/memoized-markdown";
import { __markdownParseCount, __resetMarkdownCounters } from "@/lib/markdown/normalize-cache";
type Row = { id: string; content: string };
const rowRenderCounts = new Map<string, number>();
const MarkdownRow = memo(function MarkdownRow({ id, content }: Row) {
rowRenderCounts.set(id, (rowRenderCounts.get(id) ?? 0) + 1);
return <MemoizedMarkdown content={content} />;
});
function MessageList({ rows, panel }: { rows: Row[]; panel: boolean }) {
return (
<div data-panel={String(panel)}>
{rows.map((r) => (
<MarkdownRow key={r.id} id={r.id} content={r.content} />
))}
</div>
);
}
function makeRows(n: number): Row[] {
return Array.from({ length: n }, (_, i) => ({ id: `m${i}`, content: `# Message ${i}` }));
}
/** Deep clone into fresh object refs but identical content (simulates a refetch). */
function refetch(rows: Row[]): Row[] {
return rows.map((r) => ({ id: r.id, content: r.content }));
}
function totalRowRenders(): number {
let total = 0;
for (const count of rowRenderCounts.values()) total += count;
return total;
}
describe("chat render-cost evals", () => {
beforeEach(() => {
__resetMarkdownCounters();
rowRenderCounts.clear();
});
it("scenario 1: no-op refetch causes 0 parses and 0 row re-renders", () => {
const rows = makeRows(20);
const { rerender } = render(<MessageList rows={rows} panel={false} />);
const parsesAfterMount = __markdownParseCount();
rowRenderCounts.clear();
rerender(<MessageList rows={refetch(rows)} panel={false} />);
expect(__markdownParseCount() - parsesAfterMount).toBe(0);
expect(totalRowRenders()).toBe(0);
});
it("scenario 2: one changed message causes exactly 1 new parse", () => {
const rows = makeRows(20);
const { rerender } = render(<MessageList rows={rows} panel={false} />);
const parsesAfterMount = __markdownParseCount();
rowRenderCounts.clear();
const next = refetch(rows);
next[5] = { id: next[5].id, content: "# Message 5 (edited)" };
rerender(<MessageList rows={next} panel={false} />);
expect(__markdownParseCount() - parsesAfterMount).toBe(1);
expect(rowRenderCounts.get("m5")).toBe(1);
});
it("scenario 3: unrelated parent re-render causes 0 parses", () => {
const rows = makeRows(20);
function Harness() {
const [panel, setPanel] = useState(false);
return (
<div>
<button onClick={() => setPanel((p) => !p)}>toggle</button>
<MessageList rows={rows} panel={panel} />
</div>
);
}
const { getByText } = render(<Harness />);
const parsesAfterMount = __markdownParseCount();
rowRenderCounts.clear();
act(() => {
fireEvent.click(getByText("toggle"));
});
expect(__markdownParseCount() - parsesAfterMount).toBe(0);
expect(totalRowRenders()).toBe(0);
});
it("scenario 4: identical content across two rows parses once (cache hit)", () => {
const rows: Row[] = [
{ id: "a", content: "# Same content" },
{ id: "b", content: "# Same content" },
];
render(<MessageList rows={rows} panel={false} />);
expect(__markdownParseCount()).toBe(1);
});
});
@@ -233,6 +233,15 @@ export type ChatInputBodyProps = {
editorAreaProps: ChatInputEditorAreaProps;
};
/** Glow class for the outer wrapper. The pulsing glow lives on the wrapper
* (not the inner box) because the inner box has `overflow-hidden`, which would
* clip a child pseudo-element's outer box-shadow. */
function chatInputGlowClass(isAgentBusy: boolean, isStarting: boolean): string {
if (isAgentBusy) return "chat-input-glow-running";
if (isStarting) return "chat-input-glow-starting";
return "";
}
export function ChatInputBody({
containerRef,
height,
@@ -294,7 +303,7 @@ export function ChatInputBody({
);
return (
<div className="relative">
<div className={cn("relative", chatInputGlowClass(isAgentBusy, isStarting))}>
<ResizeHandle
planModeEnabled={planModeEnabled}
isAgentBusy={isAgentBusy}
@@ -181,7 +181,6 @@ export const NativeMessageList = memo(function NativeMessageList({
messagesLoading,
isWorking,
sessionState,
taskState,
worktreePath,
onOpenFile,
}: MessageListProps) {
@@ -245,9 +244,6 @@ export const NativeMessageList = memo(function NativeMessageList({
onOpenFile={onOpenFile}
isLastGroup={item.type === "turn_group" && item.id === lastTurnGroupId}
isTurnActive={isRunning}
messages={messages}
sessionState={sessionState}
taskState={taskState}
onScrollToMessage={handleScrollToMessage}
/>
</div>
@@ -256,12 +252,7 @@ export const NativeMessageList = memo(function NativeMessageList({
<AgentStatus sessionState={sessionState} sessionId={sessionId} messages={messages} />
{(footerActionMessages ?? []).map((msg: Message) => (
<MessageRenderer
key={msg.id}
comment={msg}
isTaskDescription={false}
sessionState={sessionState}
/>
<MessageRenderer key={msg.id} comment={msg} isTaskDescription={false} />
))}
{/* Bottom anchor — browser keeps scroll pinned here when new content appends */}
@@ -0,0 +1,75 @@
import { useState } from "react";
import { fireEvent, render } from "@testing-library/react";
import { afterEach, describe, expect, it, vi } from "vitest";
import type { RenderItem } from "@/hooks/use-processed-messages";
import type { Message } from "@/lib/types/http";
const rendererSpy = vi.fn();
vi.mock("@/components/task/chat/message-renderer", () => ({
MessageRenderer: (props: { onOpenFile?: unknown }) => {
rendererSpy(props);
return <div data-testid="renderer" />;
},
}));
vi.mock("@/components/task/chat/messages/turn-group-message", () => ({
TurnGroupMessage: () => <div data-testid="turn-group" />,
}));
vi.mock("@/components/session/prepare-progress", () => ({
PrepareProgress: () => <div data-testid="prepare" />,
}));
import { MessageItem } from "./message-list-shared";
const item: RenderItem = { type: "message", message: { id: "m1" } as Message };
const noop = () => {};
const perm = new Map<string, Message>();
const kids = new Map<string, Message[]>();
function row(onOpenFile: (p: string) => void) {
return (
<MessageItem
item={item}
sessionId="s1"
permissionsByToolCallId={perm}
childrenByParentToolCallId={kids}
taskId="t1"
worktreePath="/wt"
onOpenFile={onOpenFile}
isLastGroup={false}
isTurnActive={false}
onScrollToMessage={noop}
/>
);
}
function Harness({ onOpenFile }: { onOpenFile: (p: string) => void }) {
const [, setTick] = useState(0);
return (
<div>
<button onClick={() => setTick((t) => t + 1)}>tick</button>
{row(onOpenFile)}
</div>
);
}
describe("MessageItem memo boundary", () => {
afterEach(() => {
rendererSpy.mockClear();
});
it("does not re-render the row when the parent re-renders with stable props", () => {
const { getByText } = render(<Harness onOpenFile={noop} />);
expect(rendererSpy).toHaveBeenCalledTimes(1);
fireEvent.click(getByText("tick"));
fireEvent.click(getByText("tick"));
expect(rendererSpy).toHaveBeenCalledTimes(1); // memo bailed on stable props
});
it("re-renders the row when onOpenFile identity changes (stability requirement)", () => {
const { rerender } = render(row(() => {}));
expect(rendererSpy).toHaveBeenCalledTimes(1);
rerender(row(() => {}));
expect(rendererSpy).toHaveBeenCalledTimes(2); // fresh callback ref breaks memo
});
});
@@ -1,8 +1,9 @@
"use client";
import { memo } from "react";
import { Button } from "@kandev/ui/button";
import { GridSpinner } from "@/components/grid-spinner";
import type { Message, TaskSessionState, TaskState } from "@/lib/types/http";
import type { Message, TaskSessionState } from "@/lib/types/http";
import type { RenderItem } from "@/hooks/use-processed-messages";
import { MessageRenderer } from "@/components/task/chat/message-renderer";
import { TurnGroupMessage } from "@/components/task/chat/messages/turn-group-message";
@@ -20,7 +21,6 @@ export type MessageListProps = {
messagesLoading: boolean;
isWorking: boolean;
sessionState?: TaskSessionState;
taskState?: TaskState;
worktreePath?: string;
onOpenFile?: (path: string) => void;
};
@@ -100,7 +100,7 @@ export function MessageListStatus({
);
}
export function MessageItem({
export const MessageItem = memo(function MessageItem({
item,
sessionId,
permissionsByToolCallId,
@@ -110,9 +110,6 @@ export function MessageItem({
onOpenFile,
isLastGroup,
isTurnActive,
messages,
sessionState,
taskState,
onScrollToMessage,
}: {
item: RenderItem;
@@ -124,9 +121,6 @@ export function MessageItem({
onOpenFile?: (path: string) => void;
isLastGroup: boolean;
isTurnActive: boolean;
messages: Message[];
sessionState?: TaskSessionState;
taskState?: TaskState;
onScrollToMessage: (id: string) => void;
}) {
if (item.type === "prepare_progress") {
@@ -144,9 +138,6 @@ export function MessageItem({
onOpenFile={onOpenFile}
isLastGroup={isLastGroup}
isTurnActive={isTurnActive}
allMessages={messages}
sessionState={sessionState}
taskState={taskState}
onScrollToMessage={onScrollToMessage}
/>
);
@@ -155,16 +146,13 @@ export function MessageItem({
<MessageRenderer
comment={item.message}
isTaskDescription={item.message.id === "task-description"}
sessionState={sessionState}
taskState={taskState}
taskId={taskId}
permissionsByToolCallId={permissionsByToolCallId}
childrenByParentToolCallId={childrenByParentToolCallId}
worktreePath={worktreePath}
sessionId={sessionId ?? undefined}
onOpenFile={onOpenFile}
allMessages={messages}
onScrollToMessage={onScrollToMessage}
/>
);
}
});
@@ -97,15 +97,7 @@ function useStableFirstItemIndex(items: RenderItem[]) {
function useVirtuosoCallbacks(props: VirtuosoBodyProps) {
const { items, sessionId, permissionsByToolCallId, childrenByParentToolCallId, taskId } = props;
const {
worktreePath,
onOpenFile,
lastTurnGroupId,
isRunning,
messages,
sessionState,
taskState,
} = props;
const { worktreePath, onOpenFile, lastTurnGroupId, isRunning } = props;
const { hasMore, isLoadingMore, loadMore } = props;
const virtuosoRef = useRef<VirtuosoHandle>(null);
const itemCount = items.length;
@@ -162,9 +154,6 @@ function useVirtuosoCallbacks(props: VirtuosoBodyProps) {
onOpenFile={onOpenFile}
isLastGroup={item.type === "turn_group" && item.id === lastTurnGroupId}
isTurnActive={isRunning}
messages={messages}
sessionState={sessionState}
taskState={taskState}
onScrollToMessage={handleScrollToMessage}
/>
</div>
@@ -181,9 +170,6 @@ function useVirtuosoCallbacks(props: VirtuosoBodyProps) {
onOpenFile,
lastTurnGroupId,
isRunning,
messages,
sessionState,
taskState,
handleScrollToMessage,
],
);
@@ -436,12 +422,7 @@ function useVirtuosoHeaderFooter(args: HeaderFooterArgs) {
<>
<AgentStatus sessionState={sessionState} sessionId={sessionId} messages={messages} />
{footerActions.map((msg) => (
<MessageRenderer
key={msg.id}
comment={msg}
isTaskDescription={false}
sessionState={sessionState}
/>
<MessageRenderer key={msg.id} comment={msg} isTaskDescription={false} />
))}
</>
),
@@ -511,12 +492,7 @@ export const VirtuosoMessageList = memo(function VirtuosoMessageList(props: Mess
/>
<AgentStatus sessionState={sessionState} sessionId={sessionId} messages={messages} />
{footerActions.map((msg) => (
<MessageRenderer
key={msg.id}
comment={msg}
isTaskDescription={false}
sessionState={sessionState}
/>
<MessageRenderer key={msg.id} comment={msg} isTaskDescription={false} />
))}
</SessionPanelContent>
);
@@ -3,11 +3,12 @@
import { memo, useState, useCallback, type ReactElement } from "react";
import { IconPlayerPlay } from "@tabler/icons-react";
import { Button } from "@kandev/ui/button";
import type { Message, TaskSessionState, TaskState } from "@/lib/types/http";
import type { Message, TaskSessionState } from "@/lib/types/http";
import type { ToolCallMetadata } from "@/components/task/chat/types";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildStartCreatedRequest } from "@/lib/services/session-launch-helpers";
import { useAppStore } from "@/components/state-provider";
import { useTask } from "@/hooks/use-task";
import { ChatMessage } from "@/components/task/chat/messages/chat-message";
import { PermissionRequestMessage } from "@/components/task/chat/messages/permission-request-message";
import { StatusMessage } from "@/components/task/chat/messages/status-message";
@@ -31,15 +32,12 @@ import {
type AdapterContext = {
isTaskDescription: boolean;
sessionState?: TaskSessionState;
taskState?: TaskState;
taskId?: string;
permissionsByToolCallId?: Map<string, Message>;
childrenByParentToolCallId?: Map<string, Message[]>;
worktreePath?: string;
sessionId?: string;
onOpenFile?: (path: string) => void;
allMessages?: Message[];
onScrollToMessage?: (messageId: string) => void;
};
@@ -81,6 +79,60 @@ function TaskDescriptionStartButton({ taskId, sessionId }: { taskId: string; ses
);
}
function useSessionStateValue(sessionId?: string): TaskSessionState | undefined {
return useAppStore((state) =>
sessionId ? (state.taskSessions.items[sessionId]?.state ?? undefined) : undefined,
);
}
/**
* The task-description message is the only row whose rendering depends on the
* live session/task state, so it reads them from the store directly. Keeping
* sessionState/taskState off the shared MessageRenderer props means a state
* transition no longer re-renders every message in the list.
*/
function TaskDescriptionMessage({
comment,
taskId,
sessionId,
onScrollToMessage,
}: {
comment: Message;
taskId?: string;
sessionId?: string;
onScrollToMessage?: (messageId: string) => void;
}) {
const sessionState = useSessionStateValue(sessionId);
const task = useTask(taskId ?? null);
const renderAsUser = comment.author_type === "user" || sessionState !== "FAILED";
if (!renderAsUser) {
return (
<ChatMessage
comment={comment}
label="Agent"
className="bg-muted/40 text-foreground border-border/60"
showRichBlocks={comment.type === "message" || comment.type === "content" || !comment.type}
sessionId={sessionId}
onScrollToMessage={onScrollToMessage}
/>
);
}
const showStartButton =
sessionState === "CREATED" && task?.state !== "SCHEDULING" && !!taskId && !!sessionId;
return (
<>
<ChatMessage
comment={comment}
label="You"
className="bg-primary/10 text-foreground border-primary/30"
sessionId={sessionId}
onScrollToMessage={onScrollToMessage}
/>
{showStartButton && <TaskDescriptionStartButton taskId={taskId!} sessionId={sessionId!} />}
</>
);
}
type MessageAdapter = {
matches: (comment: Message, ctx: AdapterContext) => boolean;
render: (comment: Message, ctx: AdapterContext) => ReactElement;
@@ -220,7 +272,7 @@ const adapters: MessageAdapter[] = [
const meta = comment.metadata as Record<string, unknown> | undefined;
return Array.isArray(meta?.actions) && (meta.actions as unknown[]).length > 0;
},
render: (comment, ctx) => <ActionMessage comment={comment} sessionState={ctx.sessionState} />,
render: (comment) => <ActionMessage comment={comment} />,
},
{
matches: (comment) =>
@@ -247,29 +299,25 @@ const adapters: MessageAdapter[] = [
{
matches: () => true,
render: (comment, ctx) => {
if (
comment.author_type === "user" ||
(ctx.isTaskDescription && ctx.sessionState !== "FAILED")
) {
const showStartButton =
ctx.isTaskDescription &&
ctx.sessionState === "CREATED" &&
ctx.taskState !== "SCHEDULING" &&
ctx.taskId &&
ctx.sessionId;
if (ctx.isTaskDescription) {
return (
<>
<ChatMessage
comment={comment}
label="You"
className="bg-primary/10 text-foreground border-primary/30"
allMessages={ctx.allMessages}
onScrollToMessage={ctx.onScrollToMessage}
/>
{showStartButton && (
<TaskDescriptionStartButton taskId={ctx.taskId!} sessionId={ctx.sessionId!} />
)}
</>
<TaskDescriptionMessage
comment={comment}
taskId={ctx.taskId}
sessionId={ctx.sessionId}
onScrollToMessage={ctx.onScrollToMessage}
/>
);
}
if (comment.author_type === "user") {
return (
<ChatMessage
comment={comment}
label="You"
className="bg-primary/10 text-foreground border-primary/30"
sessionId={ctx.sessionId}
onScrollToMessage={ctx.onScrollToMessage}
/>
);
}
return (
@@ -278,7 +326,7 @@ const adapters: MessageAdapter[] = [
label="Agent"
className="bg-muted/40 text-foreground border-border/60"
showRichBlocks={comment.type === "message" || comment.type === "content" || !comment.type}
allMessages={ctx.allMessages}
sessionId={ctx.sessionId}
onScrollToMessage={ctx.onScrollToMessage}
/>
);
@@ -289,43 +337,34 @@ const adapters: MessageAdapter[] = [
type MessageRendererProps = {
comment: Message;
isTaskDescription: boolean;
sessionState?: TaskSessionState;
taskState?: TaskState;
taskId?: string;
permissionsByToolCallId?: Map<string, Message>;
childrenByParentToolCallId?: Map<string, Message[]>;
worktreePath?: string;
sessionId?: string;
onOpenFile?: (path: string) => void;
allMessages?: Message[];
onScrollToMessage?: (messageId: string) => void;
};
export const MessageRenderer = memo(function MessageRenderer({
comment,
isTaskDescription,
sessionState,
taskState,
taskId,
permissionsByToolCallId,
childrenByParentToolCallId,
worktreePath,
sessionId,
onOpenFile,
allMessages,
onScrollToMessage,
}: MessageRendererProps) {
const ctx = {
isTaskDescription,
sessionState,
taskState,
taskId,
permissionsByToolCallId,
childrenByParentToolCallId,
worktreePath,
sessionId,
onOpenFile,
allMessages,
onScrollToMessage,
};
const adapter =
@@ -1,9 +1,15 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import type { ReactNode } from "react";
import { cleanup, fireEvent, render, screen, waitFor } from "@testing-library/react";
import { StateProvider } from "@/components/state-provider";
import { ActionMessage } from "./action-message";
import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http";
import {
sessionId as toSessionId,
taskId as toTaskId,
type Message,
type TaskSession,
type TaskSessionState,
} from "@/lib/types/http";
import type { AppState } from "@/lib/state/store";
const requestMock = vi.fn().mockResolvedValue({});
@@ -52,24 +58,29 @@ function retryMessage(overrides: Partial<Message> = {}): Message {
} as Message;
}
function Wrapper({ children }: { children: ReactNode }) {
return <StateProvider initialState={{}}>{children}</StateProvider>;
/** ActionMessage reads session state from the store (keyed by comment.session_id),
* so seed it via the provider instead of passing a prop. */
function renderAction(comment: Message, sessionState?: TaskSessionState) {
const initialState: Partial<AppState> = sessionState
? { taskSessions: { items: { "sess-1": { state: sessionState } as TaskSession } } }
: {};
return render(<ActionMessage comment={comment} />, {
wrapper: ({ children }) => (
<StateProvider initialState={initialState}>{children}</StateProvider>
),
});
}
describe("ActionMessage — transient retry (warning variant)", () => {
it("renders the retrying copy in amber, not red", () => {
render(<ActionMessage comment={retryMessage()} sessionState="WAITING_FOR_INPUT" />, {
wrapper: Wrapper,
});
renderAction(retryMessage(), "WAITING_FOR_INPUT");
const text = screen.getByText(/retrying in 5s \(attempt 1\/3\)/i);
expect(text.className).toContain("text-amber-600");
expect(text.className).not.toContain("text-red-600");
});
it("Cancel fires a session.recover ws_request with action cancel_retry", async () => {
render(<ActionMessage comment={retryMessage()} sessionState="WAITING_FOR_INPUT" />, {
wrapper: Wrapper,
});
renderAction(retryMessage(), "WAITING_FOR_INPUT");
fireEvent.click(screen.getByTestId(CANCEL_TEST_ID));
await waitFor(() => expect(requestMock).toHaveBeenCalledTimes(1));
expect(requestMock).toHaveBeenCalledWith("session.recover", {
@@ -80,10 +91,7 @@ describe("ActionMessage — transient retry (warning variant)", () => {
});
it("hides while the session is RUNNING (retry in flight) to avoid a stale card", () => {
const { container } = render(
<ActionMessage comment={retryMessage()} sessionState="RUNNING" />,
{ wrapper: Wrapper },
);
const { container } = renderAction(retryMessage(), "RUNNING");
expect(container.firstChild).toBeNull();
});
@@ -98,9 +106,7 @@ describe("ActionMessage — transient retry (warning variant)", () => {
],
},
} as Partial<Message>);
render(<ActionMessage comment={errorMsg} sessionState="WAITING_FOR_INPUT" />, {
wrapper: Wrapper,
});
renderAction(errorMsg, "WAITING_FOR_INPUT");
const text = screen.getByText(/Agent encountered an error/i);
expect(text.className).toContain("text-red-600");
expect(text.className).not.toContain("text-amber-600");
@@ -47,13 +47,15 @@ function isSessionActive(state?: TaskSessionState) {
return state === "RUNNING" || state === "STARTING" || state === "COMPLETED";
}
export const ActionMessage = memo(function ActionMessage({
comment,
sessionState,
}: {
comment: Message;
sessionState?: TaskSessionState;
}) {
export const ActionMessage = memo(function ActionMessage({ comment }: { comment: Message }) {
// Read session state from the store instead of receiving it as a prop, so a
// state transition doesn't re-render every message in the list (only the
// rare action messages that actually depend on it).
const sessionState = useAppStore((state) =>
comment.session_id
? (state.taskSessions.items[comment.session_id]?.state ?? undefined)
: undefined,
);
const metadata = comment.metadata as ActionMeta | undefined;
const isWarning = metadata?.variant === "warning";
const message = comment.content || "An error occurred";
@@ -1,15 +1,10 @@
"use client";
import { useState, memo } from "react";
import ReactMarkdown from "react-markdown";
import { IconMinus, IconPlus, IconMaximize } from "@tabler/icons-react";
import { Button } from "@kandev/ui/button";
import { Dialog, DialogContent, DialogHeader, DialogTitle } from "@kandev/ui/dialog";
import {
markdownComponents,
normalizeMarkdown,
remarkPlugins,
} from "@/components/shared/markdown-components";
import { MemoizedMarkdown } from "@/components/shared/memoized-markdown";
import type { Message } from "@/lib/types/http";
function parsePlanContent(text: string): { title: string; body: string } {
@@ -28,9 +23,7 @@ function PlanMarkdownBody({ text, className }: { text: string; className?: strin
<div
className={`markdown-body max-w-none text-sm [&>*]:my-2 [&>p]:my-2 [&>ul]:my-2 [&>ol]:my-2 ${className ?? ""}`}
>
<ReactMarkdown remarkPlugins={remarkPlugins} components={markdownComponents}>
{normalizeMarkdown(text)}
</ReactMarkdown>
<MemoizedMarkdown content={text} />
</div>
);
}
@@ -1,19 +1,14 @@
"use client";
import { memo, useState, useCallback } from "react";
import ReactMarkdown from "react-markdown";
import { IconWand, IconMessageDots, IconFile } from "@tabler/icons-react";
import { cn } from "@/lib/utils";
import type { Message } from "@/lib/types/http";
import { RichBlocks } from "@/components/task/chat/messages/rich-blocks";
import { MessageActions } from "@/components/task/chat/messages/message-actions";
import { useMessageNavigation } from "@/hooks/use-message-navigation";
import { useUserMessageNavigation } from "@/hooks/use-message-navigation";
import { SenderTaskBadge, type SenderTaskInfo } from "./sender-task-badge";
import {
markdownComponents,
normalizeMarkdown,
remarkPlugins,
} from "@/components/shared/markdown-components";
import { MemoizedMarkdown } from "@/components/shared/memoized-markdown";
import { ImagePreviewDialog } from "@/components/task/chat/image-preview-dialog";
import {
WorkflowStepMessageBadge,
@@ -27,7 +22,7 @@ type ChatMessageProps = {
label: string;
className: string;
showRichBlocks?: boolean;
allMessages?: Message[];
sessionId?: string | null;
onScrollToMessage?: (messageId: string) => void;
};
@@ -88,9 +83,7 @@ function renderUserMessageBody(
if (hasContent) {
return (
<div className="markdown-body markdown-body-user max-w-none">
<ReactMarkdown remarkPlugins={remarkPlugins} components={markdownComponents}>
{normalizeMarkdown(content)}
</ReactMarkdown>
<MemoizedMarkdown content={content} />
</div>
);
}
@@ -106,7 +99,7 @@ type UserMessageProps = {
comment: Message;
showRaw: boolean;
onToggleRaw: () => void;
allMessages?: Message[];
sessionId?: string | null;
onScrollToMessage?: (messageId: string) => void;
};
@@ -200,10 +193,10 @@ function UserMessageContent({
comment,
showRaw,
onToggleRaw,
allMessages,
sessionId,
onScrollToMessage,
}: UserMessageProps) {
const userNavigation = useMessageNavigation(allMessages || [], comment.id, "user");
const userNavigation = useUserMessageNavigation(sessionId ?? null, comment.id);
const {
imageAttachments,
fileAttachments,
@@ -263,15 +256,16 @@ function UserMessageContent({
showTimestamp={true}
showRawToggle={true}
hasHiddenPrompts={hasHiddenPrompts}
showNavigation={!!allMessages && allMessages.length > 0}
showNavigation={userNavigation.hasPrevious || userNavigation.hasNext}
isRawView={showRaw}
onToggleRaw={onToggleRaw}
onNavigatePrev={() => {
if (userNavigation.previous && onScrollToMessage)
onScrollToMessage(userNavigation.previous.id);
if (userNavigation.previousId && onScrollToMessage)
onScrollToMessage(userNavigation.previousId);
}}
onNavigateNext={() => {
if (userNavigation.next && onScrollToMessage) onScrollToMessage(userNavigation.next.id);
if (userNavigation.nextId && onScrollToMessage)
onScrollToMessage(userNavigation.nextId);
}}
hasPrev={userNavigation.hasPrevious}
hasNext={userNavigation.hasNext}
@@ -300,9 +294,7 @@ function AgentMessageContent({ comment, showRaw, onToggleRaw, showRichBlocks }:
</pre>
) : (
<div className="markdown-body max-w-none">
<ReactMarkdown remarkPlugins={remarkPlugins} components={markdownComponents}>
{normalizeMarkdown(comment.content || "(empty)")}
</ReactMarkdown>
<MemoizedMarkdown content={comment.content || "(empty)"} />
{showRichBlocks ? <RichBlocks comment={comment} /> : null}
</div>
)}
@@ -328,7 +320,7 @@ export const ChatMessage = memo(function ChatMessage({
label,
className,
showRichBlocks,
allMessages,
sessionId,
onScrollToMessage,
}: ChatMessageProps) {
const [showRaw, setShowRaw] = useState(false);
@@ -360,7 +352,7 @@ export const ChatMessage = memo(function ChatMessage({
comment={comment}
showRaw={showRaw}
onToggleRaw={toggleRaw}
allMessages={allMessages}
sessionId={sessionId}
onScrollToMessage={onScrollToMessage}
/>
);
@@ -4,7 +4,7 @@ import { useState, useCallback, memo, useMemo } from "react";
import { IconChevronDown, IconChevronRight } from "@tabler/icons-react";
import { GridSpinner } from "@/components/grid-spinner";
import { cn, transformPathsInText } from "@/lib/utils";
import type { Message, TaskSessionState, TaskState } from "@/lib/types/http";
import type { Message } from "@/lib/types/http";
import type { TurnGroup } from "@/hooks/use-processed-messages";
import type { ToolCallMetadata } from "@/components/task/chat/types";
import { MessageRenderer } from "@/components/task/chat/message-renderer";
@@ -21,9 +21,6 @@ type TurnGroupMessageProps = {
isLastGroup?: boolean;
/** Whether the turn is still active (agent is running) */
isTurnActive?: boolean;
allMessages?: Message[];
sessionState?: TaskSessionState;
taskState?: TaskState;
onScrollToMessage?: (messageId: string) => void;
};
@@ -154,9 +151,6 @@ type TurnGroupContentProps = {
taskId?: string;
worktreePath?: string;
onOpenFile?: (path: string) => void;
allMessages?: Message[];
sessionState?: TaskSessionState;
taskState?: TaskState;
onScrollToMessage?: (messageId: string) => void;
};
@@ -261,14 +255,11 @@ function renderMessageEntry(message: Message, props: MessageRenderProps) {
comment={message}
isTaskDescription={false}
taskId={props.taskId}
sessionState={props.sessionState}
taskState={props.taskState}
permissionsByToolCallId={props.permissionsByToolCallId}
childrenByParentToolCallId={props.childrenByParentToolCallId}
worktreePath={props.worktreePath}
sessionId={props.sessionId ?? undefined}
onOpenFile={props.onOpenFile}
allMessages={props.allMessages}
onScrollToMessage={props.onScrollToMessage}
/>
);
@@ -320,9 +311,6 @@ function TurnGroupContent({
taskId,
worktreePath,
onOpenFile,
allMessages,
sessionState,
taskState,
onScrollToMessage,
}: TurnGroupContentProps) {
const renderProps: MessageRenderProps = {
@@ -332,9 +320,6 @@ function TurnGroupContent({
taskId,
worktreePath,
onOpenFile,
allMessages,
sessionState,
taskState,
onScrollToMessage,
};
const compacted = useMemo(() => compactTurnGroupMessages(group.messages), [group.messages]);
@@ -361,9 +346,6 @@ export const TurnGroupMessage = memo(function TurnGroupMessage({
onOpenFile,
isLastGroup = false,
isTurnActive = false,
allMessages,
sessionState,
taskState,
onScrollToMessage,
}: TurnGroupMessageProps) {
const isGroupRunning = hasRunningTool(group.messages);
@@ -401,9 +383,6 @@ export const TurnGroupMessage = memo(function TurnGroupMessage({
taskId={taskId}
worktreePath={worktreePath}
onOpenFile={onOpenFile}
allMessages={allMessages}
sessionState={sessionState}
taskState={taskState}
onScrollToMessage={onScrollToMessage}
/>
)}
+10 -2
View File
@@ -10,6 +10,7 @@ import {
useMemo,
} from "react";
import { EditorContent } from "@tiptap/react";
import { useShallow } from "zustand/react/shallow";
import { useCustomPrompts } from "@/hooks/domains/settings/use-custom-prompts";
import { useAppStore, useAppStoreApi } from "@/components/state-provider";
import { getWebSocketClient } from "@/lib/ws/connection";
@@ -490,8 +491,15 @@ function TipTapPopups({
}
function useMessageHistoryForSession(sessionId: string | null): string[] {
const messages = useAppStore((s) => (sessionId ? s.messages.bySession[sessionId] : undefined));
return useMemo(() => extractUserHistory(messages ?? []), [messages]);
// Subscribe to the *derived* user history (not the raw messages array) via a
// shallow comparison: agent streaming tokens churn the messages array every
// frame but don't change the user's sent-message history, so the rich editor
// only re-renders when the history list itself changes.
return useAppStore(
useShallow((s) =>
extractUserHistory((sessionId ? s.messages.bySession[sessionId] : undefined) ?? []),
),
);
}
function useChatHistory(sessionId: string | null) {
@@ -143,7 +143,6 @@ export const TaskChatPanel = memo(function TaskChatPanel({
const {
resolvedSessionId,
session,
task,
taskId,
isWorking,
messagesLoading,
@@ -212,7 +211,6 @@ export const TaskChatPanel = memo(function TaskChatPanel({
messagesLoading={messagesLoading}
isWorking={isWorking}
sessionState={session?.state}
taskState={task?.state}
worktreePath={session?.worktree_path}
onOpenFile={onOpenFile}
/>
@@ -1,10 +1,26 @@
import { describe, expect, it } from "vitest";
import {
aggregateSidebarTasks,
buildPendingFlags,
readPendingFlags,
type SidebarStepInfo,
type WorkflowSnapshotMap,
} from "./task-session-sidebar-aggregate";
import type { KanbanState } from "@/lib/state/slices";
import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http";
function makePermissionRequest(id: string, status?: string): Message {
return {
id,
session_id: toSessionId("s1"),
task_id: toTaskId("t1"),
author_type: "agent",
content: "",
type: "permission_request",
metadata: status ? { status } : undefined,
created_at: "",
};
}
type KanbanTask = KanbanState["tasks"][number];
@@ -141,3 +157,31 @@ describe("aggregateSidebarTasks", () => {
expect(result.allTasks.map((t) => t.id)).toEqual(["task-B"]);
});
});
describe("buildPendingFlags / readPendingFlags", () => {
it("flags a session with a pending permission request", () => {
const flags = buildPendingFlags({ "sess-1": [makePermissionRequest("p1")] }, ["sess-1"]);
expect(readPendingFlags(flags, "sess-1")).toEqual({ clarification: false, permission: true });
});
it("does not flag a session whose permission request is already resolved", () => {
const flags = buildPendingFlags({ "sess-1": [makePermissionRequest("p1", "approved")] }, [
"sess-1",
]);
expect(readPendingFlags(flags, "sess-1")).toEqual({ clarification: false, permission: false });
});
it("only computes flags for the requested session ids", () => {
const flags = buildPendingFlags(
{ "sess-1": [makePermissionRequest("p1")], "sess-2": [makePermissionRequest("p2")] },
["sess-1"],
);
expect(readPendingFlags(flags, "sess-1").permission).toBe(true);
expect(readPendingFlags(flags, "sess-2").permission).toBe(false);
});
it("returns all-false for a null/unknown session", () => {
expect(readPendingFlags({}, null)).toEqual({ clarification: false, permission: false });
expect(readPendingFlags({}, "missing")).toEqual({ clarification: false, permission: false });
});
});
@@ -1,4 +1,39 @@
import type { KanbanState } from "@/lib/state/slices";
import type { Message } from "@/lib/types/http";
import {
hasPendingClarification,
hasPendingPermissionRequest,
} from "@/lib/utils/pending-clarification";
/** Flat per-session pending flags keyed for shallow comparison so the sidebar
* only re-renders when a clarification/permission flag actually flips, not on
* every streaming token that churns the messages map. */
const pendingClarKey = (sessionId: string) => `${sessionId}#clar`;
const pendingPermKey = (sessionId: string) => `${sessionId}#perm`;
export function buildPendingFlags(
bySession: Record<string, Message[] | undefined>,
sessionIds: string[],
): Record<string, boolean> {
const flags: Record<string, boolean> = {};
for (const id of sessionIds) {
const msgs = bySession[id];
flags[pendingClarKey(id)] = hasPendingClarification(msgs);
flags[pendingPermKey(id)] = hasPendingPermissionRequest(msgs);
}
return flags;
}
export function readPendingFlags(
pendingFlags: Record<string, boolean>,
sessionId?: string | null,
): { clarification: boolean; permission: boolean } {
if (!sessionId) return { clarification: false, permission: false };
return {
clarification: pendingFlags[pendingClarKey(sessionId)] ?? false,
permission: pendingFlags[pendingPermKey(sessionId)] ?? false,
};
}
export type SidebarStepInfo = {
id: string;
@@ -3,13 +3,7 @@
import { useCallback, useEffect, useMemo, useState, memo } from "react";
import { usePathname, useRouter } from "next/navigation";
import { linkToTask } from "@/lib/links";
import type {
Message,
TaskState,
TaskSession,
TaskSessionState,
Repository,
} from "@/lib/types/http";
import type { TaskState, TaskSession, TaskSessionState, Repository } from "@/lib/types/http";
import type { TaskPR } from "@/lib/types/github";
import type { KanbanState } from "@/lib/state/slices";
import type { GitStatusEntry } from "@/lib/state/slices/session-runtime/types";
@@ -33,10 +27,8 @@ import { getWebSocketClient } from "@/lib/ws/connection";
import { useArchivedTaskState } from "./task-archived-context";
import { useRepositories } from "@/hooks/domains/workspace/use-repositories";
import { useWorkspacePRs } from "@/hooks/domains/github/use-task-pr";
import {
hasPendingClarificationForSession,
hasPendingPermissionForSession,
} from "@/lib/utils/pending-clarification";
import { buildPendingFlags, readPendingFlags } from "./task-session-sidebar-aggregate";
import { useShallow } from "zustand/react/shallow";
/**
* Stabilize a derived array of primary session IDs so the reference only
@@ -98,7 +90,7 @@ type SidebarCtx = {
envIdBySessionId: Record<string, string>;
repositorySlugById: Map<string, string | undefined>;
taskPRsByTaskId: Record<string, TaskPR[] | undefined>;
messagesBySession: Record<string, Message[]>;
pendingFlags: Record<string, boolean>;
titleById: Map<string, string>;
workflowNameById: Map<string, string>;
stepTitleById: Map<string, string>;
@@ -128,14 +120,7 @@ function toSidebarItem(
const repoSlug = task.repositoryId ? ctx.repositorySlugById.get(task.repositoryId) : undefined;
// Sidebar shows just one slot; pick the primary PR (first by created_at).
const pr = ctx.taskPRsByTaskId[task.id]?.[0];
const hasPendingClarificationRequest = hasPendingClarificationForSession(
ctx.messagesBySession,
task.primarySessionId,
);
const hasPendingPermission = hasPendingPermissionForSession(
ctx.messagesBySession,
task.primarySessionId,
);
const pending = readPendingFlags(ctx.pendingFlags, task.primarySessionId);
const diffStats = resolveDiffStats(
sessionInfo.diffStats,
@@ -162,8 +147,8 @@ function toSidebarItem(
remoteExecutorType: task.primaryExecutorType ?? undefined,
remoteExecutorName: task.primaryExecutorName ?? undefined,
primarySessionId: task.primarySessionId ?? null,
hasPendingClarification: hasPendingClarificationRequest,
hasPendingPermission,
hasPendingClarification: pending.clarification,
hasPendingPermission: pending.permission,
updatedAt: sessionInfo.updatedAt ?? task.updatedAt ?? task.createdAt,
createdAt: task.createdAt,
isArchived: false as boolean,
@@ -225,7 +210,6 @@ function useSidebarData(workspaceId: string | null) {
const envIdBySessionId = useAppStore((state) => state.environmentIdBySessionId);
const repositoriesByWorkspace = useAppStore((state) => state.repositories.itemsByWorkspaceId);
const taskPRsByTaskId = useAppStore((state) => state.taskPRs.byTaskId);
const messagesBySession = useAppStore((state) => state.messages.bySession);
const archivedState = useArchivedTaskState();
const selectedTaskId = useMemo(() => {
@@ -241,6 +225,14 @@ function useSidebarData(workspaceId: string | null) {
isLoading: isLoadingWorkflow,
} = useWorkspaceSidebarTasks(workspaceId);
// Stable list of primary session IDs for the bulk-subscribe effect and the
// narrow pending-flag selector below. Derived from kanban tasks (always
// available) rather than sessionsByTaskId (loaded on-demand).
const primarySessionIds = useStablePrimarySessionIds(allTasks);
const pendingFlags = useAppStore(
useShallow((state) => buildPendingFlags(state.messages.bySession, primarySessionIds)),
);
const tasksWithRepositories = useMemo(() => {
const repositories = workspaceId ? (repositoriesByWorkspace[workspaceId] ?? []) : [];
const repositorySlugById = new Map(
@@ -255,7 +247,7 @@ function useSidebarData(workspaceId: string | null) {
envIdBySessionId,
repositorySlugById,
taskPRsByTaskId,
messagesBySession,
pendingFlags,
titleById,
workflowNameById,
stepTitleById,
@@ -279,14 +271,10 @@ function useSidebarData(workspaceId: string | null) {
gitStatusByEnvId,
envIdBySessionId,
taskPRsByTaskId,
messagesBySession,
pendingFlags,
archivedState,
]);
// Stable list of primary session IDs for the bulk-subscribe effect.
// Derived from kanban tasks (always available) rather than sessionsByTaskId (loaded on-demand).
const primarySessionIds = useStablePrimarySessionIds(allTasks);
return {
activeTaskId,
selectedTaskId,
@@ -33,6 +33,8 @@ import {
shouldRunMessageBackfill,
runBackfillRound,
autoBackfillUntilUserMessage,
nextFetchSeq,
commitFetchSeq,
MAX_AUTO_BACKFILL_PAGES,
} from "./use-session-messages";
import type { TaskSessionState } from "@/lib/types/http";
@@ -269,6 +271,32 @@ describe("runBackfillRound", () => {
});
});
describe("stale concurrent fetch guard", () => {
it("rejects an older fetch that completes after a newer one merged", () => {
const sid = "guard-sess-a";
const older = nextFetchSeq();
const newer = nextFetchSeq();
// The newer fetch finishes first and merges.
expect(commitFetchSeq(sid, newer)).toBe(true);
// The older fetch finishing late must be skipped.
expect(commitFetchSeq(sid, older)).toBe(false);
});
it("applies fetches that complete in order", () => {
const sid = "guard-sess-b";
expect(commitFetchSeq(sid, nextFetchSeq())).toBe(true);
expect(commitFetchSeq(sid, nextFetchSeq())).toBe(true);
});
it("tracks the applied sequence independently per session", () => {
const older = nextFetchSeq();
const newer = nextFetchSeq();
expect(commitFetchSeq("guard-sess-c", newer)).toBe(true);
// A different session is unaffected by another session's higher applied seq.
expect(commitFetchSeq("guard-sess-d", older)).toBe(true);
});
});
describe("autoBackfillUntilUserMessage", () => {
it("continues past three pages until a user/agent message is found", async () => {
mockListTaskSessionMessages
@@ -8,9 +8,35 @@ import { createDebugLogger, isDebug } from "@/lib/debug/log";
const INITIAL_FETCH_LIMIT = 100;
const BACKFILL_PAGE_LIMIT = 100;
const RUNNING_BACKFILL_INITIAL_DELAY_MS = 1200;
const RUNNING_BACKFILL_INTERVAL_MS = 3000;
const RUNNING_BACKFILL_INTERVAL_MS = 5000;
export const MAX_AUTO_BACKFILL_PAGES = 10;
// Monotonic guard against stale concurrent fetches. Each fetch claims an
// increasing sequence number before awaiting; a completion only merges if no
// newer fetch for the same session has already merged, so an older in-flight
// fetch cannot overwrite newer state (WS-delivered messages or a newer fetch's
// snapshot) when two fetches for the same session race.
let fetchSeqCounter = 0;
const lastAppliedFetchSeq = new Map<string, number>();
/** Claims the next monotonic fetch sequence number. */
export function nextFetchSeq(): number {
fetchSeqCounter += 1;
return fetchSeqCounter;
}
/**
* Records that the fetch identified by `seq` is about to merge for `sessionId`.
* Returns false when a newer fetch has already merged for the session, so the
* caller can skip the stale merge.
*/
export function commitFetchSeq(sessionId: string, seq: number): boolean {
const applied = lastAppliedFetchSeq.get(sessionId) ?? 0;
if (seq < applied) return false;
lastAppliedFetchSeq.set(sessionId, seq);
return true;
}
export function hasUserOrAgentMessage(messages: Message[]): boolean {
return messages.some(
(m) => m.type === "message" && (m.author_type === "user" || m.author_type === "agent"),
@@ -108,6 +134,32 @@ type MessageListResponse = { messages: Message[]; has_more?: boolean; cursor?: s
const EMPTY_MESSAGES: Message[] = [];
const EMPTY_META = { isLoading: false, hasMore: false, oldestCursor: null };
/** Debug-only summary of a fetch response (no-op unless debug logging is on). */
function logFetchSummary(
sessionId: string,
fetched: Message[],
response: MessageListResponse,
limit: number,
): void {
if (!isDebug()) return;
const summary = summarizeMessages(fetched);
debug("message.list response", {
sessionId,
hasMore: response.has_more ?? false,
cursor: response.cursor ?? null,
...summary,
});
if (fetched.length > 0 && summary.userMessageCount === 0 && summary.agentMessageCount === 0) {
debug("WARNING: fetched window contains no user/agent message rows", {
sessionId,
limit,
hasMore: response.has_more ?? false,
byType: summary.byType,
hint: "The fetch limit may be too small for this session's last turn — user prompt and agent replies live further back. Paginate or raise the limit to see them.",
});
}
}
/** Fetch latest messages via WS and merge with any that arrived via live notifications. */
async function fetchAndStoreMessages(
sessionId: string,
@@ -117,6 +169,7 @@ async function fetchAndStoreMessages(
if (!client) {
return [];
}
const seq = nextFetchSeq();
const requestParams = {
session_id: sessionId,
@@ -126,24 +179,7 @@ async function fetchAndStoreMessages(
debug("message.list request", requestParams);
const response = await client.request<MessageListResponse>("message.list", requestParams, 10000);
const fetched = [...(response.messages ?? [])].reverse();
if (isDebug()) {
const summary = summarizeMessages(fetched);
debug("message.list response", {
sessionId,
hasMore: response.has_more ?? false,
cursor: response.cursor ?? null,
...summary,
});
if (fetched.length > 0 && summary.userMessageCount === 0 && summary.agentMessageCount === 0) {
debug("WARNING: fetched window contains no user/agent message rows", {
sessionId,
limit: requestParams.limit,
hasMore: response.has_more ?? false,
byType: summary.byType,
hint: "The fetch limit may be too small for this session's last turn — user prompt and agent replies live further back. Paginate or raise the limit to see them.",
});
}
}
logFetchSummary(sessionId, fetched, response, requestParams.limit);
// Merge: keep WS-delivered messages that aren't in the fetch response.
// This prevents a slow fetch (sent before messages existed) from wiping
// messages that arrived via real-time notifications while the fetch was
@@ -158,10 +194,23 @@ async function fetchAndStoreMessages(
)
: fetched;
store.getState().setMessages(sessionId, merged, {
// Stale-fetch guard: if a newer fetch for this session already merged while
// this one was in flight, skip the merge so the older snapshot can't drop
// newer messages. The newer state is already in the store.
if (!commitFetchSeq(sessionId, seq)) {
debug("message.list stale fetch skipped", { sessionId, seq });
return store.getState().messages.bySession[sessionId] ?? merged;
}
// Idempotent merge: the store reconciles this snapshot against current state,
// preserving object/array identity for unchanged messages so the periodic
// refetch doesn't re-render the whole chat (see reconcileMessages).
store.getState().mergeMessages(sessionId, merged, {
hasMore: response.has_more ?? false,
oldestCursor: merged[0]?.id ?? null,
});
// The store now holds the identity-reconciled array; callers only read length
// and message content from the return, so `merged` is equivalent.
return merged;
}
@@ -0,0 +1,122 @@
import { afterEach, describe, expect, it } from "vitest";
import { createElement, type ReactNode } from "react";
import { act, cleanup, renderHook } from "@testing-library/react";
import { StateProvider, useAppStore } from "@/components/state-provider";
import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http";
import { useUserMessageNavigation } from "./use-message-navigation";
afterEach(() => cleanup());
const SESSION_ID = "sess-1";
function makeMessage(id: string, authorType: Message["author_type"]): Message {
return {
id,
session_id: toSessionId(SESSION_ID),
task_id: toTaskId("task-1"),
author_type: authorType,
content: "",
type: "message",
created_at: "2026-06-12T00:00:00Z",
};
}
// Render the hook alongside setMessages so a test can seed the store and read
// the resolved navigation within the same render tree.
function renderNav(currentMessageId: string, sid: string | null = SESSION_ID) {
function wrapper({ children }: { children: ReactNode }) {
return createElement(StateProvider, null, children);
}
return renderHook(
() => {
const setMessages = useAppStore((s) => s.setMessages);
const nav = useUserMessageNavigation(sid, currentMessageId);
return { setMessages, nav };
},
{ wrapper },
);
}
const NONE = { hasPrevious: false, hasNext: false, previousId: null, nextId: null };
type Nav = ReturnType<typeof renderNav>;
function seed(result: Nav["result"], messages: Message[]) {
act(() => result.current.setMessages(SESSION_ID, messages));
}
describe("useUserMessageNavigation", () => {
it("returns no navigation when the session id is null", () => {
const { result } = renderNav("u1", null);
expect(result.current.nav).toEqual(NONE);
});
it("returns no navigation when the session has no messages", () => {
const { result } = renderNav("u1");
expect(result.current.nav).toEqual(NONE);
});
it("returns no navigation for a lone user message", () => {
const { result } = renderNav("u1");
seed(result, [makeMessage("u1", "user")]);
expect(result.current.nav).toEqual(NONE);
});
it("resolves both neighbours for a middle user message", () => {
const { result } = renderNav("u2");
seed(
result,
["u1", "u2", "u3"].map((id) => makeMessage(id, "user")),
);
expect(result.current.nav).toEqual({
hasPrevious: true,
hasNext: true,
previousId: "u1",
nextId: "u3",
});
});
it("resolves only next for the first user message", () => {
const { result } = renderNav("u1");
seed(
result,
["u1", "u2"].map((id) => makeMessage(id, "user")),
);
expect(result.current.nav).toMatchObject({
hasPrevious: false,
previousId: null,
nextId: "u2",
});
});
it("resolves only previous for the last user message", () => {
const { result } = renderNav("u2");
seed(
result,
["u1", "u2"].map((id) => makeMessage(id, "user")),
);
expect(result.current.nav).toMatchObject({ hasNext: false, previousId: "u1", nextId: null });
});
it("ignores agent messages when computing neighbours", () => {
const { result } = renderNav("u2");
seed(result, [
makeMessage("u1", "user"),
makeMessage("a1", "agent"),
makeMessage("u2", "user"),
makeMessage("a2", "agent"),
makeMessage("u3", "user"),
]);
expect(result.current.nav).toMatchObject({ previousId: "u1", nextId: "u3" });
});
it("returns no navigation when the current message is not a user message", () => {
const { result } = renderNav("a1");
seed(result, [
makeMessage("u1", "user"),
makeMessage("a1", "agent"),
makeMessage("u2", "user"),
]);
expect(result.current.nav).toEqual(NONE);
});
});
+34 -30
View File
@@ -1,35 +1,39 @@
import { useMemo } from "react";
import type { Message } from "@/lib/types/http";
import { useAppStore } from "@/components/state-provider";
export function useMessageNavigation(
messages: Message[],
currentMessageId: string,
filterType: "user" | "agent",
) {
const currentIndex = useMemo(() => {
return messages.findIndex((msg) => msg.id === currentMessageId);
}, [messages, currentMessageId]);
const EMPTY_KEY = "";
const filteredIndices = useMemo(() => {
return messages
.map((msg, idx) => (msg.author_type === filterType ? idx : -1))
.filter((idx) => idx !== -1);
}, [messages, filterType]);
/**
* Prev/next navigation between user messages in a session, sourced directly from
* the store rather than a drilled `allMessages` array.
*
* The selector returns a value-stable key (the comma-joined ids of the session's
* user messages). During agent token streaming the user-message set is unchanged,
* so the key is referentially equal and the subscribing message does not
* re-render — which keeps `ChatMessage` memoized instead of re-running its
* markdown render on every streamed token.
*/
export function useUserMessageNavigation(sessionId: string | null, currentMessageId: string) {
const userIdsKey = useAppStore((state) => {
const messages = sessionId ? state.messages.bySession[sessionId] : undefined;
if (!messages) return EMPTY_KEY;
let key = EMPTY_KEY;
for (const message of messages) {
if (message.author_type === "user") key += `${message.id},`;
}
return key;
});
const previous = useMemo(() => {
const validPrev = filteredIndices.filter((idx) => idx < currentIndex);
return validPrev.length > 0 ? messages[validPrev[validPrev.length - 1]] : null;
}, [filteredIndices, currentIndex, messages]);
const next = useMemo(() => {
const validNext = filteredIndices.filter((idx) => idx > currentIndex);
return validNext.length > 0 ? messages[validNext[0]] : null;
}, [filteredIndices, currentIndex, messages]);
return {
hasPrevious: previous !== null,
hasNext: next !== null,
previous,
next,
};
return useMemo(() => {
const ids = userIdsKey ? userIdsKey.slice(0, -1).split(",") : [];
const index = ids.indexOf(currentMessageId);
const previousId = index > 0 ? ids[index - 1] : null;
const nextId = index >= 0 && index < ids.length - 1 ? ids[index + 1] : null;
return {
hasPrevious: previousId !== null,
hasNext: nextId !== null,
previousId,
nextId,
};
}, [userIdsKey, currentMessageId]);
}
@@ -12,6 +12,10 @@ import {
deduplicateAgentBootResumes,
isAgentBootResumeMessage,
isSetupScriptMessage,
messageListMapsEqual,
messageMapsEqualByIdentity,
reconcileRenderItems,
type RenderItem,
} from "./use-processed-messages";
function makeMessage(
@@ -257,3 +261,111 @@ describe("buildGroupedRenderItems prepare progress placement", () => {
expect(result.map((item) => item.type)).toEqual(["message", "prepare_progress"]);
});
});
describe("messageMapsEqualByIdentity", () => {
it("treats maps with the same keys and identical Message refs as equal", () => {
const a = makeMessage("perm-1", "permission_request");
const b = makeMessage("perm-2", "permission_request");
const first = new Map<string, Message>([
["tc-1", a],
["tc-2", b],
]);
const second = new Map<string, Message>([
["tc-1", a],
["tc-2", b],
]);
expect(messageMapsEqualByIdentity(first, second)).toBe(true);
});
it("is false when sizes differ", () => {
const a = makeMessage("perm-1", "permission_request");
const first = new Map<string, Message>([["tc-1", a]]);
const second = new Map<string, Message>([
["tc-1", a],
["tc-2", makeMessage("perm-2", "permission_request")],
]);
expect(messageMapsEqualByIdentity(first, second)).toBe(false);
});
it("is false when a key maps to a different Message ref (e.g. status changed)", () => {
const first = new Map<string, Message>([["tc-1", makeMessage("perm-1", "permission_request")]]);
const second = new Map<string, Message>([
["tc-1", makeMessage("perm-1", "permission_request")],
]);
expect(messageMapsEqualByIdentity(first, second)).toBe(false);
});
});
describe("messageListMapsEqual", () => {
it("treats maps whose array values share positional Message refs as equal", () => {
const c1 = makeMessage("child-1", "tool_call");
const c2 = makeMessage("child-2", "tool_call");
const first = new Map<string, Message[]>([["parent-1", [c1, c2]]]);
const second = new Map<string, Message[]>([["parent-1", [c1, c2]]]);
expect(messageListMapsEqual(first, second)).toBe(true);
});
it("is false when an array length changes (new child appended)", () => {
const c1 = makeMessage("child-1", "tool_call");
const first = new Map<string, Message[]>([["parent-1", [c1]]]);
const second = new Map<string, Message[]>([
["parent-1", [c1, makeMessage("child-2", "tool_call")]],
]);
expect(messageListMapsEqual(first, second)).toBe(false);
});
it("is false when a child Message ref changes positionally", () => {
const c1 = makeMessage("child-1", "tool_call");
const first = new Map<string, Message[]>([["parent-1", [c1]]]);
const second = new Map<string, Message[]>([
["parent-1", [makeMessage("child-1", "tool_call")]],
]);
expect(messageListMapsEqual(first, second)).toBe(false);
});
});
describe("reconcileRenderItems", () => {
const turnGroup = (id: string, messages: Message[]): RenderItem => ({
type: "turn_group",
id,
turnId: messages[0]?.turn_id ?? null,
messages,
});
const messageItem = (message: Message): RenderItem => ({ type: "message", message });
it("returns the prior array unchanged when nothing changed", () => {
const m1 = makeMessage("m1", "message");
const prev = [messageItem(m1)];
const next = [messageItem(m1)];
expect(reconcileRenderItems(prev, next)).toBe(prev);
});
it("reuses unchanged turn-group wrappers while replacing the changed one", () => {
const a1 = makeMessage("a1", "tool_call");
const a2 = makeMessage("a2", "tool_call");
const b1 = makeMessage("b1", "tool_call");
const b2 = makeMessage("b2", "tool_call");
const groupA = turnGroup("turn-group-a1", [a1, a2]);
const groupBPrev = turnGroup("turn-group-b1", [b1]);
const prev = [groupA, groupBPrev];
// groupA identical content (rebuilt wrapper); groupB grows by a token.
const groupBNext = turnGroup("turn-group-b1", [b1, b2]);
const next = [turnGroup("turn-group-a1", [a1, a2]), groupBNext];
const result = reconcileRenderItems(prev, next);
expect(result[0]).toBe(groupA); // stable → memo holds
expect(result[1]).toBe(groupBNext); // changed → fresh wrapper
});
it("keeps stable message wrappers and appends a new item", () => {
const m1 = makeMessage("m1", "message");
const prev = [messageItem(m1)];
const m2 = makeMessage("m2", "message");
const next = [messageItem(m1), messageItem(m2)];
const result = reconcileRenderItems(prev, next);
expect(result).not.toBe(prev);
expect(result[0]).toBe(prev[0]);
expect(result).toHaveLength(2);
});
});
+102 -6
View File
@@ -1,4 +1,4 @@
import { useEffect, useMemo } from "react";
import { useEffect, useMemo, useState } from "react";
import {
sessionId as toSessionId,
taskId as toTaskId,
@@ -157,6 +157,98 @@ function buildSubagentChildIds(childrenByParentToolCallId: Map<string, Message[]
return set;
}
/** True when both maps hold the same keys pointing at the same Message refs. */
export function messageMapsEqualByIdentity(
a: Map<string, Message>,
b: Map<string, Message>,
): boolean {
if (a.size !== b.size) return false;
for (const [key, value] of a) {
if (b.get(key) !== value) return false;
}
return true;
}
/** True when both maps hold the same keys pointing at arrays whose Message refs
* match positionally. */
export function messageListMapsEqual(
a: Map<string, Message[]>,
b: Map<string, Message[]>,
): boolean {
if (a.size !== b.size) return false;
for (const [key, value] of a) {
const other = b.get(key);
if (!other || other.length !== value.length) return false;
for (let i = 0; i < value.length; i++) {
if (value[i] !== other[i]) return false;
}
}
return true;
}
/** Keep a prior reference when the freshly-built value is content-equal. The
* per-token `messages` array churns every reference downstream; holding the
* prior value lets `memo` on each message component survive commits where the
* derived map content didn't actually change. Uses the React-endorsed
* "adjust state during render" pattern so we never mutate a ref mid-render. */
function useStableValue<T>(next: T, equal: (a: T, b: T) => boolean): T {
const [stable, setStable] = useState(next);
if (!equal(stable, next)) {
setStable(next);
return next;
}
return stable;
}
function renderItemKey(item: RenderItem): string {
return item.type === "message" ? `m:${item.message.id}` : item.id;
}
/** Content-equal when the wrapper would render identically. Turn groups compare
* their messages by positional identity so a streaming token in the active turn
* yields a fresh wrapper while quiescent earlier turns stay equal. */
function renderItemsEqual(a: RenderItem, b: RenderItem): boolean {
if (a.type !== b.type) return false;
if (a.type === "message" && b.type === "message") return a.message === b.message;
if (a.type === "prepare_progress" && b.type === "prepare_progress") return a.id === b.id;
if (a.type === "turn_group" && b.type === "turn_group") {
if (a.id !== b.id || a.turnId !== b.turnId || a.messages.length !== b.messages.length) {
return false;
}
for (let i = 0; i < a.messages.length; i++) {
if (a.messages[i] !== b.messages[i]) return false;
}
return true;
}
return false;
}
/** Reuse prior wrapper objects for items whose content didn't change so the
* `memo` on TurnGroupMessage/MessageRenderer survives new-message commits.
* Returns `prev` unchanged when every slot is positionally identical. */
export function reconcileRenderItems(prev: RenderItem[], next: RenderItem[]): RenderItem[] {
const prevByKey = new Map<string, RenderItem>();
for (const item of prev) prevByKey.set(renderItemKey(item), item);
let identical = prev.length === next.length;
const result = next.map((nextItem, i) => {
const prevItem = prevByKey.get(renderItemKey(nextItem));
const reused = prevItem && renderItemsEqual(prevItem, nextItem) ? prevItem : nextItem;
if (reused !== prev[i]) identical = false;
return reused;
});
return identical ? prev : result;
}
function useStableGroupedItems(next: RenderItem[]): RenderItem[] {
const [stable, setStable] = useState(next);
const reconciled = reconcileRenderItems(stable, next);
if (reconciled !== stable) {
setStable(reconciled);
return reconciled;
}
return stable;
}
function isRecoveryMessage(message: Message): boolean {
const meta = message.metadata as Record<string, unknown> | undefined;
return meta?.recovery_actions === true;
@@ -407,10 +499,13 @@ export function useProcessedMessages(
options: ProcessedMessagesOptions = {},
) {
const toolCallIds = useMemo(() => buildToolCallIds(messages), [messages]);
const permissionsByToolCallId = useMemo(() => buildPermissionsByToolCallId(messages), [messages]);
const childrenByParentToolCallId = useMemo(
() => buildChildrenByParentToolCallId(messages),
[messages],
const permissionsByToolCallId = useStableValue(
useMemo(() => buildPermissionsByToolCallId(messages), [messages]),
messageMapsEqualByIdentity,
);
const childrenByParentToolCallId = useStableValue(
useMemo(() => buildChildrenByParentToolCallId(messages), [messages]),
messageListMapsEqual,
);
const subagentChildIds = useMemo(
() => buildSubagentChildIds(childrenByParentToolCallId),
@@ -467,7 +562,7 @@ export function useProcessedMessages(
return { regularMessages: regular, footerActionMessages: footer };
}, [allMessages]);
const groupedItems = useMemo<RenderItem[]>(() => {
const rawGroupedItems = useMemo<RenderItem[]>(() => {
return buildGroupedItemsForHook({
regularMessages,
allSessionMessages: messages,
@@ -475,6 +570,7 @@ export function useProcessedMessages(
hasOlderMessages: options.hasOlderMessages ?? false,
});
}, [regularMessages, resolvedSessionId, messages, options.hasOlderMessages]);
const groupedItems = useStableGroupedItems(rawGroupedItems);
useDebugProcessedPipeline({
sessionId: resolvedSessionId,
@@ -0,0 +1,67 @@
import { beforeEach, describe, expect, it } from "vitest";
import {
__MAX_CACHE_ENTRIES,
__lruSize,
__markdownParseCount,
__resetMarkdownCounters,
normalizeCached,
normalizeMarkdown,
} from "./normalize-cache";
describe("normalizeCached", () => {
beforeEach(() => {
__resetMarkdownCounters();
});
it("parses once for repeated identical input (cache hit)", () => {
const input = "```go\nfunc f() {}```\nprose";
const first = normalizeCached(input);
const second = normalizeCached(input);
expect(second).toBe(first);
expect(__markdownParseCount()).toBe(1);
});
it("parses again for different input", () => {
normalizeCached("alpha");
normalizeCached("beta");
expect(__markdownParseCount()).toBe(2);
});
it("produces output byte-identical to normalizeMarkdown", () => {
const inputs = [
"```go\nfunc f() {}```\nprose",
"Use `code` inline.",
"",
"single line",
"```go\nx```\nprose\n",
];
for (const input of inputs) {
expect(normalizeCached(input)).toBe(normalizeMarkdown(input));
}
});
it("evicts oldest entries past the cap (bounded LRU)", () => {
for (let i = 0; i < __MAX_CACHE_ENTRIES + 50; i++) {
normalizeCached(`unique-content-${i}`);
}
expect(__lruSize()).toBeLessThanOrEqual(__MAX_CACHE_ENTRIES);
});
it("keeps a recently-used entry warm despite overflow", () => {
normalizeCached("keep-me");
for (let i = 0; i < __MAX_CACHE_ENTRIES; i++) {
normalizeCached(`filler-${i}`);
normalizeCached("keep-me"); // refresh recency each round
}
const before = __markdownParseCount();
normalizeCached("keep-me");
expect(__markdownParseCount()).toBe(before); // still cached, no new parse
});
it("__resetMarkdownCounters clears cache and counter", () => {
normalizeCached("something");
__resetMarkdownCounters();
expect(__markdownParseCount()).toBe(0);
expect(__lruSize()).toBe(0);
});
});
+125
View File
@@ -0,0 +1,125 @@
/**
* Pure markdown normalization plus a bounded, value-keyed LRU cache.
*
* `normalizeMarkdown` is a pure string transform (moved here from
* markdown-components so the cache has no React dependency). `normalizeCached`
* wraps it in a hard-capped LRU keyed by the raw input string, so repeated
* renders of the same content never re-run the transform. The cache is keyed by
* content only (never by task/session/message id), so it cannot leak per-task
* state: entries are anonymous strings that age out via LRU.
*/
const FENCE_OPEN_RE = /^ {0,3}(`{3,})/;
const TRAILING_FENCE_RE = /(`{3,})\s*$/;
function pureCloseLength(line: string, openCount: number): number | null {
const match = /^ {0,3}(`{3,})\s*$/.exec(line);
if (!match || match[1].length < openCount) return null;
return match[1].length;
}
function gluedCloseLength(line: string, openCount: number): number | null {
const match = TRAILING_FENCE_RE.exec(line);
if (!match || match[1].length < openCount) return null;
// Reject pure-fence lines (already handled by pureCloseLength) and lines
// where everything before the trailing run is whitespace only.
const head = line.slice(0, line.length - match[0].length).trimEnd();
if (head.length === 0) return null;
return match[1].length;
}
/**
* Pre-process a markdown string to repair fenced code blocks that have their
* closing fence glued to the last code line (`...}\`\`\`\n`prose`). Without
* this, CommonMark/GFM treats the glued backticks as code content, so the
* fence never closes and following prose gets swallowed into one huge code
* node. We split such lines into `<content>\n<backticks>` only when we're
* inside an open fence whose opener run length is ≤ the trailing run length.
*
* Pure string preprocessing, intentionally not a remark plugin.
*/
export function normalizeMarkdown(input: string): string {
if (!input || input.length === 0) return input;
const hadTrailingNewline = input.endsWith("\n");
const lines = input.split("\n");
const out: string[] = [];
let openCount: number | null = null;
for (const line of lines) {
if (openCount === null) {
const opener = FENCE_OPEN_RE.exec(line);
if (opener) openCount = opener[1].length;
out.push(line);
continue;
}
if (pureCloseLength(line, openCount) !== null) {
openCount = null;
out.push(line);
continue;
}
const glued = gluedCloseLength(line, openCount);
if (glued !== null) {
const trailingMatch = TRAILING_FENCE_RE.exec(line)!;
const head = line.slice(0, line.length - trailingMatch[0].length);
out.push(head);
out.push("`".repeat(glued));
openCount = null;
continue;
}
out.push(line);
}
const result = out.join("\n");
return hadTrailingNewline && !result.endsWith("\n") ? result + "\n" : result;
}
// ── Bounded LRU cache ───────────────────────────────────────────────
const MAX_CACHE_ENTRIES = 500;
const normalizeCache = new Map<string, string>();
let parseCount = 0;
/**
* Cached variant of {@link normalizeMarkdown}. Keyed by the raw input string;
* a hit refreshes recency (Map preserves insertion order, so delete+set moves
* the entry to the end). On overflow the oldest entry is evicted.
*/
export function normalizeCached(input: string): string {
const cached = normalizeCache.get(input);
if (cached !== undefined) {
normalizeCache.delete(input);
normalizeCache.set(input, cached);
return cached;
}
parseCount += 1;
const result = normalizeMarkdown(input);
// Evict before insert so the map never exceeds the cap. A miss means the key
// isn't present, so the cache holds at most MAX_CACHE_ENTRIES - 1 entries here.
if (normalizeCache.size >= MAX_CACHE_ENTRIES) {
const oldest = normalizeCache.keys().next().value;
if (oldest !== undefined) normalizeCache.delete(oldest);
}
normalizeCache.set(input, result);
return result;
}
// ── Test-only instrumentation ───────────────────────────────────────
/** Number of real normalize passes (cache misses) since the last reset. */
export function __markdownParseCount(): number {
return parseCount;
}
/** Clears the cache and resets the parse counter. Test-only. */
export function __resetMarkdownCounters(): void {
parseCount = 0;
normalizeCache.clear();
}
/** Current cache size. Test-only (asserts the LRU stays bounded). */
export function __lruSize(): number {
return normalizeCache.size;
}
/** Hard cap on cached entries. Test-only. */
export const __MAX_CACHE_ENTRIES = MAX_CACHE_ENTRIES;
@@ -0,0 +1,130 @@
import { describe, expect, it } from "vitest";
import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http";
import { reconcileMessages, signatureOf } from "./message-signature";
const TS = "2024-01-01T00:00:00Z";
function makeMessage(overrides: Partial<Message>): Message {
return {
id: "msg-1",
task_id: toTaskId("task-1"),
session_id: toSessionId("sess-1"),
author_type: "user",
content: "hello",
type: "message",
created_at: TS,
...overrides,
} as Message;
}
describe("signatureOf", () => {
it("returns equal signatures for messages with equal content fields", () => {
const a = makeMessage({ content: "one", metadata: { foo: { bar: 1 } } });
const b = makeMessage({ content: "one", metadata: { foo: { bar: 1 } } });
expect(signatureOf(a)).toBe(signatureOf(b));
});
it("differs when content changes", () => {
expect(signatureOf(makeMessage({ content: "one" }))).not.toBe(
signatureOf(makeMessage({ content: "two" })),
);
});
it("differs when metadata changes", () => {
expect(signatureOf(makeMessage({ metadata: { a: 1 } }))).not.toBe(
signatureOf(makeMessage({ metadata: { a: 2 } })),
);
});
it("is stable regardless of metadata key insertion order", () => {
const a = makeMessage({ metadata: { a: 1, b: 2 } });
const b = makeMessage({ metadata: { b: 2, a: 1 } });
expect(signatureOf(a)).toBe(signatureOf(b));
});
it("differs when turn_id or requests_input change", () => {
const base = makeMessage({ content: "x" });
expect(signatureOf(base)).not.toBe(signatureOf(makeMessage({ content: "x", turn_id: "t1" })));
expect(signatureOf(base)).not.toBe(
signatureOf(makeMessage({ content: "x", requests_input: true })),
);
});
it("caches per object (same object returns identical signature)", () => {
const m = makeMessage({ content: "cached" });
expect(signatureOf(m)).toBe(signatureOf(m));
});
it("short-circuits to updated_at when present (authoritative signal)", () => {
const a = makeMessage({ content: "x", updated_at: TS });
const b = makeMessage({ content: "different", updated_at: TS });
expect(signatureOf(a)).toBe(signatureOf(b));
});
it("differs when updated_at advances", () => {
const a = makeMessage({ content: "x", updated_at: TS });
const b = makeMessage({ content: "x", updated_at: "2024-01-01T00:00:01Z" });
expect(signatureOf(a)).not.toBe(signatureOf(b));
});
it("falls back to content hash when updated_at is absent", () => {
expect(signatureOf(makeMessage({ content: "one" }))).not.toBe(
signatureOf(makeMessage({ content: "two" })),
);
});
});
describe("reconcileMessages", () => {
it("returns the prev array unchanged when every message is signature-equal", () => {
const prev = [
makeMessage({ id: "a", content: "one" }),
makeMessage({ id: "b", content: "two", metadata: { foo: { bar: 1 } } }),
];
const next = [
makeMessage({ id: "a", content: "one" }),
makeMessage({ id: "b", content: "two", metadata: { foo: { bar: 1 } } }),
];
expect(reconcileMessages(prev, next)).toBe(prev);
});
it("reuses prior references for unchanged messages and keeps the changed one fresh", () => {
const prev = [
makeMessage({ id: "a", content: "one" }),
makeMessage({ id: "b", content: "two" }),
];
const next = [
makeMessage({ id: "a", content: "one" }),
makeMessage({ id: "b", content: "two-edited" }),
];
const result = reconcileMessages(prev, next);
expect(result).not.toBe(prev);
expect(result[0]).toBe(prev[0]);
expect(result[1]).toBe(next[1]);
});
it("returns next when a message is appended, reusing existing references", () => {
const prev = [makeMessage({ id: "a", content: "one" })];
const next = [
makeMessage({ id: "a", content: "one" }),
makeMessage({ id: "b", content: "two" }),
];
const result = reconcileMessages(prev, next);
expect(result).not.toBe(prev);
expect(result[0]).toBe(prev[0]);
expect(result[1]).toBe(next[1]);
});
it("returns next directly when prev is empty or undefined", () => {
const next = [makeMessage({ id: "a" })];
expect(reconcileMessages([], next)).toBe(next);
expect(reconcileMessages(undefined, next)).toBe(next);
});
it("detects a metadata-only change", () => {
const prev = [makeMessage({ id: "a", content: "one", metadata: { x: 1 } })];
const next = [makeMessage({ id: "a", content: "one", metadata: { x: 2 } })];
const result = reconcileMessages(prev, next);
expect(result).not.toBe(prev);
expect(result[0]).toBe(next[0]);
});
});
@@ -0,0 +1,83 @@
import type { Message } from "@/lib/types/http";
const SEP = "\u0000";
const signatureCache = new WeakMap<Message, string>();
/**
* djb2 string hash → unsigned base36. Non-crypto; paired with a length prefix
* and the message id (during reconciliation) so collisions cannot cause a
* visible reuse of a changed message.
*/
function hashString(input: string): string {
let hash = 5381;
for (let i = 0; i < input.length; i++) {
hash = ((hash << 5) + hash) ^ input.charCodeAt(i);
}
return (hash >>> 0).toString(36);
}
function serializeMetadata(metadata: Record<string, unknown> | undefined): string {
if (!metadata) return "";
// Sort keys so signature is stable regardless of insertion order.
const keys = Object.keys(metadata).sort();
let out = "";
for (const key of keys) out += `${key}=${JSON.stringify(metadata[key])}${SEP}`;
return out;
}
function contentHashSignature(message: Message): string {
const parts =
message.content +
SEP +
message.type +
SEP +
(message.turn_id ?? "") +
SEP +
(message.requests_input ? "1" : "0") +
SEP +
(message.raw_content ?? "") +
SEP +
serializeMetadata(message.metadata);
return `h:${parts.length}:${hashString(parts)}`;
}
/**
* Content signature of a message, cached per-object via a WeakMap so each
* Message object is hashed at most once. Two messages with the same id and the
* same signature are treated as unchanged for reconciliation. The WeakMap is
* keyed by the Message object, so entries are GC'd once messages leave the
* store (e.g. task delete/archive) — no cap or cleanup hook needed.
*
* When the backend supplies `updated_at` (the authoritative per-message change
* signal, bumped on every content/metadata mutation including streaming tokens),
* the signature short-circuits to it — an O(1) compare with no content hashing.
* Older payloads without `updated_at` fall back to the content hash.
*/
export function signatureOf(message: Message): string {
const cached = signatureCache.get(message);
if (cached !== undefined) return cached;
const signature = message.updated_at ? `u:${message.updated_at}` : contentHashSignature(message);
signatureCache.set(message, signature);
return signature;
}
/**
* Reconcile a full snapshot against the previous array, preserving object
* identity for messages whose signature is unchanged and returning the prev
* array reference itself when nothing moved (so array identity is preserved and
* no downstream re-render fires). Only genuinely-changed messages get a fresh
* reference.
*/
export function reconcileMessages(prev: Message[] | undefined, next: Message[]): Message[] {
if (!prev || prev.length === 0) return next;
const prevById = new Map<string, Message>();
for (const message of prev) prevById.set(message.id, message);
let identical = prev.length === next.length;
const result = next.map((message, i) => {
const previous = prevById.get(message.id);
const reused = previous && signatureOf(previous) === signatureOf(message) ? previous : message;
if (reused !== prev[i]) identical = false;
return reused;
});
return identical ? prev : result;
}
@@ -0,0 +1,97 @@
import { describe, it, expect } from "vitest";
import { create } from "zustand";
import { immer } from "zustand/middleware/immer";
import { createSessionSlice } from "./session-slice";
import type { SessionSlice } from "./types";
import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http";
function makeStore() {
return create<SessionSlice>()(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
immer((set) => ({ ...(createSessionSlice as any)(set) })),
);
}
const SESSION = "sess-1";
function makeMessage(overrides: Partial<Message>): Message {
return {
id: "m1",
task_id: toTaskId("task-1"),
session_id: toSessionId(SESSION),
author_type: "user",
content: "hello",
type: "message",
created_at: "2024-01-01T00:00:00Z",
...overrides,
} as Message;
}
function snapshot(): Message[] {
return [makeMessage({ id: "a", content: "one" }), makeMessage({ id: "b", content: "two" })];
}
describe("mergeMessages", () => {
it("preserves the array reference on a no-op refetch", () => {
const store = makeStore();
store.getState().mergeMessages(SESSION, snapshot());
const first = store.getState().messages.bySession[SESSION];
// Fresh, equal objects as a refetch would produce.
store.getState().mergeMessages(SESSION, snapshot());
expect(store.getState().messages.bySession[SESSION]).toBe(first);
});
it("changes the array ref and only the changed element when one message changes", () => {
const store = makeStore();
store.getState().mergeMessages(SESSION, snapshot());
const first = store.getState().messages.bySession[SESSION];
store
.getState()
.mergeMessages(SESSION, [
makeMessage({ id: "a", content: "one" }),
makeMessage({ id: "b", content: "two-edited" }),
]);
const next = store.getState().messages.bySession[SESSION];
expect(next).not.toBe(first);
expect(next[0]).toBe(first[0]); // unchanged message kept its identity
expect(next[1]).not.toBe(first[1]);
expect(next[1].content).toBe("two-edited");
});
it("appends new messages while reusing existing references", () => {
const store = makeStore();
store.getState().mergeMessages(SESSION, [makeMessage({ id: "a", content: "one" })]);
const first = store.getState().messages.bySession[SESSION];
store
.getState()
.mergeMessages(SESSION, [
makeMessage({ id: "a", content: "one" }),
makeMessage({ id: "b", content: "two" }),
]);
const next = store.getState().messages.bySession[SESSION];
expect(next).not.toBe(first);
expect(next[0]).toBe(first[0]);
expect(next).toHaveLength(2);
});
it("detects a metadata-only change", () => {
const store = makeStore();
store.getState().mergeMessages(SESSION, [makeMessage({ id: "a", metadata: { x: 1 } })]);
const first = store.getState().messages.bySession[SESSION];
store.getState().mergeMessages(SESSION, [makeMessage({ id: "a", metadata: { x: 2 } })]);
const next = store.getState().messages.bySession[SESSION];
expect(next).not.toBe(first);
expect((next[0].metadata as { x: number }).x).toBe(2);
});
it("applies metadata (hasMore / oldestCursor) on merge", () => {
const store = makeStore();
store.getState().mergeMessages(SESSION, snapshot(), { hasMore: true, oldestCursor: "a" });
const meta = store.getState().messages.metaBySession[SESSION];
expect(meta.hasMore).toBe(true);
expect(meta.oldestCursor).toBe("a");
});
});
@@ -1,6 +1,8 @@
import type { StateCreator } from "zustand";
import type { TaskSession } from "@/lib/types/http";
import { original } from "immer";
import type { Message, TaskSession } from "@/lib/types/http";
import type { SessionSlice, SessionSliceState } from "./types";
import { reconcileMessages } from "./message-signature";
import { migrateEnvKeyedData } from "@/lib/state/slices/session-runtime/session-runtime-slice";
import { prepareResultToSessionState } from "@/lib/state/slices/session-runtime/prepare-result";
import { createDebugLogger, isDebug } from "@/lib/debug/log";
@@ -175,6 +177,25 @@ function buildMessageActions(set: ImmerSet) {
);
messages[index] = merged;
}),
mergeMessages: (
sessionId: string,
messages: Parameters<SessionSlice["mergeMessages"]>[1],
meta?: Parameters<SessionSlice["mergeMessages"]>[2],
) =>
set((draft) => {
const prevDraft = draft.messages.bySession[sessionId];
const prev = (prevDraft ? (original(prevDraft) ?? prevDraft) : undefined) as
| Message[]
| undefined;
const reconciled = reconcileMessages(prev, messages);
// Only replace the array when identity actually changed, so a no-op
// refetch preserves the array reference and triggers no re-render.
if (reconciled !== prev) {
draft.messages.bySession[sessionId] = reconciled;
}
ensureMessageMeta(draft.messages.metaBySession, sessionId);
if (meta) applyMessageMeta(draft.messages.metaBySession, sessionId, meta);
}),
prependMessages: (
sessionId: string,
messages: Parameters<SessionSlice["prependMessages"]>[1],
@@ -150,6 +150,17 @@ export type SessionSliceActions = {
) => void;
addMessage: (message: Message) => void;
updateMessage: (message: Message) => void;
/**
* Idempotent full-snapshot merge: reconciles `messages` against the current
* stored array, preserving object identity for unchanged messages and the
* array reference itself when nothing changed (see `reconcileMessages`). Used
* by periodic refetches so a no-op tick triggers zero re-renders.
*/
mergeMessages: (
sessionId: string,
messages: Message[],
meta?: { hasMore?: boolean; oldestCursor?: string | null },
) => void;
prependMessages: (
sessionId: string,
messages: Message[],
+5
View File
@@ -387,6 +387,11 @@ export type AppState = {
meta?: { hasMore?: boolean; oldestCursor?: string | null },
) => void;
addMessage: (message: Message) => void;
mergeMessages: (
sessionId: string,
messages: Message[],
meta?: { hasMore?: boolean; oldestCursor?: string | null },
) => void;
addTurn: (turn: Turn) => void;
completeTurn: (sessionId: string, turnId: string, completedAt: string) => void;
setActiveTurn: (sessionId: string, turnId: string | null) => void;
+1
View File
@@ -256,6 +256,7 @@ export type MessageAddedPayload = {
metadata?: Record<string, unknown>;
requests_input?: boolean;
created_at: string;
updated_at?: string;
};
export type TaskSessionStateChangedPayload = {
+2
View File
@@ -663,6 +663,8 @@ export type Message = {
metadata?: Record<string, unknown>;
requests_input?: boolean;
created_at: string;
/** Authoritative per-message change signal; advances on every content/metadata update. */
updated_at?: string;
};
export type Turn = {
+2
View File
@@ -23,6 +23,7 @@ export function registerMessagesHandlers(store: StoreApi<AppState>): WsHandlers
metadata: payload.metadata,
requests_input: payload.requests_input,
created_at: payload.created_at,
updated_at: payload.updated_at,
});
},
"session.message.updated": (message) => {
@@ -43,6 +44,7 @@ export function registerMessagesHandlers(store: StoreApi<AppState>): WsHandlers
metadata: payload.metadata,
requests_input: payload.requests_input,
created_at: payload.created_at,
updated_at: payload.updated_at,
});
},
};