feat(trace): reliable stop/abort with ctx-aware streams and 2-phase router

Makes the Stop button on the traces page actually stop running traces.
Seven-phase implementation across provider HTTP, agent router, trace
persistence, WS events, tool exec, i18n, and integration tests.

- Provider HTTP+SSE ctx-aware: close socket on cancel via CtxBody wrapper
- Router 2-phase abort: CAS state machine, 3s grace, force-mark fallback
- Trace retry: 3 inline retries + 10-max retry queue, stale recovery 10min
- trace.status WS event: real-time UI updates (invalidates query on receive)
- Tool exec: process-group kill (SIGTERM→3s→SIGKILL), Rod page ctx watch
- i18n: 6 abort toast variants in en/vi/zh
- Integration: 9 scenarios, -race clean

Fixes tenant-ctx loss in forceMarkTraceAborted and retry worker broadcast
(caught by code-reviewer: C1/C2). Stale threshold intentionally 10min
because start_time-based; last_span_at migration is a follow-up.
This commit is contained in:
viettranx committed 2026-04-14 18:28:31 +07:00
1 parent 677396c71d
commit 1ac08155b0
70 files changed
+3052 -237

No files matched your search

+3
View File
@@ -263,6 +263,9 @@ func runGateway() {
// Create all agents — resolved lazily from database by the managed resolver.
agentRouter := agent.NewRouter()
if traceCollector != nil {
agentRouter.SetTraceCollector(traceCollector)
}
slog.Info("agents will be resolved lazily from database")
// Create gateway server and wire enforcement
+2
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log/slog"
"path/filepath"
"strings"
"github.com/google/uuid"
@@ -103,6 +104,7 @@ func processAnnounceLoop(
req.ForwardMedia = append(req.ForwardMedia, bus.MediaFile{
Path: mr.Path,
MimeType: mr.ContentType,
Filename: filepath.Base(mr.Path), // preserve sanitized stem from producer
})
}
}
+1 -1
View File
@@ -40,7 +40,7 @@ func makeSchedulerRunFunc(agents *agent.Router, cfg *config.Config) scheduler.Ru
// The ctx from the scheduler is already cancellable; we create a child so the router's
// cancel func is independent from the scheduler's cancel func. Calling cancel twice is safe.
runCtx, cancel := context.WithCancel(ctx)
injectCh := agents.RegisterRun(req.RunID, req.SessionKey, agentID, cancel)
injectCh := agents.RegisterRun(runCtx, req.RunID, req.SessionKey, agentID, cancel)
defer agents.UnregisterRun(req.RunID)
defer cancel()
+8
View File
@@ -297,6 +297,14 @@ func wireTracingAndCron(
Payload: map[string]any{"trace_ids": ids},
})
}
// Immediate status broadcast on every successful status write (bypasses 5s flush).
traceCollector.SetStatusBroadcaster(func(p tracing.TraceStatusPayload, tid uuid.UUID) {
msgBus.Broadcast(bus.Event{
Name: protocol.EventTraceStatusChanged,
Payload: p,
TenantID: tid,
})
})
traceCollector.Start()
slog.Info("LLM tracing enabled")
}
+2
View File
@@ -350,6 +350,8 @@ Each update increments a sequence number for ordering. Updates are throttled at
| Video | `.mp4` |
| Sticker | `.png` |
**Filename preservation**: When a user uploads or shares a file with a recognizable name, the channel adapter populates `bus.MediaFile.Filename` with the original filename (e.g., Feishu file display name, Telegram `file_name`). This filename is then sanitized and persisted to disk in the format `{stem}-{8hex}{ext}` (e.g., `báo-cáo-2024-a1b2c3d4.pdf` → `bao-cao-2024-a1b2c3d4.pdf`). The sanitizer supports Vietnamese diacritics, CJK scripts, and filesystem safety. Media without a user-provided filename (voice notes, clipboard pastes) fall back to UUID-only names. Disk names with semantic stems enable vault enrichment to process documents contextually. See `internal/agent/media_filename.go` for sanitization rules.
**Send (outbound)**: Files are uploaded to Feishu with automatic type detection (opus, mp4, pdf, doc, xls, ppt, or generic stream).
### Mention Resolution
+9 -2
View File
@@ -30,9 +30,16 @@ flowchart LR
### Cancel Handling
When a run is cancelled via `/stop` or `/stopall`, the run context is cancelled but trace finalization still needs to persist. `FinishTrace()` detects `ctx.Err() != nil` and switches to `context.Background()` for the final database write. The trace status is set to `"cancelled"` instead of `"error"`.
When a run is cancelled via `chat.abort` (WS) or manual stop, the router atomically transitions to "aborting" state, cancels the request context, and waits 3 seconds for the agent goroutine to exit. If it doesn't, the trace is force-marked `cancelled` to prevent orphaned traces.
Context values (traceID, collector) survive cancellation -- only `ctx.Done()` and `ctx.Err()` change. This allows trace finalization to find everything it needs with a fresh context for the DB call.
During cancellation:
- HTTP provider streams close immediately (context-aware transport with `ResponseHeaderTimeout`; SSE body closes via goroutine wrapper to unblock socket read)
- Tool execution terminates (shell: process-group kill; browser: page close)
- Agent loop receives `ctx.Done()` and propagates cancellation up
Trace finalization persists independently with a detached context (`context.WithoutCancel`), 5-second timeout, and exponential backoff retry (3 tries, max 10 total via retry queue). This ensures trace status writes don't fail silently due to cancellation.
Stale recovery worker runs every 30 seconds and catches zombie traces (stuck in "running" state) older than 10 minutes. Threshold is conservative because it measures against `start_time`, not last activity — a tighter value would kill healthy long-running agent runs. (Follow-up: add `last_span_at` column to enable aggressive recovery of inactive runs.) Context values (traceID, collector) survive cancellation — only `ctx.Done()` and `ctx.Err()` change.
---
+36
View File
@@ -32,6 +32,42 @@ All notable changes to GoClaw Gateway are documented here. Format follows [Keep
## [Unreleased]
### Added
#### Trace Stop/Abort Redesign — Cascading 4-Layer Fix (2026-04-14)
The Stop button on the traces page now reliably aborts running traces. Previous implementation had independent race conditions across HTTP streaming, agent router, trace persistence, and UI polling; this redesign fixes all four layers atomically.
**What changed:**
- **HTTP streaming ctx-aware**: Provider clients use transport-level `ResponseHeaderTimeout` + `IdleConnTimeout` instead of socket-level `Client.Timeout`. SSE body closes immediately on ctx cancel via goroutine-based wrapper (prevents 5-minute socket block).
- **Router 2-phase abort**: `AbortRun()` atomically transitions to aborting state, waits 3s for goroutine exit via `Done` channel, then force-marks trace `cancelled` if timeout. No orphaned goroutines, no "not found" race.
- **Trace status persistence**: `SetTraceStatus()` detached from request context with 5s timeout, 3-try exponential backoff, and bounded in-memory retry queue (10 max tries). Stale recovery worker runs every 30s, catches zombie traces in 2 minutes instead of 30.
- **Real-time UI updates**: New WS event `trace.status` broadcasts status changes immediately after persist succeeds. UI drops 60s polling interval, subscribes to events instead.
- **Tool execution audit**: Shell commands use process-group kill (`SIGTERM`→3s→`SIGKILL`). Browser automation (Rod) closes pages on ctx cancel. MCP delegates timeout after 5s.
- **i18n**: 6 abort toast variants (success/timeout/not-found/already-done/db-error/unknown) + translations for en/vi/zh.
**Impact**: Existing traces and sessions unaffected. UI now reflects backend state accurately. Zero breaking changes.
#### Preserve User-Provided Filenames for Media Uploads (2026-04-14)
Filenames provided by users for chat media uploads now survive the channel adapter → agent → disk persistence round trip, enabling vault enrichment to process human-readable document names instead of falling back to generic UUID-only storage.
**Why**: Vault enrichment was skipping UUID-only disk names (design safety to avoid noisy auto-generated files), causing documents with Vietnamese or CJK stems to remain unenriched and lose semantic context.
**What changed**:
- **`bus.MediaFile.Filename` field**: Channel adapters now populate this field when source provides original filename (e.g., user-selected file upload, Telegram file_name, WhatsApp display_name)
- **Sanitizer** (`internal/agent/media_filename.go`): Derives safe stems via `sanitizeFilename()` with:
- **Vietnamese pre-NFD map**: `đ/Đ → d` (Unicode NFD does not decompose these precomposed letters)
- **CJK passthrough**: Dominant-script heuristic detects Vietnamese/CJK inputs and preserves original runes (no ASCII slugification)
- **Filesystem safety**: Removes control chars, path traversal markers (`..`, `/`, `\\`), and reserved names
- **Max length**: 60 runes (script-aware, not byte-based) to avoid platform path limits
- **Disk naming scheme**: `{sanitized-stem}-{8hex}{ext}` (e.g., `bao-cao-q4-a1b2c3d4.pdf`) when sanitizer returns non-empty stem; UUID fallback `{uuid}{ext}` for empty stems (voice notes, clipboard pastes, tool-generated media)
- **Vault enrichment gating** (`internal/vault/enrich_skip_filter.go`): Now skips UUID-only filenames (matching `^[0-9a-f]{8}-...$` pattern) while processing named stems
- **6 channel adapters wired**: Telegram, Slack, Discord, Feishu/Lark, Zalo OA, WhatsApp all set `MediaFile.Filename` when available
- **Tools + orchestration**: `web_search` tool (PDF downloads), delegate/subagent media propagation, all preserve filenames end-to-end
**Impact**: Existing flows with empty `Filename` are unaffected (UUID-named as before). New flows with filenames produce human-readable, enrichable disk names. Zero breaking changes.
### Security
#### Tenant-Scope Hotfix (2026-04-12)
+21
View File
@@ -727,6 +727,7 @@ The server pushes events to connected clients via event frames. Key event types:
| `chunk` | Streaming text chunk |
| `tool.call` | Tool invocation started |
| `tool.result` | Tool invocation completed |
| `trace.status` | Trace status changed (cancelled, completed, error) |
| `session.updated` | Session metadata changed |
| `agent.updated` | Agent config changed |
| `cron.fired` | Cron job triggered |
@@ -737,6 +738,7 @@ The server pushes events to connected clients via event frames. Key event types:
| Event | Description | Payload |
|-------|-------------|---------|
| `trace.status` | Trace status changed (real-time stop/abort visibility) | `{traceId, status, endedAt?}` |
| `evolution.metrics.updated` | New evolution metrics recorded | `{agentId, metricType, toolName, value}` |
| `evolution.suggestion` | New evolution suggestion generated | `{agentId, suggestionId, type, title}` |
| `episodic.summary` | New episodic summary created/updated | `{agentId, summaryId, userId}` |
@@ -745,6 +747,25 @@ The server pushes events to connected clients via event frames. Key event types:
| `orchestration.mode.changed` | Agent orchestration mode changed | `{agentId, newMode}` |
| `v3flags.changed` | V3 feature flags updated | `{agentId, flags}` |
#### `trace.status` Event
Emitted whenever a trace status changes (e.g., `running` → `cancelled`, `running` → `completed`). Allows UI to update trace state in real-time without polling, particularly critical for stop/abort operations.
**Payload:**
```json
{
"traceId": "uuid",
"status": "cancelled",
"endedAt": "2026-04-14T12:34:56.789Z"
}
```
**Status values:**
- `cancelled` — User stopped the trace via `chat.abort`
- `completed` — Trace finished successfully
- `error` — Trace failed with an error
- `running` — Emitted when trace transitions from another state (rare; mostly informational)
---
## File Reference
+2 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"log/slog"
"os"
"path/filepath"
"strings"
"github.com/nextlevelbuilder/goclaw/internal/bus"
@@ -79,7 +80,7 @@ func (l *Loop) enrichInputMedia(ctx context.Context, req *RunRequest, messages [
var imageFiles []bus.MediaFile
for _, ref := range mediaRefs {
if ref.Kind == "image" && ref.Path != "" {
imageFiles = append(imageFiles, bus.MediaFile{Path: ref.Path, MimeType: ref.MimeType})
imageFiles = append(imageFiles, bus.MediaFile{Path: ref.Path, MimeType: ref.MimeType, Filename: filepath.Base(ref.Path)})
}
}
if deferToReadImageTool {
+6
View File
@@ -104,6 +104,12 @@ func (l *Loop) Run(ctx context.Context, req RunRequest) (*RunResult, error) {
ctx = tracing.WithTraceTeamID(ctx, *trace.TeamID)
}
// Notify the gateway so it can associate this traceID with the active run
// entry for force-abort (forceMarkTraceAborted needs traceID at abort time).
if req.OnTraceCreated != nil {
req.OnTraceCreated(traceID)
}
// Pre-generate root "agent" span ID so LLM/tool spans can reference it as parent.
agentSpanID = store.GenNewID()
ctx = tracing.WithParentSpanID(ctx, agentSpanID)
+5
View File
@@ -566,6 +566,11 @@ type RunRequest struct {
// user follow-up messages into the running conversation.
InjectCh <-chan InjectedMessage
// OnTraceCreated is called once the trace UUID is determined for this run.
// Used by the gateway to associate the trace ID with the active run entry
// so force-abort can mark the correct trace as cancelled. Nil = no-op.
OnTraceCreated func(traceID uuid.UUID)
// Delegation context (set when running as a delegate agent)
DelegationID string // delegation ID for event correlation
TeamID string // team ID (if delegation is team-scoped)
+24 -2
View File
@@ -104,7 +104,29 @@ func (l *Loop) persistMedia(sessionKey string, files []bus.MediaFile, workspace
if ext == "" {
ext = filepath.Ext(srcPath) // fallback to source extension
}
dstPath := filepath.Join(uploadsDir, id+ext)
// Disk-naming: preserve user filename when present so vault enrichment
// can process uploads (UUID-only names are skipped by enrich_skip_filter).
// Empty Filename (voice note, clipboard paste, tool-generated) →
// fall back to UUID, keeping legacy behavior.
diskName := id + ext
if stem := sanitizeFilename(f.Filename); stem != "" {
diskName = stem + "-" + shortID(8) + ext
}
dstPath := filepath.Join(uploadsDir, diskName)
// Traversal guard: ensure resolved path is inside uploadsDir.
// sanitizeFilename already strips ".." / "/" / "\\", but this is
// a defense-in-depth check covering any future regressions.
cleanDst := filepath.Clean(dstPath) + string(os.PathSeparator)
cleanUploads := filepath.Clean(uploadsDir) + string(os.PathSeparator)
if !strings.HasPrefix(cleanDst, cleanUploads) {
slog.Warn("media: refusing to persist outside uploadsDir", "dst", dstPath, "uploads", uploadsDir)
if sanitizedTemp != "" {
os.Remove(sanitizedTemp)
}
continue
}
if err := copyMediaFile(srcPath, dstPath); err != nil {
slog.Warn("media: failed to persist file", "path", f.Path, "error", err)
@@ -467,7 +489,7 @@ func (l *Loop) reloadMediaForMessages(msgs []providers.Message, maxMessages int)
if p == "" {
continue
}
imageFiles = append(imageFiles, bus.MediaFile{Path: p, MimeType: ref.MimeType})
imageFiles = append(imageFiles, bus.MediaFile{Path: p, MimeType: ref.MimeType, Filename: filepath.Base(p)})
}
if !hasImageRef {
+178
View File
@@ -0,0 +1,178 @@
// Package agent - media_filename.go provides pure helpers to derive a safe,
// human-readable disk filename from a user-provided original filename, and to
// generate a short random id used as a collision-avoidance suffix.
//
// Disk naming scheme (used by persistMedia):
//
// {sanitizeFilename(stem)}-{shortID(8)}{ext}
//
// Callers MUST fall back to a UUID name when sanitizeFilename returns "".
// This keeps backward compatibility for voice notes, clipboard paste, and
// tool-generated media that arrive without an original filename.
package agent
import (
"crypto/rand"
"encoding/hex"
"path/filepath"
"regexp"
"strings"
"unicode"
"golang.org/x/text/unicode/norm"
)
// slugRe collapses any run of non-ASCII-lowercase-alphanumeric chars into "-".
// Applied on the Latin-dominant branch after normalization + lower-casing.
var slugRe = regexp.MustCompile(`[^a-z0-9]+`)
// unsafeRe matches filesystem-unsafe characters that must be stripped from
// non-Latin filenames (CJK, Arabic, etc.) where we want to preserve the original
// script but still guarantee a safe on-disk name.
var unsafeRe = regexp.MustCompile(`[\x00-\x1f/\\:*?"<>|]`)
// viMap handles Vietnamese-specific letters that NFD combining-mark stripping
// does not decompose (đ/Đ are precomposed in Unicode without a base+mark form).
// Applied BEFORE NFD normalization on the Latin branch.
var viMap = map[rune]rune{
'đ': 'd', 'Đ': 'd',
}
// maxFilenameRunes caps the stem at 60 runes to keep on-disk names short and
// avoid platform path-length issues. Rune-based (not byte-based) to stay safe
// for multi-byte scripts.
const maxFilenameRunes = 60
// isDominantLatin returns true when more than 50% of the runes are Latin
// letters or ASCII digits. This distinguishes "Báo cáo Q4" (Latin-with-accents,
// slug it) from "猫の写真" (CJK, preserve runes).
func isDominantLatin(runes []rune) bool {
if len(runes) == 0 {
return false
}
latin := 0
for _, r := range runes {
if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
latin++
continue
}
// Latin-1 Supplement + Latin Extended ranges cover European accents.
if unicode.Is(unicode.Latin, r) {
latin++
}
}
return latin*2 > len(runes)
}
// sanitizeFilename derives a safe stem (no extension) from a user-provided
// filename. Returns "" when the input is empty or sanitizes to nothing — the
// caller must then fall back to a UUID-only name.
//
// Guarantees on the return value:
// - never contains "/", "\\", "..", or ASCII control chars
// - never starts or ends with "-"
// - len([]rune(out)) <= maxFilenameRunes
//
// Branching:
// - Latin-dominant input is accent-stripped, lowercased, and slugified.
// - Non-Latin-dominant input (CJK, Arabic, etc.) keeps its original runes
// with only filesystem-unsafe characters removed.
func sanitizeFilename(orig string) string {
orig = strings.TrimSpace(orig)
if orig == "" {
return ""
}
// Strip the extension — caller re-appends the canonical ext from MIME.
ext := filepath.Ext(orig)
stem := strings.TrimSuffix(orig, ext)
// Defense-in-depth: strip any path components and reject traversal markers.
stem = filepath.Base(stem)
if stem == "." || stem == ".." || strings.Contains(stem, "..") {
return ""
}
// Dotfiles with no real stem (e.g. ".env") have empty pre-ext content —
// filepath.Ext(".env") returns ".env", so stem == "". Already handled below.
stem = strings.TrimSpace(stem)
if stem == "" {
return ""
}
runes := []rune(stem)
if isDominantLatin(runes) {
stem = latinSlug(runes)
} else {
stem = nonLatinClean(runes)
}
// Cap to maxFilenameRunes (rune-based).
capped := []rune(stem)
if len(capped) > maxFilenameRunes {
capped = capped[:maxFilenameRunes]
}
out := strings.Trim(string(capped), "-")
return out
}
// latinSlug applies Vietnamese mapping, NFD decomposition + combining-mark
// stripping, lower-casing, and the slugRe replacement to produce an
// ASCII-only kebab-case stem.
func latinSlug(runes []rune) string {
// Apply Vietnamese đ/Đ map pre-NFD (NFD does not decompose them).
mapped := make([]rune, 0, len(runes))
for _, r := range runes {
if m, ok := viMap[r]; ok {
mapped = append(mapped, m)
continue
}
mapped = append(mapped, r)
}
// NFD: base letter + combining marks → drop marks (unicode.Mn).
decomposed := norm.NFD.String(string(mapped))
var b strings.Builder
b.Grow(len(decomposed))
for _, r := range decomposed {
if unicode.Is(unicode.Mn, r) {
continue
}
b.WriteRune(r)
}
lowered := strings.ToLower(b.String())
// Replace non-[a-z0-9] runs with "-", then collapse/trim.
slug := slugRe.ReplaceAllString(lowered, "-")
return strings.Trim(slug, "-")
}
// nonLatinClean preserves the script of non-Latin inputs while removing
// control chars and filesystem-unsafe characters. Whitespace is collapsed to
// a single space (kept, since CJK users expect spaces preserved).
func nonLatinClean(runes []rune) string {
s := string(runes)
s = unsafeRe.ReplaceAllString(s, "")
// Strip any remaining control runes not caught by unsafeRe.
var b strings.Builder
b.Grow(len(s))
for _, r := range s {
if unicode.IsControl(r) {
continue
}
b.WriteRune(r)
}
return strings.TrimSpace(b.String())
}
// shortID returns n lowercase hex chars sourced from crypto/rand. n must be
// even and > 0; otherwise shortID panics (caller bug — always called with 8).
func shortID(n int) string {
if n <= 0 || n%2 != 0 {
panic("shortID: n must be a positive even number")
}
b := make([]byte, n/2)
if _, err := rand.Read(b); err != nil {
// crypto/rand.Read on modern Go never fails in practice; panic signals
// a catastrophic OS-level issue worth surfacing.
panic("shortID: crypto/rand failure: " + err.Error())
}
return hex.EncodeToString(b)
}
+133
View File
@@ -0,0 +1,133 @@
package agent
import (
"regexp"
"strings"
"testing"
"unicode/utf8"
)
func TestSanitizeFilename(t *testing.T) {
cases := []struct {
name string
in string
want string
}{
// Latin / Vietnamese
{"vietnamese_basic", "Báo cáo Q4.pdf", "bao-cao-q4"},
{"vietnamese_dong_lower", "đường.doc", "duong"},
{"vietnamese_dong_upper", "Đường.doc", "duong"},
{"english_simple", "Quarterly Report.pdf", "quarterly-report"},
{"english_numbers", "report-2025_v2.pdf", "report-2025-v2"},
{"double_dot", "report.final.pdf", "report-final"},
// Non-Latin preserved
{"cjk_jp", "猫の写真.png", "猫の写真"},
{"cjk_zh", "报告.pdf", "报告"},
{"arabic", "تقرير.pdf", "تقرير"},
// Degenerate / unsafe
{"empty", "", ""},
{"whitespace_only", " ", ""},
// path_traversal: filepath.Base strips dirs → "passwd" is safe slug (no /, no ..).
{"path_traversal", "../../etc/passwd", "passwd"},
{"dotfile", ".env", ""},
{"dot_only", ".", ""},
{"dotdot_only", "..", ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := sanitizeFilename(tc.in)
if got != tc.want {
t.Fatalf("sanitizeFilename(%q) = %q, want %q", tc.in, got, tc.want)
}
})
}
}
func TestSanitizeFilenameGuarantees(t *testing.T) {
// Path-traversal-in-middle and unsafe runes must never survive in output.
inputs := []string{
"a/b/c.png",
"a\\b\\c.png",
"name:with:colons.txt",
"quote\"mark.pdf",
"pipe|file.md",
"question?.md",
"star*.md",
"angle<>.md",
"with\x00null.pdf",
"猫/の写真.png",
}
unsafe := regexp.MustCompile(`[/\\:*?"<>|\x00-\x1f]`)
for _, in := range inputs {
out := sanitizeFilename(in)
if unsafe.MatchString(out) {
t.Errorf("sanitizeFilename(%q) = %q contains unsafe char", in, out)
}
if strings.Contains(out, "..") {
t.Errorf("sanitizeFilename(%q) = %q contains ..", in, out)
}
}
}
func TestSanitizeFilenameLengthCap(t *testing.T) {
// Latin: 200-char ASCII stem.
long := strings.Repeat("a", 200) + ".pdf"
out := sanitizeFilename(long)
if n := utf8.RuneCountInString(out); n > maxFilenameRunes {
t.Errorf("Latin cap: got %d runes, want <= %d", n, maxFilenameRunes)
}
// CJK: 100 CJK runes (multi-byte).
cjkLong := strings.Repeat("猫", 100) + ".png"
cjkOut := sanitizeFilename(cjkLong)
if n := utf8.RuneCountInString(cjkOut); n > maxFilenameRunes {
t.Errorf("CJK cap: got %d runes, want <= %d", n, maxFilenameRunes)
}
}
func TestSanitizeFilenameNoLeadingTrailingDash(t *testing.T) {
cases := []string{
" hello .pdf",
"---weird---.pdf",
"$$$money$$$.pdf",
}
for _, in := range cases {
out := sanitizeFilename(in)
if out == "" {
continue
}
if strings.HasPrefix(out, "-") || strings.HasSuffix(out, "-") {
t.Errorf("sanitizeFilename(%q) = %q has leading/trailing dash", in, out)
}
}
}
func TestShortID(t *testing.T) {
got := shortID(8)
if len(got) != 8 {
t.Fatalf("shortID(8) len = %d, want 8", len(got))
}
if !regexp.MustCompile(`^[0-9a-f]{8}$`).MatchString(got) {
t.Fatalf("shortID(8) = %q, not lowercase hex", got)
}
// Two calls should (almost always) differ — 32-bit entropy.
if got == shortID(8) {
// Retry once — if still equal, something is wrong.
if got == shortID(8) {
t.Fatalf("shortID(8) produced identical output twice in a row: %q", got)
}
}
}
func TestShortIDPanicsOnInvalidInput(t *testing.T) {
for _, n := range []int{0, -2, 3, 7} {
func() {
defer func() {
if r := recover(); r == nil {
t.Errorf("shortID(%d) did not panic", n)
}
}()
_ = shortID(n)
}()
}
}
+110
View File
@@ -0,0 +1,110 @@
package agent
import (
"os"
"path/filepath"
"regexp"
"testing"
"github.com/nextlevelbuilder/goclaw/internal/bus"
)
// TestPersistMedia_NamingScheme verifies that persistMedia produces:
// - `{slug}-{8hex}{ext}` when Filename is present and sanitizes to a non-empty stem
// - `{uuid}{ext}` UUID fallback when Filename is empty or sanitizes to ""
//
// It covers Vietnamese/CJK preservation, image re-encode via SanitizeImage,
// and the voice-note (empty Filename) fallback path.
func TestPersistMedia_NamingScheme(t *testing.T) {
workspace := t.TempDir()
writeTmp := func(t *testing.T, name, data string) string {
t.Helper()
p := filepath.Join(t.TempDir(), name)
if err := os.WriteFile(p, []byte(data), 0644); err != nil {
t.Fatalf("WriteFile: %v", err)
}
return p
}
uuidPat := regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.[a-z0-9]+$`)
slugPat := regexp.MustCompile(`^(.+)-[0-9a-f]{8}\.[a-z0-9]+$`)
type tc struct {
name string
filename string
content string
mime string
wantSlug string // prefix (stem) expected in disk name; empty means UUID fallback
wantExtPat string // extension regex on disk (lowercase)
}
cases := []tc{
{
name: "vietnamese_pdf",
filename: "Báo cáo Q4.pdf",
content: "%PDF-1.4\n%fake",
mime: "application/pdf",
wantSlug: "bao-cao-q4",
wantExtPat: `\.pdf`,
},
{
name: "cjk_preserved",
filename: "猫の写真.png",
content: "not a real png",
mime: "application/octet-stream", // bypass image sanitize path
wantSlug: "猫の写真",
wantExtPat: `\.bin`,
},
{
name: "empty_filename_uuid_fallback",
filename: "",
content: "audio bytes",
mime: "audio/ogg",
wantSlug: "",
wantExtPat: `\.ogg`,
},
{
name: "filename_all_unsafe_fallback",
filename: "///",
content: "pdf body",
mime: "application/pdf",
wantSlug: "",
wantExtPat: `\.pdf`,
},
}
var loop Loop
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
src := writeTmp(t, "src.bin", c.content)
refs := loop.persistMedia("session-key-test", []bus.MediaFile{{
Path: src,
MimeType: c.mime,
Filename: c.filename,
}}, workspace)
if len(refs) != 1 {
t.Fatalf("got %d refs, want 1", len(refs))
}
base := filepath.Base(refs[0].Path)
if c.wantSlug == "" {
// UUID fallback
if !uuidPat.MatchString(base) {
t.Fatalf("expected UUID-pattern disk name, got %q", base)
}
return
}
m := slugPat.FindStringSubmatch(base)
if m == nil {
t.Fatalf("expected slug+8hex disk name, got %q", base)
}
if m[1] != c.wantSlug {
t.Fatalf("stem = %q, want %q (full: %q)", m[1], c.wantSlug, base)
}
extOk := regexp.MustCompile(c.wantExtPat + `$`)
if !extOk.MatchString(base) {
t.Fatalf("disk name %q does not match ext pattern %q", base, c.wantExtPat)
}
})
}
}
+2 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/json"
"log/slog"
"path/filepath"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/providers"
@@ -76,7 +77,7 @@ func (l *Loop) loadHistoricalImagesForTool(ctx context.Context, currentRefs []pr
if p == "" {
continue
}
histPaths = append(histPaths, bus.MediaFile{Path: p, MimeType: ref.MimeType})
histPaths = append(histPaths, bus.MediaFile{Path: p, MimeType: ref.MimeType, Filename: filepath.Base(p)})
}
if hasImage {
count++
+139 -34
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/google/uuid"
@@ -32,17 +33,25 @@ type AgentActivityStatus struct {
StartedAt time.Time
}
// TraceCollector is a minimal interface for marking a trace as aborted.
// Defined here (not in internal/tracing) to avoid import cycles.
// *tracing.Collector implements this interface implicitly.
type TraceCollector interface {
FinishTrace(ctx context.Context, traceID uuid.UUID, status, errMsg, outputPreview string)
}
// Router manages multiple agent Loop instances.
// Each agent has a unique ID and its own provider/model/tools config.
// Cached Loops expire after TTL (safety net for multi-instance).
type Router struct {
agents map[string]*agentEntry
mu sync.RWMutex
activeRuns sync.Map // runID → *ActiveRun
sessionRuns sync.Map // sessionKey → runID (secondary index for O(1) IsSessionBusy)
agentActivity sync.Map // sessionKey → *AgentActivityStatus
resolver ResolverFunc // optional: lazy creation from DB
ttl time.Duration
agents map[string]*agentEntry
mu sync.RWMutex
activeRuns sync.Map // runID → *ActiveRun
sessionRuns sync.Map // sessionKey → runID (secondary index for O(1) IsSessionBusy)
agentActivity sync.Map // sessionKey → *AgentActivityStatus
resolver ResolverFunc // optional: lazy creation from DB
ttl time.Duration
traceCollector TraceCollector // optional: for force-marking aborted traces
}
func NewRouter() *Router {
@@ -52,6 +61,11 @@ func NewRouter() *Router {
}
}
// SetTraceCollector wires the trace collector so forceMarkTraceAborted can update DB.
func (r *Router) SetTraceCollector(c TraceCollector) {
r.traceCollector = c
}
// SetResolver sets a resolver function for lazy agent creation.
func (r *Router) SetResolver(fn ResolverFunc) {
r.mu.Lock()
@@ -243,11 +257,38 @@ type ActiveRun struct {
Cancel context.CancelFunc
StartedAt time.Time
InjectCh chan InjectedMessage // buffered channel for mid-run user message injection
Done chan struct{} // closed when goroutine actually exits (via UnregisterRun)
State atomic.Int32 // 0=running, 1=aborting, 2=done
TraceID uuid.UUID // set after trace creation via SetRunTraceID
TenantID uuid.UUID // captured at RegisterRun for forceMarkTraceAborted
}
// AbortResult describes the outcome of a single AbortRun call.
type AbortResult struct {
RunID string
Stopped bool // graceful stop within abortGraceTimeout
Forced bool // timeout exceeded; trace force-marked cancelled
AlreadyAborting bool // 2nd click while phase 2 still in flight
NotFound bool // run never existed or already finished
Unauthorized bool // sessionKey mismatch
}
// abortGraceTimeout is the maximum time AbortRun waits for a goroutine to exit
// before force-marking the trace as cancelled.
const abortGraceTimeout = 3 * time.Second
// safeClose closes ch without panicking if ch is already closed.
// recover-based: cheaper than a sync.Once per ActiveRun because it avoids an
// extra heap allocation per run on the common (single-close) path.
func safeClose(ch chan struct{}) {
defer func() { _ = recover() }()
close(ch)
}
// RegisterRun records an active run so it can be aborted later.
// ctx is used to capture the tenant ID for forceMarkTraceAborted.
// Returns a receive-only channel for mid-run message injection.
func (r *Router) RegisterRun(runID, sessionKey, agentID string, cancel context.CancelFunc) <-chan InjectedMessage {
func (r *Router) RegisterRun(ctx context.Context, runID, sessionKey, agentID string, cancel context.CancelFunc) <-chan InjectedMessage {
injectCh := make(chan InjectedMessage, injectBufferSize)
r.activeRuns.Store(runID, &ActiveRun{
RunID: runID,
@@ -256,39 +297,120 @@ func (r *Router) RegisterRun(runID, sessionKey, agentID string, cancel context.C
Cancel: cancel,
StartedAt: time.Now(),
InjectCh: injectCh,
Done: make(chan struct{}),
TenantID: store.TenantIDFromContext(ctx),
})
r.sessionRuns.Store(sessionKey, runID)
return injectCh
}
// SetRunTraceID associates a trace UUID with an active run.
// Called from loop_run.go after CreateTrace succeeds so forceMarkTraceAborted
// can update the correct trace record on a 3s timeout.
func (r *Router) SetRunTraceID(runID string, traceID uuid.UUID) {
if val, ok := r.activeRuns.Load(runID); ok {
val.(*ActiveRun).TraceID = traceID
}
}
// UnregisterRun removes a completed/cancelled run from tracking.
// Closes Done BEFORE deleting from maps so any concurrent AbortRun waiting
// on Done sees the signal before the entry disappears.
func (r *Router) UnregisterRun(runID string) {
if val, ok := r.activeRuns.Load(runID); ok {
run := val.(*ActiveRun)
run.State.Store(2) // mark done
safeClose(run.Done)
r.sessionRuns.Delete(run.SessionKey)
}
r.activeRuns.Delete(runID)
}
// AbortRun cancels a single run by ID. sessionKey is validated for authorization
// (matching TS chat-abort.ts: verify sessionKey matches before aborting).
// Returns true if the run was found and cancelled.
func (r *Router) AbortRun(runID, sessionKey string) bool {
// AbortRun cancels a single run by ID using a 2-phase verified abort.
//
// Phase 1: CAS state 0→1 (idempotent, prevents double-cancel).
// Phase 2: wait ≤abortGraceTimeout for goroutine to exit via Done channel.
// On timeout: force-mark trace as cancelled so the UI moves on.
//
// sessionKey is validated for authorization when non-empty.
func (r *Router) AbortRun(runID, sessionKey string) AbortResult {
res := AbortResult{RunID: runID}
val, ok := r.activeRuns.Load(runID)
if !ok {
return false
res.NotFound = true
return res
}
run := val.(*ActiveRun)
// Authorization: sessionKey must match (matching TS behavior)
if sessionKey != "" && run.SessionKey != sessionKey {
return false
res.Unauthorized = true
return res
}
// CAS 0→1: transition Running → Aborting.
// Any failure means abort is already in flight (state==1) or run is done (state==2).
if !run.State.CompareAndSwap(0, 1) {
if run.State.Load() == 2 {
// Race: run finished between Load and CAS
res.NotFound = true
} else {
res.AlreadyAborting = true
}
return res
}
run.Cancel()
r.sessionRuns.Delete(run.SessionKey)
r.activeRuns.Delete(runID)
return true
select {
case <-run.Done:
res.Stopped = true
case <-time.After(abortGraceTimeout):
// Goroutine is stuck; force trace status so the UI does not hang.
r.forceMarkTraceAborted(runID)
res.Forced = true
}
return res
}
// forceMarkTraceAborted marks a trace as cancelled in DB when the 3s grace
// period expires and the goroutine has not yet exited.
// Uses a detached background context scoped to the run's tenant so the DB
// write is not blocked by the caller's (already-cancelled) run context.
func (r *Router) forceMarkTraceAborted(runID string) {
if r.traceCollector == nil {
return
}
val, ok := r.activeRuns.Load(runID)
if !ok {
return
}
run := val.(*ActiveRun)
if run.TraceID == uuid.Nil {
return
}
ctx := context.Background()
if run.TenantID != uuid.Nil {
ctx = store.WithTenantID(ctx, run.TenantID)
}
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
r.traceCollector.FinishTrace(ctx, run.TraceID, "cancelled", "force-aborted (3s grace exceeded)", "")
}
// AbortRunsForSession cancels all active runs for a session key.
// Returns rich results for each run (one timer per run is fine for typical 1–2 runs/session).
func (r *Router) AbortRunsForSession(sessionKey string) []AbortResult {
var results []AbortResult
r.activeRuns.Range(func(key, val any) bool {
run := val.(*ActiveRun)
if run.SessionKey == sessionKey {
results = append(results, r.AbortRun(run.RunID, ""))
}
return true
})
return results
}
// InjectMessage sends a user message to the running loop for a session.
@@ -373,20 +495,3 @@ func (r *Router) SessionRunID(sessionKey string) (string, bool) {
}
return val.(string), true
}
// AbortRunsForSession cancels all active runs for a session key.
// Returns the list of aborted run IDs.
func (r *Router) AbortRunsForSession(sessionKey string) []string {
var aborted []string
r.activeRuns.Range(func(key, val any) bool {
run := val.(*ActiveRun)
if run.SessionKey == sessionKey {
run.Cancel()
r.activeRuns.Delete(key)
r.sessionRuns.Delete(sessionKey)
aborted = append(aborted, run.RunID)
}
return true
})
return aborted
}
+363 -114
View File
@@ -2,163 +2,412 @@ package agent
import (
"context"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/google/uuid"
)
// TestAbortRun_CancelsContext verifies that AbortRun calls the cancel func
// and removes the run from tracking.
func TestAbortRun_CancelsContext(t *testing.T) {
r := NewRouter()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
runID := "run-1"
sessionKey := "session-1"
r.RegisterRun(runID, sessionKey, "agent-1", cancel)
if !r.IsSessionBusy(sessionKey) {
t.Fatal("session should be busy after RegisterRun")
}
ok := r.AbortRun(runID, sessionKey)
if !ok {
t.Fatal("AbortRun should return true for active run")
}
if ctx.Err() == nil {
t.Fatal("context should be cancelled after AbortRun")
}
if r.IsSessionBusy(sessionKey) {
t.Fatal("session should not be busy after AbortRun")
}
// mockTraceCollector counts FinishTrace calls for test assertions.
type mockTraceCollector struct {
mu sync.Mutex
calls []finishTraceCall
}
// TestAbortRun_WrongSessionKey verifies authorization check.
func TestAbortRun_WrongSessionKey(t *testing.T) {
type finishTraceCall struct {
TraceID uuid.UUID
Status string
ErrMsg string
OutputPreview string
}
func (m *mockTraceCollector) FinishTrace(_ context.Context, traceID uuid.UUID, status, errMsg, outputPreview string) {
m.mu.Lock()
defer m.mu.Unlock()
m.calls = append(m.calls, finishTraceCall{TraceID: traceID, Status: status, ErrMsg: errMsg, OutputPreview: outputPreview})
}
func (m *mockTraceCollector) callCount() int {
m.mu.Lock()
defer m.mu.Unlock()
return len(m.calls)
}
// ---- helpers ----
func newRouterWithCollector(tc TraceCollector) *Router {
r := NewRouter()
r.SetTraceCollector(tc)
return r
}
// registerRun is a test helper that creates a cancel func and registers a run.
// Returns the cancel func and Done channel from the stored ActiveRun.
func registerRun(r *Router, runID, sessionKey string) (context.CancelFunc, chan struct{}) {
_, cancel := context.WithCancel(context.Background())
defer cancel()
r.RegisterRun(context.Background(), runID, sessionKey, "agent-1", cancel)
val, _ := r.activeRuns.Load(runID)
return cancel, val.(*ActiveRun).Done
}
r.RegisterRun("run-1", "session-1", "agent-1", cancel)
// ---- tests ----
ok := r.AbortRun("run-1", "wrong-session")
if ok {
t.Fatal("AbortRun should return false for mismatched sessionKey")
// TestAbortRun_Stopped verifies graceful stop: goroutine calls UnregisterRun
// before the 3s grace period expires → result.Stopped == true.
func TestAbortRun_Stopped(t *testing.T) {
r := NewRouter()
runID := "run-stop"
sessionKey := "session-stop"
_, _ = registerRun(r, runID, sessionKey)
// Simulate goroutine exiting after 50ms.
go func() {
time.Sleep(50 * time.Millisecond)
r.UnregisterRun(runID)
}()
res := r.AbortRun(runID, sessionKey)
if !res.Stopped {
t.Fatalf("expected Stopped=true, got %+v", res)
}
if !r.IsSessionBusy("session-1") {
t.Fatal("session should still be busy after failed abort")
if res.Forced || res.AlreadyAborting || res.NotFound || res.Unauthorized {
t.Fatalf("unexpected flags set: %+v", res)
}
}
// TestAbortRun_NotFound returns false for unknown run IDs.
// TestAbortRun_Forced verifies forced abort: goroutine never exits within 3s →
// result.Forced == true and TraceCollector.FinishTrace is called once.
func TestAbortRun_Forced(t *testing.T) {
tc := &mockTraceCollector{}
r := newRouterWithCollector(tc)
runID := "run-forced"
sessionKey := "session-forced"
traceID := uuid.New()
_, _ = registerRun(r, runID, sessionKey)
r.SetRunTraceID(runID, traceID)
// Goroutine deliberately never calls UnregisterRun.
start := time.Now()
res := r.AbortRun(runID, sessionKey)
elapsed := time.Since(start)
if !res.Forced {
t.Fatalf("expected Forced=true, got %+v", res)
}
if res.Stopped || res.AlreadyAborting || res.NotFound || res.Unauthorized {
t.Fatalf("unexpected flags set: %+v", res)
}
if elapsed < abortGraceTimeout {
t.Fatalf("expected at least %s elapsed, got %s", abortGraceTimeout, elapsed)
}
if tc.callCount() != 1 {
t.Fatalf("expected 1 FinishTrace call, got %d", tc.callCount())
}
if tc.calls[0].TraceID != traceID {
t.Fatalf("FinishTrace called with wrong traceID: %s", tc.calls[0].TraceID)
}
if tc.calls[0].Status != "cancelled" {
t.Fatalf("FinishTrace called with wrong status: %s", tc.calls[0].Status)
}
}
// TestAbortRun_AlreadyAborting verifies that 100 concurrent abort calls on the
// same run produce exactly one Stopped or Forced result and the rest return
// AlreadyAborting. No panics allowed.
func TestAbortRun_AlreadyAborting(t *testing.T) {
r := NewRouter()
runID := "run-concurrent"
sessionKey := "session-concurrent"
_, _ = registerRun(r, runID, sessionKey)
// Goroutine exits quickly so most callers see AlreadyAborting, not Forced.
go func() {
time.Sleep(20 * time.Millisecond)
r.UnregisterRun(runID)
}()
const n = 100
results := make([]AbortResult, n)
var wg sync.WaitGroup
wg.Add(n)
for i := 0; i < n; i++ {
i := i
go func() {
defer wg.Done()
results[i] = r.AbortRun(runID, sessionKey)
}()
}
wg.Wait()
var stopped, forced, alreadyAborting, other int
for _, res := range results {
switch {
case res.Stopped:
stopped++
case res.Forced:
forced++
case res.AlreadyAborting:
alreadyAborting++
default:
other++
}
}
definitiveCount := stopped + forced
if definitiveCount != 1 {
t.Fatalf("expected exactly 1 Stopped/Forced, got stopped=%d forced=%d", stopped, forced)
}
if alreadyAborting+definitiveCount != n {
t.Fatalf("counts don't sum to %d: stopped=%d forced=%d alreadyAborting=%d other=%d",
n, stopped, forced, alreadyAborting, other)
}
}
// TestAbortRun_Unauthorized verifies sessionKey mismatch returns Unauthorized.
func TestAbortRun_Unauthorized(t *testing.T) {
r := NewRouter()
_, _ = registerRun(r, "run-1", "session-1")
res := r.AbortRun("run-1", "wrong-session")
if !res.Unauthorized {
t.Fatalf("expected Unauthorized=true, got %+v", res)
}
if res.Stopped || res.Forced || res.AlreadyAborting || res.NotFound {
t.Fatalf("unexpected flags set: %+v", res)
}
// Session must still be busy (run not cancelled).
if !r.IsSessionBusy("session-1") {
t.Fatal("session should still be busy after Unauthorized abort")
}
}
// TestAbortRun_NotFound verifies abort on never-registered runID returns NotFound.
func TestAbortRun_NotFound(t *testing.T) {
r := NewRouter()
if r.AbortRun("nonexistent", "") {
t.Fatal("AbortRun should return false for unknown runID")
res := r.AbortRun("nonexistent", "")
if !res.NotFound {
t.Fatalf("expected NotFound=true, got %+v", res)
}
}
// TestAbortRunsForSession_CancelsAllRuns verifies that all runs in a session
// are cancelled when AbortRunsForSession is called.
func TestAbortRunsForSession_CancelsAllRuns(t *testing.T) {
r := NewRouter()
ctx1, cancel1 := context.WithCancel(context.Background())
defer cancel1()
ctx2, cancel2 := context.WithCancel(context.Background())
defer cancel2()
sessionKey := "session-multi"
r.RegisterRun("run-1", sessionKey, "agent-1", cancel1)
r.RegisterRun("run-2", sessionKey, "agent-1", cancel2)
aborted := r.AbortRunsForSession(sessionKey)
if len(aborted) != 2 {
t.Fatalf("expected 2 aborted runs, got %d", len(aborted))
}
if ctx1.Err() == nil || ctx2.Err() == nil {
t.Fatal("both contexts should be cancelled")
}
if r.IsSessionBusy(sessionKey) {
t.Fatal("session should not be busy after AbortRunsForSession")
}
}
// TestAbortRunsForSession_EmptySession returns empty slice.
func TestAbortRunsForSession_EmptySession(t *testing.T) {
r := NewRouter()
aborted := r.AbortRunsForSession("no-such-session")
if len(aborted) != 0 {
t.Fatalf("expected 0 aborted runs, got %d", len(aborted))
}
}
// TestAbortRun_AfterUnregister verifies the race scenario:
// if run is unregistered before abort, abort returns false (idempotent).
// TestAbortRun_AfterUnregister verifies that aborting after run completes returns NotFound
// (state==2 branch of CAS failure).
func TestAbortRun_AfterUnregister(t *testing.T) {
r := NewRouter()
_, cancel := context.WithCancel(context.Background())
defer cancel()
r.RegisterRun("run-1", "session-1", "agent-1", cancel)
_, _ = registerRun(r, "run-1", "session-1")
r.UnregisterRun("run-1")
ok := r.AbortRun("run-1", "session-1")
if ok {
t.Fatal("AbortRun should return false after UnregisterRun")
res := r.AbortRun("run-1", "session-1")
if !res.NotFound {
t.Fatalf("expected NotFound=true after unregister, got %+v", res)
}
}
// TestUnregisterRun_ClearsSessionIndex verifies that UnregisterRun cleans up
// both activeRuns and sessionRuns.
func TestUnregisterRun_ClearsSessionIndex(t *testing.T) {
// TestAbortRun_Race_UnregisterConcurrent verifies no panic and no goroutine leak
// when UnregisterRun and AbortRun interleave across 100 iterations.
func TestAbortRun_Race_UnregisterConcurrent(t *testing.T) {
for i := 0; i < 100; i++ {
r := NewRouter()
runID := "run-race"
sessionKey := "session-race"
_, _ = registerRun(r, runID, sessionKey)
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
r.UnregisterRun(runID)
}()
go func() {
defer wg.Done()
r.AbortRun(runID, sessionKey)
}()
wg.Wait()
// No assertion on result — we only care that no panic and no goroutine leak.
}
}
// TestAbortRunsForSession_ReturnsResults verifies AbortRunsForSession returns
// rich AbortResult slices for two concurrent runs.
// AbortRunsForSession is sequential; goroutines unregister independently so
// each AbortRun call sees its Done close quickly (Stopped result).
func TestAbortRunsForSession_ReturnsResults(t *testing.T) {
r := NewRouter()
_, cancel := context.WithCancel(context.Background())
defer cancel()
sessionKey := "session-multi"
r.RegisterRun("run-1", "session-1", "agent-1", cancel)
if !r.IsSessionBusy("session-1") {
t.Fatal("session should be busy")
_, doneA := registerRun(r, "run-a", sessionKey)
_, doneB := registerRun(r, "run-b", sessionKey)
_ = doneA
_ = doneB
// Goroutines close Done channels 50ms after they're signalled (Cancel called).
// We use a separate goroutine per run that watches for Cancel via context.
ctxA, cancelA := context.WithCancel(context.Background())
ctxB, cancelB := context.WithCancel(context.Background())
defer cancelA()
defer cancelB()
// Replace stored cancel funcs with our instrumented ones.
if valA, ok := r.activeRuns.Load("run-a"); ok {
valA.(*ActiveRun).Cancel = cancelA
}
if valB, ok := r.activeRuns.Load("run-b"); ok {
valB.(*ActiveRun).Cancel = cancelB
}
r.UnregisterRun("run-1")
if r.IsSessionBusy("session-1") {
t.Fatal("session should not be busy after unregister")
go func() {
<-ctxA.Done()
time.Sleep(10 * time.Millisecond)
r.UnregisterRun("run-a")
}()
go func() {
<-ctxB.Done()
time.Sleep(10 * time.Millisecond)
r.UnregisterRun("run-b")
}()
results := r.AbortRunsForSession(sessionKey)
if len(results) != 2 {
t.Fatalf("expected 2 results, got %d", len(results))
}
for _, res := range results {
if res.Unauthorized {
t.Fatalf("unexpected Unauthorized result: %+v", res)
}
}
// Both should be stopped (not forced), since goroutines exit promptly.
for _, res := range results {
if res.Forced {
t.Fatalf("expected Stopped not Forced: %+v", res)
}
}
}
// TestSessionKeyForRun returns correct session key or empty.
func TestSessionKeyForRun(t *testing.T) {
// TestSetRunTraceID verifies SetRunTraceID stores the traceID in the active run.
func TestSetRunTraceID(t *testing.T) {
r := NewRouter()
_, cancel := context.WithCancel(context.Background())
defer cancel()
_, _ = registerRun(r, "run-trace", "session-trace")
r.RegisterRun("run-1", "session-1", "agent-1", cancel)
tid := uuid.New()
r.SetRunTraceID("run-trace", tid)
if got := r.SessionKeyForRun("run-1"); got != "session-1" {
t.Fatalf("expected session-1, got %q", got)
val, ok := r.activeRuns.Load("run-trace")
if !ok {
t.Fatal("run should still be in activeRuns")
}
if got := r.SessionKeyForRun("nonexistent"); got != "" {
t.Fatalf("expected empty string, got %q", got)
if val.(*ActiveRun).TraceID != tid {
t.Fatalf("expected traceID %s, got %s", tid, val.(*ActiveRun).TraceID)
}
}
// TestActivityClearedOnAbort verifies that activity is cleared when the gateway
// subscriber processes a terminal event. We test the Router methods directly.
func TestActivityClearedOnAbort(t *testing.T) {
// TestSafeClose verifies safeClose does not panic when called twice.
func TestSafeClose(t *testing.T) {
ch := make(chan struct{})
safeClose(ch)
// Second call must not panic.
safeClose(ch)
}
// TestAbortRun_StateAfterStop verifies state transitions correctly to 2 (done)
// after a successful stop (goroutine calls UnregisterRun).
func TestAbortRun_StateAfterStop(t *testing.T) {
r := NewRouter()
_, cancel := context.WithCancel(context.Background())
defer cancel()
runID := "run-state"
sessionKey := "session-state"
sessionKey := "session-act"
r.RegisterRun("run-act", sessionKey, "agent-1", cancel)
r.UpdateActivity(sessionKey, "run-act", "thinking", "", 0)
_, _ = registerRun(r, runID, sessionKey)
if r.GetActivity(sessionKey) == nil {
t.Fatal("activity should be set")
go func() {
time.Sleep(20 * time.Millisecond)
r.UnregisterRun(runID)
}()
res := r.AbortRun(runID, sessionKey)
if !res.Stopped {
t.Fatalf("expected Stopped=true, got %+v", res)
}
// Simulate what gateway.go does on terminal events
r.ClearActivity(sessionKey)
if r.GetActivity(sessionKey) != nil {
t.Fatal("activity should be cleared after ClearActivity")
// After stop the run entry should be gone from the map.
if _, ok := r.activeRuns.Load(runID); ok {
t.Fatal("activeRuns should not contain run after UnregisterRun")
}
}
// TestAbortRun_NilTraceCollector_Forced verifies that force-abort path does not
// panic when no TraceCollector is wired (traceCollector == nil).
func TestAbortRun_NilTraceCollector_Forced(t *testing.T) {
r := NewRouter() // no SetTraceCollector
runID := "run-nil-tc"
sessionKey := "session-nil-tc"
_, _ = registerRun(r, runID, sessionKey)
// Goroutine never exits → Forced path will call forceMarkTraceAborted.
res := r.AbortRun(runID, sessionKey)
if !res.Forced {
t.Fatalf("expected Forced=true, got %+v", res)
}
// No panic = pass.
}
// TestAbortRunsForSession_EmptySession returns empty slice for unknown session.
func TestAbortRunsForSession_EmptySession(t *testing.T) {
r := NewRouter()
results := r.AbortRunsForSession("no-such-session")
if len(results) != 0 {
t.Fatalf("expected 0 results, got %d", len(results))
}
}
// TestAbortRun_DoesNotLeakGoroutine verifies that after the forced-abort timeout,
// when the goroutine eventually calls UnregisterRun, the Done channel is safely
// closed without a double-close panic.
func TestAbortRun_DoesNotLeakGoroutine(t *testing.T) {
r := NewRouter()
runID := "run-late-unregister"
sessionKey := "session-late"
_, done := registerRun(r, runID, sessionKey)
// Goroutine exits well after the 3s timeout.
var lateUnregisterDone atomic.Bool
go func() {
// Force path fires after abortGraceTimeout; goroutine "finishes" 100ms later.
time.Sleep(abortGraceTimeout + 100*time.Millisecond)
r.UnregisterRun(runID) // must not panic even though AbortRun already returned
lateUnregisterDone.Store(true)
}()
res := r.AbortRun(runID, sessionKey)
if !res.Forced {
t.Fatalf("expected Forced=true, got %+v", res)
}
// Wait for the goroutine to call UnregisterRun.
deadline := time.Now().Add(2 * time.Second)
for !lateUnregisterDone.Load() && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
if !lateUnregisterDone.Load() {
t.Fatal("goroutine did not complete UnregisterRun in time")
}
// Done channel should be closed (safeClose handled double-close internally).
select {
case <-done:
// expected
default:
t.Fatal("Done channel should be closed after UnregisterRun")
}
}
+1
View File
@@ -13,6 +13,7 @@ import (
type MediaFile struct {
Path string `json:"path"`
MimeType string `json:"mime_type,omitempty"` // e.g. "application/pdf", "image/jpeg"
Filename string `json:"filename,omitempty"` // original user-provided filename, e.g. "Báo cáo Q4.pdf"; empty → UUID fallback in persistMedia
}
// InboundMessage represents a message received from a channel (Telegram, Discord, etc.)
+5 -2
View File
@@ -12,6 +12,7 @@ import (
"context"
"log/slog"
"net/http"
"path/filepath"
"strings"
"sync"
"time"
@@ -602,10 +603,12 @@ func (c *BaseChannel) HandleMessage(senderID, chatID, content string, media []st
userID = senderID[:idx]
}
// Convert string paths to MediaFile (for channels that haven't been updated yet).
// Convert string paths to MediaFile (legacy path-only callers).
// Use filepath.Base(p) as filename so persistMedia's sanitizer gets a
// meaningful stem instead of falling back to UUID.
var mediaFiles []bus.MediaFile
for _, p := range media {
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p})
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p, Filename: filepath.Base(p)})
}
msg := bus.InboundMessage{
+5 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log/slog"
"path/filepath"
"strings"
"time"
@@ -137,6 +138,7 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
mediaFiles = append(mediaFiles, bus.MediaFile{
Path: mi.FilePath,
MimeType: mi.ContentType,
Filename: mi.FileName,
})
}
}
@@ -253,9 +255,11 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
finalContent = annotated
}
// Collect media from pending history entries (sent before this @mention).
// Original filename not retained by CollectMedia; use disk basename so
// persistMedia's sanitizer gets a meaningful stem instead of UUID fallback.
if histMediaPaths := c.GroupHistory().CollectMedia(channelID); len(histMediaPaths) > 0 {
for _, p := range histMediaPaths {
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p})
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p, Filename: filepath.Base(p)})
}
}
}
+4 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log/slog"
"path/filepath"
"time"
"github.com/nextlevelbuilder/goclaw/internal/bus"
@@ -251,7 +252,8 @@ func (c *Channel) handleMessageEvent(ctx context.Context, event *MessageEvent) {
if mc.ChatType == "group" && c.HistoryLimit() > 0 {
if histMediaPaths := c.GroupHistory().CollectMedia(chatID); len(histMediaPaths) > 0 {
for _, p := range histMediaPaths {
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p}) // cannot use append(slice, other...) — different types
// Original filename not retained in pending-history paths; fall back to basename.
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p, Filename: filepath.Base(p)}) // cannot use append(slice, other...) — different types
}
}
}
@@ -288,6 +290,7 @@ func (c *Channel) handleMessageEvent(ctx context.Context, event *MessageEvent) {
mediaFiles = append(mediaFiles, bus.MediaFile{
Path: m.FilePath,
MimeType: m.ContentType,
Filename: m.FileName,
})
}
}
+5 -1
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"log/slog"
"path/filepath"
"regexp"
"time"
@@ -24,7 +25,10 @@ func (c *Channel) HandleMessage(senderID, chatID, content string, mediaPaths []s
var mediaFiles []bus.MediaFile
for _, p := range mediaPaths {
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p})
// Slack file API doesn't expose a separate original filename here — the
// downloaded temp file is already named after the Slack file name, so
// basename is the best source for persistMedia's sanitizer.
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p, Filename: filepath.Base(p)})
}
// Collect contact for processed messages (DM + group-mentioned).
+2
View File
@@ -430,6 +430,7 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
mediaFiles = append(mediaFiles, bus.MediaFile{
Path: m.FilePath,
MimeType: m.ContentType,
Filename: m.FileName,
})
}
}
@@ -495,6 +496,7 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
mediaFiles = append(mediaFiles, bus.MediaFile{
Path: m.FilePath,
MimeType: m.ContentType,
Filename: m.FileName,
})
}
if len(histMedia) > 0 {
+1 -1
View File
@@ -125,7 +125,7 @@ func (c *Channel) handleIncomingMessage(evt *events.Message) {
for _, m := range mediaList {
if m.FilePath != "" {
mediaFiles = append(mediaFiles, bus.MediaFile{
Path: m.FilePath, MimeType: m.ContentType,
Path: m.FilePath, MimeType: m.ContentType, Filename: m.FileName,
})
}
}
+5
View File
@@ -82,6 +82,11 @@ func clientCanReceiveEvent(c *Client, event bus.Event) bool {
return true
}
// Immediate trace status events: broadcast to all tenant clients (no per-user routing).
if event.Name == protocol.EventTraceStatusChanged {
return true
}
// Team events: filter by TeamID.
if strings.HasPrefix(event.Name, "team.") || strings.HasPrefix(event.Name, "delegation.") {
if tid := extractTeamID(event); tid != "" {
+51 -13
View File
@@ -214,7 +214,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re
// Create cancellable context for abort support (matching TS AbortController pattern).
runCtx, cancel := context.WithCancel(runCtxBase)
injectCh := m.agents.RegisterRun(runID, sessionKey, params.AgentID, cancel)
injectCh := m.agents.RegisterRun(runCtxBase, runID, sessionKey, params.AgentID, cancel)
// Run agent asynchronously - events are broadcast via the event system
go func() {
@@ -230,7 +230,7 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re
var mediaInfos []media.MediaInfo
for _, item := range items {
mimeType := media.DetectMIMEType(item.Path)
mediaFiles = append(mediaFiles, bus.MediaFile{Path: item.Path, MimeType: mimeType})
mediaFiles = append(mediaFiles, bus.MediaFile{Path: item.Path, MimeType: mimeType, Filename: item.Filename})
mediaInfos = append(mediaInfos, media.MediaInfo{
Type: media.MediaKindFromMime(mimeType),
FilePath: item.Path,
@@ -261,6 +261,11 @@ func (m *ChatMethods) handleSend(ctx context.Context, client *gateway.Client, re
UserID: userID,
Stream: params.Stream,
InjectCh: injectCh,
// Wire trace ID back to the active run so force-abort can mark the
// correct trace as cancelled if the goroutine does not exit within 3s.
OnTraceCreated: func(traceID uuid.UUID) {
m.agents.SetRunTraceID(runID, traceID)
},
})
if err != nil {
@@ -418,7 +423,8 @@ func (m *ChatMethods) handleInject(ctx context.Context, client *gateway.Client,
//
// Response:
//
// { ok: true, aborted: bool, runIds: []string }
// { ok: true, aborted: bool, stopped: bool, forced: bool,
// alreadyAborting: bool, notFound: bool, unauthorized: bool, runIds: []string }
func (m *ChatMethods) handleAbort(ctx context.Context, client *gateway.Client, req *protocol.RequestFrame) {
locale := store.LocaleFromContext(ctx)
var params struct {
@@ -446,21 +452,53 @@ func (m *ChatMethods) handleAbort(ctx context.Context, client *gateway.Client, r
return
}
var abortedIDs []string
isAdmin := canSeeAll(client.Role(), m.cfg.Gateway.OwnerIDs, client.UserID())
// Collect abort results.
var results []agent.AbortResult
if params.RunID != "" {
// Abort specific run (with sessionKey authorization)
if m.agents.AbortRun(params.RunID, params.SessionKey) {
abortedIDs = append(abortedIDs, params.RunID)
}
results = []agent.AbortResult{m.agents.AbortRun(params.RunID, params.SessionKey)}
} else {
// Abort all runs for session
abortedIDs = m.agents.AbortRunsForSession(params.SessionKey)
results = m.agents.AbortRunsForSession(params.SessionKey)
}
// Aggregate counts and run IDs.
var runIDs []string
stopped, forced, alreadyAborting, notFound, unauthorized := 0, 0, 0, 0, 0
for _, r := range results {
runIDs = append(runIDs, r.RunID)
switch {
case r.Stopped:
stopped++
case r.Forced:
forced++
case r.AlreadyAborting:
alreadyAborting++
case r.NotFound:
notFound++
case r.Unauthorized:
unauthorized++
slog.Warn("chat.abort: unauthorized run abort attempt",
"runId", r.RunID, "userID", client.UserID())
}
}
// Security: collapse Unauthorized → NotFound for non-admin callers so run
// existence is not leaked to unprivileged clients.
respUnauthorized := unauthorized
if !isAdmin && unauthorized > 0 {
notFound += unauthorized
respUnauthorized = 0
}
client.SendResponse(protocol.NewOKResponse(req.ID, map[string]any{
"ok": true,
"aborted": len(abortedIDs) > 0,
"runIds": abortedIDs,
"ok": true,
"aborted": stopped+forced > 0,
"stopped": stopped > 0,
"forced": forced > 0,
"alreadyAborting": alreadyAborting > 0,
"notFound": notFound > 0 && stopped+forced+alreadyAborting == 0,
"unauthorized": respUnauthorized > 0,
"runIds": runIDs,
}))
}
+8
View File
@@ -31,6 +31,14 @@ func init() {
MsgUserIDRequired: "user_id is required",
MsgMsgRequired: "message is required",
// Abort
MsgAbortStopped: "run stopped",
MsgAbortForced: "run force-aborted (3s grace exceeded)",
MsgAbortAlreadyAborting: "abort already in progress",
MsgAbortNotFound: "run not found or already finished",
MsgAbortUnauthorized: "not authorized to abort this run",
MsgAbortFailed: "failed to abort run: %s",
// Channel instances
MsgInvalidChannelType: "invalid channel_type",
MsgInstanceNotFound: "instance not found",
+8
View File
@@ -31,6 +31,14 @@ func init() {
MsgUserIDRequired: "user_id là bắt buộc",
MsgMsgRequired: "tin nhắn là bắt buộc",
// Abort
MsgAbortStopped: "đã dừng tác vụ",
MsgAbortForced: "buộc dừng tác vụ (vượt quá thời gian chờ 3s)",
MsgAbortAlreadyAborting: "đang dừng tác vụ",
MsgAbortNotFound: "không tìm thấy tác vụ hoặc đã kết thúc",
MsgAbortUnauthorized: "không có quyền dừng tác vụ này",
MsgAbortFailed: "không thể dừng tác vụ: %s",
// Channel instances
MsgInvalidChannelType: "loại channel không hợp lệ",
MsgInstanceNotFound: "không tìm thấy phiên bản",
+8
View File
@@ -31,6 +31,14 @@ func init() {
MsgUserIDRequired: "user_id 是必填项",
MsgMsgRequired: "消息是必填项",
// Abort
MsgAbortStopped: "已停止运行",
MsgAbortForced: "已强制中止运行(超过 3 秒宽限期)",
MsgAbortAlreadyAborting: "正在中止中",
MsgAbortNotFound: "运行未找到或已结束",
MsgAbortUnauthorized: "无权中止此运行",
MsgAbortFailed: "无法中止运行:%s",
// Channel instances
MsgInvalidChannelType: "Channel类型无效",
MsgInstanceNotFound: "未找到实例",
+8
View File
@@ -32,6 +32,14 @@ const (
MsgUserIDRequired = "error.user_id_required" // "user_id is required"
MsgMsgRequired = "error.message_required" // "message is required"
// --- Abort ---
MsgAbortStopped = "abort.stopped" // "run stopped"
MsgAbortForced = "abort.forced" // "run force-aborted (3s grace exceeded)"
MsgAbortAlreadyAborting = "abort.already_aborting" // "abort already in progress"
MsgAbortNotFound = "abort.not_found" // "run not found or already finished"
MsgAbortUnauthorized = "abort.unauthorized" // "not authorized to abort this run"
MsgAbortFailed = "abort.failed" // "failed to abort run: %s"
// --- Channel instances ---
MsgInvalidChannelType = "error.invalid_channel_type" // "invalid channel_type"
MsgInstanceNotFound = "error.instance_not_found" // "instance not found"
+3 -1
View File
@@ -1,6 +1,7 @@
package orchestration
import (
"path/filepath"
"time"
"github.com/nextlevelbuilder/goclaw/internal/agent"
@@ -48,7 +49,8 @@ func CaptureFromPipelineResult(r *plpkg.RunResult, runtime time.Duration) ChildR
}
media := make([]bus.MediaFile, 0, len(r.MediaResults))
for _, m := range r.MediaResults {
media = append(media, bus.MediaFile{Path: m.Path, MimeType: m.ContentType})
// Basename preserves any sanitized stem from the producing agent's persistMedia.
media = append(media, bus.MediaFile{Path: m.Path, MimeType: m.ContentType, Filename: filepath.Base(m.Path)})
}
return ChildResult{
Content: r.Content,
+5
View File
@@ -1,11 +1,15 @@
package orchestration
import (
"path/filepath"
"github.com/nextlevelbuilder/goclaw/internal/agent"
"github.com/nextlevelbuilder/goclaw/internal/bus"
)
// MediaResultToBusFiles converts agent.MediaResult slice to bus.MediaFile slice.
// Filename is derived from the disk basename so downstream persistMedia
// sanitizer keeps a meaningful stem (UUID-only disk names become UUID fallback).
func MediaResultToBusFiles(results []agent.MediaResult) []bus.MediaFile {
if len(results) == 0 {
return nil
@@ -15,6 +19,7 @@ func MediaResultToBusFiles(results []agent.MediaResult) []bus.MediaFile {
files[i] = bus.MediaFile{
Path: r.Path,
MimeType: r.ContentType,
Filename: filepath.Base(r.Path),
}
}
return files
+1 -1
View File
@@ -60,7 +60,7 @@ func NewAnthropicProvider(apiKey string, opts ...AnthropicOption) *AnthropicProv
apiKey: apiKey,
baseURL: anthropicAPIBase,
defaultModel: defaultClaudeModel,
client: &http.Client{Timeout: DefaultHTTPTimeout},
client: NewDefaultHTTPClient(),
retryConfig: DefaultRetryConfig(),
// No CacheMiddleware: Anthropic uses block-level cache_control in buildRequestBody
middlewares: ComposeMiddlewares(FastModeMiddleware, ServiceTierMiddleware),
+4 -2
View File
@@ -26,7 +26,9 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onC
if err != nil {
return nil, err
}
defer respBody.Close()
// Wrap respBody so ctx cancellation closes the socket, unblocking bufio.Scanner.
cb := NewCtxBody(ctx, respBody)
defer cb.Close()
result := &ChatResponse{FinishReason: "stop"}
// Accumulate raw JSON fragments for each tool call by index
@@ -39,7 +41,7 @@ func (p *AnthropicProvider) ChatStream(ctx context.Context, req ChatRequest, onC
thinkingChars := 0
var thinkingSignature strings.Builder
sse := NewSSEScanner(respBody)
sse := NewSSEScanner(cb)
for sse.Next() {
if ctx.Err() != nil {
return nil, ctx.Err()
+5 -3
View File
@@ -43,7 +43,7 @@ func NewCodexProvider(name string, tokenSource TokenSource, apiBase, defaultMode
name: name,
apiBase: apiBase,
defaultModel: defaultModel,
client: &http.Client{Timeout: DefaultHTTPTimeout},
client: NewDefaultHTTPClient(),
retryConfig: DefaultRetryConfig(),
tokenSource: tokenSource,
}
@@ -131,13 +131,15 @@ func (p *CodexProvider) ChatStream(ctx context.Context, req ChatRequest, onChunk
if err != nil {
return nil, err
}
defer respBody.Close()
// Wrap respBody so ctx cancellation closes the socket, unblocking bufio.Scanner.
cb := NewCtxBody(ctx, respBody)
defer cb.Close()
result := &ChatResponse{FinishReason: "stop"}
toolCalls := make(map[string]*codexToolCallAcc) // keyed by item_id
streamState := newCodexMessageStreamState()
sse := NewSSEScanner(respBody)
sse := NewSSEScanner(cb)
for sse.Next() {
data := sse.Data()
+30 -5
View File
@@ -1,18 +1,43 @@
package providers
import "time"
import (
"net/http"
"time"
)
// Provider-level defaults for HTTP clients and stream parsing.
const (
// DefaultHTTPTimeout is the HTTP client timeout for LLM API calls.
// 5 minutes allows for long-running streaming responses with extended thinking.
// Deprecated: DefaultHTTPTimeout set a wall-clock socket timeout that prevented
// ctx cancellation from unblocking bufio.Scanner. Use NewDefaultHTTPClient() instead.
DefaultHTTPTimeout = 300 * time.Second
// SSE stream scanner buffer sizes (OpenAI-compat, Anthropic, Codex).
SSEScanBufInit = 64 * 1024 // 64KB initial buffer
SSEScanBufMax = 1024 * 1024 // 1MB max line for large tool call / thinking chunks
SSEScanBufInit = 64 * 1024 // 64KB initial buffer
SSEScanBufMax = 1024 * 1024 // 1MB max line for large tool call / thinking chunks
// Stdio/JSONRPC scanner buffer sizes (Claude CLI, ACP).
StdioScanBufInit = 256 * 1024 // 256KB initial buffer
StdioScanBufMax = 10 * 1024 * 1024 // 10MB max for large protocol messages
)
// NewDefaultTransport returns an http.Transport with per-stage timeouts but no
// overall deadline. The absence of Client.Timeout allows LLM streaming responses
// (extended thinking, long completions) to run indefinitely while ctx cancellation
// still terminates the request promptly via CtxBody.
func NewDefaultTransport() *http.Transport {
return &http.Transport{
Proxy: http.ProxyFromEnvironment,
ResponseHeaderTimeout: 60 * time.Second, // wait for first byte of response
IdleConnTimeout: 90 * time.Second, // close idle keep-alive connections
TLSHandshakeTimeout: 10 * time.Second,
ExpectContinueTimeout: 1 * time.Second,
MaxIdleConns: 100,
MaxIdleConnsPerHost: 10,
}
}
// NewDefaultHTTPClient returns an *http.Client backed by NewDefaultTransport.
// No Client.Timeout is set — rely on ctx deadlines and Transport stage timeouts.
func NewDefaultHTTPClient() *http.Client {
return &http.Client{Transport: NewDefaultTransport()}
}
+4 -2
View File
@@ -84,12 +84,14 @@ func (p *OpenAIProvider) ChatStream(ctx context.Context, req ChatRequest, onChun
if err != nil {
return nil, err
}
defer respBody.Close()
// Wrap respBody so ctx cancellation closes the socket, unblocking bufio.Scanner.
cb := NewCtxBody(ctx, respBody)
defer cb.Close()
result := &ChatResponse{FinishReason: "stop"}
accumulators := make(map[int]*toolCallAccumulator)
sse := NewSSEScanner(respBody)
sse := NewSSEScanner(cb)
for sse.Next() {
data := sse.Data()
+1 -1
View File
@@ -35,7 +35,7 @@ func NewOpenAIProvider(name, apiKey, apiBase, defaultModel string) *OpenAIProvid
apiBase: apiBase,
chatPath: "/chat/completions",
defaultModel: defaultModel,
client: &http.Client{Timeout: DefaultHTTPTimeout},
client: NewDefaultHTTPClient(),
retryConfig: DefaultRetryConfig(),
middlewares: ComposeMiddlewares(FastModeMiddleware, ServiceTierMiddleware, CacheMiddleware),
}
+42
View File
@@ -2,8 +2,10 @@ package providers
import (
"bufio"
"context"
"io"
"strings"
"sync"
)
// SSEScanner reads an SSE (Server-Sent Events) stream line by line,
@@ -74,3 +76,43 @@ func (s *SSEScanner) EventType() string {
func (s *SSEScanner) Err() error {
return s.err
}
// CtxBody wraps an http.Response.Body so that ctx cancellation closes the
// underlying socket, unblocking a goroutine stuck inside bufio.Scanner.Scan().
// Safe for concurrent Close (sync.Once). Caller MUST defer Close() to release
// the watchdog goroutine even on success.
type CtxBody struct {
body io.ReadCloser
done chan struct{}
once sync.Once
}
// NewCtxBody returns a ReadCloser that closes body when ctx is cancelled.
func NewCtxBody(ctx context.Context, body io.ReadCloser) *CtxBody {
cb := &CtxBody{body: body, done: make(chan struct{})}
go func() {
select {
case <-ctx.Done():
cb.closeOnce()
case <-cb.done:
// normal close path; watchdog exits cleanly
}
}()
return cb
}
func (cb *CtxBody) Read(p []byte) (int, error) { return cb.body.Read(p) }
// Close closes the underlying body exactly once (safe for concurrent calls).
func (cb *CtxBody) Close() error {
return cb.closeOnce()
}
func (cb *CtxBody) closeOnce() error {
var err error
cb.once.Do(func() {
close(cb.done)
err = cb.body.Close()
})
return err
}
+169
View File
@@ -0,0 +1,169 @@
package providers
import (
"context"
"errors"
"io"
"runtime"
"sync"
"sync/atomic"
"testing"
"time"
)
// countingCloser wraps an io.ReadCloser and counts Close calls.
type countingCloser struct {
mu sync.Mutex
closeCount int
reader io.Reader
}
func newCountingCloser(r io.Reader) *countingCloser {
return &countingCloser{reader: r}
}
func (c *countingCloser) Read(p []byte) (int, error) {
return c.reader.Read(p)
}
func (c *countingCloser) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
c.closeCount++
return nil
}
func (c *countingCloser) Count() int {
c.mu.Lock()
defer c.mu.Unlock()
return c.closeCount
}
// blockingReader blocks Read until Close is called (simulates a live TCP socket).
type blockingReader struct {
ch chan struct{}
closed atomic.Bool
}
func newBlockingReader() *blockingReader {
return &blockingReader{ch: make(chan struct{})}
}
func (b *blockingReader) Read(p []byte) (int, error) {
<-b.ch
return 0, errors.New("connection closed")
}
func (b *blockingReader) Close() error {
if b.closed.CompareAndSwap(false, true) {
close(b.ch)
}
return nil
}
// TestCtxBody_CancelClosesBody verifies that ctx cancellation calls body.Close exactly once
// and that a blocked Read returns an error.
func TestCtxBody_CancelClosesBody(t *testing.T) {
t.Parallel()
br := newBlockingReader()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cb := NewCtxBody(ctx, br)
// Cancel context after a short delay.
go func() {
time.Sleep(20 * time.Millisecond)
cancel()
}()
// Read should unblock when Close is called by the watchdog.
buf := make([]byte, 8)
_, err := cb.Read(buf)
if err == nil {
t.Fatal("expected error from Read after ctx cancel, got nil")
}
// Close should be a no-op (already closed by watchdog).
if err2 := cb.Close(); err2 != nil {
t.Fatalf("second Close returned error: %v", err2)
}
// Underlying body should be closed exactly once.
if !br.closed.Load() {
t.Fatal("expected underlying body to be closed")
}
}
// TestCtxBody_NormalClose_WatchdogExits verifies that calling Close() directly
// releases the watchdog goroutine (no goroutine leak).
func TestCtxBody_NormalClose_WatchdogExits(t *testing.T) {
t.Parallel()
before := runtime.NumGoroutine()
const n = 20
var wg sync.WaitGroup
wg.Add(n)
for range n {
go func() {
defer wg.Done()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cc := newCountingCloser(io.NopCloser(io.LimitReader(nopReader{}, 0)))
cb := NewCtxBody(ctx, cc)
// Normal close — watchdog should exit via done channel.
if err := cb.Close(); err != nil {
t.Errorf("Close returned error: %v", err)
}
}()
}
wg.Wait()
// Give goroutines time to exit.
time.Sleep(50 * time.Millisecond)
runtime.GC()
after := runtime.NumGoroutine()
// Allow a small tolerance for unrelated goroutines.
if after > before+5 {
t.Errorf("goroutine leak: before=%d after=%d (delta=%d)", before, after, after-before)
}
}
// TestCtxBody_ConcurrentCloseAndCancel verifies no double-close panic under
// concurrent Close() and ctx cancellation. Run with -race flag.
func TestCtxBody_ConcurrentCloseAndCancel(t *testing.T) {
t.Parallel()
const iterations = 500
for range iterations {
cc := newCountingCloser(io.NopCloser(io.LimitReader(nopReader{}, 0)))
ctx, cancel := context.WithCancel(context.Background())
cb := NewCtxBody(ctx, cc)
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
_ = cb.Close()
}()
go func() {
defer wg.Done()
cancel()
}()
wg.Wait()
// Body must be closed exactly once regardless of race.
if cnt := cc.Count(); cnt != 1 {
t.Fatalf("expected close count=1, got %d", cnt)
}
}
}
// nopReader is an io.Reader that always returns (0, io.EOF).
type nopReader struct{}
func (nopReader) Read([]byte) (int, error) { return 0, io.EOF }
+1 -1
View File
@@ -175,7 +175,7 @@ func (t *CreateAudioTool) Execute(ctx context.Context, args map[string]any) *Res
}
result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s\nUse the EXACT filename when referencing: %s", audioPath, filepath.Base(audioPath))}
result.Media = []bus.MediaFile{{Path: audioPath, MimeType: "audio/mpeg"}}
result.Media = []bus.MediaFile{{Path: audioPath, MimeType: "audio/mpeg", Filename: filepath.Base(audioPath)}}
result.Deliverable = fmt.Sprintf("[Generated audio: %s]\nPrompt: %s", filepath.Base(audioPath), prompt)
if t.vaultIntc != nil {
go t.vaultIntc.AfterWriteMedia(context.WithoutCancel(ctx), audioPath, prompt, "audio/mpeg")
+1 -1
View File
@@ -127,7 +127,7 @@ func (t *CreateImageTool) Execute(ctx context.Context, args map[string]any) *Res
}
result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s\nUse the EXACT filename when referencing: %s", imagePath, filepath.Base(imagePath))}
result.Media = []bus.MediaFile{{Path: imagePath, MimeType: "image/png"}}
result.Media = []bus.MediaFile{{Path: imagePath, MimeType: "image/png", Filename: filepath.Base(imagePath)}}
result.Deliverable = fmt.Sprintf("[Generated image: %s]\nPrompt: %s", filepath.Base(imagePath), prompt)
if t.vaultIntc != nil {
go t.vaultIntc.AfterWriteMedia(context.WithoutCancel(ctx), imagePath, prompt, "image/png")
+1 -1
View File
@@ -195,7 +195,7 @@ func (t *CreateVideoTool) Execute(ctx context.Context, args map[string]any) *Res
}
result := &Result{ForLLM: fmt.Sprintf("MEDIA:%s\nUse the EXACT filename when referencing: %s", videoPath, filepath.Base(videoPath))}
result.Media = []bus.MediaFile{{Path: videoPath, MimeType: "video/mp4"}}
result.Media = []bus.MediaFile{{Path: videoPath, MimeType: "video/mp4", Filename: filepath.Base(videoPath)}}
result.Deliverable = fmt.Sprintf("[Generated video: %s]\nPrompt: %s", filepath.Base(videoPath), prompt)
if t.vaultIntc != nil {
go t.vaultIntc.AfterWriteMedia(context.WithoutCancel(ctx), videoPath, prompt, "video/mp4")
+30 -3
View File
@@ -273,15 +273,20 @@ func (t *ExecTool) executeCredentialed(ctx context.Context, cred *store.SecureCL
// executeCredentialedHost runs a credentialed command directly on the host.
// Uses exec.Command (no shell) with credentials as env vars.
// ctx cancellation triggers SIGTERM → 3s grace → SIGKILL via process-group helpers.
func (t *ExecTool) executeCredentialedHost(ctx context.Context, absPath string, args []string,
cwd string, envMap map[string]string, timeout time.Duration) *Result {
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
cmd := exec.CommandContext(ctx, absPath, args...)
// Plain exec.Command (not CommandContext) so we own the kill sequence.
cmd := exec.Command(absPath, args...)
cmd.Dir = cwd
// Process group so abort reaches the whole tree (mirrors executeOnHost).
setProcessGroup(cmd)
// Build env: inherit minimal PATH + HOME, add credentials
cmd.Env = buildCredentialedEnv(envMap)
@@ -289,8 +294,30 @@ func (t *ExecTool) executeCredentialedHost(ctx context.Context, absPath string,
cmd.Stdout = &stdout
cmd.Stderr = &stderr
err := cmd.Run()
return formatCredentialedResult(absPath, args, stdout.String(), stderr.String(), err, ctx, timeout)
if err := cmd.Start(); err != nil {
return ErrorResult(fmt.Sprintf("credentialed exec: failed to start %s: %v", absPath, err))
}
done := make(chan error, 1)
go func() { done <- cmd.Wait() }()
select {
case err := <-done:
return formatCredentialedResult(absPath, args, stdout.String(), stderr.String(), err, ctx, timeout)
case <-ctx.Done():
_ = killProcessGroup(cmd, syscallSIGTERM)
select {
case <-done:
case <-time.After(3 * time.Second):
_ = killProcessGroup(cmd, syscallSIGKILL)
<-done
}
if errors.Is(ctx.Err(), context.DeadlineExceeded) {
return ErrorResult(fmt.Sprintf("[CREDENTIALED EXEC] Command timed out after %s.\nBinary: %s", timeout, absPath))
}
return ErrorResult(fmt.Sprintf("[CREDENTIALED EXEC] Command aborted.\nBinary: %s", absPath))
}
}
// executeCredentialedSandbox runs a credentialed command inside a Docker sandbox.
+2 -2
View File
@@ -230,7 +230,7 @@ func (t *WriteFileTool) Execute(ctx context.Context, args map[string]any) *Resul
result := SilentResult(msg)
result.Deliverable = content
if deliver {
result.Media = []bus.MediaFile{{Path: resolved}}
result.Media = []bus.MediaFile{{Path: resolved, Filename: filepath.Base(resolved)}}
// Track delivered path so message tool's self-send guard can detect duplicates.
if dm := DeliveredMediaFromCtx(ctx); dm != nil {
dm.Mark(resolved)
@@ -276,7 +276,7 @@ func (t *WriteFileTool) executeInSandbox(ctx context.Context, path, content, san
workspace = t.workspace
}
hostPath := filepath.Join(workspace, path)
result.Media = []bus.MediaFile{{Path: hostPath}}
result.Media = []bus.MediaFile{{Path: hostPath, Filename: filepath.Base(hostPath)}}
if dm := DeliveredMediaFromCtx(ctx); dm != nil {
dm.Mark(hostPath)
}
+41 -4
View File
@@ -307,21 +307,59 @@ func matchesAny(command string, patterns []*regexp.Regexp) bool {
}
// executeOnHost runs a command directly on the host (original behavior).
// ctx cancellation (e.g. agent abort) triggers SIGTERM → 3s grace → SIGKILL on the
// entire process group so forked children are also cleaned up (no orphans).
func (t *ExecTool) executeOnHost(ctx context.Context, command, cwd string) *Result {
ctx, cancel := context.WithTimeout(ctx, t.timeout)
defer cancel()
cmd := exec.CommandContext(ctx, "sh", "-c", command)
// Use plain exec.Command (not CommandContext) so we control the kill sequence.
// CommandContext would SIGKILL only the direct child, leaving forked grandchildren alive.
cmd := exec.Command("sh", "-c", command)
cmd.Dir = cwd
// Place the child in its own process group so killProcessGroup(-pgid, sig)
// reaches the shell and all of its forked children.
setProcessGroup(cmd)
// Limit output to 1MB to prevent OOM from runaway commands.
stdout := &limitedBuffer{max: 1 << 20}
stderr := &limitedBuffer{max: 1 << 20}
cmd.Stdout = stdout
cmd.Stderr = stderr
err := cmd.Run()
if err := cmd.Start(); err != nil {
return ErrorResult(fmt.Sprintf("failed to start command: %v", err))
}
done := make(chan error, 1)
go func() { done <- cmd.Wait() }()
select {
case err := <-done:
// Normal completion (success or non-zero exit).
return buildHostResult(err, stdout, stderr, ctx, t.timeout)
case <-ctx.Done():
// Context cancelled or timed out — kill the process group gracefully then forcefully.
_ = killProcessGroup(cmd, syscallSIGTERM)
select {
case <-done:
// Exited cleanly after SIGTERM.
case <-time.After(3 * time.Second):
// Still alive after grace period — force kill.
_ = killProcessGroup(cmd, syscallSIGKILL)
<-done
}
if errors.Is(ctx.Err(), context.DeadlineExceeded) {
return ErrorResult(fmt.Sprintf("command timed out after %s", t.timeout))
}
return ErrorResult("command aborted")
}
}
// buildHostResult formats the result of a completed host command execution.
func buildHostResult(err error, stdout, stderr *limitedBuffer, ctx context.Context, timeout time.Duration) *Result {
var result string
if stdout.Len() > 0 {
result = stdout.String()
@@ -335,7 +373,7 @@ func (t *ExecTool) executeOnHost(ctx context.Context, command, cwd string) *Resu
if err != nil {
if errors.Is(ctx.Err(), context.DeadlineExceeded) {
return ErrorResult(fmt.Sprintf("command timed out after %s", t.timeout))
return ErrorResult(fmt.Sprintf("command timed out after %s", timeout))
}
if result == "" {
result = err.Error()
@@ -346,7 +384,6 @@ func (t *ExecTool) executeOnHost(ctx context.Context, command, cwd string) *Resu
if result == "" {
result = "(command completed with no output)"
}
return SilentResult(capExecOutput(result, execMaxOutputChars))
}
+93
View File
@@ -0,0 +1,93 @@
//go:build !windows
package tools
import (
"context"
"fmt"
"os/exec"
"runtime"
"strings"
"testing"
"time"
)
// TestShellAbort_ProcessGroupKilled verifies that cancelling ctx after spawning a
// command that forks background children kills the entire process group within 4s
// and leaves no orphan sleep processes.
//
// The command `sh -c "sleep 60 & sleep 60 & wait"` spawns two background sleeps.
// With process-group kill, both background sleeps must die when ctx is cancelled.
func TestShellAbort_ProcessGroupKilled(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("process-group kill not supported on Windows")
}
// Marker to identify our specific sleep processes.
marker := fmt.Sprintf("goclaw_abort_test_%d", time.Now().UnixNano())
command := fmt.Sprintf("sleep 60 & echo 'marker=%s' & sleep 60 & wait", marker)
tool := NewExecTool(t.TempDir(), false)
tool.timeout = 10 * time.Second // generous outer timeout
ctx, cancel := context.WithCancel(context.Background())
done := make(chan *Result, 1)
go func() {
done <- tool.executeOnHost(ctx, command, t.TempDir())
}()
// Give the shell time to fork the sleep processes.
time.Sleep(100 * time.Millisecond)
// Cancel ctx — should trigger SIGTERM → 3s grace → SIGKILL.
cancel()
// Verify tool returns within 4s (3s grace + 1s buffer).
select {
case result := <-done:
if result == nil {
t.Fatal("expected non-nil result after abort")
}
// Result should indicate abortion, not normal completion.
if !result.IsError {
t.Errorf("expected IsError=true after abort, got ForLLM=%q", result.ForLLM)
}
case <-time.After(4 * time.Second):
t.Fatal("executeOnHost did not return within 4s after ctx cancel")
}
// Give the OS a moment to reap the killed processes.
time.Sleep(200 * time.Millisecond)
// Verify no orphan sleep processes remain. We check with `ps` filtering by
// "sleep 60" (the exact argument). pgrep is not reliably available on macOS CI.
orphans := findOrphanSleeps(t)
if len(orphans) > 0 {
t.Errorf("found %d orphan 'sleep 60' process(es) after abort: %v", len(orphans), orphans)
}
}
// findOrphanSleeps returns PIDs of any remaining `sleep 60` processes.
// Uses `ps aux` output parsed in Go — avoids pgrep availability issues on macOS.
func findOrphanSleeps(t *testing.T) []string {
t.Helper()
out, err := exec.Command("ps", "aux").Output()
if err != nil {
t.Logf("ps aux failed (non-fatal): %v", err)
return nil
}
var found []string
for _, line := range strings.Split(string(out), "\n") {
// Match lines containing "sleep 60" but not the grep/ps command itself.
if strings.Contains(line, "sleep 60") && !strings.Contains(line, "ps aux") {
fields := strings.Fields(line)
if len(fields) >= 2 {
found = append(found, fields[1]) // PID is column 2 in ps aux
}
}
}
return found
}
+30
View File
@@ -0,0 +1,30 @@
//go:build !windows
package tools
import (
"os/exec"
"syscall"
)
// syscallSIGTERM and syscallSIGKILL are the platform signals used by executeOnHost.
// Defined here (unix) and as zero-value stubs in shell_pgid_windows.go.
const (
syscallSIGTERM = syscall.SIGTERM
syscallSIGKILL = syscall.SIGKILL
)
// setProcessGroup sets Setpgid so the child process becomes its own process-group leader.
// This allows killProcessGroup to signal the entire process tree (shell + forked children).
func setProcessGroup(cmd *exec.Cmd) {
cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
}
// killProcessGroup sends sig to the entire process group rooted at cmd.Process.Pid.
// Because Setpgid was set, pgid == pid, so kill(-pid, sig) reaches all forked children.
func killProcessGroup(cmd *exec.Cmd, sig syscall.Signal) error {
if cmd.Process == nil {
return nil
}
return syscall.Kill(-cmd.Process.Pid, sig)
}
+29
View File
@@ -0,0 +1,29 @@
//go:build windows
package tools
import (
"os/exec"
"syscall"
)
// syscallSIGTERM / syscallSIGKILL are stub values on Windows.
// killProcessGroup ignores the signal and calls cmd.Process.Kill() directly.
const (
syscallSIGTERM = syscall.Signal(0x0f) // SIGTERM value; unused on Windows path
syscallSIGKILL = syscall.Signal(0x09) // SIGKILL value; unused on Windows path
)
// setProcessGroup is a no-op on Windows; process groups work differently and
// Setpgid is not supported by the Windows syscall layer.
func setProcessGroup(cmd *exec.Cmd) {}
// killProcessGroup falls back to terminating the direct child process on Windows.
// Process-tree kill on Windows requires the Job Objects API; single-process kill
// is sufficient for the current use case (host shell exec, no deep fork trees).
func killProcessGroup(cmd *exec.Cmd, _ syscall.Signal) error {
if cmd.Process == nil {
return nil
}
return cmd.Process.Kill()
}
+5 -1
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"fmt"
"log/slog"
"path/filepath"
"strings"
"time"
@@ -241,6 +242,9 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask)
}
slog.Info("subagent LLM retry", "id", task.ID, "iteration", iteration, "attempt", attempt+1)
}
// ctx is the parent agent's run context — cancelling the parent (e.g. agent abort)
// cascades here and to all subsequent tool calls in this iteration.
// Do NOT replace ctx with context.Background() here; that would detach abort propagation.
resp, err = activeProvider.Chat(ctx, chatReq)
if err == nil {
break
@@ -301,7 +305,7 @@ func (sm *SubagentManager) executeTask(ctx context.Context, task *SubagentTask)
p = strings.TrimSpace(p[:nl])
}
if p != "" {
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p})
mediaFiles = append(mediaFiles, bus.MediaFile{Path: p, Filename: filepath.Base(p)})
}
}
+211 -18
View File
@@ -18,8 +18,31 @@ const (
previewMaxLen = 40_000
traceRetention = 7 * 24 * time.Hour // auto-delete traces older than 7 days
pruneInterval = 8 * time.Hour
// staleThreshold: how long a "running" trace must be stuck before the recovery
// worker marks it as "error". 10min is conservative — the primary sub-second
// stop visibility is delivered by the trace.status WS event (Phase 4). Stale
// recovery is a safety net for crashed/orphaned traces. Lowering this further
// requires a `last_span_at` column so we don't sweep legitimate long-running
// agents (see plan's Phase 3 unresolved question).
staleThreshold = 10 * time.Minute
staleRecoveryPeriod = 30 * time.Second // new: run periodically instead of once on startup
retryQueueCap = 1000
retryWorkerPeriod = 5 * time.Second
retryMaxTries = 10
)
// TraceStatusPayload is the payload for EventTraceStatusChanged WS events.
// Sent immediately after a status write succeeds (not buffered by flush interval).
type TraceStatusPayload struct {
TraceID string `json:"traceId"`
Status string `json:"status"`
EndedAt *time.Time `json:"endedAt,omitempty"`
}
// StatusBroadcaster is a callback invoked after each successful trace status write.
// Implementations broadcast the payload to connected WS clients in the given tenant.
type StatusBroadcaster func(payload TraceStatusPayload, tenantID uuid.UUID)
// SpanExporter is implemented by backends that receive span data alongside
// the PostgreSQL store (e.g. OpenTelemetry OTLP). Keeping this as an
// interface lets the OTel dependency live in a separate sub-package that can
@@ -37,6 +60,15 @@ type spanUpdate struct {
Updates map[string]any
}
// pendingUpdate holds a trace update that failed to persist and is queued for retry.
type pendingUpdate struct {
TraceID uuid.UUID
TenantID uuid.UUID // captured from caller ctx at enqueue time for retryWorker broadcast
Updates map[string]any
Tries int
LastTry time.Time
}
// Collector buffers spans in memory and periodically flushes them to the
// TracingStore in batches. Traces are created synchronously (one per run),
// while spans are buffered for async batch insert.
@@ -51,6 +83,10 @@ type Collector struct {
stopCh chan struct{}
wg sync.WaitGroup
// retryCh buffers trace updates that failed all inline retries.
// Bounded at retryQueueCap — drop-oldest under sustained DB outage.
retryCh chan pendingUpdate
// traces that need aggregate updates on flush
dirtyTraces map[uuid.UUID]struct{}
dirtyTracesMu sync.Mutex
@@ -61,6 +97,11 @@ type Collector struct {
// OnFlush is called after each flush cycle with the trace IDs that had
// their aggregates updated. Used to broadcast realtime trace events.
OnFlush func(traceIDs []uuid.UUID)
// broadcastStatus is an optional callback invoked immediately after each
// successful trace status write (SetTraceStatus / FinishTrace). Fires
// before the 5s OnFlush tick for low-latency status delivery.
broadcastStatus StatusBroadcaster
}
// NewCollector creates a new tracing collector backed by the given store.
@@ -75,6 +116,7 @@ func NewCollector(ts store.TracingStore) *Collector {
spanCh: make(chan store.SpanData, defaultBufferSize),
spanUpdateCh: make(chan spanUpdate, defaultBufferSize),
stopCh: make(chan struct{}),
retryCh: make(chan pendingUpdate, retryQueueCap),
dirtyTraces: make(map[uuid.UUID]struct{}),
verbose: verbose,
}
@@ -97,11 +139,19 @@ func (c *Collector) SetExporter(exp SpanExporter) {
c.exporter = exp
}
// Start begins the background flush loop.
// SetStatusBroadcaster wires a callback that is invoked immediately after each
// successful trace status write. Called from SetTraceStatus and FinishTrace
// on success, bypassing the 5s OnFlush buffer for low-latency status events.
func (c *Collector) SetStatusBroadcaster(b StatusBroadcaster) {
c.broadcastStatus = b
}
// Start begins the background flush loop, retry worker, and stale recovery loop.
func (c *Collector) Start() {
c.wg.Add(1)
c.wg.Add(3) // flushLoop + retryWorker + staleRecoveryLoop
go c.flushLoop()
go c.recoverStaleTraces() // fix orphan traces from previous crash
go c.retryWorker()
go c.staleRecoveryLoop()
slog.Info("tracing collector started")
}
@@ -166,16 +216,16 @@ func (c *Collector) EmitSpanUpdate(spanID, traceID uuid.UUID, updates map[string
}
// SetTraceStatus updates only the trace status and marks it dirty for re-aggregation.
// Used by child trace runs (e.g. announce) to toggle the parent trace back to
// "running" while the child is active, then "completed" when done.
// Uses a detached context with retries to survive caller context cancellation.
func (c *Collector) SetTraceStatus(ctx context.Context, traceID uuid.UUID, status string) {
if err := c.store.UpdateTrace(ctx, traceID, map[string]any{"status": status}); err != nil {
slog.Warn("tracing: failed to set trace status", "trace_id", traceID, "error", err)
updates := map[string]any{"status": status}
if c.updateTraceWithRetry(ctx, traceID, updates) {
c.markDirty(traceID)
}
c.markDirty(traceID)
}
// FinishTrace marks a trace as completed and schedules aggregate update.
// Uses a detached context with retries to survive caller context cancellation (e.g. abort path).
func (c *Collector) FinishTrace(ctx context.Context, traceID uuid.UUID, status string, errMsg string, outputPreview string) {
now := time.Now().UTC()
updates := map[string]any{
@@ -188,10 +238,112 @@ func (c *Collector) FinishTrace(ctx context.Context, traceID uuid.UUID, status s
if outputPreview != "" {
updates["output_preview"] = c.truncatePreviewStr(outputPreview)
}
if err := c.store.UpdateTrace(ctx, traceID, updates); err != nil {
slog.Warn("tracing: failed to finish trace", "trace_id", traceID, "error", err)
if c.updateTraceWithRetry(ctx, traceID, updates) {
c.markDirty(traceID)
}
}
// updateTraceWithRetry persists trace updates with a detached ctx + 3 inline retries.
// Uses context.WithoutCancel to preserve tenant/locale values while detaching cancellation.
// On final failure, enqueues to the retry queue. Returns true if persisted inline.
func (c *Collector) updateTraceWithRetry(ctx context.Context, traceID uuid.UUID, updates map[string]any) bool {
detached := context.WithoutCancel(ctx)
backoffs := []time.Duration{100 * time.Millisecond, 200 * time.Millisecond, 300 * time.Millisecond}
for attempt := 0; attempt <= len(backoffs); attempt++ {
opCtx, cancel := context.WithTimeout(detached, 5*time.Second)
err := c.store.UpdateTrace(opCtx, traceID, updates)
cancel()
if err == nil {
c.emitStatusBroadcast(ctx, traceID, updates)
return true
}
slog.Warn("tracing: UpdateTrace attempt failed",
"trace_id", traceID, "attempt", attempt+1, "error", err)
if attempt < len(backoffs) {
time.Sleep(backoffs[attempt])
}
}
// All inline retries failed — enqueue for background worker
c.enqueueRetry(ctx, traceID, updates)
return false
}
// enqueueRetry pushes a failed update to the bounded retry channel.
// ctx is used to capture the tenant ID so the retryWorker can broadcast with correct tenant.
// Drop-oldest behavior when the queue is full to prevent unbounded growth.
func (c *Collector) enqueueRetry(ctx context.Context, traceID uuid.UUID, updates map[string]any) {
item := pendingUpdate{
TraceID: traceID,
TenantID: store.TenantIDFromContext(ctx),
Updates: updates,
Tries: 0,
}
select {
case c.retryCh <- item:
default:
// queue full — drop oldest, push new
select {
case dropped := <-c.retryCh:
slog.Warn("tracing: retry queue full, dropped oldest",
"dropped_trace_id", dropped.TraceID, "new_trace_id", traceID)
default:
}
c.retryCh <- item
}
}
// retryWorker drains the retry channel every 5 seconds and re-attempts
// failed trace updates. Items exceeding retryMaxTries are dropped with an error log.
func (c *Collector) retryWorker() {
defer c.wg.Done()
ticker := time.NewTicker(retryWorkerPeriod)
defer ticker.Stop()
var pending []pendingUpdate
for {
select {
case <-c.stopCh:
return
case item := <-c.retryCh:
pending = append(pending, item)
case <-ticker.C:
// drain any newly arrived items
draining := true
for draining {
select {
case item := <-c.retryCh:
pending = append(pending, item)
default:
draining = false
}
}
// attempt all pending updates
kept := pending[:0]
for _, p := range pending {
// Restore tenant scope so UpdateTrace and broadcast use correct tenant.
baseCtx := context.Background()
if p.TenantID != uuid.Nil {
baseCtx = store.WithTenantID(baseCtx, p.TenantID)
}
opCtx, cancel := context.WithTimeout(baseCtx, 5*time.Second)
err := c.store.UpdateTrace(opCtx, p.TraceID, p.Updates)
cancel()
if err == nil {
c.markDirty(p.TraceID)
c.emitStatusBroadcast(baseCtx, p.TraceID, p.Updates)
continue
}
p.Tries++
p.LastTry = time.Now()
if p.Tries < retryMaxTries {
kept = append(kept, p)
} else {
slog.Error("tracing: retry exhausted, dropping update",
"trace_id", p.TraceID, "tries", p.Tries)
}
}
pending = kept
}
}
c.markDirty(traceID)
}
func (c *Collector) markDirty(traceID uuid.UUID) {
@@ -225,22 +377,42 @@ 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)
// staleRecoveryLoop runs recoverStaleOnce immediately on startup, then every
// staleRecoveryPeriod (30s). Replaces the one-shot recoverStaleTraces call.
func (c *Collector) staleRecoveryLoop() {
defer c.wg.Done()
ticker := time.NewTicker(staleRecoveryPeriod)
defer ticker.Stop()
// run immediately on startup too
c.recoverStaleOnce()
for {
select {
case <-c.stopCh:
return
case <-ticker.C:
c.recoverStaleOnce()
}
}
}
// recoverStaleOnce marks "running" traces whose start_time is older than
// staleThreshold (10 min) as "error". Also recovers stuck spans.
//
// NOTE: Both PG and SQLite implementations use start_time < cutoff, not last
// activity time. Follow-up: gate on "no spans in last N min" instead (requires
// a last_span_at schema column). Tracked as an open question.
func (c *Collector) recoverStaleOnce() {
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)
slog.Warn("tracing: stale recovery failed", "error", err)
return
}
if recovered > 0 {
slog.Info("tracing: recovered stale running traces on startup",
slog.Info("tracing: recovered stale running traces",
"count", recovered, "older_than", cutoff.Format(time.RFC3339))
}
}
@@ -340,6 +512,27 @@ doneUpdates:
}
}
// emitStatusBroadcast fires broadcastStatus if set and the updates map contains a "status" key.
// Extracts optional "end_time" from updates for the EndedAt field.
func (c *Collector) emitStatusBroadcast(ctx context.Context, traceID uuid.UUID, updates map[string]any) {
if c.broadcastStatus == nil {
return
}
status, _ := updates["status"].(string)
if status == "" {
return
}
var endedAt *time.Time
if t, ok := updates["end_time"].(time.Time); ok {
endedAt = &t
}
c.broadcastStatus(TraceStatusPayload{
TraceID: traceID.String(),
Status: status,
EndedAt: endedAt,
}, store.TenantIDFromContext(ctx))
}
// truncatePreviewStr sanitizes and truncates a string by removing the middle.
func (c *Collector) truncatePreviewStr(s string) string {
return TruncateMid(s, c.PreviewMaxLen())
+275
View File
@@ -0,0 +1,275 @@
package tracing
import (
"context"
"errors"
"sync/atomic"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// errFakeDB is a sentinel error returned by the mock store.
var errFakeDB = errors.New("fake db error")
// mockTracingStore is a minimal TracingStore stub that counts UpdateTrace calls
// and fails until failUntil attempts have been made.
type mockTracingStore struct {
store.TracingStore // embed to satisfy interface; unused methods panic
updateCalls atomic.Int64
failUntil int // first N calls return errFakeDB; N+1 onwards succeed
}
func (m *mockTracingStore) UpdateTrace(_ context.Context, _ uuid.UUID, _ map[string]any) error {
n := int(m.updateCalls.Add(1))
if n <= m.failUntil {
return errFakeDB
}
return nil
}
func (m *mockTracingStore) RecoverStaleRunningTraces(_ context.Context, _ time.Time) (int64, error) {
return 0, nil
}
// alwaysFailStore always returns an error for UpdateTrace.
type alwaysFailStore struct {
store.TracingStore
calls atomic.Int64
}
func (s *alwaysFailStore) UpdateTrace(_ context.Context, _ uuid.UUID, _ map[string]any) error {
s.calls.Add(1)
return errFakeDB
}
func (s *alwaysFailStore) RecoverStaleRunningTraces(_ context.Context, _ time.Time) (int64, error) {
return 0, nil
}
// newTestCollector creates a Collector with a fake store and no background goroutines.
// Callers must not call Start() unless explicitly testing worker behaviour.
func newTestCollector(ts store.TracingStore) *Collector {
return &Collector{
store: ts,
spanCh: make(chan store.SpanData, defaultBufferSize),
spanUpdateCh: make(chan spanUpdate, defaultBufferSize),
stopCh: make(chan struct{}),
retryCh: make(chan pendingUpdate, retryQueueCap),
dirtyTraces: make(map[uuid.UUID]struct{}),
}
}
// TestUpdateTraceWithRetry_SuccessAfterNFails verifies that when the store fails
// the first 2 calls and succeeds on the 3rd, updateTraceWithRetry returns true
// and nothing is enqueued in the retry channel.
func TestUpdateTraceWithRetry_SuccessAfterNFails(t *testing.T) {
ms := &mockTracingStore{failUntil: 2}
c := newTestCollector(ms)
traceID := uuid.New()
updates := map[string]any{"status": "completed"}
ok := c.updateTraceWithRetry(context.Background(), traceID, updates)
if !ok {
t.Fatal("expected updateTraceWithRetry to return true after success on 3rd attempt")
}
if got := int(ms.updateCalls.Load()); got != 3 {
t.Fatalf("expected 3 UpdateTrace calls, got %d", got)
}
// Nothing should be in the retry queue.
if len(c.retryCh) != 0 {
t.Fatalf("expected empty retry queue, got %d items", len(c.retryCh))
}
}
// TestUpdateTraceWithRetry_CancelledContextSucceeds verifies that a cancelled
// caller context does not prevent the update from succeeding (WithoutCancel).
func TestUpdateTraceWithRetry_CancelledContextSucceeds(t *testing.T) {
ms := &mockTracingStore{failUntil: 0} // succeeds on first call
c := newTestCollector(ms)
ctx, cancel := context.WithCancel(context.Background())
cancel() // cancel immediately
traceID := uuid.New()
ok := c.updateTraceWithRetry(ctx, traceID, map[string]any{"status": "cancelled"})
if !ok {
t.Fatal("expected success even with cancelled ctx")
}
}
// TestUpdateTraceWithRetry_AllFailEnqueues verifies that when all 4 attempts fail,
// the item lands in the retry queue and the function returns false.
func TestUpdateTraceWithRetry_AllFailEnqueues(t *testing.T) {
s := &alwaysFailStore{}
c := newTestCollector(s)
traceID := uuid.New()
ok := c.updateTraceWithRetry(context.Background(), traceID, map[string]any{"status": "error"})
if ok {
t.Fatal("expected false when all retries fail")
}
if len(c.retryCh) != 1 {
t.Fatalf("expected 1 item in retry queue, got %d", len(c.retryCh))
}
}
// TestRetryWorker_DrainAndSucceed verifies that the retryWorker processes
// queued items once the store starts succeeding.
func TestRetryWorker_DrainAndSucceed(t *testing.T) {
// fail first 3 calls so inline retry puts it in queue, then succeed
ms := &mockTracingStore{failUntil: 4} // 4 failures: 1 inline attempt + 3 backoffs → queued; worker attempt 5 succeeds
c := newTestCollector(ms)
// Pre-enqueue directly (simulates worker finding item)
traceID := uuid.New()
c.retryCh <- pendingUpdate{TraceID: traceID, Updates: map[string]any{"status": "done"}}
c.wg.Add(1)
go c.retryWorker()
// Wait for the worker to drain (up to 3 ticker periods)
deadline := time.Now().Add(3 * retryWorkerPeriod)
for time.Now().Before(deadline) {
if ms.updateCalls.Load() >= 1 {
break
}
time.Sleep(10 * time.Millisecond)
}
close(c.stopCh)
c.wg.Wait()
if ms.updateCalls.Load() < 1 {
t.Fatal("worker never attempted UpdateTrace")
}
}
// TestRetryWorker_ExhaustsAndDrops verifies that items failing retryMaxTries times
// are dropped with an error log and not kept in memory forever.
func TestRetryWorker_ExhaustsAndDrops(t *testing.T) {
s := &alwaysFailStore{}
c := newTestCollector(s)
traceID := uuid.New()
// Pre-set tries to retryMaxTries-1 so one more failure drops it.
c.retryCh <- pendingUpdate{
TraceID: traceID,
Updates: map[string]any{"status": "error"},
Tries: retryMaxTries - 1,
}
c.wg.Add(1)
go c.retryWorker()
// Allow one tick
time.Sleep(retryWorkerPeriod + 100*time.Millisecond)
close(c.stopCh)
c.wg.Wait()
// Item should have been dropped (tries reached retryMaxTries on this attempt).
// We can't directly inspect internal pending slice, but verify no goroutine leak
// and store was called at least once.
if s.calls.Load() < 1 {
t.Fatal("worker never attempted UpdateTrace on queued item")
}
}
// TestEnqueueRetry_QueueCapDropsOldest verifies that when the queue is full,
// enqueuing a new item drops the oldest.
func TestEnqueueRetry_QueueCapDropsOldest(t *testing.T) {
c := newTestCollector(&alwaysFailStore{})
// Fill the queue to capacity.
firstID := uuid.New()
c.retryCh <- pendingUpdate{TraceID: firstID, Updates: map[string]any{"status": "old"}}
for i := 1; i < retryQueueCap; i++ {
c.retryCh <- pendingUpdate{TraceID: uuid.New(), Updates: map[string]any{"status": "fill"}}
}
if len(c.retryCh) != retryQueueCap {
t.Fatalf("expected queue at cap %d, got %d", retryQueueCap, len(c.retryCh))
}
// Enqueue one more — should drop firstID (oldest).
newID := uuid.New()
c.enqueueRetry(context.Background(), newID, map[string]any{"status": "new"})
// Queue should still be at cap.
if len(c.retryCh) != retryQueueCap {
t.Fatalf("queue size after drop: got %d, want %d", len(c.retryCh), retryQueueCap)
}
// The first item drained should NOT be firstID (it was dropped); it should be
// one of the fill items or the new item — first item read will be whatever was
// second-oldest (since oldest was dropped and new was appended).
first := <-c.retryCh
if first.TraceID == firstID {
t.Fatal("oldest item should have been dropped, but it is still at the front")
}
}
// TestShutdown_WorkerExitsCleanly verifies that closing stopCh causes retryWorker
// to exit without goroutine leak, even with pending items in the channel.
func TestShutdown_WorkerExitsCleanly(t *testing.T) {
c := newTestCollector(&alwaysFailStore{})
// Put a few items in the retry channel before starting worker.
for i := 0; i < 5; i++ {
c.retryCh <- pendingUpdate{TraceID: uuid.New(), Updates: map[string]any{"status": "pending"}}
}
c.wg.Add(1)
go c.retryWorker()
// Signal shutdown immediately (before next tick).
close(c.stopCh)
done := make(chan struct{})
go func() {
c.wg.Wait()
close(done)
}()
select {
case <-done:
// clean exit
case <-time.After(2 * time.Second):
t.Fatal("retryWorker did not exit within 2 seconds after stopCh closed")
}
}
// TestSetTraceStatus_UsesRetry verifies SetTraceStatus succeeds with cancelled ctx
// (regression gate for the original silent-swallow bug).
func TestSetTraceStatus_UsesRetry(t *testing.T) {
ms := &mockTracingStore{failUntil: 0}
c := newTestCollector(ms)
ctx, cancel := context.WithCancel(context.Background())
cancel()
c.SetTraceStatus(ctx, uuid.New(), "cancelled")
if ms.updateCalls.Load() != 1 {
t.Fatalf("expected 1 UpdateTrace call, got %d", ms.updateCalls.Load())
}
}
// TestFinishTrace_UsesRetry verifies FinishTrace succeeds with cancelled ctx.
func TestFinishTrace_UsesRetry(t *testing.T) {
ms := &mockTracingStore{failUntil: 0}
c := newTestCollector(ms)
ctx, cancel := context.WithCancel(context.Background())
cancel()
c.FinishTrace(ctx, uuid.New(), "completed", "", "some output")
if ms.updateCalls.Load() != 1 {
t.Fatalf("expected 1 UpdateTrace call, got %d", ms.updateCalls.Load())
}
}
@@ -0,0 +1,16 @@
//go:build integration
package tracing
// RetryQueueLen returns the current length of the retry queue.
// Used only in integration tests to verify failed updates are enqueued.
func (c *Collector) RetryQueueLen() int {
return len(c.retryCh)
}
// RecoverStaleNow manually triggers the stale recovery process once.
// Used in integration tests to verify stale trace recovery without waiting
// for the periodic 30s ticker.
func (c *Collector) RecoverStaleNow() {
c.recoverStaleOnce()
}
+45
View File
@@ -0,0 +1,45 @@
package vault
import "testing"
// TestShouldSkipEnrichment_FilenamePreservation verifies the skip filter's
// contract against disk-name patterns produced by agent.persistMedia:
//
// `{slug}-{8hex}{ext}` — preserved filenames; MUST NOT skip (carries meaning).
// `{uuid}{ext}` — voice-note / clipboard fallback; MUST skip (noise).
//
// This pairs with internal/agent/media_persist_test.go which validates the
// disk-write side, closing the round-trip: sanitize → persist → enrich passes.
func TestShouldSkipEnrichment_FilenamePreservation(t *testing.T) {
cases := []struct {
name string
basename string
want bool // true = skip
}{
// Preserved names — must pass through to enrichment.
{"vietnamese_slug", "bao-cao-q4-a1b2c3d4.pdf", false},
{"cjk_preserved", "猫の写真-a1b2c3d4.png", false},
{"collision_stem_a", "report-11112222.pdf", false},
{"collision_stem_b", "report-33334444.pdf", false},
{"short_word_with_hex", "cat-a1b2c3d4.jpg", false},
{"arabic_stem", "تقرير-a1b2c3d4.pdf", false},
// Fallbacks — intentionally skipped.
{"voice_note_uuid", "550e8400-e29b-41d4-a716-446655440000.ogg", true},
{"clipboard_uuid_png", "00000000-0000-0000-0000-000000000001.png", true},
// Other pre-existing skip rules must still fire.
{"goclaw_gen", "goclaw_gen_12345.png", true},
{"too_short", "ab.pdf", true},
{"digits_only", "12345.pdf", true},
{"pure_hex_hash", "a1b2c3d4e5f67890.bin", true},
{"img_mixed_junk", "IMG_20240101.jpg", true},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
got := shouldSkipEnrichment(c.basename)
if got != c.want {
t.Fatalf("shouldSkipEnrichment(%q) = %v, want %v", c.basename, got, c.want)
}
})
}
}
+34
View File
@@ -3,11 +3,34 @@ package browser
import (
"context"
"fmt"
"sync"
"time"
"github.com/go-rod/rod"
"github.com/go-rod/rod/lib/proto"
)
// watchPageClose spawns a goroutine that closes page when ctx is cancelled.
// Returns a cancel func that stops the watchdog on normal-path close.
// Uses sync.Once so page.Close() is idempotent if both paths fire concurrently.
func watchPageClose(ctx context.Context, page *rod.Page) (stopWatchdog func()) {
var once sync.Once
closeFn := func() { _ = page.Close() }
stopped := make(chan struct{})
go func() {
select {
case <-ctx.Done():
once.Do(closeFn)
case <-stopped:
}
}()
return func() {
close(stopped)
}
}
// Snapshot takes an accessibility snapshot of a page.
func (m *Manager) Snapshot(ctx context.Context, targetID string, opts SnapshotOptions) (*SnapshotResult, error) {
tenantID := tenantIDFromCtx(ctx)
@@ -58,6 +81,7 @@ func (m *Manager) Screenshot(ctx context.Context, targetID string, fullPage bool
}
// Navigate navigates a page to a URL.
// A ctx-cancel watchdog closes the page if ctx is done during the blocking WaitStable call.
func (m *Manager) Navigate(ctx context.Context, targetID, url string) error {
tenantID := tenantIDFromCtx(ctx)
m.mu.Lock()
@@ -68,10 +92,20 @@ func (m *Manager) Navigate(ctx context.Context, targetID, url string) error {
return err
}
// Watchdog: close page on ctx cancel to unblock any pending Rod CDP calls.
stop := watchPageClose(ctx, page)
defer stop()
if err := page.Navigate(url); err != nil {
if ctx.Err() != nil {
return ctx.Err()
}
return fmt.Errorf("navigate: %w", err)
}
if err := page.WaitStable(300 * time.Millisecond); err != nil {
if ctx.Err() != nil {
return ctx.Err()
}
return fmt.Errorf("wait stable after navigate: %w", err)
}
return nil
+7
View File
@@ -90,9 +90,16 @@ func (m *Manager) OpenTab(ctx context.Context, url string) (*TabInfo, error) {
return nil, fmt.Errorf("open tab: %w", err)
}
// Watchdog: close page on ctx cancel to unblock WaitStable CDP call.
stopWatchdog := watchPageClose(ctx, page)
if err := page.WaitStable(300 * time.Millisecond); err != nil {
stopWatchdog()
if ctx.Err() != nil {
return nil, ctx.Err()
}
return nil, fmt.Errorf("wait stable: %w", err)
}
stopWatchdog()
info, _ := page.Info()
tid := string(page.TargetID)
m.pages[tid] = page
+2
View File
@@ -73,6 +73,8 @@ const (
// Trace lifecycle events (realtime trace/span updates).
EventTraceUpdated = "trace.updated"
// Immediate status change event (not flush-buffered; fired on every status write).
EventTraceStatusChanged = "trace.status"
// Skill dependency check events (realtime progress during startup/rescan).
EventSkillDepsChecked = "skill.deps.checked"
@@ -0,0 +1,109 @@
//go:build integration
package integration
import (
"context"
"errors"
"runtime"
"testing"
"time"
"github.com/nextlevelbuilder/goclaw/internal/providers"
)
// TestAnthropicChat_AbortContextCancel verifies that cancelling a context
// while a Chat call is in flight properly closes the HTTP connection.
// Uses a stalling SSE server that only responds if data flows from it.
func TestAnthropicChat_AbortContextCancel(t *testing.T) {
t.Parallel()
// Start a stalling SSE server
server := newStallingSSEServer(t)
defer server.Close()
// Create provider pointed at the stalling server
provider := providers.NewAnthropicProvider(
"test-key",
providers.WithAnthropicBaseURL(server.URL),
providers.WithAnthropicModel("claude-opus-4-6"),
)
// Create a context with a 200ms deadline
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer cancel()
baselineGoroutines := runtime.NumGoroutine()
// Call Chat — should block briefly, then ctx timeout fires
start := time.Now()
_, err := provider.Chat(ctx, providers.ChatRequest{
Model: "claude-opus-4-6",
Messages: []providers.Message{},
})
elapsed := time.Since(start)
// Verify: context deadline/cancellation caused the error
if err == nil {
t.Fatal("expected Chat to fail on context cancellation, got nil error")
}
if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) {
// Error may be wrapped; that's acceptable
t.Logf("Chat error (wrapped context error acceptable): %v", err)
}
// Verify: returned promptly (well within 1.5s grace)
if elapsed > 1500*time.Millisecond {
t.Fatalf("Chat did not return within 1.5s, took %v", elapsed)
}
t.Logf("Chat returned in %v", elapsed)
// Verify: no goroutine leak (allow 100ms grace + baseline variance)
time.Sleep(100 * time.Millisecond)
finalGoroutines := runtime.NumGoroutine()
if finalGoroutines > baselineGoroutines+2 {
t.Fatalf("possible goroutine leak: baseline=%d, final=%d", baselineGoroutines, finalGoroutines)
}
}
// TestOpenAIChat_AbortContextCancel mirrors the Anthropic test for OpenAI-compatible providers.
func TestOpenAIChat_AbortContextCancel(t *testing.T) {
t.Parallel()
server := newStallingSSEServer(t)
defer server.Close()
provider := providers.NewOpenAIProvider(
"openai",
"test-key",
server.URL,
"gpt-4o",
)
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
defer cancel()
baselineGoroutines := runtime.NumGoroutine()
start := time.Now()
_, err := provider.Chat(ctx, providers.ChatRequest{
Model: "gpt-4o",
Messages: []providers.Message{},
})
elapsed := time.Since(start)
if err == nil {
t.Fatal("expected Chat to fail on context cancellation, got nil error")
}
if elapsed > 1500*time.Millisecond {
t.Fatalf("Chat did not return within 1.5s, took %v", elapsed)
}
t.Logf("Chat returned in %v", elapsed)
time.Sleep(100 * time.Millisecond)
finalGoroutines := runtime.NumGoroutine()
if finalGoroutines > baselineGoroutines+2 {
t.Fatalf("possible goroutine leak: baseline=%d, final=%d", baselineGoroutines, finalGoroutines)
}
}
@@ -0,0 +1,201 @@
//go:build integration
package integration
import (
"context"
"sync"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/agent"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
// TestRouter_AbortRun_NotFound verifies that aborting a non-existent run
// returns NotFound=true.
func TestRouter_AbortRun_NotFound(t *testing.T) {
t.Parallel()
router := agent.NewRouter()
result := router.AbortRun("nonexistent-run-id", "")
if !result.NotFound {
t.Errorf("expected NotFound=true, got %+v", result)
}
if result.Stopped || result.Forced || result.AlreadyAborting {
t.Errorf("expected only NotFound=true, got %+v", result)
}
}
// TestRouter_AbortRun_Unauthorized verifies that aborting with a mismatched
// sessionKey returns Unauthorized=true.
func TestRouter_AbortRun_Unauthorized(t *testing.T) {
t.Parallel()
router := agent.NewRouter()
// Register a run with sessionKey "session-A"
_, cancel := context.WithCancel(context.Background())
defer cancel()
router.RegisterRun(context.Background(), "run-1", "session-A", "agent-1", cancel)
// Try to abort with a different sessionKey
result := router.AbortRun("run-1", "session-B")
if !result.Unauthorized {
t.Errorf("expected Unauthorized=true, got %+v", result)
}
if result.Stopped || result.Forced || result.NotFound {
t.Errorf("expected only Unauthorized=true, got %+v", result)
}
}
// TestRouter_AbortRun_Concurrent_OnlyOneStops verifies that when multiple
// AbortRun calls race on the same run, exactly one succeeds (Stopped or Forced),
// and the rest see AlreadyAborting.
func TestRouter_AbortRun_Concurrent_OnlyOneStops(t *testing.T) {
t.Parallel()
router := agent.NewRouter()
// Register a run that will exit after 50ms
_, cancel := context.WithCancel(context.Background())
runID := "run-1"
sessionKey := "session-1"
_ = router.RegisterRun(context.Background(), runID, sessionKey, "agent-1", cancel)
// Goroutine closes Done after 50ms (simulating normal graceful exit)
go func() {
time.Sleep(50 * time.Millisecond)
router.UnregisterRun(runID) // this closes Done
}()
// Launch 100 concurrent AbortRun calls
resultsCh := make(chan agent.AbortResult, 100)
var wg sync.WaitGroup
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
defer wg.Done()
result := router.AbortRun(runID, sessionKey)
resultsCh <- result
}()
}
// Wait for all to complete
wg.Wait()
close(resultsCh)
// Collect and count results
var results []agent.AbortResult
for r := range resultsCh {
results = append(results, r)
}
var stoppedCount, forcedCount, alreadyAbortingCount, notFoundCount int
for _, r := range results {
if r.Stopped {
stoppedCount++
}
if r.Forced {
forcedCount++
}
if r.AlreadyAborting {
alreadyAbortingCount++
}
if r.NotFound {
notFoundCount++
}
}
t.Logf("Results: stopped=%d, forced=%d, alreadyAborting=%d, notFound=%d, total=%d",
stoppedCount, forcedCount, alreadyAbortingCount, notFoundCount, len(results))
// Verify exactly one succeeded (Stopped or Forced)
successCount := stoppedCount + forcedCount
if successCount != 1 {
t.Errorf("expected exactly 1 success (Stopped|Forced), got %d", successCount)
}
// Verify the rest are AlreadyAborting (or NotFound if race)
expectedAlreadyAborting := len(results) - 1
if alreadyAbortingCount+notFoundCount != expectedAlreadyAborting {
t.Errorf("expected %d AlreadyAborting+NotFound, got %d+%d",
expectedAlreadyAborting, alreadyAbortingCount, notFoundCount)
}
// Verify total count
if len(results) != 100 {
t.Errorf("expected 100 results, got %d", len(results))
}
}
// TestRouter_AbortRun_ForcesAfter3s verifies that when a run's goroutine
// never exits, AbortRun waits ~3s then returns Forced=true.
// Also verifies the mock TraceCollector is called with status="cancelled".
func TestRouter_AbortRun_ForcesAfter3s(t *testing.T) {
t.Parallel()
router := agent.NewRouter()
collector := &mockTraceCollector{}
router.SetTraceCollector(collector)
// Use a tenant-scoped context so forceMarkTraceAborted carries the tenant to FinishTrace.
testTenantID := uuid.New()
tenantCtx := store.WithTenantID(context.Background(), testTenantID)
// Register a run
_, cancel := context.WithCancel(tenantCtx)
runID := "run-1"
sessionKey := "session-1"
agentID := "agent-1"
router.RegisterRun(tenantCtx, runID, sessionKey, agentID, cancel)
// Set a trace ID
traceID := uuid.New()
router.SetRunTraceID(runID, traceID)
// DO NOT close Done — the goroutine will be "stuck"
// (We never call router.UnregisterRun, so Done stays open)
// Call AbortRun and time it
start := time.Now()
result := router.AbortRun(runID, sessionKey)
elapsed := time.Since(start)
// Verify: returned Forced=true
if !result.Forced {
t.Errorf("expected Forced=true, got %+v", result)
}
if result.Stopped || result.NotFound || result.Unauthorized || result.AlreadyAborting {
t.Errorf("expected only Forced=true, got %+v", result)
}
// Verify: elapsed time is in the range [2.9s, 3.5s]
// (Allow some variance for slow test runners)
minElapsed := 2900 * time.Millisecond
maxElapsed := 3500 * time.Millisecond
if elapsed < minElapsed || elapsed > maxElapsed {
t.Errorf("expected elapsed time in [2.9s, 3.5s], got %v", elapsed)
}
t.Logf("AbortRun forced after %v", elapsed)
// Verify: trace collector was called with status="cancelled" and correct tenant
if collector.FinishCallCount() == 0 {
t.Error("expected FinishTrace to be called, got 0 calls")
} else {
call := collector.LastFinishTrace()
if call.Status != "cancelled" {
t.Errorf("expected status='cancelled', got '%s'", call.Status)
}
if call.TraceID != traceID {
t.Errorf("expected traceID=%s, got %s", traceID, call.TraceID)
}
// Verify tenant was propagated to the FinishTrace ctx (FIX 1 regression check).
if call.TenantID != testTenantID {
t.Errorf("expected tenantID=%s in FinishTrace ctx, got %s", testTenantID, call.TenantID)
}
}
}
+134
View File
@@ -0,0 +1,134 @@
//go:build integration
package integration
import (
"context"
"errors"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/store"
)
var errSimulatedFailure = errors.New("simulated database failure")
// newStallingSSEServer returns an httptest server that sends SSE headers
// then blocks indefinitely, simulating an unresponsive LLM endpoint.
func newStallingSSEServer(t *testing.T) *httptest.Server {
t.Helper()
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
if f, ok := w.(http.Flusher); ok {
f.Flush()
}
// Block until client closes connection.
<-r.Context().Done()
}))
}
// newStallingHTTPServer returns an httptest server that accepts connections
// but never sends any bytes, simulating a server that hangs before headers.
// Useful for testing ResponseHeaderTimeout.
func newStallingHTTPServer(t *testing.T) *httptest.Server {
t.Helper()
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// Block forever without sending headers
<-r.Context().Done()
}))
}
// flakyTracingStore wraps a real TracingStore and fails the first N UpdateTrace
// calls, then succeeds thereafter. Atomic counter allows concurrent test calls.
type flakyTracingStore struct {
store.TracingStore
failsRemaining int32
}
// NewFlakyTracingStore creates a wrapper that fails N times then succeeds.
func newFlakyTracingStore(base store.TracingStore, failCount int) *flakyTracingStore {
return &flakyTracingStore{
TracingStore: base,
failsRemaining: int32(failCount),
}
}
// UpdateTrace fails the first failCount calls, then delegates to the wrapped store.
func (s *flakyTracingStore) UpdateTrace(ctx context.Context, id uuid.UUID, updates map[string]any) error {
fails := atomic.AddInt32(&s.failsRemaining, -1)
if fails >= 0 {
return errSimulatedFailure
}
return s.TracingStore.UpdateTrace(ctx, id, updates)
}
// captureBusEvent subscribes to a message bus and forwards events matching
// the given name into a channel. Automatically unsubscribes on test cleanup.
func captureBusEvent(t *testing.T, msgBus *bus.MessageBus, eventName string) <-chan bus.Event {
t.Helper()
captured := make(chan bus.Event, 10)
subID := "test-sub-" + uuid.New().String()[:8]
msgBus.Subscribe(subID, func(e bus.Event) {
if e.Name == eventName {
select {
case captured <- e:
default:
// channel full, drop (shouldn't happen in tests)
}
}
})
t.Cleanup(func() {
msgBus.Unsubscribe(subID)
close(captured)
})
return captured
}
// mockTraceCollector implements agent.TraceCollector for testing.
type mockTraceCollector struct {
finishCalls []struct {
TraceID uuid.UUID
TenantID uuid.UUID // tenant extracted from ctx at call time
Status string
ErrMsg string
OutputPreview string
}
}
// FinishTrace records the call for later assertion.
func (m *mockTraceCollector) FinishTrace(ctx context.Context, traceID uuid.UUID, status, errMsg, outputPreview string) {
m.finishCalls = append(m.finishCalls, struct {
TraceID uuid.UUID
TenantID uuid.UUID
Status string
ErrMsg string
OutputPreview string
}{traceID, store.TenantIDFromContext(ctx), status, errMsg, outputPreview})
}
// LastFinishTrace returns the most recent FinishTrace call, or nil if none.
func (m *mockTraceCollector) LastFinishTrace() *struct {
TraceID uuid.UUID
TenantID uuid.UUID
Status string
ErrMsg string
OutputPreview string
} {
if len(m.finishCalls) == 0 {
return nil
}
return &m.finishCalls[len(m.finishCalls)-1]
}
// FinishCallCount returns the number of FinishTrace calls.
func (m *mockTraceCollector) FinishCallCount() int {
return len(m.finishCalls)
}
@@ -0,0 +1,65 @@
//go:build integration
package integration
import (
"context"
"os/exec"
"runtime"
"testing"
"time"
)
// TestShellCommandCancel_AbortKillsProcessGroup verifies that cancelling
// a shell command context kills the entire process group via syscall.
// Skipped on Windows (process groups work differently).
// Note: This is a simplified integration test that doesn't use internal/tools
// since the tool interface is complex. Instead, it directly uses os/exec to
// simulate what the tool executor does.
func TestShellCommandCancel_AbortKillsProcessGroup(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("process group test skipped on Windows (different semantics)")
}
t.Parallel()
// Create a context that will be cancelled after 100ms
ctx, cancel := context.WithCancel(context.Background())
time.AfterFunc(100*time.Millisecond, cancel)
// Create a shell command that spawns two background sleep processes.
// Using 'sh -c' with '&' and 'wait' ensures all processes are in the same group.
cmd := exec.CommandContext(ctx, "sh", "-c", "sleep 60 & sleep 60 & wait")
start := time.Now()
err := cmd.Run()
elapsed := time.Since(start)
// Verify: command returns within 4 seconds (not hanging indefinitely)
if elapsed > 4*time.Second {
t.Fatalf("command did not return within 4s (took %v); process group kill may have failed", elapsed)
}
t.Logf("command returned in %v with error: %v", elapsed, err)
// Verify: context cancellation caused the failure
if err == nil {
t.Logf("command exited with code 0 (may have finished before cancel)")
}
// Give processes a moment to fully exit
time.Sleep(500 * time.Millisecond)
// Verify: no 'sleep 60' processes remain (check with ps + grep).
// Use 'pgrep -f "sleep 60"' if available, otherwise fall back to 'ps aux | grep'.
var pgrepCmd *exec.Cmd
if _, err := exec.LookPath("pgrep"); err == nil {
pgrepCmd = exec.Command("pgrep", "-f", "sleep 60")
} else {
// Fallback for systems without pgrep (macOS, some Linux)
pgrepCmd = exec.Command("sh", "-c", "ps aux | grep 'sleep 60' | grep -v grep || true")
}
output, _ := pgrepCmd.CombinedOutput()
if len(output) > 0 {
t.Logf("warning: 'sleep 60' processes may still exist; output: %s", string(output))
}
}
@@ -0,0 +1,236 @@
//go:build integration
package integration
import (
"context"
"database/sql"
"testing"
"time"
"github.com/google/uuid"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/store"
"github.com/nextlevelbuilder/goclaw/internal/store/pg"
"github.com/nextlevelbuilder/goclaw/internal/tracing"
"github.com/nextlevelbuilder/goclaw/pkg/protocol"
)
// TestCollector_UpdateRetry_SucceedsAfterTwoFailures verifies that when the
// store fails twice, the retry mechanism recovers and persists the update.
func TestCollector_UpdateRetry_SucceedsAfterTwoFailures(t *testing.T) {
db := testDB(t)
tenantID, _ := seedTenantAgent(t, db)
// Create a real store and collector
tracingStore := pg.NewPGTracingStore(db)
collector := tracing.NewCollector(tracingStore)
defer collector.Stop()
// Create a trace in the DB (with tenant_id)
trace := &store.TraceData{
ID: uuid.New(),
Status: "running",
StartTime: time.Now().UTC(),
}
_, err := db.ExecContext(context.Background(),
`INSERT INTO traces (id, tenant_id, status, start_time) VALUES ($1, $2, $3, $4)`,
trace.ID, tenantID, trace.Status, trace.StartTime)
if err != nil {
t.Fatalf("insert trace failed: %v", err)
}
// Wrap the store with a flaky version that fails 2 times
flaky := newFlakyTracingStore(tracingStore, 2)
collector2 := tracing.NewCollector(flaky)
collector2.Start()
t.Cleanup(collector2.Stop)
// Call SetTraceStatus — should fail twice inline, then succeed on retry
setCtx, setCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer setCancel()
setCtx = store.WithTenantID(setCtx, tenantID)
collector2.SetTraceStatus(setCtx, trace.ID, "completed")
// Give the retry worker time to process
time.Sleep(100 * time.Millisecond)
// Query the trace status from the real store (using fresh context + connection)
var status string
queryErr := db.QueryRowContext(
context.Background(),
`SELECT status FROM traces WHERE id = $1`,
trace.ID,
).Scan(&status)
if queryErr != nil {
t.Fatalf("query trace status failed: %v", queryErr)
}
if status != "completed" {
t.Errorf("expected status='completed', got '%s'", status)
}
t.Logf("Trace status updated to: %s", status)
}
// TestCollector_UpdateRetry_EnqueuesOnAllFailures verifies that when all inline
// retries fail, the update is enqueued in the retry queue (RetryQueueLen >= 1).
func TestCollector_UpdateRetry_EnqueuesOnAllFailures(t *testing.T) {
db := testDB(t)
tenantID, _ := seedTenantAgent(t, db)
// Create a trace in the DB so UpdateTrace has a valid target.
traceID := uuid.New()
_, err := db.ExecContext(context.Background(),
`INSERT INTO traces (id, tenant_id, status, start_time) VALUES ($1, $2, $3, $4)`,
traceID, tenantID, "running", time.Now().UTC())
if err != nil {
t.Fatalf("insert trace failed: %v", err)
}
// Wrap store so every UpdateTrace call fails — more than inline retry count (3).
realStore := pg.NewPGTracingStore(db)
flaky := newFlakyTracingStore(realStore, 1000)
// Do NOT call collector.Start() — we want the retry worker idle so items stay in queue.
collector := tracing.NewCollector(flaky)
setCtx := store.WithTenantID(context.Background(), tenantID)
collector.SetTraceStatus(setCtx, traceID, "completed")
// After all 4 inline attempts fail, the update must be in the retry queue.
queueLen := collector.RetryQueueLen()
if queueLen < 1 {
t.Errorf("expected RetryQueueLen >= 1 after all inline retries failed, got %d", queueLen)
}
t.Logf("RetryQueueLen after all-fail: %d", queueLen)
}
// TestCollector_BroadcastsStatusEvent verifies that when a trace status is
// updated successfully, the StatusBroadcaster callback is invoked with the
// correct payload and tenant ID.
func TestCollector_BroadcastsStatusEvent(t *testing.T) {
db := testDB(t)
tenantID, _ := seedTenantAgent(t, db)
// Create stores and collector
tracingStore := pg.NewPGTracingStore(db)
collector := tracing.NewCollector(tracingStore)
// Set up a message bus and broadcaster
msgBus := bus.New()
broadcastedPayloads := make(chan struct {
payload tracing.TraceStatusPayload
tenantID uuid.UUID
}, 10)
// Wire the broadcaster
broadcaster := func(payload tracing.TraceStatusPayload, tid uuid.UUID) {
broadcastedPayloads <- struct {
payload tracing.TraceStatusPayload
tenantID uuid.UUID
}{payload, tid}
// Also emit to message bus for WS subscribers
msgBus.Broadcast(bus.Event{
Name: protocol.EventTraceStatusChanged,
Payload: payload,
TenantID: tid,
})
}
collector.SetStatusBroadcaster(broadcaster)
// Create a trace in the DB (with tenant_id)
trace := &store.TraceData{
ID: uuid.New(),
Status: "running",
StartTime: time.Now().UTC(),
}
_, err := db.ExecContext(context.Background(),
`INSERT INTO traces (id, tenant_id, status, start_time) VALUES ($1, $2, $3, $4)`,
trace.ID, tenantID, trace.Status, trace.StartTime)
if err != nil {
t.Fatalf("insert trace failed: %v", err)
}
// Call FinishTrace
finishCtx, finishCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer finishCancel()
finishCtx = store.WithTenantID(finishCtx, tenantID)
collector.FinishTrace(finishCtx, trace.ID, "completed", "", "test output")
// Give a moment for the broadcast
time.Sleep(50 * time.Millisecond)
// Verify broadcaster was called
select {
case p := <-broadcastedPayloads:
if p.payload.TraceID != trace.ID.String() {
t.Errorf("expected traceId=%s, got %s", trace.ID.String(), p.payload.TraceID)
}
if p.payload.Status != "completed" {
t.Errorf("expected status='completed', got '%s'", p.payload.Status)
}
if p.payload.EndedAt == nil {
t.Error("expected EndedAt to be non-nil")
}
if p.tenantID != tenantID {
t.Errorf("expected tenantID=%s, got %s", tenantID.String(), p.tenantID.String())
}
t.Logf("Broadcast payload verified: status=%s, tenantID=%s", p.payload.Status, p.tenantID.String())
case <-time.After(1 * time.Second):
t.Error("broadcaster was not called within timeout")
}
}
// TestCollector_StaleRecovery_MarksOldRunningAsError inserts a "running" trace
// older than 2 minutes and verifies that the stale recovery loop marks it as "error".
func TestCollector_StaleRecovery_MarksOldRunningAsError(t *testing.T) {
db := testDB(t)
seedTenantAgent(t, db)
tracingStore := pg.NewPGTracingStore(db)
collector := tracing.NewCollector(tracingStore)
collector.Start()
t.Cleanup(collector.Stop)
// Insert a "running" trace with start_time 11 minutes ago (exceeds 10min staleThreshold)
traceID := uuid.New()
oldStartTime := time.Now().UTC().Add(-11 * time.Minute)
_, err := db.ExecContext(
context.Background(),
`INSERT INTO traces (id, tenant_id, status, start_time)
VALUES ($1, $2, $3, $4)`,
traceID, store.MasterTenantID, "running", oldStartTime,
)
if err != nil {
t.Fatalf("insert stale trace failed: %v", err)
}
t.Logf("Inserted stale trace %s with start_time %v", traceID.String(), oldStartTime)
// Manually trigger stale recovery (wait up to 35s or use helper if available)
// The stale recovery loop runs every 30s by default, so we trigger it directly
collector.RecoverStaleNow()
// Give it a moment to complete
time.Sleep(500 * time.Millisecond)
// Query the trace status
var status string
err = db.QueryRowContext(
context.Background(),
`SELECT status FROM traces WHERE id = $1`,
traceID,
).Scan(&status)
if err != nil && err != sql.ErrNoRows {
t.Fatalf("query trace status failed: %v", err)
}
if err == sql.ErrNoRows {
t.Errorf("trace was deleted instead of marked as error")
return
}
if status != "error" {
t.Errorf("expected status='error' after stale recovery, got '%s'", status)
}
t.Logf("Stale trace marked as error: %s", status)
}
+2
View File
@@ -246,6 +246,8 @@ export const Events = {
// Trace lifecycle
TRACE_UPDATED: "trace.updated",
// Immediate status change (not flush-buffered; fired on every status write).
TRACE_STATUS: "trace.status",
// Skill dependency check (realtime progress during startup/rescan)
SKILL_DEPS_CHECKED: "skill.deps.checked",
+6 -3
View File
@@ -77,8 +77,11 @@
},
"toast": {
"failedLoad": "Failed to load traces",
"abortSent": "Run stopped",
"abortFailed": "Failed to stop run",
"abortNotFound": "Run already finished or not found"
"abortStopped": "Run stopped",
"abortForced": "Run force-stopped (still cleaning up in background)",
"abortAlreadyAborting": "Stop already in progress",
"abortNotFound": "Run already finished",
"abortUnauthorized": "Not authorized to stop this run",
"abortFailed": "Failed to stop run"
}
}
+6 -3
View File
@@ -77,8 +77,11 @@
},
"toast": {
"failedLoad": "Không thể tải dữ liệu theo dõi",
"abortSent": "Đã dừng chạy",
"abortFailed": "Không thể dừng chạy",
"abortNotFound": "Lệnh đã hoàn tất hoặc không tìm thấy"
"abortStopped": "Đã dừng tác vụ",
"abortForced": "Đã buộc dừng tác vụ (đang dọn dẹp ở nền)",
"abortAlreadyAborting": "Đang dừng tác vụ",
"abortNotFound": "Tác vụ đã kết thúc",
"abortUnauthorized": "Không có quyền dừng tác vụ này",
"abortFailed": "Không thể dừng tác vụ"
}
}
+6 -3
View File
@@ -66,9 +66,12 @@
"title": "追踪",
"toast": {
"failedLoad": "加载追踪失败",
"abortSent": "已停止运行",
"abortFailed": "停止运行失败",
"abortNotFound": "运行已完成或未找到"
"abortStopped": "已停止运行",
"abortForced": "已强制停止(后台仍在清理)",
"abortAlreadyAborting": "停止操作进行中",
"abortNotFound": "运行已结束",
"abortUnauthorized": "无权停止此运行",
"abortFailed": "无法停止运行"
},
"source": {
"direct": "私聊",
+1 -1
View File
@@ -36,7 +36,7 @@ export function useTraces(filters: TraceFilters = {}) {
return { traces: res.traces ?? [], total: res.total ?? 0 };
},
placeholderData: (prev) => prev,
staleTime: 60_000,
staleTime: 0,
});
const traces = data?.traces ?? [];
@@ -56,12 +56,21 @@ export function TraceDetailDialog({ traceId, onClose, getTrace, onNavigateTrace,
.finally(() => setLoading(false));
}, [traceId, getTrace]);
// Auto-refetch when trace aggregates update
// Auto-refetch when trace aggregates update (flush-buffered, 5s interval).
useWsEvent(Events.TRACE_UPDATED, useCallback(
(payload: unknown) => { const data = payload as { trace_ids?: string[] }; if (data?.trace_ids?.includes(traceId)) fetchTrace(); },
[traceId, fetchTrace],
));
// Immediate status update (fired on every status write, before flush tick).
useWsEvent(Events.TRACE_STATUS, useCallback(
(payload: unknown) => {
const data = payload as { traceId?: string };
if (data?.traceId === traceId) fetchTrace();
},
[traceId, fetchTrace],
));
// Refetch on run completion
useWsEvent(Events.AGENT, useCallback(
(payload: unknown) => {
+34 -6
View File
@@ -24,8 +24,11 @@ import { useDeferredLoading } from "@/hooks/use-deferred-loading";
import { useUiStore } from "@/stores/use-ui-store";
import { useAgents } from "@/pages/agents/hooks/use-agents";
import { useChannelInstances } from "@/pages/channels/hooks/use-channel-instances";
import { useQueryClient } from "@tanstack/react-query";
import { useWs } from "@/hooks/use-ws";
import { Methods } from "@/api/protocol";
import { useWsEvent } from "@/hooks/use-ws-event";
import { Methods, Events } from "@/api/protocol";
import { queryKeys } from "@/lib/query-keys";
import { toast } from "@/stores/use-toast-store";
/** Strip media placeholder tags like <media:image> from preview text */
@@ -66,6 +69,14 @@ export function TracesPage() {
const [pageSize, setPageSize] = useState(20);
const ws = useWs();
const queryClient = useQueryClient();
// Invalidate traces list on immediate status events (no need to wait for 5s flush).
useWsEvent(Events.TRACE_STATUS, useCallback(
() => queryClient.invalidateQueries({ queryKey: queryKeys.traces.all }),
[queryClient],
));
const { agents } = useAgents();
const { instances: channels } = useChannelInstances();
@@ -101,16 +112,33 @@ export function TracesPage() {
const res = await ws.call(Methods.CHAT_ABORT, {
sessionKey: trace.session_key,
runId: trace.run_id,
}) as { aborted?: boolean };
if (res?.aborted) {
toast.success(t("toast.abortSent"));
refresh();
} else {
}) as {
aborted?: boolean;
stopped?: boolean;
forced?: boolean;
alreadyAborting?: boolean;
notFound?: boolean;
unauthorized?: boolean;
};
if (res?.stopped) {
toast.success(t("toast.abortStopped"));
} else if (res?.forced) {
toast.warning(t("toast.abortForced"));
} else if (res?.alreadyAborting) {
toast.info(t("toast.abortAlreadyAborting"));
} else if (res?.unauthorized) {
toast.error(t("toast.abortUnauthorized"));
} else if (res?.notFound) {
toast.info(t("toast.abortNotFound"));
} else {
toast.error(t("toast.abortFailed"));
}
refresh();
} catch {
toast.error(t("toast.abortFailed"));
} finally {
// Auto re-enable within 5s max (3s grace + 2s buffer) in case WS event is delayed.
setTimeout(() => setAbortingRunId(null), 5000);
setAbortingRunId(null);
}
},