| 1 | package builtin |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "regexp" |
| 8 | "strings" |
| 9 | |
| 10 | "reasonix/internal/evidence" |
| 11 | "reasonix/internal/jobs" |
| 12 | "reasonix/internal/planmode" |
| 13 | "reasonix/internal/tool" |
| 14 | ) |
| 15 | |
| 16 | // bash_output / kill_shell / wait operate the background jobs registered by |
| 17 | // bash(run_in_background) and task(run_in_background). They reach the session's |
| 18 | // job manager through the call context (jobs.FromContext) — the agent stamps it |
| 19 | // onto every tool call — and degrade to a clear error when it isn't available |
| 20 | // (a headless context with no manager). Together they poll a job's new output, |
| 21 | // terminate a job, and block until jobs finish. |
| 22 | |
| 23 | func init() { |
| 24 | tool.RegisterBuiltin(bashOutput{}) |
| 25 | tool.RegisterBuiltin(killShell{}) |
| 26 | tool.RegisterBuiltin(waitJob{}) |
| 27 | } |
| 28 | |
| 29 | // --- bash_output: poll a background job's new output (non-blocking) --- |
| 30 | |
| 31 | type bashOutput struct{} |
| 32 | |
| 33 | func (bashOutput) Name() string { return "bash_output" } |
| 34 | |
| 35 | func (bashOutput) Description() string { |
| 36 | return "Read new output from a background job started with bash(run_in_background=true) or task(run_in_background=true). Returns the output produced since the last bash_output call for that job, plus its status (running/done/failed/killed). Does not block." |
| 37 | } |
| 38 | |
| 39 | func (bashOutput) Schema() json.RawMessage { |
| 40 | return json.RawMessage(`{"type":"object","properties":{"job_id":{"type":"string","description":"The background job id (e.g. \"bash-1\") returned when it was started."},"filter":{"type":"string","description":"Optional regular expression; only matching lines of the new output are returned."}},"required":["job_id"]}`) |
| 41 | } |
| 42 | |
| 43 | func (bashOutput) ReadOnly() bool { return true } |
| 44 | |
| 45 | func (bashOutput) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 46 | var p struct { |
| 47 | JobID string `json:"job_id"` |
| 48 | Filter string `json:"filter"` |
| 49 | } |
| 50 | if err := json.Unmarshal(args, &p); err != nil { |
| 51 | return "", fmt.Errorf("invalid args: %w", err) |
| 52 | } |
| 53 | if p.JobID == "" { |
| 54 | return "", fmt.Errorf("job_id is required") |
| 55 | } |
| 56 | jm, ok := jobs.FromContext(ctx) |
| 57 | if !ok { |
| 58 | return "", fmt.Errorf("background jobs are not available in this context") |
| 59 | } |
| 60 | text, status, found := jm.OutputForSession(jobs.SessionFromContext(ctx), p.JobID) |
| 61 | if !found { |
| 62 | return "", fmt.Errorf("no background job %q", p.JobID) |
| 63 | } |
| 64 | if status != jobs.Running { |
| 65 | collectBackgroundEvidence(ctx, jm, p.JobID) |
| 66 | } |
| 67 | if p.Filter != "" && text != "" { |
| 68 | filtered, err := filterLines(text, p.Filter) |
| 69 | if err != nil { |
| 70 | return "", err |
| 71 | } |
| 72 | text = filtered |
| 73 | } |
| 74 | header := fmt.Sprintf("[%s] %s", p.JobID, status) |
| 75 | if strings.TrimSpace(text) == "" { |
| 76 | return header + "\n(no new output)", nil |
| 77 | } |
| 78 | return header + "\n" + text, nil |
| 79 | } |
| 80 | |
| 81 | // filterLines keeps only the lines of s matching the regular expression re. |
| 82 | func filterLines(s, re string) (string, error) { |
| 83 | rx, err := regexp.Compile(re) |
| 84 | if err != nil { |
| 85 | return "", fmt.Errorf("invalid filter regexp: %w", err) |
| 86 | } |
| 87 | var keep []string |
| 88 | for _, line := range strings.Split(s, "\n") { |
| 89 | if rx.MatchString(line) { |
| 90 | keep = append(keep, line) |
| 91 | } |
| 92 | } |
| 93 | return strings.Join(keep, "\n"), nil |
| 94 | } |
| 95 | |
| 96 | // --- kill_shell: terminate a running background job --- |
| 97 | |
| 98 | type killShell struct{} |
| 99 | |
| 100 | func (killShell) Name() string { return "kill_shell" } |
| 101 | |
| 102 | func (killShell) Description() string { |
| 103 | return "Terminate a running background job (bash or task) started with run_in_background. A no-op if the job has already finished or the id is unknown." |
| 104 | } |
| 105 | |
| 106 | func (killShell) Schema() json.RawMessage { |
| 107 | return json.RawMessage(`{"type":"object","properties":{"job_id":{"type":"string","description":"The background job id to terminate (e.g. \"bash-1\")."}},"required":["job_id"]}`) |
| 108 | } |
| 109 | |
| 110 | func (killShell) ReadOnly() bool { return false } |
| 111 | |
| 112 | func (killShell) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 113 | var p struct { |
| 114 | JobID string `json:"job_id"` |
| 115 | } |
| 116 | if err := json.Unmarshal(args, &p); err != nil { |
| 117 | return "", fmt.Errorf("invalid args: %w", err) |
| 118 | } |
| 119 | if p.JobID == "" { |
| 120 | return "", fmt.Errorf("job_id is required") |
| 121 | } |
| 122 | jm, ok := jobs.FromContext(ctx) |
| 123 | if !ok { |
| 124 | return "", fmt.Errorf("background jobs are not available in this context") |
| 125 | } |
| 126 | if jm.KillForSession(jobs.SessionFromContext(ctx), p.JobID) { |
| 127 | return fmt.Sprintf("Killed background job %q.", p.JobID), nil |
| 128 | } |
| 129 | return fmt.Sprintf("Background job %q was not running (already finished or unknown).", p.JobID), nil |
| 130 | } |
| 131 | |
| 132 | // --- wait: block until background jobs finish, then return their results --- |
| 133 | |
| 134 | type waitJob struct{} |
| 135 | |
| 136 | func (waitJob) Name() string { return "wait" } |
| 137 | |
| 138 | func (waitJob) Description() string { |
| 139 | return "Block until background jobs finish, then return each job's status and final output/answer. Use to collect the result of a task(run_in_background) or bash(run_in_background) before continuing. Omit job_ids to wait for every running job." |
| 140 | } |
| 141 | |
| 142 | func (waitJob) Schema() json.RawMessage { |
| 143 | return json.RawMessage(`{"type":"object","properties":{"job_ids":{"type":"array","items":{"type":"string"},"description":"Background job ids to wait for. Omit to wait for every currently-running job."},"timeout_seconds":{"type":"integer","description":"Optional maximum seconds to block before returning current progress. Omit to wait until the jobs finish.","minimum":1}}}`) |
| 144 | } |
| 145 | |
| 146 | func (waitJob) ReadOnly() bool { return true } |
| 147 | |
| 148 | func (waitJob) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 149 | var p struct { |
| 150 | JobIDs []string `json:"job_ids"` |
| 151 | TimeoutSeconds int `json:"timeout_seconds"` |
| 152 | } |
| 153 | if len(args) > 0 { |
| 154 | if err := json.Unmarshal(args, &p); err != nil { |
| 155 | return "", fmt.Errorf("invalid args: %w", err) |
| 156 | } |
| 157 | } |
| 158 | jm, ok := jobs.FromContext(ctx) |
| 159 | if !ok { |
| 160 | return "", fmt.Errorf("background jobs are not available in this context") |
| 161 | } |
| 162 | results := jm.WaitForSession(ctx, jobs.SessionFromContext(ctx), p.JobIDs, p.TimeoutSeconds) |
| 163 | if len(results) == 0 { |
| 164 | return "No background jobs to wait for.", nil |
| 165 | } |
| 166 | var b strings.Builder |
| 167 | for i, r := range results { |
| 168 | if r.Status != jobs.Running { |
| 169 | collectBackgroundEvidence(ctx, jm, r.ID) |
| 170 | } |
| 171 | if i > 0 { |
| 172 | b.WriteString("\n\n") |
| 173 | } |
| 174 | label := r.ID |
| 175 | if r.Label != "" { |
| 176 | label = fmt.Sprintf("%s (%s)", r.ID, r.Label) |
| 177 | } |
| 178 | fmt.Fprintf(&b, "[%s] %s", label, r.Status) |
| 179 | if strings.TrimSpace(r.Output) != "" { |
| 180 | b.WriteString("\n" + r.Output) |
| 181 | } |
| 182 | } |
| 183 | return b.String(), nil |
| 184 | } |
| 185 | |
| 186 | func collectBackgroundEvidence(ctx context.Context, jm *jobs.Manager, jobID string) { |
| 187 | // A Plan turn should not consume a finished background writer's mutation |
| 188 | // receipts before the workflow reaches execution. Writers may still run after |
| 189 | // Permissions approval; leave their evidence on the job so the first |
| 190 | // post-approval collection can merge and audit it. |
| 191 | if planmode.Active(ctx) { |
| 192 | return |
| 193 | } |
| 194 | ledger, ok := evidence.FromContext(ctx) |
| 195 | if !ok || ledger == nil || jm == nil { |
| 196 | return |
| 197 | } |
| 198 | session := jobs.SessionFromContext(ctx) |
| 199 | // A non-Running status from bash_output/wait does not guarantee the job's |
| 200 | // run goroutine has actually flushed PublishEvidence and closed done: kill_shell |
| 201 | // flips status to Killed synchronously, well before its cancelled goroutine |
| 202 | // unwinds. Check readiness before noting the lease — noting it on an empty, |
| 203 | // not-yet-ready read would dedupe away every later retry in this turn (the |
| 204 | // lease is idempotent per turn) while the job later publishes real mutation |
| 205 | // evidence nobody ever merges or reviews. |
| 206 | summary, ready := jm.TryLeaseEvidenceForSession(session, jobID) |
| 207 | if !ready { |
| 208 | return |
| 209 | } |
| 210 | // Note the lease before merging so a second wait/bash_output in the same |
| 211 | // turn does not double-count. The merge is provisional: the lease does not |
| 212 | // consume, so if this turn fails the agent never commits and the next turn |
| 213 | // re-collects. The agent commits leased jobs only after the turn passes its |
| 214 | // delivery gates. |
| 215 | if !ledger.NoteBackgroundLease(session, jobID) { |
| 216 | return |
| 217 | } |
| 218 | ledger.MergeChild(summary) |
| 219 | } |
| 220 |