Merge pull request #138 from bnema/feat/in-process-cron-scheduler

feat(agent): add in-process cron scheduler

commit: 29be61caae3a3b596c357d4ef62ab93b2441bb2e
author: bnema · 2mo ago · 2026-08-17 05:26 UTC
committer: GitHub · 2mo ago · 2026-08-17 05:26 UTC
merge of: d7aa01a + cc1e45a · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown
refs: tag: v0.20.0

13 files · +1266 -10

M README.md +2 -1


@@ -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		-	- nine core built-in tools (read, write, edit, bash, create_worktree, grep, lsp, web_search, and update_goal); 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,6 +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 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.
258	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	261

 - one static binary.
 - 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.
-- nine core built-in tools (read, write, edit, bash, create_worktree, grep, lsp, web_search, and update_goal); 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.
+- 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.
 - three run modes (interactive tui, print, json).
 - built-in telegram bot.
 - 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).
 - `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).
 - `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).
 - `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`.
+- `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.
 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.

M packages/agent/cli.go +88 -5


@@ -19,6 +19,7 @@ import (
19	19		"github.com/bnema/zut/packages/agent/extensions"
20	20		"github.com/bnema/zut/packages/agent/extproto"
21	21		"github.com/bnema/zut/packages/agent/modes"
	22	+	"github.com/bnema/zut/packages/agent/scheduler"
22	23		"github.com/bnema/zut/packages/agent/skills"
23	24		"github.com/bnema/zut/packages/agent/subagents"
24	25		"github.com/bnema/zut/packages/agent/tools"

@@ -1484,6 +1485,21 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1484	1485		var sessionTransitionMu sync.RWMutex
1485	1486		activeProvider := r.Provider
1486	1487		activeModel := r.Model
	1488	+	var sess *core.Session
	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	+	}
1487	1503
1488	1504		// Build the extension manager BEFORE the agent so we can fold
1489	1505		// extension-defined tools into the registry. Attach the interactive

@@ -1665,6 +1681,30 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1665	1681		return a
1666	1682		}
1667	1683
	1684	+	scheduleTool := &tools.ScheduleTool{
	1685	+	Engine: scheduleEngine,
	1686	+	SessionID: func() string {
	1687	+	persistMu.Lock()
	1688	+	defer persistMu.Unlock()
	1689	+	if sess == nil {
	1690	+	return ""
	1691	+	}
	1692	+	return sess.ID
	1693	+	},
	1694	+	Location: func() *time.Location { return time.Local },
	1695	+	}
	1696	+	schedulerAllowed := !args.NoTools && (!args.ToolsSet && len(args.Tools) == 0 \|\| toolListContains(args.Tools, tools.ScheduleToolName))
	1697	+	installSchedulerTool := func(registry core.Registry) core.Registry {
	1698	+	if schedulerAllowed {
	1699	+	registry[tools.ScheduleToolName] = scheduleTool
	1700	+	}
	1701	+	return registry
	1702	+	}
	1703	+	installSchedulerTool(r.ToolRegistry)
	1704	+	prepareInteractiveRegistryWithScheduler := func(registry core.Registry) core.Registry {
	1705	+	return prepareInteractiveRegistry(installSchedulerTool(registry))
	1706	+	}
	1707	+
1668	1708		var ag *core.Agent
1669	1709		buildAgent := func() (*core.Agent, string, string, error) {
1670	1710		resolved, err := Resolve(args, true)

@@ -1676,6 +1716,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1676	1716		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)
1677	1717		runtime.SetFastMode(resolved.FastMode)
1678	1718		resolved.MergeExtensionTools(extToolAdapter)
	1719	+	installSchedulerTool(resolved.ToolRegistry)
1679	1720		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)
1680	1721		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil
1681	1722		}

@@ -1698,6 +1739,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1698	1739		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)
1699	1740		runtime.SetFastMode(resolved.FastMode)
1700	1741		resolved.MergeExtensionTools(extToolAdapter)
	1742	+	installSchedulerTool(resolved.ToolRegistry)
1701	1743		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)
1702	1744		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil
1703	1745		}

@@ -1728,6 +1770,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1728	1770		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)
1729	1771		runtime.SetFastMode(resolved.FastMode)
1730	1772		resolved.MergeExtensionTools(extToolAdapter)
	1773	+	installSchedulerTool(resolved.ToolRegistry)
1731	1774		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)
1732	1775		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil
1733	1776		}

@@ -1747,12 +1790,11 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1747	1790		if current == nil {
1748	1791		return
1749	1792		}
1750		-	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv); err != nil && iv != nil {
	1793	+	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv); err != nil && iv != nil {
1751	1794		iv.ReportError(err)
1752	1795		}
1753	1796		})
1754	1797
1755		-	var sess *core.Session
1756	1798		var sessBaselineMsgs int // messages already on disk when current session opened
1757	1799		var sessionTitlePending bool
1758	1800		if !args.NoSess && ag != nil {

@@ -2476,14 +2518,14 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2476	2518		_ = extMgr.Reload(context.Background(), 2*time.Second)
2477	2519		}
2478	2520		if current != nil {
2479		-	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv); err != nil {
	2521	+	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv); err != nil {
2480	2522		iv.ReportError(err)
2481	2523		}
2482	2524		}
2483	2525		},
2484	2526		RefreshTools: func() error {
2485	2527		current := liveInteractiveAgent(iv, ag)
2486		-	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv)
	2528	+	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv)
2487	2529		if err != nil {
2488	2530		if !errors.Is(err, errInteractiveAgentChanged) {
2489	2531		setWebSearchAvailable(false)

@@ -2496,7 +2538,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2496	2538		SetWebSearchAvailable: setWebSearchAvailable,
2497	2539		RefreshPrompt: func() error {
2498	2540		current := liveInteractiveAgent(iv, ag)
2499		-	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv)
	2541	+	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv)
2500	2542		if err != nil {
2501	2543		return err
2502	2544		}

@@ -2682,6 +2724,47 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2682	2724		},
2683	2725		})
2684	2726		extHooks.attachInteractive(iv)
	2727	+	schedulerCtx, cancelScheduler := context.WithCancel(ctx)
	2728	+	defer cancelScheduler()
	2729	+	go func() {
	2730	+	if err := scheduleEngine.Run(schedulerCtx, func(taskCtx context.Context, task scheduler.Task) error {
	2731	+	unlockSession := lockScheduledSession(task.SessionID)
	2732	+	defer unlockSession()
	2733	+	// Keep a background session run mutually exclusive with /session
	2734	+	// transitions. The active target uses Interactive's distinct-turn
	2735	+	// ingress instead, so it never injects into an in-flight model loop.
	2736	+	sessionTransitionMu.RLock()
	2737	+	defer sessionTransitionMu.RUnlock()
	2738	+	persistMu.Lock()
	2739	+	activeTarget := sess != nil && sess.ID == task.SessionID
	2740	+	persistMu.Unlock()
	2741	+	if activeTarget {
	2742	+	return iv.SubmitFollowUp(taskCtx, task.Message)
	2743	+	}
	2744	+	// A resumed background agent must schedule any follow-up work against
	2745	+	// the transcript it resumed, not whichever session is visible in the
	2746	+	// interactive TUI while it runs.
	2747	+	backgroundRegistry := func(registry core.Registry) core.Registry {
	2748	+	if schedulerAllowed {
	2749	+	registry[tools.ScheduleToolName] = &tools.ScheduleTool{
	2750	+	Engine: scheduleEngine,
	2751	+	SessionID: func() string { return task.SessionID },
	2752	+	Location: func() *time.Location { return time.Local },
	2753	+	}
	2754	+	}
	2755	+	return registry
	2756	+	}
	2757	+	err := runScheduledSession(taskCtx, task, args, r, extMgr, backgroundRegistry)
	2758	+	if err != nil {
	2759	+	iv.Notify("scheduler", "error", fmt.Sprintf("scheduled task %s failed: %v", task.ID, err))
	2760	+	} else {
	2761	+	iv.Notify("scheduler", "success", fmt.Sprintf("scheduled task %s completed in session %s", task.ID, task.SessionID))
	2762	+	}
	2763	+	return err
	2764	+	}); err != nil && schedulerCtx.Err() == nil {
	2765	+	iv.ReportError(fmt.Errorf("scheduler stopped: %w", err))
	2766	+	}
	2767	+	}()
2685	2768		go func() {
2686	2769		<-reloadDone
2687	2770		for _, reloadErr := range reloadErrs {


@@ -19,6 +19,7 @@ import (
19		"github.com/bnema/zut/packages/agent/extensions"	19		"github.com/bnema/zut/packages/agent/extensions"
20		"github.com/bnema/zut/packages/agent/extproto"	20		"github.com/bnema/zut/packages/agent/extproto"
21		"github.com/bnema/zut/packages/agent/modes"	21		"github.com/bnema/zut/packages/agent/modes"
			22	+	"github.com/bnema/zut/packages/agent/scheduler"
22		"github.com/bnema/zut/packages/agent/skills"	23		"github.com/bnema/zut/packages/agent/skills"
23		"github.com/bnema/zut/packages/agent/subagents"	24		"github.com/bnema/zut/packages/agent/subagents"
24		"github.com/bnema/zut/packages/agent/tools"	25		"github.com/bnema/zut/packages/agent/tools"

@@ -1484,6 +1485,21 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1484		var sessionTransitionMu sync.RWMutex	1485		var sessionTransitionMu sync.RWMutex
1485		activeProvider := r.Provider	1486		activeProvider := r.Provider
1486		activeModel := r.Model	1487		activeModel := r.Model
			1488	+	var sess *core.Session
			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	+	}
1487			1503
1488		// Build the extension manager BEFORE the agent so we can fold	1504		// Build the extension manager BEFORE the agent so we can fold
1489		// extension-defined tools into the registry. Attach the interactive	1505		// extension-defined tools into the registry. Attach the interactive

@@ -1665,6 +1681,30 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1665		return a	1681		return a
1666		}	1682		}
1667			1683
			1684	+	scheduleTool := &tools.ScheduleTool{
			1685	+	Engine: scheduleEngine,
			1686	+	SessionID: func() string {
			1687	+	persistMu.Lock()
			1688	+	defer persistMu.Unlock()
			1689	+	if sess == nil {
			1690	+	return ""
			1691	+	}
			1692	+	return sess.ID
			1693	+	},
			1694	+	Location: func() *time.Location { return time.Local },
			1695	+	}
			1696	+	schedulerAllowed := !args.NoTools && (!args.ToolsSet && len(args.Tools) == 0 \|\| toolListContains(args.Tools, tools.ScheduleToolName))
			1697	+	installSchedulerTool := func(registry core.Registry) core.Registry {
			1698	+	if schedulerAllowed {
			1699	+	registry[tools.ScheduleToolName] = scheduleTool
			1700	+	}
			1701	+	return registry
			1702	+	}
			1703	+	installSchedulerTool(r.ToolRegistry)
			1704	+	prepareInteractiveRegistryWithScheduler := func(registry core.Registry) core.Registry {
			1705	+	return prepareInteractiveRegistry(installSchedulerTool(registry))
			1706	+	}
			1707	+
1668		var ag *core.Agent	1708		var ag *core.Agent
1669		buildAgent := func() (*core.Agent, string, string, error) {	1709		buildAgent := func() (*core.Agent, string, string, error) {
1670		resolved, err := Resolve(args, true)	1710		resolved, err := Resolve(args, true)

@@ -1676,6 +1716,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1676		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)	1716		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)
1677		runtime.SetFastMode(resolved.FastMode)	1717		runtime.SetFastMode(resolved.FastMode)
1678		resolved.MergeExtensionTools(extToolAdapter)	1718		resolved.MergeExtensionTools(extToolAdapter)
			1719	+	installSchedulerTool(resolved.ToolRegistry)
1679		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)	1720		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)
1680		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil	1721		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil
1681		}	1722		}

@@ -1698,6 +1739,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1698		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)	1739		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)
1699		runtime.SetFastMode(resolved.FastMode)	1740		runtime.SetFastMode(resolved.FastMode)
1700		resolved.MergeExtensionTools(extToolAdapter)	1741		resolved.MergeExtensionTools(extToolAdapter)
			1742	+	installSchedulerTool(resolved.ToolRegistry)
1701		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)	1743		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)
1702		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil	1744		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil
1703		}	1745		}

@@ -1728,6 +1770,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1728		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)	1770		runtime.SetProviderSettings(resolved.BaseURL, resolved.InsecureTLS)
1729		runtime.SetFastMode(resolved.FastMode)	1771		runtime.SetFastMode(resolved.FastMode)
1730		resolved.MergeExtensionTools(extToolAdapter)	1772		resolved.MergeExtensionTools(extToolAdapter)
			1773	+	installSchedulerTool(resolved.ToolRegistry)
1731		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)	1774		prepareResolvedInteractiveRegistry(resolved.ToolRegistry, resolved.WebSearchPolicy)
1732		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil	1775		return wireAgentExt(resolved.NewAgent()), resolved.Provider, resolved.Model, nil
1733		}	1776		}

@@ -1747,12 +1790,11 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1747		if current == nil {	1790		if current == nil {
1748		return	1791		return
1749		}	1792		}
1750	-	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv); err != nil && iv != nil {	1793	+	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv); err != nil && iv != nil {
1751		iv.ReportError(err)	1794		iv.ReportError(err)
1752		}	1795		}
1753		})	1796		})
1754			1797
1755	-	var sess *core.Session
1756		var sessBaselineMsgs int // messages already on disk when current session opened	1798		var sessBaselineMsgs int // messages already on disk when current session opened
1757		var sessionTitlePending bool	1799		var sessionTitlePending bool
1758		if !args.NoSess && ag != nil {	1800		if !args.NoSess && ag != nil {

@@ -2476,14 +2518,14 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2476		_ = extMgr.Reload(context.Background(), 2*time.Second)	2518		_ = extMgr.Reload(context.Background(), 2*time.Second)
2477		}	2519		}
2478		if current != nil {	2520		if current != nil {
2479	-	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv); err != nil {	2521	+	if _, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv); err != nil {
2480		iv.ReportError(err)	2522		iv.ReportError(err)
2481		}	2523		}
2482		}	2524		}
2483		},	2525		},
2484		RefreshTools: func() error {	2526		RefreshTools: func() error {
2485		current := liveInteractiveAgent(iv, ag)	2527		current := liveInteractiveAgent(iv, ag)
2486	-	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv)	2528	+	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv)
2487		if err != nil {	2529		if err != nil {
2488		if !errors.Is(err, errInteractiveAgentChanged) {	2530		if !errors.Is(err, errInteractiveAgentChanged) {
2489		setWebSearchAvailable(false)	2531		setWebSearchAvailable(false)

@@ -2496,7 +2538,7 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2496		SetWebSearchAvailable: setWebSearchAvailable,	2538		SetWebSearchAvailable: setWebSearchAvailable,
2497		RefreshPrompt: func() error {	2539		RefreshPrompt: func() error {
2498		current := liveInteractiveAgent(iv, ag)	2540		current := liveInteractiveAgent(iv, ag)
2499	-	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistry, iv)	2541	+	webSearchPolicy, err := refreshAgentToolsAndPrompt(args, sharedSandbox, extToolAdapter, current, prepareInteractiveRegistryWithScheduler, iv)
2500		if err != nil {	2542		if err != nil {
2501		return err	2543		return err
2502		}	2544		}

@@ -2682,6 +2724,47 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2682		},	2724		},
2683		})	2725		})
2684		extHooks.attachInteractive(iv)	2726		extHooks.attachInteractive(iv)
			2727	+	schedulerCtx, cancelScheduler := context.WithCancel(ctx)
			2728	+	defer cancelScheduler()
			2729	+	go func() {
			2730	+	if err := scheduleEngine.Run(schedulerCtx, func(taskCtx context.Context, task scheduler.Task) error {
			2731	+	unlockSession := lockScheduledSession(task.SessionID)
			2732	+	defer unlockSession()
			2733	+	// Keep a background session run mutually exclusive with /session
			2734	+	// transitions. The active target uses Interactive's distinct-turn
			2735	+	// ingress instead, so it never injects into an in-flight model loop.
			2736	+	sessionTransitionMu.RLock()
			2737	+	defer sessionTransitionMu.RUnlock()
			2738	+	persistMu.Lock()
			2739	+	activeTarget := sess != nil && sess.ID == task.SessionID
			2740	+	persistMu.Unlock()
			2741	+	if activeTarget {
			2742	+	return iv.SubmitFollowUp(taskCtx, task.Message)
			2743	+	}
			2744	+	// A resumed background agent must schedule any follow-up work against
			2745	+	// the transcript it resumed, not whichever session is visible in the
			2746	+	// interactive TUI while it runs.
			2747	+	backgroundRegistry := func(registry core.Registry) core.Registry {
			2748	+	if schedulerAllowed {
			2749	+	registry[tools.ScheduleToolName] = &tools.ScheduleTool{
			2750	+	Engine: scheduleEngine,
			2751	+	SessionID: func() string { return task.SessionID },
			2752	+	Location: func() *time.Location { return time.Local },
			2753	+	}
			2754	+	}
			2755	+	return registry
			2756	+	}
			2757	+	err := runScheduledSession(taskCtx, task, args, r, extMgr, backgroundRegistry)
			2758	+	if err != nil {
			2759	+	iv.Notify("scheduler", "error", fmt.Sprintf("scheduled task %s failed: %v", task.ID, err))
			2760	+	} else {
			2761	+	iv.Notify("scheduler", "success", fmt.Sprintf("scheduled task %s completed in session %s", task.ID, task.SessionID))
			2762	+	}
			2763	+	return err
			2764	+	}); err != nil && schedulerCtx.Err() == nil {
			2765	+	iv.ReportError(fmt.Errorf("scheduler stopped: %w", err))
			2766	+	}
			2767	+	}()
2685		go func() {	2768		go func() {
2686		<-reloadDone	2769		<-reloadDone
2687		for _, reloadErr := range reloadErrs {	2770		for _, reloadErr := range reloadErrs {

M packages/agent/modes/interactive.go +10 -0


@@ -612,6 +612,11 @@ func encodeCompactHandoff(state compactContinuationState) json.RawMessage {
612	612		return encoded
613	613		}
614	614
	615	+	type scheduledFollowUp struct {
	616	+	text string
	617	+	accepted chan error
	618	+	}
	619	+
615	620		type Interactive struct {
616	621		cfg InteractiveConfig
617	622		view *tui.View

@@ -728,6 +733,11 @@ type Interactive struct {
728	733		// above the status bar as "sliding in: ..." chips.
729	734		queued []string
730	735
	736	+	// scheduled holds prompt-only follow-ups injected by the in-process
	737	+	// scheduler. Unlike agent.QueueMessage, these always wait until the
	738	+	// current turn has completed before starting their own turn.
	739	+	scheduled []scheduledFollowUp
	740	+
731	741		// runCtx is the top-level context passed to Run(). Follow-up turns
732	742		// drained from `queued` are started against this context so they
733	743		// survive past the ctx of the key event that enqueued them.


@@ -612,6 +612,11 @@ func encodeCompactHandoff(state compactContinuationState) json.RawMessage {
612	return encoded	612		return encoded
613	}	613		}
614		614
		615	+	type scheduledFollowUp struct {
		616	+	text string
		617	+	accepted chan error
		618	+	}
		619	+
615	type Interactive struct {	620		type Interactive struct {
616	cfg InteractiveConfig	621		cfg InteractiveConfig
617	view *tui.View	622		view *tui.View

@@ -728,6 +733,11 @@ type Interactive struct {
728	// above the status bar as "sliding in: ..." chips.	733		// above the status bar as "sliding in: ..." chips.
729	queued []string	734		queued []string
730		735
		736	+	// scheduled holds prompt-only follow-ups injected by the in-process
		737	+	// scheduler. Unlike agent.QueueMessage, these always wait until the
		738	+	// current turn has completed before starting their own turn.
		739	+	scheduled []scheduledFollowUp
		740	+
731	// runCtx is the top-level context passed to Run(). Follow-up turns	741		// runCtx is the top-level context passed to Run(). Follow-up turns
732	// drained from `queued` are started against this context so they	742		// drained from `queued` are started against this context so they
733	// survive past the ctx of the key event that enqueued them.	743		// survive past the ctx of the key event that enqueued them.

M packages/agent/modes/interactive_host.go +30 -0


@@ -201,6 +201,36 @@ func (i *Interactive) SubmitSlash(text string) {
201	201		func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
202	202		i.submitOrQueue(text, images, true)
203	203		}
	204	+
	205	+	// SubmitFollowUp submits a scheduler-owned prompt without interpreting shell
	206	+	// escapes or steering an active agent loop. A busy interactive waits for its
	207	+	// current turn to end, then starts this prompt as a distinct follow-up turn.
	208	+	func (i *Interactive) SubmitFollowUp(ctx context.Context, text string) error {
	209	+	text = strings.TrimSpace(text)
	210	+	if text == "" {
	211	+	return fmt.Errorf("scheduled prompt is empty")
	212	+	}
	213	+	i.mu.Lock()
	214	+	if i.agent == nil {
	215	+	i.mu.Unlock()
	216	+	return fmt.Errorf("no agent running; log in first")
	217	+	}
	218	+	if i.busy {
	219	+	accepted := make(chan error, 1)
	220	+	i.scheduled = append(i.scheduled, scheduledFollowUp{text: text, accepted: accepted})
	221	+	i.mu.Unlock()
	222	+	i.invalidate()
	223	+	select {
	224	+	case err := <-accepted:
	225	+	return err
	226	+	case <-ctx.Done():
	227	+	return ctx.Err()
	228	+	}
	229	+	}
	230	+	i.mu.Unlock()
	231	+	i.startTurn(i.runCtx, text)
	232	+	return nil
	233	+	}
204	234		func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {
205	235		if cmd, ok := shellEscapeCommand(text); ok {
206	236		i.startShellEscape(i.runCtx, cmd)


@@ -201,6 +201,36 @@ func (i *Interactive) SubmitSlash(text string) {
201	func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {	201		func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
202	i.submitOrQueue(text, images, true)	202		i.submitOrQueue(text, images, true)
203	}	203		}
		204	+
		205	+	// SubmitFollowUp submits a scheduler-owned prompt without interpreting shell
		206	+	// escapes or steering an active agent loop. A busy interactive waits for its
		207	+	// current turn to end, then starts this prompt as a distinct follow-up turn.
		208	+	func (i *Interactive) SubmitFollowUp(ctx context.Context, text string) error {
		209	+	text = strings.TrimSpace(text)
		210	+	if text == "" {
		211	+	return fmt.Errorf("scheduled prompt is empty")
		212	+	}
		213	+	i.mu.Lock()
		214	+	if i.agent == nil {
		215	+	i.mu.Unlock()
		216	+	return fmt.Errorf("no agent running; log in first")
		217	+	}
		218	+	if i.busy {
		219	+	accepted := make(chan error, 1)
		220	+	i.scheduled = append(i.scheduled, scheduledFollowUp{text: text, accepted: accepted})
		221	+	i.mu.Unlock()
		222	+	i.invalidate()
		223	+	select {
		224	+	case err := <-accepted:
		225	+	return err
		226	+	case <-ctx.Done():
		227	+	return ctx.Err()
		228	+	}
		229	+	}
		230	+	i.mu.Unlock()
		231	+	i.startTurn(i.runCtx, text)
		232	+	return nil
		233	+	}
204	func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {	234		func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, userInput bool) {
205	if cmd, ok := shellEscapeCommand(text); ok {	235		if cmd, ok := shellEscapeCommand(text); ok {
206	i.startShellEscape(i.runCtx, cmd)	236		i.startShellEscape(i.runCtx, cmd)

M packages/agent/modes/interactive_turn.go +18 -4


@@ -337,14 +337,22 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
337	337		next, i.queued = i.queued[0], i.queued[1:]
338	338		hasNext = true
339	339		}
340		-	// If the turn was cancelled or errored, drop the queue so the
	340	+	// If the turn was cancelled or errored, drop ordinary input so the
341	341		// user isn't bombarded with stale messages after an interrupt.
	342	+	// Scheduler follow-ups deliberately remain: they are distinct turns
	343	+	// whose due time must not let them steer this failed turn.
342	344		if ctx.Err() != nil \|\| (err != nil && !recoverContextOverflow) {
343	345		i.queued = nil
344	346		if i.agent != nil {
345	347		i.agent.DrainQueuedMessages()
346	348		}
347	349		}
	350	+	var scheduled scheduledFollowUp
	351	+	var hasScheduled bool
	352	+	if !awaitingPre && !hasNext && len(i.scheduled) > 0 {
	353	+	scheduled, i.scheduled = i.scheduled[0], i.scheduled[1:]
	354	+	hasScheduled = true
	355	+	}
348	356		// Decide whether the next thing to do is an auto-compaction.
349	357		// Only fires when the turn completed cleanly AND no host-side
350	358		// or agent-side queued messages are waiting (otherwise a queued

@@ -371,7 +379,7 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
371	379		handoff, persistHandoff = i.resetCompactContinuationLocked()
372	380		}
373	381		}
374		-	if !continueStatusRescue && (i.agent == nil \|\| ctx.Err() != nil \|\| err != nil \|\| awaitingPre \|\| hasNext \|\| continueQueued \|\| offer \|\| recoverContextOverflow \|\| (!shouldAutoCompact && !statusRescueActive)) {
	382	+	if !continueStatusRescue && (i.agent == nil \|\| ctx.Err() != nil \|\| err != nil \|\| awaitingPre \|\| hasNext \|\| hasScheduled \|\| continueQueued \|\| offer \|\| recoverContextOverflow \|\| (!shouldAutoCompact && !statusRescueActive)) {
375	383		handoff, persistHandoff = i.resetCompactContinuationLocked()
376	384		}
377	385		// The agent run can finish before the paced final text reaches the

@@ -381,8 +389,8 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
381	389		if recoverContextOverflow \|\| shouldAutoCompact {
382	390		i.resetStreamingStateLocked()
383	391		}
384		-	alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext \|\| agentQueued > 0 \|\| continueGoal, offer, recoverContextOverflow, shouldAutoCompact)
385		-	i.busy = hasNext \|\| continueQueued \|\| continueStatusRescue \|\| recoverContextOverflow \|\| shouldAutoCompact
	392	+	alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext \|\| hasScheduled \|\| agentQueued > 0 \|\| continueGoal, offer, recoverContextOverflow, shouldAutoCompact)
	393	+	i.busy = hasNext \|\| hasScheduled \|\| continueQueued \|\| continueStatusRescue \|\| recoverContextOverflow \|\| shouldAutoCompact
386	394		i.mu.Unlock()
387	395		if persistHandoff {
388	396		i.persistCompactHandoff(handoff)

@@ -402,6 +410,12 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
402	410		switch {
403	411		case hasNext:
404	412		i.startTurn(parent, next)
	413	+	case hasScheduled:
	414	+	// The scheduler holds the session transition read lock until this
	415	+	// acknowledgement, so accepting here binds the prompt to the agent
	416	+	// that just completed rather than a subsequently loaded session.
	417	+	scheduled.accepted <- nil
	418	+	i.startTurn(parent, scheduled.text)
405	419		case continueQueued:
406	420		i.startTurnRequest(parent, "", nil, true, false)
407	421		case continueStatusRescue:

 			next, i.queued = i.queued[0], i.queued[1:]
 			hasNext = true
 		}
-		// If the turn was cancelled or errored, drop the queue so the
+		// If the turn was cancelled or errored, drop ordinary input so the
 		// user isn't bombarded with stale messages after an interrupt.
+		// Scheduler follow-ups deliberately remain: they are distinct turns
+		// whose due time must not let them steer this failed turn.
 		if ctx.Err() != nil || (err != nil && !recoverContextOverflow) {
 			i.queued = nil
 			if i.agent != nil {
 				i.agent.DrainQueuedMessages()
 			}
 		}
+		var scheduled scheduledFollowUp
+		var hasScheduled bool
+		if !awaitingPre && !hasNext && len(i.scheduled) > 0 {
+			scheduled, i.scheduled = i.scheduled[0], i.scheduled[1:]
+			hasScheduled = true
+		}
 		// Decide whether the next thing to do is an auto-compaction.
 		// Only fires when the turn completed cleanly AND no host-side
 		// or agent-side queued messages are waiting (otherwise a queued
 				handoff, persistHandoff = i.resetCompactContinuationLocked()
 			}
 		}
-		if !continueStatusRescue && (i.agent == nil || ctx.Err() != nil || err != nil || awaitingPre || hasNext || continueQueued || offer || recoverContextOverflow || (!shouldAutoCompact && !statusRescueActive)) {
+		if !continueStatusRescue && (i.agent == nil || ctx.Err() != nil || err != nil || awaitingPre || hasNext || hasScheduled || continueQueued || offer || recoverContextOverflow || (!shouldAutoCompact && !statusRescueActive)) {
 			handoff, persistHandoff = i.resetCompactContinuationLocked()
 		}
 		// The agent run can finish before the paced final text reaches the
 		if recoverContextOverflow || shouldAutoCompact {
 			i.resetStreamingStateLocked()
 		}
-		alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext || agentQueued > 0 || continueGoal, offer, recoverContextOverflow, shouldAutoCompact)
+		alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext || hasScheduled || agentQueued > 0 || continueGoal, offer, recoverContextOverflow, shouldAutoCompact)
-		i.busy = hasNext || continueQueued || continueStatusRescue || recoverContextOverflow || shouldAutoCompact
+		i.busy = hasNext || hasScheduled || continueQueued || continueStatusRescue || recoverContextOverflow || shouldAutoCompact
 		i.mu.Unlock()
 		if persistHandoff {
 			i.persistCompactHandoff(handoff)
 		switch {
 		case hasNext:
 			i.startTurn(parent, next)
+		case hasScheduled:
+			// The scheduler holds the session transition read lock until this
+			// acknowledgement, so accepting here binds the prompt to the agent
+			// that just completed rather than a subsequently loaded session.
+			scheduled.accepted <- nil
+			i.startTurn(parent, scheduled.text)
 		case continueQueued:
 			i.startTurnRequest(parent, "", nil, true, false)
 		case continueStatusRescue:

A packages/agent/scheduler/cron.go +208 -0

@@ -0,0 +1,208 @@
	1	+	// Package scheduler provides portable, in-process calendar scheduling.
	2	+	package scheduler
	3	+
	4	+	import (
	5	+	"fmt"
	6	+	"strconv"
	7	+	"strings"
	8	+	"time"
	9	+	)
	10	+
	11	+	// Cron is a five-field cron schedule (minute, hour, day of month, month, day
	12	+	// of week). It evaluates calendar times in the location captured when the
	13	+	// schedule is created. A repeated local minute during a DST fall-back matches
	14	+	// twice; a local minute skipped during a spring-forward does not match.
	15	+	type Cron struct {
	16	+	minute field
	17	+	hour field
	18	+	day field
	19	+	month field
	20	+	week field
	21	+	loc *time.Location
	22	+	}
	23	+
	24	+	type field struct {
	25	+	values []bool
	26	+	wildcard bool
	27	+	}
	28	+
	29	+	var cronMacros = map[string]string{
	30	+	"@yearly": "0 0 1 1 *",
	31	+	"@annually": "0 0 1 1 *",
	32	+	"@monthly": "0 0 1 * *",
	33	+	"@weekly": "0 0 * * 0",
	34	+	"@daily": "0 0 * * *",
	35	+	"@midnight": "0 0 * * *",
	36	+	"@hourly": "0 * * * *",
	37	+	}
	38	+
	39	+	var monthNames = map[string]int{
	40	+	"jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6,
	41	+	"jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12,
	42	+	}
	43	+
	44	+	var weekdayNames = map[string]int{
	45	+	"sun": 0, "mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6,
	46	+	}
	47	+
	48	+	// ParseCron parses a conventional five-field cron expression. Month and
	49	+	// weekday names use the English three-letter abbreviations. The supplied
	50	+	// location is retained by the schedule, so later changes to time.Local do not
	51	+	// change an existing task's calendar semantics.
	52	+	func ParseCron(expression string, location *time.Location) (Cron, error) {
	53	+	if location == nil {
	54	+	return Cron{}, fmt.Errorf("cron timezone is required")
	55	+	}
	56	+	expression = strings.ToLower(strings.TrimSpace(expression))
	57	+	if replacement, ok := cronMacros[expression]; ok {
	58	+	expression = replacement
	59	+	}
	60	+	parts := strings.Fields(expression)
	61	+	if len(parts) != 5 {
	62	+	return Cron{}, fmt.Errorf("cron requires five fields (minute hour day-of-month month day-of-week)")
	63	+	}
	64	+	minute, err := parseField(parts[0], 0, 59, nil, false)
	65	+	if err != nil {
	66	+	return Cron{}, fmt.Errorf("invalid cron minute: %w", err)
	67	+	}
	68	+	hour, err := parseField(parts[1], 0, 23, nil, false)
	69	+	if err != nil {
	70	+	return Cron{}, fmt.Errorf("invalid cron hour: %w", err)
	71	+	}
	72	+	day, err := parseField(parts[2], 1, 31, nil, false)
	73	+	if err != nil {
	74	+	return Cron{}, fmt.Errorf("invalid cron day-of-month: %w", err)
	75	+	}
	76	+	month, err := parseField(parts[3], 1, 12, monthNames, false)
	77	+	if err != nil {
	78	+	return Cron{}, fmt.Errorf("invalid cron month: %w", err)
	79	+	}
	80	+	week, err := parseField(parts[4], 0, 7, weekdayNames, true)
	81	+	if err != nil {
	82	+	return Cron{}, fmt.Errorf("invalid cron day-of-week: %w", err)
	83	+	}
	84	+	return Cron{minute: minute, hour: hour, day: day, month: month, week: week, loc: location}, nil
	85	+	}
	86	+
	87	+	// Next returns the first schedule occurrence strictly after after. An invalid
	88	+	// calendar expression such as 31 February has no occurrence and returns the
	89	+	// zero time.
	90	+	func (c Cron) Next(after time.Time) time.Time {
	91	+	if c.loc == nil {
	92	+	return time.Time{}
	93	+	}
	94	+	candidate := after.In(c.loc).Truncate(time.Minute).Add(time.Minute)
	95	+	// Eight years covers every leap-year placement while keeping impossible
	96	+	// expressions bounded. Normal schedules resolve in a handful of steps.
	97	+	const maxMinutes = 8 * 366 * 24 * 60
	98	+	for range maxMinutes {
	99	+	if c.matches(candidate) {
	100	+	return candidate
	101	+	}
	102	+	candidate = candidate.Add(time.Minute)
	103	+	}
	104	+	return time.Time{}
	105	+	}
	106	+
	107	+	func (c Cron) matches(t time.Time) bool {
	108	+	if !c.minute.matches(t.Minute()) \|\| !c.hour.matches(t.Hour()) \|\|
	109	+	!c.month.matches(int(t.Month())) {
	110	+	return false
	111	+	}
	112	+	dom := c.day.matches(t.Day())
	113	+	dow := c.week.matches(int(t.Weekday()))
	114	+	// Traditional cron treats restricted day-of-month and day-of-week fields
	115	+	// as an OR. When either is *, the other field controls matching.
	116	+	if !c.day.wildcard && !c.week.wildcard {
	117	+	return dom \|\| dow
	118	+	}
	119	+	return dom && dow
	120	+	}
	121	+
	122	+	func (f field) matches(value int) bool {
	123	+	return value >= 0 && value < len(f.values) && f.values[value]
	124	+	}
	125	+
	126	+	func parseField(raw string, min, max int, aliases map[string]int, sundaySeven bool) (field, error) {
	127	+	out := field{values: make([]bool, max+1), wildcard: raw == "*"}
	128	+	if raw == "" {
	129	+	return field{}, fmt.Errorf("empty field")
	130	+	}
	131	+	for _, part := range strings.Split(raw, ",") {
	132	+	if part == "" {
	133	+	return field{}, fmt.Errorf("empty list item")
	134	+	}
	135	+	base, step, hasStep, err := splitStep(part)
	136	+	if err != nil {
	137	+	return field{}, err
	138	+	}
	139	+	start, end := min, max
	140	+	switch {
	141	+	case base == "*":
	142	+	case strings.Contains(base, "-"):
	143	+	pieces := strings.Split(base, "-")
	144	+	if len(pieces) != 2 \|\| pieces[0] == "" \|\| pieces[1] == "" {
	145	+	return field{}, fmt.Errorf("invalid range %q", base)
	146	+	}
	147	+	start, err = parseCronValue(pieces[0], min, max, aliases, sundaySeven)
	148	+	if err != nil {
	149	+	return field{}, err
	150	+	}
	151	+	end, err = parseCronValue(pieces[1], min, max, aliases, sundaySeven)
	152	+	if err != nil {
	153	+	return field{}, err
	154	+	}
	155	+	if start > end {
	156	+	return field{}, fmt.Errorf("range %q descends", base)
	157	+	}
	158	+	default:
	159	+	if hasStep {
	160	+	return field{}, fmt.Errorf("step requires * or a range")
	161	+	}
	162	+	start, err = parseCronValue(base, min, max, aliases, sundaySeven)
	163	+	if err != nil {
	164	+	return field{}, err
	165	+	}
	166	+	end = start
	167	+	}
	168	+	if !hasStep {
	169	+	step = 1
	170	+	}
	171	+	for value := start; value <= end; value += step {
	172	+	out.values[value] = true
	173	+	}
	174	+	}
	175	+	return out, nil
	176	+	}
	177	+
	178	+	func splitStep(part string) (base string, step int, hasStep bool, err error) {
	179	+	pieces := strings.Split(part, "/")
	180	+	if len(pieces) == 1 {
	181	+	return part, 0, false, nil
	182	+	}
	183	+	if len(pieces) != 2 \|\| pieces[0] == "" \|\| pieces[1] == "" {
	184	+	return "", 0, false, fmt.Errorf("invalid step %q", part)
	185	+	}
	186	+	step, err = strconv.Atoi(pieces[1])
	187	+	if err != nil \|\| step <= 0 {
	188	+	return "", 0, false, fmt.Errorf("invalid step %q", part)
	189	+	}
	190	+	return pieces[0], step, true, nil
	191	+	}
	192	+
	193	+	func parseCronValue(raw string, min, max int, aliases map[string]int, sundaySeven bool) (int, error) {
	194	+	if value, ok := aliases[raw]; ok {
	195	+	return value, nil
	196	+	}
	197	+	value, err := strconv.Atoi(raw)
	198	+	if err != nil {
	199	+	return 0, fmt.Errorf("%q is not a number", raw)
	200	+	}
	201	+	if sundaySeven && value == 7 {
	202	+	return 0, nil
	203	+	}
	204	+	if value < min \|\| value > max {
	205	+	return 0, fmt.Errorf("%d is outside %d-%d", value, min, max)
	206	+	}
	207	+	return value, nil
	208	+	}

@@ -0,0 +1,208 @@
	1	+	// Package scheduler provides portable, in-process calendar scheduling.
	2	+	package scheduler
	3	+
	4	+	import (
	5	+	"fmt"
	6	+	"strconv"
	7	+	"strings"
	8	+	"time"
	9	+	)
	10	+
	11	+	// Cron is a five-field cron schedule (minute, hour, day of month, month, day
	12	+	// of week). It evaluates calendar times in the location captured when the
	13	+	// schedule is created. A repeated local minute during a DST fall-back matches
	14	+	// twice; a local minute skipped during a spring-forward does not match.
	15	+	type Cron struct {
	16	+	minute field
	17	+	hour field
	18	+	day field
	19	+	month field
	20	+	week field
	21	+	loc *time.Location
	22	+	}
	23	+
	24	+	type field struct {
	25	+	values []bool
	26	+	wildcard bool
	27	+	}
	28	+
	29	+	var cronMacros = map[string]string{
	30	+	"@yearly": "0 0 1 1 *",
	31	+	"@annually": "0 0 1 1 *",
	32	+	"@monthly": "0 0 1 * *",
	33	+	"@weekly": "0 0 * * 0",
	34	+	"@daily": "0 0 * * *",
	35	+	"@midnight": "0 0 * * *",
	36	+	"@hourly": "0 * * * *",
	37	+	}
	38	+
	39	+	var monthNames = map[string]int{
	40	+	"jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6,
	41	+	"jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12,
	42	+	}
	43	+
	44	+	var weekdayNames = map[string]int{
	45	+	"sun": 0, "mon": 1, "tue": 2, "wed": 3, "thu": 4, "fri": 5, "sat": 6,
	46	+	}
	47	+
	48	+	// ParseCron parses a conventional five-field cron expression. Month and
	49	+	// weekday names use the English three-letter abbreviations. The supplied
	50	+	// location is retained by the schedule, so later changes to time.Local do not
	51	+	// change an existing task's calendar semantics.
	52	+	func ParseCron(expression string, location *time.Location) (Cron, error) {
	53	+	if location == nil {
	54	+	return Cron{}, fmt.Errorf("cron timezone is required")
	55	+	}
	56	+	expression = strings.ToLower(strings.TrimSpace(expression))
	57	+	if replacement, ok := cronMacros[expression]; ok {
	58	+	expression = replacement
	59	+	}
	60	+	parts := strings.Fields(expression)
	61	+	if len(parts) != 5 {
	62	+	return Cron{}, fmt.Errorf("cron requires five fields (minute hour day-of-month month day-of-week)")
	63	+	}
	64	+	minute, err := parseField(parts[0], 0, 59, nil, false)
	65	+	if err != nil {
	66	+	return Cron{}, fmt.Errorf("invalid cron minute: %w", err)
	67	+	}
	68	+	hour, err := parseField(parts[1], 0, 23, nil, false)
	69	+	if err != nil {
	70	+	return Cron{}, fmt.Errorf("invalid cron hour: %w", err)
	71	+	}
	72	+	day, err := parseField(parts[2], 1, 31, nil, false)
	73	+	if err != nil {
	74	+	return Cron{}, fmt.Errorf("invalid cron day-of-month: %w", err)
	75	+	}
	76	+	month, err := parseField(parts[3], 1, 12, monthNames, false)
	77	+	if err != nil {
	78	+	return Cron{}, fmt.Errorf("invalid cron month: %w", err)
	79	+	}
	80	+	week, err := parseField(parts[4], 0, 7, weekdayNames, true)
	81	+	if err != nil {
	82	+	return Cron{}, fmt.Errorf("invalid cron day-of-week: %w", err)
	83	+	}
	84	+	return Cron{minute: minute, hour: hour, day: day, month: month, week: week, loc: location}, nil
	85	+	}
	86	+
	87	+	// Next returns the first schedule occurrence strictly after after. An invalid
	88	+	// calendar expression such as 31 February has no occurrence and returns the
	89	+	// zero time.
	90	+	func (c Cron) Next(after time.Time) time.Time {
	91	+	if c.loc == nil {
	92	+	return time.Time{}
	93	+	}
	94	+	candidate := after.In(c.loc).Truncate(time.Minute).Add(time.Minute)
	95	+	// Eight years covers every leap-year placement while keeping impossible
	96	+	// expressions bounded. Normal schedules resolve in a handful of steps.
	97	+	const maxMinutes = 8 * 366 * 24 * 60
	98	+	for range maxMinutes {
	99	+	if c.matches(candidate) {
	100	+	return candidate
	101	+	}
	102	+	candidate = candidate.Add(time.Minute)
	103	+	}
	104	+	return time.Time{}
	105	+	}
	106	+
	107	+	func (c Cron) matches(t time.Time) bool {
	108	+	if !c.minute.matches(t.Minute()) \|\| !c.hour.matches(t.Hour()) \|\|
	109	+	!c.month.matches(int(t.Month())) {
	110	+	return false
	111	+	}
	112	+	dom := c.day.matches(t.Day())
	113	+	dow := c.week.matches(int(t.Weekday()))
	114	+	// Traditional cron treats restricted day-of-month and day-of-week fields
	115	+	// as an OR. When either is *, the other field controls matching.
	116	+	if !c.day.wildcard && !c.week.wildcard {
	117	+	return dom \|\| dow
	118	+	}
	119	+	return dom && dow
	120	+	}
	121	+
	122	+	func (f field) matches(value int) bool {
	123	+	return value >= 0 && value < len(f.values) && f.values[value]
	124	+	}
	125	+
	126	+	func parseField(raw string, min, max int, aliases map[string]int, sundaySeven bool) (field, error) {
	127	+	out := field{values: make([]bool, max+1), wildcard: raw == "*"}
	128	+	if raw == "" {
	129	+	return field{}, fmt.Errorf("empty field")
	130	+	}
	131	+	for _, part := range strings.Split(raw, ",") {
	132	+	if part == "" {
	133	+	return field{}, fmt.Errorf("empty list item")
	134	+	}
	135	+	base, step, hasStep, err := splitStep(part)
	136	+	if err != nil {
	137	+	return field{}, err
	138	+	}
	139	+	start, end := min, max
	140	+	switch {
	141	+	case base == "*":
	142	+	case strings.Contains(base, "-"):
	143	+	pieces := strings.Split(base, "-")
	144	+	if len(pieces) != 2 \|\| pieces[0] == "" \|\| pieces[1] == "" {
	145	+	return field{}, fmt.Errorf("invalid range %q", base)
	146	+	}
	147	+	start, err = parseCronValue(pieces[0], min, max, aliases, sundaySeven)
	148	+	if err != nil {
	149	+	return field{}, err
	150	+	}
	151	+	end, err = parseCronValue(pieces[1], min, max, aliases, sundaySeven)
	152	+	if err != nil {
	153	+	return field{}, err
	154	+	}
	155	+	if start > end {
	156	+	return field{}, fmt.Errorf("range %q descends", base)
	157	+	}
	158	+	default:
	159	+	if hasStep {
	160	+	return field{}, fmt.Errorf("step requires * or a range")
	161	+	}
	162	+	start, err = parseCronValue(base, min, max, aliases, sundaySeven)
	163	+	if err != nil {
	164	+	return field{}, err
	165	+	}
	166	+	end = start
	167	+	}
	168	+	if !hasStep {
	169	+	step = 1
	170	+	}
	171	+	for value := start; value <= end; value += step {
	172	+	out.values[value] = true
	173	+	}
	174	+	}
	175	+	return out, nil
	176	+	}
	177	+
	178	+	func splitStep(part string) (base string, step int, hasStep bool, err error) {
	179	+	pieces := strings.Split(part, "/")
	180	+	if len(pieces) == 1 {
	181	+	return part, 0, false, nil
	182	+	}
	183	+	if len(pieces) != 2 \|\| pieces[0] == "" \|\| pieces[1] == "" {
	184	+	return "", 0, false, fmt.Errorf("invalid step %q", part)
	185	+	}
	186	+	step, err = strconv.Atoi(pieces[1])
	187	+	if err != nil \|\| step <= 0 {
	188	+	return "", 0, false, fmt.Errorf("invalid step %q", part)
	189	+	}
	190	+	return pieces[0], step, true, nil
	191	+	}
	192	+
	193	+	func parseCronValue(raw string, min, max int, aliases map[string]int, sundaySeven bool) (int, error) {
	194	+	if value, ok := aliases[raw]; ok {
	195	+	return value, nil
	196	+	}
	197	+	value, err := strconv.Atoi(raw)
	198	+	if err != nil {
	199	+	return 0, fmt.Errorf("%q is not a number", raw)
	200	+	}
	201	+	if sundaySeven && value == 7 {
	202	+	return 0, nil
	203	+	}
	204	+	if value < min \|\| value > max {
	205	+	return 0, fmt.Errorf("%d is outside %d-%d", value, min, max)
	206	+	}
	207	+	return value, nil
	208	+	}

A packages/agent/scheduler/cron_test.go +80 -0

@@ -0,0 +1,80 @@
	1	+	package scheduler
	2	+
	3	+	import (
	4	+	"testing"
	5	+	"time"
	6	+	)
	7	+
	8	+	func TestParseCronNext(t *testing.T) {
	9	+	loc := time.FixedZone("UTC+2", 26060)
	10	+	schedule, err := ParseCron("/15 9-10 * 1-5", loc)
	11	+	if err != nil {
	12	+	t.Fatalf("ParseCron() error = %v", err)
	13	+	}
	14	+
	15	+	after := time.Date(2026, time.August, 14, 9, 14, 45, 0, loc) // Friday
	16	+	want := time.Date(2026, time.August, 14, 9, 15, 0, 0, loc)
	17	+	if got := schedule.Next(after); !got.Equal(want) {
	18	+	t.Fatalf("Next() = %v, want %v", got, want)
	19	+	}
	20	+	}
	21	+
	22	+	func TestParseCronDayOfMonthOrDayOfWeek(t *testing.T) {
	23	+	loc := time.UTC
	24	+	schedule, err := ParseCron("0 9 15 * 1", loc)
	25	+	if err != nil {
	26	+	t.Fatalf("ParseCron() error = %v", err)
	27	+	}
	28	+
	29	+	// Vixie cron semantics: when both fields are restricted, either match is
	30	+	// sufficient. Monday 2026-06-08 is not the fifteenth.
	31	+	after := time.Date(2026, time.June, 7, 9, 0, 0, 0, loc)
	32	+	want := time.Date(2026, time.June, 8, 9, 0, 0, 0, loc)
	33	+	if got := schedule.Next(after); !got.Equal(want) {
	34	+	t.Fatalf("Next() = %v, want %v", got, want)
	35	+	}
	36	+	}
	37	+
	38	+	func TestParseCronDSTTransitions(t *testing.T) {
	39	+	loc, err := time.LoadLocation("America/New_York")
	40	+	if err != nil {
	41	+	t.Fatalf("LoadLocation() error = %v", err)
	42	+	}
	43	+	fallBack, err := ParseCron("30 1 * * *", loc)
	44	+	if err != nil {
	45	+	t.Fatalf("ParseCron() error = %v", err)
	46	+	}
	47	+	first := fallBack.Next(time.Date(2026, time.November, 1, 0, 59, 0, 0, loc))
	48	+	second := fallBack.Next(first)
	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	+	}
	52	+
	53	+	springForward, err := ParseCron("30 2 * * *", loc)
	54	+	if err != nil {
	55	+	t.Fatalf("ParseCron() error = %v", err)
	56	+	}
	57	+	got := springForward.Next(time.Date(2026, time.March, 8, 1, 59, 0, 0, loc))
	58	+	want := time.Date(2026, time.March, 9, 2, 30, 0, 0, loc)
	59	+	if !got.Equal(want) {
	60	+	t.Fatalf("spring-forward Next() = %v, want %v", got, want)
	61	+	}
	62	+	}
	63	+
	64	+	func TestParseCronRejectsInvalidField(t *testing.T) {
	65	+	if _, err := ParseCron("61 * * * *", time.UTC); err == nil {
	66	+	t.Fatal("ParseCron() error = nil, want invalid minute error")
	67	+	}
	68	+	}
	69	+
	70	+	func TestParseCronSupportsMacro(t *testing.T) {
	71	+	schedule, err := ParseCron("@daily", time.UTC)
	72	+	if err != nil {
	73	+	t.Fatalf("ParseCron() error = %v", err)
	74	+	}
	75	+	got := schedule.Next(time.Date(2026, time.August, 14, 23, 59, 0, 0, time.UTC))
	76	+	want := time.Date(2026, time.August, 15, 0, 0, 0, 0, time.UTC)
	77	+	if !got.Equal(want) {
	78	+	t.Fatalf("Next() = %v, want %v", got, want)
	79	+	}
	80	+	}

@@ -0,0 +1,80 @@
	1	+	package scheduler
	2	+
	3	+	import (
	4	+	"testing"
	5	+	"time"
	6	+	)
	7	+
	8	+	func TestParseCronNext(t *testing.T) {
	9	+	loc := time.FixedZone("UTC+2", 26060)
	10	+	schedule, err := ParseCron("/15 9-10 * 1-5", loc)
	11	+	if err != nil {
	12	+	t.Fatalf("ParseCron() error = %v", err)
	13	+	}
	14	+
	15	+	after := time.Date(2026, time.August, 14, 9, 14, 45, 0, loc) // Friday
	16	+	want := time.Date(2026, time.August, 14, 9, 15, 0, 0, loc)
	17	+	if got := schedule.Next(after); !got.Equal(want) {
	18	+	t.Fatalf("Next() = %v, want %v", got, want)
	19	+	}
	20	+	}
	21	+
	22	+	func TestParseCronDayOfMonthOrDayOfWeek(t *testing.T) {
	23	+	loc := time.UTC
	24	+	schedule, err := ParseCron("0 9 15 * 1", loc)
	25	+	if err != nil {
	26	+	t.Fatalf("ParseCron() error = %v", err)
	27	+	}
	28	+
	29	+	// Vixie cron semantics: when both fields are restricted, either match is
	30	+	// sufficient. Monday 2026-06-08 is not the fifteenth.
	31	+	after := time.Date(2026, time.June, 7, 9, 0, 0, 0, loc)
	32	+	want := time.Date(2026, time.June, 8, 9, 0, 0, 0, loc)
	33	+	if got := schedule.Next(after); !got.Equal(want) {
	34	+	t.Fatalf("Next() = %v, want %v", got, want)
	35	+	}
	36	+	}
	37	+
	38	+	func TestParseCronDSTTransitions(t *testing.T) {
	39	+	loc, err := time.LoadLocation("America/New_York")
	40	+	if err != nil {
	41	+	t.Fatalf("LoadLocation() error = %v", err)
	42	+	}
	43	+	fallBack, err := ParseCron("30 1 * * *", loc)
	44	+	if err != nil {
	45	+	t.Fatalf("ParseCron() error = %v", err)
	46	+	}
	47	+	first := fallBack.Next(time.Date(2026, time.November, 1, 0, 59, 0, 0, loc))
	48	+	second := fallBack.Next(first)
	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	+	}
	52	+
	53	+	springForward, err := ParseCron("30 2 * * *", loc)
	54	+	if err != nil {
	55	+	t.Fatalf("ParseCron() error = %v", err)
	56	+	}
	57	+	got := springForward.Next(time.Date(2026, time.March, 8, 1, 59, 0, 0, loc))
	58	+	want := time.Date(2026, time.March, 9, 2, 30, 0, 0, loc)
	59	+	if !got.Equal(want) {
	60	+	t.Fatalf("spring-forward Next() = %v, want %v", got, want)
	61	+	}
	62	+	}
	63	+
	64	+	func TestParseCronRejectsInvalidField(t *testing.T) {
	65	+	if _, err := ParseCron("61 * * * *", time.UTC); err == nil {
	66	+	t.Fatal("ParseCron() error = nil, want invalid minute error")
	67	+	}
	68	+	}
	69	+
	70	+	func TestParseCronSupportsMacro(t *testing.T) {
	71	+	schedule, err := ParseCron("@daily", time.UTC)
	72	+	if err != nil {
	73	+	t.Fatalf("ParseCron() error = %v", err)
	74	+	}
	75	+	got := schedule.Next(time.Date(2026, time.August, 14, 23, 59, 0, 0, time.UTC))
	76	+	want := time.Date(2026, time.August, 15, 0, 0, 0, 0, time.UTC)
	77	+	if !got.Equal(want) {
	78	+	t.Fatalf("Next() = %v, want %v", got, want)
	79	+	}
	80	+	}

A packages/agent/scheduler/scheduler.go +322 -0

@@ -0,0 +1,322 @@
	1	+	package scheduler
	2	+
	3	+	import (
	4	+	"context"
	5	+	"fmt"
	6	+	"sort"
	7	+	"strings"
	8	+	"sync"
	9	+	"time"
	10	+
	11	+	"github.com/google/uuid"
	12	+	)
	13	+
	14	+	// Task is one recurring prompt owned by a session branch. Tasks are kept only
	15	+	// in process memory; callers are responsible for stopping the engine with the
	16	+	// host process.
	17	+	type Task struct {
	18	+	ID string
	19	+	SessionID string
	20	+	Cron string
	21	+	Timezone string
	22	+	Message string
	23	+	CreatedAt time.Time
	24	+	NextRun time.Time
	25	+	LastRun time.Time
	26	+	RunCount int
	27	+	LastError string
	28	+
	29	+	schedule Cron
	30	+	queued bool
	31	+	running bool
	32	+	}
	33	+
	34	+	// NewTaskInput describes a task to add to an Engine. Location is captured at
	35	+	// creation rather than consulting time.Local during later evaluations.
	36	+	type NewTaskInput struct {
	37	+	SessionID string
	38	+	Cron string
	39	+	Message string
	40	+	Location *time.Location
	41	+	}
	42	+
	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.
	46	+	type Engine struct {
	47	+	mu sync.Mutex
	48	+	now func() time.Time
	49	+	tasks map[string]Task
	50	+	wake chan struct{}
	51	+	}
	52	+
	53	+	// NewEngine creates an empty scheduler. now is injectable for tests; nil uses
	54	+	// time.Now.
	55	+	func NewEngine(now func() time.Time) *Engine {
	56	+	if now == nil {
	57	+	now = time.Now
	58	+	}
	59	+	return &Engine{now: now, tasks: make(map[string]Task), wake: make(chan struct{}, 1)}
	60	+	}
	61	+
	62	+	// Add validates and records a new recurring task.
	63	+	func (e *Engine) Add(input NewTaskInput) (Task, error) {
	64	+	if e == nil {
	65	+	return Task{}, fmt.Errorf("scheduler is not initialized")
	66	+	}
	67	+	input.SessionID = strings.TrimSpace(input.SessionID)
	68	+	input.Cron = strings.TrimSpace(input.Cron)
	69	+	input.Message = strings.TrimSpace(input.Message)
	70	+	if input.SessionID == "" {
	71	+	return Task{}, fmt.Errorf("session ID is required")
	72	+	}
	73	+	if input.Message == "" {
	74	+	return Task{}, fmt.Errorf("scheduled message is required")
	75	+	}
	76	+	if input.Location == nil {
	77	+	return Task{}, fmt.Errorf("schedule timezone is required")
	78	+	}
	79	+	schedule, err := ParseCron(input.Cron, input.Location)
	80	+	if err != nil {
	81	+	return Task{}, err
	82	+	}
	83	+	now := e.now()
	84	+	next := schedule.Next(now)
	85	+	if next.IsZero() {
	86	+	return Task{}, fmt.Errorf("cron has no occurrence in the next eight years")
	87	+	}
	88	+	task := Task{
	89	+	ID: uuid.NewString(),
	90	+	SessionID: input.SessionID,
	91	+	Cron: input.Cron,
	92	+	Timezone: input.Location.String(),
	93	+	Message: input.Message,
	94	+	CreatedAt: now,
	95	+	NextRun: next,
	96	+	schedule: schedule,
	97	+	}
	98	+	e.mu.Lock()
	99	+	e.tasks[task.ID] = task
	100	+	e.mu.Unlock()
	101	+	e.signalWake()
	102	+	return task, nil
	103	+	}
	104	+
	105	+	// Cancel removes a scheduled task. It returns false when the ID is unknown.
	106	+	func (e *Engine) Cancel(id string) bool {
	107	+	if e == nil {
	108	+	return false
	109	+	}
	110	+	e.mu.Lock()
	111	+	_, ok := e.tasks[id]
	112	+	if ok {
	113	+	delete(e.tasks, id)
	114	+	}
	115	+	e.mu.Unlock()
	116	+	if ok {
	117	+	e.signalWake()
	118	+	}
	119	+	return ok
	120	+	}
	121	+
	122	+	// Get returns a task snapshot by ID.
	123	+	func (e *Engine) Get(id string) (Task, bool) {
	124	+	if e == nil {
	125	+	return Task{}, false
	126	+	}
	127	+	e.mu.Lock()
	128	+	defer e.mu.Unlock()
	129	+	task, ok := e.tasks[id]
	130	+	return task, ok
	131	+	}
	132	+
	133	+	// List returns snapshots ordered by next execution and then ID.
	134	+	func (e *Engine) List() []Task {
	135	+	if e == nil {
	136	+	return nil
	137	+	}
	138	+	e.mu.Lock()
	139	+	tasks := make([]Task, 0, len(e.tasks))
	140	+	for _, task := range e.tasks {
	141	+	tasks = append(tasks, task)
	142	+	}
	143	+	e.mu.Unlock()
	144	+	sort.Slice(tasks, func(i, j int) bool {
	145	+	if tasks[i].NextRun.Equal(tasks[j].NextRun) {
	146	+	return tasks[i].ID < tasks[j].ID
	147	+	}
	148	+	return tasks[i].NextRun.Before(tasks[j].NextRun)
	149	+	})
	150	+	return tasks
	151	+	}
	152	+
	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 {
	157	+	if e == nil {
	158	+	return fmt.Errorf("scheduler is not initialized")
	159	+	}
	160	+	if executor == nil {
	161	+	return fmt.Errorf("scheduler executor is required")
	162	+	}
	163	+	for {
	164	+	if ctx.Err() != nil {
	165	+	return nil
	166	+	}
	167	+	now := e.now()
	168	+	e.dispatchDue(ctx, now, executor)
	169	+	next, ok := e.nextRun()
	170	+	if !ok {
	171	+	select {
	172	+	case <-ctx.Done():
	173	+	return nil
	174	+	case <-e.wake:
	175	+	}
	176	+	continue
	177	+	}
	178	+	wait := time.Until(next)
	179	+	if wait < 0 {
	180	+	wait = 0
	181	+	}
	182	+	timer := time.NewTimer(wait)
	183	+	select {
	184	+	case <-ctx.Done():
	185	+	if !timer.Stop() {
	186	+	select {
	187	+	case <-timer.C:
	188	+	default:
	189	+	}
	190	+	}
	191	+	return nil
	192	+	case <-e.wake:
	193	+	if !timer.Stop() {
	194	+	select {
	195	+	case <-timer.C:
	196	+	default:
	197	+	}
	198	+	}
	199	+	case <-timer.C:
	200	+	}
	201	+	}
	202	+	}
	203	+
	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 {
	210	+	return
	211	+	}
	212	+	for _, task := range e.takeDue(now) {
	213	+	go e.dispatchTask(ctx, task, executor)
	214	+	}
	215	+	}
	216	+
	217	+	func (e *Engine) takeDue(now time.Time) []Task {
	218	+	e.mu.Lock()
	219	+	defer e.mu.Unlock()
	220	+	due := make([]Task, 0)
	221	+	for id, task := range e.tasks {
	222	+	if task.queued \|\| task.running \|\| task.NextRun.After(now) {
	223	+	continue
	224	+	}
	225	+	task.queued = true
	226	+	e.tasks[id] = task
	227	+	due = append(due, task)
	228	+	}
	229	+	sort.Slice(due, func(i, j int) bool {
	230	+	if due[i].NextRun.Equal(due[j].NextRun) {
	231	+	return due[i].ID < due[j].ID
	232	+	}
	233	+	return due[i].NextRun.Before(due[j].NextRun)
	234	+	})
	235	+	return due
	236	+	}
	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	+
	274	+	func (e *Engine) finish(id string, scheduledFor time.Time, err error) {
	275	+	e.mu.Lock()
	276	+	task, ok := e.tasks[id]
	277	+	if !ok \|\| !task.NextRun.Equal(scheduledFor) {
	278	+	e.mu.Unlock()
	279	+	return // cancelled or replaced while the executor ran
	280	+	}
	281	+	task.running = false
	282	+	task.LastRun = scheduledFor
	283	+	task.RunCount++
	284	+	if err != nil {
	285	+	task.LastError = err.Error()
	286	+	} else {
	287	+	task.LastError = ""
	288	+	}
	289	+	// Calculate from completion time, rather than repeatedly from the prior
	290	+	// deadline, so a sleeping machine or a slow task never creates a burst of
	291	+	// stale catch-up executions.
	292	+	task.NextRun = task.schedule.Next(e.now())
	293	+	if task.NextRun.IsZero() {
	294	+	delete(e.tasks, id)
	295	+	} else {
	296	+	e.tasks[id] = task
	297	+	}
	298	+	e.mu.Unlock()
	299	+	e.signalWake()
	300	+	}
	301	+
	302	+	func (e *Engine) nextRun() (time.Time, bool) {
	303	+	e.mu.Lock()
	304	+	defer e.mu.Unlock()
	305	+	var next time.Time
	306	+	for _, task := range e.tasks {
	307	+	if task.queued \|\| task.running {
	308	+	continue
	309	+	}
	310	+	if next.IsZero() \|\| task.NextRun.Before(next) {
	311	+	next = task.NextRun
	312	+	}
	313	+	}
	314	+	return next, !next.IsZero()
	315	+	}
	316	+
	317	+	func (e *Engine) signalWake() {
	318	+	select {
	319	+	case e.wake <- struct{}{}:
	320	+	default:
	321	+	}
	322	+	}

@@ -0,0 +1,322 @@
	1	+	package scheduler
	2	+
	3	+	import (
	4	+	"context"
	5	+	"fmt"
	6	+	"sort"
	7	+	"strings"
	8	+	"sync"
	9	+	"time"
	10	+
	11	+	"github.com/google/uuid"
	12	+	)
	13	+
	14	+	// Task is one recurring prompt owned by a session branch. Tasks are kept only
	15	+	// in process memory; callers are responsible for stopping the engine with the
	16	+	// host process.
	17	+	type Task struct {
	18	+	ID string
	19	+	SessionID string
	20	+	Cron string
	21	+	Timezone string
	22	+	Message string
	23	+	CreatedAt time.Time
	24	+	NextRun time.Time
	25	+	LastRun time.Time
	26	+	RunCount int
	27	+	LastError string
	28	+
	29	+	schedule Cron
	30	+	queued bool
	31	+	running bool
	32	+	}
	33	+
	34	+	// NewTaskInput describes a task to add to an Engine. Location is captured at
	35	+	// creation rather than consulting time.Local during later evaluations.
	36	+	type NewTaskInput struct {
	37	+	SessionID string
	38	+	Cron string
	39	+	Message string
	40	+	Location *time.Location
	41	+	}
	42	+
	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.
	46	+	type Engine struct {
	47	+	mu sync.Mutex
	48	+	now func() time.Time
	49	+	tasks map[string]Task
	50	+	wake chan struct{}
	51	+	}
	52	+
	53	+	// NewEngine creates an empty scheduler. now is injectable for tests; nil uses
	54	+	// time.Now.
	55	+	func NewEngine(now func() time.Time) *Engine {
	56	+	if now == nil {
	57	+	now = time.Now
	58	+	}
	59	+	return &Engine{now: now, tasks: make(map[string]Task), wake: make(chan struct{}, 1)}
	60	+	}
	61	+
	62	+	// Add validates and records a new recurring task.
	63	+	func (e *Engine) Add(input NewTaskInput) (Task, error) {
	64	+	if e == nil {
	65	+	return Task{}, fmt.Errorf("scheduler is not initialized")
	66	+	}
	67	+	input.SessionID = strings.TrimSpace(input.SessionID)
	68	+	input.Cron = strings.TrimSpace(input.Cron)
	69	+	input.Message = strings.TrimSpace(input.Message)
	70	+	if input.SessionID == "" {
	71	+	return Task{}, fmt.Errorf("session ID is required")
	72	+	}
	73	+	if input.Message == "" {
	74	+	return Task{}, fmt.Errorf("scheduled message is required")
	75	+	}
	76	+	if input.Location == nil {
	77	+	return Task{}, fmt.Errorf("schedule timezone is required")
	78	+	}
	79	+	schedule, err := ParseCron(input.Cron, input.Location)
	80	+	if err != nil {
	81	+	return Task{}, err
	82	+	}
	83	+	now := e.now()
	84	+	next := schedule.Next(now)
	85	+	if next.IsZero() {
	86	+	return Task{}, fmt.Errorf("cron has no occurrence in the next eight years")
	87	+	}
	88	+	task := Task{
	89	+	ID: uuid.NewString(),
	90	+	SessionID: input.SessionID,
	91	+	Cron: input.Cron,
	92	+	Timezone: input.Location.String(),
	93	+	Message: input.Message,
	94	+	CreatedAt: now,
	95	+	NextRun: next,
	96	+	schedule: schedule,
	97	+	}
	98	+	e.mu.Lock()
	99	+	e.tasks[task.ID] = task
	100	+	e.mu.Unlock()
	101	+	e.signalWake()
	102	+	return task, nil
	103	+	}
	104	+
	105	+	// Cancel removes a scheduled task. It returns false when the ID is unknown.
	106	+	func (e *Engine) Cancel(id string) bool {
	107	+	if e == nil {
	108	+	return false
	109	+	}
	110	+	e.mu.Lock()
	111	+	_, ok := e.tasks[id]
	112	+	if ok {
	113	+	delete(e.tasks, id)
	114	+	}
	115	+	e.mu.Unlock()
	116	+	if ok {
	117	+	e.signalWake()
	118	+	}
	119	+	return ok
	120	+	}
	121	+
	122	+	// Get returns a task snapshot by ID.
	123	+	func (e *Engine) Get(id string) (Task, bool) {
	124	+	if e == nil {
	125	+	return Task{}, false
	126	+	}
	127	+	e.mu.Lock()
	128	+	defer e.mu.Unlock()
	129	+	task, ok := e.tasks[id]
	130	+	return task, ok
	131	+	}
	132	+
	133	+	// List returns snapshots ordered by next execution and then ID.
	134	+	func (e *Engine) List() []Task {
	135	+	if e == nil {
	136	+	return nil
	137	+	}
	138	+	e.mu.Lock()
	139	+	tasks := make([]Task, 0, len(e.tasks))
	140	+	for _, task := range e.tasks {
	141	+	tasks = append(tasks, task)
	142	+	}
	143	+	e.mu.Unlock()
	144	+	sort.Slice(tasks, func(i, j int) bool {
	145	+	if tasks[i].NextRun.Equal(tasks[j].NextRun) {
	146	+	return tasks[i].ID < tasks[j].ID
	147	+	}
	148	+	return tasks[i].NextRun.Before(tasks[j].NextRun)
	149	+	})
	150	+	return tasks
	151	+	}
	152	+
	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 {
	157	+	if e == nil {
	158	+	return fmt.Errorf("scheduler is not initialized")
	159	+	}
	160	+	if executor == nil {
	161	+	return fmt.Errorf("scheduler executor is required")
	162	+	}
	163	+	for {
	164	+	if ctx.Err() != nil {
	165	+	return nil
	166	+	}
	167	+	now := e.now()
	168	+	e.dispatchDue(ctx, now, executor)
	169	+	next, ok := e.nextRun()
	170	+	if !ok {
	171	+	select {
	172	+	case <-ctx.Done():
	173	+	return nil
	174	+	case <-e.wake:
	175	+	}
	176	+	continue
	177	+	}
	178	+	wait := time.Until(next)
	179	+	if wait < 0 {
	180	+	wait = 0
	181	+	}
	182	+	timer := time.NewTimer(wait)
	183	+	select {
	184	+	case <-ctx.Done():
	185	+	if !timer.Stop() {
	186	+	select {
	187	+	case <-timer.C:
	188	+	default:
	189	+	}
	190	+	}
	191	+	return nil
	192	+	case <-e.wake:
	193	+	if !timer.Stop() {
	194	+	select {
	195	+	case <-timer.C:
	196	+	default:
	197	+	}
	198	+	}
	199	+	case <-timer.C:
	200	+	}
	201	+	}
	202	+	}
	203	+
	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 {
	210	+	return
	211	+	}
	212	+	for _, task := range e.takeDue(now) {
	213	+	go e.dispatchTask(ctx, task, executor)
	214	+	}
	215	+	}
	216	+
	217	+	func (e *Engine) takeDue(now time.Time) []Task {
	218	+	e.mu.Lock()
	219	+	defer e.mu.Unlock()
	220	+	due := make([]Task, 0)
	221	+	for id, task := range e.tasks {
	222	+	if task.queued \|\| task.running \|\| task.NextRun.After(now) {
	223	+	continue
	224	+	}
	225	+	task.queued = true
	226	+	e.tasks[id] = task
	227	+	due = append(due, task)
	228	+	}
	229	+	sort.Slice(due, func(i, j int) bool {
	230	+	if due[i].NextRun.Equal(due[j].NextRun) {
	231	+	return due[i].ID < due[j].ID
	232	+	}
	233	+	return due[i].NextRun.Before(due[j].NextRun)
	234	+	})
	235	+	return due
	236	+	}
	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	+
	274	+	func (e *Engine) finish(id string, scheduledFor time.Time, err error) {
	275	+	e.mu.Lock()
	276	+	task, ok := e.tasks[id]
	277	+	if !ok \|\| !task.NextRun.Equal(scheduledFor) {
	278	+	e.mu.Unlock()
	279	+	return // cancelled or replaced while the executor ran
	280	+	}
	281	+	task.running = false
	282	+	task.LastRun = scheduledFor
	283	+	task.RunCount++
	284	+	if err != nil {
	285	+	task.LastError = err.Error()
	286	+	} else {
	287	+	task.LastError = ""
	288	+	}
	289	+	// Calculate from completion time, rather than repeatedly from the prior
	290	+	// deadline, so a sleeping machine or a slow task never creates a burst of
	291	+	// stale catch-up executions.
	292	+	task.NextRun = task.schedule.Next(e.now())
	293	+	if task.NextRun.IsZero() {
	294	+	delete(e.tasks, id)
	295	+	} else {
	296	+	e.tasks[id] = task
	297	+	}
	298	+	e.mu.Unlock()
	299	+	e.signalWake()
	300	+	}
	301	+
	302	+	func (e *Engine) nextRun() (time.Time, bool) {
	303	+	e.mu.Lock()
	304	+	defer e.mu.Unlock()
	305	+	var next time.Time
	306	+	for _, task := range e.tasks {
	307	+	if task.queued \|\| task.running {
	308	+	continue
	309	+	}
	310	+	if next.IsZero() \|\| task.NextRun.Before(next) {
	311	+	next = task.NextRun
	312	+	}
	313	+	}
	314	+	return next, !next.IsZero()
	315	+	}
	316	+
	317	+	func (e *Engine) signalWake() {
	318	+	select {
	319	+	case e.wake <- struct{}{}:
	320	+	default:
	321	+	}
	322	+	}

A packages/agent/scheduler/scheduler_test.go +231 -0

@@ -0,0 +1,231 @@
	1	+	package scheduler
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+	"testing"
	8	+	"time"
	9	+	)
	10	+
	11	+	func TestEngineDispatchDueReschedulesTask(t *testing.T) {
	12	+	loc := time.UTC
	13	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, loc)
	14	+	engine := NewEngine(func() time.Time { return now })
	15	+	created, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "check status", Location: loc})
	16	+	if err != nil {
	17	+	t.Fatalf("Add() error = %v", err)
	18	+	}
	19	+
	20	+	done := make(chan Task, 1)
	21	+	now = now.Add(time.Minute)
	22	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
	23	+	done <- task
	24	+	return nil
	25	+	})
	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")
	33	+	}
	34	+
	35	+	eventually(t, func() bool {
	36	+	tasks := engine.List()
	37	+	return len(tasks) == 1 && tasks[0].RunCount == 1
	38	+	})
	39	+	tasks := engine.List()
	40	+	if want := now.Add(time.Minute); !tasks[0].NextRun.Equal(want) {
	41	+	t.Fatalf("NextRun = %v, want %v", tasks[0].NextRun, want)
	42	+	}
	43	+	if !tasks[0].LastRun.Equal(now) \|\| tasks[0].RunCount != 1 {
	44	+	t.Fatalf("task after execution = %#v", tasks[0])
	45	+	}
	46	+	}
	47	+
	48	+	func TestEngineRecordsExecutionFailure(t *testing.T) {
	49	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	50	+	engine := NewEngine(func() time.Time { return now })
	51	+	task, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "run", Location: time.UTC})
	52	+	if err != nil {
	53	+	t.Fatalf("Add() error = %v", err)
	54	+	}
	55	+
	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	+	})
	61	+	}
	62	+
	63	+	func TestEngineCancel(t *testing.T) {
	64	+	engine := NewEngine(nil)
	65	+	task, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "@hourly", Message: "run", Location: time.UTC})
	66	+	if err != nil {
	67	+	t.Fatalf("Add() error = %v", err)
	68	+	}
	69	+	if !engine.Cancel(task.ID) {
	70	+	t.Fatal("Cancel() = false, want true")
	71	+	}
	72	+	if _, ok := engine.Get(task.ID); ok {
	73	+	t.Fatal("Get() found cancelled task")
	74	+	}
	75	+	}
	76	+
	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)
	198	+	ctx, cancel := context.WithCancel(context.Background())
	199	+	cancel()
	200	+	called := false
	201	+	if err := engine.Run(ctx, func(context.Context, Task) error {
	202	+	called = true
	203	+	return nil
	204	+	}); err != nil {
	205	+	t.Fatalf("Run() error = %v", err)
	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")
	231	+	}

@@ -0,0 +1,231 @@
	1	+	package scheduler
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+	"testing"
	8	+	"time"
	9	+	)
	10	+
	11	+	func TestEngineDispatchDueReschedulesTask(t *testing.T) {
	12	+	loc := time.UTC
	13	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, loc)
	14	+	engine := NewEngine(func() time.Time { return now })
	15	+	created, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "check status", Location: loc})
	16	+	if err != nil {
	17	+	t.Fatalf("Add() error = %v", err)
	18	+	}
	19	+
	20	+	done := make(chan Task, 1)
	21	+	now = now.Add(time.Minute)
	22	+	engine.dispatchDue(context.Background(), now, func(_ context.Context, task Task) error {
	23	+	done <- task
	24	+	return nil
	25	+	})
	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")
	33	+	}
	34	+
	35	+	eventually(t, func() bool {
	36	+	tasks := engine.List()
	37	+	return len(tasks) == 1 && tasks[0].RunCount == 1
	38	+	})
	39	+	tasks := engine.List()
	40	+	if want := now.Add(time.Minute); !tasks[0].NextRun.Equal(want) {
	41	+	t.Fatalf("NextRun = %v, want %v", tasks[0].NextRun, want)
	42	+	}
	43	+	if !tasks[0].LastRun.Equal(now) \|\| tasks[0].RunCount != 1 {
	44	+	t.Fatalf("task after execution = %#v", tasks[0])
	45	+	}
	46	+	}
	47	+
	48	+	func TestEngineRecordsExecutionFailure(t *testing.T) {
	49	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	50	+	engine := NewEngine(func() time.Time { return now })
	51	+	task, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "* * * * *", Message: "run", Location: time.UTC})
	52	+	if err != nil {
	53	+	t.Fatalf("Add() error = %v", err)
	54	+	}
	55	+
	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	+	})
	61	+	}
	62	+
	63	+	func TestEngineCancel(t *testing.T) {
	64	+	engine := NewEngine(nil)
	65	+	task, err := engine.Add(NewTaskInput{SessionID: "session-1", Cron: "@hourly", Message: "run", Location: time.UTC})
	66	+	if err != nil {
	67	+	t.Fatalf("Add() error = %v", err)
	68	+	}
	69	+	if !engine.Cancel(task.ID) {
	70	+	t.Fatal("Cancel() = false, want true")
	71	+	}
	72	+	if _, ok := engine.Get(task.ID); ok {
	73	+	t.Fatal("Get() found cancelled task")
	74	+	}
	75	+	}
	76	+
	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)
	198	+	ctx, cancel := context.WithCancel(context.Background())
	199	+	cancel()
	200	+	called := false
	201	+	if err := engine.Run(ctx, func(context.Context, Task) error {
	202	+	called = true
	203	+	return nil
	204	+	}); err != nil {
	205	+	t.Fatalf("Run() error = %v", err)
	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")
	231	+	}

A packages/agent/scheduler_runtime.go +84 -0

@@ -0,0 +1,84 @@
	1	+	package agent
	2	+
	3	+	import (
	4	+	"context"
	5	+	"fmt"
	6	+	"strings"
	7	+	"sync"
	8	+
	9	+	"github.com/bnema/zut/packages/agent/extensions"
	10	+	"github.com/bnema/zut/packages/agent/scheduler"
	11	+	"github.com/bnema/zut/packages/core"
	12	+	"github.com/bnema/zut/packages/provider"
	13	+	)
	14	+
	15	+	// runScheduledSession reconstructs an inactive session in the current zut
	16	+	// process and appends the resulting turn to its existing transcript. The
	17	+	// caller serializes this with session transitions before invoking it.
	18	+	func runScheduledSession(ctx context.Context, task scheduler.Task, args Args, base Resolved, extMgr *extensions.Manager, prepareRegistry func(core.Registry) core.Registry) error {
	19	+	path, err := core.FindManagedSessionByID(ctx, ZutHome(), task.SessionID)
	20	+	if err != nil {
	21	+	return fmt.Errorf("find scheduled session %q: %w", task.SessionID, err)
	22	+	}
	23	+	if path == "" {
	24	+	return fmt.Errorf("scheduled session %q no longer exists", task.SessionID)
	25	+	}
	26	+	sess, messages, err := core.OpenSession(path)
	27	+	if err != nil {
	28	+	return fmt.Errorf("open scheduled session: %w", err)
	29	+	}
	30	+	defer sess.Close()
	31	+
	32	+	providerName := strings.TrimSpace(sess.Meta.Provider)
	33	+	if providerName == "" {
	34	+	providerName = base.Provider
	35	+	}
	36	+	model := strings.TrimSpace(sess.Meta.Model)
	37	+	if model == "" {
	38	+	model = base.Model
	39	+	}
	40	+	ag, _, _, err := buildNonInteractiveSessionAgentWithRegistry(ctx, args, base, extMgr, providerName, model, prepareRegistry)
	41	+	if err != nil {
	42	+	return fmt.Errorf("build scheduled session agent: %w", err)
	43	+	}
	44	+	ag.SetMessages(trimMessagesForResume(messages, 100))
	45	+	if cumulative, lastTurn, usageErr := core.SessionUsageDetail(path); usageErr == nil {
	46	+	ag.SeedCost(cumulative)
	47	+	ag.SeedLastTurnUsage(lastTurn)
	48	+	}
	49	+	ag.SetSessionTimeContext(sess.Meta.Started, sess.Meta.Timezone, sess.Meta.TimezoneOffset)
	50	+	var appendMu sync.Mutex
	51	+	var appendErr error
	52	+	recordAppendError := func(err error) {
	53	+	if err == nil {
	54	+	return
	55	+	}
	56	+	appendMu.Lock()
	57	+	defer appendMu.Unlock()
	58	+	if appendErr == nil {
	59	+	appendErr = err
	60	+	}
	61	+	}
	62	+	ag.OnMessageAppended = func(message provider.Message) {
	63	+	recordAppendError(sess.AppendMessage(message))
	64	+	}
	65	+	ag.OnUsage = func(cumulative provider.Usage) {
	66	+	recordAppendError(sess.AppendUsage(cumulative, cumulative))
	67	+	}
	68	+	ag.OnTranscriptCompacted = func(messages []provider.Message) {
	69	+	recordAppendError(sess.AppendCompaction(messages))
	70	+	}
	71	+	if err := ag.Prompt(ctx, task.Message, nil, func(core.AgentEvent) {}); err != nil {
	72	+	return err
	73	+	}
	74	+	appendMu.Lock()
	75	+	err = appendErr
	76	+	appendMu.Unlock()
	77	+	if err != nil {
	78	+	return fmt.Errorf("append scheduled transcript: %w", err)
	79	+	}
	80	+	if err := sess.Flush(); err != nil {
	81	+	return fmt.Errorf("flush scheduled session: %w", err)
	82	+	}
	83	+	return nil
	84	+	}

@@ -0,0 +1,84 @@
	1	+	package agent
	2	+
	3	+	import (
	4	+	"context"
	5	+	"fmt"
	6	+	"strings"
	7	+	"sync"
	8	+
	9	+	"github.com/bnema/zut/packages/agent/extensions"
	10	+	"github.com/bnema/zut/packages/agent/scheduler"
	11	+	"github.com/bnema/zut/packages/core"
	12	+	"github.com/bnema/zut/packages/provider"
	13	+	)
	14	+
	15	+	// runScheduledSession reconstructs an inactive session in the current zut
	16	+	// process and appends the resulting turn to its existing transcript. The
	17	+	// caller serializes this with session transitions before invoking it.
	18	+	func runScheduledSession(ctx context.Context, task scheduler.Task, args Args, base Resolved, extMgr *extensions.Manager, prepareRegistry func(core.Registry) core.Registry) error {
	19	+	path, err := core.FindManagedSessionByID(ctx, ZutHome(), task.SessionID)
	20	+	if err != nil {
	21	+	return fmt.Errorf("find scheduled session %q: %w", task.SessionID, err)
	22	+	}
	23	+	if path == "" {
	24	+	return fmt.Errorf("scheduled session %q no longer exists", task.SessionID)
	25	+	}
	26	+	sess, messages, err := core.OpenSession(path)
	27	+	if err != nil {
	28	+	return fmt.Errorf("open scheduled session: %w", err)
	29	+	}
	30	+	defer sess.Close()
	31	+
	32	+	providerName := strings.TrimSpace(sess.Meta.Provider)
	33	+	if providerName == "" {
	34	+	providerName = base.Provider
	35	+	}
	36	+	model := strings.TrimSpace(sess.Meta.Model)
	37	+	if model == "" {
	38	+	model = base.Model
	39	+	}
	40	+	ag, _, _, err := buildNonInteractiveSessionAgentWithRegistry(ctx, args, base, extMgr, providerName, model, prepareRegistry)
	41	+	if err != nil {
	42	+	return fmt.Errorf("build scheduled session agent: %w", err)
	43	+	}
	44	+	ag.SetMessages(trimMessagesForResume(messages, 100))
	45	+	if cumulative, lastTurn, usageErr := core.SessionUsageDetail(path); usageErr == nil {
	46	+	ag.SeedCost(cumulative)
	47	+	ag.SeedLastTurnUsage(lastTurn)
	48	+	}
	49	+	ag.SetSessionTimeContext(sess.Meta.Started, sess.Meta.Timezone, sess.Meta.TimezoneOffset)
	50	+	var appendMu sync.Mutex
	51	+	var appendErr error
	52	+	recordAppendError := func(err error) {
	53	+	if err == nil {
	54	+	return
	55	+	}
	56	+	appendMu.Lock()
	57	+	defer appendMu.Unlock()
	58	+	if appendErr == nil {
	59	+	appendErr = err
	60	+	}
	61	+	}
	62	+	ag.OnMessageAppended = func(message provider.Message) {
	63	+	recordAppendError(sess.AppendMessage(message))
	64	+	}
	65	+	ag.OnUsage = func(cumulative provider.Usage) {
	66	+	recordAppendError(sess.AppendUsage(cumulative, cumulative))
	67	+	}
	68	+	ag.OnTranscriptCompacted = func(messages []provider.Message) {
	69	+	recordAppendError(sess.AppendCompaction(messages))
	70	+	}
	71	+	if err := ag.Prompt(ctx, task.Message, nil, func(core.AgentEvent) {}); err != nil {
	72	+	return err
	73	+	}
	74	+	appendMu.Lock()
	75	+	err = appendErr
	76	+	appendMu.Unlock()
	77	+	if err != nil {
	78	+	return fmt.Errorf("append scheduled transcript: %w", err)
	79	+	}
	80	+	if err := sess.Flush(); err != nil {
	81	+	return fmt.Errorf("flush scheduled session: %w", err)
	82	+	}
	83	+	return nil
	84	+	}

A packages/agent/tools/schedule.go +122 -0

@@ -0,0 +1,122 @@
	1	+	package tools
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"encoding/json"
	7	+	"fmt"
	8	+	"io"
	9	+	"strings"
	10	+	"time"
	11	+
	12	+	"github.com/bnema/zut/packages/agent/scheduler"
	13	+	"github.com/bnema/zut/packages/core"
	14	+	"github.com/bnema/zut/packages/provider"
	15	+	)
	16	+
	17	+	const scheduleToolSchema = `{"type":"object","properties":{"action":{"type":"string","enum":["add","list","cancel"],"description":"Scheduling operation."},"cron":{"type":"string","description":"Five-field cron expression, for example '0 9 * * 1-5'. Required for add."},"message":{"type":"string","description":"Prompt to run in this session when due. Required for add."},"id":{"type":"string","description":"Task ID. Required for cancel."}},"required":["action"],"additionalProperties":false}`
	18	+
	19	+	// ScheduleToolName is the stable built-in scheduler tool name.
	20	+	const ScheduleToolName = "schedule"
	21	+
	22	+	// ScheduleTool manages process-lifetime scheduled follow-up prompts. Its
	23	+	// host-provided session callback prevents an agent from selecting another
	24	+	// session as a task target.
	25	+	type ScheduleTool struct {
	26	+	Engine *scheduler.Engine
	27	+	SessionID func() string
	28	+	Location func() *time.Location
	29	+	}
	30	+
	31	+	type scheduleToolArgs struct {
	32	+	Action string `json:"action"`
	33	+	Cron string `json:"cron,omitempty"`
	34	+	Message string `json:"message,omitempty"`
	35	+	ID string `json:"id,omitempty"`
	36	+	}
	37	+
	38	+	func (t *ScheduleTool) Name() string { return ScheduleToolName }
	39	+
	40	+	func (t *ScheduleTool) Description() string {
	41	+	return "Create, list, or cancel in-process cron follow-ups for the current session. Scheduled work runs only while this zut process remains open."
	42	+	}
	43	+
	44	+	func (t *ScheduleTool) Schema() json.RawMessage { return json.RawMessage(scheduleToolSchema) }
	45	+
	46	+	func (t *ScheduleTool) Execute(_ context.Context, raw json.RawMessage, _ func(string)) (core.ToolResult, error) {
	47	+	if t == nil \|\| t.Engine == nil \|\| t.SessionID == nil {
	48	+	return scheduleToolError("scheduler is unavailable"), nil
	49	+	}
	50	+	var args scheduleToolArgs
	51	+	decoder := json.NewDecoder(bytes.NewReader(raw))
	52	+	decoder.DisallowUnknownFields()
	53	+	if err := decoder.Decode(&args); err != nil {
	54	+	return scheduleToolError("invalid arguments"), nil
	55	+	}
	56	+	if err := decoder.Decode(&struct{}{}); err != io.EOF {
	57	+	return scheduleToolError("invalid arguments"), nil
	58	+	}
	59	+	args.Action = strings.TrimSpace(args.Action)
	60	+	args.Cron = strings.TrimSpace(args.Cron)
	61	+	args.Message = strings.TrimSpace(args.Message)
	62	+	args.ID = strings.TrimSpace(args.ID)
	63	+
	64	+	switch args.Action {
	65	+	case "add":
	66	+	if args.Cron == "" \|\| args.Message == "" {
	67	+	return scheduleToolError("cron and message are required when adding a task"), nil
	68	+	}
	69	+	location := time.Local
	70	+	if t.Location != nil && t.Location() != nil {
	71	+	location = t.Location()
	72	+	}
	73	+	task, err := t.Engine.Add(scheduler.NewTaskInput{
	74	+	SessionID: t.SessionID(),
	75	+	Cron: args.Cron,
	76	+	Message: args.Message,
	77	+	Location: location,
	78	+	})
	79	+	if err != nil {
	80	+	return scheduleToolError(err.Error()), nil
	81	+	}
	82	+	return scheduleToolResult(fmt.Sprintf("scheduled %s for %s (%s)", task.ID, task.NextRun.Format(time.RFC3339), task.Timezone))
	83	+	case "list":
	84	+	tasks := t.Engine.List()
	85	+	currentSession := t.SessionID()
	86	+	lines := make([]string, 0, len(tasks))
	87	+	for _, task := range tasks {
	88	+	if task.SessionID != currentSession {
	89	+	continue
	90	+	}
	91	+	line := fmt.Sprintf("%s %s next %s", task.ID, task.Cron, task.NextRun.Format(time.RFC3339))
	92	+	if task.LastError != "" {
	93	+	line += " last error: " + task.LastError
	94	+	}
	95	+	lines = append(lines, line)
	96	+	}
	97	+	if len(lines) == 0 {
	98	+	return scheduleToolResult("no scheduled tasks for this session")
	99	+	}
	100	+	return scheduleToolResult(strings.Join(lines, "\n"))
	101	+	case "cancel":
	102	+	if args.ID == "" {
	103	+	return scheduleToolError("id is required when cancelling a task"), nil
	104	+	}
	105	+	task, ok := t.Engine.Get(args.ID)
	106	+	if !ok \|\| task.SessionID != t.SessionID() {
	107	+	return scheduleToolError("scheduled task not found in this session"), nil
	108	+	}
	109	+	t.Engine.Cancel(args.ID)
	110	+	return scheduleToolResult("cancelled scheduled task " + args.ID)
	111	+	default:
	112	+	return scheduleToolError("action must be add, list, or cancel"), nil
	113	+	}
	114	+	}
	115	+
	116	+	func scheduleToolResult(text string) (core.ToolResult, error) {
	117	+	return core.ToolResult{Content: []provider.Content{provider.TextBlock{Text: text}}}, nil
	118	+	}
	119	+
	120	+	func scheduleToolError(text string) core.ToolResult {
	121	+	return core.ToolResult{Content: []provider.Content{provider.TextBlock{Text: text}}, IsError: true}
	122	+	}

@@ -0,0 +1,122 @@
	1	+	package tools
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"encoding/json"
	7	+	"fmt"
	8	+	"io"
	9	+	"strings"
	10	+	"time"
	11	+
	12	+	"github.com/bnema/zut/packages/agent/scheduler"
	13	+	"github.com/bnema/zut/packages/core"
	14	+	"github.com/bnema/zut/packages/provider"
	15	+	)
	16	+
	17	+	const scheduleToolSchema = `{"type":"object","properties":{"action":{"type":"string","enum":["add","list","cancel"],"description":"Scheduling operation."},"cron":{"type":"string","description":"Five-field cron expression, for example '0 9 * * 1-5'. Required for add."},"message":{"type":"string","description":"Prompt to run in this session when due. Required for add."},"id":{"type":"string","description":"Task ID. Required for cancel."}},"required":["action"],"additionalProperties":false}`
	18	+
	19	+	// ScheduleToolName is the stable built-in scheduler tool name.
	20	+	const ScheduleToolName = "schedule"
	21	+
	22	+	// ScheduleTool manages process-lifetime scheduled follow-up prompts. Its
	23	+	// host-provided session callback prevents an agent from selecting another
	24	+	// session as a task target.
	25	+	type ScheduleTool struct {
	26	+	Engine *scheduler.Engine
	27	+	SessionID func() string
	28	+	Location func() *time.Location
	29	+	}
	30	+
	31	+	type scheduleToolArgs struct {
	32	+	Action string `json:"action"`
	33	+	Cron string `json:"cron,omitempty"`
	34	+	Message string `json:"message,omitempty"`
	35	+	ID string `json:"id,omitempty"`
	36	+	}
	37	+
	38	+	func (t *ScheduleTool) Name() string { return ScheduleToolName }
	39	+
	40	+	func (t *ScheduleTool) Description() string {
	41	+	return "Create, list, or cancel in-process cron follow-ups for the current session. Scheduled work runs only while this zut process remains open."
	42	+	}
	43	+
	44	+	func (t *ScheduleTool) Schema() json.RawMessage { return json.RawMessage(scheduleToolSchema) }
	45	+
	46	+	func (t *ScheduleTool) Execute(_ context.Context, raw json.RawMessage, _ func(string)) (core.ToolResult, error) {
	47	+	if t == nil \|\| t.Engine == nil \|\| t.SessionID == nil {
	48	+	return scheduleToolError("scheduler is unavailable"), nil
	49	+	}
	50	+	var args scheduleToolArgs
	51	+	decoder := json.NewDecoder(bytes.NewReader(raw))
	52	+	decoder.DisallowUnknownFields()
	53	+	if err := decoder.Decode(&args); err != nil {
	54	+	return scheduleToolError("invalid arguments"), nil
	55	+	}
	56	+	if err := decoder.Decode(&struct{}{}); err != io.EOF {
	57	+	return scheduleToolError("invalid arguments"), nil
	58	+	}
	59	+	args.Action = strings.TrimSpace(args.Action)
	60	+	args.Cron = strings.TrimSpace(args.Cron)
	61	+	args.Message = strings.TrimSpace(args.Message)
	62	+	args.ID = strings.TrimSpace(args.ID)
	63	+
	64	+	switch args.Action {
	65	+	case "add":
	66	+	if args.Cron == "" \|\| args.Message == "" {
	67	+	return scheduleToolError("cron and message are required when adding a task"), nil
	68	+	}
	69	+	location := time.Local
	70	+	if t.Location != nil && t.Location() != nil {
	71	+	location = t.Location()
	72	+	}
	73	+	task, err := t.Engine.Add(scheduler.NewTaskInput{
	74	+	SessionID: t.SessionID(),
	75	+	Cron: args.Cron,
	76	+	Message: args.Message,
	77	+	Location: location,
	78	+	})
	79	+	if err != nil {
	80	+	return scheduleToolError(err.Error()), nil
	81	+	}
	82	+	return scheduleToolResult(fmt.Sprintf("scheduled %s for %s (%s)", task.ID, task.NextRun.Format(time.RFC3339), task.Timezone))
	83	+	case "list":
	84	+	tasks := t.Engine.List()
	85	+	currentSession := t.SessionID()
	86	+	lines := make([]string, 0, len(tasks))
	87	+	for _, task := range tasks {
	88	+	if task.SessionID != currentSession {
	89	+	continue
	90	+	}
	91	+	line := fmt.Sprintf("%s %s next %s", task.ID, task.Cron, task.NextRun.Format(time.RFC3339))
	92	+	if task.LastError != "" {
	93	+	line += " last error: " + task.LastError
	94	+	}
	95	+	lines = append(lines, line)
	96	+	}
	97	+	if len(lines) == 0 {
	98	+	return scheduleToolResult("no scheduled tasks for this session")
	99	+	}
	100	+	return scheduleToolResult(strings.Join(lines, "\n"))
	101	+	case "cancel":
	102	+	if args.ID == "" {
	103	+	return scheduleToolError("id is required when cancelling a task"), nil
	104	+	}
	105	+	task, ok := t.Engine.Get(args.ID)
	106	+	if !ok \|\| task.SessionID != t.SessionID() {
	107	+	return scheduleToolError("scheduled task not found in this session"), nil
	108	+	}
	109	+	t.Engine.Cancel(args.ID)
	110	+	return scheduleToolResult("cancelled scheduled task " + args.ID)
	111	+	default:
	112	+	return scheduleToolError("action must be add, list, or cancel"), nil
	113	+	}
	114	+	}
	115	+
	116	+	func scheduleToolResult(text string) (core.ToolResult, error) {
	117	+	return core.ToolResult{Content: []provider.Content{provider.TextBlock{Text: text}}}, nil
	118	+	}
	119	+
	120	+	func scheduleToolError(text string) core.ToolResult {
	121	+	return core.ToolResult{Content: []provider.Content{provider.TextBlock{Text: text}}, IsError: true}
	122	+	}

A packages/agent/tools/schedule_test.go +49 -0

@@ -0,0 +1,49 @@
	1	+	package tools
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/zut/packages/agent/scheduler"
	10	+	"github.com/bnema/zut/packages/core"
	11	+	"github.com/bnema/zut/packages/provider"
	12	+	)
	13	+
	14	+	func TestScheduleToolScopesTasksToCurrentSession(t *testing.T) {
	15	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	16	+	engine := scheduler.NewEngine(func() time.Time { return now })
	17	+	tool := &ScheduleTool{
	18	+	Engine: engine,
	19	+	SessionID: func() string { return "session-a" },
	20	+	Location: func() *time.Location { return time.UTC },
	21	+	}
	22	+
	23	+	ctx := context.Background()
	24	+	result, err := tool.Execute(ctx, json.RawMessage(`{"action":"add","cron":"@hourly","message":"check build"}`), nil)
	25	+	if err != nil \|\| result.IsError {
	26	+	t.Fatalf("Execute(add) = %#v, %v", result, err)
	27	+	}
	28	+	if got := resultText(result); got == "" {
	29	+	t.Fatal("Execute(add) returned no confirmation")
	30	+	}
	31	+
	32	+	otherSession, err := engine.Add(scheduler.NewTaskInput{SessionID: "session-b", Cron: "@hourly", Message: "hidden", Location: time.UTC})
	33	+	if err != nil {
	34	+	t.Fatalf("Engine.Add() error = %v", err)
	35	+	}
	36	+	result, err = tool.Execute(ctx, json.RawMessage(`{"action":"cancel","id":"`+otherSession.ID+`"}`), nil)
	37	+	if err != nil \|\| !result.IsError {
	38	+	t.Fatalf("Execute(cancel other session) = %#v, %v; want scoped error", result, err)
	39	+	}
	40	+	}
	41	+
	42	+	func resultText(result core.ToolResult) string {
	43	+	for _, content := range result.Content {
	44	+	if text, ok := content.(provider.TextBlock); ok {
	45	+	return text.Text
	46	+	}
	47	+	}
	48	+	return ""
	49	+	}

@@ -0,0 +1,49 @@
	1	+	package tools
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/zut/packages/agent/scheduler"
	10	+	"github.com/bnema/zut/packages/core"
	11	+	"github.com/bnema/zut/packages/provider"
	12	+	)
	13	+
	14	+	func TestScheduleToolScopesTasksToCurrentSession(t *testing.T) {
	15	+	now := time.Date(2026, time.August, 15, 12, 0, 0, 0, time.UTC)
	16	+	engine := scheduler.NewEngine(func() time.Time { return now })
	17	+	tool := &ScheduleTool{
	18	+	Engine: engine,
	19	+	SessionID: func() string { return "session-a" },
	20	+	Location: func() *time.Location { return time.UTC },
	21	+	}
	22	+
	23	+	ctx := context.Background()
	24	+	result, err := tool.Execute(ctx, json.RawMessage(`{"action":"add","cron":"@hourly","message":"check build"}`), nil)
	25	+	if err != nil \|\| result.IsError {
	26	+	t.Fatalf("Execute(add) = %#v, %v", result, err)
	27	+	}
	28	+	if got := resultText(result); got == "" {
	29	+	t.Fatal("Execute(add) returned no confirmation")
	30	+	}
	31	+
	32	+	otherSession, err := engine.Add(scheduler.NewTaskInput{SessionID: "session-b", Cron: "@hourly", Message: "hidden", Location: time.UTC})
	33	+	if err != nil {
	34	+	t.Fatalf("Engine.Add() error = %v", err)
	35	+	}
	36	+	result, err = tool.Execute(ctx, json.RawMessage(`{"action":"cancel","id":"`+otherSession.ID+`"}`), nil)
	37	+	if err != nil \|\| !result.IsError {
	38	+	t.Fatalf("Execute(cancel other session) = %#v, %v; want scoped error", result, err)
	39	+	}
	40	+	}
	41	+
	42	+	func resultText(result core.ToolResult) string {
	43	+	for _, content := range result.Content {
	44	+	if text, ok := content.(provider.TextBlock); ok {
	45	+	return text.Text
	46	+	}
	47	+	}
	48	+	return ""
	49	+	}

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)