fix(agent): harden scheduled task dispatch

commit: cc1e45a6407461b29bcc2102e0cb32f6eaf71452
author: brice · 2mo ago · 2026-08-15 15:58 UTC
parent: 3d78ab0
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +287 -55

M README.md +2 -2


@@ -9,7 +9,7 @@ Yet another coding agent harness, lightweight and written (vibe-slopped) in go.
9	9
10	10		- one static binary.
11	11		- built-in providers for Anthropic, OpenAI/Codex/Responses, Kimi, DeepSeek, Google Gemini/Vertex, GitHub Copilot, Bedrock, Azure OpenAI, OpenRouter, Groq, Cerebras, xAI, Together, Hugging Face, Mistral, Moonshot, Z.AI, Xiaomi, MiniMax, Fireworks, Vercel AI Gateway, OpenCode, Cloudflare AI, and Ollama/local models.
12		-	- ten core built-in tools (read, write, edit, bash, create_worktree, grep, lsp, web_search, update_goal, and schedule); the conditional `skill` tool is available when skills are enabled. See [docs/web-search.md](docs/web-search.md) for web-search egress and availability boundaries.
	12	+	- ten core built-in tools (read, write, edit, bash, create_worktree, grep, lsp, web_search, update_goal, and schedule; `schedule` is interactive-only); the conditional `skill` tool is available when skills are enabled. See [docs/web-search.md](docs/web-search.md) for web-search egress and availability boundaries.
13	13		- three run modes (interactive tui, print, json).
14	14		- built-in telegram bot.
15	15		- extensions in any language via subprocess + json-rpc. None installed by default; opt in with `zut ext install` or `zut --ext`. See [docs/extensions.md](docs/extensions.md).

@@ -255,7 +255,7 @@ Print-mode stats contain `provider`, `model`, `prompt_tokens`, `reasoning_tokens
255	255		- `lsp`: query configured language servers and linters for diagnostics, definitions, references, hover information, symbols, renames, code actions, capabilities, and raw protocol requests. LSP servers use stdio JSON-RPC; project `lsp.json` files can add or override servers and CLI linters. Diagnostics are bounded, sorted, deduplicated by path/severity/code/start position, and repeated issues are grouped before they reach the model. See [docs/lsp.md](docs/lsp.md).
256	256		- `web_search`: search the public web through DuckDuckGo HTML and return bounded source titles, URLs, and snippets. Enabled by default for normal CLI sessions; it is unavailable to bot, default SDK, and packaged-agent runs. See [docs/web-search.md](docs/web-search.md).
257	257		- `update_goal`: let the main agent start its own mission when none is active, set the next bounded goal within an active mission, or mark the active goal `done` or `blocked`. Pausing, resuming, and clearing missions remain user-controlled through `/goal`.
258		-	- `schedule`: add, list, or cancel recurring five-field cron prompts for the current session. A task captures the machine's current timezone when it is created. While this zut process remains open, a due task waits for an active target turn to finish, or resumes an inactive target session in the background; its output is recorded in that target transcript. Resumed background tasks also show a completion notification in the current TUI. Scheduling is in-memory only: quitting zut cancels pending tasks, and schedules do not survive restart or reboot.
	258	+	- `schedule`: add, list, or cancel recurring five-field cron prompts for the current interactive session. A task captures the machine's current timezone when it is created. While this zut process remains open, a due task waits for an active target turn to finish, or resumes an inactive target session in the background; its output is recorded in that target transcript. Resumed background tasks also show a completion notification in the current TUI. Scheduling is in-memory only: quitting zut cancels pending tasks, and schedules do not survive restart or reboot. Print, stream, JSON, and RPC modes do not expose this tool.
259	259
260	260		When the sandbox is on (see `/jail`), filesystem tools, `grep`, and LSP workspace edits refuse paths outside the session cwd. `create_worktree` also requires its repository root, configured worktree destination, optional `.gitignore`, and Git metadata to remain inside the jail. Jail does not sandbox web-search network egress.
261	261


@@ -9,7 +9,7 @@ Yet another coding agent harness, lightweight and written (vibe-slopped) in go.
9			9
10		- one static binary.	10		- one static binary.
11		- built-in providers for Anthropic, OpenAI/Codex/Responses, Kimi, DeepSeek, Google Gemini/Vertex, GitHub Copilot, Bedrock, Azure OpenAI, OpenRouter, Groq, Cerebras, xAI, Together, Hugging Face, Mistral, Moonshot, Z.AI, Xiaomi, MiniMax, Fireworks, Vercel AI Gateway, OpenCode, Cloudflare AI, and Ollama/local models.	11		- built-in providers for Anthropic, OpenAI/Codex/Responses, Kimi, DeepSeek, Google Gemini/Vertex, GitHub Copilot, Bedrock, Azure OpenAI, OpenRouter, Groq, Cerebras, xAI, Together, Hugging Face, Mistral, Moonshot, Z.AI, Xiaomi, MiniMax, Fireworks, Vercel AI Gateway, OpenCode, Cloudflare AI, and Ollama/local models.
12	-	- ten core built-in tools (read, write, edit, bash, create_worktree, grep, lsp, web_search, update_goal, and schedule); the conditional `skill` tool is available when skills are enabled. See [docs/web-search.md](docs/web-search.md) for web-search egress and availability boundaries.	12	+	- ten core built-in tools (read, write, edit, bash, create_worktree, grep, lsp, web_search, update_goal, and schedule; `schedule` is interactive-only); the conditional `skill` tool is available when skills are enabled. See [docs/web-search.md](docs/web-search.md) for web-search egress and availability boundaries.
13		- three run modes (interactive tui, print, json).	13		- three run modes (interactive tui, print, json).
14		- built-in telegram bot.	14		- built-in telegram bot.
15		- extensions in any language via subprocess + json-rpc. None installed by default; opt in with `zut ext install` or `zut --ext`. See [docs/extensions.md](docs/extensions.md).	15		- extensions in any language via subprocess + json-rpc. None installed by default; opt in with `zut ext install` or `zut --ext`. See [docs/extensions.md](docs/extensions.md).

@@ -255,7 +255,7 @@ Print-mode stats contain `provider`, `model`, `prompt_tokens`, `reasoning_tokens
255		- `lsp`: query configured language servers and linters for diagnostics, definitions, references, hover information, symbols, renames, code actions, capabilities, and raw protocol requests. LSP servers use stdio JSON-RPC; project `lsp.json` files can add or override servers and CLI linters. Diagnostics are bounded, sorted, deduplicated by path/severity/code/start position, and repeated issues are grouped before they reach the model. See [docs/lsp.md](docs/lsp.md).	255		- `lsp`: query configured language servers and linters for diagnostics, definitions, references, hover information, symbols, renames, code actions, capabilities, and raw protocol requests. LSP servers use stdio JSON-RPC; project `lsp.json` files can add or override servers and CLI linters. Diagnostics are bounded, sorted, deduplicated by path/severity/code/start position, and repeated issues are grouped before they reach the model. See [docs/lsp.md](docs/lsp.md).
256		- `web_search`: search the public web through DuckDuckGo HTML and return bounded source titles, URLs, and snippets. Enabled by default for normal CLI sessions; it is unavailable to bot, default SDK, and packaged-agent runs. See [docs/web-search.md](docs/web-search.md).	256		- `web_search`: search the public web through DuckDuckGo HTML and return bounded source titles, URLs, and snippets. Enabled by default for normal CLI sessions; it is unavailable to bot, default SDK, and packaged-agent runs. See [docs/web-search.md](docs/web-search.md).
257		- `update_goal`: let the main agent start its own mission when none is active, set the next bounded goal within an active mission, or mark the active goal `done` or `blocked`. Pausing, resuming, and clearing missions remain user-controlled through `/goal`.	257		- `update_goal`: let the main agent start its own mission when none is active, set the next bounded goal within an active mission, or mark the active goal `done` or `blocked`. Pausing, resuming, and clearing missions remain user-controlled through `/goal`.
258	-	- `schedule`: add, list, or cancel recurring five-field cron prompts for the current session. A task captures the machine's current timezone when it is created. While this zut process remains open, a due task waits for an active target turn to finish, or resumes an inactive target session in the background; its output is recorded in that target transcript. Resumed background tasks also show a completion notification in the current TUI. Scheduling is in-memory only: quitting zut cancels pending tasks, and schedules do not survive restart or reboot.	258	+	- `schedule`: add, list, or cancel recurring five-field cron prompts for the current interactive session. A task captures the machine's current timezone when it is created. While this zut process remains open, a due task waits for an active target turn to finish, or resumes an inactive target session in the background; its output is recorded in that target transcript. Resumed background tasks also show a completion notification in the current TUI. Scheduling is in-memory only: quitting zut cancels pending tasks, and schedules do not survive restart or reboot. Print, stream, JSON, and RPC modes do not expose this tool.
259			259
260		When the sandbox is on (see `/jail`), filesystem tools, `grep`, and LSP workspace edits refuse paths outside the session cwd. `create_worktree` also requires its repository root, configured worktree destination, optional `.gitignore`, and Git metadata to remain inside the jail. Jail does not sandbox web-search network egress.	260		When the sandbox is on (see `/jail`), filesystem tools, `grep`, and LSP workspace edits refuse paths outside the session cwd. `create_worktree` also requires its repository root, configured worktree destination, optional `.gitignore`, and Git metadata to remain inside the jail. Jail does not sandbox web-search network egress.
261			261

