| 1 | package builtin |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "regexp" |
| 8 | "strings" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/evidence" |
| 12 | "reasonix/internal/jobs" |
| 13 | "reasonix/internal/planmode" |
| 14 | "reasonix/internal/tool" |
| 15 | ) |
| 16 | |
| 17 | // job_output / job_kill operate background jobs registered by shell and task |
| 18 | // run_in_background calls; legacy aliases remain for replay. They reach the session's |
| 19 | // job manager through the call context (jobs.FromContext) — the agent stamps it |
| 20 | // onto every tool call — and degrade to a clear error when it isn't available |
| 21 | // (a headless context with no manager). Together they poll a job's new output, |
| 22 | // terminate a job, and block until jobs finish. |
| 23 | |
| 24 | func init() { |
| 25 | tool.RegisterBuiltin(jobOutput{}) |
| 26 | tool.RegisterBuiltin(jobKill{}) |
| 27 | tool.RegisterBuiltin(bashOutput{}) |
| 28 | tool.RegisterBuiltin(killShell{}) |
| 29 | tool.RegisterBuiltin(waitJob{}) |
| 30 | } |
| 31 | |
| 32 | const ( |
| 33 | jobOutputDefaultWait = 30 * time.Second |
| 34 | jobOutputMaxWait = 10 * time.Minute |
| 35 | ) |
| 36 | |
| 37 | // job_output is the provider-facing Harness-compatible job reader. Legacy |
| 38 | // bash_output and wait remain registered for replaying older sessions. |
| 39 | type jobOutput struct{} |
| 40 | |
| 41 | func (jobOutput) Name() string { return "job_output" } |
| 42 | |
| 43 | func (jobOutput) Description() string { |
| 44 | return "Read output from a background job. Reads are non-blocking unless wait=true; every response includes the current status. Do not busy-poll a running job." |
| 45 | } |
| 46 | |
| 47 | func (jobOutput) Schema() json.RawMessage { |
| 48 | return json.RawMessage(`{"type":"object","properties":{"job_id":{"type":"string","description":"Job id returned by the tool that started the background work."},"wait":{"type":"boolean","description":"Wait until the job finishes or timeout_ms elapses. A timeout leaves the job running."},"timeout_ms":{"type":"integer","minimum":1,"maximum":600000,"description":"Maximum wait in milliseconds. Defaults to 30000 and is capped at 600000."},"filter":{"type":"string","description":"Optional regular expression; only matching lines of new output are returned."}},"required":["job_id"]}`) |
| 49 | } |
| 50 | |
| 51 | func (jobOutput) ReadOnly() bool { return true } |
| 52 | |
| 53 | func (jobOutput) ProviderVisible(ctx context.Context) bool { |
| 54 | _, ok := jobs.FromContext(ctx) |
| 55 | return ok |
| 56 | } |
| 57 | |
| 58 | func (jobOutput) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 59 | result, err := (jobOutput{}).ExecuteDetailed(ctx, args) |
| 60 | return result.Output, err |
| 61 | } |
| 62 | |
| 63 | func (jobOutput) ExecutionDescriptor(json.RawMessage) *tool.ShellExecution { return nil } |
| 64 | |
| 65 | func (jobOutput) ExecuteDetailed(ctx context.Context, args json.RawMessage) (tool.DetailedResult, error) { |
| 66 | var p struct { |
| 67 | JobID string `json:"job_id"` |
| 68 | Wait bool `json:"wait"` |
| 69 | TimeoutMS int `json:"timeout_ms"` |
| 70 | Filter string `json:"filter"` |
| 71 | } |
| 72 | if err := json.Unmarshal(args, &p); err != nil { |
| 73 | return tool.DetailedResult{}, fmt.Errorf("invalid args: %w", err) |
| 74 | } |
| 75 | p.JobID = strings.TrimSpace(p.JobID) |
| 76 | if p.JobID == "" { |
| 77 | return tool.DetailedResult{}, fmt.Errorf("job_id is required") |
| 78 | } |
| 79 | if p.TimeoutMS < 0 { |
| 80 | return tool.DetailedResult{}, fmt.Errorf("timeout_ms must be positive") |
| 81 | } |
| 82 | // Validate before waiting or consuming the manager's incremental cursor. |
| 83 | if _, err := regexp.Compile(p.Filter); err != nil { |
| 84 | return tool.DetailedResult{}, fmt.Errorf("invalid filter: %w", err) |
| 85 | } |
| 86 | jm, ok := jobs.FromContext(ctx) |
| 87 | if !ok { |
| 88 | return tool.DetailedResult{}, fmt.Errorf("background jobs are not available in this context") |
| 89 | } |
| 90 | session := jobs.SessionFromContext(ctx) |
| 91 | if p.Wait { |
| 92 | waitFor := jobOutputDefaultWait |
| 93 | if p.TimeoutMS > 0 { |
| 94 | waitFor = cappedMilliseconds(p.TimeoutMS, jobOutputMaxWait) |
| 95 | } |
| 96 | if waitFor > jobOutputMaxWait { |
| 97 | waitFor = jobOutputMaxWait |
| 98 | } |
| 99 | waitCtx, cancel := context.WithTimeout(ctx, waitFor) |
| 100 | _ = jm.WaitForSession(waitCtx, session, []string{p.JobID}, 0) |
| 101 | cancel() |
| 102 | } |
| 103 | text, status, found := jm.OutputForSession(session, p.JobID) |
| 104 | if !found { |
| 105 | return tool.DetailedResult{}, fmt.Errorf("no background job %q", p.JobID) |
| 106 | } |
| 107 | if status != jobs.Running { |
| 108 | collectBackgroundEvidence(ctx, jm, p.JobID) |
| 109 | } |
| 110 | if p.Filter != "" && text != "" { |
| 111 | filtered, err := filterLines(text, p.Filter) |
| 112 | if err != nil { |
| 113 | return tool.DetailedResult{}, err |
| 114 | } |
| 115 | text = filtered |
| 116 | } |
| 117 | body := strings.TrimRight(text, "\n") |
| 118 | if strings.TrimSpace(body) == "" { |
| 119 | body = "(no new output)" |
| 120 | } |
| 121 | return tool.DetailedResult{ |
| 122 | Output: fmt.Sprintf("%s\n[status: %s]", body, status), |
| 123 | Execution: jm.ExecutionForSession(session, p.JobID), |
| 124 | }, nil |
| 125 | } |
| 126 | |
| 127 | // job_kill is the provider-facing Harness-compatible cancellation tool. |
| 128 | type jobKill struct{} |
| 129 | |
| 130 | func (jobKill) Name() string { return "job_kill" } |
| 131 | |
| 132 | func (jobKill) Description() string { |
| 133 | return "Request cancellation of a running background job by job id. Returns immediately; the process tree settles as killed once shutdown completes." |
| 134 | } |
| 135 | |
| 136 | func (jobKill) Schema() json.RawMessage { |
| 137 | return json.RawMessage(`{"type":"object","properties":{"job_id":{"type":"string","description":"Job id returned by the tool that started the background work."},"reason":{"type":"string","description":"Optional short reason for stopping the job."}},"required":["job_id"]}`) |
| 138 | } |
| 139 | |
| 140 | func (jobKill) ReadOnly() bool { return false } |
| 141 | |
| 142 | func (jobKill) ProviderVisible(ctx context.Context) bool { |
| 143 | _, ok := jobs.FromContext(ctx) |
| 144 | return ok |
| 145 | } |
| 146 | |
| 147 | func (jobKill) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 148 | var p struct { |
| 149 | JobID string `json:"job_id"` |
| 150 | Reason string `json:"reason"` |
| 151 | } |
| 152 | if err := json.Unmarshal(args, &p); err != nil { |
| 153 | return "", fmt.Errorf("invalid args: %w", err) |
| 154 | } |
| 155 | p.JobID = strings.TrimSpace(p.JobID) |
| 156 | if p.JobID == "" { |
| 157 | return "", fmt.Errorf("job_id is required") |
| 158 | } |
| 159 | jm, ok := jobs.FromContext(ctx) |
| 160 | if !ok { |
| 161 | return "", fmt.Errorf("background jobs are not available in this context") |
| 162 | } |
| 163 | if jm.KillForSession(jobs.SessionFromContext(ctx), p.JobID) { |
| 164 | return fmt.Sprintf("Requested cancellation of job %q.\n[status: killed]", p.JobID), nil |
| 165 | } |
| 166 | return fmt.Sprintf("Job %q had already finished or is unknown.", p.JobID), nil |
| 167 | } |
| 168 | |
| 169 | // bash_output: poll a background job's new output (non-blocking) |
| 170 | |
| 171 | type bashOutput struct{} |
| 172 | |
| 173 | func (bashOutput) Name() string { return "bash_output" } |
| 174 | |
| 175 | func (bashOutput) Description() string { |
| 176 | return "Legacy alias: read new output from a background job without blocking. New calls should use job_output." |
| 177 | } |
| 178 | |
| 179 | func (bashOutput) Schema() json.RawMessage { |
| 180 | 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"]}`) |
| 181 | } |
| 182 | |
| 183 | func (bashOutput) ReadOnly() bool { return true } |
| 184 | |
| 185 | func (bashOutput) HiddenFromCapabilityCatalog() bool { return true } |
| 186 | |
| 187 | func (bashOutput) ProviderVisible(ctx context.Context) bool { |
| 188 | _, ok := jobs.FromContext(ctx) |
| 189 | return ok |
| 190 | } |
| 191 | |
| 192 | func (bashOutput) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 193 | var p struct { |
| 194 | JobID string `json:"job_id"` |
| 195 | Filter string `json:"filter"` |
| 196 | } |
| 197 | if err := json.Unmarshal(args, &p); err != nil { |
| 198 | return "", fmt.Errorf("invalid args: %w", err) |
| 199 | } |
| 200 | if p.JobID == "" { |
| 201 | return "", fmt.Errorf("job_id is required") |
| 202 | } |
| 203 | if _, err := regexp.Compile(p.Filter); err != nil { |
| 204 | return "", fmt.Errorf("invalid filter regexp: %w", err) |
| 205 | } |
| 206 | jm, ok := jobs.FromContext(ctx) |
| 207 | if !ok { |
| 208 | return "", fmt.Errorf("background jobs are not available in this context") |
| 209 | } |
| 210 | text, status, found := jm.OutputForSession(jobs.SessionFromContext(ctx), p.JobID) |
| 211 | if !found { |
| 212 | return "", fmt.Errorf("no background job %q", p.JobID) |
| 213 | } |
| 214 | if status != jobs.Running { |
| 215 | collectBackgroundEvidence(ctx, jm, p.JobID) |
| 216 | } |
| 217 | if p.Filter != "" && text != "" { |
| 218 | filtered, err := filterLines(text, p.Filter) |
| 219 | if err != nil { |
| 220 | return "", err |
| 221 | } |
| 222 | text = filtered |
| 223 | } |
| 224 | header := fmt.Sprintf("[%s] %s", p.JobID, status) |
| 225 | if strings.TrimSpace(text) == "" { |
| 226 | return header + "\n(no new output)", nil |
| 227 | } |
| 228 | return header + "\n" + text, nil |
| 229 | } |
| 230 | |
| 231 | // filterLines keeps only the lines of s matching the regular expression re. |
| 232 | func filterLines(s, re string) (string, error) { |
| 233 | rx, err := regexp.Compile(re) |
| 234 | if err != nil { |
| 235 | return "", fmt.Errorf("invalid filter regexp: %w", err) |
| 236 | } |
| 237 | var keep []string |
| 238 | for line := range strings.SplitSeq(s, "\n") { |
| 239 | if rx.MatchString(line) { |
| 240 | keep = append(keep, line) |
| 241 | } |
| 242 | } |
| 243 | return strings.Join(keep, "\n"), nil |
| 244 | } |
| 245 | |
| 246 | // kill_shell: terminate a running background job |
| 247 | |
| 248 | type killShell struct{} |
| 249 | |
| 250 | func (killShell) Name() string { return "kill_shell" } |
| 251 | |
| 252 | func (killShell) Description() string { |
| 253 | return "Legacy alias: terminate a running background job. New calls should use job_kill." |
| 254 | } |
| 255 | |
| 256 | func (killShell) Schema() json.RawMessage { |
| 257 | return json.RawMessage(`{"type":"object","properties":{"job_id":{"type":"string","description":"The background job id to terminate (e.g. \"bash-1\")."}},"required":["job_id"]}`) |
| 258 | } |
| 259 | |
| 260 | func (killShell) ReadOnly() bool { return false } |
| 261 | |
| 262 | func (killShell) HiddenFromCapabilityCatalog() bool { return true } |
| 263 | |
| 264 | func (killShell) ProviderVisible(ctx context.Context) bool { |
| 265 | _, ok := jobs.FromContext(ctx) |
| 266 | return ok |
| 267 | } |
| 268 | |
| 269 | func (killShell) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 270 | var p struct { |
| 271 | JobID string `json:"job_id"` |
| 272 | } |
| 273 | if err := json.Unmarshal(args, &p); err != nil { |
| 274 | return "", fmt.Errorf("invalid args: %w", err) |
| 275 | } |
| 276 | if p.JobID == "" { |
| 277 | return "", fmt.Errorf("job_id is required") |
| 278 | } |
| 279 | jm, ok := jobs.FromContext(ctx) |
| 280 | if !ok { |
| 281 | return "", fmt.Errorf("background jobs are not available in this context") |
| 282 | } |
| 283 | if jm.KillForSession(jobs.SessionFromContext(ctx), p.JobID) { |
| 284 | return fmt.Sprintf("Killed background job %q.", p.JobID), nil |
| 285 | } |
| 286 | return fmt.Sprintf("Background job %q was not running (already finished or unknown).", p.JobID), nil |
| 287 | } |
| 288 | |
| 289 | // wait: block until background jobs finish, then return their results |
| 290 | |
| 291 | type waitJob struct{} |
| 292 | |
| 293 | func (waitJob) Name() string { return "wait" } |
| 294 | |
| 295 | func (waitJob) Description() string { |
| 296 | return "Legacy multi-job wait retained for old sessions. New calls should use job_output with wait=true." |
| 297 | } |
| 298 | |
| 299 | func (waitJob) Schema() json.RawMessage { |
| 300 | 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}}}`) |
| 301 | } |
| 302 | |
| 303 | func (waitJob) ReadOnly() bool { return true } |
| 304 | |
| 305 | func (waitJob) HiddenFromCapabilityCatalog() bool { return true } |
| 306 | |
| 307 | func (waitJob) ProviderVisible(ctx context.Context) bool { |
| 308 | _, ok := jobs.FromContext(ctx) |
| 309 | return ok |
| 310 | } |
| 311 | |
| 312 | func (waitJob) Execute(ctx context.Context, args json.RawMessage) (string, error) { |
| 313 | var p struct { |
| 314 | JobIDs []string `json:"job_ids"` |
| 315 | TimeoutSeconds int `json:"timeout_seconds"` |
| 316 | } |
| 317 | if len(args) > 0 { |
| 318 | if err := json.Unmarshal(args, &p); err != nil { |
| 319 | return "", fmt.Errorf("invalid args: %w", err) |
| 320 | } |
| 321 | } |
| 322 | jm, ok := jobs.FromContext(ctx) |
| 323 | if !ok { |
| 324 | return "", fmt.Errorf("background jobs are not available in this context") |
| 325 | } |
| 326 | results := jm.WaitForSession(ctx, jobs.SessionFromContext(ctx), p.JobIDs, p.TimeoutSeconds) |
| 327 | if len(results) == 0 { |
| 328 | return "No background jobs to wait for.", nil |
| 329 | } |
| 330 | var b strings.Builder |
| 331 | for i, r := range results { |
| 332 | if r.Status != jobs.Running { |
| 333 | collectBackgroundEvidence(ctx, jm, r.ID) |
| 334 | } |
| 335 | if i > 0 { |
| 336 | b.WriteString("\n\n") |
| 337 | } |
| 338 | label := r.ID |
| 339 | if r.Label != "" { |
| 340 | label = fmt.Sprintf("%s (%s)", r.ID, r.Label) |
| 341 | } |
| 342 | fmt.Fprintf(&b, "[%s] %s", label, r.Status) |
| 343 | if strings.TrimSpace(r.Output) != "" { |
| 344 | b.WriteString("\n" + r.Output) |
| 345 | } |
| 346 | } |
| 347 | return b.String(), nil |
| 348 | } |
| 349 | |
| 350 | func collectBackgroundEvidence(ctx context.Context, jm *jobs.Manager, jobID string) { |
| 351 | // A Plan turn should not consume a finished background writer's mutation |
| 352 | // receipts before the workflow reaches execution. Writers may still run after |
| 353 | // Permissions approval; leave their evidence on the job so the first |
| 354 | // post-approval collection can merge and audit it. |
| 355 | if planmode.Active(ctx) { |
| 356 | return |
| 357 | } |
| 358 | ledger, ok := evidence.FromContext(ctx) |
| 359 | if !ok || ledger == nil || jm == nil { |
| 360 | return |
| 361 | } |
| 362 | session := jobs.SessionFromContext(ctx) |
| 363 | // A non-Running status from bash_output/wait does not guarantee the job's |
| 364 | // run goroutine has actually flushed PublishEvidence and closed done: kill_shell |
| 365 | // flips status to Killed synchronously, well before its cancelled goroutine |
| 366 | // unwinds. Check readiness before noting the lease — noting it on an empty, |
| 367 | // not-yet-ready read would dedupe away every later retry in this turn (the |
| 368 | // lease is idempotent per turn) while the job later publishes real mutation |
| 369 | // evidence nobody ever merges or reviews. |
| 370 | summary, ready := jm.TryLeaseEvidenceForSession(session, jobID) |
| 371 | if !ready { |
| 372 | return |
| 373 | } |
| 374 | // Note the lease before merging so a second wait/bash_output in the same |
| 375 | // turn does not double-count. The merge is provisional: the lease does not |
| 376 | // consume, so if this turn fails the agent never commits and the next turn |
| 377 | // re-collects. The agent commits leased jobs only after the turn passes its |
| 378 | // delivery gates. |
| 379 | if !ledger.NoteBackgroundLease(session, jobID) { |
| 380 | return |
| 381 | } |
| 382 | ledger.MergeChild(summary) |
| 383 | } |
| 384 |