diff --git a/internal/agent/loop_cancel_test.go b/internal/agent/loop_cancel_test.go index e31ad470..731758af 100644 --- a/internal/agent/loop_cancel_test.go +++ b/internal/agent/loop_cancel_test.go @@ -18,7 +18,7 @@ func TestParallelToolCollection_ContextCancel(t *testing.T) { var err error collectLoop: - for i := 0; i < 3; i++ { + for range 3 { select { case r, ok := <-resultCh: if !ok { @@ -50,7 +50,7 @@ func TestParallelToolCollection_AllComplete(t *testing.T) { collected := make([]indexedResult, 0, 3) collectLoop: - for i := 0; i < 3; i++ { + for range 3 { select { case r, ok := <-resultCh: if !ok { diff --git a/internal/agent/loop_history.go b/internal/agent/loop_history.go index 90e39cf3..260ea925 100644 --- a/internal/agent/loop_history.go +++ b/internal/agent/loop_history.go @@ -438,10 +438,7 @@ func (l *Loop) resolveSkillsSummary(ctx context.Context, skillFilter []string) s // Cap description length to match BuildSummary() truncation (skillDescMaxLen=200 runes). totalChars := 0 for _, s := range filtered { - descLen := len(s.Description) - if descLen > 200 { - descLen = 200 - } + descLen := min(len(s.Description), 200) totalChars += len(s.Name) + descLen + 10 // +10 for XML tags overhead } estimatedTokens := totalChars / 4 diff --git a/internal/agent/loop_run.go b/internal/agent/loop_run.go index 37f80372..d308a9f0 100644 --- a/internal/agent/loop_run.go +++ b/internal/agent/loop_run.go @@ -118,6 +118,25 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { runStart := time.Now().UTC() + // Safety net: ensure root traces are ALWAYS finalized, even on panic or goroutine leak. + // Normal-path finalization sets traceFinalized=true; this defer only acts if it wasn't. + var traceFinalized bool + if !isChildTrace && l.traceCollector != nil && traceID != uuid.Nil { + defer func() { + if traceFinalized { + return + } + slog.Warn("tracing: safety-net finalizing orphan trace", + "trace_id", traceID, "agent", l.id, "session", req.SessionKey) + safeCtx := context.WithoutCancel(ctx) + if agentSpanID != uuid.Nil { + l.emitAgentSpanEnd(safeCtx, agentSpanID, runStart, nil, context.Canceled) + } + l.traceCollector.FinishTrace(safeCtx, traceID, store.TraceStatusError, + "trace finalized by safety net (likely panic or goroutine leak)", "") + }() + } + // Emit running agent span immediately so it's visible in the trace UI. if agentSpanID != uuid.Nil { var agentSpanOpts []spanOption @@ -183,6 +202,7 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { // Use background context when the run context is cancelled (/stop command) // so the DB update still succeeds. if !isChildTrace && l.traceCollector != nil && traceID != uuid.Nil { + traceFinalized = true traceCtx := ctx traceStatus := store.TraceStatusError if ctx.Err() != nil { @@ -214,6 +234,7 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) { Payload: completedPayload, }) if !isChildTrace && l.traceCollector != nil && traceID != uuid.Nil { + traceFinalized = true if result != nil { l.traceCollector.FinishTrace(ctx, traceID, store.TraceStatusCompleted, "", truncateStr(result.Content, l.traceCollector.PreviewMaxLen())) } else { diff --git a/internal/agent/pruning.go b/internal/agent/pruning.go index bc5793be..7a5025ec 100644 --- a/internal/agent/pruning.go +++ b/internal/agent/pruning.go @@ -296,10 +296,7 @@ func estimateMessageChars(m providers.Message) int { // hasImportantTail checks if the last ~500 chars of content contain error/summary keywords. func hasImportantTail(content string) bool { runes := []rune(content) - checkLen := 500 - if checkLen > len(runes) { - checkLen = len(runes) - } + checkLen := min(500, len(runes)) tail := string(runes[len(runes)-checkLen:]) return importantTailRe.MatchString(tail) } diff --git a/internal/providers/claude_cli_chat.go b/internal/providers/claude_cli_chat.go index cb8ec571..a1278c11 100644 --- a/internal/providers/claude_cli_chat.go +++ b/internal/providers/claude_cli_chat.go @@ -112,6 +112,7 @@ func (p *ClaudeCLIProvider) ChatStream(ctx context.Context, req ChatRequest, onC } cmd := exec.CommandContext(ctx, p.cliPath, args...) + cmd.WaitDelay = 5 * time.Second // force-close pipes if process lingers after kill cmd.Dir = workDir cmd.Env = filterCLIEnv(os.Environ()) if stdin != nil { @@ -151,6 +152,9 @@ func (p *ClaudeCLIProvider) ChatStream(ctx context.Context, req ChatRequest, onC var contentBuf strings.Builder for scanner.Scan() { + if ctx.Err() != nil { + break // context cancelled (abort) → exit immediately + } line := scanner.Bytes() if len(line) == 0 { continue @@ -201,6 +205,12 @@ func (p *ClaudeCLIProvider) ChatStream(ctx context.Context, req ChatRequest, onC } } + // Context cancelled (abort): best-effort reap (bounded by WaitDelay), then return. + if ctx.Err() != nil { + _ = cmd.Wait() + return nil, ctx.Err() + } + if err := scanner.Err(); err != nil { return nil, fmt.Errorf("claude-cli: stream read error: %w", err) } diff --git a/internal/store/pg/tracing.go b/internal/store/pg/tracing.go index 706a4925..545d19a3 100644 --- a/internal/store/pg/tracing.go +++ b/internal/store/pg/tracing.go @@ -564,3 +564,26 @@ func (s *PGTracingStore) DeleteTracesOlderThan(ctx context.Context, cutoff time. } return res.RowsAffected() } + +// RecoverStaleRunningTraces marks traces stuck in "running" since before cutoff as "error". +// Also recovers their stuck spans. Called on startup to fix orphans from crashes. +func (s *PGTracingStore) RecoverStaleRunningTraces(ctx context.Context, cutoff time.Time) (int64, error) { + // Recover stuck spans first. + _, err := s.db.ExecContext(ctx, + `UPDATE spans SET status = 'error', error = 'recovered: server restart', + end_time = NOW(), duration_ms = EXTRACT(EPOCH FROM (NOW() - start_time))::int * 1000 + WHERE status = 'running' AND start_time < $1`, cutoff) + if err != nil { + return 0, fmt.Errorf("recover stale spans: %w", err) + } + + res, err := s.db.ExecContext(ctx, + `UPDATE traces SET status = 'error', + error = 'recovered: stuck in running state (server restart)', + end_time = NOW(), duration_ms = EXTRACT(EPOCH FROM (NOW() - start_time))::int * 1000 + WHERE status = 'running' AND start_time < $1`, cutoff) + if err != nil { + return 0, fmt.Errorf("recover stale running traces: %w", err) + } + return res.RowsAffected() +} diff --git a/internal/store/sqlitestore/tracing.go b/internal/store/sqlitestore/tracing.go index c0947893..e7be4f19 100644 --- a/internal/store/sqlitestore/tracing.go +++ b/internal/store/sqlitestore/tracing.go @@ -251,6 +251,29 @@ func (s *SQLiteTracingStore) DeleteTracesOlderThan(ctx context.Context, cutoff t return res.RowsAffected() } +// RecoverStaleRunningTraces marks traces stuck in "running" since before cutoff as "error". +// Also recovers their stuck spans. Called on startup to fix orphans from crashes. +func (s *SQLiteTracingStore) RecoverStaleRunningTraces(ctx context.Context, cutoff time.Time) (int64, error) { + // Recover stuck spans first. + _, err := s.db.ExecContext(ctx, + `UPDATE spans SET status = 'error', error = 'recovered: server restart', + end_time = datetime('now'), duration_ms = CAST((julianday('now') - julianday(start_time)) * 86400000 AS INTEGER) + WHERE status = 'running' AND start_time < ?`, cutoff) + if err != nil { + return 0, fmt.Errorf("recover stale spans: %w", err) + } + + res, err := s.db.ExecContext(ctx, + `UPDATE traces SET status = 'error', + error = 'recovered: stuck in running state (server restart)', + end_time = datetime('now'), duration_ms = CAST((julianday('now') - julianday(start_time)) * 86400000 AS INTEGER) + WHERE status = 'running' AND start_time < ?`, cutoff) + if err != nil { + return 0, fmt.Errorf("recover stale running traces: %w", err) + } + return res.RowsAffected() +} + // ListCodexPoolSpans is not supported in SQLite (Codex pool is a standard-edition feature). func (s *SQLiteTracingStore) ListCodexPoolSpans(_ context.Context, _, _ uuid.UUID, _ []string, _ int) ([]store.CodexPoolSpan, error) { return nil, nil diff --git a/internal/store/tracing_store.go b/internal/store/tracing_store.go index 961028e1..93a2af04 100644 --- a/internal/store/tracing_store.go +++ b/internal/store/tracing_store.go @@ -159,6 +159,9 @@ type TracingStore interface { // Maintenance DeleteTracesOlderThan(ctx context.Context, cutoff time.Time) (int64, error) + // RecoverStaleRunningTraces marks traces stuck in "running" since before cutoff as "error". + // Returns count of recovered traces. Called on startup to fix orphans from crashes. + RecoverStaleRunningTraces(ctx context.Context, cutoff time.Time) (int64, error) // ListCodexPoolSpans returns recent LLM call spans for agents using Codex OAuth pool providers. ListCodexPoolSpans(ctx context.Context, agentID, tenantID uuid.UUID, poolProviders []string, limit int) ([]CodexPoolSpan, error) diff --git a/internal/tracing/collector.go b/internal/tracing/collector.go index a6ba4ec2..5b84c3f5 100644 --- a/internal/tracing/collector.go +++ b/internal/tracing/collector.go @@ -101,6 +101,7 @@ func (c *Collector) SetExporter(exp SpanExporter) { func (c *Collector) Start() { c.wg.Add(1) go c.flushLoop() + go c.recoverStaleTraces() // fix orphan traces from previous crash slog.Info("tracing collector started") } @@ -224,6 +225,26 @@ func (c *Collector) flushLoop() { } } +// recoverStaleTraces marks "running" traces older than 30 min as "error". +// Called once on startup to fix orphans left by a previous crash or stuck goroutine. +func (c *Collector) recoverStaleTraces() { + const staleThreshold = 30 * time.Minute + cutoff := time.Now().UTC().Add(-staleThreshold) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + recovered, err := c.store.RecoverStaleRunningTraces(ctx, cutoff) + if err != nil { + slog.Warn("tracing: failed to recover stale running traces", "error", err) + return + } + if recovered > 0 { + slog.Info("tracing: recovered stale running traces on startup", + "count", recovered, "older_than", cutoff.Format(time.RFC3339)) + } +} + // pruneOldTraces deletes traces and spans older than traceRetention. func (c *Collector) pruneOldTraces() { cutoff := time.Now().UTC().Add(-traceRetention) diff --git a/pkg/browser/browser.go b/pkg/browser/browser.go index c6ee4146..6a51dac7 100644 --- a/pkg/browser/browser.go +++ b/pkg/browser/browser.go @@ -15,6 +15,7 @@ import ( type Manager struct { mu sync.Mutex browser *rod.Browser + launcher *launcher.Launcher // retained for PID-based cleanup on crash refs *RefStore pages map[string]*rod.Page // targetID → page console map[string][]ConsoleMessage // targetID → console messages @@ -107,12 +108,7 @@ func (m *Manager) Start(ctx context.Context) error { } // Connection dead — clean up and reconnect m.logger.Info("browser connection lost, reconnecting") - m.closeTenantContextsLocked() - m.browser = nil - m.pages = make(map[string]*rod.Page) - m.console = make(map[string][]ConsoleMessage) - m.pageTenants = make(map[string]string) - m.refs = NewRefStore() + m.cleanupDeadBrowserLocked() } var controlURL string @@ -126,23 +122,45 @@ func (m *Manager) Start(ctx context.Context) error { controlURL = u m.logger.Info("connecting to remote Chrome", "cdp", controlURL, "remote", m.remoteURL) } else { - // Local Chrome — launch via rod launcher + // Local Chrome — launch via rod launcher with stability flags + launchCtx, launchCancel := context.WithTimeout(ctx, 30*time.Second) + defer launchCancel() + l := launcher.New(). + Context(launchCtx). + Leakless(true). Headless(m.headless). Set("disable-gpu"). Set("no-first-run"). - Set("no-default-browser-check") + Set("no-default-browser-check"). + Set("disable-dev-shm-usage"). + Set("disable-software-rasterizer"). + Set("disable-extensions"). + Set("disable-background-networking"). + Set("disable-renderer-backgrounding"). + Set("disable-background-timer-throttling"). + Set("disable-backgrounding-occluded-windows") u, err := l.Launch() if err != nil { return fmt.Errorf("launch Chrome: %w", err) } controlURL = u - m.logger.Info("Chrome launched", "cdp", controlURL, "headless", m.headless) + m.launcher = l + m.logger.Info("Chrome launched", "cdp", controlURL, "headless", m.headless, "pid", l.PID()) } - b := rod.New().ControlURL(controlURL) + connectCtx, connectCancel := context.WithTimeout(ctx, 15*time.Second) + defer connectCancel() + + b := rod.New().Context(connectCtx).ControlURL(controlURL) if err := b.Connect(); err != nil { + // If local launch succeeded but connect failed, kill the orphan process + if m.launcher != nil { + m.launcher.Kill() + m.launcher.Cleanup() + m.launcher = nil + } return fmt.Errorf("connect to Chrome: %w", err) } @@ -182,6 +200,12 @@ func (m *Manager) Stop(ctx context.Context) error { if m.remoteURL == "" { // Local Chrome — close the browser process err = m.browser.Close() + // Force-kill via launcher if retained + if m.launcher != nil { + m.launcher.Kill() + m.launcher.Cleanup() + m.launcher = nil + } } // Remote Chrome — just drop the connection; sidecar stays alive @@ -203,6 +227,23 @@ func (m *Manager) closeTenantContextsLocked() { m.tenantCtxs = make(map[string]*rod.Browser) } +// cleanupDeadBrowserLocked resets all state and kills any orphan Chrome process. +// Must be called with mu held. +func (m *Manager) cleanupDeadBrowserLocked() { + m.closeTenantContextsLocked() + if m.launcher != nil { + m.launcher.Kill() + m.launcher.Cleanup() + m.launcher = nil + } + m.browser = nil + m.pages = make(map[string]*rod.Page) + m.console = make(map[string][]ConsoleMessage) + m.pageTenants = make(map[string]string) + m.pageLastUsed = make(map[string]time.Time) + m.refs = NewRefStore() +} + // MasterTenantID is the well-known master tenant UUID string. // Pages opened without a tenant context or by the master tenant use the main browser directly. const MasterTenantID = "0193a5b0-7000-7000-8000-000000000001" diff --git a/ui/web/src/i18n/locales/en/agents.json b/ui/web/src/i18n/locales/en/agents.json index a9067358..a401060b 100644 --- a/ui/web/src/i18n/locales/en/agents.json +++ b/ui/web/src/i18n/locales/en/agents.json @@ -161,7 +161,6 @@ "extraHint": "Only enabled and signed-in OpenAI Codex OAuth aliases can participate.", "emptyExtras": "Sign in another OpenAI Codex OAuth alias to unlock pooling.", "noReadyExtras": "Sign in and enable another OpenAI Codex OAuth alias to add it here.", - "clickToAdd": "Click to add", "extraSelectableHint": "Only enabled, signed-in OpenAI Codex OAuth aliases can enter the active pool.", "selectedAccountsLabel": "Accounts in Pool", "emptySelected": "No extra pool members selected yet.", @@ -607,10 +606,14 @@ "userIdPlaceholder": "Enter user ID (e.g. telegram:386246614)", "warning": "When sharing is enabled, all shared users can read and write files in the same workspace directory. Memory sharing is controlled separately. Context files remain isolated per user.", "memoryGroupLabel": "Memory & Knowledge Graph", + "memoryGroupDescription": "Control memory and knowledge graph isolation between users", "folderGroupLabel": "Workspace Folders", - "shareMemory": "Shared Memory & KG", - "shareMemoryTip": "All users access the same memory and knowledge graph. Independent of workspace folder sharing. When off, each user has their own isolated memory space.", - "shareMemoryNote": "Toggling this does not migrate data — existing per-user memories become inaccessible in shared mode." + "shareMemory": "Shared Memory", + "shareMemoryTip": "All users access the same memory store. When off, each user has their own isolated memory space. KG sharing is controlled separately below.", + "shareMemoryNote": "Toggling this does not migrate data — existing per-user memories become inaccessible in shared mode.", + "shareKG": "Shared Knowledge Graph", + "shareKGTip": "All users access the same knowledge graph. When off, each user has their own isolated KG with global canonical fallback for read operations.", + "shareKGNote": "Toggling this does not migrate data — existing per-user KG entities become inaccessible in shared mode." }, "compaction": { "title": "Compaction", diff --git a/ui/web/src/i18n/locales/vi/agents.json b/ui/web/src/i18n/locales/vi/agents.json index bf57ebb9..2a1a042b 100644 --- a/ui/web/src/i18n/locales/vi/agents.json +++ b/ui/web/src/i18n/locales/vi/agents.json @@ -161,7 +161,6 @@ "extraHint": "Chỉ các bí danh OpenAI Codex OAuth đã bật và đã đăng nhập mới được tham gia.", "emptyExtras": "Đăng nhập thêm một bí danh OpenAI Codex OAuth để bật pooling.", "noReadyExtras": "Đăng nhập và bật thêm một bí danh OpenAI Codex OAuth để thêm vào đây.", - "clickToAdd": "Bấm để thêm", "extraSelectableHint": "Chỉ các bí danh OpenAI Codex OAuth đã bật và đã đăng nhập mới được vào active pool.", "selectedAccountsLabel": "Tài khoản trong pool", "emptySelected": "Chưa chọn pool member bổ sung nào.", @@ -607,10 +606,14 @@ "userIdPlaceholder": "Nhập user ID (vd: telegram:386246614)", "warning": "Khi bật chia sẻ, tất cả người dùng được chia sẻ có thể đọc và ghi file trong cùng thư mục workspace. Chia sẻ bộ nhớ được cài đặt riêng. File ngữ cảnh vẫn được cô lập theo từng người dùng.", "memoryGroupLabel": "Memory & Knowledge Graph", + "memoryGroupDescription": "Kiểm soát cô lập memory và knowledge graph giữa các người dùng", "folderGroupLabel": "Thư mục Workspace", - "shareMemory": "Chia sẻ Memory & KG", - "shareMemoryTip": "Tất cả người dùng truy cập chung Memory và Knowledge Graph. Độc lập với việc chia sẻ thư mục workspace. Khi tắt, mỗi người dùng có vùng nhớ riêng.", - "shareMemoryNote": "Bật/tắt không di chuyển dữ liệu — memory per-user trước đó sẽ không truy cập được trong chế độ shared." + "shareMemory": "Chia sẻ Memory", + "shareMemoryTip": "Tất cả người dùng truy cập chung Memory. Khi tắt, mỗi người dùng có vùng nhớ riêng. Chia sẻ KG được cài đặt riêng bên dưới.", + "shareMemoryNote": "Bật/tắt không di chuyển dữ liệu — memory per-user trước đó sẽ không truy cập được trong chế độ shared.", + "shareKG": "Chia sẻ Knowledge Graph", + "shareKGTip": "Tất cả người dùng truy cập chung Knowledge Graph. Khi tắt, mỗi người dùng có KG riêng với fallback đọc từ global canonical.", + "shareKGNote": "Bật/tắt không di chuyển dữ liệu — KG per-user trước đó sẽ không truy cập được trong chế độ shared." }, "compaction": { "title": "Nén ngữ cảnh", diff --git a/ui/web/src/i18n/locales/zh/agents.json b/ui/web/src/i18n/locales/zh/agents.json index 3277cba5..422eb1c3 100644 --- a/ui/web/src/i18n/locales/zh/agents.json +++ b/ui/web/src/i18n/locales/zh/agents.json @@ -161,7 +161,6 @@ "extraHint": "只有已启用且已登录的 OpenAI Codex OAuth 别名才能参与。", "emptyExtras": "再登录一个 OpenAI Codex OAuth 别名即可启用 pooling。", "noReadyExtras": "请先登录并启用另一个 OpenAI Codex OAuth 别名,然后再添加到这里。", - "clickToAdd": "点击添加", "extraSelectableHint": "只有已启用且已登录的 OpenAI Codex OAuth 别名才能进入 active pool。", "selectedAccountsLabel": "池中的账户", "emptySelected": "还没有选择额外的池成员。", @@ -607,10 +606,14 @@ "userIdPlaceholder": "输入用户ID(例如 telegram:386246614)", "warning": "启用共享后,所有共享用户可以读写同一工作区目录中的文件。记忆共享单独控制。上下文文件仍按用户隔离。", "memoryGroupLabel": "记忆与知识图谱", + "memoryGroupDescription": "控制用户之间的记忆和知识图谱隔离", "folderGroupLabel": "工作区文件夹", - "shareMemory": "共享记忆与知识图谱", - "shareMemoryTip": "所有用户访问相同的记忆和知识图谱。与工作区文件夹共享无关。关闭时,每个用户拥有独立的记忆空间。", - "shareMemoryNote": "切换不会迁移数据——切换到共享模式后,现有的用户独立记忆将无法访问。" + "shareMemory": "共享记忆", + "shareMemoryTip": "所有用户访问相同的记忆存储。关闭时,每个用户拥有独立的记忆空间。知识图谱共享在下方单独控制。", + "shareMemoryNote": "切换不会迁移数据——切换到共享模式后,现有的用户独立记忆将无法访问。", + "shareKG": "共享知识图谱", + "shareKGTip": "所有用户访问相同的知识图谱。关闭时,每个用户拥有独立的知识图谱,读取操作会回退到全局规范数据。", + "shareKGNote": "切换不会迁移数据——切换到共享模式后,现有的用户独立知识图谱实体将无法访问。" }, "compaction": { "title": "压缩", diff --git a/ui/web/src/pages/agents/agent-detail/config-sections/workspace-sharing-section.tsx b/ui/web/src/pages/agents/agent-detail/config-sections/workspace-sharing-section.tsx index b00a0e8d..450a11a6 100644 --- a/ui/web/src/pages/agents/agent-detail/config-sections/workspace-sharing-section.tsx +++ b/ui/web/src/pages/agents/agent-detail/config-sections/workspace-sharing-section.tsx @@ -52,23 +52,40 @@ export function WorkspaceSharingSection({ value, onChange }: WorkspaceSharingSec

{t(`${s}.memoryGroupLabel`)}

-

{t(`${s}.shareMemoryTip`)}

+

{t(`${s}.memoryGroupDescription`)}

-
-
- {t(`${s}.shareMemory`)} - onChange({ ...value, share_memory: v })} - /> +
+
+
+ {t(`${s}.shareMemory`)} + onChange({ ...value, share_memory: v })} + /> +
+ {value.share_memory && ( +

+ {t(`${s}.shareMemoryNote`)} +

+ )} +
+ +
+
+ {t(`${s}.shareKG`)} + onChange({ ...value, share_knowledge_graph: v })} + /> +
+ {value.share_knowledge_graph && ( +

+ {t(`${s}.shareKGNote`)} +

+ )}
- {value.share_memory && ( -

- {t(`${s}.shareMemoryNote`)} -

- )}
diff --git a/ui/web/src/types/agent.ts b/ui/web/src/types/agent.ts index 375eef0b..d3ab9ef6 100644 --- a/ui/web/src/types/agent.ts +++ b/ui/web/src/types/agent.ts @@ -74,6 +74,7 @@ export interface WorkspaceSharingConfig { shared_group?: boolean; shared_users?: string[]; share_memory?: boolean; + share_knowledge_graph?: boolean; } export type ChatGPTOAuthRoutingStrategy =