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