返回 DeepSeek-Reasonix
task.go
根目录 / internal / cli / task.go
1 package cli
2
3 import (
4 "context"
5 "encoding/json"
6 "flag"
7 "fmt"
8 "os"
9 "os/signal"
10 "strings"
11 "time"
12
13 "reasonix/internal/taskmonitor"
14 )
15
16 const (
17 taskCommandUsage = "usage: reasonix task <list|show|monitor|status|events|stop|cancel|requeue|open-session|tmux> [flags]"
18 taskMonitorUsage = "usage: reasonix task monitor <list|status|events|stop|cancel|requeue|open-session> [flags]"
19 taskTmuxUsage = "usage: reasonix task tmux <attach|status|open|detach>"
20 )
21
22 // taskStore is the taskmonitor.Store used by the task CLI commands.
23 // When nil, the CLI defaults to a FileStore backed by .reasonix/tasks
24 // under the project directory.
25 var taskStore taskmonitor.Store
26
27 // taskJobKiller is an optional JobKiller for stopping running tasks.
28 // It is injected by the main wiring or by cli.go when a controller is
29 // available (Desktop or running session). When nil, kill is a no-op.
30 var taskJobKiller taskmonitor.JobKiller
31
32 // SetTaskJobKiller sets the JobKiller for control subcommands.
33 // Called by the wiring when a controller with jobs.Manager is available.
34 func SetTaskJobKiller(k taskmonitor.JobKiller) { taskJobKiller = k }
35
36 // The monitor commands are a content-free machine interface. Scrub optional
37 // free-form summaries at the output boundary as well as at current write sites
38 // so snapshots persisted by older versions cannot disclose paths or commands.
39 func contentFreeTaskSnapshot(s taskmonitor.TaskSnapshot) taskmonitor.TaskSnapshot {
40 s.ErrorSummary = ""
41 return s
42 }
43
44 func contentFreeTaskSnapshots(tasks []taskmonitor.TaskSnapshot) []taskmonitor.TaskSnapshot {
45 if tasks == nil {
46 return nil
47 }
48 contentFree := make([]taskmonitor.TaskSnapshot, len(tasks))
49 for i := range tasks {
50 contentFree[i] = contentFreeTaskSnapshot(tasks[i])
51 }
52 return contentFree
53 }
54
55 func contentFreeTaskEvents(events []taskmonitor.TaskEvent) []taskmonitor.TaskEvent {
56 if events == nil {
57 return nil
58 }
59 contentFree := make([]taskmonitor.TaskEvent, len(events))
60 for i := range events {
61 contentFree[i] = events[i]
62 contentFree[i].ErrorSummary = ""
63 }
64 return contentFree
65 }
66
67 func taskCommand(args []string) int {
68 if len(args) == 0 {
69 fmt.Fprintln(os.Stderr, taskCommandUsage)
70 return 2
71 }
72 store := taskStore
73 if store == nil {
74 store = taskmonitor.NewFileStore(".reasonix/tasks")
75 }
76 switch args[0] {
77 case "list":
78 // Keep the pre-task-monitor machine contract intact. New monitor
79 // commands live below `task monitor` so existing callers do not see a
80 // different schema or task identity model under the same command.
81 return runTaskCommand(args, os.Stdout)
82 case "show":
83 return runTaskCommand(args, os.Stdout)
84 case "monitor":
85 return taskMonitorCommand(store, args[1:])
86 case "machine-list":
87 return runTaskCommand(append([]string{"list"}, args[1:]...), os.Stdout)
88 case "machine-show":
89 return runTaskCommand(append([]string{"show"}, args[1:]...), os.Stdout)
90 case "status":
91 return taskStatusCmd(store, args[1:])
92 case "events":
93 return taskEventsCmd(store, args[1:])
94 case "stop":
95 return taskStopCmd(store, args[1:])
96 case "cancel":
97 return taskCancelCmd(store, args[1:])
98 case "requeue":
99 return taskRequeueCmd(store, args[1:])
100 case "open-session":
101 return taskOpenSessionCmd(store, args[1:])
102 case "tmux":
103 return taskTmuxCmd(store, args[1:])
104 default:
105 fmt.Fprintf(os.Stderr, "unknown task subcommand: %s\n", args[0])
106 fmt.Fprintln(os.Stderr, taskCommandUsage)
107 return 2
108 }
109 }
110
111 func taskMonitorCommand(store taskmonitor.Store, args []string) int {
112 if len(args) == 0 {
113 fmt.Fprintln(os.Stderr, taskMonitorUsage)
114 return 2
115 }
116 switch args[0] {
117 case "list":
118 return taskListCmd(store, args[1:])
119 case "status":
120 return taskStatusCmd(store, args[1:])
121 case "events":
122 return taskEventsCmd(store, args[1:])
123 case "stop":
124 return taskStopCmd(store, args[1:])
125 case "cancel":
126 return taskCancelCmd(store, args[1:])
127 case "requeue":
128 return taskRequeueCmd(store, args[1:])
129 case "open-session":
130 return taskOpenSessionCmd(store, args[1:])
131 default:
132 fmt.Fprintf(os.Stderr, "unknown task monitor subcommand: %s\n", args[0])
133 fmt.Fprintln(os.Stderr, taskMonitorUsage)
134 return 2
135 }
136 }
137
138 func taskTmuxCmd(store taskmonitor.Store, args []string) int {
139 if len(args) == 0 {
140 fmt.Fprintln(os.Stderr, taskTmuxUsage)
141 return 2
142 }
143 a := taskmonitor.NewTmuxAdapter(store, ".reasonix/tasks")
144 switch args[0] {
145 case "attach":
146 return taskTmuxAttachCmd(a, args[1:])
147 case "status":
148 return taskTmuxStatusCmd(a, args[1:])
149 case "open":
150 return taskTmuxOpenCmd(a, args[1:])
151 case "detach":
152 return taskTmuxDetachCmd(a, args[1:])
153 default:
154 fmt.Fprintf(os.Stderr, "unknown task tmux subcommand: %s\n", args[0])
155 fmt.Fprintln(os.Stderr, taskTmuxUsage)
156 return 2
157 }
158 }
159
160 func taskTmuxFlags(name string, args []string) (string, string, bool, *flag.FlagSet, int) {
161 fs := flag.NewFlagSet(name, flag.ContinueOnError)
162 dir := fs.String("dir", "", "project directory scope")
163 session := fs.String("session", "", "tmux session name")
164 jsonOut := fs.Bool("json", false, "output as JSON")
165 if err := fs.Parse(reorderTaskID(fs, args)); err != nil {
166 return "", "", false, fs, 2
167 }
168 return *dir, *session, *jsonOut, fs, 0
169 }
170
171 // reorderTaskID lets users place the positional task ID before or after flags.
172 // The standard flag package stops parsing at the first positional argument.
173 func reorderTaskID(fs *flag.FlagSet, args []string) []string {
174 flags := make([]string, 0, len(args))
175 positionals := make([]string, 0, 1)
176 for i := 0; i < len(args); i++ {
177 arg := args[i]
178 if arg == "--" {
179 flags = append(flags, arg)
180 positionals = append(positionals, args[i+1:]...)
181 break
182 }
183
184 name, inlineValue := taskFlagName(arg)
185 if name == "" {
186 positionals = append(positionals, arg)
187 continue
188 }
189
190 flags = append(flags, arg)
191 registered := fs.Lookup(name)
192 if registered == nil || inlineValue {
193 continue
194 }
195 if boolean, ok := registered.Value.(interface{ IsBoolFlag() bool }); ok && boolean.IsBoolFlag() {
196 continue
197 }
198 if i+1 < len(args) {
199 flags = append(flags, args[i+1])
200 i++
201 }
202 }
203 return append(flags, positionals...)
204 }
205
206 func taskFlagName(arg string) (name string, inlineValue bool) {
207 if arg == "-" || !strings.HasPrefix(arg, "-") {
208 return "", false
209 }
210 name = strings.TrimPrefix(arg, "-")
211 name = strings.TrimPrefix(name, "-")
212 if name == "" {
213 return "", false
214 }
215 if before, _, ok := strings.Cut(name, "="); ok {
216 return before, true
217 }
218 return name, false
219 }
220
221 func printTmuxResult(r taskmonitor.TmuxResult, jsonOut bool) int {
222 if !jsonOut {
223 fmt.Fprintln(os.Stderr, "tmux task commands require --json")
224 return 2
225 }
226 if err := json.NewEncoder(os.Stdout).Encode(r); err != nil {
227 return 1
228 }
229 if r.Error != nil {
230 return 1
231 }
232 return 0
233 }
234
235 func taskTmuxAttachCmd(a *taskmonitor.TmuxAdapter, args []string) int {
236 dir, session, jsonOut, fs, code := taskTmuxFlags("task tmux attach", args)
237 if code != 0 || fs.Arg(0) == "" {
238 fmt.Fprintln(os.Stderr, "usage: reasonix task tmux attach <id> --json [--dir DIR] [--session NAME]")
239 return 2
240 }
241 return printTmuxResult(a.Attach(context.Background(), dir, fs.Arg(0), session), jsonOut)
242 }
243
244 func taskTmuxStatusCmd(a *taskmonitor.TmuxAdapter, args []string) int {
245 dir, _, jsonOut, fs, code := taskTmuxFlags("task tmux status", args)
246 if code != 0 || fs.Arg(0) == "" {
247 fmt.Fprintln(os.Stderr, "usage: reasonix task tmux status <id> --json [--dir DIR]")
248 return 2
249 }
250 return printTmuxResult(a.Status(context.Background(), dir, fs.Arg(0)), jsonOut)
251 }
252
253 func taskTmuxOpenCmd(a *taskmonitor.TmuxAdapter, args []string) int {
254 dir, _, jsonOut, fs, code := taskTmuxFlags("task tmux open", args)
255 if code != 0 || fs.Arg(0) == "" {
256 fmt.Fprintln(os.Stderr, "usage: reasonix task tmux open <id> --json [--dir DIR]")
257 return 2
258 }
259 return printTmuxResult(a.Open(context.Background(), dir, fs.Arg(0)), jsonOut)
260 }
261
262 func taskTmuxDetachCmd(a *taskmonitor.TmuxAdapter, args []string) int {
263 dir, _, jsonOut, fs, code := taskTmuxFlags("task tmux detach", args)
264 if code != 0 || fs.Arg(0) == "" {
265 fmt.Fprintln(os.Stderr, "usage: reasonix task tmux detach <id> --json [--dir DIR]")
266 return 2
267 }
268 return printTmuxResult(a.Detach(context.Background(), dir, fs.Arg(0)), jsonOut)
269 }
270
271 // list
272
273 func taskListCmd(store taskmonitor.Store, args []string) int {
274 fs := flag.NewFlagSet("task list", flag.ContinueOnError)
275 jsonOut := fs.Bool("json", false, "output as JSON")
276 dir := fs.String("dir", "", "project directory scope")
277 if err := fs.Parse(args); err != nil {
278 return 2
279 }
280 if !*jsonOut {
281 fmt.Fprintln(os.Stderr, "task list requires --json")
282 return 2
283 }
284
285 ctx := context.Background()
286 tasks, err := store.ListTasks(ctx, *dir)
287 if err != nil {
288 fmt.Fprintln(os.Stderr, err)
289 return 1
290 }
291 tasks = contentFreeTaskSnapshots(tasks)
292 output := struct {
293 SchemaVersion int `json:"schema_version"`
294 Tasks []taskmonitor.TaskSnapshot `json:"tasks"`
295 }{SchemaVersion: 1, Tasks: tasks}
296 if tasks == nil {
297 output.Tasks = []taskmonitor.TaskSnapshot{}
298 }
299 enc := json.NewEncoder(os.Stdout)
300 enc.SetIndent("", " ")
301 if err := enc.Encode(output); err != nil {
302 fmt.Fprintln(os.Stderr, err)
303 return 1
304 }
305 return 0
306 }
307
308 // status
309
310 func taskStatusCmd(store taskmonitor.Store, args []string) int {
311 fs := flag.NewFlagSet("task status", flag.ContinueOnError)
312 jsonOut := fs.Bool("json", false, "output as JSON")
313 dir := fs.String("dir", "", "project directory scope")
314 if err := fs.Parse(reorderTaskID(fs, args)); err != nil {
315 return 2
316 }
317 if !*jsonOut {
318 fmt.Fprintln(os.Stderr, "task status requires --json")
319 return 2
320 }
321 id := fs.Arg(0)
322 if id == "" {
323 fmt.Fprintln(os.Stderr, "usage: reasonix task status <id> --json [--dir DIR]")
324 return 2
325 }
326
327 ctx := context.Background()
328 snap, err := store.GetTask(ctx, *dir, id)
329 if err != nil {
330 fmt.Fprintln(os.Stderr, err)
331 return 1
332 }
333 output := struct {
334 SchemaVersion int `json:"schema_version"`
335 Task *taskmonitor.TaskSnapshot `json:"task"`
336 }{SchemaVersion: 1}
337 if snap != nil {
338 contentFree := contentFreeTaskSnapshot(*snap)
339 output.Task = &contentFree
340 }
341 enc := json.NewEncoder(os.Stdout)
342 enc.SetIndent("", " ")
343 if err := enc.Encode(output); err != nil {
344 fmt.Fprintln(os.Stderr, err)
345 return 1
346 }
347 return 0
348 }
349
350 // events
351
352 func taskEventsCmd(store taskmonitor.Store, args []string) int {
353 fs := flag.NewFlagSet("task events", flag.ContinueOnError)
354 jsonOut := fs.Bool("json", false, "output as JSON array")
355 jsonl := fs.Bool("jsonl", false, "output as JSONL stream")
356 dir := fs.String("dir", "", "project directory scope")
357 after := fs.Int("after", 0, "only events with Sequence > N")
358 follow := fs.Bool("follow", false, "poll for new events until interrupted")
359 if err := fs.Parse(reorderTaskID(fs, args)); err != nil {
360 return 2
361 }
362 if !*jsonOut && !*jsonl {
363 fmt.Fprintln(os.Stderr, "task events requires --json or --jsonl")
364 return 2
365 }
366 id := fs.Arg(0)
367 if id == "" {
368 fmt.Fprintln(os.Stderr, "usage: reasonix task events <id> --json|--jsonl [--dir DIR] [--after N] [--follow]")
369 return 2
370 }
371
372 ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
373 defer cancel()
374
375 cursor := *after
376
377 for {
378 events, err := store.ListEvents(ctx, *dir, id, cursor)
379 if err != nil {
380 if ctx.Err() != nil {
381 return 0 // cancelled
382 }
383 fmt.Fprintln(os.Stderr, err)
384 return 1
385 }
386
387 events = contentFreeTaskEvents(events)
388
389 // Find max sequence to update cursor
390 for _, e := range events {
391 if e.Sequence > cursor {
392 cursor = e.Sequence
393 }
394 }
395
396 if *jsonl {
397 enc := json.NewEncoder(os.Stdout)
398 for _, e := range events {
399 if err := enc.Encode(e); err != nil {
400 return 1
401 }
402 }
403 } else {
404 // --json: output as JSON array
405 output := struct {
406 SchemaVersion int `json:"schema_version"`
407 TaskID string `json:"task_id"`
408 Events []taskmonitor.TaskEvent `json:"events"`
409 }{SchemaVersion: 1, TaskID: id, Events: events}
410 if events == nil {
411 output.Events = []taskmonitor.TaskEvent{}
412 }
413 enc := json.NewEncoder(os.Stdout)
414 enc.SetIndent("", " ")
415 if err := enc.Encode(output); err != nil {
416 return 1
417 }
418 }
419
420 if !*follow {
421 break
422 }
423 // Check if task has reached a terminal state
424 snap, _ := store.GetTask(ctx, *dir, id)
425 if snap != nil && snap.State.Terminal() && len(events) == 0 {
426 break
427 }
428
429 select {
430 case <-ctx.Done():
431 return 0
432 case <-time.After(500 * time.Millisecond):
433 }
434 }
435 return 0
436 }
437
438 // control commands
439
440 func taskStopCmd(store taskmonitor.Store, args []string) int {
441 fs := flag.NewFlagSet("task stop", flag.ContinueOnError)
442 jsonOut := fs.Bool("json", false, "output as JSON")
443 dir := fs.String("dir", "", "project directory scope")
444 expectedVersion := fs.Uint64("expected-version", 0, "expected task version for CAS")
445 reason := fs.String("reason", "", "reason for stopping")
446 idemKey := fs.String("idempotency-key", "", "idempotency key")
447 if err := fs.Parse(reorderTaskID(fs, args)); err != nil {
448 return 2
449 }
450 if !*jsonOut {
451 fmt.Fprintln(os.Stderr, "task stop requires --json")
452 return 2
453 }
454 id := fs.Arg(0)
455 if id == "" {
456 fmt.Fprintln(os.Stderr, "usage: reasonix task stop <id> --expected-version N --json")
457 return 2
458 }
459
460 ws, ok := store.(taskmonitor.WriteStore)
461 if !ok {
462 fmt.Fprintln(os.Stderr, "task stop: store does not support writes")
463 return 1
464 }
465 cs := taskmonitor.NewControlService(ws)
466 res, err := cs.StopTaskWithKiller(context.Background(), *dir, id, *expectedVersion, *reason, *idemKey, taskJobKiller)
467 return outputControlResult(res, err)
468 }
469
470 func taskCancelCmd(store taskmonitor.Store, args []string) int {
471 fs := flag.NewFlagSet("task cancel", flag.ContinueOnError)
472 jsonOut := fs.Bool("json", false, "output as JSON")
473 dir := fs.String("dir", "", "project directory scope")
474 expectedVersion := fs.Uint64("expected-version", 0, "expected task version for CAS")
475 reason := fs.String("reason", "", "reason for cancelling")
476 idemKey := fs.String("idempotency-key", "", "idempotency key")
477 if err := fs.Parse(reorderTaskID(fs, args)); err != nil {
478 return 2
479 }
480 if !*jsonOut {
481 fmt.Fprintln(os.Stderr, "task cancel requires --json")
482 return 2
483 }
484 id := fs.Arg(0)
485 if id == "" {
486 fmt.Fprintln(os.Stderr, "usage: reasonix task cancel <id> --expected-version N --json")
487 return 2
488 }
489
490 ws, ok := store.(taskmonitor.WriteStore)
491 if !ok {
492 fmt.Fprintln(os.Stderr, "task cancel: store does not support writes")
493 return 1
494 }
495 cs := taskmonitor.NewControlService(ws)
496 res, err := cs.CancelTaskWithKiller(context.Background(), *dir, id, *expectedVersion, *reason, *idemKey, taskJobKiller)
497 return outputControlResult(res, err)
498 }
499
500 func taskRequeueCmd(store taskmonitor.Store, args []string) int {
501 fs := flag.NewFlagSet("task requeue", flag.ContinueOnError)
502 jsonOut := fs.Bool("json", false, "output as JSON")
503 dir := fs.String("dir", "", "project directory scope")
504 expectedVersion := fs.Uint64("expected-version", 0, "expected task version for CAS")
505 idemKey := fs.String("idempotency-key", "", "idempotency key")
506 if err := fs.Parse(reorderTaskID(fs, args)); err != nil {
507 return 2
508 }
509 if !*jsonOut {
510 fmt.Fprintln(os.Stderr, "task requeue requires --json")
511 return 2
512 }
513 id := fs.Arg(0)
514 if id == "" {
515 fmt.Fprintln(os.Stderr, "usage: reasonix task requeue <id> --expected-version N --json")
516 return 2
517 }
518
519 ws, ok := store.(taskmonitor.WriteStore)
520 if !ok {
521 fmt.Fprintln(os.Stderr, "task requeue: store does not support writes")
522 return 1
523 }
524 cs := taskmonitor.NewControlService(ws)
525 res, err := cs.RequeueTask(context.Background(), *dir, id, *expectedVersion, *idemKey)
526 return outputControlResult(res, err)
527 }
528
529 func taskOpenSessionCmd(store taskmonitor.Store, args []string) int {
530 fs := flag.NewFlagSet("task open-session", flag.ContinueOnError)
531 jsonOut := fs.Bool("json", false, "output as JSON")
532 dir := fs.String("dir", "", "project directory scope")
533 if err := fs.Parse(reorderTaskID(fs, args)); err != nil {
534 return 2
535 }
536 if !*jsonOut {
537 fmt.Fprintln(os.Stderr, "task open-session requires --json")
538 return 2
539 }
540 id := fs.Arg(0)
541 if id == "" {
542 fmt.Fprintln(os.Stderr, "usage: reasonix task open-session <id> --json")
543 return 2
544 }
545
546 // open-session is read-only — use Store directly
547 snap, err := store.GetTask(context.Background(), *dir, id)
548 if err != nil {
549 fmt.Fprintln(os.Stderr, err)
550 return 1
551 }
552 res := taskmonitor.ControlResult{
553 SchemaVersion: 1,
554 Command: "open_session",
555 TaskID: id,
556 }
557 if snap == nil {
558 res.Error = &taskmonitor.CtrlError{Code: taskmonitor.ErrTaskNotFound, Message: "task not found"}
559 return outputControlResult(res, nil)
560 }
561 res.SessionID = snap.SessionID
562 res.State = snap.State
563 res.Version = snap.Version
564 res.Accepted = true
565 return outputControlResult(res, nil)
566 }
567
568 func outputControlResult(res taskmonitor.ControlResult, err error) int {
569 if err != nil {
570 fmt.Fprintln(os.Stderr, err)
571 return 1
572 }
573 enc := json.NewEncoder(os.Stdout)
574 enc.SetIndent("", " ")
575 if err := enc.Encode(res); err != nil {
576 fmt.Fprintln(os.Stderr, err)
577 return 1
578 }
579 if !res.Accepted && !res.Idempotent {
580 return 1
581 }
582 return 0
583 }
584
584 lines GO