feat(infra): tracing recovery, browser cleanup, CLI fixes, UI workspace split (#709)

- Tracing: recover stale running traces/spans on startup (PG + SQLite)
- Browser: Chrome orphan cleanup via launcher PID, timeouts, Leakless
- Claude CLI: WaitDelay 5s + context-cancel early exit
- Agent loop: safety-net defer to finalize orphan root traces
- UI: split workspace sharing into separate Memory and KG toggles
- Minor: for-range idiom, min() builtin
This commit is contained in:
viettranx
2026-04-05 21:32:59 +07:00
parent 7d7b716074
commit 41e6c8f5cc
15 changed files with 208 additions and 45 deletions
+2 -2
View File
@@ -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 {
+1 -4
View File
@@ -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
+21
View File
@@ -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 {
+1 -4
View File
@@ -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)
}
+10
View File
@@ -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)
}
+23
View File
@@ -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()
}
+23
View File
@@ -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
+3
View File
@@ -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)
+21
View File
@@ -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)
+51 -10
View File
@@ -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"
+7 -4
View File
@@ -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",
+7 -4
View File
@@ -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",
+7 -4
View File
@@ -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": "压缩",
@@ -52,23 +52,40 @@ export function WorkspaceSharingSection({ value, onChange }: WorkspaceSharingSec
</div>
<div>
<h3 className="text-sm font-semibold">{t(`${s}.memoryGroupLabel`)}</h3>
<p className="text-xs text-muted-foreground">{t(`${s}.shareMemoryTip`)}</p>
<p className="text-xs text-muted-foreground">{t(`${s}.memoryGroupDescription`)}</p>
</div>
</div>
<div className={`rounded-lg border p-3 sm:p-4 ${value.share_memory ? "border-orange-400/60 bg-orange-50/30 dark:border-orange-500/30 dark:bg-orange-950/10" : ""}`}>
<div className="flex items-center justify-between">
<InfoLabel tip={t(`${s}.shareMemoryTip`)}>{t(`${s}.shareMemory`)}</InfoLabel>
<Switch
checked={value.share_memory ?? false}
onCheckedChange={(v) => onChange({ ...value, share_memory: v })}
/>
<div className="space-y-2">
<div className={`rounded-lg border p-3 sm:p-4 ${value.share_memory ? "border-orange-400/60 bg-orange-50/30 dark:border-orange-500/30 dark:bg-orange-950/10" : ""}`}>
<div className="flex items-center justify-between">
<InfoLabel tip={t(`${s}.shareMemoryTip`)}>{t(`${s}.shareMemory`)}</InfoLabel>
<Switch
checked={value.share_memory ?? false}
onCheckedChange={(v) => onChange({ ...value, share_memory: v })}
/>
</div>
{value.share_memory && (
<p className="mt-2 text-xs text-orange-600 dark:text-orange-400">
{t(`${s}.shareMemoryNote`)}
</p>
)}
</div>
<div className={`rounded-lg border p-3 sm:p-4 ${value.share_knowledge_graph ? "border-orange-400/60 bg-orange-50/30 dark:border-orange-500/30 dark:bg-orange-950/10" : ""}`}>
<div className="flex items-center justify-between">
<InfoLabel tip={t(`${s}.shareKGTip`)}>{t(`${s}.shareKG`)}</InfoLabel>
<Switch
checked={value.share_knowledge_graph ?? false}
onCheckedChange={(v) => onChange({ ...value, share_knowledge_graph: v })}
/>
</div>
{value.share_knowledge_graph && (
<p className="mt-2 text-xs text-orange-600 dark:text-orange-400">
{t(`${s}.shareKGNote`)}
</p>
)}
</div>
{value.share_memory && (
<p className="mt-2 text-xs text-orange-600 dark:text-orange-400">
{t(`${s}.shareMemoryNote`)}
</p>
)}
</div>
</section>
+1
View File
@@ -74,6 +74,7 @@ export interface WorkspaceSharingConfig {
shared_group?: boolean;
shared_users?: string[];
share_memory?: boolean;
share_knowledge_graph?: boolean;
}
export type ChatGPTOAuthRoutingStrategy =