| 1 | package taskmonitor |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | crand "crypto/rand" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "sync" |
| 11 | "sync/atomic" |
| 12 | "time" |
| 13 | |
| 14 | "reasonix/internal/jobs" |
| 15 | ) |
| 16 | |
| 17 | // TaskRecorder bridges jobs.Manager lifecycle events into the Task Store. It |
| 18 | // is the write side of task monitoring: RecordStart persists a running |
| 19 | // snapshot, RecordDone advances it to its terminal state. All failures are |
| 20 | // swallowed — monitoring is best-effort and must never break the job pipeline. |
| 21 | // The store's per-task lock keeps concurrent recorders (CLI + Desktop) safe. |
| 22 | type TaskRecorder struct { |
| 23 | store WriteStore |
| 24 | projectDir string |
| 25 | sessionIDFn func() string |
| 26 | mu sync.Mutex |
| 27 | monitorIDs map[string]string |
| 28 | heartbeats map[string]context.CancelFunc |
| 29 | runtimeOwnerID string |
| 30 | } |
| 31 | |
| 32 | // NewTaskRecorder returns a TaskRecorder writing to store under projectDir. |
| 33 | // sessionIDFn is called per record so the snapshot reflects the session id at |
| 34 | // creation time (controllers resolve their session path lazily); it may return |
| 35 | // "" when no session is bound yet. |
| 36 | func NewTaskRecorder(store WriteStore, projectDir string, sessionIDFn func() string) *TaskRecorder { |
| 37 | return &TaskRecorder{store: store, projectDir: projectDir, sessionIDFn: sessionIDFn, monitorIDs: make(map[string]string), heartbeats: make(map[string]context.CancelFunc), runtimeOwnerID: newRuntimeOwnerID()} |
| 38 | } |
| 39 | |
| 40 | const ( |
| 41 | runtimeLeaseTTL = 30 * time.Second |
| 42 | runtimeHeartbeatEvery = 5 * time.Second |
| 43 | ) |
| 44 | |
| 45 | var runtimeOwnerSequence atomic.Uint64 |
| 46 | |
| 47 | func newRuntimeOwnerID() string { |
| 48 | var nonce [16]byte |
| 49 | if _, err := crand.Read(nonce[:]); err == nil { |
| 50 | return hex.EncodeToString(nonce[:]) |
| 51 | } |
| 52 | h := sha256.Sum256([]byte(fmt.Sprintf("%d:%d", timeNow().UnixNano(), runtimeOwnerSequence.Add(1)))) |
| 53 | return hex.EncodeToString(h[:16]) |
| 54 | } |
| 55 | |
| 56 | // monitorTaskID creates a globally unique monitor identity for a job within |
| 57 | // a session. jobs.Manager IDs are local to a manager and restart from task-1 |
| 58 | // for every session, so persisting the raw job ID would cause cross-session |
| 59 | // overwrites in the shared project store. |
| 60 | func monitorTaskID(sessionID, jobID string) string { |
| 61 | if sessionID == "" { |
| 62 | return jobID |
| 63 | } |
| 64 | id := fmt.Sprintf("%s--%s", sessionID, jobID) |
| 65 | if len(id) <= maxFieldLen { |
| 66 | return id |
| 67 | } |
| 68 | h := sha256.Sum256([]byte(sessionID)) |
| 69 | return hex.EncodeToString(h[:8]) + "--" + jobID |
| 70 | } |
| 71 | |
| 72 | func sessionlessMonitorTaskID(jobID string) string { |
| 73 | var nonce [8]byte |
| 74 | if _, err := crand.Read(nonce[:]); err != nil { |
| 75 | // crypto/rand failure is exceptional; retain a bounded, non-path-like |
| 76 | // identity rather than falling back to the colliding raw job ID. |
| 77 | h := sha256.Sum256([]byte(fmt.Sprintf("%s:%d", jobID, timeNow().UnixNano()))) |
| 78 | return hex.EncodeToString(h[:8]) + "--" + jobID |
| 79 | } |
| 80 | return hex.EncodeToString(nonce[:]) + "--" + jobID |
| 81 | } |
| 82 | |
| 83 | func (r *TaskRecorder) rememberMonitorID(jobID, monitorID string) { |
| 84 | r.mu.Lock() |
| 85 | r.monitorIDs[jobID] = monitorID |
| 86 | r.mu.Unlock() |
| 87 | } |
| 88 | |
| 89 | func (r *TaskRecorder) lookupMonitorID(jobID string) (string, bool) { |
| 90 | r.mu.Lock() |
| 91 | monitorID, ok := r.monitorIDs[jobID] |
| 92 | r.mu.Unlock() |
| 93 | return monitorID, ok |
| 94 | } |
| 95 | |
| 96 | func (r *TaskRecorder) startHeartbeat(monitorID string) { |
| 97 | ctx, cancel := context.WithCancel(context.Background()) |
| 98 | r.mu.Lock() |
| 99 | if old := r.heartbeats[monitorID]; old != nil { |
| 100 | old() |
| 101 | } |
| 102 | r.heartbeats[monitorID] = cancel |
| 103 | r.mu.Unlock() |
| 104 | go func() { |
| 105 | ticker := time.NewTicker(runtimeHeartbeatEvery) |
| 106 | defer ticker.Stop() |
| 107 | for { |
| 108 | select { |
| 109 | case <-ctx.Done(): |
| 110 | return |
| 111 | case <-ticker.C: |
| 112 | if !r.renewHeartbeat(ctx, monitorID) { |
| 113 | return |
| 114 | } |
| 115 | } |
| 116 | } |
| 117 | }() |
| 118 | } |
| 119 | |
| 120 | func (r *TaskRecorder) renewHeartbeat(ctx context.Context, monitorID string) bool { |
| 121 | renewed, err := r.store.RenewRuntimeLease(ctx, r.projectDir, monitorID, r.runtimeOwnerID, timeNow().Add(runtimeLeaseTTL)) |
| 122 | return err == nil && renewed |
| 123 | } |
| 124 | |
| 125 | func (r *TaskRecorder) stopHeartbeat(monitorID string) { |
| 126 | r.mu.Lock() |
| 127 | if cancel := r.heartbeats[monitorID]; cancel != nil { |
| 128 | cancel() |
| 129 | delete(r.heartbeats, monitorID) |
| 130 | } |
| 131 | r.mu.Unlock() |
| 132 | } |
| 133 | |
| 134 | // RecordStart implements jobs.TaskRecorder. |
| 135 | func (r *TaskRecorder) RecordStart(id, kind, label string) { |
| 136 | ctx := context.Background() |
| 137 | now := timeNow() |
| 138 | sessionID := "" |
| 139 | if r.sessionIDFn != nil { |
| 140 | sessionID = r.sessionIDFn() |
| 141 | } |
| 142 | monitorID := monitorTaskID(sessionID, id) |
| 143 | if sessionID == "" { |
| 144 | monitorID = sessionlessMonitorTaskID(id) |
| 145 | } |
| 146 | r.rememberMonitorID(id, monitorID) |
| 147 | snap := TaskSnapshot{ |
| 148 | SchemaVersion: 1, |
| 149 | TaskID: monitorID, |
| 150 | JobID: id, |
| 151 | SessionID: sessionID, |
| 152 | State: TaskStateRunning, |
| 153 | RuntimeState: RuntimeStateAlive, |
| 154 | RuntimeLeaseUntil: now.Add(runtimeLeaseTTL), |
| 155 | RuntimeOwnerID: r.runtimeOwnerID, |
| 156 | Version: 1, |
| 157 | CreatedAt: now, |
| 158 | UpdatedAt: now, |
| 159 | } |
| 160 | if err := r.store.SaveTask(ctx, r.projectDir, snap); err != nil { |
| 161 | return |
| 162 | } |
| 163 | _ = r.store.AppendAuditEvent(ctx, r.projectDir, TaskEvent{ |
| 164 | Timestamp: now, EventType: "state_change", |
| 165 | TaskID: monitorID, SessionID: sessionID, State: TaskStateRunning, |
| 166 | RuntimeState: RuntimeStateAlive, |
| 167 | }) |
| 168 | r.startHeartbeat(monitorID) |
| 169 | } |
| 170 | |
| 171 | // RecordDone implements jobs.TaskRecorder. |
| 172 | func (r *TaskRecorder) RecordDone(id string, st jobs.Status, jobErr error) { |
| 173 | ctx := context.Background() |
| 174 | target := terminalState(st) |
| 175 | if target == "" { |
| 176 | return // non-terminal/unknown status: leave the snapshot untouched |
| 177 | } |
| 178 | monitorID, ok := r.lookupMonitorID(id) |
| 179 | if !ok { |
| 180 | return // no matching lifecycle was recorded by this recorder |
| 181 | } |
| 182 | r.stopHeartbeat(monitorID) |
| 183 | const maxSaveAttempts = 4 |
| 184 | for attempt := 0; attempt < maxSaveAttempts; attempt++ { |
| 185 | cur, gerr := r.store.GetTask(ctx, r.projectDir, monitorID) |
| 186 | if gerr != nil || cur == nil { |
| 187 | return // never recorded (recorder attached after the job started) |
| 188 | } |
| 189 | if cur.RuntimeOwnerID != "" && cur.RuntimeOwnerID != r.runtimeOwnerID { |
| 190 | return // a newer recorder generation owns this reused task identity |
| 191 | } |
| 192 | now := timeNow() |
| 193 | cur.State = target |
| 194 | cur.RuntimeState = RuntimeStateExited |
| 195 | cur.RuntimeLeaseUntil = time.Time{} |
| 196 | cur.RuntimeOwnerID = "" |
| 197 | cur.Version++ |
| 198 | cur.UpdatedAt = now |
| 199 | cur.ErrorSummary = "" |
| 200 | if jobErr != nil { |
| 201 | cur.ErrorCode = "job_failed" |
| 202 | } |
| 203 | if serr := r.store.SaveTask(ctx, r.projectDir, *cur); serr != nil { |
| 204 | if errors.Is(serr, ErrStoreVersionConflict) { |
| 205 | continue |
| 206 | } |
| 207 | return |
| 208 | } |
| 209 | _ = r.store.AppendAuditEvent(ctx, r.projectDir, TaskEvent{ |
| 210 | Timestamp: now, EventType: "state_change", |
| 211 | TaskID: monitorID, SessionID: cur.SessionID, State: target, |
| 212 | RuntimeState: RuntimeStateExited, |
| 213 | ErrorCode: cur.ErrorCode, ErrorSummary: cur.ErrorSummary, |
| 214 | }) |
| 215 | return |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // terminalState maps a job status to the task state it reports. Unknown or |
| 220 | // non-terminal statuses map to "" (no update). |
| 221 | func terminalState(st jobs.Status) TaskState { |
| 222 | switch st { |
| 223 | case jobs.Done: |
| 224 | return TaskStateSucceeded |
| 225 | case jobs.Failed: |
| 226 | return TaskStateFailed |
| 227 | case jobs.Killed, jobs.Interrupted: |
| 228 | return TaskStateCancelled |
| 229 | default: |
| 230 | return "" |
| 231 | } |
| 232 | } |
| 233 |