diff --git a/cmd/gateway.go b/cmd/gateway.go index 9ad41ca8..5e1adb59 100644 --- a/cmd/gateway.go +++ b/cmd/gateway.go @@ -440,7 +440,7 @@ func runGateway() { // Register all RPC methods server.SetLogTee(logTee) server.SetRuntimeLogsHandler(httpapi.NewRuntimeLogsHandler(logTee)) - pairingMethods, heartbeatMethods, chatMethods, cfgPermsMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions, pgStores.SystemConfigs, pgStores.Tenants, pgStores.SkillTenantCfgs, audioMgr, usageCapSvc) + pairingMethods, heartbeatMethods, chatMethods, cfgPermsMethods := registerAllMethods(server, agentRouter, pgStores.Sessions, pgStores.RunTimeline, pgStores.Cron, pgStores.Pairing, cfg, cfgPath, workspace, dataDir, msgBus, execApprovalMgr, pgStores.Agents, pgStores.Skills, pgStores.ConfigSecrets, pgStores.Teams, contextFileInterceptor, logTee, pgStores.Heartbeats, pgStores.ConfigPermissions, pgStores.SystemConfigs, pgStores.Tenants, pgStores.SkillTenantCfgs, audioMgr, usageCapSvc) // Phase 3: Agent hooks RPC methods (hooks.list/create/update/delete/toggle/test/history). if hs, ok := pgStores.Hooks.(hooks.HookStore); ok && hs != nil { diff --git a/cmd/gateway_http_handlers.go b/cmd/gateway_http_handlers.go index 2f255e3f..f5d1bc20 100644 --- a/cmd/gateway_http_handlers.go +++ b/cmd/gateway_http_handlers.go @@ -48,7 +48,7 @@ func wireHTTP(stores *store.Stores, defaultWorkspace, dataDir, bundledSkillsDir } if stores != nil && stores.Tracing != nil { - tracesH = httpapi.NewTracesHandler(stores.Tracing) + tracesH = httpapi.NewTracesHandler(stores.Tracing, stores.RunTimeline) } if stores != nil && stores.MCP != nil { diff --git a/cmd/gateway_managed.go b/cmd/gateway_managed.go index 87809779..179ce8da 100644 --- a/cmd/gateway_managed.go +++ b/cmd/gateway_managed.go @@ -204,6 +204,7 @@ func wireExtras( sharedHookHandlers = handlers slog.Info("agent hooks dispatcher wired", "handlers", "command,http,prompt") } + timelineRecorder := agent.NewRunTimelineRecorder(stores.RunTimeline) resolver := agent.NewManagedResolver(agent.ResolverDeps{ AgentStore: stores.Agents, @@ -299,6 +300,7 @@ func wireExtras( Payload: event, TenantID: event.TenantID, }) + timelineRecorder.Record(event) }, }) agentRouter.SetResolver(resolver) diff --git a/cmd/gateway_methods.go b/cmd/gateway_methods.go index 7d7f0664..f63282ef 100644 --- a/cmd/gateway_methods.go +++ b/cmd/gateway_methods.go @@ -15,7 +15,7 @@ import ( usagecaps "github.com/nextlevelbuilder/goclaw/internal/usage/caps" ) -func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore, contextFileInterceptor *tools.ContextFileInterceptor, logTee *gateway.LogTee, heartbeatStore store.HeartbeatStore, configPermStore store.ConfigPermissionStore, sysConfigStore store.SystemConfigStore, tenantStore store.TenantStore, skillTenantCfgStore store.SkillTenantConfigStore, audioMgr *audio.Manager, usageCapSvc *usagecaps.Service) (*methods.PairingMethods, *methods.HeartbeatMethods, *methods.ChatMethods, *methods.ConfigPermissionsMethods) { +func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore store.SessionStore, runTimeline store.RunTimelineStore, cronStore store.CronStore, pairingStore store.PairingStore, cfg *config.Config, cfgPath, workspace, dataDir string, msgBus *bus.MessageBus, execApprovalMgr *tools.ExecApprovalManager, agentStore store.AgentStore, skillStore store.SkillStore, configSecretsStore store.ConfigSecretsStore, teamStore store.TeamStore, contextFileInterceptor *tools.ContextFileInterceptor, logTee *gateway.LogTee, heartbeatStore store.HeartbeatStore, configPermStore store.ConfigPermissionStore, sysConfigStore store.SystemConfigStore, tenantStore store.TenantStore, skillTenantCfgStore store.SkillTenantConfigStore, audioMgr *audio.Manager, usageCapSvc *usagecaps.Service) (*methods.PairingMethods, *methods.HeartbeatMethods, *methods.ChatMethods, *methods.ConfigPermissionsMethods) { router := server.Router() // Phase 1: Core methods @@ -25,6 +25,7 @@ func registerAllMethods(server *gateway.Server, agents *agent.Router, sessStore chatMethods.Register(router) methods.NewAgentsMethods(agents, cfg, cfgPath, workspace, agentStore, contextFileInterceptor, msgBus).Register(router) methods.NewSessionsMethods(sessStore, msgBus, cfg).Register(router) + methods.NewRunTimelineMethods(runTimeline, cfg).Register(router) configMethods := methods.NewConfigMethods(cfg, cfgPath, configSecretsStore, msgBus) if sysConfigStore != nil { configMethods.SetSystemConfigSync(func(ctx context.Context, c *config.Config) { diff --git a/docs/18-http-api.md b/docs/18-http-api.md index fbabdba6..44bedddf 100644 --- a/docs/18-http-api.md +++ b/docs/18-http-api.md @@ -1297,6 +1297,7 @@ LLM call tracing and cost analysis. | `GET` | `/v1/traces/follow` | Poll trace changes for one session or agent | | `GET` | `/v1/traces/{traceID}` | Get trace with spans | | `GET` | `/v1/traces/{traceID}/export` | Export trace tree (gzipped JSON) | +| `GET` | `/v1/runs/{runID}/timeline` | Get persisted run archive timeline items | **Filters:** `agent_id`, `user_id`, `session_key`, `status`, `channel` @@ -1314,6 +1315,42 @@ Follow response: } ``` +### Run Timeline + +`GET /v1/runs/{runID}/timeline` returns display-safe archive entries for one +agent run. Optional query params: `session_key`, `limit` (default 200, max 500), +and `offset`. Non-admin callers only receive entries owned by their effective +`X-GoClaw-User-Id`. + +Timeline items are ordered by run sequence for `run_id` reads and include only +safe previews for tool arguments/results. Raw thinking is not persisted. + +Example: + +```json +{ + "run_id": "run-123", + "session_key": "agent:demo:direct:user-1", + "items": [{ + "id": "019e...", + "run_id": "run-123", + "session_key": "agent:demo:direct:user-1", + "seq": 2, + "item_type": "tool.call", + "status": "running", + "title": "web_fetch", + "preview": "{\"url\":\"https://example.com\"}", + "tool_name": "web_fetch", + "tool_call_id": "call_123", + "trace_id": "019e...", + "span_id": "019e...", + "created_at": "2026-05-29T10:00:00Z" + }], + "limit": 200, + "offset": 0 +} +``` + ### Costs | Method | Path | Description | diff --git a/docs/19-websocket-rpc.md b/docs/19-websocket-rpc.md index c1d77d57..91a52a7e 100644 --- a/docs/19-websocket-rpc.md +++ b/docs/19-websocket-rpc.md @@ -232,10 +232,56 @@ Delete an agent (admin only). | `sessions.patch` | Update label, model, metadata | | `sessions.delete` | Delete session | | `sessions.reset` | Clear session messages | +| `run.timeline.get` | Get archived run/session timeline items | **`sessions.list` request:** `{agentId, limit, offset}` **Response:** `{sessions[], total, limit, offset}` +### `run.timeline.get` + +Fetch display-safe timeline entries captured during agent runs. Pass `runId` for +one run, or `sessionKey` for the session archive panel. At least one is +required. `limit` defaults to `200` and is capped at `500`; `offset` paginates. +Viewer role can read this method. Non-admin callers only receive entries whose +`user_id` matches their connected user. + +**Request:** + +```json +{ + "runId": "run-123", + "sessionKey": "agent:demo:direct:user-1", + "limit": 100, + "offset": 0 +} +``` + +**Response:** + +```json +{ + "runId": "run-123", + "sessionKey": "agent:demo:direct:user-1", + "items": [{ + "id": "019e...", + "run_id": "run-123", + "session_key": "agent:demo:direct:user-1", + "seq": 1, + "item_type": "assistant.message", + "status": "completed", + "title": "assistant", + "preview": "I will check that now.", + "created_at": "2026-05-29T10:00:00Z" + }], + "limit": 100, + "offset": 0 +} +``` + +Timeline items include `activity`, `assistant.message`, `tool.call`, +`tool.result`, and `run.status`. Tool entries store bounded previews only; +raw reasoning/thinking is not persisted. + --- ## 5. Config diff --git a/docs/project-changelog.md b/docs/project-changelog.md index 93825551..5f049baf 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -6,6 +6,33 @@ Significant changes, features, and fixes in reverse chronological order. ## 2026-05-29 +### Archived run timeline (issue #76) + +Adds persisted run archive timeline entries for session detail review. + +**New** + +- `run_timeline_items` schema for PostgreSQL and SQLite/Lite, with tenant/run + sequence uniqueness and session/run indexes. +- Best-effort agent event recorder for `activity`, `assistant.message`, + `tool.call`, `tool.result`, and `run.status` entries. +- HTTP `GET /v1/runs/{runID}/timeline` and WS `run.timeline.get` read APIs, + including tenant/user visibility filtering. +- Web session detail timeline panel using the WS API and localized labels. + +**Security** + +- Tool arguments/results are persisted as bounded previews only. +- Raw thinking/reasoning is not persisted. +- Non-admin reads are filtered to the connected/effective user. + +**Tests** + +- Added store coverage for PG and SQLite, recorder tests, HTTP/WS API tests, + permission classification coverage, and UI display mapping tests. + +--- + ### Shell security group disabled state persistence (issue #75) **Fixes** diff --git a/internal/agent/loop_pipeline_tool_callbacks.go b/internal/agent/loop_pipeline_tool_callbacks.go index 765903df..4749ecfe 100644 --- a/internal/agent/loop_pipeline_tool_callbacks.go +++ b/internal/agent/loop_pipeline_tool_callbacks.go @@ -232,10 +232,16 @@ func (l *Loop) recordToolMetric(ctx context.Context, sessionKey, toolName string func makeToolEmitRun(l *Loop, req *RunRequest) func(AgentEvent) { return func(event AgentEvent) { event.RunKind = req.RunKind + event.DelegationID = req.DelegationID + event.TeamID = req.TeamID + event.TeamTaskID = req.TeamTaskID + event.ParentAgentID = req.ParentAgentID event.SessionKey = req.SessionKey event.SenderID = req.SenderID event.UserID = req.UserID event.Channel = req.Channel + event.ChatID = req.ChatID + event.TenantID = l.tenantID l.emit(event) } } diff --git a/internal/agent/run_timeline_recorder.go b/internal/agent/run_timeline_recorder.go new file mode 100644 index 00000000..b5140d92 --- /dev/null +++ b/internal/agent/run_timeline_recorder.go @@ -0,0 +1,334 @@ +package agent + +import ( + "context" + "encoding/json" + "log/slog" + "regexp" + "strings" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/internal/tools" + "github.com/nextlevelbuilder/goclaw/internal/tracing" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +const runTimelinePreviewLimit = 2000 + +// RunTimelineRecorder persists display-safe run events without blocking delivery. +type RunTimelineRecorder struct { + store store.RunTimelineStore + timeout time.Duration + + mu sync.Mutex + nextSeq map[string]int +} + +func NewRunTimelineRecorder(timelineStore store.RunTimelineStore) *RunTimelineRecorder { + return &RunTimelineRecorder{ + store: timelineStore, + timeout: 2 * time.Second, + nextSeq: make(map[string]int), + } +} + +func (r *RunTimelineRecorder) Record(event AgentEvent) { + if r == nil || r.store == nil { + return + } + if event.RunID == "" || event.SessionKey == "" || event.TenantID == uuid.Nil { + return + } + if _, _, ok := timelineKindForEvent(event); !ok { + return + } + seq := r.reserveSeq(event.RunID) + item, ok := runTimelineItemFromEvent(event, seq) + if !ok { + return + } + if isTerminalRunTimelineEvent(event.Type) { + defer r.forgetRun(event.RunID) + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), r.timeout) + defer cancel() + ctx = store.WithTenantID(ctx, item.TenantID) + if err := r.store.AppendRunTimelineItem(ctx, &item); err != nil { + slog.Warn("run_timeline.persist_failed", "run_id", event.RunID, "event", event.Type, "error", err) + } + }() +} + +func (r *RunTimelineRecorder) reserveSeq(runID string) int { + r.mu.Lock() + defer r.mu.Unlock() + r.nextSeq[runID]++ + return r.nextSeq[runID] +} + +func (r *RunTimelineRecorder) forgetRun(runID string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.nextSeq, runID) +} + +func isTerminalRunTimelineEvent(eventType string) bool { + switch eventType { + case protocol.AgentEventRunCompleted, protocol.AgentEventRunFailed, protocol.AgentEventRunCancelled: + return true + default: + return false + } +} + +func runTimelineItemFromEvent(event AgentEvent, seq int) (store.RunTimelineItem, bool) { + if event.RunID == "" || event.SessionKey == "" || event.TenantID == uuid.Nil { + return store.RunTimelineItem{}, false + } + itemType, status, ok := timelineKindForEvent(event) + if !ok { + return store.RunTimelineItem{}, false + } + metadata := timelineMetadata(event) + agentUUID, agentIsUUID := parseOptionalUUID(event.AgentID) + if event.AgentID != "" && !agentIsUUID { + metadata["agent_key"] = event.AgentID + } + traceID, spanID := payloadTraceIDs(event.Payload) + item := store.RunTimelineItem{ + TenantID: event.TenantID, + RunID: event.RunID, + SessionKey: event.SessionKey, + AgentID: agentUUID, + UserID: event.UserID, + Channel: event.Channel, + ChatID: event.ChatID, + Seq: seq, + ItemType: itemType, + Status: status, + Title: timelineTitle(event), + Preview: timelinePreview(event), + ToolName: payloadString(event.Payload, "name"), + ToolCallID: payloadString(event.Payload, "id"), + TraceID: traceID, + SpanID: spanID, + Metadata: mustJSON(metadata), + } + return item, true +} + +func timelineKindForEvent(event AgentEvent) (string, string, bool) { + switch event.Type { + case protocol.AgentEventRunStarted: + return store.RunTimelineItemTypeRunStatus, store.RunTimelineStatusStarted, true + case protocol.AgentEventRunCompleted: + return store.RunTimelineItemTypeRunStatus, store.RunTimelineStatusCompleted, true + case protocol.AgentEventRunFailed: + return store.RunTimelineItemTypeRunStatus, store.RunTimelineStatusFailed, true + case protocol.AgentEventRunCancelled: + return store.RunTimelineItemTypeRunStatus, store.RunTimelineStatusCancelled, true + case protocol.AgentEventActivity: + return store.RunTimelineItemTypeActivity, store.RunTimelineStatusRunning, true + case protocol.AgentEventBlockReply: + return store.RunTimelineItemTypeAssistantMessage, store.RunTimelineStatusCompleted, true + case protocol.AgentEventToolCall: + return store.RunTimelineItemTypeToolCall, store.RunTimelineStatusRunning, true + case protocol.AgentEventToolResult: + if payloadBool(event.Payload, "is_error") { + return store.RunTimelineItemTypeToolResult, store.RunTimelineStatusFailed, true + } + return store.RunTimelineItemTypeToolResult, store.RunTimelineStatusCompleted, true + default: + return "", "", false + } +} + +func timelineTitle(event AgentEvent) string { + if name := payloadString(event.Payload, "name"); name != "" { + return name + } + switch event.Type { + case protocol.AgentEventRunStarted: + return "Run started" + case protocol.AgentEventRunCompleted: + return "Run completed" + case protocol.AgentEventRunFailed: + return "Run failed" + case protocol.AgentEventRunCancelled: + return "Run cancelled" + case protocol.AgentEventBlockReply: + return "Assistant message" + case protocol.AgentEventActivity: + return "Activity" + default: + return event.Type + } +} + +func timelinePreview(event AgentEvent) string { + switch event.Type { + case protocol.AgentEventRunStarted: + return sanitizeTimelinePreview(payloadString(event.Payload, "message")) + case protocol.AgentEventRunCompleted, protocol.AgentEventBlockReply: + return sanitizeTimelinePreview(payloadString(event.Payload, "content")) + case protocol.AgentEventRunFailed: + return sanitizeTimelinePreview(payloadString(event.Payload, "error")) + case protocol.AgentEventActivity: + return sanitizeTimelinePreview(payloadAnyString(event.Payload)) + case protocol.AgentEventToolCall: + return sanitizeTimelinePreview(payloadJSON(event.Payload, "arguments")) + case protocol.AgentEventToolResult: + if result := payloadString(event.Payload, "result"); result != "" { + return sanitizeTimelinePreview(result) + } + return sanitizeTimelinePreview(payloadString(event.Payload, "content")) + default: + return "" + } +} + +func sanitizeTimelinePreview(value string) string { + value = strings.TrimSpace(stripThinkingTags(value)) + value = stripDeliveryFileTokens(value) + value = tools.ScrubCredentials(value) + return tracing.TruncateMid(value, runTimelinePreviewLimit) +} + +var deliveryFileTokenRe = regexp.MustCompile(`([?&])ft=[^)\]'"<>\s&]+`) + +func stripDeliveryFileTokens(value string) string { + if !strings.Contains(value, "ft=") { + return value + } + value = deliveryFileTokenRe.ReplaceAllStringFunc(value, func(match string) string { + if strings.HasPrefix(match, "&") { + return "" + } + return "?" + }) + value = strings.ReplaceAll(value, "?&", "?") + value = strings.ReplaceAll(value, "?)", ")") + value = strings.ReplaceAll(value, "?]", "]") + value = strings.TrimSuffix(value, "?") + return strings.TrimSuffix(value, "&") +} + +func timelineMetadata(event AgentEvent) map[string]any { + metadata := map[string]any{"event_type": event.Type} + if event.RunKind != "" { + metadata["run_kind"] = event.RunKind + } + if event.DelegationID != "" { + metadata["delegation_id"] = event.DelegationID + } + if event.TeamID != "" { + metadata["team_id"] = event.TeamID + } + if event.TeamTaskID != "" { + metadata["team_task_id"] = event.TeamTaskID + } + if event.ParentAgentID != "" { + metadata["parent_agent_id"] = event.ParentAgentID + } + if event.SenderID != "" { + metadata["sender_id"] = event.SenderID + } + if payloadBool(event.Payload, "is_error") { + metadata["is_error"] = true + } + return metadata +} + +func payloadString(payload any, key string) string { + switch m := payload.(type) { + case map[string]string: + return m[key] + case map[string]any: + if v, ok := m[key].(string); ok { + return v + } + } + return "" +} + +func payloadBool(payload any, key string) bool { + if m, ok := payload.(map[string]any); ok { + if v, ok := m[key].(bool); ok { + return v + } + } + return false +} + +func payloadAnyString(payload any) string { + switch v := payload.(type) { + case string: + return v + case map[string]string: + if c := v["content"]; c != "" { + return c + } + if m := v["message"]; m != "" { + return m + } + case map[string]any: + for _, key := range []string{"content", "message", "status", "step"} { + if s, ok := v[key].(string); ok && s != "" { + return s + } + } + } + raw, _ := json.Marshal(payload) + return string(raw) +} + +func payloadJSON(payload any, key string) string { + if m, ok := payload.(map[string]any); ok { + if v, has := m[key]; has { + raw, _ := json.Marshal(v) + return string(raw) + } + } + return "" +} + +func payloadTraceIDs(payload any) (*uuid.UUID, *uuid.UUID) { + traceID := parsePayloadUUID(payload, "trace_id", "traceId") + spanID := parsePayloadUUID(payload, "span_id", "spanId") + return traceID, spanID +} + +func parsePayloadUUID(payload any, keys ...string) *uuid.UUID { + for _, key := range keys { + if v := payloadString(payload, key); v != "" { + if parsed, err := uuid.Parse(v); err == nil { + return &parsed + } + } + } + return nil +} + +func parseOptionalUUID(value string) (*uuid.UUID, bool) { + if value == "" { + return nil, false + } + parsed, err := uuid.Parse(value) + if err != nil { + return nil, false + } + return &parsed, true +} + +func mustJSON(v any) json.RawMessage { + raw, err := json.Marshal(v) + if err != nil { + return json.RawMessage(`{}`) + } + return raw +} diff --git a/internal/agent/run_timeline_recorder_test.go b/internal/agent/run_timeline_recorder_test.go new file mode 100644 index 00000000..2d2080b9 --- /dev/null +++ b/internal/agent/run_timeline_recorder_test.go @@ -0,0 +1,171 @@ +package agent + +import ( + "context" + "strings" + "sync" + "testing" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +type recordingTimelineStore struct { + mu sync.Mutex + items []store.RunTimelineItem +} + +func (s *recordingTimelineStore) AppendRunTimelineItem(_ context.Context, item *store.RunTimelineItem) error { + s.mu.Lock() + defer s.mu.Unlock() + s.items = append(s.items, *item) + return nil +} + +func (s *recordingTimelineStore) ListRunTimelineItems(context.Context, store.RunTimelineListOpts) ([]store.RunTimelineItem, error) { + return nil, nil +} + +func TestRunTimelineItemFromEventScrubsToolArguments(t *testing.T) { + tenantID := uuid.Must(uuid.NewV7()) + item, ok := runTimelineItemFromEvent(AgentEvent{ + Type: protocol.AgentEventToolCall, + AgentID: "default", + RunID: "run-1", + UserID: "user-1", + Channel: "web", + ChatID: "chat-1", + SessionKey: "session-1", + TenantID: tenantID, + Payload: map[string]any{ + "name": "exec_command", + "id": "call-1", + "arguments": map[string]any{"cmd": "echo sk-abcdefghijklmnopqrstuvwxyz123456"}, + }, + }, 7) + if !ok { + t.Fatal("expected timeline item") + } + if item.ItemType != store.RunTimelineItemTypeToolCall { + t.Fatalf("ItemType = %q", item.ItemType) + } + if item.Seq != 7 { + t.Fatalf("Seq = %d, want 7", item.Seq) + } + if item.ToolName != "exec_command" || item.ToolCallID != "call-1" { + t.Fatalf("tool fields = %q/%q", item.ToolName, item.ToolCallID) + } + if strings.Contains(item.Preview, "sk-abcdefghijklmnopqrstuvwxyz123456") { + t.Fatalf("preview leaked secret: %s", item.Preview) + } + if !strings.Contains(item.Preview, "[REDACTED]") { + t.Fatalf("preview missing redaction: %s", item.Preview) + } + if item.AgentID != nil { + t.Fatalf("AgentID = %v, want nil for agent key", item.AgentID) + } + if !strings.Contains(string(item.Metadata), `"agent_key":"default"`) { + t.Fatalf("metadata missing agent_key: %s", item.Metadata) + } +} + +func TestRunTimelineItemFromEventDropsUnsupportedAndThinking(t *testing.T) { + tenantID := uuid.Must(uuid.NewV7()) + if _, ok := runTimelineItemFromEvent(AgentEvent{ + Type: protocol.ChatEventThinking, + RunID: "run-1", + SessionKey: "session-1", + TenantID: tenantID, + }, 1); ok { + t.Fatal("thinking event should not be archived") + } + + item, ok := runTimelineItemFromEvent(AgentEvent{ + Type: protocol.AgentEventRunCompleted, + RunID: "run-1", + SessionKey: "session-1", + TenantID: tenantID, + Payload: map[string]any{ + "content": "visible hidden chain done", + "thinking": "raw hidden chain", + }, + }, 1) + if !ok { + t.Fatal("expected completed item") + } + if strings.Contains(item.Preview, "hidden chain") || strings.Contains(item.Preview, "raw hidden") { + t.Fatalf("preview leaked thinking: %q", item.Preview) + } + if item.Preview != "visible done" { + t.Fatalf("Preview = %q", item.Preview) + } +} + +func TestRunTimelinePreviewStripsDeliveryFileTokens(t *testing.T) { + tenantID := uuid.Must(uuid.NewV7()) + item, ok := runTimelineItemFromEvent(AgentEvent{ + Type: protocol.AgentEventRunCompleted, + RunID: "run-1", + SessionKey: "session-1", + TenantID: tenantID, + Payload: map[string]any{ + "content": "See ![a](/v1/files/work/a.png?ft=signed.123) and /v1/media/b.txt?x=1&ft=stale.456", + }, + }, 1) + if !ok { + t.Fatal("expected completed item") + } + if strings.Contains(item.Preview, "ft=") || strings.Contains(item.Preview, "signed.123") || strings.Contains(item.Preview, "stale.456") { + t.Fatalf("preview leaked delivery token: %q", item.Preview) + } + if !strings.Contains(item.Preview, "/v1/files/work/a.png") || !strings.Contains(item.Preview, "/v1/media/b.txt?x=1") { + t.Fatalf("preview lost clean file URLs: %q", item.Preview) + } +} + +func TestRunTimelineRecorderOnlyTracksSupportedActiveRuns(t *testing.T) { + tenantID := uuid.Must(uuid.NewV7()) + recorder := NewRunTimelineRecorder(&recordingTimelineStore{}) + base := AgentEvent{ + RunID: "run-1", + SessionKey: "session-1", + TenantID: tenantID, + } + + recorder.Record(AgentEvent{Type: protocol.ChatEventThinking, RunID: base.RunID, SessionKey: base.SessionKey, TenantID: base.TenantID}) + if got := recorderTrackedRuns(recorder); got != 0 { + t.Fatalf("tracked runs after unsupported event = %d, want 0", got) + } + + recorder.Record(AgentEvent{Type: protocol.AgentEventToolCall, RunID: base.RunID, SessionKey: base.SessionKey, TenantID: base.TenantID}) + if got := recorderTrackedRuns(recorder); got != 1 { + t.Fatalf("tracked runs after supported event = %d, want 1", got) + } + if seq := recorderSeq(recorder, base.RunID); seq != 1 { + t.Fatalf("seq after first supported event = %d, want 1", seq) + } + + recorder.Record(AgentEvent{Type: protocol.ChatEventThinking, RunID: base.RunID, SessionKey: base.SessionKey, TenantID: base.TenantID}) + if seq := recorderSeq(recorder, base.RunID); seq != 1 { + t.Fatalf("seq after unsupported event = %d, want 1", seq) + } + + recorder.Record(AgentEvent{Type: protocol.AgentEventRunCompleted, RunID: base.RunID, SessionKey: base.SessionKey, TenantID: base.TenantID}) + if got := recorderTrackedRuns(recorder); got != 0 { + t.Fatalf("tracked runs after terminal event = %d, want 0", got) + } +} + +func recorderTrackedRuns(r *RunTimelineRecorder) int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.nextSeq) +} + +func recorderSeq(r *RunTimelineRecorder, runID string) int { + r.mu.Lock() + defer r.mu.Unlock() + return r.nextSeq[runID] +} diff --git a/internal/gateway/methods/run_timeline.go b/internal/gateway/methods/run_timeline.go new file mode 100644 index 00000000..7eb3c241 --- /dev/null +++ b/internal/gateway/methods/run_timeline.go @@ -0,0 +1,94 @@ +package methods + +import ( + "context" + "encoding/json" + "log/slog" + + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/gateway" + "github.com/nextlevelbuilder/goclaw/internal/i18n" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +// RunTimelineMethods handles archived run timeline reads. +type RunTimelineMethods struct { + timeline store.RunTimelineStore + cfg *config.Config +} + +func NewRunTimelineMethods(timeline store.RunTimelineStore, cfg *config.Config) *RunTimelineMethods { + return &RunTimelineMethods{timeline: timeline, cfg: cfg} +} + +func (m *RunTimelineMethods) Register(router *gateway.MethodRouter) { + router.Register(protocol.MethodRunTimelineGet, m.handleGet) +} + +type runTimelineGetParams struct { + RunID string `json:"runId"` + SessionKey string `json:"sessionKey"` + Limit int `json:"limit"` + Offset int `json:"offset"` +} + +func (m *RunTimelineMethods) handleGet(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) { + locale := store.LocaleFromContext(ctx) + if m.timeline == nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRunTimelineUnavailable))) + return + } + var params runTimelineGetParams + if req.Params != nil { + if err := json.Unmarshal(req.Params, ¶ms); err != nil { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidJSON))) + return + } + } + if params.RunID == "" && params.SessionKey == "" { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgRequired, "runId or sessionKey"))) + return + } + if params.Limit <= 0 || params.Limit > 500 { + params.Limit = 200 + } + if params.Offset < 0 { + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, i18n.T(locale, i18n.MsgInvalidRequest, "offset must be non-negative"))) + return + } + items, err := m.timeline.ListRunTimelineItems(ctx, store.RunTimelineListOpts{ + RunID: params.RunID, + SessionKey: params.SessionKey, + Limit: params.Limit, + Offset: params.Offset, + }) + if err != nil { + slog.Warn("run_timeline.get_failed", "run_id", params.RunID, "session_key", params.SessionKey, "error", err) + client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInternal, i18n.T(locale, i18n.MsgInternalError, "run timeline"))) + return + } + if !canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID()) { + items = filterRunTimelineItemsByUser(items, client.UserID()) + } + client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{ + "runId": params.RunID, + "sessionKey": params.SessionKey, + "items": items, + "limit": params.Limit, + "offset": params.Offset, + })) +} + +func filterRunTimelineItemsByUser(items []store.RunTimelineItem, userID string) []store.RunTimelineItem { + if userID == "" { + return nil + } + filtered := items[:0] + for _, item := range items { + if item.UserID == userID { + filtered = append(filtered, item) + } + } + return filtered +} diff --git a/internal/gateway/methods/run_timeline_test.go b/internal/gateway/methods/run_timeline_test.go new file mode 100644 index 00000000..ba4d4d07 --- /dev/null +++ b/internal/gateway/methods/run_timeline_test.go @@ -0,0 +1,115 @@ +package methods + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/config" + "github.com/nextlevelbuilder/goclaw/internal/gateway" + "github.com/nextlevelbuilder/goclaw/internal/permissions" + "github.com/nextlevelbuilder/goclaw/internal/store" + "github.com/nextlevelbuilder/goclaw/pkg/protocol" +) + +type stubRunTimelineStore struct { + opts []store.RunTimelineListOpts + items []store.RunTimelineItem + err error +} + +func (s *stubRunTimelineStore) AppendRunTimelineItem(context.Context, *store.RunTimelineItem) error { + return nil +} + +func (s *stubRunTimelineStore) ListRunTimelineItems(_ context.Context, opts store.RunTimelineListOpts) ([]store.RunTimelineItem, error) { + s.opts = append(s.opts, opts) + if s.err != nil { + return nil, s.err + } + return s.items, nil +} + +func TestRunTimelineGetScopesViewerByUser(t *testing.T) { + timeline := &stubRunTimelineStore{ + items: []store.RunTimelineItem{ + {RunID: "run-1", UserID: "caller", Seq: 1, Preview: "visible"}, + {RunID: "run-1", UserID: "other", Seq: 2, Preview: "hidden"}, + }, + } + m := NewRunTimelineMethods(timeline, &config.Config{}) + tenantID := uuid.Must(uuid.NewV7()) + client, responses := gateway.NewCapturingTestClient(permissions.RoleViewer, tenantID, "caller", 1) + ctx := store.WithTenantID(context.Background(), tenantID) + m.handleGet(ctx, client, sessionReqFrame(t, protocol.MethodRunTimelineGet, map[string]any{"runId": "run-1"})) + + resp := readTimelineResponse(t, responses) + if resp.Error != nil { + t.Fatalf("unexpected error: %+v", resp.Error) + } + data, ok := resp.Payload.(map[string]any) + if !ok { + t.Fatalf("payload type = %T", resp.Payload) + } + rawItems, ok := data["items"].([]any) + if !ok { + t.Fatalf("items type = %T", data["items"]) + } + if len(rawItems) != 1 { + t.Fatalf("len(items) = %d, want 1", len(rawItems)) + } + if timeline.opts[0].RunID != "run-1" { + t.Fatalf("RunID = %q", timeline.opts[0].RunID) + } +} + +func TestRunTimelineGetRejectsNegativeOffset(t *testing.T) { + timeline := &stubRunTimelineStore{} + m := NewRunTimelineMethods(timeline, &config.Config{}) + tenantID := uuid.Must(uuid.NewV7()) + client, responses := gateway.NewCapturingTestClient(permissions.RoleViewer, tenantID, "caller", 1) + ctx := store.WithTenantID(context.Background(), tenantID) + m.handleGet(ctx, client, sessionReqFrame(t, protocol.MethodRunTimelineGet, map[string]any{ + "runId": "run-1", + "offset": -1, + })) + + resp := readTimelineResponse(t, responses) + if resp.Error == nil || resp.Error.Code != protocol.ErrInvalidRequest { + t.Fatalf("error = %+v, want INVALID_REQUEST", resp.Error) + } + if len(timeline.opts) != 0 { + t.Fatalf("store called with opts: %+v", timeline.opts) + } +} + +func TestRunTimelineGetHidesStoreErrorDetail(t *testing.T) { + timeline := &stubRunTimelineStore{err: errors.New("pq: syntax error at or near \"OFFSET -1\"")} + m := NewRunTimelineMethods(timeline, &config.Config{}) + tenantID := uuid.Must(uuid.NewV7()) + client, responses := gateway.NewCapturingTestClient(permissions.RoleViewer, tenantID, "caller", 1) + ctx := store.WithTenantID(context.Background(), tenantID) + m.handleGet(ctx, client, sessionReqFrame(t, protocol.MethodRunTimelineGet, map[string]any{"runId": "run-1"})) + + resp := readTimelineResponse(t, responses) + if resp.Error == nil || resp.Error.Code != protocol.ErrInternal { + t.Fatalf("error = %+v, want INTERNAL_ERROR", resp.Error) + } + if strings.Contains(resp.Error.Message, "OFFSET -1") || strings.Contains(resp.Error.Message, "pq:") { + t.Fatalf("error leaked store detail: %q", resp.Error.Message) + } +} + +func readTimelineResponse(t *testing.T, ch <-chan []byte) protocol.ResponseFrame { + t.Helper() + raw := <-ch + var resp protocol.ResponseFrame + if err := json.Unmarshal(raw, &resp); err != nil { + t.Fatalf("decode response: %v", err) + } + return resp +} diff --git a/internal/http/run_timeline_test.go b/internal/http/run_timeline_test.go new file mode 100644 index 00000000..e9138704 --- /dev/null +++ b/internal/http/run_timeline_test.go @@ -0,0 +1,85 @@ +package http + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + "github.com/nextlevelbuilder/goclaw/internal/crypto" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +type stubRunTimelineStore struct { + opts []store.RunTimelineListOpts + items []store.RunTimelineItem +} + +func (s *stubRunTimelineStore) AppendRunTimelineItem(context.Context, *store.RunTimelineItem) error { + return nil +} + +func (s *stubRunTimelineStore) ListRunTimelineItems(_ context.Context, opts store.RunTimelineListOpts) ([]store.RunTimelineItem, error) { + s.opts = append(s.opts, opts) + return s.items, nil +} + +func TestRunTimelineHTTPScopesViewerByUser(t *testing.T) { + token := setupTraceReadToken(t, "caller") + timeline := &stubRunTimelineStore{ + items: []store.RunTimelineItem{ + {RunID: "run-1", UserID: "caller", Seq: 1, Preview: "visible"}, + {RunID: "run-1", UserID: "other", Seq: 2, Preview: "hidden"}, + }, + } + mux := http.NewServeMux() + NewTracesHandler(&mockTracingStore{}, timeline).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/runs/run-1/timeline?session_key=session-1&limit=20", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } + var body struct { + Items []store.RunTimelineItem `json:"items"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode body: %v", err) + } + if len(body.Items) != 1 || body.Items[0].UserID != "caller" { + t.Fatalf("items = %+v", body.Items) + } + if len(timeline.opts) != 1 { + t.Fatalf("List calls = %d, want 1", len(timeline.opts)) + } + if timeline.opts[0].RunID != "run-1" || timeline.opts[0].SessionKey != "session-1" { + t.Fatalf("opts = %+v", timeline.opts[0]) + } +} + +func TestRunTimelineHTTPAdminSeesTenantItems(t *testing.T) { + token := "timeline-admin-key" + setupTestCache(t, map[string]*store.APIKeyData{ + crypto.HashAPIKey(token): {ID: uuid.New(), Scopes: []string{"operator.admin"}, OwnerID: "admin"}, + }) + timeline := &stubRunTimelineStore{ + items: []store.RunTimelineItem{{RunID: "run-1", UserID: "other", Seq: 1}}, + } + mux := http.NewServeMux() + NewTracesHandler(&mockTracingStore{}, timeline).RegisterRoutes(mux) + + req := httptest.NewRequest(http.MethodGet, "/v1/runs/run-1/timeline", nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + mux.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/http/traces.go b/internal/http/traces.go index dcde4417..a1a27146 100644 --- a/internal/http/traces.go +++ b/internal/http/traces.go @@ -8,6 +8,7 @@ import ( "log/slog" "net/http" "strconv" + "strings" "time" "github.com/google/uuid" @@ -19,12 +20,17 @@ import ( // TracesHandler handles LLM trace listing and detail endpoints. type TracesHandler struct { - tracing store.TracingStore + tracing store.TracingStore + runTimeline store.RunTimelineStore } // NewTracesHandler creates a handler for trace management endpoints. -func NewTracesHandler(tracing store.TracingStore) *TracesHandler { - return &TracesHandler{tracing: tracing} +func NewTracesHandler(tracing store.TracingStore, timelines ...store.RunTimelineStore) *TracesHandler { + h := &TracesHandler{tracing: tracing} + if len(timelines) > 0 { + h.runTimeline = timelines[0] + } + return h } // RegisterRoutes registers trace routes on the given mux. @@ -33,6 +39,7 @@ func (h *TracesHandler) RegisterRoutes(mux *http.ServeMux) { mux.HandleFunc("GET /v1/traces/follow", h.authMiddleware(h.handleFollow)) mux.HandleFunc("GET /v1/traces/{traceID}/export", h.authMiddleware(h.handleExport)) mux.HandleFunc("GET /v1/traces/{traceID}", h.authMiddleware(h.handleGet)) + mux.HandleFunc("GET /v1/runs/{runID}/timeline", h.authMiddleware(h.handleRunTimeline)) mux.HandleFunc("GET /v1/costs/summary", h.authMiddleware(h.handleCostSummary)) } @@ -256,6 +263,67 @@ func (h *TracesHandler) handleCostSummary(w http.ResponseWriter, r *http.Request writeJSON(w, http.StatusOK, map[string]any{"rows": rows}) } +func (h *TracesHandler) handleRunTimeline(w http.ResponseWriter, r *http.Request) { + locale := store.LocaleFromContext(r.Context()) + if h.runTimeline == nil { + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": i18n.T(locale, i18n.MsgRunTimelineUnavailable)}) + return + } + runID := r.PathValue("runID") + if strings.TrimSpace(runID) == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": i18n.T(locale, i18n.MsgRequired, "run_id")}) + return + } + opts := store.RunTimelineListOpts{ + RunID: runID, + SessionKey: r.URL.Query().Get("session_key"), + Limit: 200, + } + if v := r.URL.Query().Get("limit"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n > 0 { + if n > 500 { + n = 500 + } + opts.Limit = n + } + } + if v := r.URL.Query().Get("offset"); v != "" { + if n, err := strconv.Atoi(v); err == nil && n >= 0 { + opts.Offset = n + } + } + items, err := h.runTimeline.ListRunTimelineItems(r.Context(), opts) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + auth := resolveAuth(r) + if !permissions.HasMinRole(auth.Role, permissions.RoleAdmin) { + callerID := store.UserIDFromContext(r.Context()) + items = filterTimelineItemsByUser(items, callerID) + } + writeJSON(w, http.StatusOK, map[string]any{ + "run_id": runID, + "session_key": opts.SessionKey, + "items": items, + "limit": opts.Limit, + "offset": opts.Offset, + }) +} + +func filterTimelineItemsByUser(items []store.RunTimelineItem, userID string) []store.RunTimelineItem { + if userID == "" { + return nil + } + filtered := items[:0] + for _, item := range items { + if item.UserID == userID { + filtered = append(filtered, item) + } + } + return filtered +} + // traceExportEntry is a trace with its spans and recursive sub-traces. type traceExportEntry struct { Trace store.TraceData `json:"trace"` diff --git a/internal/i18n/catalog_en.go b/internal/i18n/catalog_en.go index 1bbcfb60..60943dff 100644 --- a/internal/i18n/catalog_en.go +++ b/internal/i18n/catalog_en.go @@ -59,23 +59,24 @@ func init() { MsgSenderIDRequired: "sender_id is required", // HTTP API - MsgInvalidAuth: "invalid authentication", - MsgMsgsRequired: "messages is required", - MsgUserIDHeader: "X-GoClaw-User-Id header is required", - MsgFileTooLarge: "file too large or invalid multipart form", - MsgMissingFileField: "missing 'file' field", - MsgInvalidFilename: "invalid filename", - MsgChannelKeyReq: "channel and key are required", - MsgMethodNotAllowed: "method not allowed", - MsgStreamingNotSupported: "streaming not supported", - MsgOwnerOnly: "only owner can %s", - MsgNoAccess: "no access to this %s", - MsgAlreadySummoning: "agent is already being summoned", - MsgSummoningUnavailable: "summoning not available", - MsgNoDescription: "agent has no description to resummon from", - MsgSummonCancelled: "summon cancelled by user", - MsgCannotCancel: "agent is not being summoned", - MsgInvalidPath: "invalid path", + MsgInvalidAuth: "invalid authentication", + MsgMsgsRequired: "messages is required", + MsgUserIDHeader: "X-GoClaw-User-Id header is required", + MsgFileTooLarge: "file too large or invalid multipart form", + MsgMissingFileField: "missing 'file' field", + MsgInvalidFilename: "invalid filename", + MsgChannelKeyReq: "channel and key are required", + MsgMethodNotAllowed: "method not allowed", + MsgStreamingNotSupported: "streaming not supported", + MsgOwnerOnly: "only owner can %s", + MsgNoAccess: "no access to this %s", + MsgAlreadySummoning: "agent is already being summoned", + MsgSummoningUnavailable: "summoning not available", + MsgRunTimelineUnavailable: "run timeline not available", + MsgNoDescription: "agent has no description to resummon from", + MsgSummonCancelled: "summon cancelled by user", + MsgCannotCancel: "agent is not being summoned", + MsgInvalidPath: "invalid path", // Browser cookies MsgBrowserCookieTooMany: "too many browser cookies in one sync request", @@ -308,11 +309,11 @@ func init() { MsgGrantEnvRevealLimit: "rate limit exceeded for env reveal — try again later", // Git credential adapter - MsgGitCredHostMismatch: "stored git credential is for %s but command targets %s", - MsgGitCredNoMatch: "no git credential configured for host %s", - MsgGitCredUnsupportedType: "git credential type %q is not supported", - MsgGitCredTokenInvalid: "stored git token is invalid or empty", - MsgGitCredTokenControlChar: "stored git token contains forbidden control characters", + MsgGitCredHostMismatch: "stored git credential is for %s but command targets %s", + MsgGitCredNoMatch: "no git credential configured for host %s", + MsgGitCredUnsupportedType: "git credential type %q is not supported", + MsgGitCredTokenInvalid: "stored git token is invalid or empty", + MsgGitCredTokenControlChar: "stored git token contains forbidden control characters", MsgGitCredHostUserinfoRejected: "git URL with embedded userinfo is rejected as ambiguous", MsgGitCredSSHPassphraseUnsupported: "passphrase-protected SSH keys are not supported; remove the passphrase with `ssh-keygen -p` before saving", MsgGitCredSSHKeyInvalid: "SSH private key invalid: %s", diff --git a/internal/i18n/catalog_vi.go b/internal/i18n/catalog_vi.go index 78b31cd1..df769c49 100644 --- a/internal/i18n/catalog_vi.go +++ b/internal/i18n/catalog_vi.go @@ -59,23 +59,24 @@ func init() { MsgSenderIDRequired: "sender_id là bắt buộc", // HTTP API - MsgInvalidAuth: "xác thực không hợp lệ", - MsgMsgsRequired: "messages là bắt buộc", - MsgUserIDHeader: "header X-GoClaw-User-Id là bắt buộc", - MsgFileTooLarge: "tệp quá lớn hoặc form multipart không hợp lệ", - MsgMissingFileField: "thiếu trường 'file'", - MsgInvalidFilename: "tên tệp không hợp lệ", - MsgChannelKeyReq: "channel và key là bắt buộc", - MsgMethodNotAllowed: "phương thức không được phép", - MsgStreamingNotSupported: "streaming không được hỗ trợ", - MsgOwnerOnly: "chỉ chủ sở hữu mới có thể %s", - MsgNoAccess: "không có quyền truy cập %s này", - MsgAlreadySummoning: "agent đang được triệu hồi", - MsgSummoningUnavailable: "triệu hồi không khả dụng", - MsgNoDescription: "agent không có mô tả để triệu hồi lại", - MsgSummonCancelled: "đã huỷ triệu hồi", - MsgCannotCancel: "agent không trong trạng thái đang triệu hồi", - MsgInvalidPath: "đường dẫn không hợp lệ", + MsgInvalidAuth: "xác thực không hợp lệ", + MsgMsgsRequired: "messages là bắt buộc", + MsgUserIDHeader: "header X-GoClaw-User-Id là bắt buộc", + MsgFileTooLarge: "tệp quá lớn hoặc form multipart không hợp lệ", + MsgMissingFileField: "thiếu trường 'file'", + MsgInvalidFilename: "tên tệp không hợp lệ", + MsgChannelKeyReq: "channel và key là bắt buộc", + MsgMethodNotAllowed: "phương thức không được phép", + MsgStreamingNotSupported: "streaming không được hỗ trợ", + MsgOwnerOnly: "chỉ chủ sở hữu mới có thể %s", + MsgNoAccess: "không có quyền truy cập %s này", + MsgAlreadySummoning: "agent đang được triệu hồi", + MsgSummoningUnavailable: "triệu hồi không khả dụng", + MsgRunTimelineUnavailable: "timeline lượt chạy không khả dụng", + MsgNoDescription: "agent không có mô tả để triệu hồi lại", + MsgSummonCancelled: "đã huỷ triệu hồi", + MsgCannotCancel: "agent không trong trạng thái đang triệu hồi", + MsgInvalidPath: "đường dẫn không hợp lệ", // Browser cookies MsgBrowserCookieTooMany: "quá nhiều cookie trình duyệt trong một yêu cầu đồng bộ", @@ -308,11 +309,11 @@ func init() { MsgGrantEnvRevealLimit: "đã vượt giới hạn yêu cầu xem env — vui lòng thử lại sau", // Git credential adapter - MsgGitCredHostMismatch: "thông tin xác thực git đã lưu dành cho %s nhưng lệnh đang nhắm tới %s", - MsgGitCredNoMatch: "không có thông tin xác thực git cho host %s", - MsgGitCredUnsupportedType: "loại thông tin xác thực git %q không được hỗ trợ", - MsgGitCredTokenInvalid: "token git đã lưu không hợp lệ hoặc rỗng", - MsgGitCredTokenControlChar: "token git đã lưu chứa ký tự điều khiển bị cấm", + MsgGitCredHostMismatch: "thông tin xác thực git đã lưu dành cho %s nhưng lệnh đang nhắm tới %s", + MsgGitCredNoMatch: "không có thông tin xác thực git cho host %s", + MsgGitCredUnsupportedType: "loại thông tin xác thực git %q không được hỗ trợ", + MsgGitCredTokenInvalid: "token git đã lưu không hợp lệ hoặc rỗng", + MsgGitCredTokenControlChar: "token git đã lưu chứa ký tự điều khiển bị cấm", MsgGitCredHostUserinfoRejected: "URL git chứa userinfo nhúng bị từ chối vì gây nhập nhằng", MsgGitCredSSHPassphraseUnsupported: "khóa SSH có passphrase chưa được hỗ trợ; chạy `ssh-keygen -p` để gỡ passphrase trước khi lưu", MsgGitCredSSHKeyInvalid: "khóa riêng SSH không hợp lệ: %s", diff --git a/internal/i18n/catalog_zh.go b/internal/i18n/catalog_zh.go index 2fddc6af..2053cfa9 100644 --- a/internal/i18n/catalog_zh.go +++ b/internal/i18n/catalog_zh.go @@ -59,23 +59,24 @@ func init() { MsgSenderIDRequired: "sender_id 是必填项", // HTTP API - MsgInvalidAuth: "身份验证无效", - MsgMsgsRequired: "messages 是必填项", - MsgUserIDHeader: "需要 X-GoClaw-User-Id 请求头", - MsgFileTooLarge: "文件过大或 multipart 表单无效", - MsgMissingFileField: "缺少 'file' 字段", - MsgInvalidFilename: "文件名无效", - MsgChannelKeyReq: "channel 和 key 是必填项", - MsgMethodNotAllowed: "不允许的请求方法", - MsgStreamingNotSupported: "不支持流式传输", - MsgOwnerOnly: "只有所有者才能%s", - MsgNoAccess: "无权访问此%s", - MsgAlreadySummoning: "Agent正在被召唤中", - MsgSummoningUnavailable: "召唤功能不可用", - MsgNoDescription: "Agent没有可供重新召唤的描述", - MsgSummonCancelled: "已取消召唤", - MsgCannotCancel: "Agent 未处于召唤状态", - MsgInvalidPath: "路径无效", + MsgInvalidAuth: "身份验证无效", + MsgMsgsRequired: "messages 是必填项", + MsgUserIDHeader: "需要 X-GoClaw-User-Id 请求头", + MsgFileTooLarge: "文件过大或 multipart 表单无效", + MsgMissingFileField: "缺少 'file' 字段", + MsgInvalidFilename: "文件名无效", + MsgChannelKeyReq: "channel 和 key 是必填项", + MsgMethodNotAllowed: "不允许的请求方法", + MsgStreamingNotSupported: "不支持流式传输", + MsgOwnerOnly: "只有所有者才能%s", + MsgNoAccess: "无权访问此%s", + MsgAlreadySummoning: "Agent正在被召唤中", + MsgSummoningUnavailable: "召唤功能不可用", + MsgRunTimelineUnavailable: "运行时间线不可用", + MsgNoDescription: "Agent没有可供重新召唤的描述", + MsgSummonCancelled: "已取消召唤", + MsgCannotCancel: "Agent 未处于召唤状态", + MsgInvalidPath: "路径无效", // Browser cookies MsgBrowserCookieTooMany: "单次同步请求中的浏览器 Cookie 过多", @@ -308,11 +309,11 @@ func init() { MsgGrantEnvRevealLimit: "env 查看请求超出速率限制，请稍后再试", // Git 凭据适配器 - MsgGitCredHostMismatch: "已存储的 git 凭据属于 %s，但命令目标是 %s", - MsgGitCredNoMatch: "没有为主机 %s 配置 git 凭据", - MsgGitCredUnsupportedType: "不支持的 git 凭据类型 %q", - MsgGitCredTokenInvalid: "已存储的 git token 无效或为空", - MsgGitCredTokenControlChar: "已存储的 git token 包含被禁止的控制字符", + MsgGitCredHostMismatch: "已存储的 git 凭据属于 %s，但命令目标是 %s", + MsgGitCredNoMatch: "没有为主机 %s 配置 git 凭据", + MsgGitCredUnsupportedType: "不支持的 git 凭据类型 %q", + MsgGitCredTokenInvalid: "已存储的 git token 无效或为空", + MsgGitCredTokenControlChar: "已存储的 git token 包含被禁止的控制字符", MsgGitCredHostUserinfoRejected: "git URL 中嵌入的 userinfo 因含义不明被拒绝", MsgGitCredSSHPassphraseUnsupported: "暂不支持带密码短语的 SSH 私钥；请先用 `ssh-keygen -p` 移除密码短语再保存", MsgGitCredSSHKeyInvalid: "SSH 私钥无效:%s", diff --git a/internal/i18n/keys.go b/internal/i18n/keys.go index 9f93a239..4516d15f 100644 --- a/internal/i18n/keys.go +++ b/internal/i18n/keys.go @@ -60,23 +60,24 @@ const ( MsgSenderIDRequired = "error.sender_id_required" // "sender_id is required" // --- HTTP API --- - MsgInvalidAuth = "error.invalid_auth" // "invalid authentication" - MsgMsgsRequired = "error.messages_required" // "messages is required" - MsgUserIDHeader = "error.user_id_header" // "X-GoClaw-User-Id header is required" - MsgFileTooLarge = "error.file_too_large" // "file too large or invalid multipart form" - MsgMissingFileField = "error.missing_file_field" // "missing 'file' field" - MsgInvalidFilename = "error.invalid_filename" // "invalid filename" - MsgChannelKeyReq = "error.channel_key_required" // "channel and key are required" - MsgMethodNotAllowed = "error.method_not_allowed" // "method not allowed" - MsgStreamingNotSupported = "error.streaming_not_supported" // "streaming not supported" - MsgOwnerOnly = "error.owner_only" // "only owner can %s" - MsgNoAccess = "error.no_access" // "no access to this %s" - MsgAlreadySummoning = "error.already_summoning" // "agent is already being summoned" - MsgSummoningUnavailable = "error.summoning_unavailable" // "summoning not available" - MsgNoDescription = "error.no_description" // "agent has no description to resummon from" - MsgSummonCancelled = "info.summon_cancelled" // "summon cancelled by user" - MsgCannotCancel = "error.cannot_cancel_summon" // "agent is not being summoned" - MsgInvalidPath = "error.invalid_path" // "invalid path" + MsgInvalidAuth = "error.invalid_auth" // "invalid authentication" + MsgMsgsRequired = "error.messages_required" // "messages is required" + MsgUserIDHeader = "error.user_id_header" // "X-GoClaw-User-Id header is required" + MsgFileTooLarge = "error.file_too_large" // "file too large or invalid multipart form" + MsgMissingFileField = "error.missing_file_field" // "missing 'file' field" + MsgInvalidFilename = "error.invalid_filename" // "invalid filename" + MsgChannelKeyReq = "error.channel_key_required" // "channel and key are required" + MsgMethodNotAllowed = "error.method_not_allowed" // "method not allowed" + MsgStreamingNotSupported = "error.streaming_not_supported" // "streaming not supported" + MsgOwnerOnly = "error.owner_only" // "only owner can %s" + MsgNoAccess = "error.no_access" // "no access to this %s" + MsgAlreadySummoning = "error.already_summoning" // "agent is already being summoned" + MsgSummoningUnavailable = "error.summoning_unavailable" // "summoning not available" + MsgRunTimelineUnavailable = "error.run_timeline_unavailable" // "run timeline not available" + MsgNoDescription = "error.no_description" // "agent has no description to resummon from" + MsgSummonCancelled = "info.summon_cancelled" // "summon cancelled by user" + MsgCannotCancel = "error.cannot_cancel_summon" // "agent is not being summoned" + MsgInvalidPath = "error.invalid_path" // "invalid path" // --- Browser cookies --- MsgBrowserCookieTooMany = "error.browser_cookie_too_many" // "too many browser cookies in one sync request" @@ -340,16 +341,16 @@ const ( MsgGrantEnvRevealLimit = "error.grant_env_reveal_limit" // "rate limit exceeded for env reveal" // --- Git credential adapter (Phase 3+) --- - MsgGitCredHostMismatch = "error.git_cred_host_mismatch" // "stored credential is for %s but command targets %s" - MsgGitCredNoMatch = "error.git_cred_no_match" // "no git credential configured for host %s" - MsgGitCredUnsupportedType = "error.git_cred_unsupported_type" // "git credential type %q is not supported" - MsgGitCredTokenInvalid = "error.git_cred_token_invalid" // "stored git token is invalid or empty" - MsgGitCredTokenControlChar = "error.git_cred_token_control_char" // "stored git token contains forbidden control characters" - MsgGitCredHostUserinfoRejected = "error.git_cred_host_userinfo_rejected" // "git URL with embedded userinfo is rejected as ambiguous" + MsgGitCredHostMismatch = "error.git_cred_host_mismatch" // "stored credential is for %s but command targets %s" + MsgGitCredNoMatch = "error.git_cred_no_match" // "no git credential configured for host %s" + MsgGitCredUnsupportedType = "error.git_cred_unsupported_type" // "git credential type %q is not supported" + MsgGitCredTokenInvalid = "error.git_cred_token_invalid" // "stored git token is invalid or empty" + MsgGitCredTokenControlChar = "error.git_cred_token_control_char" // "stored git token contains forbidden control characters" + MsgGitCredHostUserinfoRejected = "error.git_cred_host_userinfo_rejected" // "git URL with embedded userinfo is rejected as ambiguous" MsgGitCredSSHPassphraseUnsupported = "error.git_cred_ssh_passphrase_unsupported" // "passphrase-protected SSH keys not supported in v1" - MsgGitCredSSHKeyInvalid = "error.git_cred_ssh_key_invalid" // "SSH private key invalid: %s" - MsgGitCredHostScopeRequired = "error.git_cred_host_scope_required" // "host_scope required for credential_type %s" - MsgGitCredHostScopeInvalid = "error.git_cred_host_scope_invalid" // "host_scope %q is not a valid hostname" - MsgGitCredBlobMissingField = "error.git_cred_blob_missing_field" // "blob missing required field %q" - MsgGitCredUnsupportedCredType = "error.git_cred_unsupported_cred_type" // "credential_type %q is not supported" + MsgGitCredSSHKeyInvalid = "error.git_cred_ssh_key_invalid" // "SSH private key invalid: %s" + MsgGitCredHostScopeRequired = "error.git_cred_host_scope_required" // "host_scope required for credential_type %s" + MsgGitCredHostScopeInvalid = "error.git_cred_host_scope_invalid" // "host_scope %q is not a valid hostname" + MsgGitCredBlobMissingField = "error.git_cred_blob_missing_field" // "blob missing required field %q" + MsgGitCredUnsupportedCredType = "error.git_cred_unsupported_cred_type" // "credential_type %q is not supported" ) diff --git a/internal/permissions/policy.go b/internal/permissions/policy.go index 67042578..4aa75ff7 100644 --- a/internal/permissions/policy.go +++ b/internal/permissions/policy.go @@ -367,6 +367,7 @@ func isReadMethod(method string) bool { // Sessions read protocol.MethodSessionsList, protocol.MethodSessionsPreview, + protocol.MethodRunTimelineGet, // Skills read protocol.MethodSkillsList, diff --git a/internal/store/pg/factory.go b/internal/store/pg/factory.go index bde42a89..6cee2461 100644 --- a/internal/store/pg/factory.go +++ b/internal/store/pg/factory.go @@ -33,6 +33,7 @@ func NewPGStores(cfg store.StoreConfig) (*store.Stores, error) { Agents: NewPGAgentStore(db), Providers: NewPGProviderStore(db, cfg.EncryptionKey), Tracing: NewPGTracingStore(db), + RunTimeline: NewPGRunTimelineStore(db), MCP: NewPGMCPServerStore(db, cfg.EncryptionKey), ChannelInstances: NewPGChannelInstanceStore(db, cfg.EncryptionKey), ConfigSecrets: NewPGConfigSecretsStore(db, cfg.EncryptionKey), diff --git a/internal/store/pg/run_timeline.go b/internal/store/pg/run_timeline.go new file mode 100644 index 00000000..cfd3e138 --- /dev/null +++ b/internal/store/pg/run_timeline.go @@ -0,0 +1,164 @@ +package pg + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// PGRunTimelineStore implements store.RunTimelineStore backed by PostgreSQL. +type PGRunTimelineStore struct { + db *sql.DB +} + +func NewPGRunTimelineStore(db *sql.DB) *PGRunTimelineStore { + return &PGRunTimelineStore{db: db} +} + +func (s *PGRunTimelineStore) AppendRunTimelineItem(ctx context.Context, item *store.RunTimelineItem) error { + if item.ID == uuid.Nil { + item.ID = store.GenNewID() + } + if item.CreatedAt.IsZero() { + item.CreatedAt = time.Now() + } + tenantID := tenantIDForInsert(ctx) + item.TenantID = tenantID + metadata := item.Metadata + if len(metadata) == 0 { + metadata = []byte(`{}`) + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO run_timeline_items + (id, tenant_id, run_id, session_key, agent_id, user_id, channel, chat_id, seq, + item_type, status, title, preview, content, tool_name, tool_call_id, trace_id, span_id, + metadata, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, + $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20) + ON CONFLICT (tenant_id, run_id, seq) DO UPDATE SET + session_key = EXCLUDED.session_key, + agent_id = EXCLUDED.agent_id, + user_id = EXCLUDED.user_id, + channel = EXCLUDED.channel, + chat_id = EXCLUDED.chat_id, + item_type = EXCLUDED.item_type, + status = EXCLUDED.status, + title = EXCLUDED.title, + preview = EXCLUDED.preview, + content = '', + tool_name = EXCLUDED.tool_name, + tool_call_id = EXCLUDED.tool_call_id, + trace_id = EXCLUDED.trace_id, + span_id = EXCLUDED.span_id, + metadata = EXCLUDED.metadata, + created_at = EXCLUDED.created_at`, + item.ID, tenantID, item.RunID, item.SessionKey, nilUUID(item.AgentID), nilStr(item.UserID), + nilStr(item.Channel), nilStr(item.ChatID), item.Seq, item.ItemType, nilStr(item.Status), + nilStr(item.Title), nilStr(item.Preview), "", nilStr(item.ToolName), nilStr(item.ToolCallID), + nilUUID(item.TraceID), nilUUID(item.SpanID), jsonOrEmpty(metadata), item.CreatedAt, + ) + if err == nil { + item.Content = "" + } + return err +} + +func (s *PGRunTimelineStore) ListRunTimelineItems(ctx context.Context, opts store.RunTimelineListOpts) ([]store.RunTimelineItem, error) { + where, args := buildRunTimelineWhere(ctx, opts) + limit := opts.Limit + if limit <= 0 || limit > 500 { + limit = 200 + } + q := `SELECT id, tenant_id, run_id, session_key, agent_id, user_id, channel, chat_id, seq, + item_type, status, title, preview, COALESCE(content, '') AS content, tool_name, tool_call_id, + trace_id, span_id, COALESCE(metadata, '{}'::jsonb) AS metadata, created_at + FROM run_timeline_items` + where + + runTimelineOrderBy(opts) + + fmt.Sprintf(" OFFSET %d LIMIT %d", opts.Offset, limit) + + var rows []runTimelineRow + if err := pkgSqlxDB.SelectContext(ctx, &rows, q, args...); err != nil { + return nil, err + } + items := make([]store.RunTimelineItem, len(rows)) + for i, row := range rows { + items[i] = row.toStore() + } + return items, nil +} + +func runTimelineOrderBy(opts store.RunTimelineListOpts) string { + if opts.RunID != "" { + return " ORDER BY seq ASC, created_at ASC" + } + return " ORDER BY created_at ASC, seq ASC" +} + +func buildRunTimelineWhere(ctx context.Context, opts store.RunTimelineListOpts) (string, []any) { + var conditions []string + var args []any + argIdx := 1 + if !store.IsCrossTenant(ctx) { + tenantID := store.TenantIDFromContext(ctx) + if tenantID == uuid.Nil { + return " WHERE 1=0", nil + } + conditions = append(conditions, fmt.Sprintf("tenant_id = $%d", argIdx)) + args = append(args, tenantID) + argIdx++ + } + if opts.RunID != "" { + conditions = append(conditions, fmt.Sprintf("run_id = $%d", argIdx)) + args = append(args, opts.RunID) + argIdx++ + } + if opts.SessionKey != "" { + conditions = append(conditions, fmt.Sprintf("session_key = $%d", argIdx)) + args = append(args, opts.SessionKey) + } + if len(conditions) == 0 { + return " WHERE 1=0", nil + } + return " WHERE " + strings.Join(conditions, " AND "), args +} + +type runTimelineRow struct { + ID uuid.UUID `db:"id"` + TenantID uuid.UUID `db:"tenant_id"` + RunID string `db:"run_id"` + SessionKey string `db:"session_key"` + AgentID *uuid.UUID `db:"agent_id"` + UserID *string `db:"user_id"` + Channel *string `db:"channel"` + ChatID *string `db:"chat_id"` + Seq int `db:"seq"` + ItemType string `db:"item_type"` + Status *string `db:"status"` + Title *string `db:"title"` + Preview *string `db:"preview"` + Content string `db:"content"` + ToolName *string `db:"tool_name"` + ToolCallID *string `db:"tool_call_id"` + TraceID *uuid.UUID `db:"trace_id"` + SpanID *uuid.UUID `db:"span_id"` + Metadata json.RawMessage `db:"metadata"` + CreatedAt time.Time `db:"created_at"` +} + +func (r runTimelineRow) toStore() store.RunTimelineItem { + return store.RunTimelineItem{ + ID: r.ID, TenantID: r.TenantID, RunID: r.RunID, SessionKey: r.SessionKey, + AgentID: r.AgentID, UserID: derefStr(r.UserID), Channel: derefStr(r.Channel), + ChatID: derefStr(r.ChatID), Seq: r.Seq, ItemType: r.ItemType, + Status: derefStr(r.Status), Title: derefStr(r.Title), Preview: derefStr(r.Preview), + Content: r.Content, ToolName: derefStr(r.ToolName), ToolCallID: derefStr(r.ToolCallID), + TraceID: r.TraceID, SpanID: r.SpanID, Metadata: r.Metadata, CreatedAt: r.CreatedAt, + } +} diff --git a/internal/store/pg/run_timeline_test.go b/internal/store/pg/run_timeline_test.go new file mode 100644 index 00000000..27610f6e --- /dev/null +++ b/internal/store/pg/run_timeline_test.go @@ -0,0 +1,107 @@ +package pg + +import ( + "context" + "encoding/json" + "testing" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestPGRunTimelineStoreAppendAndListBySeq(t *testing.T) { + db := hooksTestDB(t) + timeline := NewPGRunTimelineStore(db) + tenantID, agentID := seedTenantAndAgent(t, db) + ctx := store.WithTenantID(context.Background(), tenantID) + + items := []store.RunTimelineItem{ + { + RunID: "pg-run-1", + SessionKey: "agent:default:direct:user-1", + AgentID: &agentID, + UserID: "user-1", + Channel: "web", + Seq: 2, + ItemType: store.RunTimelineItemTypeToolCall, + Status: store.RunTimelineStatusRunning, + Title: "exec_command", + Preview: `{"cmd":"date"}`, + ToolName: "exec_command", + ToolCallID: "call-1", + Metadata: json.RawMessage(`{"safe":true}`), + }, + { + RunID: "pg-run-1", + SessionKey: "agent:default:direct:user-1", + UserID: "user-1", + Channel: "web", + Seq: 1, + ItemType: store.RunTimelineItemTypeRunStatus, + Status: store.RunTimelineStatusStarted, + Title: "Run started", + Preview: "hello", + }, + } + for i := range items { + if err := timeline.AppendRunTimelineItem(ctx, &items[i]); err != nil { + t.Fatalf("AppendRunTimelineItem(%d): %v", i, err) + } + } + + got, err := timeline.ListRunTimelineItems(ctx, store.RunTimelineListOpts{RunID: "pg-run-1", Limit: 10}) + if err != nil { + t.Fatalf("ListRunTimelineItems: %v", err) + } + if len(got) != 2 { + t.Fatalf("len(items) = %d, want 2", len(got)) + } + if got[0].Seq != 1 || got[1].Seq != 2 { + t.Fatalf("seq order = [%d,%d], want [1,2]", got[0].Seq, got[1].Seq) + } + if got[1].Content != "" { + t.Fatalf("Content persisted = %q, want empty preview-only archive", got[1].Content) + } +} + +func TestPGRunTimelineStoreTenantScope(t *testing.T) { + db := hooksTestDB(t) + timeline := NewPGRunTimelineStore(db) + tenantID, _ := seedTenantAndAgent(t, db) + ctxA := store.WithTenantID(context.Background(), tenantID) + ctxB := store.WithTenantID(context.Background(), uuid.Must(uuid.NewV7())) + + item := store.RunTimelineItem{ + RunID: "pg-run-scope", + SessionKey: "agent:default:direct:user-1", + Seq: 1, + ItemType: store.RunTimelineItemTypeAssistantMessage, + Status: store.RunTimelineStatusCompleted, + Title: "Assistant", + Preview: "Visible only to tenant A", + Content: `{"raw":"must not persist"}`, + } + if err := timeline.AppendRunTimelineItem(ctxA, &item); err != nil { + t.Fatalf("AppendRunTimelineItem: %v", err) + } + + gotA, err := timeline.ListRunTimelineItems(ctxA, store.RunTimelineListOpts{RunID: "pg-run-scope"}) + if err != nil { + t.Fatalf("List tenant A: %v", err) + } + if len(gotA) != 1 { + t.Fatalf("tenant A len = %d, want 1", len(gotA)) + } + if gotA[0].Content != "" { + t.Fatalf("Content = %q, want empty", gotA[0].Content) + } + + gotB, err := timeline.ListRunTimelineItems(ctxB, store.RunTimelineListOpts{RunID: "pg-run-scope"}) + if err != nil { + t.Fatalf("List tenant B: %v", err) + } + if len(gotB) != 0 { + t.Fatalf("tenant B len = %d, want 0", len(gotB)) + } +} diff --git a/internal/store/run_timeline_store.go b/internal/store/run_timeline_store.go new file mode 100644 index 00000000..1dda34e9 --- /dev/null +++ b/internal/store/run_timeline_store.go @@ -0,0 +1,64 @@ +package store + +import ( + "context" + "encoding/json" + "time" + + "github.com/google/uuid" +) + +const ( + RunTimelineItemTypeActivity = "activity" + RunTimelineItemTypeAssistantMessage = "assistant.message" + RunTimelineItemTypeToolCall = "tool.call" + RunTimelineItemTypeToolResult = "tool.result" + RunTimelineItemTypeRunStatus = "run.status" +) + +const ( + RunTimelineStatusStarted = "started" + RunTimelineStatusRunning = "running" + RunTimelineStatusCompleted = "completed" + RunTimelineStatusFailed = "failed" + RunTimelineStatusCancelled = "cancelled" +) + +// RunTimelineItem is a persisted, display-safe archive entry for one agent run. +type RunTimelineItem struct { + ID uuid.UUID `json:"id" db:"id"` + TenantID uuid.UUID `json:"tenant_id" db:"tenant_id"` + RunID string `json:"run_id" db:"run_id"` + SessionKey string `json:"session_key" db:"session_key"` + AgentID *uuid.UUID `json:"agent_id,omitempty" db:"agent_id"` + UserID string `json:"user_id,omitempty" db:"user_id"` + Channel string `json:"channel,omitempty" db:"channel"` + ChatID string `json:"chat_id,omitempty" db:"chat_id"` + Seq int `json:"seq" db:"seq"` + ItemType string `json:"item_type" db:"item_type"` + Status string `json:"status,omitempty" db:"status"` + Title string `json:"title,omitempty" db:"title"` + Preview string `json:"preview,omitempty" db:"preview"` + Content string `json:"content,omitempty" db:"content"` + ToolName string `json:"tool_name,omitempty" db:"tool_name"` + ToolCallID string `json:"tool_call_id,omitempty" db:"tool_call_id"` + TraceID *uuid.UUID `json:"trace_id,omitempty" db:"trace_id"` + SpanID *uuid.UUID `json:"span_id,omitempty" db:"span_id"` + Metadata json.RawMessage `json:"metadata,omitempty" db:"metadata"` + CreatedAt time.Time `json:"created_at" db:"created_at"` +} + +// RunTimelineListOpts scopes a timeline read. RunID is preferred; SessionKey is +// a fallback for session archive views that need the latest runs in a session. +type RunTimelineListOpts struct { + RunID string + SessionKey string + Limit int + Offset int +} + +// RunTimelineStore appends and lists archived agent run timeline entries. +type RunTimelineStore interface { + AppendRunTimelineItem(ctx context.Context, item *RunTimelineItem) error + ListRunTimelineItems(ctx context.Context, opts RunTimelineListOpts) ([]RunTimelineItem, error) +} diff --git a/internal/store/sqlitestore/factory.go b/internal/store/sqlitestore/factory.go index 527a6447..296ca92d 100644 --- a/internal/store/sqlitestore/factory.go +++ b/internal/store/sqlitestore/factory.go @@ -36,43 +36,44 @@ func NewSQLiteStores(cfg store.StoreConfig) (*store.Stores, error) { } sqliteStores := &store.Stores{ - DB: db, - Sessions: NewSQLiteSessionStore(db), - Agents: NewSQLiteAgentStore(db), - Providers: NewSQLiteProviderStore(db, cfg.EncryptionKey), - Tracing: NewSQLiteTracingStore(db), - ConfigSecrets: NewSQLiteConfigSecretsStore(db, cfg.EncryptionKey), - BuiltinTools: NewSQLiteBuiltinToolStore(db), - Heartbeats: NewSQLiteHeartbeatStore(db), - Tenants: NewSQLiteTenantStore(db), - BuiltinToolTenantCfgs: NewSQLiteBuiltinToolTenantConfigStore(db), - SkillTenantCfgs: NewSQLiteSkillTenantConfigStore(db), - SystemConfigs: NewSQLiteSystemConfigStore(db), - Snapshots: NewSQLiteSnapshotStore(db), - Cron: NewSQLiteCronStore(db), - ChannelInstances: NewSQLiteChannelInstanceStore(db, cfg.EncryptionKey), - Pairing: NewSQLitePairingStore(db), - PendingMessages: NewSQLitePendingMessageStore(db), - Contacts: NewSQLiteContactStore(db), - Teams: NewSQLiteTeamStore(db), - Skills: NewSQLiteSkillStore(db, cfg.SkillsStorageDir), - MCP: NewSQLiteMCPServerStore(db, cfg.EncryptionKey), - Activity: NewSQLiteActivityStore(db), - APIKeys: NewSQLiteAPIKeyStore(db), - ConfigPermissions: NewSQLiteConfigPermissionStore(db), - BrowserCookies: NewSQLiteBrowserCookieStore(db, cfg.EncryptionKey), - Memory: NewSQLiteMemoryStore(db), - SubagentTasks: NewSQLiteSubagentTaskStore(db), - AgentLinks: NewSQLiteAgentLinkStore(db), - SecureCLI: secureCLI, - SecureCLIGrants: NewSQLiteSecureCLIAgentGrantStore(db, cfg.EncryptionKey), - Episodic: NewSQLiteEpisodicStore(db), - EvolutionMetrics: NewSQLiteEvolutionMetricsStore(db), - EvolutionSuggestions: NewSQLiteEvolutionSuggestionStore(db), - KnowledgeGraph: NewSQLiteKnowledgeGraphStore(db), - Vault: NewSQLiteVaultStore(db), - BitrixPortals: NewSQLiteBitrixPortalStore(db, cfg.EncryptionKey), - Hooks: NewSQLiteHookStore(db), + DB: db, + Sessions: NewSQLiteSessionStore(db), + Agents: NewSQLiteAgentStore(db), + Providers: NewSQLiteProviderStore(db, cfg.EncryptionKey), + Tracing: NewSQLiteTracingStore(db), + RunTimeline: NewSQLiteRunTimelineStore(db), + ConfigSecrets: NewSQLiteConfigSecretsStore(db, cfg.EncryptionKey), + BuiltinTools: NewSQLiteBuiltinToolStore(db), + Heartbeats: NewSQLiteHeartbeatStore(db), + Tenants: NewSQLiteTenantStore(db), + BuiltinToolTenantCfgs: NewSQLiteBuiltinToolTenantConfigStore(db), + SkillTenantCfgs: NewSQLiteSkillTenantConfigStore(db), + SystemConfigs: NewSQLiteSystemConfigStore(db), + Snapshots: NewSQLiteSnapshotStore(db), + Cron: NewSQLiteCronStore(db), + ChannelInstances: NewSQLiteChannelInstanceStore(db, cfg.EncryptionKey), + Pairing: NewSQLitePairingStore(db), + PendingMessages: NewSQLitePendingMessageStore(db), + Contacts: NewSQLiteContactStore(db), + Teams: NewSQLiteTeamStore(db), + Skills: NewSQLiteSkillStore(db, cfg.SkillsStorageDir), + MCP: NewSQLiteMCPServerStore(db, cfg.EncryptionKey), + Activity: NewSQLiteActivityStore(db), + APIKeys: NewSQLiteAPIKeyStore(db), + ConfigPermissions: NewSQLiteConfigPermissionStore(db), + BrowserCookies: NewSQLiteBrowserCookieStore(db, cfg.EncryptionKey), + Memory: NewSQLiteMemoryStore(db), + SubagentTasks: NewSQLiteSubagentTaskStore(db), + AgentLinks: NewSQLiteAgentLinkStore(db), + SecureCLI: secureCLI, + SecureCLIGrants: NewSQLiteSecureCLIAgentGrantStore(db, cfg.EncryptionKey), + Episodic: NewSQLiteEpisodicStore(db), + EvolutionMetrics: NewSQLiteEvolutionMetricsStore(db), + EvolutionSuggestions: NewSQLiteEvolutionSuggestionStore(db), + KnowledgeGraph: NewSQLiteKnowledgeGraphStore(db), + Vault: NewSQLiteVaultStore(db), + BitrixPortals: NewSQLiteBitrixPortalStore(db, cfg.EncryptionKey), + Hooks: NewSQLiteHookStore(db), Webhooks: NewSQLiteWebhookStore(db), WebhookCalls: NewSQLiteWebhookCallStore(db), Workstations: NewSQLiteWorkstationStore(db, cfg.EncryptionKey), diff --git a/internal/store/sqlitestore/run_timeline.go b/internal/store/sqlitestore/run_timeline.go new file mode 100644 index 00000000..c3c4723d --- /dev/null +++ b/internal/store/sqlitestore/run_timeline.go @@ -0,0 +1,155 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "database/sql" + "fmt" + "strings" + "time" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +// SQLiteRunTimelineStore implements store.RunTimelineStore backed by SQLite. +type SQLiteRunTimelineStore struct { + db *sql.DB +} + +func NewSQLiteRunTimelineStore(db *sql.DB) *SQLiteRunTimelineStore { + return &SQLiteRunTimelineStore{db: db} +} + +func (s *SQLiteRunTimelineStore) AppendRunTimelineItem(ctx context.Context, item *store.RunTimelineItem) error { + if item.ID == uuid.Nil { + item.ID = store.GenNewID() + } + if item.CreatedAt.IsZero() { + item.CreatedAt = time.Now() + } + tenantID := tenantIDForInsert(ctx) + item.TenantID = tenantID + metadata := item.Metadata + if len(metadata) == 0 { + metadata = []byte(`{}`) + } + _, err := s.db.ExecContext(ctx, + `INSERT INTO run_timeline_items + (id, tenant_id, run_id, session_key, agent_id, user_id, channel, chat_id, seq, + item_type, status, title, preview, content, tool_name, tool_call_id, trace_id, span_id, + metadata, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (tenant_id, run_id, seq) DO UPDATE SET + session_key = excluded.session_key, + agent_id = excluded.agent_id, + user_id = excluded.user_id, + channel = excluded.channel, + chat_id = excluded.chat_id, + item_type = excluded.item_type, + status = excluded.status, + title = excluded.title, + preview = excluded.preview, + content = '', + tool_name = excluded.tool_name, + tool_call_id = excluded.tool_call_id, + trace_id = excluded.trace_id, + span_id = excluded.span_id, + metadata = excluded.metadata, + created_at = excluded.created_at`, + item.ID, tenantID, item.RunID, item.SessionKey, nilUUID(item.AgentID), nilStr(item.UserID), + nilStr(item.Channel), nilStr(item.ChatID), item.Seq, item.ItemType, nilStr(item.Status), + nilStr(item.Title), nilStr(item.Preview), "", nilStr(item.ToolName), nilStr(item.ToolCallID), + nilUUID(item.TraceID), nilUUID(item.SpanID), string(metadata), item.CreatedAt, + ) + if err == nil { + item.Content = "" + } + return err +} + +func (s *SQLiteRunTimelineStore) ListRunTimelineItems(ctx context.Context, opts store.RunTimelineListOpts) ([]store.RunTimelineItem, error) { + where, args := buildRunTimelineWhere(ctx, opts) + limit := opts.Limit + if limit <= 0 || limit > 500 { + limit = 200 + } + q := `SELECT id, tenant_id, run_id, session_key, agent_id, user_id, channel, chat_id, seq, + item_type, status, title, preview, COALESCE(content, '') AS content, tool_name, tool_call_id, + trace_id, span_id, COALESCE(metadata, '{}') AS metadata, created_at + FROM run_timeline_items` + where + + runTimelineOrderBy(opts) + + fmt.Sprintf(" LIMIT %d OFFSET %d", limit, opts.Offset) + + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + return scanRunTimelineRows(rows) +} + +func runTimelineOrderBy(opts store.RunTimelineListOpts) string { + if opts.RunID != "" { + return " ORDER BY seq ASC, created_at ASC" + } + return " ORDER BY created_at ASC, seq ASC" +} + +func buildRunTimelineWhere(ctx context.Context, opts store.RunTimelineListOpts) (string, []any) { + var conditions []string + var args []any + if !store.IsCrossTenant(ctx) { + tenantID := store.TenantIDFromContext(ctx) + if tenantID == uuid.Nil { + return " WHERE 1=0", nil + } + conditions = append(conditions, "tenant_id = ?") + args = append(args, tenantID) + } + if opts.RunID != "" { + conditions = append(conditions, "run_id = ?") + args = append(args, opts.RunID) + } + if opts.SessionKey != "" { + conditions = append(conditions, "session_key = ?") + args = append(args, opts.SessionKey) + } + if len(conditions) == 0 { + return " WHERE 1=0", nil + } + return " WHERE " + strings.Join(conditions, " AND "), args +} + +func scanRunTimelineRows(rows *sql.Rows) ([]store.RunTimelineItem, error) { + var items []store.RunTimelineItem + for rows.Next() { + var item store.RunTimelineItem + var agentID, traceID, spanID *uuid.UUID + var userID, channel, chatID, status, title, preview, toolName, toolCallID sql.NullString + var metadata string + var createdAt sqliteTime + if err := rows.Scan(&item.ID, &item.TenantID, &item.RunID, &item.SessionKey, &agentID, + &userID, &channel, &chatID, &item.Seq, &item.ItemType, &status, &title, &preview, + &item.Content, &toolName, &toolCallID, &traceID, &spanID, &metadata, &createdAt); err != nil { + return nil, err + } + item.AgentID = agentID + item.TraceID = traceID + item.SpanID = spanID + item.UserID = userID.String + item.Channel = channel.String + item.ChatID = chatID.String + item.Status = status.String + item.Title = title.String + item.Preview = preview.String + item.ToolName = toolName.String + item.ToolCallID = toolCallID.String + item.Metadata = []byte(metadata) + item.CreatedAt = createdAt.Time + items = append(items, item) + } + return items, rows.Err() +} diff --git a/internal/store/sqlitestore/run_timeline_test.go b/internal/store/sqlitestore/run_timeline_test.go new file mode 100644 index 00000000..139be542 --- /dev/null +++ b/internal/store/sqlitestore/run_timeline_test.go @@ -0,0 +1,160 @@ +//go:build sqlite || sqliteonly + +package sqlitestore + +import ( + "context" + "database/sql" + "encoding/json" + "testing" + + "github.com/google/uuid" + + "github.com/nextlevelbuilder/goclaw/internal/store" +) + +func TestSQLiteRunTimelineStoreAppendAndListBySeq(t *testing.T) { + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + + timeline := NewSQLiteRunTimelineStore(db) + ctx := store.WithTenantID(context.Background(), store.MasterTenantID) + agentID := uuid.Must(uuid.NewV7()) + seedSQLiteRunTimelineAgent(t, db, store.MasterTenantID, agentID) + + items := []store.RunTimelineItem{ + { + RunID: "run-1", + SessionKey: "agent:default:direct:user-1", + AgentID: &agentID, + UserID: "user-1", + Channel: "web", + Seq: 2, + ItemType: store.RunTimelineItemTypeToolCall, + Status: store.RunTimelineStatusRunning, + Title: "read_file", + Preview: `{"path":"/tmp/a.txt"}`, + ToolName: "read_file", + ToolCallID: "call-1", + Metadata: json.RawMessage(`{"safe":true}`), + }, + { + RunID: "run-1", + SessionKey: "agent:default:direct:user-1", + UserID: "user-1", + Channel: "web", + Seq: 1, + ItemType: store.RunTimelineItemTypeRunStatus, + Status: store.RunTimelineStatusStarted, + Title: "Run started", + Preview: "hello", + }, + } + for i := range items { + if err := timeline.AppendRunTimelineItem(ctx, &items[i]); err != nil { + t.Fatalf("AppendRunTimelineItem(%d): %v", i, err) + } + } + + got, err := timeline.ListRunTimelineItems(ctx, store.RunTimelineListOpts{RunID: "run-1", Limit: 10}) + if err != nil { + t.Fatalf("ListRunTimelineItems: %v", err) + } + if len(got) != 2 { + t.Fatalf("len(items) = %d, want 2", len(got)) + } + if got[0].Seq != 1 || got[1].Seq != 2 { + t.Fatalf("seq order = [%d,%d], want [1,2]", got[0].Seq, got[1].Seq) + } + if got[1].Content != "" { + t.Fatalf("Content persisted = %q, want empty preview-only archive", got[1].Content) + } + if got[1].AgentID == nil || *got[1].AgentID != agentID { + t.Fatalf("AgentID = %v, want %s", got[1].AgentID, agentID) + } +} + +func TestSQLiteRunTimelineStoreTenantScope(t *testing.T) { + db := openTestDB(t) + if err := EnsureSchema(db); err != nil { + t.Fatalf("EnsureSchema: %v", err) + } + + timeline := NewSQLiteRunTimelineStore(db) + tenantA := uuid.Must(uuid.NewV7()) + tenantB := uuid.Must(uuid.NewV7()) + seedSQLiteRunTimelineTenant(t, db, tenantA) + seedSQLiteRunTimelineTenant(t, db, tenantB) + ctxA := store.WithTenantID(context.Background(), tenantA) + ctxB := store.WithTenantID(context.Background(), tenantB) + + item := store.RunTimelineItem{ + RunID: "run-shared", + SessionKey: "agent:default:direct:user-1", + Seq: 1, + ItemType: store.RunTimelineItemTypeAssistantMessage, + Status: store.RunTimelineStatusCompleted, + Title: "Assistant", + Preview: "Visible only to tenant A", + Content: `{"raw":"must not persist"}`, + } + if err := timeline.AppendRunTimelineItem(ctxA, &item); err != nil { + t.Fatalf("AppendRunTimelineItem: %v", err) + } + + gotA, err := timeline.ListRunTimelineItems(ctxA, store.RunTimelineListOpts{RunID: "run-shared"}) + if err != nil { + t.Fatalf("List tenant A: %v", err) + } + if len(gotA) != 1 { + t.Fatalf("tenant A len = %d, want 1", len(gotA)) + } + if gotA[0].Content != "" { + t.Fatalf("Content = %q, want empty", gotA[0].Content) + } + + gotB, err := timeline.ListRunTimelineItems(ctxB, store.RunTimelineListOpts{RunID: "run-shared"}) + if err != nil { + t.Fatalf("List tenant B: %v", err) + } + if len(gotB) != 0 { + t.Fatalf("tenant B len = %d, want 0", len(gotB)) + } + + gotNoTenant, err := timeline.ListRunTimelineItems(context.Background(), store.RunTimelineListOpts{RunID: "run-shared"}) + if err != nil { + t.Fatalf("List no tenant: %v", err) + } + if len(gotNoTenant) != 0 { + t.Fatalf("no-tenant len = %d, want fail-closed 0", len(gotNoTenant)) + } +} + +func seedSQLiteRunTimelineTenant(t *testing.T, db execer, tenantID uuid.UUID) { + t.Helper() + if _, err := db.Exec( + `INSERT OR IGNORE INTO tenants (id, name, slug, status) + VALUES (?, ?, ?, 'active')`, + tenantID, "timeline-test-"+tenantID.String()[:8], "timeline-"+tenantID.String(), + ); err != nil { + t.Fatalf("seed tenant: %v", err) + } +} + +func seedSQLiteRunTimelineAgent(t *testing.T, db execer, tenantID, agentID uuid.UUID) { + t.Helper() + seedSQLiteRunTimelineTenant(t, db, tenantID) + if _, err := db.Exec( + `INSERT OR IGNORE INTO agents (id, tenant_id, agent_key, agent_type, status, provider, model, owner_id) + VALUES (?, ?, ?, 'predefined', 'active', 'test', 'test-model', 'owner')`, + agentID, tenantID, "timeline-agent-"+agentID.String(), + ); err != nil { + t.Fatalf("seed agent: %v", err) + } +} + +type execer interface { + Exec(query string, args ...any) (sql.Result, error) +} diff --git a/internal/store/sqlitestore/schema.go b/internal/store/sqlitestore/schema.go index a834844b..8a60d4f0 100644 --- a/internal/store/sqlitestore/schema.go +++ b/internal/store/sqlitestore/schema.go @@ -16,7 +16,7 @@ var schemaSQL string // SchemaVersion is the current SQLite schema version. // Bump this when adding new migration steps below. -const SchemaVersion = 42 +const SchemaVersion = 43 // migrations maps version → SQL to apply when upgrading FROM that version. // schema.sql always represents the LATEST full schema (for fresh DBs). @@ -772,6 +772,37 @@ CREATE INDEX IF NOT EXISTS idx_browser_cookies_expires_at 40: `ALTER TABLE secure_cli_user_credentials ADD COLUMN host_scope TEXT;`, // Version 41 → 42: credential adapter framework — adapter_name on binaries. 41: `ALTER TABLE secure_cli_binaries ADD COLUMN adapter_name TEXT;`, + // Version 42 → 43: archived agent run timeline. + 42: `CREATE TABLE IF NOT EXISTS run_timeline_items ( + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + run_id TEXT NOT NULL, + session_key TEXT NOT NULL, + agent_id TEXT REFERENCES agents(id) ON DELETE SET NULL, + user_id TEXT, + channel TEXT, + chat_id TEXT, + seq INTEGER NOT NULL, + item_type TEXT NOT NULL, + status TEXT, + title TEXT, + preview TEXT, + content TEXT NOT NULL DEFAULT '', + tool_name TEXT, + tool_call_id TEXT, + trace_id TEXT REFERENCES traces(id) ON DELETE SET NULL, + span_id TEXT REFERENCES spans(id) ON DELETE SET NULL, + metadata TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE (tenant_id, run_id, seq) +); +CREATE INDEX IF NOT EXISTS idx_run_timeline_run_seq + ON run_timeline_items (tenant_id, run_id, seq); +CREATE INDEX IF NOT EXISTS idx_run_timeline_session_time + ON run_timeline_items (tenant_id, session_key, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_run_timeline_trace + ON run_timeline_items (tenant_id, trace_id) + WHERE trace_id IS NOT NULL;`, } // addHooksTables is the SQLite incremental migration for schema v19 → v20. diff --git a/internal/store/sqlitestore/schema.sql b/internal/store/sqlitestore/schema.sql index 7b837760..af889780 100644 --- a/internal/store/sqlitestore/schema.sql +++ b/internal/store/sqlitestore/schema.sql @@ -591,6 +591,42 @@ CREATE INDEX IF NOT EXISTS idx_traces_team ON traces(team_id, created_at DESC) W CREATE INDEX IF NOT EXISTS idx_traces_tenant ON traces(tenant_id); CREATE INDEX IF NOT EXISTS idx_traces_tenant_time ON traces(tenant_id, created_at DESC); +-- ============================================================ +-- Table: run_timeline_items +-- ============================================================ + +CREATE TABLE IF NOT EXISTS run_timeline_items ( + id TEXT NOT NULL PRIMARY KEY, + tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + run_id TEXT NOT NULL, + session_key TEXT NOT NULL, + agent_id TEXT REFERENCES agents(id) ON DELETE SET NULL, + user_id TEXT, + channel TEXT, + chat_id TEXT, + seq INTEGER NOT NULL, + item_type TEXT NOT NULL, + status TEXT, + title TEXT, + preview TEXT, + content TEXT NOT NULL DEFAULT '', + tool_name TEXT, + tool_call_id TEXT, + trace_id TEXT REFERENCES traces(id) ON DELETE SET NULL, + span_id TEXT REFERENCES spans(id) ON DELETE SET NULL, + metadata TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now')), + UNIQUE (tenant_id, run_id, seq) +); + +CREATE INDEX IF NOT EXISTS idx_run_timeline_run_seq + ON run_timeline_items (tenant_id, run_id, seq); +CREATE INDEX IF NOT EXISTS idx_run_timeline_session_time + ON run_timeline_items (tenant_id, session_key, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_run_timeline_trace + ON run_timeline_items (tenant_id, trace_id) + WHERE trace_id IS NOT NULL; + -- ============================================================ -- Table: spans -- ============================================================ diff --git a/internal/store/stores.go b/internal/store/stores.go index 7a1a9887..e48ef9b8 100644 --- a/internal/store/stores.go +++ b/internal/store/stores.go @@ -13,6 +13,7 @@ type Stores struct { Agents AgentStore Providers ProviderStore Tracing TracingStore + RunTimeline RunTimelineStore MCP MCPServerStore ChannelInstances ChannelInstanceStore ConfigSecrets ConfigSecretsStore diff --git a/internal/upgrade/version.go b/internal/upgrade/version.go index 0033f111..279d5bfd 100644 --- a/internal/upgrade/version.go +++ b/internal/upgrade/version.go @@ -2,4 +2,4 @@ package upgrade // RequiredSchemaVersion is the schema migration version this binary requires. // Bump this whenever adding a new SQL migration file. -const RequiredSchemaVersion uint = 73 +const RequiredSchemaVersion uint = 74 diff --git a/migrations/000074_run_timeline_items.down.sql b/migrations/000074_run_timeline_items.down.sql new file mode 100644 index 00000000..0ca66cff --- /dev/null +++ b/migrations/000074_run_timeline_items.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS run_timeline_items; diff --git a/migrations/000074_run_timeline_items.up.sql b/migrations/000074_run_timeline_items.up.sql new file mode 100644 index 00000000..5cb71c1c --- /dev/null +++ b/migrations/000074_run_timeline_items.up.sql @@ -0,0 +1,27 @@ +CREATE TABLE run_timeline_items ( + id UUID PRIMARY KEY DEFAULT uuid_generate_v7(), + tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE, + run_id TEXT NOT NULL, + session_key VARCHAR(500) NOT NULL, + agent_id UUID REFERENCES agents(id) ON DELETE SET NULL, + user_id VARCHAR(255), + channel VARCHAR(50), + chat_id VARCHAR(255), + seq INT NOT NULL, + item_type VARCHAR(40) NOT NULL, + status VARCHAR(40), + title TEXT, + preview TEXT, + content TEXT NOT NULL DEFAULT '', + tool_name VARCHAR(255), + tool_call_id VARCHAR(255), + trace_id UUID REFERENCES traces(id) ON DELETE SET NULL, + span_id UUID REFERENCES spans(id) ON DELETE SET NULL, + metadata JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (tenant_id, run_id, seq) +); + +CREATE INDEX idx_run_timeline_run_seq ON run_timeline_items(tenant_id, run_id, seq); +CREATE INDEX idx_run_timeline_session_time ON run_timeline_items(tenant_id, session_key, created_at DESC); +CREATE INDEX idx_run_timeline_trace ON run_timeline_items(tenant_id, trace_id) WHERE trace_id IS NOT NULL; diff --git a/pkg/protocol/methods.go b/pkg/protocol/methods.go index 73ad36f2..a8d0e150 100644 --- a/pkg/protocol/methods.go +++ b/pkg/protocol/methods.go @@ -41,6 +41,7 @@ const ( MethodSessionsDelete = "sessions.delete" MethodSessionsReset = "sessions.reset" MethodSessionsCompact = "sessions.compact" + MethodRunTimelineGet = "run.timeline.get" // System MethodConnect = "connect" diff --git a/plans/260528-1804-archived-run-timeline/phase-01-tdd-contract-and-schema.md b/plans/260528-1804-archived-run-timeline/phase-01-tdd-contract-and-schema.md new file mode 100644 index 00000000..6efc057c --- /dev/null +++ b/plans/260528-1804-archived-run-timeline/phase-01-tdd-contract-and-schema.md @@ -0,0 +1,128 @@ +--- +phase: 1 +title: "TDD Contract and Schema" +status: complete +priority: P1 +effort: "1d" +dependencies: [] +--- + +# Phase 1: TDD Contract and Schema + +## Context Links + +- GitHub issue: `digitopvn/goclaw#76` +- Related issue boundary: `digitopvn/goclaw#67` +- Current schema: `migrations/000001_init_schema.up.sql` +- SQLite schema: `internal/store/sqlitestore/schema.sql` +- Trace store reference: `internal/store/tracing_store.go` + +## Overview + +Define the durable run timeline contract and add tests before schema/store code. This phase locks the archive semantics so later implementation does not drift into issue #67 delivery behavior. + +## Key Insights + +- `sessions.messages` is useful for conversation history but not reliable enough for exact live event rhythm. +- `traces` and `spans` are diagnostic product data with different retention/debug semantics. +- Archive timeline needs its own append-only table because ordering and privacy rules differ. + +## Requirements + +- Functional: define timeline item DTOs for activity, assistant message, tool call, tool result, and run status. +- Functional: expose Phase 1 storage as previews only for tool args/results. +- Functional: include `run_id`, `session_key`, `agent_id`, `tenant_id`, `seq`, `type`, `status`, timestamps, optional `trace_id`, optional `span_id`. +- Non-functional: preserve strict sequence ordering under parallel tool calls. +- Non-functional: support both PostgreSQL and SQLite/Lite. +- Non-functional: no raw hidden thinking persistence. + +## Architecture + +Use a dedicated append-only product table instead of reconstructing from `sessions.messages` or `spans`. + +Proposed table: + +```text +run_timeline_items +- id uuid/text primary key +- tenant_id uuid/text not null +- run_id text not null +- session_key text not null +- agent_id uuid/text nullable +- user_id text nullable +- channel text nullable +- chat_id text nullable +- seq bigint not null +- item_type text not null +- status text nullable +- title text nullable +- preview text nullable +- content text nullable +- tool_name text nullable +- tool_call_id text nullable +- trace_id uuid/text nullable +- span_id uuid/text nullable +- metadata json/jsonb nullable +- created_at timestamp not null +``` + +Index: +- `(tenant_id, run_id, seq)` +- `(tenant_id, session_key, created_at desc)` +- optional `(trace_id)` where supported + +## Related Code Files + +- Create: `internal/store/run_timeline_store.go` +- Create: `internal/store/pg/run_timeline.go` +- Create: `internal/store/sqlitestore/run_timeline.go` +- Create: `internal/store/pg/run_timeline_test.go` +- Create: `internal/store/sqlitestore/run_timeline_test.go` +- Create: `migrations/0000XX_run_timeline_items.up.sql` +- Create: `migrations/0000XX_run_timeline_items.down.sql` +- Modify: `internal/store/sqlitestore/schema.sql` +- Modify: `internal/store/sqlitestore/schema.go` +- Modify: `internal/upgrade/version.go` + +## Implementation Steps + +1. Write store contract tests first: + - append items with explicit sequence. + - list by `tenant_id + run_id` ordered by `seq`. + - reject cross-tenant reads. + - persist preview fields but not full tool JSON. +2. Write schema migration tests or startup schema assertions for PostgreSQL and SQLite. +3. Add store interface and DTO structs. +4. Add PostgreSQL migration and implementation. +5. Add SQLite full schema and incremental migration. Bump `SchemaVersion`. +6. Bump `RequiredSchemaVersion` for PostgreSQL. +7. Confirm migration numbering against current `origin/dev` before implementation. + +## Success Criteria + +- [ ] Store tests fail before implementation and pass after. +- [ ] Both DB backends expose the same timeline item fields. +- [ ] Sequence order is deterministic and not timestamp-dependent. +- [ ] Tool args/results have preview fields only in Phase 1. +- [ ] SQLite and PostgreSQL schema versions are both updated. + +## Todo List + +- [ ] Add failing PostgreSQL store tests. +- [ ] Add failing SQLite store tests. +- [ ] Define store interface and DTOs. +- [ ] Add PostgreSQL migration. +- [ ] Add SQLite full schema and incremental migration. +- [ ] Bump both schema version choke points. + +## Risk Assessment + +Main risk: schema drift between Standard and Lite. Mitigation: add both migrations in the same phase and compile/test `-tags sqliteonly`. + +## Security Considerations + +Timeline rows are tenant-scoped product data. All store list queries must include tenant scope. Phase 1 must not persist full tool argument/result JSON. + +## Next Steps + +Proceed to Phase 2 only after both DB backends pass contract tests. diff --git a/plans/260528-1804-archived-run-timeline/phase-02-timeline-persistence.md b/plans/260528-1804-archived-run-timeline/phase-02-timeline-persistence.md new file mode 100644 index 00000000..ef45a5f8 --- /dev/null +++ b/plans/260528-1804-archived-run-timeline/phase-02-timeline-persistence.md @@ -0,0 +1,104 @@ +--- +phase: 2 +title: "Timeline Persistence" +status: complete +priority: P1 +effort: "1.5d" +dependencies: [1] +--- + +# Phase 2: Timeline Persistence + +## Context Links + +- Event enrichment: `internal/agent/loop_run.go` +- Tool events: `internal/agent/loop_pipeline_tool_callbacks.go` +- Block replies: `internal/pipeline/think_stage.go` +- Final/block reply dedup reference: `internal/pipeline/observe_stage.go` + +## Overview + +Persist timeline items from existing agent events without changing delivery behavior. This is an event capture layer, not a new chat behavior engine. + +## Key Insights + +- The shared enriched emitter boundary already carries run, user, channel, chat, session, and tenant context. +- Parallel tool calls can emit from goroutines, so `seq` must be assigned before async persistence. +- `block.reply` already excludes final answers; this helps avoid final duplicate archive items. + +## Requirements + +- Functional: capture `run.started`, `activity`, `block.reply`, `tool.call`, `tool.result`, `run.completed`, `run.failed`, `run.cancelled`. +- Functional: link tool timeline items to trace/span IDs where the current code can provide them without invasive rewrites. +- Functional: store final answer once; avoid duplicating the last `block.reply` as final answer. +- Non-functional: best-effort persistence must not block agent execution if DB write fails. +- Non-functional: preserve tenant/session/user/channel context already attached in `AgentEvent`. + +## Architecture + +Introduce a small timeline recorder owned by the agent loop. It receives enriched `AgentEvent` values from the same emitter boundary used by live WS/channel broadcast. + +Ordering: +- Maintain per-run atomic sequence counter in the loop/request scope. +- Assign `seq` before async persistence. +- For parallel tool calls, sequence reflects event emission order, not completion timestamp. + +Privacy: +- `tool.call`: title/tool name + safe args preview. No full arguments. +- `tool.result`: status + safe result preview. No full result. +- `thinking`: redacted `activity` marker only. +- `block.reply` and final answer: content stored because user-visible. + +## Related Code Files + +- Create: `internal/agent/run_timeline_recorder.go` +- Create: `internal/agent/run_timeline_recorder_test.go` +- Modify: `internal/agent/loop_types.go` +- Modify: `internal/agent/loop_run.go` +- Modify: `internal/agent/loop_pipeline_callbacks.go` +- Modify: `internal/agent/loop_pipeline_tool_callbacks.go` +- Modify: `internal/agent/loop_tracing.go` only if span ID linkage can be added narrowly. + +## Implementation Steps + +1. Write recorder unit tests first: + - event to item mapping. + - preview truncation and redaction. + - no raw thinking. + - final answer dedup when same as last block reply. + - tool error status mapping. +2. Add a `RunTimelineStore` dependency to loop config/wiring. +3. Add recorder creation in `Loop.Run` after run context is known. +4. Wrap `emitRun` so each relevant event is recorded and then emitted live. +5. Keep recorder DB writes best-effort with bounded timeout/logging. +6. Add tests for sequential and parallel tool event ordering. +7. Verify no #67 behavior changes: no ack, no message splitting, no channel delivery toggles. + +## Success Criteria + +- [ ] Every supported event type maps to exactly one timeline item shape. +- [ ] Tool args/results are persisted as previews only. +- [ ] Final answer is not duplicated after a matching block reply. +- [ ] Recorder failures do not fail the agent run. +- [ ] Parallel tool call and result events stay ordered by assigned `seq`. + +## Todo List + +- [ ] Add recorder mapper tests. +- [ ] Add privacy/redaction tests. +- [ ] Add sequence ordering tests. +- [ ] Wire store dependency into loop config. +- [ ] Wrap enriched event emitter with recorder. +- [ ] Add best-effort persistence timeout/logging. + +## Risk Assessment + +Main risk: recording at too many call sites creates inconsistent item shapes. Mitigation: record at the shared enriched emitter boundary and keep payload normalization centralized. + +## Security Considerations + +Use the same sanitizer/redaction approach across tool call and result previews. Never persist raw thinking. Log recorder failures without dumping payloads. + +## Next Steps + +Proceed to Phase 3 after recorder tests prove stable event-to-item mapping. diff --git a/plans/260528-1804-archived-run-timeline/phase-03-http-and-ws-timeline-apis.md b/plans/260528-1804-archived-run-timeline/phase-03-http-and-ws-timeline-apis.md new file mode 100644 index 00000000..8ebb6e11 --- /dev/null +++ b/plans/260528-1804-archived-run-timeline/phase-03-http-and-ws-timeline-apis.md @@ -0,0 +1,110 @@ +--- +phase: 3 +title: "HTTP and WS Timeline APIs" +status: complete +priority: P1 +effort: "1d" +dependencies: [1, 2] +--- + +# Phase 3: HTTP and WS Timeline APIs + +## Context Links + +- HTTP trace permissions reference: `internal/http/traces.go` +- WS chat methods reference: `internal/gateway/methods/chat.go` +- API docs: `docs/18-http-api.md` +- WS docs: `docs/19-websocket-rpc.md` + +## Overview + +Expose Phase 1 archive timeline through both HTTP and WebSocket RPC. User decision: both APIs are in scope for Phase 1. + +## Key Insights + +- HTTP is useful for direct UI fetch and future export. +- WS RPC keeps parity with existing dashboard control-plane patterns. +- Both paths must use one DTO mapper to avoid response drift. + +## Requirements + +- Functional: HTTP endpoint returns timeline by run ID. +- Functional: WS RPC returns the same DTO by run ID. +- Functional: support optional session key guard when provided. +- Functional: enforce tenant/user permissions; non-admin users can only access their own run/session timeline. +- Functional: include trace ID/span ID references for admin debug, but not full trace content. +- Non-functional: response shape must be stable for web UI and future exports. + +## Architecture + +HTTP route recommendation: + +```text +GET /v1/runs/{runID}/timeline?session_key= +``` + +WS method recommendation: + +```text +run.timeline.get +params: { runId: string, sessionKey?: string } +response: { runId, sessionKey, items: RunTimelineItem[] } +``` + +DTO should be shared between HTTP and WS methods to avoid drift. + +## Related Code Files + +- Create: `internal/http/run_timeline.go` +- Create: `internal/http/run_timeline_test.go` +- Create: `internal/gateway/methods/run_timeline.go` +- Create: `internal/gateway/methods/run_timeline_test.go` +- Modify: `internal/http/server.go` or route registration file +- Modify: `internal/gateway/methods/registry.go` or method registration file +- Modify: `pkg/protocol` only if method constants live there +- Modify: `ui/web/src/api/protocol.ts` +- Modify: `docs/18-http-api.md` +- Modify: `docs/19-websocket-rpc.md` + +## Implementation Steps + +1. Write HTTP auth/permission tests first: + - admin can read tenant run timeline. + - owner/user can read own run timeline. + - other user receives not found/forbidden without leaking existence. + - invalid run ID returns localized bad request. +2. Write WS RPC tests with equivalent permission cases. +3. Add shared response DTO and mapper. +4. Register HTTP route. +5. Register WS method. +6. Add docs for both APIs. +7. Confirm endpoint names do not conflict with existing `/v1/traces` routes. + +## Success Criteria + +- [ ] HTTP and WS return identical timeline item fields. +- [ ] Permission tests cover admin, owner, and unrelated user. +- [ ] Trace/span links are IDs only. +- [ ] API docs include request and response examples. +- [ ] No public share or export API is introduced in Phase 1. + +## Todo List + +- [ ] Add HTTP permission tests. +- [ ] Add WS permission tests. +- [ ] Add shared DTO mapper. +- [ ] Register HTTP route. +- [ ] Register WS method. +- [ ] Update HTTP and WS docs. + +## Risk Assessment + +Main risk: leaking run existence across tenants/users. Mitigation: scope all queries by tenant and session/user where available, and mirror trace handler non-admin behavior. + +## Security Considerations + +Return not found for cross-tenant or unauthorized access when possible. Do not include full trace/span payloads, only IDs. + +## Next Steps + +Proceed to Phase 4 after API parity tests pass. diff --git a/plans/260528-1804-archived-run-timeline/phase-04-session-archive-ui.md b/plans/260528-1804-archived-run-timeline/phase-04-session-archive-ui.md new file mode 100644 index 00000000..814e4622 --- /dev/null +++ b/plans/260528-1804-archived-run-timeline/phase-04-session-archive-ui.md @@ -0,0 +1,107 @@ +--- +phase: 4 +title: "Session Archive UI" +status: complete +priority: P2 +effort: "1.5d" +dependencies: [3] +--- + +# Phase 4: Session Archive UI + +## Context Links + +- Live chat event handling: `ui/web/src/pages/chat/hooks/use-chat-messages.ts` +- Tool card reference: `ui/web/src/components/chat/tool-call-card.tsx` +- Session detail page: `ui/web/src/pages/sessions/session-detail-page.tsx` +- Trace detail page: `ui/web/src/pages/traces/trace-detail-dialog.tsx` + +## Overview + +Add the Phase 1 archive UI under session detail as the recommended first surface. Trace pages link out for admin/debug, but the primary user journey is reviewing a session run archive. + +## Key Insights + +- Session detail is the right Phase 1 home because users review archived conversations there. +- Trace detail is a secondary admin/debug path, not the primary archive UX. +- Current `ToolCallCard` can leak full args/results on expand, so archive UI needs preview-only rendering. + +## Requirements + +- Functional: session detail shows available runs for the session and opens a Claude-like timeline. +- Functional: render assistant messages, activity markers, combined tool cards, and final status in strict order. +- Functional: tool cards collapsed by default and show preview only. +- Functional: admin/debug users see trace/span links when available. +- Non-functional: mobile-friendly, no nested cards, no raw args/results expansion in Phase 1. +- Non-functional: preserve current live chat UI behavior. + +## Architecture + +Recommended UI placement: +- Primary: `ui/web/src/pages/sessions/session-detail-page.tsx` +- Add a run archive drawer/panel or route-scoped section from session detail. +- Use current `ToolCallCard` patterns but create archive-safe component that never displays full raw JSON in Phase 1. +- Trace detail may include a "View archive timeline" link if `run_id` exists. + +Why this recommendation: +- Users looking at old work start from sessions, not traces. +- Trace detail is developer/admin diagnostic UI. +- Public share page belongs to later phase after privacy model is proven. + +## Related Code Files + +- Create: `ui/web/src/types/run-timeline.ts` +- Create: `ui/web/src/pages/sessions/run-timeline-panel.tsx` +- Create: `ui/web/src/pages/sessions/run-timeline-item.tsx` +- Create: `ui/web/src/pages/sessions/hooks/use-run-timeline.ts` +- Create: `ui/web/src/pages/sessions/__tests__/run-timeline-panel.test.tsx` +- Modify: `ui/web/src/pages/sessions/session-detail-page.tsx` +- Modify: `ui/web/src/pages/traces/trace-detail-dialog.tsx` +- Modify: `ui/web/src/api/protocol.ts` +- Modify: `ui/web/src/i18n/locales/{en,vi,zh}/sessions.json` + +## Implementation Steps + +1. Write component tests first: + - assistant intermediate and final messages render in order. + - tool call/result combine into one collapsed card by `tool_call_id`. + - preview appears; raw args/results do not. + - trace link appears only when item has trace/span IDs and user can see debug affordance. + - mobile layout does not depend on fixed wide columns. +2. Add API hook using HTTP fetch first; use WS RPC only where current page pattern prefers WS. +3. Add run selector/list in session detail using existing session/run data where possible. +4. Add timeline panel/drawer. +5. Add trace detail link to timeline when `run_id` and `session_key` exist. +6. Add i18n keys in all three locale files before wiring UI text. +7. Verify no current chat live event rendering regresses. + +## Success Criteria + +- [ ] Session detail can open an archive timeline for a run. +- [ ] Timeline order matches backend `seq`. +- [ ] Tool previews are collapsed and safe by default. +- [ ] Trace/span links are available for admin/debug users. +- [ ] UI works on mobile widths without horizontal overflow. +- [ ] Existing live chat `ActiveRunZone` still behaves unchanged. + +## Todo List + +- [ ] Add timeline TS types. +- [ ] Add timeline fetch hook. +- [ ] Add archive-safe timeline item component. +- [ ] Add session detail panel/drawer entry. +- [ ] Add trace detail link. +- [ ] Add i18n keys for en/vi/zh. +- [ ] Add component tests. + +## Risk Assessment + +Main risk: reusing live `ToolCallCard` leaks raw arguments/results. Mitigation: introduce archive-safe rendering that accepts preview fields only. + +## Security Considerations + +Do not render raw payload fields in Phase 1. Treat trace links as debug affordances and hide them when permission data is unavailable. + +## Next Steps + +Proceed to Phase 5 after UI tests and manual mobile review pass. diff --git a/plans/260528-1804-archived-run-timeline/phase-05-validation-and-issue-handoff.md b/plans/260528-1804-archived-run-timeline/phase-05-validation-and-issue-handoff.md new file mode 100644 index 00000000..ad715bd4 --- /dev/null +++ b/plans/260528-1804-archived-run-timeline/phase-05-validation-and-issue-handoff.md @@ -0,0 +1,98 @@ +--- +phase: 5 +title: "Validation and Issue Handoff" +status: complete +priority: P1 +effort: "0.5d" +dependencies: [1, 2, 3, 4] +--- + +# Phase 5: Validation and Issue Handoff + +## Context Links + +- Plan overview: `plans/260528-1804-archived-run-timeline/plan.md` +- GitHub issue: `digitopvn/goclaw#76` +- Related issue boundary: `digitopvn/goclaw#67` + +## Overview + +Run compile/test gates, verify issue #76 acceptance criteria, and update the GitHub issue with the final implementation summary. + +## Key Insights + +- This is a cross-surface feature: store, agent loop, HTTP, WS, and UI all need verification. +- Lite/SQLite must be explicitly checked because schema drift can break desktop startup. +- #67 behavior must remain absent. + +## Requirements + +- Functional: verify backend persistence, HTTP API, WS RPC, and web UI behavior. +- Functional: confirm no #67 behavior was implemented accidentally. +- Non-functional: run both PostgreSQL and SQLite/Lite compile gates. +- Non-functional: update issue #76 with plan/implementation result and filepath. + +## Architecture + +Validation focuses on product behavior: +- timeline stored as product data. +- trace/span IDs linked for debug only. +- archive UI under session detail. +- previews only in Phase 1. + +## Related Code Files + +- Modify: `docs/18-http-api.md` +- Modify: `docs/19-websocket-rpc.md` +- Modify: `docs/project-changelog.md` if implementation lands. +- Read: `plans/260528-1804-archived-run-timeline/plan.md` + +## Implementation Steps + +1. Run focused backend tests: + - stores, HTTP handler, WS method, agent recorder. +2. Run SQLite-specific tests and build. +3. Run full Go compile/vet gates: + - `go build ./...` + - `go build -tags sqliteonly ./...` + - `go vet ./...` +4. Run web tests/build: + - `cd ui/web && pnpm test -- --run` + - `cd ui/web && pnpm build` +5. Run `git diff --check`. +6. Manually verify a real or fixture run timeline: + - block reply appears before tool call/result where emitted. + - final answer appears once. + - tool preview only. + - trace link exists when trace IDs are available. +7. Reply to GitHub issue #76 with summary, validation results, and plan/implementation filepath. + +## Success Criteria + +- [x] All planned acceptance criteria are checked against tests or manual verification. +- [x] Both HTTP and WS API paths are verified. +- [x] No quick ack or message splitting behavior from #67 is present. +- [x] GitHub issue #76 has a concise update with filepath. +- [x] Any unresolved questions are listed at the end. + +## Todo List + +- [x] Run focused backend tests. +- [x] Run SQLite/Lite tests and build. +- [x] Run Go build/vet gates. +- [x] Run web tests/build. +- [x] Run `git diff --check`. +- [x] Manually verify timeline behavior. +- [x] Reply to issue #76. + +## Risk Assessment + +Main risk: validation misses Lite/SQLite regression. Mitigation: treat `go build -tags sqliteonly ./...` and SQLite store tests as blocking. + +## Security Considerations + +Validation must include permission boundary tests and preview-only assertions. Do not paste sensitive tool payloads into GitHub issue updates. + +## Next Steps + +After implementation validation, ship via normal branch/PR flow and update issue #76 with commit/PR references. diff --git a/plans/260528-1804-archived-run-timeline/plan.md b/plans/260528-1804-archived-run-timeline/plan.md new file mode 100644 index 00000000..67582f9a --- /dev/null +++ b/plans/260528-1804-archived-run-timeline/plan.md @@ -0,0 +1,79 @@ +--- +title: "Archived Interleaved Run Timeline" +description: "TDD plan for issue #76: persist safe run timeline previews and render a Claude-like session archive, without implementing issue #67 chat delivery heuristics." +status: complete +priority: P2 +branch: "codex/issue-76-run-timeline-plan" +tags: [timeline, archive, traces, web-ui, tdd, issue-76] +blockedBy: [] +blocks: [] +created: "2026-05-28T11:04:29.547Z" +createdBy: "ck:plan" +source: skill +--- + +# Archived Interleaved Run Timeline + +## Overview + +Implement `digitopvn/goclaw#76`: persist assistant intermediate messages, tool calls, tool results, activity markers, and final run status as an ordered archive timeline. Phase 1 exposes both HTTP and WS RPC fetch APIs, stores safe previews only, and links to traces/spans for admin debug. + +Recommended Phase 1 UI: add the archive inside session detail first, with a per-run timeline panel/drawer and trace links for admin/debug. This is closest to user intent ("what happened in this run?") and avoids building a public share surface before the archive contract is proven. + +Hard boundary with related issue `#67`: this plan does not add quick acknowledgement, multi-message final splitting, channel spam controls, or delivery heuristics. It archives existing structured events; #67 can later reuse the same timeline model for live/channel behavior. + +## Current Code Context + +- Runtime event enrichment: `internal/agent/loop_run.go` +- Tool events: `internal/agent/loop_pipeline_tool_callbacks.go` +- Intermediate assistant content: `internal/pipeline/think_stage.go`, `internal/pipeline/observe_stage.go` +- Existing trace product data: `internal/store/tracing_store.go`, `internal/http/traces.go` +- Existing session messages: `sessions.messages` JSONB in `migrations/000001_init_schema.up.sql` +- Existing live UI event handling: `ui/web/src/pages/chat/hooks/use-chat-messages.ts` +- Existing tool card UI: `ui/web/src/components/chat/tool-call-card.tsx` + +## Phases + +| Phase | Name | Status | +|-------|------|--------| +| 1 | [TDD Contract and Schema](./phase-01-tdd-contract-and-schema.md) | Complete | +| 2 | [Timeline Persistence](./phase-02-timeline-persistence.md) | Complete | +| 3 | [HTTP and WS Timeline APIs](./phase-03-http-and-ws-timeline-apis.md) | Complete | +| 4 | [Session Archive UI](./phase-04-session-archive-ui.md) | Complete | +| 5 | [Validation and Issue Handoff](./phase-05-validation-and-issue-handoff.md) | Complete | + +## Dependencies + +- GitHub issue: `digitopvn/goclaw#76` +- Related non-overlap issue: `digitopvn/goclaw#67` +- No blocking plan found. Existing pending issue #80 plan targets skills export UI and does not overlap this feature. + +## Data Contract Target + +Timeline item kinds: +- `activity` +- `assistant.message` +- `tool.call` +- `tool.result` +- `run.status` + +Storage rule: +- Store previews only for tool args/results in Phase 1. +- Store assistant `block.reply` and final answer content because they are user-visible. +- Store `thinking` as redacted marker only; do not persist raw reasoning. +- Link to trace/span IDs when available for admin debug. + +## Validation Commands + +Run focused tests first, then full compile gates: + +```bash +go test ./internal/store/pg ./internal/store/sqlitestore ./internal/http ./internal/gateway/methods ./internal/agent ./internal/pipeline +go test -tags sqliteonly ./internal/store/sqlitestore ./internal/gateway/methods +go build ./... +go build -tags sqliteonly ./... +go vet ./... +cd ui/web && pnpm test -- --run +cd ui/web && pnpm build +git diff --check +``` diff --git a/ui/web/src/api/protocol.ts b/ui/web/src/api/protocol.ts index c841c5e7..0b3b1caa 100644 --- a/ui/web/src/api/protocol.ts +++ b/ui/web/src/api/protocol.ts @@ -79,6 +79,7 @@ export const Methods = { SESSIONS_PATCH: "sessions.patch", SESSIONS_DELETE: "sessions.delete", SESSIONS_RESET: "sessions.reset", + RUN_TIMELINE_GET: "run.timeline.get", // Phase 2 - NEEDED SKILLS_LIST: "skills.list", diff --git a/ui/web/src/i18n/locales/en/sessions.json b/ui/web/src/i18n/locales/en/sessions.json index a2935f64..c46c057f 100644 --- a/ui/web/src/i18n/locales/en/sessions.json +++ b/ui/web/src/i18n/locales/en/sessions.json @@ -32,7 +32,19 @@ "showDetails": "show details", "hide": "hide", "systemMessage": "System Message", - "messages": "messages" + "messages": "messages", + "timeline": { + "title": "Run timeline", + "loading": "loading", + "count": "{{count}} items", + "runStatus": "run", + "assistant": "assistant", + "toolCall": "tool call", + "toolResult": "tool result", + "trace": "trace", + "activity": "activity", + "failed": "failed" + } }, "toast": { "deleted": "Session deleted", diff --git a/ui/web/src/i18n/locales/vi/sessions.json b/ui/web/src/i18n/locales/vi/sessions.json index 66a48466..ea378575 100644 --- a/ui/web/src/i18n/locales/vi/sessions.json +++ b/ui/web/src/i18n/locales/vi/sessions.json @@ -32,7 +32,19 @@ "showDetails": "xem chi tiết", "hide": "ẩn", "systemMessage": "Tin nhắn hệ thống", - "messages": "tin nhắn" + "messages": "tin nhắn", + "timeline": { + "title": "Timeline lượt chạy", + "loading": "đang tải", + "count": "{{count}} mục", + "runStatus": "lượt chạy", + "assistant": "assistant", + "toolCall": "gọi tool", + "toolResult": "kết quả tool", + "trace": "trace", + "activity": "hoạt động", + "failed": "lỗi" + } }, "toast": { "deleted": "Đã xóa session", diff --git a/ui/web/src/i18n/locales/zh/sessions.json b/ui/web/src/i18n/locales/zh/sessions.json index 6ea76314..30214f0d 100644 --- a/ui/web/src/i18n/locales/zh/sessions.json +++ b/ui/web/src/i18n/locales/zh/sessions.json @@ -32,7 +32,19 @@ "showDetails": "显示详情", "hide": "隐藏", "systemMessage": "系统消息", - "messages": "条消息" + "messages": "条消息", + "timeline": { + "title": "运行时间线", + "loading": "加载中", + "count": "{{count}} 项", + "runStatus": "运行", + "assistant": "助手", + "toolCall": "工具调用", + "toolResult": "工具结果", + "trace": "追踪", + "activity": "活动", + "failed": "失败" + } }, "toast": { "deleted": "Session已删除", diff --git a/ui/web/src/lib/query-keys.ts b/ui/web/src/lib/query-keys.ts index 3ce184e1..268323bf 100644 --- a/ui/web/src/lib/query-keys.ts +++ b/ui/web/src/lib/query-keys.ts @@ -23,6 +23,7 @@ export const queryKeys = { sessions: { all: ["sessions"] as const, list: (params: Record) => ["sessions", params] as const, + timeline: (params: Record) => ["sessions", "timeline", params] as const, }, traces: { all: ["traces"] as const, diff --git a/ui/web/src/pages/sessions/hooks/use-run-timeline.ts b/ui/web/src/pages/sessions/hooks/use-run-timeline.ts new file mode 100644 index 00000000..68aec52d --- /dev/null +++ b/ui/web/src/pages/sessions/hooks/use-run-timeline.ts @@ -0,0 +1,31 @@ +import { useQuery } from "@tanstack/react-query"; +import { Methods } from "@/api/protocol"; +import { useWs } from "@/hooks/use-ws"; +import { useAuthStore } from "@/stores/use-auth-store"; +import { queryKeys } from "@/lib/query-keys"; +import type { RunTimelineResponse } from "@/types/run-timeline"; + +interface UseRunTimelineOptions { + runId?: string; + sessionKey?: string; + limit?: number; +} + +export function useRunTimeline({ runId, sessionKey, limit = 100 }: UseRunTimelineOptions) { + const ws = useWs(); + const connected = useAuthStore((s) => s.connected); + + return useQuery({ + queryKey: queryKeys.sessions.timeline({ runId, sessionKey, limit }), + queryFn: async () => { + if (!ws.isConnected) return { items: [], limit, offset: 0 } satisfies RunTimelineResponse; + return ws.call(Methods.RUN_TIMELINE_GET, { + runId, + sessionKey, + limit, + }); + }, + staleTime: 10_000, + enabled: connected && Boolean(runId || sessionKey), + }); +} diff --git a/ui/web/src/pages/sessions/run-timeline-panel.test.ts b/ui/web/src/pages/sessions/run-timeline-panel.test.ts new file mode 100644 index 00000000..804d1def --- /dev/null +++ b/ui/web/src/pages/sessions/run-timeline-panel.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from "vitest"; +import { getRunTimelineDisplay } from "./run-timeline-panel"; + +describe("getRunTimelineDisplay", () => { + it("marks failed items as destructive", () => { + const display = getRunTimelineDisplay({ item_type: "tool.result", status: "failed" }); + expect(display.labelKey).toBe("detail.timeline.failed"); + expect(display.dotClass).toContain("destructive"); + }); + + it("distinguishes tool call and assistant message items", () => { + expect(getRunTimelineDisplay({ item_type: "tool.call", status: "running" }).labelKey).toBe( + "detail.timeline.toolCall", + ); + expect(getRunTimelineDisplay({ item_type: "assistant.message", status: "completed" }).labelKey).toBe( + "detail.timeline.assistant", + ); + }); +}); diff --git a/ui/web/src/pages/sessions/run-timeline-panel.tsx b/ui/web/src/pages/sessions/run-timeline-panel.tsx new file mode 100644 index 00000000..f1e4a9f6 --- /dev/null +++ b/ui/web/src/pages/sessions/run-timeline-panel.tsx @@ -0,0 +1,146 @@ +import { useState } from "react"; +import { useTranslation } from "react-i18next"; +import { Link } from "react-router"; +import { + Activity, + CheckCircle2, + ChevronDown, + ChevronRight, + Clock, + ExternalLink, + MessageSquareText, + Wrench, + XCircle, +} from "lucide-react"; +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { formatDate } from "@/lib/format"; +import type { RunTimelineItem, RunTimelineItemType } from "@/types/run-timeline"; + +interface RunTimelinePanelProps { + items: RunTimelineItem[]; + loading?: boolean; +} + +export function RunTimelinePanel({ items, loading }: RunTimelinePanelProps) { + const { t } = useTranslation("sessions"); + const [open, setOpen] = useState(false); + + if (!loading && items.length === 0) return null; + + return ( +
+ + + {open && ( +
+ {items.map((item, index) => { + const display = getRunTimelineDisplay(item); + const Icon = display.icon; + return ( +
+
+
+ +
+ {index < items.length - 1 &&
} +
+
+
+ + {t(display.labelKey)} + + {item.title || t(display.labelKey)} + {item.status && {item.status}} +
+ {item.preview && ( + 
+                      {item.preview}
+
+ )} +
+ {formatDate(item.created_at)} + {item.tool_call_id && {item.tool_call_id}} + {item.trace_id && ( + + )} + {item.run_id && {item.run_id}} +
+
+
+ ); + })} +
+ )} +
+ ); +} + +export function getRunTimelineDisplay(item: Pick) { + if (item.status === "failed") { + return { + icon: XCircle, + dotClass: "bg-destructive/10 text-destructive", + labelKey: "detail.timeline.failed", + }; + } + if (item.item_type === "tool.call" || item.item_type === "tool.result") { + return { + icon: Wrench, + dotClass: "bg-amber-500/10 text-amber-700 dark:text-amber-300", + labelKey: item.item_type === "tool.call" ? "detail.timeline.toolCall" : "detail.timeline.toolResult", + }; + } + if (item.item_type === "assistant.message") { + return { + icon: MessageSquareText, + dotClass: "bg-blue-500/10 text-blue-700 dark:text-blue-300", + labelKey: "detail.timeline.assistant", + }; + } + if (item.item_type === "activity") { + return { + icon: Activity, + dotClass: "bg-primary/10 text-primary", + labelKey: "detail.timeline.activity", + }; + } + return { + icon: item.status === "completed" ? CheckCircle2 : Clock, + dotClass: statusDotClass(item.status), + labelKey: "detail.timeline.runStatus", + }; +} + +function statusDotClass(status: RunTimelineItem["status"]) { + switch (status) { + case "completed": + return "bg-emerald-500/10 text-emerald-700 dark:text-emerald-300"; + case "cancelled": + return "bg-muted text-muted-foreground"; + case "running": + case "started": + return "bg-primary/10 text-primary"; + default: + return "bg-muted text-muted-foreground"; + } +} + +export type { RunTimelineItemType }; diff --git a/ui/web/src/pages/sessions/session-detail-page.tsx b/ui/web/src/pages/sessions/session-detail-page.tsx index 04eee12f..6d40a97b 100644 --- a/ui/web/src/pages/sessions/session-detail-page.tsx +++ b/ui/web/src/pages/sessions/session-detail-page.tsx @@ -14,6 +14,8 @@ import type { SessionInfo, SessionPreview, Message } from "@/types/session"; import type { ChatMessage, AgentEventPayload, ToolStreamEntry } from "@/types/chat"; import { messageToTimestamp } from "@/lib/message-utils"; import { SystemMessageBlock, SummaryBlock } from "./session-message-blocks"; +import { useRunTimeline } from "./hooks/use-run-timeline"; +import { RunTimelinePanel } from "./run-timeline-panel"; /** Check if a message is an internal system message (subagent results, cron, etc.) */ function isSystemMessage(msg: ChatMessage): boolean { @@ -61,6 +63,11 @@ export function SessionDetailPage({ const [titleDraft, setTitleDraft] = useState(""); const parsed = parseSessionKey(session.key); + const { + data: runTimelineData, + isFetching: runTimelineFetching, + refetch: refetchRunTimeline, + } = useRunTimeline({ sessionKey: session.key, limit: 100 }); const loadMessages = useCallback(() => { onPreview(session.key) @@ -119,12 +126,14 @@ export function SessionDetailPage({ if (!event) return; if ( (event.type === "run.completed" || event.type === "run.failed" || event.type === "run.cancelled") && - event.agentId === parsed.agentId + event.agentId === parsed.agentId && + (!event.sessionKey || event.sessionKey === session.key) ) { debouncedRefresh(); + void refetchRunTimeline(); } }, - [debouncedRefresh, parsed.agentId], + [debouncedRefresh, parsed.agentId, refetchRunTimeline, session.key], ); useWsEvent(Events.AGENT, handleAgentEvent); @@ -220,6 +229,8 @@ export function SessionDetailPage({ )} + + {/* Messages */}
{loading && messages.length === 0 ? ( @@ -271,4 +282,3 @@ export function SessionDetailPage({
); } - diff --git a/ui/web/src/types/run-timeline.ts b/ui/web/src/types/run-timeline.ts new file mode 100644 index 00000000..60f29b2b --- /dev/null +++ b/ui/web/src/types/run-timeline.ts @@ -0,0 +1,36 @@ +export type RunTimelineItemType = + | "activity" + | "assistant.message" + | "tool.call" + | "tool.result" + | "run.status"; + +export interface RunTimelineItem { + id: string; + tenant_id?: string; + run_id: string; + session_key: string; + agent_id?: string; + user_id?: string; + channel?: string; + chat_id?: string; + seq: number; + item_type: RunTimelineItemType; + status?: "started" | "running" | "completed" | "failed" | "cancelled" | string; + title?: string; + preview?: string; + tool_name?: string; + tool_call_id?: string; + trace_id?: string; + span_id?: string; + metadata?: Record; + created_at: string; +} + +export interface RunTimelineResponse { + runId?: string; + sessionKey?: string; + items: RunTimelineItem[]; + limit: number; + offset: number; +}