1package agent
2
3import (
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.
18func 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 if sess.ID != task.SessionID {
31 closeErr := sess.Close()
32 if closeErr != nil {
33 return fmt.Errorf("scheduled session %q changed during resume: %w", task.SessionID, closeErr)
34 }
35 return fmt.Errorf("scheduled session %q changed during resume", task.SessionID)
36 }
37 defer sess.Close()
38
39 providerName := strings.TrimSpace(sess.Meta.Provider)
40 if providerName == "" {
41 providerName = base.Provider
42 }
43 model := strings.TrimSpace(sess.Meta.Model)
44 if model == "" {
45 model = base.Model
46 }
47 ag, _, _, err := buildNonInteractiveSessionAgentWithRegistry(ctx, args, base, extMgr, providerName, model, prepareRegistry)
48 if err != nil {
49 return fmt.Errorf("build scheduled session agent: %w", err)
50 }
51 ag.SetMessages(trimMessagesForResume(messages, 100))
52 // Seed the agent-owned plan from the reconstructed session so a scheduled
53 // turn sees the persisted checklist instead of empty core state.
54 ag.SetPlan(sess.Meta.PlanSteps())
55 // Route tool-result commits through the session so plan (and goal)
56 // mutations the scheduled turn makes are durable, matching the other modes.
57 ag.CommitToolResult = func(_ string, result core.ToolResult) error {
58 return persistToolResultState(nil, sess, result)
59 }
60 if cumulative, lastTurn, usageErr := core.SessionUsageDetail(path); usageErr == nil {
61 ag.SeedCost(cumulative)
62 ag.SeedLastTurnUsage(lastTurn)
63 }
64 ag.SetSessionTimeContext(sess.Meta.Started, sess.Meta.Timezone, sess.Meta.TimezoneOffset)
65 var appendMu sync.Mutex
66 var appendErr error
67 recordAppendError := func(err error) {
68 if err == nil {
69 return
70 }
71 appendMu.Lock()
72 defer appendMu.Unlock()
73 if appendErr == nil {
74 appendErr = err
75 }
76 }
77 ag.OnMessageAppended = func(message provider.Message) {
78 recordAppendError(sess.AppendMessage(message))
79 }
80 ag.OnUsage = func(cumulative provider.Usage) {
81 recordAppendError(sess.AppendUsage(cumulative, cumulative))
82 }
83 ag.OnTranscriptCompacted = func(messages []provider.Message) {
84 recordAppendError(sess.AppendCompaction(messages))
85 }
86 if err := ag.Prompt(ctx, task.Message, nil, func(core.AgentEvent) {}); err != nil {
87 return err
88 }
89 appendMu.Lock()
90 err = appendErr
91 appendMu.Unlock()
92 if err != nil {
93 return fmt.Errorf("append scheduled transcript: %w", err)
94 }
95 if err := sess.Flush(); err != nil {
96 return fmt.Errorf("flush scheduled session: %w", err)
97 }
98 return nil
99}