M packages/agent/cli.go +18 -3


@@ -1487,6 +1487,19 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1487	1487		activeModel := r.Model
1488	1488		var sess *core.Session
1489	1489		scheduleEngine := scheduler.NewEngine(nil)
	1490	+	var scheduledSessionMu sync.Mutex
	1491	+	scheduledSessionLocks := make(map[string]*sync.Mutex)
	1492	+	lockScheduledSession := func(sessionID string) func() {
	1493	+	scheduledSessionMu.Lock()
	1494	+	lock := scheduledSessionLocks[sessionID]
	1495	+	if lock == nil {
	1496	+	lock = new(sync.Mutex)
	1497	+	scheduledSessionLocks[sessionID] = lock
	1498	+	}
	1499	+	scheduledSessionMu.Unlock()
	1500	+	lock.Lock()
	1501	+	return lock.Unlock
	1502	+	}
1490	1503
1491	1504		// Build the extension manager BEFORE the agent so we can fold
1492	1505		// extension-defined tools into the registry. Attach the interactive

@@ -2714,7 +2727,9 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2714	2727		schedulerCtx, cancelScheduler := context.WithCancel(ctx)
2715	2728		defer cancelScheduler()
2716	2729		go func() {
2717		-	if err := scheduleEngine.Run(schedulerCtx, func(task scheduler.Task) error {
	2730	+	if err := scheduleEngine.Run(schedulerCtx, func(taskCtx context.Context, task scheduler.Task) error {
	2731	+	unlockSession := lockScheduledSession(task.SessionID)
	2732	+	defer unlockSession()
2718	2733		// Keep a background session run mutually exclusive with /session
2719	2734		// transitions. The active target uses Interactive's distinct-turn
2720	2735		// ingress instead, so it never injects into an in-flight model loop.

@@ -2724,7 +2739,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2724	2739		activeTarget := sess != nil && sess.ID == task.SessionID
2725	2740		persistMu.Unlock()
2726	2741		if activeTarget {
2727		-	return iv.SubmitFollowUp(schedulerCtx, task.Message)
	2742	+	return iv.SubmitFollowUp(taskCtx, task.Message)
2728	2743		}
2729	2744		// A resumed background agent must schedule any follow-up work against
2730	2745		// the transcript it resumed, not whichever session is visible in the

@@ -2739,7 +2754,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2739	2754		}
2740	2755		return registry
2741	2756		}
2742		-	err := runScheduledSession(schedulerCtx, task, args, r, extMgr, backgroundRegistry)
	2757	+	err := runScheduledSession(taskCtx, task, args, r, extMgr, backgroundRegistry)
2743	2758		if err != nil {
2744	2759		iv.Notify("scheduler", "error", fmt.Sprintf("scheduled task %s failed: %v", task.ID, err))
2745	2760		} else {

 	activeModel := r.Model
 	var sess *core.Session
 	scheduleEngine := scheduler.NewEngine(nil)
+	var scheduledSessionMu sync.Mutex
+	scheduledSessionLocks := make(map[string]*sync.Mutex)
+	lockScheduledSession := func(sessionID string) func() {
+		scheduledSessionMu.Lock()
+		lock := scheduledSessionLocks[sessionID]
+		if lock == nil {
+			lock = new(sync.Mutex)
+			scheduledSessionLocks[sessionID] = lock
+		}
+		scheduledSessionMu.Unlock()
+		lock.Lock()
+		return lock.Unlock
+	}
 	// Build the extension manager BEFORE the agent so we can fold
 	// extension-defined tools into the registry. Attach the interactive
 	schedulerCtx, cancelScheduler := context.WithCancel(ctx)
 	defer cancelScheduler()
 	go func() {
-		if err := scheduleEngine.Run(schedulerCtx, func(task scheduler.Task) error {
+		if err := scheduleEngine.Run(schedulerCtx, func(taskCtx context.Context, task scheduler.Task) error {
+			unlockSession := lockScheduledSession(task.SessionID)
+			defer unlockSession()
 			// Keep a background session run mutually exclusive with /session
 			// transitions. The active target uses Interactive's distinct-turn
 			// ingress instead, so it never injects into an in-flight model loop.
 			activeTarget := sess != nil && sess.ID == task.SessionID
 			persistMu.Unlock()
 			if activeTarget {
-				return iv.SubmitFollowUp(schedulerCtx, task.Message)
+				return iv.SubmitFollowUp(taskCtx, task.Message)
 			}
 			// A resumed background agent must schedule any follow-up work against
 			// the transcript it resumed, not whichever session is visible in the
 				}
 				return registry
 			}
-			err := runScheduledSession(schedulerCtx, task, args, r, extMgr, backgroundRegistry)
+			err := runScheduledSession(taskCtx, task, args, r, extMgr, backgroundRegistry)
 			if err != nil {
 				iv.Notify("scheduler", "error", fmt.Sprintf("scheduled task %s failed: %v", task.ID, err))
 			} else {

M packages/agent/scheduler/cron_test.go +2 -2


@@ -46,8 +46,8 @@ func TestParseCronDSTTransitions(t *testing.T) {
46	46		}
47	47		first := fallBack.Next(time.Date(2026, time.November, 1, 0, 59, 0, 0, loc))
48	48		second := fallBack.Next(first)
49		-	if first.Hour() != 1 \|\| first.Minute() != 30 \|\| second.Hour() != 1 \|\| second.Minute() != 30 \|\| first.Equal(second) {
50		-	t.Fatalf("fall-back occurrences = %v, %v; want two distinct 01:30 occurrences", first, second)
	49	+	if first.Hour() != 1 \|\| first.Minute() != 30 \|\| second.Hour() != 1 \|\| second.Minute() != 30 \|\| first.Equal(second) \|\| second.Sub(first) != time.Hour {
	50	+	t.Fatalf("fall-back occurrences = %v, %v; want 01:30 exactly one elapsed hour apart", first, second)
51	51		}
52	52
53	53		springForward, err := ParseCron("30 2 * * *", loc)


@@ -46,8 +46,8 @@ func TestParseCronDSTTransitions(t *testing.T) {
46		}	46		}
47		first := fallBack.Next(time.Date(2026, time.November, 1, 0, 59, 0, 0, loc))	47		first := fallBack.Next(time.Date(2026, time.November, 1, 0, 59, 0, 0, loc))
48		second := fallBack.Next(first)	48		second := fallBack.Next(first)
49	-	if first.Hour() != 1 \|\| first.Minute() != 30 \|\| second.Hour() != 1 \|\| second.Minute() != 30 \|\| first.Equal(second) {	49	+	if first.Hour() != 1 \|\| first.Minute() != 30 \|\| second.Hour() != 1 \|\| second.Minute() != 30 \|\| first.Equal(second) \|\| second.Sub(first) != time.Hour {
50	-	t.Fatalf("fall-back occurrences = %v, %v; want two distinct 01:30 occurrences", first, second)	50	+	t.Fatalf("fall-back occurrences = %v, %v; want 01:30 exactly one elapsed hour apart", first, second)
51		}	51		}
52			52
53		springForward, err := ParseCron("30 2 * * *", loc)	53		springForward, err := ParseCron("30 2 * * *", loc)

M packages/agent/scheduler/scheduler.go +71 -21


@@ -27,6 +27,8 @@ type Task struct {
27	27		LastError string
28	28
29	29		schedule Cron
	30	+	queued bool
	31	+	running bool
30	32		}
31	33
32	34		// NewTaskInput describes a task to add to an Engine. Location is captured at

@@ -38,9 +40,9 @@ type NewTaskInput struct {
38	40		Location *time.Location
39	41		}
40	42
41		-	// Engine serializes task scheduling and execution through one timer loop. Its
42		-	// callback runs outside the engine lock but is intentionally called one task
43		-	// at a time so a session cannot receive overlapping scheduled turns.
	43	+	// Engine owns due-task claiming through one timer loop. Executors run outside
	44	+	// the lock, allowing independent session tasks to progress concurrently; the
	45	+	// host remains responsible for serializing work for one session.
44	46		type Engine struct {
45	47		mu sync.Mutex
46	48		now func() time.Time

@@ -148,10 +150,10 @@ func (e *Engine) List() []Task {
148	150		return tasks
149	151		}
150	152
151		-	// Run waits for due tasks until ctx is cancelled. The executor is called
152		-	// synchronously for each due task, which prevents a scheduled task from
153		-	// overlapping a prior scheduled turn.
154		-	func (e *Engine) Run(ctx context.Context, executor func(Task) error) error {
	153	+	// Run waits for due tasks until ctx is cancelled. Each claimed task receives
	154	+	// ctx and executes asynchronously; cancellation prevents new callbacks and is
	155	+	// propagated to callbacks already in flight.
	156	+	func (e *Engine) Run(ctx context.Context, executor func(context.Context, Task) error) error {
155	157		if e == nil {
156	158		return fmt.Errorf("scheduler is not initialized")
157	159		}

@@ -159,8 +161,11 @@ func (e *Engine) Run(ctx context.Context, executor func(Task) error) error {
159	161		return fmt.Errorf("scheduler executor is required")
160	162		}
161	163		for {
	164	+	if ctx.Err() != nil {
	165	+	return nil
	166	+	}
162	167		now := e.now()
163		-	e.dispatchDue(now, executor)
	168	+	e.dispatchDue(ctx, now, executor)
164	169		next, ok := e.nextRun()
165	170		if !ok {
166	171		select {

@@ -196,16 +201,16 @@ func (e *Engine) Run(ctx context.Context, executor func(Task) error) error {
196	201		}
197	202		}
198	203
199		-	// dispatchDue executes all tasks due at or before now. Run is its only
200		-	// production caller; keeping it private preserves Engine's single executor
201		-	// ownership and prevents concurrent duplicate dispatches.
202		-	func (e *Engine) dispatchDue(now time.Time, executor func(Task) error) {
203		-	if e == nil \|\| executor == nil {
	204	+	// dispatchDue reserves every due task before launching an asynchronous
	205	+	// dispatch. A reservation prevents concurrent Run callers from selecting the
	206	+	// same task. dispatchTask claims it immediately before invoking the executor,
	207	+	// so Cancel can still win while a task is waiting to begin.
	208	+	func (e *Engine) dispatchDue(ctx context.Context, now time.Time, executor func(context.Context, Task) error) {
	209	+	if e == nil \|\| executor == nil \|\| ctx.Err() != nil {
204	210		return
205	211		}
206	212		for _, task := range e.takeDue(now) {
207		-	err := executor(task)
208		-	e.finish(task.ID, task.NextRun, err)
	213	+	go e.dispatchTask(ctx, task, executor)
209	214		}
210	215		}
211	216

@@ -213,10 +218,13 @@ func (e *Engine) takeDue(now time.Time) []Task {
213	218		e.mu.Lock()
214	219		defer e.mu.Unlock()
215	220		due := make([]Task, 0)
216		-	for _, task := range e.tasks {
217		-	if !task.NextRun.After(now) {
218		-	due = append(due, task)
	221	+	for id, task := range e.tasks {
	222	+	if task.queued \|\| task.running \|\| task.NextRun.After(now) {
	223	+	continue
219	224		}
	225	+	task.queued = true
	226	+	e.tasks[id] = task
	227	+	due = append(due, task)
220	228		}
221	229		sort.Slice(due, func(i, j int) bool {
222	230		if due[i].NextRun.Equal(due[j].NextRun) {

@@ -227,13 +235,50 @@ func (e *Engine) takeDue(now time.Time) []Task {
227	235		return due
228	236		}
229	237
	238	+	func (e *Engine) dispatchTask(ctx context.Context, queued Task, executor func(context.Context, Task) error) {
	239	+	if ctx.Err() != nil {
	240	+	e.releaseQueued(queued.ID, queued.NextRun)
	241	+	return
	242	+	}
	243	+	e.mu.Lock()
	244	+	task, ok := e.tasks[queued.ID]
	245	+	if !ok \|\| !task.queued \|\| !task.NextRun.Equal(queued.NextRun) {
	246	+	e.mu.Unlock()
	247	+	return // cancelled or superseded before it was claimed
	248	+	}
	249	+	task.queued = false
	250	+	task.running = true
	251	+	e.tasks[task.ID] = task
	252	+	e.mu.Unlock()
	253	+	if ctx.Err() != nil {
	254	+	e.finish(task.ID, task.NextRun, ctx.Err())
	255	+	return
	256	+	}
	257	+	err := executor(ctx, task)
	258	+	e.finish(task.ID, task.NextRun, err)
	259	+	}
	260	+
	261	+	func (e *Engine) releaseQueued(id string, scheduledFor time.Time) {
	262	+	e.mu.Lock()
	263	+	task, ok := e.tasks[id]
	264	+	if !ok \|\| !task.queued \|\| !task.NextRun.Equal(scheduledFor) {
	265	+	e.mu.Unlock()
	266	+	return
	267	+	}
	268	+	task.queued = false
	269	+	e.tasks[id] = task
	270	+	e.mu.Unlock()
	271	+	e.signalWake()
	272	+	}
	273	+
230	274		func (e *Engine) finish(id string, scheduledFor time.Time, err error) {
231	275		e.mu.Lock()
232		-	defer e.mu.Unlock()
233	276		task, ok := e.tasks[id]
234	277		if !ok \|\| !task.NextRun.Equal(scheduledFor) {
	278	+	e.mu.Unlock()
235	279		return // cancelled or replaced while the executor ran
236	280		}
	281	+	task.running = false
237	282		task.LastRun = scheduledFor
238	283		task.RunCount++
239	284		if err != nil {

@@ -247,9 +292,11 @@ func (e *Engine) finish(id string, scheduledFor time.Time, err error) {
247	292		task.NextRun = task.schedule.Next(e.now())
248	293		if task.NextRun.IsZero() {
249	294		delete(e.tasks, id)
250		-	return
	295	+	} else {
	296	+	e.tasks[id] = task
251	297		}
252		-	e.tasks[id] = task
	298	+	e.mu.Unlock()
	299	+	e.signalWake()
253	300		}
254	301
255	302		func (e *Engine) nextRun() (time.Time, bool) {

@@ -257,6 +304,9 @@ func (e *Engine) nextRun() (time.Time, bool) {
257	304		defer e.mu.Unlock()
258	305		var next time.Time
259	306		for _, task := range e.tasks {
	307	+	if task.queued \|\| task.running {
	308	+	continue
	309	+	}
260	310		if next.IsZero() \|\| task.NextRun.Before(next) {
261	311		next = task.NextRun
262	312		}

 	LastError string
 	schedule Cron
+	queued   bool
+	running  bool
 }
 // NewTaskInput describes a task to add to an Engine. Location is captured at
 	Location  *time.Location
 }
-// Engine serializes task scheduling and execution through one timer loop. Its
+// Engine owns due-task claiming through one timer loop. Executors run outside
-// callback runs outside the engine lock but is intentionally called one task
+// the lock, allowing independent session tasks to progress concurrently; the
-// at a time so a session cannot receive overlapping scheduled turns.
+// host remains responsible for serializing work for one session.
 type Engine struct {
 	mu    sync.Mutex
 	now   func() time.Time
 	return tasks
 }
-// Run waits for due tasks until ctx is cancelled. The executor is called
+// Run waits for due tasks until ctx is cancelled. Each claimed task receives
-// synchronously for each due task, which prevents a scheduled task from
+// ctx and executes asynchronously; cancellation prevents new callbacks and is
-// overlapping a prior scheduled turn.
+// propagated to callbacks already in flight.
-func (e *Engine) Run(ctx context.Context, executor func(Task) error) error {
+func (e *Engine) Run(ctx context.Context, executor func(context.Context, Task) error) error {
 	if e == nil {
 		return fmt.Errorf("scheduler is not initialized")
 	}
 		return fmt.Errorf("scheduler executor is required")
 	}
 	for {
+		if ctx.Err() != nil {
+			return nil
+		}
 		now := e.now()
-		e.dispatchDue(now, executor)
+		e.dispatchDue(ctx, now, executor)
 		next, ok := e.nextRun()
 		if !ok {
 			select {
 	}
 }
-// dispatchDue executes all tasks due at or before now. Run is its only
+// dispatchDue reserves every due task before launching an asynchronous
-// production caller; keeping it private preserves Engine's single executor
+// dispatch. A reservation prevents concurrent Run callers from selecting the
-// ownership and prevents concurrent duplicate dispatches.
+// same task. dispatchTask claims it immediately before invoking the executor,
-func (e *Engine) dispatchDue(now time.Time, executor func(Task) error) {
+// so Cancel can still win while a task is waiting to begin.
-	if e == nil || executor == nil {
+func (e *Engine) dispatchDue(ctx context.Context, now time.Time, executor func(context.Context, Task) error) {
+	if e == nil || executor == nil || ctx.Err() != nil {
 		return
 	}
 	for _, task := range e.takeDue(now) {
-		err := executor(task)
+		go e.dispatchTask(ctx, task, executor)
-		e.finish(task.ID, task.NextRun, err)
 	}
 }
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	due := make([]Task, 0)
-	for _, task := range e.tasks {
+	for id, task := range e.tasks {
-		if !task.NextRun.After(now) {
+		if task.queued || task.running || task.NextRun.After(now) {
-			due = append(due, task)
+			continue
 		}
+		task.queued = true
+		e.tasks[id] = task
+		due = append(due, task)
 	}
 	sort.Slice(due, func(i, j int) bool {
 		if due[i].NextRun.Equal(due[j].NextRun) {
 	return due
 }
+func (e *Engine) dispatchTask(ctx context.Context, queued Task, executor func(context.Context, Task) error) {
+	if ctx.Err() != nil {
+		e.releaseQueued(queued.ID, queued.NextRun)
+		return
+	}
+	e.mu.Lock()
+	task, ok := e.tasks[queued.ID]
+	if !ok || !task.queued || !task.NextRun.Equal(queued.NextRun) {
+		e.mu.Unlock()
+		return // cancelled or superseded before it was claimed
+	}
+	task.queued = false
+	task.running = true
+	e.tasks[task.ID] = task
+	e.mu.Unlock()
+	if ctx.Err() != nil {
+		e.finish(task.ID, task.NextRun, ctx.Err())
+		return
+	}
+	err := executor(ctx, task)
+	e.finish(task.ID, task.NextRun, err)
+}
+func (e *Engine) releaseQueued(id string, scheduledFor time.Time) {
+	e.mu.Lock()
+	task, ok := e.tasks[id]
+	if !ok || !task.queued || !task.NextRun.Equal(scheduledFor) {
+		e.mu.Unlock()
+		return
+	}
+	task.queued = false
+	e.tasks[id] = task
+	e.mu.Unlock()
+	e.signalWake()
+}
 func (e *Engine) finish(id string, scheduledFor time.Time, err error) {
 	e.mu.Lock()
-	defer e.mu.Unlock()
 	task, ok := e.tasks[id]
 	if !ok || !task.NextRun.Equal(scheduledFor) {
+		e.mu.Unlock()
 		return // cancelled or replaced while the executor ran
 	}
+	task.running = false
 	task.LastRun = scheduledFor
 	task.RunCount++
 	if err != nil {
 	task.NextRun = task.schedule.Next(e.now())
 	if task.NextRun.IsZero() {
 		delete(e.tasks, id)
-		return
+	} else {
+		e.tasks[id] = task
 	}
-	e.tasks[id] = task
+	e.mu.Unlock()
+	e.signalWake()
 }
 func (e *Engine) nextRun() (time.Time, bool) {
 	defer e.mu.Unlock()
 	var next time.Time
 	for _, task := range e.tasks {
+		if task.queued || task.running {
+			continue
+		}
 		if next.IsZero() || task.NextRun.Before(next) {
 			next = task.NextRun
 		}

M packages/agent/scheduler/scheduler_test.go +171 -25


@@ -3,6 +3,7 @@ package scheduler
3	3		import (
4	4		"context"
5	5		"errors"
	6	+	"sync"
6	7		"testing"
7	8		"time"
8	9		)

@@ -11,30 +12,31 @@ func TestEngineDispatchDueReschedulesTask(t *testing.T) {
11	12		loc := time.UTC
12	13		now := time.Date(2026, time.August, 15, 12, 0, 0, 0, loc)
13	14		engine := NewEngine(func() time.Time { return now })
14		-	created, err := engine.Add(NewTaskInput{
15		-	SessionID: "session-1",
16		-	Cron: "* * * * *",
17		-	Message: "check status",
18		-	Location: loc,
19		-	})
	15	+	created, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "check status", Location: loc})
20	16		if err != nil {
21	17		t.Fatalf("Add() error = %v", err)
22	18		}
23	19
24		-	var executed []Task
	20	+	done := make(chan Task, 1)
25	21		now = now.Add(time.Minute)
26		-	engine.dispatchDue(now, func(task Task) error {
27		-	executed = append(executed, task)
	22	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
	23	+	done <- task
28	24		return nil
29	25		})
30		-	if len(executed) != 1 \|\| executed[0].ID != created.ID {
31		-	t.Fatalf("executed = %#v, want task %q", executed, created.ID)
	26	+	select {
	27	+	case task := <-done:
	28	+	if task.ID != created.ID {
	29	+	t.Fatalf("executed task ID = %q, want %q", task.ID, created.ID)
	30	+	}
	31	+	case <-time.After(time.Second):
	32	+	t.Fatal("due task was not executed")
32	33		}
33	34
	35	+	eventually(t, func() bool {
	36	+	tasks := engine.List()
	37	+	return len(tasks) == 1 && tasks[0].RunCount == 1
	38	+	})
34	39		tasks := engine.List()
35		-	if len(tasks) != 1 {
36		-	t.Fatalf("List() length = %d, want 1", len(tasks))
37		-	}
38	40		if want := now.Add(time.Minute); !tasks[0].NextRun.Equal(want) {
39	41		t.Fatalf("NextRun = %v, want %v", tasks[0].NextRun, want)
40	42		}

@@ -51,14 +53,11 @@ func TestEngineRecordsExecutionFailure(t *testing.T) {
51	53		t.Fatalf("Add() error = %v", err)
52	54		}
53	55
54		-	engine.dispatchDue(now.Add(time.Minute), func(Task) error { return errors.New("provider unavailable") })
55		-	got, ok := engine.Get(task.ID)
56		-	if !ok {
57		-	t.Fatal("Get() did not return scheduled task")
58		-	}
59		-	if got.LastError != "provider unavailable" {
60		-	t.Fatalf("LastError = %q", got.LastError)
61		-	}
	56	+	engine.dispatchDue(context.Background(), now.Add(time.Minute), func(context.Context, Task) error { return errors.New("provider unavailable") })
	57	+	eventually(t, func() bool {
	58	+	got, ok := engine.Get(task.ID)
	59	+	return ok && got.LastError == "provider unavailable"
	60	+	})
62	61		}
63	62
64	63		func TestEngineCancel(t *testing.T) {

@@ -75,11 +74,158 @@ func TestEngineCancel(t *testing.T) {
75	74		}
76	75		}
77	76
78		-	func TestEngineRunStopsWithContext(t *testing.T) {
79		-	engine := NewEngine(nil)
	77	+	func TestEngineDispatchesLaterTaskWhileEarlierTaskBlocks(t *testing.T) {
	78	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	79	+	engine := NewEngine(func() time.Time { return now })
	80	+	first := testTask(t, "a", now)
	81	+	engine.tasks[first.ID] = first
	82	+
	83	+	started := make(chan struct{})
	84	+	release := make(chan struct{})
	85	+	secondDone := make(chan struct{})
	86	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
	87	+	switch task.ID {
	88	+	case "a":
	89	+	close(started)
	90	+	<-release
	91	+	case "b":
	92	+	close(secondDone)
	93	+	}
	94	+	return nil
	95	+	})
	96	+	<-started
	97	+	if _, ok := engine.nextRun(); ok {
	98	+	t.Fatal("nextRun() considered an in-flight task due")
	99	+	}
	100	+	second := testTask(t, "b", now)
	101	+	engine.mu.Lock()
	102	+	engine.tasks[second.ID] = second
	103	+	engine.mu.Unlock()
	104	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
	105	+	if task.ID == "b" {
	106	+	close(secondDone)
	107	+	}
	108	+	return nil
	109	+	})
	110	+	select {
	111	+	case <-secondDone:
	112	+	case <-time.After(time.Second):
	113	+	t.Fatal("later due task waited for earlier blocked task")
	114	+	}
	115	+	close(release)
	116	+	}
	117	+
	118	+	func TestEngineClaimsTaskOnceAcrossConcurrentDispatches(t *testing.T) {
	119	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	120	+	engine := NewEngine(func() time.Time { return now })
	121	+	engine.tasks["due"] = testTask(t, "due", now)
	122	+
	123	+	started := make(chan struct{})
	124	+	release := make(chan struct{})
	125	+	var callsMu sync.Mutex
	126	+	calls := 0
	127	+	executor := func(context.Context, Task) error {
	128	+	callsMu.Lock()
	129	+	calls++
	130	+	callsMu.Unlock()
	131	+	close(started)
	132	+	<-release
	133	+	return nil
	134	+	}
	135	+	var dispatched sync.WaitGroup
	136	+	for range 2 {
	137	+	dispatched.Add(1)
	138	+	go func() {
	139	+	defer dispatched.Done()
	140	+	engine.dispatchDue(context.Background(), now, executor)
	141	+	}()
	142	+	}
	143	+	dispatched.Wait()
	144	+	<-started
	145	+	callsMu.Lock()
	146	+	gotCalls := calls
	147	+	callsMu.Unlock()
	148	+	if gotCalls != 1 {
	149	+	t.Fatalf("executor calls = %d, want 1", gotCalls)
	150	+	}
	151	+	close(release)
	152	+	}
	153	+
	154	+	func TestEngineCancelQueuedTaskBeforeCallback(t *testing.T) {
	155	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	156	+	engine := NewEngine(func() time.Time { return now })
	157	+	first := testTask(t, "a", now)
	158	+	second := testTask(t, "b", now)
	159	+	engine.tasks[first.ID] = first
	160	+	engine.tasks[second.ID] = second
	161	+	due := engine.takeDue(now)
	162	+	if len(due) != 2 {
	163	+	t.Fatalf("takeDue() returned %d tasks, want 2", len(due))
	164	+	}
	165	+
	166	+	started := make(chan struct{})
	167	+	release := make(chan struct{})
	168	+	var wg sync.WaitGroup
	169	+	wg.Add(1)
	170	+	go func() {
	171	+	defer wg.Done()
	172	+	engine.dispatchTask(context.Background(), due[0], func(context.Context, Task) error {
	173	+	close(started)
	174	+	<-release
	175	+	return nil
	176	+	})
	177	+	}()
	178	+	<-started
	179	+	if !engine.Cancel(second.ID) {
	180	+	t.Fatal("Cancel(second) = false, want true")
	181	+	}
	182	+	calledSecond := false
	183	+	engine.dispatchTask(context.Background(), due[1], func(context.Context, Task) error {
	184	+	calledSecond = true
	185	+	return nil
	186	+	})
	187	+	close(release)
	188	+	wg.Wait()
	189	+	if calledSecond {
	190	+	t.Fatal("cancelled queued task invoked its executor")
	191	+	}
	192	+	}
	193	+
	194	+	func TestEngineRunSkipsDueTasksWhenContextCancelled(t *testing.T) {
	195	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	196	+	engine := NewEngine(func() time.Time { return now })
	197	+	engine.tasks["due"] = testTask(t, "due", now)
80	198		ctx, cancel := context.WithCancel(context.Background())
81	199		cancel()
82		-	if err := engine.Run(ctx, func(Task) error { return nil }); err != nil {
	200	+	called := false
	201	+	if err := engine.Run(ctx, func(context.Context, Task) error {
	202	+	called = true
	203	+	return nil
	204	+	}); err != nil {
83	205		t.Fatalf("Run() error = %v", err)
84	206		}
	207	+	if called {
	208	+	t.Fatal("Run() invoked executor after cancellation")
	209	+	}
	210	+	}
	211	+
	212	+	func testTask(t *testing.T, id string, now time.Time) Task {
	213	+	t.Helper()
	214	+	schedule, err := ParseCron("* * * * *", time.UTC)
	215	+	if err != nil {
	216	+	t.Fatal(err)
	217	+	}
	218	+	return Task{ID: id, SessionID: id, Cron: "* * * * *", Message: id, NextRun: now, schedule: schedule}
	219	+	}
	220	+
	221	+	func eventually(t *testing.T, condition func() bool) {
	222	+	t.Helper()
	223	+	deadline := time.Now().Add(time.Second)
	224	+	for time.Now().Before(deadline) {
	225	+	if condition() {
	226	+	return
	227	+	}
	228	+	time.Sleep(time.Millisecond)
	229	+	}
	230	+	t.Fatal("condition did not become true before deadline")
85	231		}


@@ -3,6 +3,7 @@ package scheduler
3		import (	3		import (
4		"context"	4		"context"
5		"errors"	5		"errors"
			6	+	"sync"
6		"testing"	7		"testing"
7		"time"	8		"time"
8		)	9		)

@@ -11,30 +12,31 @@ func TestEngineDispatchDueReschedulesTask(t *testing.T) {
11		loc := time.UTC	12		loc := time.UTC
12		now := time.Date(2026, time.August, 15, 12, 0, 0, 0, loc)	13		now := time.Date(2026, time.August, 15, 12, 0, 0, 0, loc)
13		engine := NewEngine(func() time.Time { return now })	14		engine := NewEngine(func() time.Time { return now })
14	-	created, err := engine.Add(NewTaskInput{	15	+	created, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "check status", Location: loc})
15	-	SessionID: "session-1",
16	-	Cron: "* * * * *",
17	-	Message: "check status",
18	-	Location: loc,
19	-	})
20		if err != nil {	16		if err != nil {
21		t.Fatalf("Add() error = %v", err)	17		t.Fatalf("Add() error = %v", err)
22		}	18		}
23			19
24	-	var executed []Task	20	+	done := make(chan Task, 1)
25		now = now.Add(time.Minute)	21		now = now.Add(time.Minute)
26	-	engine.dispatchDue(now, func(task Task) error {	22	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
27	-	executed = append(executed, task)	23	+	done <- task
28		return nil	24		return nil
29		})	25		})
30	-	if len(executed) != 1 \|\| executed[0].ID != created.ID {	26	+	select {
31	-	t.Fatalf("executed = %#v, want task %q", executed, created.ID)	27	+	case task := <-done:
			28	+	if task.ID != created.ID {
			29	+	t.Fatalf("executed task ID = %q, want %q", task.ID, created.ID)
			30	+	}
			31	+	case <-time.After(time.Second):
			32	+	t.Fatal("due task was not executed")
32		}	33		}
33			34
			35	+	eventually(t, func() bool {
			36	+	tasks := engine.List()
			37	+	return len(tasks) == 1 && tasks[0].RunCount == 1
			38	+	})
34		tasks := engine.List()	39		tasks := engine.List()
35	-	if len(tasks) != 1 {
36	-	t.Fatalf("List() length = %d, want 1", len(tasks))
37	-	}
38		if want := now.Add(time.Minute); !tasks[0].NextRun.Equal(want) {	40		if want := now.Add(time.Minute); !tasks[0].NextRun.Equal(want) {
39		t.Fatalf("NextRun = %v, want %v", tasks[0].NextRun, want)	41		t.Fatalf("NextRun = %v, want %v", tasks[0].NextRun, want)
40		}	42		}

@@ -51,14 +53,11 @@ func TestEngineRecordsExecutionFailure(t *testing.T) {
51		t.Fatalf("Add() error = %v", err)	53		t.Fatalf("Add() error = %v", err)
52		}	54		}
53			55
54	-	engine.dispatchDue(now.Add(time.Minute), func(Task) error { return errors.New("provider unavailable") })	56	+	engine.dispatchDue(context.Background(), now.Add(time.Minute), func(context.Context, Task) error { return errors.New("provider unavailable") })
55	-	got, ok := engine.Get(task.ID)	57	+	eventually(t, func() bool {
56	-	if !ok {	58	+	got, ok := engine.Get(task.ID)
57	-	t.Fatal("Get() did not return scheduled task")	59	+	return ok && got.LastError == "provider unavailable"
58	-	}	60	+	})
59	-	if got.LastError != "provider unavailable" {
60	-	t.Fatalf("LastError = %q", got.LastError)
61	-	}
62		}	61		}
63			62
64		func TestEngineCancel(t *testing.T) {	63		func TestEngineCancel(t *testing.T) {

@@ -75,11 +74,158 @@ func TestEngineCancel(t *testing.T) {
75		}	74		}
76		}	75		}
77			76
78	-	func TestEngineRunStopsWithContext(t *testing.T) {	77	+	func TestEngineDispatchesLaterTaskWhileEarlierTaskBlocks(t *testing.T) {
79	-	engine := NewEngine(nil)	78	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
			79	+	engine := NewEngine(func() time.Time { return now })
			80	+	first := testTask(t, "a", now)
			81	+	engine.tasks[first.ID] = first
			82	+
			83	+	started := make(chan struct{})
			84	+	release := make(chan struct{})
			85	+	secondDone := make(chan struct{})
			86	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
			87	+	switch task.ID {
			88	+	case "a":
			89	+	close(started)
			90	+	<-release
			91	+	case "b":
			92	+	close(secondDone)
			93	+	}
			94	+	return nil
			95	+	})
			96	+	<-started
			97	+	if _, ok := engine.nextRun(); ok {
			98	+	t.Fatal("nextRun() considered an in-flight task due")
			99	+	}
			100	+	second := testTask(t, "b", now)
			101	+	engine.mu.Lock()
			102	+	engine.tasks[second.ID] = second
			103	+	engine.mu.Unlock()
			104	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
			105	+	if task.ID == "b" {
			106	+	close(secondDone)
			107	+	}
			108	+	return nil
			109	+	})
			110	+	select {
			111	+	case <-secondDone:
			112	+	case <-time.After(time.Second):
			113	+	t.Fatal("later due task waited for earlier blocked task")
			114	+	}
			115	+	close(release)
			116	+	}
			117	+
			118	+	func TestEngineClaimsTaskOnceAcrossConcurrentDispatches(t *testing.T) {
			119	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
			120	+	engine := NewEngine(func() time.Time { return now })
			121	+	engine.tasks["due"] = testTask(t, "due", now)
			122	+
			123	+	started := make(chan struct{})
			124	+	release := make(chan struct{})
			125	+	var callsMu sync.Mutex
			126	+	calls := 0
			127	+	executor := func(context.Context, Task) error {
			128	+	callsMu.Lock()
			129	+	calls++
			130	+	callsMu.Unlock()
			131	+	close(started)
			132	+	<-release
			133	+	return nil
			134	+	}
			135	+	var dispatched sync.WaitGroup
			136	+	for range 2 {
			137	+	dispatched.Add(1)
			138	+	go func() {
			139	+	defer dispatched.Done()
			140	+	engine.dispatchDue(context.Background(), now, executor)
			141	+	}()
			142	+	}
			143	+	dispatched.Wait()
			144	+	<-started
			145	+	callsMu.Lock()
			146	+	gotCalls := calls
			147	+	callsMu.Unlock()
			148	+	if gotCalls != 1 {
			149	+	t.Fatalf("executor calls = %d, want 1", gotCalls)
			150	+	}
			151	+	close(release)
			152	+	}
			153	+
			154	+	func TestEngineCancelQueuedTaskBeforeCallback(t *testing.T) {
			155	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
			156	+	engine := NewEngine(func() time.Time { return now })
			157	+	first := testTask(t, "a", now)
			158	+	second := testTask(t, "b", now)
			159	+	engine.tasks[first.ID] = first
			160	+	engine.tasks[second.ID] = second
			161	+	due := engine.takeDue(now)
			162	+	if len(due) != 2 {
			163	+	t.Fatalf("takeDue() returned %d tasks, want 2", len(due))
			164	+	}
			165	+
			166	+	started := make(chan struct{})
			167	+	release := make(chan struct{})
			168	+	var wg sync.WaitGroup
			169	+	wg.Add(1)
			170	+	go func() {
			171	+	defer wg.Done()
			172	+	engine.dispatchTask(context.Background(), due[0], func(context.Context, Task) error {
			173	+	close(started)
			174	+	<-release
			175	+	return nil
			176	+	})
			177	+	}()
			178	+	<-started
			179	+	if !engine.Cancel(second.ID) {
			180	+	t.Fatal("Cancel(second) = false, want true")
			181	+	}
			182	+	calledSecond := false
			183	+	engine.dispatchTask(context.Background(), due[1], func(context.Context, Task) error {
			184	+	calledSecond = true
			185	+	return nil
			186	+	})
			187	+	close(release)
			188	+	wg.Wait()
			189	+	if calledSecond {
			190	+	t.Fatal("cancelled queued task invoked its executor")
			191	+	}
			192	+	}
			193	+
			194	+	func TestEngineRunSkipsDueTasksWhenContextCancelled(t *testing.T) {
			195	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
			196	+	engine := NewEngine(func() time.Time { return now })
			197	+	engine.tasks["due"] = testTask(t, "due", now)
80		ctx, cancel := context.WithCancel(context.Background())	198		ctx, cancel := context.WithCancel(context.Background())
81		cancel()	199		cancel()
82	-	if err := engine.Run(ctx, func(Task) error { return nil }); err != nil {	200	+	called := false
			201	+	if err := engine.Run(ctx, func(context.Context, Task) error {
			202	+	called = true
			203	+	return nil
			204	+	}); err != nil {
83		t.Fatalf("Run() error = %v", err)	205		t.Fatalf("Run() error = %v", err)
84		}	206		}
			207	+	if called {
			208	+	t.Fatal("Run() invoked executor after cancellation")
			209	+	}
			210	+	}
			211	+
			212	+	func testTask(t *testing.T, id string, now time.Time) Task {
			213	+	t.Helper()
			214	+	schedule, err := ParseCron("* * * * *", time.UTC)
			215	+	if err != nil {
			216	+	t.Fatal(err)
			217	+	}
			218	+	return Task{ID: id, SessionID: id, Cron: "* * * * *", Message: id, NextRun: now, schedule: schedule}
			219	+	}
			220	+
			221	+	func eventually(t *testing.T, condition func() bool) {
			222	+	t.Helper()
			223	+	deadline := time.Now().Add(time.Second)
			224	+	for time.Now().Before(deadline) {
			225	+	if condition() {
			226	+	return
			227	+	}
			228	+	time.Sleep(time.Millisecond)
			229	+	}
			230	+	t.Fatal("condition did not become true before deadline")
85		}	231		}

M packages/agent/scheduler_runtime.go +1 -2


@@ -3,7 +3,6 @@ package agent
3	3		import (
4	4		"context"
5	5		"fmt"
6		-	"path/filepath"
7	6		"strings"
8	7		"sync"
9	8

@@ -17,7 +16,7 @@ import (
17	16		// process and appends the resulting turn to its existing transcript. The
18	17		// caller serializes this with session transitions before invoking it.
19	18		func runScheduledSession(ctx context.Context, task scheduler.Task, args Args, base Resolved, extMgr *extensions.Manager, prepareRegistry func(core.Registry) core.Registry) error {
20		-	path, err := core.FindManagedSessionByID(ctx, filepath.Join(ZutHome(), "sessions"), task.SessionID)
	19	+	path, err := core.FindManagedSessionByID(ctx, ZutHome(), task.SessionID)
21	20		if err != nil {
22	21		return fmt.Errorf("find scheduled session %q: %w", task.SessionID, err)
23	22		}

 import (
 	"context"
 	"fmt"
-	"path/filepath"
 	"strings"
 	"sync"
 // process and appends the resulting turn to its existing transcript. The
 // caller serializes this with session transitions before invoking it.
 func runScheduledSession(ctx context.Context, task scheduler.Task, args Args, base Resolved, extMgr *extensions.Manager, prepareRegistry func(core.Registry) core.Registry) error {
-	path, err := core.FindManagedSessionByID(ctx, filepath.Join(ZutHome(), "sessions"), task.SessionID)
+	path, err := core.FindManagedSessionByID(ctx, ZutHome(), task.SessionID)
 	if err != nil {
 		return fmt.Errorf("find scheduled session %q: %w", task.SessionID, err)
 	}

M packages/core/session_portable_test.go +22 -0


@@ -37,6 +37,28 @@ func TestFindManagedSessionByIDStrictScopesAllManagedStoresAndReturnsReadErrors(
37	37		t.Fatalf("managed session path = %q, want %q", got, session.Path)
38	38		}
39	39
	40	+	agentRoot := filepath.Join(root, "sessions", "agents", "scheduled-worker")
	41	+	agentSession, err := NewSession(agentRoot, cwd, "provider", "model", "test")
	42	+	if err != nil {
	43	+	t.Fatal(err)
	44	+	}
	45	+	if err := agentSession.AppendMessage(provider.Message{
	46	+	Role: provider.RoleUser,
	47	+	Content: []provider.Content{provider.TextBlock{Text: "agent managed"}},
	48	+	}); err != nil {
	49	+	t.Fatal(err)
	50	+	}
	51	+	if err := agentSession.Close(); err != nil {
	52	+	t.Fatal(err)
	53	+	}
	54	+	got, err = FindManagedSessionByID(context.Background(), root, agentSession.ID)
	55	+	if err != nil {
	56	+	t.Fatalf("FindManagedSessionByID(agent session): %v", err)
	57	+	}
	58	+	if got != agentSession.Path {
	59	+	t.Fatalf("agent managed session path = %q, want %q", got, agentSession.Path)
	60	+	}
	61	+
40	62		badPath := filepath.Join(SessionsDir(root, cwd), "bad.jsonl")
41	63		if err := os.WriteFile(badPath, []byte("not json\n"), 0o644); err != nil {
42	64		t.Fatal(err)


@@ -37,6 +37,28 @@ func TestFindManagedSessionByIDStrictScopesAllManagedStoresAndReturnsReadErrors(
37	t.Fatalf("managed session path = %q, want %q", got, session.Path)	37		t.Fatalf("managed session path = %q, want %q", got, session.Path)
38	}	38		}
39		39
		40	+	agentRoot := filepath.Join(root, "sessions", "agents", "scheduled-worker")
		41	+	agentSession, err := NewSession(agentRoot, cwd, "provider", "model", "test")
		42	+	if err != nil {
		43	+	t.Fatal(err)
		44	+	}
		45	+	if err := agentSession.AppendMessage(provider.Message{
		46	+	Role: provider.RoleUser,
		47	+	Content: []provider.Content{provider.TextBlock{Text: "agent managed"}},
		48	+	}); err != nil {
		49	+	t.Fatal(err)
		50	+	}
		51	+	if err := agentSession.Close(); err != nil {
		52	+	t.Fatal(err)
		53	+	}
		54	+	got, err = FindManagedSessionByID(context.Background(), root, agentSession.ID)
		55	+	if err != nil {
		56	+	t.Fatalf("FindManagedSessionByID(agent session): %v", err)
		57	+	}
		58	+	if got != agentSession.Path {
		59	+	t.Fatalf("agent managed session path = %q, want %q", got, agentSession.Path)
		60	+	}
		61	+
40	badPath := filepath.Join(SessionsDir(root, cwd), "bad.jsonl")	62		badPath := filepath.Join(SessionsDir(root, cwd), "bad.jsonl")
41	if err := os.WriteFile(badPath, []byte("not json\n"), 0o644); err != nil {	63		if err := os.WriteFile(badPath, []byte("not json\n"), 0o644); err != nil {
42	t.Fatal(err)	64		t.Fatal(err)