feat: enhance filesystem security with symlink/hardlink protection, implement typing indicators, and improve scheduler queue management for stale requests.

This commit is contained in:
viettranx committed 2026-02-26 22:47:59 +07:00
1 parent d5cc5a745d
commit 2ac44ce937
12 files changed
+968 -58

No files matched your search

+24 -2
View File
@@ -5,11 +5,13 @@ import (
"fmt"
"log/slog"
"sync"
"time"
"github.com/bwmarrin/discordgo"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
"github.com/nextlevelbuilder/goclaw/internal/channels/typing"
"github.com/nextlevelbuilder/goclaw/internal/config"
)
@@ -20,6 +22,7 @@ type Channel struct {
config config.DiscordConfig
botUserID string // populated on start
placeholders sync.Map // channelID string → messageID string
typingCtrls sync.Map // channelID string → *typing.Controller
}
// New creates a new Discord channel from config.
@@ -85,6 +88,11 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) error {
return fmt.Errorf("empty chat ID for discord send")
}
// Stop typing indicator controller
if ctrl, ok := c.typingCtrls.LoadAndDelete(channelID); ok {
ctrl.(*typing.Controller).Stop()
}
content := msg.Content
// NO_REPLY cleanup: content is empty when agent suppresses reply.
@@ -208,8 +216,22 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate
"preview", channels.Truncate(content, 50),
)
// Send typing indicator
_ = c.session.ChannelTyping(channelID)
// Send typing indicator with keepalive + TTL safety net.
// Discord typing expires after 10s, so keepalive every 9s.
// TTL auto-stops after 60s to prevent stuck indicators.
typingCtrl := typing.New(typing.Options{
MaxDuration: 60 * time.Second,
KeepaliveInterval: 9 * time.Second,
StartFn: func() error {
return c.session.ChannelTyping(channelID)
},
})
// Stop previous typing controller for this channel (if any)
if prev, ok := c.typingCtrls.Load(channelID); ok {
prev.(*typing.Controller).Stop()
}
c.typingCtrls.Store(channelID, typingCtrl)
typingCtrl.Start()
// Send placeholder "Thinking..." message
placeholder, err := c.session.ChannelMessageSend(channelID, "Thinking...")
+1
View File
@@ -28,6 +28,7 @@ type Channel struct {
teamStore store.TeamStore // for /tasks, /task_detail commands (nil in standalone)
placeholders sync.Map // localKey string → messageID int
stopThinking sync.Map // localKey string → *thinkingCancel
typingCtrls sync.Map // localKey string → *typing.Controller
streams sync.Map // localKey string → *DraftStream (streaming preview)
reactions sync.Map // localKey string → *StatusReactionController
pairingReplySent sync.Map // userID string → time.Time (debounce pairing replies)
+20 -5
View File
@@ -12,6 +12,7 @@ import (
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels"
"github.com/nextlevelbuilder/goclaw/internal/channels/typing"
)
// handleMessage processes an incoming Telegram update.
@@ -286,13 +287,27 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) {
}
}
// Send typing indicator (TS ref: buildTypingThreadParams — General topic ID=1 is OK for typing).
// Send typing indicator with keepalive + TTL safety net.
// Telegram typing expires after 5s, so keepalive every 4s.
// TTL auto-stops after 60s to prevent stuck indicators.
chatIDObj := tu.ID(chatID)
typingAction := tu.ChatAction(chatIDObj, telego.ChatActionTyping)
if messageThreadID > 0 {
typingAction.MessageThreadID = messageThreadID
typingCtrl := typing.New(typing.Options{
MaxDuration: 60 * time.Second,
KeepaliveInterval: 4 * time.Second,
StartFn: func() error {
action := tu.ChatAction(chatIDObj, telego.ChatActionTyping)
if messageThreadID > 0 {
action.MessageThreadID = messageThreadID
}
return c.bot.SendChatAction(ctx, action)
},
})
// Stop previous typing controller for this chat/topic (if any)
if prev, ok := c.typingCtrls.Load(localKey); ok {
prev.(*typing.Controller).Stop()
}
_ = c.bot.SendChatAction(ctx, typingAction)
c.typingCtrls.Store(localKey, typingCtrl)
typingCtrl.Start()
// Stop previous thinking animation for this chat/topic
if prevStop, ok := c.stopThinking.Load(localKey); ok {
+6
View File
@@ -12,6 +12,7 @@ import (
tu "github.com/mymmrac/telego/telegoutil"
"github.com/nextlevelbuilder/goclaw/internal/bus"
"github.com/nextlevelbuilder/goclaw/internal/channels/typing"
)
// Error patterns for graceful handling (matching TS error constants in send.ts).
@@ -57,6 +58,11 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error {
c.stopThinking.Delete(localKey)
}
// Stop typing indicator controller (TTL keepalive)
if ctrl, ok := c.typingCtrls.LoadAndDelete(localKey); ok {
ctrl.(*typing.Controller).Stop()
}
// NO_REPLY cleanup: content is empty when agent suppresses reply (prompt injection, etc.).
// Clean up placeholder, then return without sending any message.
if msg.Content == "" && len(msg.Media) == 0 {
+204
View File
@@ -0,0 +1,204 @@
// Package typing provides a channel-agnostic typing indicator controller
// with TTL safety net, keepalive support, and post-close guards.
//
// The controller prevents stuck typing indicators by:
// - Auto-stopping after a configurable TTL (default 60s)
// - Requiring both MarkRunComplete + MarkDispatchIdle for graceful cleanup
// - Guarding against post-close keepalive invocations
package typing
import (
"log/slog"
"sync"
"time"
)
// Options configures a typing indicator controller.
type Options struct {
// MaxDuration is the TTL safety net. If the indicator hasn't been
// stopped after this duration, it auto-stops. 0 disables the TTL.
// Default: 60s.
MaxDuration time.Duration
// KeepaliveInterval is how often to re-send the typing action.
// Telegram typing expires after 5s, Discord after 10s.
// 0 disables keepalive (single fire-and-forget).
KeepaliveInterval time.Duration
// StartFn sends the channel-specific typing indicator.
// Called on Start() and on each keepalive tick.
StartFn func() error
// StopFn sends the channel-specific stop-typing signal.
// Optional — some channels (Telegram) auto-stop on message send.
StopFn func() error
}
// Controller manages the lifecycle of a typing indicator.
// It is safe for concurrent use.
type Controller struct {
mu sync.Mutex
// State flags
closed bool // post-close guard: prevents stale startFn calls
runComplete bool // signal 1: agent finished processing
dispatchIdle bool // signal 2: message delivery finished
stopSent bool // prevents duplicate stopFn calls
// Configuration
maxDuration time.Duration
keepaliveInterval time.Duration
startFn func() error
stopFn func() error
// Timers
ttlTimer *time.Timer
keepaliveDone chan struct{}
}
// New creates a typing controller with the given options.
func New(opts Options) *Controller {
maxDur := opts.MaxDuration
if maxDur == 0 {
maxDur = 60 * time.Second
}
return &Controller{
maxDuration: maxDur,
keepaliveInterval: opts.KeepaliveInterval,
startFn: opts.StartFn,
stopFn: opts.StopFn,
}
}
// Start begins the typing indicator, TTL timer, and keepalive loop.
func (c *Controller) Start() {
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return
}
// Fire initial typing action
c.fireStart()
// Start TTL safety net
if c.maxDuration > 0 {
c.ttlTimer = time.AfterFunc(c.maxDuration, func() {
c.mu.Lock()
if !c.closed {
slog.Debug("typing: TTL exceeded, auto-stopping", "ttl", c.maxDuration)
c.forceStop()
}
c.mu.Unlock()
})
}
// Start keepalive loop
if c.keepaliveInterval > 0 {
c.keepaliveDone = make(chan struct{})
go c.keepaliveLoop()
}
}
// Stop forcefully stops the typing indicator and all timers.
// Safe to call multiple times (idempotent).
func (c *Controller) Stop() {
c.mu.Lock()
defer c.mu.Unlock()
c.forceStop()
}
// MarkRunComplete signals that the agent has finished processing.
// Cleanup happens only when both MarkRunComplete and MarkDispatchIdle have been called.
func (c *Controller) MarkRunComplete() {
c.mu.Lock()
defer c.mu.Unlock()
c.runComplete = true
c.tryCleanup()
}
// MarkDispatchIdle signals that message delivery has completed.
// Cleanup happens only when both MarkRunComplete and MarkDispatchIdle have been called.
func (c *Controller) MarkDispatchIdle() {
c.mu.Lock()
defer c.mu.Unlock()
c.dispatchIdle = true
c.tryCleanup()
}
// tryCleanup runs cleanup only when both completion signals have been received.
// Must be called with c.mu held.
func (c *Controller) tryCleanup() {
if c.runComplete && c.dispatchIdle && !c.closed {
c.forceStop()
}
}
// forceStop cancels all timers and sends stop signal.
// Must be called with c.mu held.
func (c *Controller) forceStop() {
if c.closed {
return
}
c.closed = true
// Cancel TTL timer
if c.ttlTimer != nil {
c.ttlTimer.Stop()
c.ttlTimer = nil
}
// Stop keepalive loop
if c.keepaliveDone != nil {
close(c.keepaliveDone)
c.keepaliveDone = nil
}
// Send stop signal
c.fireStop()
}
// fireStart invokes the channel-specific start function.
// Must be called with c.mu held.
func (c *Controller) fireStart() {
if c.closed || c.startFn == nil {
return
}
if err := c.startFn(); err != nil {
slog.Debug("typing: startFn error", "error", err)
}
}
// fireStop invokes the channel-specific stop function (once).
// Must be called with c.mu held.
func (c *Controller) fireStop() {
if c.stopSent || c.stopFn == nil {
return
}
c.stopSent = true
if err := c.stopFn(); err != nil {
slog.Debug("typing: stopFn error", "error", err)
}
}
// keepaliveLoop periodically re-sends the typing indicator.
func (c *Controller) keepaliveLoop() {
ticker := time.NewTicker(c.keepaliveInterval)
defer ticker.Stop()
for {
select {
case <-c.keepaliveDone:
return
case <-ticker.C:
c.mu.Lock()
if c.closed {
c.mu.Unlock()
return
}
c.fireStart()
c.mu.Unlock()
}
}
}
+186
View File
@@ -0,0 +1,186 @@
package typing
import (
"sync/atomic"
"testing"
"time"
)
func TestTTLAutoStop(t *testing.T) {
var startCount, stopCount atomic.Int32
ctrl := New(Options{
MaxDuration: 100 * time.Millisecond,
KeepaliveInterval: 0,
StartFn: func() error { startCount.Add(1); return nil },
StopFn: func() error { stopCount.Add(1); return nil },
})
ctrl.Start()
// Should have started once
if startCount.Load() != 1 {
t.Fatalf("expected 1 start call, got %d", startCount.Load())
}
// Wait for TTL to expire
time.Sleep(200 * time.Millisecond)
if stopCount.Load() != 1 {
t.Fatalf("expected 1 stop call after TTL, got %d", stopCount.Load())
}
// Verify closed
ctrl.mu.Lock()
closed := ctrl.closed
ctrl.mu.Unlock()
if !closed {
t.Fatal("expected controller to be closed after TTL")
}
}
func TestPostCloseGuard(t *testing.T) {
var startCount atomic.Int32
ctrl := New(Options{
MaxDuration: 10 * time.Second,
StartFn: func() error { startCount.Add(1); return nil },
})
ctrl.Start()
if startCount.Load() != 1 {
t.Fatalf("expected 1 start call, got %d", startCount.Load())
}
ctrl.Stop()
// Start after stop should be no-op
ctrl.Start()
if startCount.Load() != 1 {
t.Fatalf("expected no additional start calls after Stop, got %d", startCount.Load())
}
}
func TestDualSignalsRequired(t *testing.T) {
var stopCount atomic.Int32
ctrl := New(Options{
MaxDuration: 10 * time.Second,
StartFn: func() error { return nil },
StopFn: func() error { stopCount.Add(1); return nil },
})
ctrl.Start()
// Only run complete — should NOT stop
ctrl.MarkRunComplete()
time.Sleep(20 * time.Millisecond)
if stopCount.Load() != 0 {
t.Fatal("expected no stop after only MarkRunComplete")
}
// Now dispatch idle — should trigger cleanup
ctrl.MarkDispatchIdle()
time.Sleep(20 * time.Millisecond)
if stopCount.Load() != 1 {
t.Fatalf("expected 1 stop after both signals, got %d", stopCount.Load())
}
}
func TestDualSignalsReverseOrder(t *testing.T) {
var stopCount atomic.Int32
ctrl := New(Options{
MaxDuration: 10 * time.Second,
StartFn: func() error { return nil },
StopFn: func() error { stopCount.Add(1); return nil },
})
ctrl.Start()
// Dispatch idle first — should NOT stop
ctrl.MarkDispatchIdle()
time.Sleep(20 * time.Millisecond)
if stopCount.Load() != 0 {
t.Fatal("expected no stop after only MarkDispatchIdle")
}
// Now run complete — should trigger cleanup
ctrl.MarkRunComplete()
time.Sleep(20 * time.Millisecond)
if stopCount.Load() != 1 {
t.Fatalf("expected 1 stop after both signals, got %d", stopCount.Load())
}
}
func TestKeepalive(t *testing.T) {
var startCount atomic.Int32
ctrl := New(Options{
MaxDuration: 2 * time.Second,
KeepaliveInterval: 30 * time.Millisecond,
StartFn: func() error { startCount.Add(1); return nil },
})
ctrl.Start()
time.Sleep(120 * time.Millisecond)
ctrl.Stop()
// Initial start + at least 2 keepalive ticks
count := startCount.Load()
if count < 3 {
t.Fatalf("expected at least 3 start calls (1 initial + keepalive ticks), got %d", count)
}
}
func TestKeepaliveStopsAfterClose(t *testing.T) {
var startCount atomic.Int32
ctrl := New(Options{
MaxDuration: 2 * time.Second,
KeepaliveInterval: 20 * time.Millisecond,
StartFn: func() error { startCount.Add(1); return nil },
})
ctrl.Start()
time.Sleep(60 * time.Millisecond)
ctrl.Stop()
countAtStop := startCount.Load()
// Wait to verify no more keepalive ticks
time.Sleep(80 * time.Millisecond)
countAfter := startCount.Load()
if countAfter != countAtStop {
t.Fatalf("expected no start calls after Stop, got %d more", countAfter-countAtStop)
}
}
func TestStopIdempotent(t *testing.T) {
var stopCount atomic.Int32
ctrl := New(Options{
MaxDuration: 10 * time.Second,
StartFn: func() error { return nil },
StopFn: func() error { stopCount.Add(1); return nil },
})
ctrl.Start()
ctrl.Stop()
ctrl.Stop()
ctrl.Stop()
if stopCount.Load() != 1 {
t.Fatalf("expected exactly 1 stop call, got %d", stopCount.Load())
}
}
func TestNilFunctions(t *testing.T) {
// Should not panic with nil start/stop functions
ctrl := New(Options{
MaxDuration: 50 * time.Millisecond,
})
ctrl.Start()
time.Sleep(100 * time.Millisecond)
ctrl.Stop()
}
+11 -4
View File
@@ -11,6 +11,7 @@ import (
"context"
"fmt"
"os/exec"
"path/filepath"
"strings"
)
@@ -104,16 +105,22 @@ func (b *FsBridge) Stat(ctx context.Context, path string) (string, error) {
}
// resolvePath resolves a path relative to the container workdir.
// Validates that absolute paths stay within the workdir (defense in depth).
func (b *FsBridge) resolvePath(path string) string {
if path == "" || path == "." {
return b.workdir
}
// Absolute paths pass through
if strings.HasPrefix(path, "/") {
return path
// Validate absolute paths stay within workdir (defense in depth,
// container is already sandboxed with read-only FS + cap-drop ALL).
cleaned := filepath.Clean(path)
if cleaned == b.workdir || strings.HasPrefix(cleaned, b.workdir+"/") {
return cleaned
}
return b.workdir // fallback to workdir for escapes
}
// Relative paths resolve against workdir
return b.workdir + "/" + path
// Relative paths: use filepath.Join for proper normalization
return filepath.Clean(filepath.Join(b.workdir, path))
}
// dockerExec runs a command inside the container and returns stdout, stderr, exit code.
+12
View File
@@ -8,4 +8,16 @@ var (
// ErrQueueDropped is returned when a queued message is evicted to make room (drop=old policy).
ErrQueueDropped = errors.New("message dropped from queue")
// ErrMessageStale is returned when a queued message is skipped because it was
// enqueued before an abort (/stopall) and is no longer relevant.
ErrMessageStale = errors.New("message stale: enqueued before abort")
// ErrGatewayDraining is returned when the gateway is shutting down and
// new requests cannot be accepted.
ErrGatewayDraining = errors.New("gateway is shutting down, please retry shortly")
// ErrLaneCleared is returned when a session queue is reset (e.g. during restart)
// and all pending requests are drained.
ErrLaneCleared = errors.New("session queue cleared during reset")
)
+80 -25
View File
@@ -2,6 +2,7 @@ package scheduler
import (
"context"
"log/slog"
"sync"
"time"
@@ -61,8 +62,9 @@ type TokenEstimateFunc func(sessionKey string) (tokens int, contextWindow int)
// PendingRequest is a queued agent run awaiting execution.
type PendingRequest struct {
Req agent.RunRequest
ResultCh chan RunOutcome
Req agent.RunRequest
ResultCh chan RunOutcome
EnqueuedAt time.Time // timestamp when enqueued, used for stale message detection
}
// RunOutcome is the result of a scheduled agent run.
@@ -71,6 +73,12 @@ type RunOutcome struct {
Err error
}
// activeRunEntry tracks a running agent execution with its generation.
type activeRunEntry struct {
cancel context.CancelFunc
generation uint64
}
// SessionQueue manages agent runs for a single session key.
// Supports configurable concurrency: 1 (serial) or N (concurrent).
type SessionQueue struct {
@@ -80,13 +88,15 @@ type SessionQueue struct {
laneMgr *LaneManager
lane string
mu sync.Mutex
queue []*PendingRequest
activeRuns map[string]context.CancelFunc // runID → cancel
activeOrder []string // FIFO order of active runIDs
maxConcurrent int // effective limit (from config or per-session override)
timer *time.Timer // debounce timer
parentCtx context.Context // stored from first Enqueue call
mu sync.Mutex
queue []*PendingRequest
activeRuns map[string]activeRunEntry // runID → entry (with generation)
activeOrder []string // FIFO order of active runIDs
maxConcurrent int // effective limit (from config or per-session override)
timer *time.Timer // debounce timer
parentCtx context.Context // stored from first Enqueue call
abortCutoffTime time.Time // messages enqueued before this are stale
generation uint64 // bumped on Reset() to ignore stale completions
tokenEstimateFn TokenEstimateFunc // optional: for adaptive throttle
}
@@ -103,7 +113,7 @@ func NewSessionQueue(key, lane string, cfg QueueConfig, laneMgr *LaneManager, ru
runFn: runFn,
laneMgr: laneMgr,
lane: lane,
activeRuns: make(map[string]context.CancelFunc),
activeRuns: make(map[string]activeRunEntry),
maxConcurrent: maxC,
}
}
@@ -148,7 +158,7 @@ func (sq *SessionQueue) hasCapacity() bool {
// Returns a channel that receives the result when the run completes.
func (sq *SessionQueue) Enqueue(ctx context.Context, req agent.RunRequest) <-chan RunOutcome {
outcome := make(chan RunOutcome, 1)
pending := &PendingRequest{Req: req, ResultCh: outcome}
pending := &PendingRequest{Req: req, ResultCh: outcome, EnqueuedAt: time.Now()}
sq.mu.Lock()
defer sq.mu.Unlock()
@@ -161,8 +171,8 @@ func (sq *SessionQueue) Enqueue(ctx context.Context, req agent.RunRequest) <-cha
switch sq.config.Mode {
case QueueModeInterrupt:
// Cancel all active runs
for runID, cancel := range sq.activeRuns {
cancel()
for runID, entry := range sq.activeRuns {
entry.cancel()
delete(sq.activeRuns, runID)
}
sq.activeOrder = nil
@@ -223,8 +233,28 @@ func (sq *SessionQueue) startAvailable(ctx context.Context) {
}
// startOne picks the first queued request and runs it in the lane.
// Skips stale messages that were enqueued before the last abort cutoff.
// Must be called with sq.mu held.
func (sq *SessionQueue) startOne(ctx context.Context) {
// Skip stale messages enqueued before the last /stopall abort cutoff.
for len(sq.queue) > 0 {
head := sq.queue[0]
if !sq.abortCutoffTime.IsZero() && head.EnqueuedAt.Before(sq.abortCutoffTime) {
sq.queue = sq.queue[1:]
head.ResultCh <- RunOutcome{Err: ErrMessageStale}
close(head.ResultCh)
slog.Debug("scheduler: skipped stale message",
"session", sq.key,
"enqueued", head.EnqueuedAt,
"cutoff", sq.abortCutoffTime,
)
continue
}
// Clear cutoff once a non-stale message is found
sq.abortCutoffTime = time.Time{}
break
}
if len(sq.queue) == 0 {
return
}
@@ -234,7 +264,7 @@ func (sq *SessionQueue) startOne(ctx context.Context) {
runID := pending.Req.RunID
runCtx, cancel := context.WithCancel(ctx)
sq.activeRuns[runID] = cancel
sq.activeRuns[runID] = activeRunEntry{cancel: cancel, generation: sq.generation}
sq.activeOrder = append(sq.activeOrder, runID)
lane := sq.laneMgr.Get(sq.lane)
@@ -242,14 +272,16 @@ func (sq *SessionQueue) startOne(ctx context.Context) {
lane = sq.laneMgr.Get(LaneMain)
}
gen := sq.generation // capture generation under lock
if lane == nil {
// No lane available — run directly
go sq.executeRun(runCtx, runID, pending)
go sq.executeRun(runCtx, runID, gen, pending)
return
}
err := lane.Submit(ctx, func() {
sq.executeRun(runCtx, runID, pending)
sq.executeRun(runCtx, runID, gen, pending)
})
if err != nil {
pending.ResultCh <- RunOutcome{Err: err}
@@ -261,14 +293,21 @@ func (sq *SessionQueue) startOne(ctx context.Context) {
}
// executeRun runs the agent and then starts the next queued message(s) if capacity allows.
func (sq *SessionQueue) executeRun(ctx context.Context, runID string, pending *PendingRequest) {
func (sq *SessionQueue) executeRun(ctx context.Context, runID string, runGeneration uint64, pending *PendingRequest) {
result, err := sq.runFn(ctx, pending.Req)
pending.ResultCh <- RunOutcome{Result: result, Err: err}
close(pending.ResultCh)
sq.mu.Lock()
delete(sq.activeRuns, runID)
sq.removeFromOrder(runID)
// Check generation: ignore stale completions from a previous generation.
if entry, ok := sq.activeRuns[runID]; ok && entry.generation == sq.generation {
delete(sq.activeRuns, runID)
sq.removeFromOrder(runID)
} else if runGeneration != sq.generation {
// Stale completion from old generation — skip cleanup.
sq.mu.Unlock()
return
}
if sq.hasCapacity() && len(sq.queue) > 0 {
// Use parentCtx (not the per-run ctx which may be cancelled)
@@ -330,7 +369,7 @@ func (sq *SessionQueue) drainQueue(outcome RunOutcome) {
}
// CancelOne stops the oldest active run (FIFO).
// Does NOT drain the pending queue. Used by /stop command.
// Does NOT drain the pending queue or set abort cutoff. Used by /stop command.
// Returns true if an active run was actually cancelled.
func (sq *SessionQueue) CancelOne() bool {
sq.mu.Lock()
@@ -342,8 +381,8 @@ func (sq *SessionQueue) CancelOne() bool {
// Cancel the oldest active run
runID := sq.activeOrder[0]
if cancel, ok := sq.activeRuns[runID]; ok {
cancel()
if entry, ok := sq.activeRuns[runID]; ok {
entry.cancel()
delete(sq.activeRuns, runID)
sq.activeOrder = sq.activeOrder[1:]
return true
@@ -352,15 +391,18 @@ func (sq *SessionQueue) CancelOne() bool {
}
// CancelAll stops all active runs and drains all pending requests.
// Sets abort cutoff so stale queued messages are skipped on next schedule.
// Used by /stopall command.
// Returns true if any active run was actually cancelled.
func (sq *SessionQueue) CancelAll() bool {
sq.mu.Lock()
defer sq.mu.Unlock()
sq.abortCutoffTime = time.Now() // mark cutoff for stale message skipping
cancelled := false
for runID, cancel := range sq.activeRuns {
cancel()
for runID, entry := range sq.activeRuns {
entry.cancel()
delete(sq.activeRuns, runID)
cancelled = true
}
@@ -395,4 +437,17 @@ func (sq *SessionQueue) QueueLen() int {
return len(sq.queue)
}
// Reset bumps the generation counter, cancels all active runs, and drains
// the pending queue. Stale completions from the old generation are ignored.
// Used during in-process restart (e.g. SIGUSR1).
func (sq *SessionQueue) Reset() {
sq.mu.Lock()
defer sq.mu.Unlock()
sq.generation++
for _, entry := range sq.activeRuns {
entry.cancel()
}
sq.activeRuns = make(map[string]activeRunEntry)
sq.activeOrder = nil
sq.drainQueue(RunOutcome{Err: ErrLaneCleared})
}
+24
View File
@@ -4,6 +4,7 @@ import (
"context"
"log/slog"
"sync"
"sync/atomic"
"github.com/nextlevelbuilder/goclaw/internal/agent"
)
@@ -22,6 +23,7 @@ type Scheduler struct {
config QueueConfig
runFn RunFunc
mu sync.RWMutex
draining atomic.Bool // set during graceful shutdown to reject new requests
tokenEstimateFn TokenEstimateFunc // optional: for adaptive throttle
}
@@ -45,15 +47,35 @@ func (s *Scheduler) SetTokenEstimateFunc(fn TokenEstimateFunc) {
s.tokenEstimateFn = fn
}
// MarkDraining signals that the gateway is shutting down.
// New Schedule/ScheduleWithOpts calls will return ErrGatewayDraining immediately.
// Active runs continue to completion.
func (s *Scheduler) MarkDraining() {
s.draining.Store(true)
slog.Info("scheduler: marked as draining, new requests will be rejected")
}
// Schedule submits a run request to the appropriate session queue and lane.
// Returns a channel that receives the result when the run completes.
func (s *Scheduler) Schedule(ctx context.Context, lane string, req agent.RunRequest) <-chan RunOutcome {
if s.draining.Load() {
ch := make(chan RunOutcome, 1)
ch <- RunOutcome{Err: ErrGatewayDraining}
close(ch)
return ch
}
sq := s.getOrCreateSession(req.SessionKey, lane)
return sq.Enqueue(ctx, req)
}
// ScheduleWithOpts submits a run request with per-session overrides.
func (s *Scheduler) ScheduleWithOpts(ctx context.Context, lane string, req agent.RunRequest, opts ScheduleOpts) <-chan RunOutcome {
if s.draining.Load() {
ch := make(chan RunOutcome, 1)
ch <- RunOutcome{Err: ErrGatewayDraining}
close(ch)
return ch
}
sq := s.getOrCreateSession(req.SessionKey, lane)
if opts.MaxConcurrent > 0 {
sq.SetMaxConcurrent(opts.MaxConcurrent)
@@ -115,7 +137,9 @@ func (s *Scheduler) CancelOneSession(sessionKey string) bool {
}
// Stop shuts down all lanes and clears session queues.
// Automatically marks the scheduler as draining before stopping.
func (s *Scheduler) Stop() {
s.MarkDraining()
s.lanes.StopAll()
}
+282
View File
@@ -0,0 +1,282 @@
package tools
import (
"os"
"path/filepath"
"runtime"
"testing"
)
// helper to create a temp workspace with files
func setupWorkspace(t *testing.T) string {
t.Helper()
dir := t.TempDir()
// Create a normal file
if err := os.WriteFile(filepath.Join(dir, "hello.txt"), []byte("hello"), 0644); err != nil {
t.Fatal(err)
}
// Create a subdirectory
sub := filepath.Join(dir, "subdir")
if err := os.MkdirAll(sub, 0755); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(sub, "nested.txt"), []byte("nested"), 0644); err != nil {
t.Fatal(err)
}
return dir
}
func TestResolvePath_NormalFile(t *testing.T) {
ws := setupWorkspace(t)
resolved, err := resolvePath("hello.txt", ws, true)
if err != nil {
t.Fatalf("expected success, got: %v", err)
}
if filepath.Base(resolved) != "hello.txt" {
t.Fatalf("expected hello.txt, got: %s", resolved)
}
}
func TestResolvePath_NestedFile(t *testing.T) {
ws := setupWorkspace(t)
resolved, err := resolvePath("subdir/nested.txt", ws, true)
if err != nil {
t.Fatalf("expected success, got: %v", err)
}
if filepath.Base(resolved) != "nested.txt" {
t.Fatalf("expected nested.txt, got: %s", resolved)
}
}
func TestResolvePath_AbsolutePath(t *testing.T) {
ws := setupWorkspace(t)
absPath := filepath.Join(ws, "hello.txt")
resolved, err := resolvePath(absPath, ws, true)
if err != nil {
t.Fatalf("expected success, got: %v", err)
}
if resolved != absPath {
// canonical path might differ if ws has symlinks (e.g. /tmp on macOS)
realAbs, _ := filepath.EvalSymlinks(absPath)
if resolved != realAbs {
t.Fatalf("expected %s or %s, got: %s", absPath, realAbs, resolved)
}
}
}
func TestResolvePath_TraversalBlocked(t *testing.T) {
ws := setupWorkspace(t)
_, err := resolvePath("../../etc/passwd", ws, true)
if err == nil {
t.Fatal("expected error for path traversal, got nil")
}
}
func TestResolvePath_AbsoluteEscapeBlocked(t *testing.T) {
ws := setupWorkspace(t)
_, err := resolvePath("/etc/passwd", ws, true)
if err == nil {
t.Fatal("expected error for absolute path outside workspace, got nil")
}
}
func TestResolvePath_SymlinkEscapeBlocked(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("symlinks require special privileges on Windows")
}
ws := setupWorkspace(t)
// Create a file outside workspace
outside := t.TempDir()
secret := filepath.Join(outside, "secret.txt")
if err := os.WriteFile(secret, []byte("secret"), 0644); err != nil {
t.Fatal(err)
}
// Create symlink inside workspace pointing outside
link := filepath.Join(ws, "evil_link")
if err := os.Symlink(secret, link); err != nil {
t.Fatal(err)
}
_, err := resolvePath("evil_link", ws, true)
if err == nil {
t.Fatal("expected error for symlink escaping workspace, got nil")
}
}
func TestResolvePath_SymlinkInsideWorkspaceAllowed(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("symlinks require special privileges on Windows")
}
ws := setupWorkspace(t)
// Create symlink pointing to a file within workspace
target := filepath.Join(ws, "hello.txt")
link := filepath.Join(ws, "good_link")
if err := os.Symlink(target, link); err != nil {
t.Fatal(err)
}
resolved, err := resolvePath("good_link", ws, true)
if err != nil {
t.Fatalf("expected success for symlink within workspace, got: %v", err)
}
// Should resolve to canonical path of target
realTarget, _ := filepath.EvalSymlinks(target)
if resolved != realTarget {
t.Fatalf("expected %s, got: %s", realTarget, resolved)
}
}
func TestResolvePath_BrokenSymlinkBlocked(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("symlinks require special privileges on Windows")
}
ws := setupWorkspace(t)
// Create symlink pointing to non-existent file outside workspace
link := filepath.Join(ws, "broken_link")
if err := os.Symlink("/nonexistent/secret", link); err != nil {
t.Fatal(err)
}
_, err := resolvePath("broken_link", ws, true)
if err == nil {
t.Fatal("expected error for broken symlink outside workspace, got nil")
}
}
func TestResolvePath_DirSymlinkEscapeBlocked(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("symlinks require special privileges on Windows")
}
ws := setupWorkspace(t)
// Create a directory symlink pointing outside workspace
outside := t.TempDir()
if err := os.WriteFile(filepath.Join(outside, "secret.txt"), []byte("secret"), 0644); err != nil {
t.Fatal(err)
}
link := filepath.Join(ws, "evil_dir")
if err := os.Symlink(outside, link); err != nil {
t.Fatal(err)
}
_, err := resolvePath("evil_dir/secret.txt", ws, true)
if err == nil {
t.Fatal("expected error for directory symlink escape, got nil")
}
}
func TestResolvePath_NonExistentFileInWorkspace(t *testing.T) {
ws := setupWorkspace(t)
resolved, err := resolvePath("new_file.txt", ws, true)
if err != nil {
t.Fatalf("expected success for non-existent file in workspace, got: %v", err)
}
if filepath.Dir(resolved) == "" {
t.Fatal("expected resolved path to have directory")
}
}
func TestResolvePath_UnrestrictedAllowsEscape(t *testing.T) {
ws := setupWorkspace(t)
// restrict=false should allow any path
resolved, err := resolvePath("/etc/hosts", ws, false)
if err != nil {
t.Fatalf("expected success with restrict=false, got: %v", err)
}
if resolved != "/etc/hosts" {
t.Fatalf("expected /etc/hosts, got: %s", resolved)
}
}
func TestCheckHardlink_NormalFile(t *testing.T) {
dir := t.TempDir()
f := filepath.Join(dir, "normal.txt")
if err := os.WriteFile(f, []byte("data"), 0644); err != nil {
t.Fatal(err)
}
if err := checkHardlink(f); err != nil {
t.Fatalf("expected no error for normal file, got: %v", err)
}
}
func TestCheckHardlink_HardlinkedFileBlocked(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("hardlinks behave differently on Windows")
}
dir := t.TempDir()
original := filepath.Join(dir, "original.txt")
if err := os.WriteFile(original, []byte("data"), 0644); err != nil {
t.Fatal(err)
}
hardlink := filepath.Join(dir, "hardlink.txt")
if err := os.Link(original, hardlink); err != nil {
t.Fatal(err)
}
// Both original and hardlink should be rejected (nlink=2)
if err := checkHardlink(original); err == nil {
t.Fatal("expected error for hardlinked file (original), got nil")
}
if err := checkHardlink(hardlink); err == nil {
t.Fatal("expected error for hardlinked file (link), got nil")
}
}
func TestCheckHardlink_DirectoryAllowed(t *testing.T) {
dir := t.TempDir()
// Directories naturally have nlink > 1, should be exempt
if err := checkHardlink(dir); err != nil {
t.Fatalf("expected no error for directory, got: %v", err)
}
}
func TestCheckHardlink_NonExistent(t *testing.T) {
if err := checkHardlink("/nonexistent/path"); err != nil {
t.Fatalf("expected no error for non-existent file, got: %v", err)
}
}
func TestCheckDeniedPath(t *testing.T) {
ws := setupWorkspace(t)
wsReal, _ := filepath.EvalSymlinks(ws)
denied := filepath.Join(wsReal, ".goclaw", "secrets")
if err := os.MkdirAll(filepath.Dir(denied), 0755); err != nil {
t.Fatal(err)
}
err := checkDeniedPath(denied, ws, []string{".goclaw"})
if err == nil {
t.Fatal("expected error for denied path, got nil")
}
// Non-denied path should pass
err = checkDeniedPath(filepath.Join(wsReal, "hello.txt"), ws, []string{".goclaw"})
if err != nil {
t.Fatalf("expected no error for non-denied path, got: %v", err)
}
}
func TestIsPathInside(t *testing.T) {
tests := []struct {
child, parent string
want bool
}{
{"/a/b/c", "/a/b", true},
{"/a/b", "/a/b", true},
{"/a/bc", "/a/b", false}, // not a child, just prefix match
{"/a", "/a/b", false},
{"/x/y", "/a/b", false},
}
for _, tt := range tests {
got := isPathInside(tt.child, tt.parent)
if got != tt.want {
t.Errorf("isPathInside(%q, %q) = %v, want %v", tt.child, tt.parent, got, tt.want)
}
}
}
+118 -22
View File
@@ -7,6 +7,7 @@ import (
"os"
"path/filepath"
"strings"
"syscall"
"github.com/nextlevelbuilder/goclaw/internal/sandbox"
)
@@ -157,12 +158,27 @@ func resolvePathWithAllowed(path, workspace string, restrict bool, allowedPrefix
return resolved, nil
}
// If restricted and denied, check if path falls under an allowed prefix.
// Resolve symlinks in the candidate path for safe comparison.
cleaned := filepath.Clean(path)
absPath, _ := filepath.Abs(cleaned)
real, evalErr := filepath.EvalSymlinks(absPath)
if evalErr != nil {
// Try resolving parent for non-existent files
parentReal, parentErr := filepath.EvalSymlinks(filepath.Dir(absPath))
if parentErr != nil {
return "", err
}
real = filepath.Join(parentReal, filepath.Base(absPath))
}
for _, prefix := range allowedPrefixes {
absPrefix, _ := filepath.Abs(prefix)
if strings.HasPrefix(cleaned, absPrefix) {
slog.Debug("read_file: allowed by prefix", "path", cleaned, "prefix", absPrefix)
return cleaned, nil
prefixReal, prefixErr := filepath.EvalSymlinks(absPrefix)
if prefixErr != nil {
prefixReal = absPrefix
}
if isPathInside(real, prefixReal) {
slog.Debug("read_file: allowed by prefix", "path", real, "prefix", prefixReal)
return real, nil
}
}
slog.Warn("read_file: access denied", "path", cleaned, "workspace", workspace, "allowedPrefixes", allowedPrefixes)
@@ -171,15 +187,21 @@ func resolvePathWithAllowed(path, workspace string, restrict bool, allowedPrefix
// checkDeniedPath returns an error if the resolved path falls under any denied prefix.
// Denied prefixes are relative to the workspace (e.g. ".goclaw" denies workspace/.goclaw/).
// The resolved path should already be canonical (from resolvePath with restrict=true).
func checkDeniedPath(resolved, workspace string, deniedPrefixes []string) error {
if len(deniedPrefixes) == 0 {
return nil
}
absResolved, _ := filepath.Abs(resolved)
absWorkspace, _ := filepath.Abs(workspace)
// Resolve workspace to canonical form for consistent comparison.
wsReal, err := filepath.EvalSymlinks(absWorkspace)
if err != nil {
wsReal = absWorkspace
}
for _, prefix := range deniedPrefixes {
denied := filepath.Join(absWorkspace, prefix)
if strings.HasPrefix(absResolved, denied) {
denied := filepath.Join(wsReal, prefix)
if isPathInside(absResolved, denied) {
return fmt.Errorf("access denied: path %s is restricted", prefix)
}
}
@@ -187,27 +209,101 @@ func checkDeniedPath(resolved, workspace string, deniedPrefixes []string) error
}
// resolvePath resolves a path relative to the workspace and validates it.
// When restrict=true, resolves symlinks to canonical paths and rejects
// paths that escape the workspace boundary (symlink/hardlink attacks).
func resolvePath(path, workspace string, restrict bool) (string, error) {
var resolved string
if filepath.IsAbs(path) {
if restrict {
absWorkspace, _ := filepath.Abs(workspace)
if !strings.HasPrefix(path, absWorkspace) {
return "", fmt.Errorf("access denied: path outside workspace")
resolved = filepath.Clean(path)
} else {
resolved = filepath.Clean(filepath.Join(workspace, path))
}
if !restrict {
return resolved, nil
}
// Resolve workspace to canonical path (follow symlinks in workspace path itself).
absWorkspace, _ := filepath.Abs(workspace)
wsReal, err := filepath.EvalSymlinks(absWorkspace)
if err != nil {
wsReal = absWorkspace // workspace doesn't exist yet — use as-is
}
// Resolve the target path to canonical form (follows all symlinks).
absResolved, _ := filepath.Abs(resolved)
real, err := filepath.EvalSymlinks(absResolved)
if err != nil {
if os.IsNotExist(err) {
// Check if the path itself is a symlink (broken/dangling).
// Lstat doesn't follow symlinks, so it succeeds even for broken ones.
if linfo, lerr := os.Lstat(absResolved); lerr == nil && linfo.Mode()&os.ModeSymlink != 0 {
// It's a broken symlink — read target and validate.
target, readErr := os.Readlink(absResolved)
if readErr != nil {
return "", fmt.Errorf("access denied: cannot resolve symlink")
}
if !filepath.IsAbs(target) {
target = filepath.Join(filepath.Dir(absResolved), target)
}
target = filepath.Clean(target)
if !isPathInside(target, wsReal) {
slog.Warn("security.broken_symlink_escape", "path", path, "target", target, "workspace", wsReal)
return "", fmt.Errorf("access denied: broken symlink target outside workspace")
}
real = target
} else {
// Truly non-existent file (not a symlink): resolve parent and re-validate.
parentReal, parentErr := filepath.EvalSymlinks(filepath.Dir(absResolved))
if parentErr != nil {
return "", fmt.Errorf("access denied: cannot resolve path")
}
real = filepath.Join(parentReal, filepath.Base(absResolved))
}
}
return filepath.Clean(path), nil
}
resolved := filepath.Join(workspace, path)
resolved = filepath.Clean(resolved)
if restrict {
absWorkspace, _ := filepath.Abs(workspace)
absResolved, _ := filepath.Abs(resolved)
if !strings.HasPrefix(absResolved, absWorkspace) {
return "", fmt.Errorf("access denied: path outside workspace")
} else {
// Permission error or other — reject.
slog.Warn("security.path_resolve_failed", "path", path, "error", err)
return "", fmt.Errorf("access denied: cannot resolve path")
}
}
return resolved, nil
// Validate canonical path stays within canonical workspace.
if !isPathInside(real, wsReal) {
slog.Warn("security.path_escape", "path", path, "resolved", real, "workspace", wsReal)
return "", fmt.Errorf("access denied: path outside workspace")
}
// Reject hardlinked files (nlink > 1) to prevent hardlink-based escapes.
if err := checkHardlink(real); err != nil {
return "", err
}
return real, nil
}
// isPathInside checks whether child is inside or equal to parent directory.
func isPathInside(child, parent string) bool {
if child == parent {
return true
}
return strings.HasPrefix(child, parent+string(filepath.Separator))
}
// checkHardlink rejects regular files with nlink > 1 (hardlink attack prevention).
// Directories naturally have nlink > 1 and are exempt.
func checkHardlink(path string) error {
info, err := os.Lstat(path)
if err != nil {
return nil // non-existent files are OK — will fail at read/write
}
if info.IsDir() {
return nil
}
if stat, ok := info.Sys().(*syscall.Stat_t); ok {
if stat.Nlink > 1 {
slog.Warn("security.hardlink_rejected", "path", path, "nlink", stat.Nlink)
return fmt.Errorf("access denied: hardlinked file not allowed")
}
}
return nil
}