| 1 | // Package jobs is the session-scoped background-job registry behind the agent's |
| 2 | // background tools (bash run_in_background, task run_in_background) and the |
| 3 | // job_output / job_kill tools (plus replay-only legacy aliases). A Manager owns a context whose lifetime |
| 4 | // is the session, NOT a single turn — so a job started in one turn keeps running |
| 5 | // across turns and is cancelled only when the controller closes (or job_kill is |
| 6 | // called). Tools reach the Manager through the call context (WithManager / |
| 7 | // FromContext), the same injection pattern the `ask` tool uses for the asker. |
| 8 | // |
| 9 | // The Manager emits a user-visible Notice when a job starts and finishes, and |
| 10 | // accumulates a one-line completion summary that the controller drains into the |
| 11 | // next turn (DrainCompletedNote) so the model itself learns of completions. |
| 12 | package jobs |
| 13 | |
| 14 | import ( |
| 15 | "context" |
| 16 | "crypto/rand" |
| 17 | "encoding/hex" |
| 18 | "fmt" |
| 19 | "io" |
| 20 | "os" |
| 21 | "path/filepath" |
| 22 | "runtime/debug" |
| 23 | "slices" |
| 24 | "sort" |
| 25 | "strings" |
| 26 | "sync" |
| 27 | "sync/atomic" |
| 28 | "time" |
| 29 | |
| 30 | "reasonix/internal/event" |
| 31 | "reasonix/internal/evidence" |
| 32 | "reasonix/internal/nilutil" |
| 33 | "reasonix/internal/tool" |
| 34 | ) |
| 35 | |
| 36 | var renamePath = os.Rename |
| 37 | var repairArtifactMeta = writeMeta |
| 38 | |
| 39 | var ( |
| 40 | managerOwnerSeq atomic.Uint64 |
| 41 | liveManagerOwners = struct { |
| 42 | sync.RWMutex |
| 43 | ids map[string]struct{} |
| 44 | }{ids: map[string]struct{}{}} |
| 45 | ) |
| 46 | |
| 47 | // Status is a job's lifecycle state. |
| 48 | type Status string |
| 49 | |
| 50 | const ( |
| 51 | Running Status = "running" |
| 52 | Done Status = "done" |
| 53 | Failed Status = "failed" |
| 54 | Killed Status = "killed" |
| 55 | Interrupted Status = "interrupted" |
| 56 | ) |
| 57 | |
| 58 | // DefaultTeardownGrace bounds Close and destroy waits for non-cooperative jobs. |
| 59 | const DefaultTeardownGrace = 15 * time.Second |
| 60 | |
| 61 | // View is a read-only snapshot of a job for the status bar. |
| 62 | type View struct { |
| 63 | ID string `json:"id"` |
| 64 | Kind string `json:"kind"` |
| 65 | Label string `json:"label"` |
| 66 | Status string `json:"status"` |
| 67 | StartedAt int64 `json:"startedAt"` // unix milliseconds |
| 68 | } |
| 69 | |
| 70 | // Result is one job's terminal (or current) state returned by Wait. |
| 71 | type Result struct { |
| 72 | ID string |
| 73 | Kind string |
| 74 | Label string |
| 75 | Status Status |
| 76 | Output string // the terminal result text, or the streamed buffer when no result was set |
| 77 | } |
| 78 | |
| 79 | // TeardownJob identifies a job that is still unwinding after teardown waited. |
| 80 | type TeardownJob struct { |
| 81 | ID string |
| 82 | Kind string |
| 83 | Label string |
| 84 | Waited time.Duration |
| 85 | } |
| 86 | |
| 87 | // TeardownResult reports jobs that did not unwind within the teardown grace. |
| 88 | type TeardownResult struct { |
| 89 | TimedOut []TeardownJob |
| 90 | } |
| 91 | |
| 92 | type teardownTarget struct { |
| 93 | info TeardownJob |
| 94 | done <-chan struct{} |
| 95 | } |
| 96 | |
| 97 | // SessionTeardown is the destroy handle for a session's owned background jobs. |
| 98 | type SessionTeardown struct { |
| 99 | SessionID string |
| 100 | targets []teardownTarget |
| 101 | } |
| 102 | |
| 103 | // Async reports whether the handle has jobs to wait on. |
| 104 | func (h SessionTeardown) Async() bool { return len(h.targets) > 0 } |
| 105 | |
| 106 | // DoneChannels returns each target's completion channel for legacy callers. |
| 107 | func (h SessionTeardown) DoneChannels() []<-chan struct{} { |
| 108 | out := make([]<-chan struct{}, 0, len(h.targets)) |
| 109 | for _, target := range h.targets { |
| 110 | out = append(out, target.done) |
| 111 | } |
| 112 | return out |
| 113 | } |
| 114 | |
| 115 | // Job is one background job. The mutex guards the streaming buffer and the |
| 116 | // terminal fields; the run goroutine writes them, readers (Output/Wait/snapshots) |
| 117 | // take the same lock. |
| 118 | type Job struct { |
| 119 | ID string |
| 120 | Kind string // "bash" | "pwsh" | "task" |
| 121 | Label string |
| 122 | SessionID string |
| 123 | |
| 124 | mu sync.Mutex |
| 125 | tail []byte |
| 126 | readOffset int64 |
| 127 | status Status |
| 128 | clock jobClock |
| 129 | outcome jobOutcome |
| 130 | cancel context.CancelFunc |
| 131 | done chan struct{} |
| 132 | |
| 133 | artifactPath string |
| 134 | artifactMetaPath string |
| 135 | artifactStatus Status // last metadata phase; may precede published terminal status |
| 136 | artifactFile *os.File |
| 137 | artifactComplete bool |
| 138 | artifactErr string |
| 139 | tombstone bool |
| 140 | |
| 141 | evidence evidence.ChildEvidenceSummary |
| 142 | evidenceCommitted bool |
| 143 | execution *tool.ShellExecution |
| 144 | } |
| 145 | |
| 146 | // Manager is the session's background-job table. It is safe for concurrent use. |
| 147 | type Manager struct { |
| 148 | runtimeObservers runtimeObservers |
| 149 | sink event.Sink |
| 150 | root context.Context |
| 151 | cancel context.CancelFunc |
| 152 | wg sync.WaitGroup |
| 153 | onJobStart func(done <-chan struct{}) |
| 154 | ownerID string |
| 155 | ownerDone sync.Once |
| 156 | // sessionOwnershipProbe authorizes destructive repair of persisted running |
| 157 | // artifacts. A nil probe is conservative: an observer that cannot prove it |
| 158 | // owns the transcript must never publish an interrupted tombstone. |
| 159 | sessionOwnershipProbe func(path string) bool |
| 160 | |
| 161 | mu sync.Mutex |
| 162 | seq int |
| 163 | jobs map[string]*Job |
| 164 | order []string |
| 165 | completed []completion // finished-job summaries awaiting drain into the next turn |
| 166 | active string |
| 167 | destroying map[string]bool |
| 168 | artifactDirs map[string]string |
| 169 | loaded map[string]bool |
| 170 | tempRoot string |
| 171 | reservations map[string]int |
| 172 | |
| 173 | stalledWarning time.Duration |
| 174 | teardownGrace time.Duration |
| 175 | |
| 176 | taskRecorder TaskRecorder // optional task-monitoring lifecycle hook |
| 177 | } |
| 178 | |
| 179 | type completion struct { |
| 180 | sessionID string |
| 181 | text string |
| 182 | } |
| 183 | |
| 184 | // Option configures a Manager. |
| 185 | type Option func(*Manager) |
| 186 | |
| 187 | // TaskRecorder observes background-job lifecycle for task monitoring. The |
| 188 | // store-backed write side lives outside jobs (typically internal/taskmonitor); |
| 189 | // jobs only calls the hooks. RecordStart runs on the caller's goroutine, |
| 190 | // RecordDone on the job's own goroutine — implementations must be safe for |
| 191 | // concurrent use and must not block or fail the job pipeline (best-effort). |
| 192 | type TaskRecorder interface { |
| 193 | RecordStart(id, kind, label string) |
| 194 | RecordDone(id string, st Status, err error) |
| 195 | } |
| 196 | |
| 197 | // WithStalledWarningAfter enables one stalled warning per job after d without |
| 198 | // job-owned visible output. A non-positive duration disables stalled warnings. |
| 199 | func WithStalledWarningAfter(d time.Duration) Option { |
| 200 | return func(m *Manager) { |
| 201 | if d > 0 { |
| 202 | m.stalledWarning = d |
| 203 | } |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | // WithTeardownGrace overrides the Close/destroy grace window. Tests can set a |
| 208 | // short value; production uses DefaultTeardownGrace. |
| 209 | func WithTeardownGrace(d time.Duration) Option { |
| 210 | return func(m *Manager) { |
| 211 | if d >= 0 { |
| 212 | m.teardownGrace = d |
| 213 | } |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | // WithJobStartObserver observes every registered background job before its |
| 218 | // goroutine starts. Delivery uses this to retain a workspace writer lease over |
| 219 | // the job's opening writes. The callback must return quickly. |
| 220 | func WithJobStartObserver(observer func(done <-chan struct{})) Option { |
| 221 | return func(m *Manager) { m.onJobStart = observer } |
| 222 | } |
| 223 | |
| 224 | // WithSessionOwnershipProbe supplies the runtime ownership check used when |
| 225 | // loading persisted Running artifacts. The probe must return true only when the |
| 226 | // current runtime owns the session transcript for writing. |
| 227 | func WithSessionOwnershipProbe(probe func(path string) bool) Option { |
| 228 | return func(m *Manager) { m.sessionOwnershipProbe = probe } |
| 229 | } |
| 230 | |
| 231 | // WithTaskRecorder installs an optional background-job lifecycle recorder for |
| 232 | // task monitoring. A nil recorder disables recording. |
| 233 | func WithTaskRecorder(r TaskRecorder) Option { |
| 234 | return func(m *Manager) { m.taskRecorder = r } |
| 235 | } |
| 236 | |
| 237 | // SetTaskRecorder installs (or clears, with nil) the lifecycle recorder after |
| 238 | // construction. Controllers that assemble their job manager before the |
| 239 | // recorder's dependencies (workspace root, session id) are known use this. |
| 240 | func (m *Manager) SetTaskRecorder(r TaskRecorder) { m.taskRecorder = r } |
| 241 | |
| 242 | // TeardownGrace reports the manager's configured close/destroy wait window. |
| 243 | func (m *Manager) TeardownGrace() time.Duration { return m.teardownGrace } |
| 244 | |
| 245 | // NewManager returns a Manager whose jobs run under a fresh session-scoped |
| 246 | // context (cancelled by Close). sink receives job-lifecycle notices; pass the |
| 247 | // session's synchronized sink (event.Sync) since jobs emit from goroutines. |
| 248 | func NewManager(sink event.Sink, opts ...Option) *Manager { |
| 249 | if nilutil.IsNil(sink) { |
| 250 | sink = event.Discard |
| 251 | } |
| 252 | root, cancel := context.WithCancel(context.Background()) |
| 253 | tempRoot, _ := os.MkdirTemp("", "reasonix-jobs-*") |
| 254 | m := &Manager{ |
| 255 | sink: sink, |
| 256 | root: root, |
| 257 | cancel: cancel, |
| 258 | jobs: map[string]*Job{}, |
| 259 | destroying: map[string]bool{}, |
| 260 | artifactDirs: map[string]string{}, |
| 261 | reservations: map[string]int{}, |
| 262 | loaded: map[string]bool{}, |
| 263 | tempRoot: tempRoot, |
| 264 | teardownGrace: DefaultTeardownGrace, |
| 265 | ownerID: newManagerOwnerID(), |
| 266 | } |
| 267 | registerManagerOwner(m.ownerID) |
| 268 | for _, opt := range opts { |
| 269 | if opt != nil { |
| 270 | opt(m) |
| 271 | } |
| 272 | } |
| 273 | return m |
| 274 | } |
| 275 | |
| 276 | func newManagerOwnerID() string { |
| 277 | var token [16]byte |
| 278 | if _, err := rand.Read(token[:]); err == nil { |
| 279 | return hex.EncodeToString(token[:]) |
| 280 | } |
| 281 | return fmt.Sprintf("%d-%d-%d", os.Getpid(), time.Now().UnixNano(), managerOwnerSeq.Add(1)) |
| 282 | } |
| 283 | |
| 284 | func registerManagerOwner(ownerID string) { |
| 285 | ownerID = strings.TrimSpace(ownerID) |
| 286 | if ownerID == "" { |
| 287 | return |
| 288 | } |
| 289 | liveManagerOwners.Lock() |
| 290 | liveManagerOwners.ids[ownerID] = struct{}{} |
| 291 | liveManagerOwners.Unlock() |
| 292 | } |
| 293 | |
| 294 | func managerOwnerIsLive(ownerID string) bool { |
| 295 | ownerID = strings.TrimSpace(ownerID) |
| 296 | if ownerID == "" { |
| 297 | return false |
| 298 | } |
| 299 | liveManagerOwners.RLock() |
| 300 | _, ok := liveManagerOwners.ids[ownerID] |
| 301 | liveManagerOwners.RUnlock() |
| 302 | return ok |
| 303 | } |
| 304 | |
| 305 | func (m *Manager) releaseOwner() { |
| 306 | if m == nil { |
| 307 | return |
| 308 | } |
| 309 | m.ownerDone.Do(func() { |
| 310 | liveManagerOwners.Lock() |
| 311 | delete(liveManagerOwners.ids, m.ownerID) |
| 312 | liveManagerOwners.Unlock() |
| 313 | }) |
| 314 | } |
| 315 | |
| 316 | // jobWriter appends a job's streamed output under its lock so a concurrent |
| 317 | // Output read never races the producing goroutine. |
| 318 | type jobWriter struct{ j *Job } |
| 319 | |
| 320 | func (w jobWriter) Write(p []byte) (int, error) { |
| 321 | w.j.mu.Lock() |
| 322 | defer w.j.mu.Unlock() |
| 323 | w.j.clock.activityAt = nowMs() |
| 324 | w.j.tail = appendTail(w.j.tail, p, defaultTailBytes) |
| 325 | if w.j.artifactFile != nil { |
| 326 | if _, err := w.j.artifactFile.Write(p); err != nil { |
| 327 | w.j.artifactErr = err.Error() |
| 328 | } |
| 329 | } |
| 330 | return len(p), nil |
| 331 | } |
| 332 | |
| 333 | // Start launches run on a goroutine under the manager's session context and |
| 334 | // returns the job immediately. run streams output to the writer and returns the |
| 335 | // terminal result text (a task's final answer; a bash job streams everything to |
| 336 | // the buffer and returns ""). The job is marked killed when its context was |
| 337 | // cancelled, failed on any other error, else done. |
| 338 | func (m *Manager) Start(kind, label string, run func(ctx context.Context, out io.Writer) (string, error)) *Job { |
| 339 | return m.StartForSession("", kind, label, run) |
| 340 | } |
| 341 | |
| 342 | // validatePathSegment rejects values that would let parentSession or kind |
| 343 | // escape the temp-root fallback built by artifactDirLocked. Persistent artifact |
| 344 | // directories bound by SetActiveSessionPath are trusted store paths and are |
| 345 | // intentionally outside that temp root. The check is intentionally conservative: |
| 346 | // it forbids any path-separator character (forward slash, backslash), NUL, and |
| 347 | // any control character. Empty parentSession is allowed (the unscoped default); |
| 348 | // kind must be non-empty. |
| 349 | // |
| 350 | // See #6932. Before this check existed, a malicious or malformed parentSession |
| 351 | // such as "../../etc" combined with filepath.Join(tempRoot, parentSession, id) |
| 352 | // resolved to a directory outside the manager's temp root, allowing the |
| 353 | // subsequent os.MkdirAll + os.OpenFile to create files at locations controlled |
| 354 | // by the caller (subject to the running process's filesystem permissions). |
| 355 | func validatePathSegment(name, field string) error { |
| 356 | if field == "kind" && name == "" { |
| 357 | return fmt.Errorf("jobs: %s must not be empty", field) |
| 358 | } |
| 359 | for i, r := range name { |
| 360 | switch { |
| 361 | case r < 0x20 || r == 0x7f: |
| 362 | return fmt.Errorf("jobs: %s contains control character 0x%02x at index %d", field, r, i) |
| 363 | case r == '/' || r == '\\': |
| 364 | return fmt.Errorf("jobs: %s contains path separator %q at index %d", field, r, i) |
| 365 | } |
| 366 | } |
| 367 | if name == "." || name == ".." { |
| 368 | return fmt.Errorf("jobs: %s is reserved (%q)", field, name) |
| 369 | } |
| 370 | return nil |
| 371 | } |
| 372 | |
| 373 | // startInvalid registers a job that failed validation BEFORE any goroutine or |
| 374 | // artifact was created. The job is observable to Wait / list calls as Failed |
| 375 | // with the validation error recorded in artifactErr, and no run goroutine is |
| 376 | // started so the manager's wg is unaffected. |
| 377 | func (m *Manager) startInvalid(parentSession, kind, label string, validationErr error) *Job { |
| 378 | finishedAt := nowMs() |
| 379 | m.mu.Lock() |
| 380 | m.seq++ |
| 381 | id := fmt.Sprintf("invalid-%d", m.seq) |
| 382 | j := &Job{ |
| 383 | ID: id, |
| 384 | Kind: kind, |
| 385 | Label: label, |
| 386 | SessionID: parentSession, |
| 387 | status: Failed, |
| 388 | clock: jobClock{startedAt: finishedAt, activityAt: finishedAt, finishedAt: finishedAt}, |
| 389 | outcome: jobOutcome{returned: true}, |
| 390 | cancel: func() {}, |
| 391 | done: make(chan struct{}), |
| 392 | artifactComplete: false, |
| 393 | artifactErr: validationErr.Error(), |
| 394 | } |
| 395 | key := jobKey(parentSession, id) |
| 396 | m.jobs[key] = j |
| 397 | m.order = append(m.order, key) |
| 398 | m.mu.Unlock() |
| 399 | close(j.done) |
| 400 | m.recordCompletion(j, Failed, validationErr) |
| 401 | m.notifyRuntime(parentSession, id) |
| 402 | return j |
| 403 | } |
| 404 | |
| 405 | func runRecovered(ctx context.Context, out io.Writer, run func(context.Context, io.Writer) (string, error)) (result string, err error) { |
| 406 | defer func() { |
| 407 | if r := recover(); r != nil { |
| 408 | err = fmt.Errorf("internal error: panic: %v\n%s", r, debug.Stack()) |
| 409 | } |
| 410 | }() |
| 411 | return run(ctx, out) |
| 412 | } |
| 413 | |
| 414 | func (m *Manager) openArtifactLocked(parentSession, id string) (logPath, metaPath string, file *os.File, artifactErr string) { |
| 415 | dir := m.artifactDirLocked(parentSession) |
| 416 | if dir == "" { |
| 417 | return "", "", nil, "artifact directory unavailable" |
| 418 | } |
| 419 | if err := ensurePrivateArtifactDir(dir); err != nil { |
| 420 | return filepath.Join(dir, id+jobLogExt), filepath.Join(dir, id+jobMetaExt), nil, err.Error() |
| 421 | } |
| 422 | logPath = filepath.Join(dir, id+jobLogExt) |
| 423 | metaPath = filepath.Join(dir, id+jobMetaExt) |
| 424 | f, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) |
| 425 | if err != nil { |
| 426 | return logPath, metaPath, nil, err.Error() |
| 427 | } |
| 428 | // O_TRUNC does not apply the requested mode to an existing artifact. Tighten |
| 429 | // it before any raw tool output is written so upgrades cannot append secrets |
| 430 | // to a legacy 0644 log. |
| 431 | if err := f.Chmod(0o600); err != nil { |
| 432 | _ = f.Close() |
| 433 | return logPath, metaPath, nil, err.Error() |
| 434 | } |
| 435 | return logPath, metaPath, f, "" |
| 436 | } |
| 437 | |
| 438 | func ensurePrivateArtifactDir(dir string) error { |
| 439 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 440 | return err |
| 441 | } |
| 442 | // MkdirAll leaves an existing 0755 directory unchanged. |
| 443 | return os.Chmod(dir, 0o700) |
| 444 | } |
| 445 | |
| 446 | func (m *Manager) artifactDirLocked(parentSession string) string { |
| 447 | parentSession = strings.TrimSpace(parentSession) |
| 448 | if parentSession != "" { |
| 449 | if dir := strings.TrimSpace(m.artifactDirs[parentSession]); dir != "" { |
| 450 | return dir |
| 451 | } |
| 452 | } |
| 453 | if strings.TrimSpace(m.tempRoot) == "" { |
| 454 | return "" |
| 455 | } |
| 456 | if parentSession == "" { |
| 457 | return filepath.Join(m.tempRoot, "default") |
| 458 | } |
| 459 | return filepath.Join(m.tempRoot, parentSession) |
| 460 | } |
| 461 | |
| 462 | func (m *Manager) writeJobMetaLocked(j *Job, st Status) error { |
| 463 | j.artifactStatus = st |
| 464 | if j.artifactMetaPath == "" { |
| 465 | return nil |
| 466 | } |
| 467 | meta := artifactMeta{ |
| 468 | ID: j.ID, |
| 469 | Kind: j.Kind, |
| 470 | Label: j.Label, |
| 471 | SessionID: j.SessionID, |
| 472 | OwnerID: m.ownerID, |
| 473 | Status: st, |
| 474 | StartedAt: j.clock.startedAt, |
| 475 | FinishedAt: j.clock.finishedAt, |
| 476 | ArtifactComplete: st != Running && j.artifactComplete && j.artifactErr == "", |
| 477 | ArtifactError: j.artifactErr, |
| 478 | LogPath: filepath.Base(j.artifactPath), |
| 479 | } |
| 480 | if j.Kind == "task" { |
| 481 | meta.MutationEvidenceVersion = mutationEvidenceVersion |
| 482 | meta.MutationEvidence = mutationEvidenceForArtifact(j.evidence) |
| 483 | } |
| 484 | return writeMeta(j.artifactMetaPath, meta) |
| 485 | } |
| 486 | |
| 487 | func mutationEvidenceForArtifact(summary evidence.ChildEvidenceSummary) *artifactMutationEvidence { |
| 488 | firstMutation := -1 |
| 489 | for i, receipt := range summary.Receipts { |
| 490 | if receipt.Success && receipt.Mutation { |
| 491 | firstMutation = i |
| 492 | break |
| 493 | } |
| 494 | } |
| 495 | if firstMutation < 0 { |
| 496 | return nil |
| 497 | } |
| 498 | return &artifactMutationEvidence{ |
| 499 | Paths: summary.MutationPaths(), |
| 500 | } |
| 501 | } |
| 502 | |
| 503 | func mutationEvidenceFromArtifact(meta artifactMeta) evidence.ChildEvidenceSummary { |
| 504 | if meta.Kind != "task" { |
| 505 | return evidence.ChildEvidenceSummary{} |
| 506 | } |
| 507 | if meta.MutationEvidenceVersion != mutationEvidenceVersion { |
| 508 | // Any version this build cannot parse — a pre-feature artifact |
| 509 | // (version 0) or one written by a newer build — is treated as an |
| 510 | // opaque mutation. A missing summary only proves the mutation state |
| 511 | // was not recorded, not that the task made no changes: a legacy |
| 512 | // background writer task collected after upgrade could carry real, |
| 513 | // edits. Preserve the existing unknown-mutation compatibility record; |
| 514 | // it never implies verification or creates an acceptance requirement. |
| 515 | return opaqueRecoveredTaskMutation() |
| 516 | } |
| 517 | if meta.MutationEvidence == nil { |
| 518 | // Same-version artifact with no summary: this build DID record the |
| 519 | // mutation state and found none, so there is genuinely nothing to |
| 520 | // recover. |
| 521 | return evidence.ChildEvidenceSummary{} |
| 522 | } |
| 523 | |
| 524 | paths := append([]string(nil), meta.MutationEvidence.Paths...) |
| 525 | // Historical risk labels do not erase observed paths or create obligations. |
| 526 | return evidence.ChildEvidenceSummary{Receipts: []evidence.Receipt{{ |
| 527 | ToolName: recoveredBackgroundTaskToolName, |
| 528 | Success: true, |
| 529 | Write: true, |
| 530 | Mutation: true, |
| 531 | Paths: paths, |
| 532 | }}} |
| 533 | } |
| 534 | |
| 535 | func opaqueRecoveredTaskMutation() evidence.ChildEvidenceSummary { |
| 536 | return evidence.ChildEvidenceSummary{Receipts: []evidence.Receipt{{ |
| 537 | ToolName: recoveredBackgroundTaskToolName, |
| 538 | Success: true, |
| 539 | Write: true, |
| 540 | Mutation: true, |
| 541 | }}} |
| 542 | } |
| 543 | |
| 544 | func (m *Manager) artifactTargetDirForJob(j *Job) string { |
| 545 | if j == nil { |
| 546 | return "" |
| 547 | } |
| 548 | m.mu.Lock() |
| 549 | defer m.mu.Unlock() |
| 550 | session := strings.TrimSpace(j.SessionID) |
| 551 | if session == "" { |
| 552 | return "" |
| 553 | } |
| 554 | return strings.TrimSpace(m.artifactDirs[session]) |
| 555 | } |
| 556 | |
| 557 | func (j *Job) noteArtifactErr(msg string) { |
| 558 | msg = strings.TrimSpace(msg) |
| 559 | if msg == "" { |
| 560 | return |
| 561 | } |
| 562 | if j.artifactErr == "" { |
| 563 | j.artifactErr = msg |
| 564 | } else { |
| 565 | j.artifactErr += "; " + msg |
| 566 | } |
| 567 | j.artifactComplete = false |
| 568 | } |
| 569 | |
| 570 | func (j *Job) moveArtifactToDirLocked(dir string) error { |
| 571 | dir = strings.TrimSpace(dir) |
| 572 | if dir == "" || j.artifactPath == "" { |
| 573 | return nil |
| 574 | } |
| 575 | if filepath.Clean(filepath.Dir(j.artifactPath)) == filepath.Clean(dir) { |
| 576 | return nil |
| 577 | } |
| 578 | if err := ensurePrivateArtifactDir(dir); err != nil { |
| 579 | return err |
| 580 | } |
| 581 | newLogPath := filepath.Join(dir, filepath.Base(j.artifactPath)) |
| 582 | if err := moveArtifactFile(j.artifactPath, newLogPath); err != nil { |
| 583 | return err |
| 584 | } |
| 585 | j.artifactPath = newLogPath |
| 586 | if j.artifactMetaPath != "" { |
| 587 | j.artifactMetaPath = filepath.Join(dir, filepath.Base(j.artifactMetaPath)) |
| 588 | } |
| 589 | return nil |
| 590 | } |
| 591 | |
| 592 | func (m *Manager) monitorStalled(parentSession string, j *Job) { |
| 593 | defer m.wg.Done() |
| 594 | timer := time.NewTimer(m.stalledWarning) |
| 595 | defer timer.Stop() |
| 596 | for { |
| 597 | select { |
| 598 | case <-j.done: |
| 599 | return |
| 600 | case <-timer.C: |
| 601 | j.mu.Lock() |
| 602 | if j.outcome.returned || j.status != Running { |
| 603 | j.mu.Unlock() |
| 604 | return |
| 605 | } |
| 606 | idle := time.Since(time.UnixMilli(j.clock.activityAt)) |
| 607 | if idle >= m.stalledWarning && !j.clock.stalled { |
| 608 | j.clock.stalled = true |
| 609 | j.mu.Unlock() |
| 610 | m.recordStalled(parentSession, j.ID, j.Kind, j.Label) |
| 611 | return |
| 612 | } |
| 613 | wait := m.stalledWarning - idle |
| 614 | if wait <= 0 { |
| 615 | wait = m.stalledWarning |
| 616 | } |
| 617 | j.mu.Unlock() |
| 618 | timer.Reset(wait) |
| 619 | } |
| 620 | } |
| 621 | } |
| 622 | |
| 623 | // recordCompletion queues the finished-job summary for DrainCompletedNote and |
| 624 | // emits a closing Notice (warn for a failure, info otherwise). |
| 625 | func (m *Manager) recordCompletion(j *Job, st Status, err error) string { |
| 626 | id, kind, label := j.ID, j.Kind, j.Label |
| 627 | tag := id |
| 628 | if label != "" { |
| 629 | tag = fmt.Sprintf("%s (%s)", id, label) |
| 630 | } |
| 631 | shouldEmit := false |
| 632 | m.mu.Lock() |
| 633 | parentSession := strings.TrimSpace(j.SessionID) |
| 634 | if parentSession != "" && m.destroying[parentSession] { |
| 635 | m.mu.Unlock() |
| 636 | return parentSession |
| 637 | } |
| 638 | m.completed = append(m.completed, completion{ |
| 639 | sessionID: parentSession, |
| 640 | text: fmt.Sprintf("%s — %s", tag, st), |
| 641 | }) |
| 642 | active := m.active |
| 643 | shouldEmit = active == "" || parentSession == "" || active == parentSession |
| 644 | m.mu.Unlock() |
| 645 | |
| 646 | if !nilutil.IsNil(m.taskRecorder) { |
| 647 | m.taskRecorder.RecordDone(id, st, err) |
| 648 | } |
| 649 | |
| 650 | level, text := event.LevelInfo, fmt.Sprintf("background %s finished: %s", kind, id) |
| 651 | detail := "" |
| 652 | switch st { |
| 653 | case Failed: |
| 654 | level, text = event.LevelWarn, fmt.Sprintf("background %s failed: needs attention", kind) |
| 655 | detail = fmt.Sprintf("background %s failed: %s — %v", kind, id, err) |
| 656 | case Killed: |
| 657 | text = fmt.Sprintf("background %s killed: %s", kind, id) |
| 658 | } |
| 659 | if shouldEmit { |
| 660 | m.sink.Emit(event.Event{Kind: event.Notice, Code: event.NoticeCodeBackgroundJobFinished, Level: level, Text: text, Detail: detail}) |
| 661 | } |
| 662 | return parentSession |
| 663 | } |
| 664 | |
| 665 | func (m *Manager) recordStalled(parentSession, id, kind, label string) { |
| 666 | tag := id |
| 667 | if label != "" { |
| 668 | tag = fmt.Sprintf("%s (%s)", id, label) |
| 669 | } |
| 670 | parentSession = strings.TrimSpace(parentSession) |
| 671 | m.mu.Lock() |
| 672 | if parentSession != "" && m.destroying[parentSession] { |
| 673 | m.mu.Unlock() |
| 674 | return |
| 675 | } |
| 676 | quietFor := m.stalledWarning.Round(time.Second) |
| 677 | text := fmt.Sprintf("%s is still running after %s with no visible output — a quiet long-running job can look like this and is not necessarily stuck. If it should have finished, inspect it with job_output, or stop it with job_kill. Tune or disable this check with tools.background_jobs.stalled_warning_seconds in your config (0 disables).", tag, quietFor) |
| 678 | m.completed = append(m.completed, completion{sessionID: parentSession, text: text}) |
| 679 | active := m.active |
| 680 | shouldEmit := active == "" || parentSession == "" || active == parentSession |
| 681 | notice := event.Event{Kind: event.Notice, Level: event.LevelWarn, |
| 682 | Text: fmt.Sprintf("background %s still running after %s with no visible output: %s", kind, quietFor, id), |
| 683 | Detail: "A quiet long-running job can look like this, so this is a heads-up, not an error. If it should have finished, inspect with job_output, or stop it with job_kill. Set tools.background_jobs.stalled_warning_seconds to 0 in your config to disable this notice."} |
| 684 | m.mu.Unlock() |
| 685 | if shouldEmit { |
| 686 | m.sink.Emit(notice) |
| 687 | } |
| 688 | } |
| 689 | |
| 690 | func (m *Manager) get(parentSession, id string) *Job { |
| 691 | m.mu.Lock() |
| 692 | defer m.mu.Unlock() |
| 693 | return m.findJobLocked(parentSession, id) |
| 694 | } |
| 695 | |
| 696 | func (m *Manager) findJobLocked(parentSession, id string) *Job { |
| 697 | parentSession = strings.TrimSpace(parentSession) |
| 698 | id = strings.TrimSpace(id) |
| 699 | if parentSession != "" { |
| 700 | return m.jobs[jobKey(parentSession, id)] |
| 701 | } |
| 702 | for _, key := range m.order { |
| 703 | j := m.jobs[key] |
| 704 | if j != nil && j.ID == id { |
| 705 | return j |
| 706 | } |
| 707 | } |
| 708 | return nil |
| 709 | } |
| 710 | |
| 711 | // Output returns the job's output produced since the last Output call plus its |
| 712 | // current status. ok is false when the id is unknown. |
| 713 | func (m *Manager) Output(id string) (text string, status Status, ok bool) { |
| 714 | return m.OutputForSession("", id) |
| 715 | } |
| 716 | |
| 717 | // OutputForSession returns output only when id belongs to parentSession. Empty |
| 718 | // parentSession preserves the legacy unscoped behavior. |
| 719 | func (m *Manager) OutputForSession(parentSession, id string) (text string, status Status, ok bool) { |
| 720 | j := m.get(parentSession, id) |
| 721 | if j == nil { |
| 722 | return "", "", false |
| 723 | } |
| 724 | j.mu.Lock() |
| 725 | defer j.mu.Unlock() |
| 726 | if j.artifactPath != "" { |
| 727 | text = j.readArtifactSinceOffsetLocked() |
| 728 | } else { |
| 729 | full := string(j.tail) |
| 730 | if j.readOffset < int64(len(full)) { |
| 731 | text = full[j.readOffset:] |
| 732 | j.readOffset = int64(len(full)) |
| 733 | } |
| 734 | } |
| 735 | // A task job streams nothing to the tail buffer — its answer lands in |
| 736 | // outcome.text. Surface it once when terminal with no buffered output, so a |
| 737 | // task's answer is visible here too (bash_output promises task support). |
| 738 | if text == "" && j.status != Running && j.outcome.text != "" && !j.outcome.read { |
| 739 | text = j.outcome.text |
| 740 | j.outcome.read = true |
| 741 | } |
| 742 | if j.artifactErr != "" { |
| 743 | if text != "" { |
| 744 | text += "\n" |
| 745 | } |
| 746 | text += "job artifact incomplete: " + j.artifactErr |
| 747 | } |
| 748 | return text, j.status, true |
| 749 | } |
| 750 | |
| 751 | func (j *Job) readArtifactSinceOffsetLocked() string { |
| 752 | f, err := os.Open(j.artifactPath) |
| 753 | if err != nil { |
| 754 | if j.artifactErr == "" { |
| 755 | j.artifactErr = err.Error() |
| 756 | } |
| 757 | return "" |
| 758 | } |
| 759 | defer f.Close() |
| 760 | info, err := f.Stat() |
| 761 | if err != nil { |
| 762 | if j.artifactErr == "" { |
| 763 | j.artifactErr = err.Error() |
| 764 | } |
| 765 | return "" |
| 766 | } |
| 767 | size := info.Size() |
| 768 | if j.readOffset > size { |
| 769 | j.readOffset = size |
| 770 | return "" |
| 771 | } |
| 772 | if _, err := f.Seek(j.readOffset, io.SeekStart); err != nil { |
| 773 | if j.artifactErr == "" { |
| 774 | j.artifactErr = err.Error() |
| 775 | } |
| 776 | return "" |
| 777 | } |
| 778 | b, err := io.ReadAll(f) |
| 779 | if err != nil { |
| 780 | if j.artifactErr == "" { |
| 781 | j.artifactErr = err.Error() |
| 782 | } |
| 783 | return "" |
| 784 | } |
| 785 | text := string(b) |
| 786 | j.readOffset = size |
| 787 | return text |
| 788 | } |
| 789 | |
| 790 | // readArtifactAllLocked deliberately reads raw bytes: the artifact is captured |
| 791 | // subprocess output (possibly binary), not a user-edited config file, and the |
| 792 | // incremental reader (readArtifactSinceOffsetLocked) is raw byte-offset based — |
| 793 | // decoding only the whole-file path would render the same artifact in two |
| 794 | // different encodings and could garble binary output via UTF-16 misdetection. |
| 795 | func (j *Job) readArtifactAllLocked() string { |
| 796 | if j.artifactPath == "" { |
| 797 | return "" |
| 798 | } |
| 799 | b, err := os.ReadFile(j.artifactPath) |
| 800 | if err != nil { |
| 801 | if j.artifactErr == "" { |
| 802 | j.artifactErr = err.Error() |
| 803 | } |
| 804 | return "" |
| 805 | } |
| 806 | return string(b) |
| 807 | } |
| 808 | |
| 809 | // Kill cancels a running job. Returns false when the id is unknown or the job has |
| 810 | // already finished. |
| 811 | func (m *Manager) Kill(id string) bool { |
| 812 | return m.KillForSession("", id) |
| 813 | } |
| 814 | |
| 815 | // KillForSession cancels a running job only when it belongs to parentSession. |
| 816 | // Empty parentSession preserves the legacy unscoped behavior. |
| 817 | func (m *Manager) KillForSession(parentSession, id string) bool { |
| 818 | j := m.get(parentSession, id) |
| 819 | if j == nil { |
| 820 | return false |
| 821 | } |
| 822 | j.mu.Lock() |
| 823 | running := j.status == Running |
| 824 | if running { |
| 825 | // Flip to Killed synchronously so Output/Wait reflect the kill the instant |
| 826 | // it's requested, not whenever the run goroutine's cmd.Run returns (which |
| 827 | // trails by WaitDelay while a cancelled process tree tears down). The |
| 828 | // goroutine still sets Killed + records completion on return; this only |
| 829 | // fires when the job is actually Running, so a job that just finished |
| 830 | // keeps its real terminal status. |
| 831 | j.status = Killed |
| 832 | } |
| 833 | j.mu.Unlock() |
| 834 | if !running { |
| 835 | return false |
| 836 | } |
| 837 | j.cancel() |
| 838 | return true |
| 839 | } |
| 840 | |
| 841 | // Wait blocks until the named jobs (or every currently-running job when ids is |
| 842 | // empty) reach a terminal state, or ctx is cancelled, or timeoutSec elapses |
| 843 | // (0 = no timeout). It returns each target's snapshot regardless of why it |
| 844 | // returned, so a timeout still reports partial progress. |
| 845 | func (m *Manager) Wait(ctx context.Context, ids []string, timeoutSec int) []Result { |
| 846 | return m.WaitForSession(ctx, "", ids, timeoutSec) |
| 847 | } |
| 848 | |
| 849 | // WaitForSession waits only on jobs owned by parentSession. Empty parentSession |
| 850 | // preserves the legacy unscoped behavior. |
| 851 | func (m *Manager) WaitForSession(ctx context.Context, parentSession string, ids []string, timeoutSec int) []Result { |
| 852 | targets := m.resolve(parentSession, ids) |
| 853 | if len(targets) == 0 { |
| 854 | return nil |
| 855 | } |
| 856 | var timeout <-chan time.Time |
| 857 | if timeoutSec > 0 { |
| 858 | t := time.NewTimer(time.Duration(timeoutSec) * time.Second) |
| 859 | defer t.Stop() |
| 860 | timeout = t.C |
| 861 | } |
| 862 | for _, j := range targets { |
| 863 | select { |
| 864 | case <-j.done: |
| 865 | case <-ctx.Done(): |
| 866 | return m.results(targets) |
| 867 | case <-timeout: |
| 868 | return m.results(targets) |
| 869 | } |
| 870 | } |
| 871 | return m.results(targets) |
| 872 | } |
| 873 | |
| 874 | // resolve maps requested ids to jobs; an empty list selects all running jobs. |
| 875 | func (m *Manager) resolve(parentSession string, ids []string) []*Job { |
| 876 | m.mu.Lock() |
| 877 | defer m.mu.Unlock() |
| 878 | var out []*Job |
| 879 | if len(ids) == 0 { |
| 880 | for _, key := range m.order { |
| 881 | j := m.jobs[key] |
| 882 | if !sessionMatches(parentSession, j.SessionID) { |
| 883 | continue |
| 884 | } |
| 885 | j.mu.Lock() |
| 886 | running := j.status == Running |
| 887 | j.mu.Unlock() |
| 888 | if running { |
| 889 | out = append(out, j) |
| 890 | } |
| 891 | } |
| 892 | return out |
| 893 | } |
| 894 | for _, id := range ids { |
| 895 | if j := m.findJobLocked(parentSession, id); j != nil { |
| 896 | out = append(out, j) |
| 897 | } |
| 898 | } |
| 899 | return out |
| 900 | } |
| 901 | |
| 902 | func (m *Manager) results(targets []*Job) []Result { |
| 903 | out := make([]Result, 0, len(targets)) |
| 904 | for _, j := range targets { |
| 905 | j.mu.Lock() |
| 906 | text := j.outcome.text |
| 907 | if text == "" && j.artifactPath != "" { |
| 908 | text = j.readArtifactAllLocked() |
| 909 | } |
| 910 | if text == "" { |
| 911 | text = string(j.tail) |
| 912 | } |
| 913 | if j.artifactErr != "" { |
| 914 | if text != "" { |
| 915 | text += "\n" |
| 916 | } |
| 917 | text += "job artifact incomplete: " + j.artifactErr |
| 918 | } |
| 919 | out = append(out, Result{ID: j.ID, Kind: j.Kind, Label: j.Label, Status: j.status, Output: text}) |
| 920 | j.mu.Unlock() |
| 921 | } |
| 922 | return out |
| 923 | } |
| 924 | |
| 925 | // Running returns a snapshot of the still-running jobs (for the status bar). |
| 926 | func (m *Manager) Running() []View { |
| 927 | return m.RunningForSession("") |
| 928 | } |
| 929 | |
| 930 | // RunningForSession returns still-running jobs owned by parentSession. Empty |
| 931 | // parentSession preserves the legacy unscoped behavior. |
| 932 | func (m *Manager) RunningForSession(parentSession string) []View { |
| 933 | m.mu.Lock() |
| 934 | defer m.mu.Unlock() |
| 935 | var out []View |
| 936 | for _, key := range m.order { |
| 937 | j := m.jobs[key] |
| 938 | if !sessionMatches(parentSession, j.SessionID) { |
| 939 | continue |
| 940 | } |
| 941 | select { |
| 942 | case <-j.done: |
| 943 | continue |
| 944 | default: |
| 945 | } |
| 946 | j.mu.Lock() |
| 947 | // A cancellation request flips the persisted/result status to Killed |
| 948 | // synchronously, but the process tree may still be unwinding. Keep the job |
| 949 | // on the operational running surface until its done channel closes so |
| 950 | // Desktop rebuild guards and Delivery workspace leases cannot declare the |
| 951 | // runtime idle early. The public view remains "running" while a stop is |
| 952 | // in flight; clients may render a local "stopping" state after they |
| 953 | // request cancellation. |
| 954 | out = append(out, View{ID: j.ID, Kind: j.Kind, Label: j.Label, Status: string(Running), StartedAt: j.clock.startedAt}) |
| 955 | j.mu.Unlock() |
| 956 | } |
| 957 | return out |
| 958 | } |
| 959 | |
| 960 | // ReserveStartForSession atomically reserves capacity for a job start. The |
| 961 | // caller must release the reservation after StartForSession has registered the |
| 962 | // job (or when setup fails). Running jobs and in-flight start reservations both |
| 963 | // count toward limit, so concurrent callers cannot overshoot it. |
| 964 | func (m *Manager) ReserveStartForSession(parentSession, kind string, limit int) (release func(), running int, ok bool) { |
| 965 | if limit <= 0 { |
| 966 | return func() {}, 0, true |
| 967 | } |
| 968 | parentSession = strings.TrimSpace(parentSession) |
| 969 | key := jobKey(parentSession, kind) |
| 970 | m.mu.Lock() |
| 971 | for _, jobKey := range m.order { |
| 972 | j := m.jobs[jobKey] |
| 973 | if j == nil || !sessionMatches(parentSession, j.SessionID) || j.Kind != kind { |
| 974 | continue |
| 975 | } |
| 976 | select { |
| 977 | case <-j.done: |
| 978 | default: |
| 979 | running++ |
| 980 | } |
| 981 | } |
| 982 | running += m.reservations[key] |
| 983 | if running >= limit { |
| 984 | m.mu.Unlock() |
| 985 | return func() {}, running, false |
| 986 | } |
| 987 | m.reservations[key]++ |
| 988 | m.mu.Unlock() |
| 989 | |
| 990 | var once sync.Once |
| 991 | release = func() { |
| 992 | once.Do(func() { |
| 993 | m.mu.Lock() |
| 994 | m.reservations[key]-- |
| 995 | if m.reservations[key] == 0 { |
| 996 | delete(m.reservations, key) |
| 997 | } |
| 998 | m.mu.Unlock() |
| 999 | }) |
| 1000 | } |
| 1001 | return release, running, true |
| 1002 | } |
| 1003 | |
| 1004 | // HasUnfinishedForSession reports whether parentSession owns any job whose |
| 1005 | // goroutine has not fully exited yet. Empty parentSession preserves the legacy |
| 1006 | // unscoped behavior. |
| 1007 | func (m *Manager) HasUnfinishedForSession(parentSession string) bool { |
| 1008 | m.mu.Lock() |
| 1009 | defer m.mu.Unlock() |
| 1010 | for _, key := range m.order { |
| 1011 | j := m.jobs[key] |
| 1012 | if !sessionMatches(parentSession, j.SessionID) { |
| 1013 | continue |
| 1014 | } |
| 1015 | select { |
| 1016 | case <-j.done: |
| 1017 | default: |
| 1018 | return true |
| 1019 | } |
| 1020 | } |
| 1021 | return false |
| 1022 | } |
| 1023 | |
| 1024 | // DrainCompletedNote returns (and clears) a one-line summary of jobs that |
| 1025 | // finished since the last drain, for the controller to fold into the next turn |
| 1026 | // so the model learns of completions. "" when nothing finished. |
| 1027 | func (m *Manager) DrainCompletedNote() string { |
| 1028 | return m.DrainCompletedNoteForSession("") |
| 1029 | } |
| 1030 | |
| 1031 | // DrainCompletedNoteForSession drains completion notes for parentSession only. |
| 1032 | // Notes for other sessions stay queued until that session becomes active again. |
| 1033 | // Empty parentSession preserves the legacy unscoped behavior. |
| 1034 | func (m *Manager) DrainCompletedNoteForSession(parentSession string) string { |
| 1035 | m.mu.Lock() |
| 1036 | var c []string |
| 1037 | if strings.TrimSpace(parentSession) == "" { |
| 1038 | for _, item := range m.completed { |
| 1039 | c = append(c, item.text) |
| 1040 | } |
| 1041 | m.completed = nil |
| 1042 | } else { |
| 1043 | remaining := m.completed[:0] |
| 1044 | for _, item := range m.completed { |
| 1045 | if item.sessionID == parentSession { |
| 1046 | c = append(c, item.text) |
| 1047 | } else { |
| 1048 | remaining = append(remaining, item) |
| 1049 | } |
| 1050 | } |
| 1051 | m.completed = remaining |
| 1052 | } |
| 1053 | m.mu.Unlock() |
| 1054 | if len(c) == 0 { |
| 1055 | return "" |
| 1056 | } |
| 1057 | return "Background job updates since your last message: " + strings.Join(c, "; ") + |
| 1058 | ". Read their output with job_output if you still need it." |
| 1059 | } |
| 1060 | |
| 1061 | // SetActiveSession controls which session receives lifecycle notices for jobs |
| 1062 | // that finish asynchronously. Empty active session preserves legacy behavior. |
| 1063 | func (m *Manager) SetActiveSession(parentSession string) { |
| 1064 | m.mu.Lock() |
| 1065 | m.active = strings.TrimSpace(parentSession) |
| 1066 | m.mu.Unlock() |
| 1067 | } |
| 1068 | |
| 1069 | // validateTrustedSessionPath performs defense-in-depth syntax validation on a |
| 1070 | // transcript path already trusted by the store/controller layer. It rejects |
| 1071 | // control characters, but deliberately preserves separators and `..`: those are |
| 1072 | // valid host-path syntax, and rejecting them without a trusted root would break |
| 1073 | // legitimate relative paths without establishing filesystem containment. |
| 1074 | func validateTrustedSessionPath(sessionPath string) error { |
| 1075 | if sessionPath == "" { |
| 1076 | return fmt.Errorf("jobs: sessionPath must not be empty") |
| 1077 | } |
| 1078 | for i, r := range sessionPath { |
| 1079 | if r < 0x20 || r == 0x7f { |
| 1080 | return fmt.Errorf("jobs: sessionPath contains control character 0x%02x at index %d", r, i) |
| 1081 | } |
| 1082 | } |
| 1083 | return nil |
| 1084 | } |
| 1085 | |
| 1086 | // SetActiveSessionPath binds a parent session id to its persistent transcript |
| 1087 | // path, migrates any temporary artifacts, and loads completed job tombstones from |
| 1088 | // the session sidecar. sessionPath must come from the trusted store/controller |
| 1089 | // path; this method does not establish filesystem containment on its own. |
| 1090 | func (m *Manager) SetActiveSessionPath(parentSession, sessionPath string) { |
| 1091 | defer func() { m.notifyRuntime("", "") }() |
| 1092 | parentSession = strings.TrimSpace(parentSession) |
| 1093 | sessionPath = strings.TrimSpace(sessionPath) |
| 1094 | // Preserve the legacy active-only behavior for calls without a complete |
| 1095 | // binding. In particular, an empty path is not an error or filesystem input. |
| 1096 | if parentSession == "" || sessionPath == "" { |
| 1097 | m.mu.Lock() |
| 1098 | m.active = parentSession |
| 1099 | m.mu.Unlock() |
| 1100 | return |
| 1101 | } |
| 1102 | // Reject malformed trusted paths before any filesystem side effect. This is |
| 1103 | // syntax hardening, not a boundary for arbitrary caller-controlled paths. |
| 1104 | if err := validateTrustedSessionPath(sessionPath); err != nil { |
| 1105 | m.mu.Lock() |
| 1106 | m.active = parentSession |
| 1107 | // A rejected rebinding must not leave future jobs writing to a stale |
| 1108 | // transcript that happened to use the same parent session id. |
| 1109 | delete(m.artifactDirs, parentSession) |
| 1110 | delete(m.loaded, parentSession) |
| 1111 | m.mu.Unlock() |
| 1112 | m.sink.Emit(event.Event{ |
| 1113 | Kind: event.Notice, |
| 1114 | Level: event.LevelWarn, |
| 1115 | Text: "Ignoring SetActiveSessionPath with invalid session path", |
| 1116 | Detail: fmt.Sprintf("session %q: %v", parentSession, err), |
| 1117 | }) |
| 1118 | return |
| 1119 | } |
| 1120 | m.mu.Lock() |
| 1121 | m.active = parentSession |
| 1122 | oldDir := m.artifactDirLocked(parentSession) |
| 1123 | adoptDefault := false |
| 1124 | if _, hasDir := m.artifactDirs[parentSession]; !hasDir && m.hasUnscopedJobsLocked() { |
| 1125 | oldDir = m.artifactDirLocked("") |
| 1126 | adoptDefault = true |
| 1127 | } |
| 1128 | newDir := ArtifactDir(sessionPath) |
| 1129 | m.artifactDirs[parentSession] = newDir |
| 1130 | loaded := m.loaded[parentSession] |
| 1131 | m.mu.Unlock() |
| 1132 | |
| 1133 | var migrationErr error |
| 1134 | if oldDir != "" && newDir != "" && oldDir != newDir { |
| 1135 | oldSession := parentSession |
| 1136 | if adoptDefault { |
| 1137 | oldSession = "" |
| 1138 | } |
| 1139 | migrationErr = m.migrateArtifactDirForSession(oldSession, oldDir, newDir) |
| 1140 | } |
| 1141 | if adoptDefault { |
| 1142 | m.mu.Lock() |
| 1143 | adopted := m.adoptUnscopedJobsLocked(parentSession) |
| 1144 | m.mu.Unlock() |
| 1145 | for _, j := range adopted { |
| 1146 | j.mu.Lock() |
| 1147 | st := j.artifactStatus |
| 1148 | if st == "" { |
| 1149 | st = j.status |
| 1150 | } |
| 1151 | if err := m.writeJobMetaLocked(j, st); err != nil { |
| 1152 | j.noteArtifactErr("ownership metadata: " + err.Error()) |
| 1153 | } |
| 1154 | j.mu.Unlock() |
| 1155 | } |
| 1156 | } |
| 1157 | if migrationErr != nil { |
| 1158 | m.recordArtifactMigrationError(parentSession, migrationErr) |
| 1159 | } |
| 1160 | if !loaded { |
| 1161 | m.loadSessionArtifacts(parentSession, sessionPath, newDir) |
| 1162 | } |
| 1163 | } |
| 1164 | |
| 1165 | func (m *Manager) hasUnscopedJobsLocked() bool { |
| 1166 | for _, j := range m.jobs { |
| 1167 | if j != nil && strings.TrimSpace(j.SessionID) == "" { |
| 1168 | return true |
| 1169 | } |
| 1170 | } |
| 1171 | return false |
| 1172 | } |
| 1173 | |
| 1174 | func (m *Manager) adoptUnscopedJobsLocked(parentSession string) []*Job { |
| 1175 | var adopted []*Job |
| 1176 | parentSession = strings.TrimSpace(parentSession) |
| 1177 | if parentSession == "" { |
| 1178 | return adopted |
| 1179 | } |
| 1180 | for i := range m.completed { |
| 1181 | if strings.TrimSpace(m.completed[i].sessionID) == "" { |
| 1182 | m.completed[i].sessionID = parentSession |
| 1183 | } |
| 1184 | } |
| 1185 | for oldKey, j := range m.jobs { |
| 1186 | if j == nil || strings.TrimSpace(j.SessionID) != "" { |
| 1187 | continue |
| 1188 | } |
| 1189 | newKey := jobKey(parentSession, j.ID) |
| 1190 | if existing := m.jobs[newKey]; existing != nil && existing != j { |
| 1191 | j.mu.Lock() |
| 1192 | j.artifactErr = "migration: job id collision while adopting temporary session" |
| 1193 | j.artifactComplete = false |
| 1194 | j.mu.Unlock() |
| 1195 | continue |
| 1196 | } |
| 1197 | delete(m.jobs, oldKey) |
| 1198 | // Manager readers and artifact writers use distinct locks. Ownership |
| 1199 | // changes hold both, always in manager-before-job order. |
| 1200 | j.mu.Lock() |
| 1201 | j.SessionID = parentSession |
| 1202 | j.mu.Unlock() |
| 1203 | adopted = append(adopted, j) |
| 1204 | m.jobs[newKey] = j |
| 1205 | for i, key := range m.order { |
| 1206 | if key == oldKey { |
| 1207 | m.order[i] = newKey |
| 1208 | } |
| 1209 | } |
| 1210 | } |
| 1211 | return adopted |
| 1212 | } |
| 1213 | |
| 1214 | func (m *Manager) recordArtifactMigrationError(parentSession string, err error) { |
| 1215 | text := "job artifact migration failed: " + err.Error() |
| 1216 | m.mu.Lock() |
| 1217 | for _, j := range m.jobs { |
| 1218 | if j == nil || !sessionMatches(parentSession, j.SessionID) { |
| 1219 | continue |
| 1220 | } |
| 1221 | j.mu.Lock() |
| 1222 | if j.artifactErr == "" { |
| 1223 | j.artifactErr = "migration: " + err.Error() |
| 1224 | j.artifactComplete = false |
| 1225 | } |
| 1226 | j.mu.Unlock() |
| 1227 | } |
| 1228 | active := m.active |
| 1229 | m.mu.Unlock() |
| 1230 | if active == "" || active == parentSession { |
| 1231 | m.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: "Job artifact migration failed.", Detail: text}) |
| 1232 | } |
| 1233 | } |
| 1234 | |
| 1235 | type artifactMigrationJob struct { |
| 1236 | job *Job |
| 1237 | wasOpen bool |
| 1238 | } |
| 1239 | |
| 1240 | func (m *Manager) migrateArtifactDirForSession(parentSession, oldDir, newDir string) error { |
| 1241 | locked := m.lockArtifactJobsForMigration(parentSession, oldDir) |
| 1242 | defer unlockArtifactMigrationJobs(locked) |
| 1243 | skip := openArtifactMigrationFiles(locked) |
| 1244 | migrateErr := migrateArtifactDirSkipping(oldDir, newDir, skip) |
| 1245 | if migrateErr == nil { |
| 1246 | rebaseArtifactMigrationJobs(locked, newDir) |
| 1247 | } |
| 1248 | return migrateErr |
| 1249 | } |
| 1250 | |
| 1251 | func (m *Manager) lockArtifactJobsForMigration(parentSession, dir string) []artifactMigrationJob { |
| 1252 | parentSession = strings.TrimSpace(parentSession) |
| 1253 | dir = filepath.Clean(strings.TrimSpace(dir)) |
| 1254 | m.mu.Lock() |
| 1255 | jobs := make([]*Job, 0, len(m.jobs)) |
| 1256 | for _, j := range m.jobs { |
| 1257 | if j == nil || strings.TrimSpace(j.SessionID) != parentSession { |
| 1258 | continue |
| 1259 | } |
| 1260 | jobs = append(jobs, j) |
| 1261 | } |
| 1262 | m.mu.Unlock() |
| 1263 | sort.Slice(jobs, func(i, k int) bool { |
| 1264 | return jobs[i].ID < jobs[k].ID |
| 1265 | }) |
| 1266 | locked := make([]artifactMigrationJob, 0, len(jobs)) |
| 1267 | for _, j := range jobs { |
| 1268 | j.mu.Lock() |
| 1269 | if !artifactPathInDir(j.artifactPath, dir) { |
| 1270 | j.mu.Unlock() |
| 1271 | continue |
| 1272 | } |
| 1273 | locked = append(locked, artifactMigrationJob{job: j, wasOpen: j.artifactFile != nil}) |
| 1274 | } |
| 1275 | return locked |
| 1276 | } |
| 1277 | |
| 1278 | func artifactPathInDir(path, dir string) bool { |
| 1279 | path = filepath.Clean(strings.TrimSpace(path)) |
| 1280 | dir = filepath.Clean(strings.TrimSpace(dir)) |
| 1281 | if path == "." || dir == "." { |
| 1282 | return false |
| 1283 | } |
| 1284 | return filepath.Dir(path) == dir |
| 1285 | } |
| 1286 | |
| 1287 | func openArtifactMigrationFiles(jobs []artifactMigrationJob) map[string]bool { |
| 1288 | skip := map[string]bool{} |
| 1289 | for _, item := range jobs { |
| 1290 | j := item.job |
| 1291 | if j == nil || j.artifactFile == nil { |
| 1292 | continue |
| 1293 | } |
| 1294 | if j.artifactPath != "" { |
| 1295 | skip[filepath.Base(j.artifactPath)] = true |
| 1296 | } |
| 1297 | if j.artifactMetaPath != "" { |
| 1298 | skip[filepath.Base(j.artifactMetaPath)] = true |
| 1299 | } |
| 1300 | } |
| 1301 | return skip |
| 1302 | } |
| 1303 | |
| 1304 | func rebaseArtifactMigrationJobs(jobs []artifactMigrationJob, dir string) { |
| 1305 | for _, item := range jobs { |
| 1306 | j := item.job |
| 1307 | if j == nil || item.wasOpen { |
| 1308 | continue |
| 1309 | } |
| 1310 | if j.artifactPath != "" { |
| 1311 | j.artifactPath = filepath.Join(dir, filepath.Base(j.artifactPath)) |
| 1312 | } |
| 1313 | if j.artifactMetaPath != "" { |
| 1314 | j.artifactMetaPath = filepath.Join(dir, filepath.Base(j.artifactMetaPath)) |
| 1315 | } |
| 1316 | } |
| 1317 | } |
| 1318 | |
| 1319 | func unlockArtifactMigrationJobs(jobs []artifactMigrationJob) { |
| 1320 | for _, v := range slices.Backward(jobs) { |
| 1321 | if v.job != nil { |
| 1322 | v.job.mu.Unlock() |
| 1323 | } |
| 1324 | } |
| 1325 | } |
| 1326 | |
| 1327 | func migrateArtifactDir(src, dst string) error { |
| 1328 | return migrateArtifactDirSkipping(src, dst, nil) |
| 1329 | } |
| 1330 | |
| 1331 | func migrateArtifactDirSkipping(src, dst string, skip map[string]bool) error { |
| 1332 | entries, err := os.ReadDir(src) |
| 1333 | if err != nil { |
| 1334 | if os.IsNotExist(err) { |
| 1335 | return nil |
| 1336 | } |
| 1337 | return err |
| 1338 | } |
| 1339 | if err := ensurePrivateArtifactDir(dst); err != nil { |
| 1340 | return err |
| 1341 | } |
| 1342 | for _, entry := range entries { |
| 1343 | if entry.IsDir() { |
| 1344 | continue |
| 1345 | } |
| 1346 | if skip[entry.Name()] { |
| 1347 | continue |
| 1348 | } |
| 1349 | if err := moveArtifactFile(filepath.Join(src, entry.Name()), filepath.Join(dst, entry.Name())); err != nil { |
| 1350 | return err |
| 1351 | } |
| 1352 | } |
| 1353 | _ = os.Remove(src) |
| 1354 | return nil |
| 1355 | } |
| 1356 | |
| 1357 | func moveArtifactFile(src, dst string) error { |
| 1358 | // A rename preserves the source mode, so tighten legacy artifacts before |
| 1359 | // either the fast rename or the cross-device copy fallback. |
| 1360 | if err := os.Chmod(src, 0o600); err != nil { |
| 1361 | return err |
| 1362 | } |
| 1363 | if err := renamePath(src, dst); err == nil { |
| 1364 | return nil |
| 1365 | } |
| 1366 | if err := copyArtifactFile(src, dst); err != nil { |
| 1367 | return err |
| 1368 | } |
| 1369 | return os.Remove(src) |
| 1370 | } |
| 1371 | |
| 1372 | func copyArtifactFile(src, dst string) error { |
| 1373 | in, err := os.Open(src) |
| 1374 | if err != nil { |
| 1375 | return err |
| 1376 | } |
| 1377 | defer in.Close() |
| 1378 | out, err := os.OpenFile(dst, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) |
| 1379 | if err != nil { |
| 1380 | return err |
| 1381 | } |
| 1382 | if err := out.Chmod(0o600); err != nil { |
| 1383 | _ = out.Close() |
| 1384 | _ = os.Remove(dst) |
| 1385 | return err |
| 1386 | } |
| 1387 | _, copyErr := io.Copy(out, in) |
| 1388 | closeErr := out.Close() |
| 1389 | if copyErr != nil { |
| 1390 | _ = os.Remove(dst) |
| 1391 | return copyErr |
| 1392 | } |
| 1393 | if closeErr != nil { |
| 1394 | _ = os.Remove(dst) |
| 1395 | return closeErr |
| 1396 | } |
| 1397 | return nil |
| 1398 | } |
| 1399 | |
| 1400 | func (m *Manager) loadSessionArtifacts(parentSession, sessionPath, dir string) { |
| 1401 | entries, err := os.ReadDir(dir) |
| 1402 | if err != nil { |
| 1403 | m.mu.Lock() |
| 1404 | m.loaded[parentSession] = true |
| 1405 | m.mu.Unlock() |
| 1406 | return |
| 1407 | } |
| 1408 | var loaded []*Job |
| 1409 | deferredLiveOwner := false |
| 1410 | var repairErrors []string |
| 1411 | maxSeq := 0 |
| 1412 | for _, entry := range entries { |
| 1413 | if entry.IsDir() || filepath.Ext(entry.Name()) != jobMetaExt { |
| 1414 | continue |
| 1415 | } |
| 1416 | metaPath := filepath.Join(dir, entry.Name()) |
| 1417 | meta, err := readMeta(metaPath) |
| 1418 | if err != nil || strings.TrimSpace(meta.ID) == "" { |
| 1419 | continue |
| 1420 | } |
| 1421 | id := strings.TrimSpace(meta.ID) |
| 1422 | if seq := maxJobSeq(id); seq > maxSeq { |
| 1423 | maxSeq = seq |
| 1424 | } |
| 1425 | // A persisted Running record may belong to another manager in this |
| 1426 | // process or to another Reasonix process entirely. Only the runtime that |
| 1427 | // owns the session lease may repair an abandoned record as Interrupted. |
| 1428 | // Observers without proof of ownership defer the artifact and leave the |
| 1429 | // session reloadable for a later owned bind. |
| 1430 | if meta.Status == Running { |
| 1431 | if managerOwnerIsLive(meta.OwnerID) { |
| 1432 | deferredLiveOwner = true |
| 1433 | continue |
| 1434 | } |
| 1435 | if m.sessionOwnershipProbe == nil || !m.sessionOwnershipProbe(sessionPath) { |
| 1436 | deferredLiveOwner = true |
| 1437 | continue |
| 1438 | } |
| 1439 | meta.Status = Interrupted |
| 1440 | if meta.FinishedAt == 0 { |
| 1441 | meta.FinishedAt = nowMs() |
| 1442 | } |
| 1443 | meta.ArtifactComplete = false |
| 1444 | if err := repairArtifactMeta(metaPath, meta); err != nil { |
| 1445 | // Do not publish an in-memory Interrupted tombstone when the durable |
| 1446 | // state still says Running. Keep the session reloadable so a later bind |
| 1447 | // can retry the repair, and surface the failure instead of letting live |
| 1448 | // and machine-facing status silently disagree. |
| 1449 | deferredLiveOwner = true |
| 1450 | repairErrors = append(repairErrors, fmt.Sprintf("repair job %s metadata: %v", id, err)) |
| 1451 | continue |
| 1452 | } |
| 1453 | } |
| 1454 | done := make(chan struct{}) |
| 1455 | close(done) |
| 1456 | logPath := filepath.Join(dir, id+jobLogExt) |
| 1457 | if strings.TrimSpace(meta.LogPath) != "" { |
| 1458 | logPath = filepath.Join(dir, filepath.Base(meta.LogPath)) |
| 1459 | } |
| 1460 | loaded = append(loaded, &Job{ |
| 1461 | ID: id, |
| 1462 | Kind: meta.Kind, |
| 1463 | Label: meta.Label, |
| 1464 | SessionID: parentSession, |
| 1465 | status: meta.Status, |
| 1466 | clock: jobClock{startedAt: meta.StartedAt, finishedAt: meta.FinishedAt, activityAt: meta.FinishedAt}, |
| 1467 | done: done, |
| 1468 | artifactPath: logPath, |
| 1469 | artifactMetaPath: filepath.Join(dir, id+jobMetaExt), |
| 1470 | artifactComplete: meta.ArtifactComplete, |
| 1471 | artifactErr: meta.ArtifactError, |
| 1472 | tombstone: true, |
| 1473 | evidence: mutationEvidenceFromArtifact(meta), |
| 1474 | }) |
| 1475 | } |
| 1476 | if len(repairErrors) > 0 { |
| 1477 | m.sink.Emit(event.Event{ |
| 1478 | Kind: event.Notice, |
| 1479 | Level: event.LevelWarn, |
| 1480 | Text: "Background job recovery did not complete.", |
| 1481 | Detail: strings.Join(repairErrors, "; "), |
| 1482 | }) |
| 1483 | } |
| 1484 | m.mu.Lock() |
| 1485 | defer m.mu.Unlock() |
| 1486 | for _, j := range loaded { |
| 1487 | key := jobKey(parentSession, j.ID) |
| 1488 | if _, exists := m.jobs[key]; exists { |
| 1489 | continue |
| 1490 | } |
| 1491 | m.jobs[key] = j |
| 1492 | m.order = append(m.order, key) |
| 1493 | } |
| 1494 | if maxSeq > m.seq { |
| 1495 | m.seq = maxSeq |
| 1496 | } |
| 1497 | m.loaded[parentSession] = !deferredLiveOwner |
| 1498 | } |
| 1499 | |
| 1500 | // BeginDestroySession marks a parent session as being removed from active use |
| 1501 | // and cancels its running jobs. WaitTeardown waits for the returned handle. |
| 1502 | func (m *Manager) BeginDestroySession(parentSession string) SessionTeardown { |
| 1503 | parentSession = strings.TrimSpace(parentSession) |
| 1504 | if parentSession == "" { |
| 1505 | return SessionTeardown{} |
| 1506 | } |
| 1507 | var cancels []context.CancelFunc |
| 1508 | var targets []teardownTarget |
| 1509 | m.mu.Lock() |
| 1510 | m.destroying[parentSession] = true |
| 1511 | remaining := m.completed[:0] |
| 1512 | for _, item := range m.completed { |
| 1513 | if item.sessionID != parentSession { |
| 1514 | remaining = append(remaining, item) |
| 1515 | } |
| 1516 | } |
| 1517 | m.completed = remaining |
| 1518 | for _, key := range m.order { |
| 1519 | j := m.jobs[key] |
| 1520 | if !sessionMatches(parentSession, j.SessionID) { |
| 1521 | continue |
| 1522 | } |
| 1523 | j.mu.Lock() |
| 1524 | switch j.status { |
| 1525 | case Running: |
| 1526 | j.status = Killed |
| 1527 | cancels = append(cancels, j.cancel) |
| 1528 | targets = append(targets, teardownTarget{info: TeardownJob{ID: j.ID, Kind: j.Kind, Label: j.Label}, done: j.done}) |
| 1529 | case Killed: |
| 1530 | targets = append(targets, teardownTarget{info: TeardownJob{ID: j.ID, Kind: j.Kind, Label: j.Label}, done: j.done}) |
| 1531 | } |
| 1532 | j.mu.Unlock() |
| 1533 | } |
| 1534 | m.mu.Unlock() |
| 1535 | for _, cancel := range cancels { |
| 1536 | cancel() |
| 1537 | } |
| 1538 | return SessionTeardown{SessionID: parentSession, targets: targets} |
| 1539 | } |
| 1540 | |
| 1541 | // DestroySession preserves the legacy channel-based destroy API. |
| 1542 | func (m *Manager) DestroySession(parentSession string) []<-chan struct{} { |
| 1543 | return m.BeginDestroySession(parentSession).DoneChannels() |
| 1544 | } |
| 1545 | |
| 1546 | // WaitTeardown waits for a destroy handle to unwind up to grace. A timed-out |
| 1547 | // result means the caller should defer physical cleanup until the jobs exit. |
| 1548 | func (m *Manager) WaitTeardown(ctx context.Context, h SessionTeardown, grace time.Duration) TeardownResult { |
| 1549 | result, timedOut := waitTeardownTargets(ctx, h.targets, grace) |
| 1550 | if timedOut { |
| 1551 | m.emitTeardownTimeout("destroy session "+h.SessionID, result) |
| 1552 | } |
| 1553 | return result |
| 1554 | } |
| 1555 | |
| 1556 | // IsDestroying reports whether parentSession is in the destroy window. Empty |
| 1557 | // parent sessions are never considered destroyed. |
| 1558 | func (m *Manager) IsDestroying(parentSession string) bool { |
| 1559 | parentSession = strings.TrimSpace(parentSession) |
| 1560 | if parentSession == "" { |
| 1561 | return false |
| 1562 | } |
| 1563 | m.mu.Lock() |
| 1564 | defer m.mu.Unlock() |
| 1565 | return m.destroying[parentSession] |
| 1566 | } |
| 1567 | |
| 1568 | // FinishDestroySession ends the destroy window after all owned jobs have unwound |
| 1569 | // and persistent cleanup/move work has completed. |
| 1570 | func (m *Manager) FinishDestroySession(parentSession string) { |
| 1571 | parentSession = strings.TrimSpace(parentSession) |
| 1572 | if parentSession == "" { |
| 1573 | return |
| 1574 | } |
| 1575 | m.mu.Lock() |
| 1576 | delete(m.destroying, parentSession) |
| 1577 | delete(m.artifactDirs, parentSession) |
| 1578 | delete(m.loaded, parentSession) |
| 1579 | m.purgeSessionLocked(parentSession) |
| 1580 | m.mu.Unlock() |
| 1581 | } |
| 1582 | |
| 1583 | func (m *Manager) purgeSessionLocked(parentSession string) { |
| 1584 | kept := m.order[:0] |
| 1585 | for _, key := range m.order { |
| 1586 | j := m.jobs[key] |
| 1587 | if j == nil || sessionMatches(parentSession, j.SessionID) { |
| 1588 | delete(m.jobs, key) |
| 1589 | continue |
| 1590 | } |
| 1591 | kept = append(kept, key) |
| 1592 | } |
| 1593 | m.order = kept |
| 1594 | } |
| 1595 | |
| 1596 | // Close cancels the session context and waits briefly for every background job |
| 1597 | // goroutine to return before unblocking. If a non-cooperative job ignores |
| 1598 | // cancellation, cleanup of the temporary artifact root continues in the |
| 1599 | // background after the goroutines eventually unwind. |
| 1600 | func (m *Manager) Close() { |
| 1601 | _ = m.CloseWithGrace(m.teardownGrace) |
| 1602 | } |
| 1603 | |
| 1604 | // CloseAsync cancels the manager and returns immediately. It is used when a |
| 1605 | // caller has already begun session-specific teardown and owns the delayed |
| 1606 | // persistent cleanup, but still needs the manager's root context and temporary |
| 1607 | // artifact root released eventually. |
| 1608 | func (m *Manager) CloseAsync() { |
| 1609 | m.cancel() |
| 1610 | go func() { |
| 1611 | m.wg.Wait() |
| 1612 | m.releaseOwner() |
| 1613 | m.removeTempRoot() |
| 1614 | }() |
| 1615 | } |
| 1616 | |
| 1617 | // CloseWithGrace is Close with an explicit wait window, used by tests and |
| 1618 | // callers that need to surface non-cooperative jobs. |
| 1619 | func (m *Manager) CloseWithGrace(grace time.Duration) TeardownResult { |
| 1620 | m.cancel() |
| 1621 | done := make(chan struct{}) |
| 1622 | go func() { |
| 1623 | m.wg.Wait() |
| 1624 | m.releaseOwner() |
| 1625 | close(done) |
| 1626 | }() |
| 1627 | result, timedOut := waitTeardownTargets(context.Background(), m.closeTargets(), grace, done) |
| 1628 | if timedOut { |
| 1629 | m.emitTeardownTimeout("close", result) |
| 1630 | go func() { |
| 1631 | <-done |
| 1632 | m.removeTempRoot() |
| 1633 | }() |
| 1634 | return result |
| 1635 | } |
| 1636 | m.removeTempRoot() |
| 1637 | return result |
| 1638 | } |
| 1639 | |
| 1640 | func waitTeardownTargets(ctx context.Context, targets []teardownTarget, grace time.Duration, allDone ...<-chan struct{}) (TeardownResult, bool) { |
| 1641 | if ctx == nil { |
| 1642 | ctx = context.Background() |
| 1643 | } |
| 1644 | start := time.Now() |
| 1645 | var timeout <-chan time.Time |
| 1646 | if grace >= 0 { |
| 1647 | timer := time.NewTimer(grace) |
| 1648 | defer timer.Stop() |
| 1649 | timeout = timer.C |
| 1650 | } |
| 1651 | if len(allDone) > 0 && allDone[0] != nil { |
| 1652 | select { |
| 1653 | case <-allDone[0]: |
| 1654 | return TeardownResult{}, false |
| 1655 | case <-ctx.Done(): |
| 1656 | return teardownTimedOut(targets, time.Since(start)), false |
| 1657 | case <-timeout: |
| 1658 | return teardownTimedOut(targets, time.Since(start)), true |
| 1659 | } |
| 1660 | } |
| 1661 | for _, target := range targets { |
| 1662 | select { |
| 1663 | case <-target.done: |
| 1664 | case <-ctx.Done(): |
| 1665 | return teardownTimedOut(targets, time.Since(start)), false |
| 1666 | case <-timeout: |
| 1667 | return teardownTimedOut(targets, time.Since(start)), true |
| 1668 | } |
| 1669 | } |
| 1670 | return TeardownResult{}, false |
| 1671 | } |
| 1672 | |
| 1673 | func teardownTimedOut(targets []teardownTarget, waited time.Duration) TeardownResult { |
| 1674 | var out []TeardownJob |
| 1675 | for _, target := range targets { |
| 1676 | select { |
| 1677 | case <-target.done: |
| 1678 | continue |
| 1679 | default: |
| 1680 | } |
| 1681 | info := target.info |
| 1682 | info.Waited = waited |
| 1683 | out = append(out, info) |
| 1684 | } |
| 1685 | return TeardownResult{TimedOut: out} |
| 1686 | } |
| 1687 | |
| 1688 | func (m *Manager) closeTargets() []teardownTarget { |
| 1689 | m.mu.Lock() |
| 1690 | defer m.mu.Unlock() |
| 1691 | var targets []teardownTarget |
| 1692 | for _, key := range m.order { |
| 1693 | j := m.jobs[key] |
| 1694 | if j == nil { |
| 1695 | continue |
| 1696 | } |
| 1697 | select { |
| 1698 | case <-j.done: |
| 1699 | continue |
| 1700 | default: |
| 1701 | } |
| 1702 | j.mu.Lock() |
| 1703 | switch j.status { |
| 1704 | case Running: |
| 1705 | j.status = Killed |
| 1706 | targets = append(targets, teardownTarget{info: TeardownJob{ID: j.ID, Kind: j.Kind, Label: j.Label}, done: j.done}) |
| 1707 | case Killed: |
| 1708 | targets = append(targets, teardownTarget{info: TeardownJob{ID: j.ID, Kind: j.Kind, Label: j.Label}, done: j.done}) |
| 1709 | } |
| 1710 | j.mu.Unlock() |
| 1711 | } |
| 1712 | return targets |
| 1713 | } |
| 1714 | |
| 1715 | func (m *Manager) emitTeardownTimeout(action string, result TeardownResult) { |
| 1716 | if len(result.TimedOut) == 0 { |
| 1717 | return |
| 1718 | } |
| 1719 | var b strings.Builder |
| 1720 | fmt.Fprintf(&b, "background job teardown timed out during %s", strings.TrimSpace(action)) |
| 1721 | for i, job := range result.TimedOut { |
| 1722 | if i == 0 { |
| 1723 | b.WriteString(": ") |
| 1724 | } else { |
| 1725 | b.WriteString("; ") |
| 1726 | } |
| 1727 | fmt.Fprintf(&b, "%s kind=%s", job.ID, job.Kind) |
| 1728 | if strings.TrimSpace(job.Label) != "" { |
| 1729 | fmt.Fprintf(&b, " label=%q", job.Label) |
| 1730 | } |
| 1731 | if job.Waited > 0 { |
| 1732 | fmt.Fprintf(&b, " waited=%s", job.Waited.Round(time.Millisecond)) |
| 1733 | } |
| 1734 | } |
| 1735 | m.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: "Background job teardown timed out.", Detail: b.String()}) |
| 1736 | } |
| 1737 | |
| 1738 | func (m *Manager) removeTempRoot() { |
| 1739 | if m.tempRoot != "" { |
| 1740 | _ = os.RemoveAll(m.tempRoot) |
| 1741 | } |
| 1742 | } |
| 1743 | |
| 1744 | func nowMs() int64 { return time.Now().UnixMilli() } |
| 1745 | |
| 1746 | func startedText(kind, id, label string) string { |
| 1747 | if label != "" { |
| 1748 | return fmt.Sprintf("background %s started: %s (%s)", kind, id, label) |
| 1749 | } |
| 1750 | return fmt.Sprintf("background %s started: %s", kind, id) |
| 1751 | } |
| 1752 | |
| 1753 | func (m *Manager) emitIfActive(parentSession string, ev event.Event) { |
| 1754 | m.mu.Lock() |
| 1755 | active := m.active |
| 1756 | m.mu.Unlock() |
| 1757 | if active == "" || strings.TrimSpace(parentSession) == "" || active == strings.TrimSpace(parentSession) { |
| 1758 | m.sink.Emit(ev) |
| 1759 | } |
| 1760 | } |
| 1761 | |
| 1762 | func sessionMatches(filter, jobSession string) bool { |
| 1763 | filter = strings.TrimSpace(filter) |
| 1764 | return filter == "" || strings.TrimSpace(jobSession) == filter |
| 1765 | } |
| 1766 | |
| 1767 | func jobKey(parentSession, id string) string { |
| 1768 | return strings.TrimSpace(parentSession) + "\x00" + strings.TrimSpace(id) |
| 1769 | } |
| 1770 | |
| 1771 | // call-context injection (mirrors agent.CallContext) |
| 1772 | |
| 1773 | type ctxKey struct{} |
| 1774 | type sessionCtxKey struct{} |
| 1775 | type jobCtxKey struct{} |
| 1776 | type noManager struct{} |
| 1777 | |
| 1778 | // WithManager stamps ctx with the job manager so tools can reach it via |
| 1779 | // FromContext. The agent sets this on every tool call's context. |
| 1780 | func WithManager(ctx context.Context, m *Manager) context.Context { |
| 1781 | return context.WithValue(ctx, ctxKey{}, m) |
| 1782 | } |
| 1783 | |
| 1784 | // WithoutManager shadows an ancestor manager. Child agents without an owned |
| 1785 | // Jobs manager must not operate the parent's background jobs by inheritance. |
| 1786 | func WithoutManager(ctx context.Context) context.Context { |
| 1787 | return context.WithValue(ctx, ctxKey{}, noManager{}) |
| 1788 | } |
| 1789 | |
| 1790 | // FromContext returns the job manager set by the agent, if any. ok is false for a |
| 1791 | // plain context (headless tests, calls outside the run loop). |
| 1792 | func FromContext(ctx context.Context) (*Manager, bool) { |
| 1793 | m, ok := ctx.Value(ctxKey{}).(*Manager) |
| 1794 | return m, ok && m != nil |
| 1795 | } |
| 1796 | |
| 1797 | // WithSession stamps ctx with the active parent session ID for session-scoped job |
| 1798 | // operations. |
| 1799 | func WithSession(ctx context.Context, parentSession string) context.Context { |
| 1800 | return context.WithValue(ctx, sessionCtxKey{}, strings.TrimSpace(parentSession)) |
| 1801 | } |
| 1802 | |
| 1803 | // SessionFromContext returns the active parent session ID for job ownership and |
| 1804 | // filtering. Empty means no session scope is available. |
| 1805 | func SessionFromContext(ctx context.Context) string { |
| 1806 | session, _ := ctx.Value(sessionCtxKey{}).(string) |
| 1807 | return strings.TrimSpace(session) |
| 1808 | } |
| 1809 | |
| 1810 | // PublishEvidence attaches a background agent's host-observed receipts to its |
| 1811 | // job. The receipts stay independent of the parent turn ledger until the |
| 1812 | // parent collects the terminal result with wait or bash_output. |
| 1813 | func PublishEvidence(ctx context.Context, summary evidence.ChildEvidenceSummary) { |
| 1814 | j, _ := ctx.Value(jobCtxKey{}).(*Job) |
| 1815 | if j == nil || len(summary.Receipts) == 0 { |
| 1816 | return |
| 1817 | } |
| 1818 | j.mu.Lock() |
| 1819 | mergePublishedEvidence(&j.evidence, summary) |
| 1820 | j.mu.Unlock() |
| 1821 | } |
| 1822 | |
| 1823 | // LeaseEvidenceForSession returns a copy of a terminal job's evidence without |
| 1824 | // consuming it. Collection is only provisional: the receipts merge into the |
| 1825 | // collecting turn's ledger, but that ledger is discarded if the turn is |
| 1826 | // cancelled, errors, or the process exits before the turn commits. Consuming |
| 1827 | // here would then lose the mutation for good — the parent's next turn resets its |
| 1828 | // ledger and this job would report nothing, so a background change would ship |
| 1829 | // unreviewed. The evidence is drained only by CommitEvidenceForSession, which |
| 1830 | // the agent calls after the collecting turn passes its delivery gates. A |
| 1831 | // committed job returns empty so a re-poll after successful delivery does not |
| 1832 | // re-demand review. |
| 1833 | func (m *Manager) LeaseEvidenceForSession(parentSession, id string) evidence.ChildEvidenceSummary { |
| 1834 | summary, _ := m.tryLeaseEvidenceForSession(parentSession, id) |
| 1835 | return summary |
| 1836 | } |
| 1837 | |
| 1838 | // TryLeaseEvidenceForSession is LeaseEvidenceForSession plus a ready flag that |
| 1839 | // separates "terminal evidence available" (possibly empty — a committed job or |
| 1840 | // one with no mutations) from "not ready to lease yet": unknown job, still |
| 1841 | // running, or killed but its run goroutine has not yet flushed PublishEvidence |
| 1842 | // and closed done. KillForSession flips status to Killed synchronously, well |
| 1843 | // before the goroutine actually returns, so a bash_output poll that lands in |
| 1844 | // that window must not treat the empty read as final. Callers that record a |
| 1845 | // lease (collectBackgroundEvidence) must gate on ready so they never note a |
| 1846 | // lease before the evidence exists — noting it early would let a later commit |
| 1847 | // drain evidence nobody ever merged or reviewed. |
| 1848 | func (m *Manager) TryLeaseEvidenceForSession(parentSession, id string) (evidence.ChildEvidenceSummary, bool) { |
| 1849 | return m.tryLeaseEvidenceForSession(parentSession, id) |
| 1850 | } |
| 1851 | |
| 1852 | func (m *Manager) tryLeaseEvidenceForSession(parentSession, id string) (evidence.ChildEvidenceSummary, bool) { |
| 1853 | j := m.get(parentSession, id) |
| 1854 | if j == nil { |
| 1855 | return evidence.ChildEvidenceSummary{}, false |
| 1856 | } |
| 1857 | j.mu.Lock() |
| 1858 | defer j.mu.Unlock() |
| 1859 | select { |
| 1860 | case <-j.done: |
| 1861 | default: |
| 1862 | return evidence.ChildEvidenceSummary{}, false |
| 1863 | } |
| 1864 | if j.evidenceCommitted { |
| 1865 | return evidence.ChildEvidenceSummary{}, true |
| 1866 | } |
| 1867 | out := make([]evidence.Receipt, len(j.evidence.Receipts)) |
| 1868 | copy(out, j.evidence.Receipts) |
| 1869 | return evidence.ChildEvidenceSummary{Receipts: out, WorkspaceRoot: j.evidence.WorkspaceRoot}, true |
| 1870 | } |
| 1871 | |
| 1872 | // PendingEvidenceJobIDsForSession returns the IDs of parentSession's terminal |
| 1873 | // jobs that carry uncommitted mutation evidence — a prior turn leased it but |
| 1874 | // never delivered (the turn failed or was cancelled, and the next turn's Reset |
| 1875 | // wiped it from the per-turn ledger), or the process restarted before any turn |
| 1876 | // collected it at all. The agent re-leases these at the start of every turn so |
| 1877 | // a turn that never calls wait/bash_output still surfaces the pending mutation |
| 1878 | // to its final-readiness checks instead of silently shipping it unreviewed. |
| 1879 | func (m *Manager) PendingEvidenceJobIDsForSession(parentSession string) []string { |
| 1880 | m.mu.Lock() |
| 1881 | defer m.mu.Unlock() |
| 1882 | var ids []string |
| 1883 | for _, key := range m.order { |
| 1884 | j := m.jobs[key] |
| 1885 | if j == nil || !sessionMatches(parentSession, j.SessionID) { |
| 1886 | continue |
| 1887 | } |
| 1888 | j.mu.Lock() |
| 1889 | terminal := false |
| 1890 | select { |
| 1891 | case <-j.done: |
| 1892 | terminal = true |
| 1893 | default: |
| 1894 | } |
| 1895 | pending := terminal && !j.evidenceCommitted && len(j.evidence.Receipts) > 0 |
| 1896 | j.mu.Unlock() |
| 1897 | if pending { |
| 1898 | ids = append(ids, j.ID) |
| 1899 | } |
| 1900 | } |
| 1901 | return ids |
| 1902 | } |
| 1903 | |
| 1904 | // CommitEvidenceForSession permanently consumes a terminal job's evidence after |
| 1905 | // the collecting turn has accounted for it (passed final-readiness). It clears |
| 1906 | // the in-memory copy and drains the persisted mutation summary so neither a |
| 1907 | // same-process re-poll nor a restart resurrects receipts the delivered turn |
| 1908 | // already reviewed. Best-effort on the disk rewrite — a failed rewrite merely |
| 1909 | // restores the conservative resurrection behavior. |
| 1910 | func (m *Manager) CommitEvidenceForSession(parentSession, id string) { |
| 1911 | j := m.get(parentSession, id) |
| 1912 | if j == nil { |
| 1913 | return |
| 1914 | } |
| 1915 | j.mu.Lock() |
| 1916 | defer j.mu.Unlock() |
| 1917 | select { |
| 1918 | case <-j.done: |
| 1919 | default: |
| 1920 | return |
| 1921 | } |
| 1922 | if j.evidenceCommitted { |
| 1923 | return |
| 1924 | } |
| 1925 | hadEvidence := len(j.evidence.Receipts) > 0 |
| 1926 | j.evidenceCommitted = true |
| 1927 | j.evidence = evidence.ChildEvidenceSummary{} |
| 1928 | if hadEvidence { |
| 1929 | if err := m.writeJobMetaLocked(j, j.status); err != nil { |
| 1930 | j.noteArtifactErr("evidence drain: " + err.Error()) |
| 1931 | } |
| 1932 | } |
| 1933 | } |
| 1934 |