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