diff --git a/cmd/spinloop/complete.go b/cmd/spinloop/complete.go index e295168d..258323c8 100644 --- a/cmd/spinloop/complete.go +++ b/cmd/spinloop/complete.go @@ -217,6 +217,16 @@ func noPositionals(_ *cobra.Command, _ []string, _ string) ([]string, cobra.Shel return nil, cobra.ShellCompDirectiveNoFileComp } +// itemIDSlot is the work subcommands' own positional — an item's id. There +// is no local registry to draw candidates from the way aliasSlot has +// aliasNames (an id only means anything against a specific run's work +// list, reached over the network, which completion does not call out to): +// it offers none, only suppressing file completion so typing an id is not +// interrupted by filename suggestions that are never valid ids. +func itemIDSlot(_ *cobra.Command, _ []string, _ string) ([]string, cobra.ShellCompDirective) { + return nil, cobra.ShellCompDirectiveNoFileComp +} + // aliasSlot is the Spinloop slot: registered names plus a path, for the first // positional only. func aliasSlot(_ *cobra.Command, args []string, _ string) ([]string, cobra.ShellCompDirective) { diff --git a/cmd/spinloop/complete_test.go b/cmd/spinloop/complete_test.go index 4ff9666b..7f88f7ad 100644 --- a/cmd/spinloop/complete_test.go +++ b/cmd/spinloop/complete_test.go @@ -229,6 +229,25 @@ func TestComplete_UnaliasOffersAliasNames(t *testing.T) { } } +// TestComplete_WorkItemIDSlotsOfferNoFilePaths checks work logs/abort/ +// remove's own positional: no registry to draw a real item id from, so it +// offers no candidates, but must still suppress file completion — an id is +// never a filename, and before this the slot fell through to the default, +// offering the working directory's own files. +func TestComplete_WorkItemIDSlotsOfferNoFilePaths(t *testing.T) { + isolateConfig(t) + + for _, sub := range []string{"logs", "abort", "remove"} { + got, directive := complete(t, "work", sub, "") + if len(got) != 0 { + t.Errorf("work %s: candidates offered where there is no registry to draw from: %v", sub, got) + } + if directive != directiveNoFile { + t.Errorf("work %s: directive = %q, want %q (an id is never a file path)", sub, directive, directiveNoFile) + } + } +} + // TestComplete_SpinloopCommandsOfferAliasesAndPaths checks the commands that take // either. func TestComplete_SpinloopCommandsOfferAliasesAndPaths(t *testing.T) { diff --git a/cmd/spinloop/work.go b/cmd/spinloop/work.go index 58993dba..7eaddff1 100644 --- a/cmd/spinloop/work.go +++ b/cmd/spinloop/work.go @@ -1,16 +1,15 @@ // The `work` command family: the orchestrator's work list — the backlog the // orchestrator works — worked from the shell as a client of the work list API -// the orchestrator serves: add an item, read the work, stop a running item, -// remove an item. The commands name the API's address and present its token; -// the run's view of the items is the source of truth, and a refusal reads the -// way the API states it. +// the orchestrator serves: add an item, read the work, read an item's kept +// output, stop a running item, remove an item. The commands name the API's +// address and present its token; the run's view of the items is the source +// of truth, and a refusal reads the way the API states it. package main import ( "bytes" "context" "encoding/json" - "errors" "fmt" "io" "net/http" @@ -40,7 +39,8 @@ func workCmd() *cobra.Command { Short: "work the orchestrator's work list", Long: `works the orchestrator's work list — the backlog it works — from the shell, as a client of the work list API the orchestrator serves: add an -item, read the work, stop a running item, remove an item. +item, read the work, read an item's kept output, stop a running item, +remove an item. Each subcommand takes --url, the API's base address, and presents the API's token — from --api-token, else --api-token-file, else the SPINLOOP_API_TOKEN @@ -54,6 +54,7 @@ fails before it calls the API, naming the flag.`, workListCmd(), workAbortCmd(), workRemoveCmd(), + workLogsCmd(), ) return c } @@ -122,6 +123,17 @@ func workRequest(base, token, method, path string, body any) ([]byte, error) { return data, nil } +// workAPIErr is a non-2xx answer from the work list API: Error() reads the +// way the API states the refusal, and status is kept so a caller that +// cares — a follow noticing its item was removed — can tell a "not found" +// apart from any other refusal without parsing the message. +type workAPIErr struct { + status int + msg string +} + +func (e *workAPIErr) Error() string { return e.msg } + // workAPIError is a non-2xx answer from the work list API, as the error the // command reports: the API's message where it states one — the refusal reads // the way the API states it — the status and the address where it does not. @@ -131,10 +143,11 @@ func workAPIError(base string, status int, data []byte) error { Message string `json:"message"` } `json:"error"` } + msg := fmt.Sprintf("the work list API at %s answered %d", base, status) if err := json.Unmarshal(data, &out); err == nil && out.Error.Message != "" { - return errors.New(out.Error.Message) + msg = out.Error.Message } - return fmt.Errorf("the work list API at %s answered %d", base, status) + return &workAPIErr{status: status, msg: msg} } // workAddBody is one item as the work list API takes it: the fields the items @@ -366,6 +379,7 @@ the command reports its answer: a refusal reads the way the API states it.`, } fs := c.Flags() workAPIFlags(fs, &base, &apiToken, &apiTokenFile) + c.ValidArgsFunction = itemIDSlot return c } @@ -401,5 +415,6 @@ reads the way the API states it.`, } fs := c.Flags() workAPIFlags(fs, &base, &apiToken, &apiTokenFile) + c.ValidArgsFunction = itemIDSlot return c } diff --git a/cmd/spinloop/work_logs.go b/cmd/spinloop/work_logs.go new file mode 100644 index 00000000..b40e7b3b --- /dev/null +++ b/cmd/spinloop/work_logs.go @@ -0,0 +1,182 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + "time" + + "github.com/spf13/cobra" + + "github.com/spinloop-ai/spinloop/internal/orchestrator" +) + +// workLogsInterval is how often a follow asks for more output, and checks +// the item's own state. A variable so tests need not wait on it. +var workLogsInterval = 1 * time.Second + +// workLogsCmd builds `work logs`. +func workLogsCmd() *cobra.Command { + var base, apiToken, apiTokenFile string + var follow bool + c := &cobra.Command{ + Use: "logs ", + Short: "print an item's kept agent output", + Long: `prints an item's kept agent output, through the work list API the +orchestrator serves. An item with no output yet is printed as empty, not a +fault, and an id the run does not carry is refused, naming it, the way the +API states it. + +-f/--follow polls for new output and prints it as it arrives, the way +tail -f does: it keeps polling through backlog and running — an item +named before, or just as, it starts is still followed — and stops once +the item is done or failed, or the operator interrupts.`, + Args: cobra.ExactArgs(1), + SilenceErrors: true, + SilenceUsage: true, + RunE: func(_ *cobra.Command, args []string) error { + id := args[0] + b, token, err := workTarget("work logs", base, apiToken, apiTokenFile) + if err != nil { + return err + } + if follow { + return followWorkLogs(b, token, id) + } + return runWorkLogsOnce(b, token, id, os.Stdout) + }, + } + fs := c.Flags() + workAPIFlags(fs, &base, &apiToken, &apiTokenFile) + fs.BoolVarP(&follow, "follow", "f", false, "keep printing new output as it arrives") + c.ValidArgsFunction = itemIDSlot + return c +} + +// runWorkLogsOnce prints an item's kept output once. +func runWorkLogsOnce(base, token, id string, w io.Writer) error { + log, err := workLogFetch(base, token, id) + if err != nil { + return err + } + fmt.Fprint(w, log) + return nil +} + +// workLogReply is the work list API's own answer shape for an item's log: +// carrying no field at all where there is none yet, the way handleLog +// answers `"log": null`. +type workLogReply struct { + Log *string `json:"log"` +} + +// workLogFetch reads an item's kept output through the work list API. An +// item with none yet answers "", not an error. +func workLogFetch(base, token, id string) (string, error) { + data, err := workRequest(base, token, http.MethodGet, "/v1/items/"+url.PathEscape(id)+"/log", nil) + if err != nil { + return "", err + } + var out workLogReply + if err := json.Unmarshal(data, &out); err != nil { + return "", fmt.Errorf("reading item %q's log from the work list API at %s: %w", id, base, err) + } + if out.Log == nil { + return "", nil + } + return *out.Log, nil +} + +// workItemState reads one item's state from the work list API's own list: +// found and its state where the list still carries the id, found false +// where it has dropped out of the list entirely — a removal. +func workItemState(base, token, id string) (state string, found bool, err error) { + data, err := workRequest(base, token, http.MethodGet, "/v1/items", nil) + if err != nil { + return "", false, err + } + var out struct { + Data []orchestrator.ItemView `json:"data"` + } + if err := json.Unmarshal(data, &out); err != nil { + return "", false, fmt.Errorf("reading the work list from the work list API at %s: %w", base, err) + } + for _, v := range out.Data { + if v.ID == id { + return v.State, true, nil + } + } + return "", false, nil +} + +// followWorkLogs polls an item's log and its own state until it ends, or +// the operator interrupts. +func followWorkLogs(base, token, id string) error { + return followUntilInterrupted(func(ctx context.Context) error { + return followWorkLogsLoop(ctx, base, token, id, os.Stdout) + }) +} + +// followWorkLogsLoop is the polling itself, with the interrupt wiring left +// to its caller so it can be driven directly. Each tick fetches the log, +// prints what is new, then checks the item's own state: the loop keeps +// going while it is backlog or running, and ends — after that tick's log +// poll, so nothing written right at the end is missed — once the state is +// done or failed, or the item has dropped out of the list entirely (a +// removal mid-follow), which ends the follow the same clean way. +func followWorkLogsLoop(ctx context.Context, base, token, id string, w io.Writer) error { + var last string + for { + log, err := workLogFetch(base, token, id) + if err != nil { + if workItemGone(err) { + return nil + } + return err + } + last = printWorkLogSuffix(w, log, last) + + state, found, err := workItemState(base, token, id) + if err != nil { + return err + } + if !found || (state != orchestrator.StateBacklog && state != orchestrator.StateRunning) { + return nil + } + + select { + case <-ctx.Done(): + return nil + case <-time.After(workLogsInterval): + } + } +} + +// printWorkLogSuffix prints whatever of log is new since last, and returns +// log as the new "last" for the next call: where log still starts with +// last, only the suffix beyond it is new; otherwise — the kept output +// changed underneath, which does not happen in an item's own lifetime, but +// is not worth crashing on — log is printed in full, as if nothing had +// been printed yet. +func printWorkLogSuffix(w io.Writer, log, last string) string { + if strings.HasPrefix(log, last) { + fmt.Fprint(w, log[len(last):]) + } else { + fmt.Fprint(w, log) + } + return log +} + +// workItemGone reports whether err is the work list API's own "id not +// carried" refusal (a 404) — what a removal mid-follow leaves behind. A +// follow reads that as a clean end, not a failure. +func workItemGone(err error) bool { + var apiErr *workAPIErr + return errors.As(err, &apiErr) && apiErr.status == http.StatusNotFound +} diff --git a/cmd/spinloop/work_logs_test.go b/cmd/spinloop/work_logs_test.go new file mode 100644 index 00000000..f0221ad3 --- /dev/null +++ b/cmd/spinloop/work_logs_test.go @@ -0,0 +1,397 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +// --- 1: a single read ------------------------------------------------------- + +func TestWorkLogs_CallsTheLogPathAndPrintsTheOutput(t *testing.T) { + base, got := workAPIStub(t, http.StatusOK, map[string]any{"id": "a", "log": "hello\n"}) + out, err := runWork(t, "logs", "a", "--url", base, "--api-token", "tok") + if err != nil { + t.Fatalf("work logs: %v (out %s)", err, out) + } + if out != "hello\n" { + t.Errorf("out = %q, want %q", out, "hello\n") + } + if got.method != http.MethodGet || got.path != "/v1/items/a/log" { + t.Errorf("logs calls the API's log path for the id: %s %s", got.method, got.path) + } + if got.bearer != "Bearer tok" { + t.Errorf("bearer = %q, want the token", got.bearer) + } +} + +func TestWorkLogs_NoOutputYetPrintsNothingNotAFault(t *testing.T) { + base, _ := workAPIStub(t, http.StatusOK, map[string]any{"id": "a", "log": nil}) + out, err := runWork(t, "logs", "a", "--url", base) + if err != nil { + t.Fatalf("an item with no output yet should not fail: %v", err) + } + if out != "" { + t.Errorf("out = %q, want empty", out) + } +} + +func TestWorkLogsCmd_NoURLFailsBeforeTheCall(t *testing.T) { + _, got := workAPIStub(t, http.StatusOK, map[string]any{"id": "a", "log": "hi\n"}) + _, err := runWork(t, "logs", "a") + if err == nil || !strings.Contains(err.Error(), "--url") { + t.Errorf("no --url, the command fails naming the flag: %v", err) + } + if got.method != "" { + t.Errorf("no address, no call: the stub saw %s %s", got.method, got.path) + } +} + +func TestWorkLogs_TheRefusalReadsTheWayTheAPIStatesIt(t *testing.T) { + base, _ := workAPIStub(t, http.StatusNotFound, + workAPIErrorReply(`the items file carries no item with id "ghost"`)) + _, err := runWork(t, "logs", "ghost", "--url", base) + if err == nil || !strings.Contains(err.Error(), `no item with id "ghost"`) { + t.Errorf("an id the run does not carry is refused, naming it: %v", err) + } +} + +func TestWorkItemState_ReadsTheMatchingItemsState(t *testing.T) { + base, _ := workAPIStub(t, http.StatusOK, map[string]any{"data": []map[string]any{ + {"id": "a", "state": "running"}, + {"id": "b", "state": "done"}, + }}) + state, found, err := workItemState(base, "", "b") + if err != nil { + t.Fatalf("workItemState: %v", err) + } + if !found || state != "done" { + t.Errorf("found = %v, state = %q, want found=true state=done", found, state) + } +} + +func TestWorkItemState_NotFoundWhenTheListDoesNotCarryIt(t *testing.T) { + base, _ := workAPIStub(t, http.StatusOK, map[string]any{"data": []map[string]any{{"id": "a", "state": "running"}}}) + _, found, err := workItemState(base, "", "ghost") + if err != nil { + t.Fatalf("workItemState: %v", err) + } + if found { + t.Error("an id the list does not carry should answer found=false, not an error — a removal leaves the list this way, not a refusal") + } +} + +func TestWorkItemState_AnAPIFailurePropagates(t *testing.T) { + _, _, err := workItemState("http://127.0.0.1:1", "", "a") + if err == nil { + t.Error("the work list API being unreachable should fail, not silently answer not-found") + } +} + +func TestWorkItemState_AMalformedAnswerFails(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Write([]byte("not json")) + })) + t.Cleanup(srv.Close) + _, _, err := workItemState(srv.URL, "", "a") + if err == nil { + t.Error("a malformed answer should fail, not silently answer not-found") + } +} + +func TestWorkLogFetch_AMalformedAnswerFails(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Write([]byte("not json")) + })) + t.Cleanup(srv.Close) + _, err := workLogFetch(srv.URL, "", "a") + if err == nil { + t.Error("a malformed answer should fail") + } +} + +func TestPrintWorkLogSuffix(t *testing.T) { + for _, tc := range []struct { + name, log, last, wantPrinted, wantReturn string + }{ + {"grows from empty", "first\n", "", "first\n", "first\n"}, + {"prints only the new suffix", "first\nsecond\n", "first\n", "second\n", "first\nsecond\n"}, + {"nothing new prints nothing", "first\n", "first\n", "", "first\n"}, + {"content changed underneath prints in full", "replaced\n", "first\n", "replaced\n", "replaced\n"}, + } { + t.Run(tc.name, func(t *testing.T) { + var buf bytes.Buffer + got := printWorkLogSuffix(&buf, tc.log, tc.last) + if buf.String() != tc.wantPrinted { + t.Errorf("printed = %q, want %q", buf.String(), tc.wantPrinted) + } + if got != tc.wantReturn { + t.Errorf("returned = %q, want %q", got, tc.wantReturn) + } + }) + } +} + +// --- 2: following ------------------------------------------------------------ + +// TestWorkLogs_FollowFlagRunsTheFollowPath drives -f through the real +// command, not followWorkLogsLoop directly, so workLogsCmd's own follow +// branch and followWorkLogs's followUntilInterrupted wiring both get +// exercised — the item is already done on the first poll, so the follow +// ends on its own without needing an interrupt or a shortened interval. +func TestWorkLogs_FollowFlagRunsTheFollowPath(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("GET /v1/items/a/log", func(w http.ResponseWriter, _ *http.Request) { + json.NewEncoder(w).Encode(map[string]any{"id": "a", "log": "done already\n"}) + }) + mux.HandleFunc("GET /v1/items", func(w http.ResponseWriter, _ *http.Request) { + json.NewEncoder(w).Encode(map[string]any{"data": []map[string]any{{"id": "a", "state": "done"}}}) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + + out, err := runWork(t, "logs", "a", "-f", "--url", srv.URL) + if err != nil { + t.Fatalf("work logs -f: %v", err) + } + if out != "done already\n" { + t.Errorf("out = %q, want %q", out, "done already\n") + } +} + +func TestWorkLogs_WithoutFollowMakesOneRequest(t *testing.T) { + var mu sync.Mutex + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + calls++ + mu.Unlock() + json.NewEncoder(w).Encode(map[string]any{"id": "a", "log": "hi\n"}) + })) + t.Cleanup(srv.Close) + + out, err := runWork(t, "logs", "a", "--url", srv.URL) + if err != nil { + t.Fatalf("work logs: %v", err) + } + if out != "hi\n" { + t.Errorf("out = %q, want %q", out, "hi\n") + } + mu.Lock() + defer mu.Unlock() + if calls != 1 { + t.Errorf("without --follow, one request should be made, got %d", calls) + } +} + +// workLogsFollowStub answers /v1/items//log and /v1/items across a +// sequence of ticks it advances on every /v1/items poll (the loop's own +// last call each tick, after the log poll) — a state's own scenario +// controls when the item ends. +type workLogsFollowStub struct { + mu sync.Mutex + tick int + logs []string // one per tick; the last entry repeats past the end + state []string // one per tick; "" (past the end of the slice) means removed +} + +func (s *workLogsFollowStub) server(t *testing.T) string { + t.Helper() + mux := http.NewServeMux() + mux.HandleFunc("GET /v1/items/a/log", func(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + i := s.tick + if i >= len(s.logs) { + i = len(s.logs) - 1 + } + log := s.logs[i] + s.mu.Unlock() + json.NewEncoder(w).Encode(map[string]any{"id": "a", "log": log}) + }) + mux.HandleFunc("GET /v1/items", func(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + i := s.tick + var state string + found := i < len(s.state) && s.state[i] != "" + if found { + state = s.state[i] + } + s.tick++ + s.mu.Unlock() + var data []map[string]any + if found { + data = []map[string]any{{"id": "a", "state": state}} + } + json.NewEncoder(w).Encode(map[string]any{"data": data}) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + return srv.URL +} + +func TestFollowWorkLogsLoop_PrintsOnlyNewOutputEachTick(t *testing.T) { + prev := workLogsInterval + workLogsInterval = time.Millisecond + t.Cleanup(func() { workLogsInterval = prev }) + + stub := &workLogsFollowStub{ + logs: []string{"first\n", "first\nsecond\n", "first\nsecond\n"}, + state: []string{"running", "running", "done"}, + } + base := stub.server(t) + + var buf bytes.Buffer + if err := followWorkLogsLoop(context.Background(), base, "", "a", &buf); err != nil { + t.Fatalf("followWorkLogsLoop: %v", err) + } + if got := buf.String(); got != "first\nsecond\n" { + t.Errorf("out = %q, want %q (no duplicated suffix)", got, "first\nsecond\n") + } +} + +func TestFollowWorkLogsLoop_KeepsPollingThroughTheBacklog(t *testing.T) { + prev := workLogsInterval + workLogsInterval = time.Millisecond + t.Cleanup(func() { workLogsInterval = prev }) + + stub := &workLogsFollowStub{ + logs: []string{"", "", "started\n"}, + state: []string{"backlog", "backlog", "running", "done"}, + } + base := stub.server(t) + + var buf bytes.Buffer + if err := followWorkLogsLoop(context.Background(), base, "", "a", &buf); err != nil { + t.Fatalf("followWorkLogsLoop: %v", err) + } + if got := buf.String(); got != "started\n" { + t.Errorf("out = %q, want %q — a backlog item should be waited on, not treated as ended", got, "started\n") + } +} + +func TestFollowWorkLogsLoop_EndsOnceTheItemIsDoneAfterAFinalPoll(t *testing.T) { + prev := workLogsInterval + workLogsInterval = time.Millisecond + t.Cleanup(func() { workLogsInterval = prev }) + + stub := &workLogsFollowStub{ + logs: []string{"running\n", "running\nthe end\n"}, + state: []string{"running", "done"}, + } + base := stub.server(t) + + var buf bytes.Buffer + if err := followWorkLogsLoop(context.Background(), base, "", "a", &buf); err != nil { + t.Fatalf("followWorkLogsLoop: %v", err) + } + if got := buf.String(); !strings.Contains(got, "the end\n") { + t.Errorf("out = %q, want the final tick's own new output printed before ending", got) + } +} + +func TestFollowWorkLogsLoop_EndsCleanlyWhenTheItemIsRemoved(t *testing.T) { + prev := workLogsInterval + workLogsInterval = time.Millisecond + t.Cleanup(func() { workLogsInterval = prev }) + + mux := http.NewServeMux() + var mu sync.Mutex + removed := false + mux.HandleFunc("GET /v1/items/a/log", func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + gone := removed + mu.Unlock() + if gone { + w.WriteHeader(http.StatusNotFound) + json.NewEncoder(w).Encode(map[string]any{"error": map[string]any{"message": `the items file carries no item with id "a"`}}) + return + } + json.NewEncoder(w).Encode(map[string]any{"id": "a", "log": "running\n"}) + }) + mux.HandleFunc("GET /v1/items", func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + removed = true + mu.Unlock() + json.NewEncoder(w).Encode(map[string]any{"data": []map[string]any{{"id": "a", "state": "running"}}}) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + + var buf bytes.Buffer + if err := followWorkLogsLoop(context.Background(), srv.URL, "", "a", &buf); err != nil { + t.Fatalf("an item removed mid-follow should end cleanly, not fail: %v", err) + } +} + +func TestFollowWorkLogsLoop_ALogFetchFailurePropagates(t *testing.T) { + // A real failure (not a 404 "not found") reading the log itself must end + // the follow as an error too, distinct from workItemGone's clean end. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + t.Cleanup(srv.Close) + + var buf bytes.Buffer + err := followWorkLogsLoop(context.Background(), srv.URL, "", "a", &buf) + if err == nil { + t.Error("a real failure reading the log should end the follow as an error") + } +} + +func TestFollowWorkLogsLoop_AStateCheckFailurePropagates(t *testing.T) { + prev := workLogsInterval + workLogsInterval = time.Millisecond + t.Cleanup(func() { workLogsInterval = prev }) + + // The log fetch succeeds, but the list — the loop's second call each + // tick — answers a real failure, not a clean "not found": that must + // end the follow as an error, not as if the item had simply ended. + mux := http.NewServeMux() + mux.HandleFunc("GET /v1/items/a/log", func(w http.ResponseWriter, _ *http.Request) { + json.NewEncoder(w).Encode(map[string]any{"id": "a", "log": "hi\n"}) + }) + mux.HandleFunc("GET /v1/items", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + + var buf bytes.Buffer + err := followWorkLogsLoop(context.Background(), srv.URL, "", "a", &buf) + if err == nil { + t.Error("a real failure reading the item's state should end the follow as an error") + } +} + +func TestFollowWorkLogsLoop_StopsWhenCancelled(t *testing.T) { + prev := workLogsInterval + workLogsInterval = time.Millisecond + t.Cleanup(func() { workLogsInterval = prev }) + + mux := http.NewServeMux() + mux.HandleFunc("GET /v1/items/a/log", func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(map[string]any{"id": "a", "log": "still going\n"}) + }) + mux.HandleFunc("GET /v1/items", func(w http.ResponseWriter, r *http.Request) { + json.NewEncoder(w).Encode(map[string]any{"data": []map[string]any{{"id": "a", "state": "running"}}}) + }) + srv := httptest.NewServer(mux) + t.Cleanup(srv.Close) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(20 * time.Millisecond) + cancel() + }() + + var buf bytes.Buffer + if err := followWorkLogsLoop(ctx, srv.URL, "", "a", &buf); err != nil { + t.Fatalf("a cancelled follow is a clean exit, got: %v", err) + } +} diff --git a/docs/README.md b/docs/README.md index 98daf3f7..d49444cc 100644 --- a/docs/README.md +++ b/docs/README.md @@ -64,7 +64,7 @@ Four words carry the whole tool: | [`spinloop fleet`](commands/fleet.md) | Drive the engines on every machine you run: start, stop, deploy, route | | [`spinloop gateway`](commands/gateway.md) | Serve the fleet under one OpenAI-compatible endpoint | | [`spinloop orchestrator`](commands/orchestrator.md) | Work a backlog of items against the fleet, at the fleet's declared pace | -| [`spinloop work`](commands/work.md) | Drive the orchestrator's work list from the shell: add, list, abort, remove | +| [`spinloop work`](commands/work.md) | Drive the orchestrator's work list from the shell: add, list, logs, abort, remove | | [`spinloop remote`](commands/remote.md) | Run the model on a cloud GPU that stops when you do | | [`spinloop hf`](commands/hf.md) | Write a `Spinloop` for a Hugging Face model, from its page reference | | [`spinloop harness`](commands/harness.md) | Configure the agent (add, remove, apply, unapply, show, export), launch it (open), and set the default (config) | diff --git a/docs/commands/work.md b/docs/commands/work.md index 38d88f19..766cbb65 100644 --- a/docs/commands/work.md +++ b/docs/commands/work.md @@ -2,12 +2,13 @@ Work the [orchestrator](orchestrator.md)'s work list — the backlog it works — from the shell, as a client of the [work list API](orchestrator.md#the-work-list-api) -the orchestrator serves: add an item, read the work, stop a running item, -remove an item. +the orchestrator serves: add an item, read the work, read an item's kept +output, stop a running item, remove an item. ```sh spinloop work add --url http://127.0.0.1:4010 --id fix-parser --instructions "fix the failing tests" --dir ./parser spinloop work list --url http://127.0.0.1:4010 +spinloop work logs --url http://127.0.0.1:4010 fix-parser -f spinloop work abort --url http://127.0.0.1:4010 fix-parser spinloop work remove --url http://127.0.0.1:4010 docs-refresh ``` @@ -76,6 +77,28 @@ record is `backlog`. The list reads the work list API the orchestrator serves — the run's view of the items, the source of truth. `spinloop work ls` is the same list. +## Reading an item's log + +```sh +spinloop work logs --url http://127.0.0.1:4010 fix-parser +``` + +Prints an item's kept agent output through the API's log path. An item with +no output yet — still `backlog`, or `running` with nothing written so far — +prints as empty, not a fault. An id the run does not carry is refused, +naming it. + +```sh +spinloop work logs --url http://127.0.0.1:4010 fix-parser -f +``` + +`-f`/`--follow` polls for new output and prints it as it arrives, the way +`tail -f` does. It keeps polling through `backlog` and `running` — run it +right after `work add` and it waits for the item to start, then streams +its output — and stops once the item is `done` or `failed`, once it drops +out of the work list (a `work remove` elsewhere), or on your own +interrupt. + ## Aborting an item ```sh @@ -122,7 +145,7 @@ once the item is out, and the command reports its answer. | Flag | Meaning | | ---- | ------- | -| `--url
` | The work list API's base address — `add`, `list`, `abort`, `remove` | +| `--url
` | The work list API's base address — `add`, `list`, `logs`, `abort`, `remove` | | `--api-token ` | The work list API's bearer token — every subcommand | | `--api-token-file ` | The file the work list API's bearer token stands in — every subcommand | | `--id ` | The item's id — `add` | @@ -130,6 +153,7 @@ once the item is out, and the command reports its answer. | `--dir ` | The directory the agent works in — `add` | | `--tag ` | A tag the item carries; repeatable — `add` | | `--priority ` | The item's priority, higher first — `add` | +| `-f`, `--follow` | Keep printing new output as it arrives — `logs` | Where neither token flag is given, the token comes from the `SPINLOOP_API_TOKEN` environment, and a loopback API needs none at all. diff --git a/docs/work-items.md b/docs/work-items.md index 17ac902c..d2e6a4e6 100644 --- a/docs/work-items.md +++ b/docs/work-items.md @@ -119,7 +119,8 @@ it again. From the shell, the [`spinloop work`](commands/work.md) family drives this backlog through the run's [work list API](commands/orchestrator.md#the-work-list-api): `work list` reports every item with its state — one plain line per item, a -dash where a value is absent; `work abort ` stops a running item and puts +dash where a value is absent; `work logs ` (`-f` to follow) prints an +item's kept agent output; `work abort ` stops a running item and puts it back in the backlog; `work remove ` takes an item out of the file, its state, and its log. The commands name the API's address with `--url` and present its token, and a refusal reads the way the API states it. @@ -145,6 +146,6 @@ giving only the gateway's address. - [`spinloop orchestrator`](commands/orchestrator.md) — the full command reference, and the [work list API](commands/orchestrator.md#the-work-list-api) a client works the backlog through - [`spinloop work`](commands/work.md) — the backlog driven from the shell, - through the run's work list API: add, list, abort, remove + through the run's work list API: add, list, logs, abort, remove - [The fleet file](commands/fleet.md) — tags, concurrency, and waking - [The gateway](commands/gateway.md) — the front door the orchestrator reads and routes through diff --git a/openspec/changes/archive/2026-09-19-work-logs-command/.openspec.yaml b/openspec/changes/archive/2026-09-19-work-logs-command/.openspec.yaml new file mode 100644 index 00000000..eaa6b1cd --- /dev/null +++ b/openspec/changes/archive/2026-09-19-work-logs-command/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-09-19 diff --git a/openspec/changes/archive/2026-09-19-work-logs-command/design.md b/openspec/changes/archive/2026-09-19-work-logs-command/design.md new file mode 100644 index 00000000..2500bc22 --- /dev/null +++ b/openspec/changes/archive/2026-09-19-work-logs-command/design.md @@ -0,0 +1,100 @@ +## Context + +`cmd/spinloop/work.go` already has `workTarget` (resolves `--url` and the +token) and `workRequest` (one call to the work list API, a non-2xx turned +into an error carrying the API's own message) — every other `work` +subcommand is built on these two. The API's `GET /v1/items/{id}/log` +(`internal/orchestrator/api.go`'s `handleLog`) already answers `{"id": +"", "log": ""}` or `{"id": "", "log": null}`, and refuses +an unknown id the way `remove`/`abort` already do (`WorkList.Log`, an +`errMissing` the API turns into a 404 `workAPIError` already reads). +`GET /v1/items` already answers each item's `state` (`ItemView`, the same +type `work list` already decodes). + +`cmd/spinloop/follow.go`'s `followUntilInterrupted` is the shared +poll-under-a-cancellable-context loop `fleet logs -f` and `remote logs -f` +already use: it hands the loop a `context.Context` an interrupt cancels, +and treats that cancellation as a clean end, not a failure. + +## Goals / Non-Goals + +**Goals:** +- `work logs ` prints an item's kept output once, empty where there is + none yet, refusing an id the API does not carry the way `abort`/`remove` + already do +- `work logs -f ` polls for new output and prints it as it arrives, + stopping once the item ends (`done`/`failed`) or the operator interrupts +- Reuse `workTarget`/`workRequest`/`followUntilInterrupted` as they stand; + no change to the API, no new endpoint + +**Non-Goals:** +- An offset/range parameter on the log endpoint (`fleet logs -f`'s + `NextOffset` pattern) — the log endpoint already answers the whole + content each call, small enough (an item's own kept output) that + re-fetching it whole each poll is not worth a protocol change for +- A `--limit` flag trimming to the last N lines (`fleet logs`'s own) — an + item's kept output is one agent's run, not a fleet-wide tail; the + problem `--limit` solves does not apply +- Interleaving several items' output — `work logs` takes exactly one id, + like `abort`/`remove` already do + +## Decisions + +### The follow loop diffs by string prefix, not by offset + +Each poll re-fetches the whole log via the same `GET +/v1/items/{id}/log` call a plain `work logs` makes. The loop keeps the +length of what it has already printed; where the new answer's content +still starts with what was printed last time, only the suffix is new and +gets printed. Content that does not start that way (the kept output +disappeared, or changed underneath — neither expected in the item's own +lifetime, but not a case to crash on) is printed in full, as if nothing +had been printed yet: simpler than reconciling a diff against something +the API gives no way to address a byte range of. + +### Following waits through the backlog, stops at an ended state + +`-f` polls `GET /v1/items` alongside the log, on the same interval, to +read the one item's own `state` (`orchestrator.ItemView`, decoded the way +`work list` already does; the item missing from the list is read the same +as the log call's own "id not carried" refusal). The loop treats +`backlog` and `running` alike — keep polling — and stops once the state +is `done` or `failed`, after one final log poll so nothing the item wrote +right at the end is missed. This is what makes `work logs -f` on an item +that has not started yet a plain "wait for it, then follow" instead of an +immediate end: the proposal's own "the way `tail -f` does" reads more +naturally as "watch this item's output happen" than "only ever look if it +is already running". + +### The poll interval is a package var, like `fleetLogsInterval` + +`workLogsInterval = 1 * time.Second` — shorter than `fleetLogsInterval`'s +3s, since an item's own agent output is usually what the operator is +actively watching right after starting or aborting it, not a fleet-wide +tail checked in passing. A package var, not a flag, matching +`fleetLogsInterval`'s own precedent: a test overrides it directly rather +than the command needing a hidden flag for it. + +### Two API calls per poll, not one + +The log endpoint carries no state, and the list endpoint carries no +single item's full log efficiently addressed — combining them into one +new endpoint would be a real API change for a command that otherwise +needs none. Two small calls a second is well inside what a local +orchestrator's work list API is built for (`work list` alone already +polls the same list endpoint from an operator's shell without concern). + +## Risks / Trade-offs + +- **A poll can catch an item between "done" and its log's last write + settling.** Not possible here: the orchestrator writes the item's kept + output before it records the state that ends it (the log is captured by + the agent's own process exit, the state written after `Wait` returns), + so by the time a poll sees `done`/`failed` the log call already carries + everything. The design's own "one final log poll" after the state ends + is what makes this true from the client's side too, not just the + server's. +- **Two calls a second, for as long as a follow runs.** Bounded by the + operator's own patience — a follow that outlives the item ends itself; + one left running against a long backlog item is no worse than `work + list` run in a manual polling loop already is. diff --git a/openspec/changes/archive/2026-09-19-work-logs-command/proposal.md b/openspec/changes/archive/2026-09-19-work-logs-command/proposal.md new file mode 100644 index 00000000..5cb9b870 --- /dev/null +++ b/openspec/changes/archive/2026-09-19-work-logs-command/proposal.md @@ -0,0 +1,40 @@ +## Why + +An item's kept agent output can only be read today by reaching the +orchestrator's own filesystem directly — `work list`/`add`/`abort`/`remove` +are already clients of the work list API (#228), but there is no client for +`GET /v1/items/{id}/log`. That defeats the point for a remote client — a +GitHub Action, a CI job — that only has network access to the API. (#231) + +## What Changes + +- `spinloop work logs ` reads an item's kept output through the work + list API's existing `GET /v1/items/{id}/log`, the same `--url`/token + shape as the rest of the `work` family. +- `-f`/`--follow` polls the API for new output and prints it as it arrives, + the way `tail -f` does, reusing the `followUntilInterrupted` helper + `fleet logs -f`/`remote logs -f` already share. It keeps polling through + `backlog` and `running` — an item named before or just as it starts still + gets followed — and stops once the item's state is `done` or `failed`, or + on the operator's interrupt. +- An item with no output yet is printed as empty, not a fault, matching the + API's own `log: null` answer. +- An id the run does not carry is refused, naming it, the way the other + subcommands already are. + +## Capabilities + +### Modified Capabilities + +- `work-commands`: the top-level `work` group's subcommand list gains + `logs`; a new requirement covers reading and following an item's log + through the work list API. + +## Impact + +- `cmd/spinloop/work.go`: `workLogsCmd`, reusing `workTarget`/`workRequest` + the other subcommands already use +- `cmd/spinloop/follow.go`: `followUntilInterrupted` gains a second, third + and fourth caller +- `openspec/specs/work-commands/spec.md`: the subcommand list and a new + requirement diff --git a/openspec/changes/archive/2026-09-19-work-logs-command/specs/work-commands/spec.md b/openspec/changes/archive/2026-09-19-work-logs-command/specs/work-commands/spec.md new file mode 100644 index 00000000..3aa9618d --- /dev/null +++ b/openspec/changes/archive/2026-09-19-work-logs-command/specs/work-commands/spec.md @@ -0,0 +1,97 @@ +## MODIFIED Requirements + +### Requirement: The work commands as work list clients + +`spinloop work` SHALL be a top-level command group with the subcommands add, +list, abort, remove and logs, each a client of the orchestrator's work list +API. Every subcommand SHALL take a `--url` flag naming the API's base +address, and SHALL present the API's token as a bearer on every request it +makes — resolved from `--api-token`, else `--api-token-file`, else the +`SPINLOOP_API_TOKEN` environment variable, two of the flags given at once +being a refusal naming both. A subcommand that names no `--url` SHALL fail +before it calls the API, naming the flag. The commands SHALL be clients of +the API alone: they SHALL NOT read or write the items file, the state, or +the logs directly, and SHALL NOT take any lock beside them. + +#### Scenario: The API's address is named + +- **WHEN** the operator runs a work command naming the API's address with + `--url` +- **THEN** it calls that API, presenting the token as a bearer + +#### Scenario: No address is named + +- **WHEN** the operator runs a work command with no `--url` +- **THEN** it fails, naming the `--url` flag, and calls no API + +#### Scenario: Two token flags at once + +- **WHEN** the operator gives both `--api-token` and `--api-token-file` +- **THEN** the command fails, naming both flags, and calls no API + +#### Scenario: The token comes from the environment + +- **WHEN** the operator names the API's address, sets no token flag, and the + `SPINLOOP_API_TOKEN` environment is set +- **THEN** the command presents that value as the bearer + +## ADDED Requirements + +### Requirement: Reading an item's log through the work list API + +`spinloop work logs ` SHALL call the API's `GET /v1/items/{id}/log` +path for the item's kept agent output and print it. An item the API +answers as carrying no output yet SHALL be printed as empty, not treated +as a fault. An id the run does not carry SHALL be refused, naming it, the +way the API states it. + +Given `-f`/`--follow`, the command SHALL instead poll the API for new +output and print it as it arrives, the way `tail -f` does: it SHALL keep +polling while the item's state is `backlog` or `running` — an item named +before, or just as, it starts is still followed — and SHALL stop once the +API reports the item `done` or `failed`, printing whatever output arrived +up to that point first. The operator's interrupt SHALL also end a follow, +cleanly. The command SHALL NOT itself read the items file, the state, or +the logs directly: the API's answers are the whole source of what it +prints and when it stops. + +#### Scenario: An item's kept output is printed + +- **WHEN** the operator runs `work logs` naming an item with kept output +- **THEN** the command prints it + +#### Scenario: An item with no output yet is empty, not a fault + +- **WHEN** the operator runs `work logs` naming an item the API answers + with no output yet +- **THEN** the command prints nothing, and does not fail + +#### Scenario: An id the run does not carry is refused + +- **WHEN** the operator runs `work logs` naming an id the API does not + carry +- **THEN** the command fails, naming the id, the way the API states it + +#### Scenario: A follow streams new output as it arrives + +- **WHEN** the operator runs `work logs -f` on a running item, and its + agent produces more output +- **THEN** the command prints the new output as later polls see it + +#### Scenario: A follow waits through the backlog + +- **WHEN** the operator runs `work logs -f` on an item the run has not yet + started +- **THEN** the command keeps polling rather than ending, and starts + printing output once the item runs + +#### Scenario: A follow ends once the item ends + +- **WHEN** a followed item's state becomes `done` or `failed` +- **THEN** the command prints whatever output arrived up to that point and + ends + +#### Scenario: A follow ends on the operator's interrupt + +- **WHEN** the operator interrupts a running follow +- **THEN** the command ends cleanly, without reporting a failure diff --git a/openspec/changes/archive/2026-09-19-work-logs-command/tasks.md b/openspec/changes/archive/2026-09-19-work-logs-command/tasks.md new file mode 100644 index 00000000..d230c0d3 --- /dev/null +++ b/openspec/changes/archive/2026-09-19-work-logs-command/tasks.md @@ -0,0 +1,49 @@ +## 1. A single read + +- [x] 1.1 Add `workLogsCmd` (`work logs `), `--url`/token flags via + `workAPIFlags`, calling `GET /v1/items/{id}/log` via `workRequest` and + printing the log field — verify with a test asserting the printed output + matches a fake API's answer +- [x] 1.2 An item with no output yet (`log: null`) prints nothing and does + not fail — verify with a test against a fake API answering `log: null` +- [x] 1.3 An id the API refuses (404, unknown id) fails the command naming + the id, the way `abort`/`remove` already report a refusal — verify with + a test against a fake API answering the refusal +- [x] 1.4 Wire `workLogsCmd` into `workCmd`'s subcommands — verify with + `spinloop work logs --help` listing it and `spinloop work --help` + listing `logs` among the subcommands (no `--url` completion exists on + any sibling subcommand to mirror — it names a free-text address, not a + file or an enum) + +## 2. Following + +- [x] 2.1 Add `-f`/`--follow` to `workLogsCmd`; unset, behavior is + unchanged from task 1 — verify with a test asserting the flag defaults + false and a single read still happens +- [x] 2.2 Add `workLogsInterval` (1s, a package var) and + `followWorkLogsLoop`: each tick, `GET /v1/items/{id}/log`, printing only + the new suffix when the new answer still starts with what was already + printed, the whole answer otherwise — verify with a test driving the + loop over a sequence of fake answers, asserting only the new bytes are + printed each time +- [x] 2.3 Each tick also reads the item's `state` from `GET /v1/items` + (decoded as `[]orchestrator.ItemView`, matching `work list`'s own + decode); the loop keeps polling through `backlog`/`running` and stops + once the state is `done`/`failed`, after one final log poll — verify + with a test asserting the loop keeps polling while state is + `backlog`/`running`, printing nothing new once no more arrives, and + ends the first tick it sees `done`/`failed`, having already printed that + tick's log +- [x] 2.4 An item that drops out of the list mid-follow (removed) ends the + follow the same way an ended state does, without failing — verify with + a test whose fake API stops carrying the id partway through +- [x] 2.5 Wire `-f` through `followUntilInterrupted`, matching `fleet logs + -f`/`remote logs -f`'s own wiring: an interrupt ends the follow cleanly, + not as a failure — verify with a test cancelling the loop's context + directly (the same pattern `follow_test.go` already uses for the shared + helper) and asserting a nil return + +## 3. Tests and the suite + +- [x] 3.1 Run the full suite — verify with `go test ./... -race -cover` + passing, no regressions, `cmd/spinloop` coverage holding at or above 80% diff --git a/openspec/specs/work-commands/spec.md b/openspec/specs/work-commands/spec.md index 6e71dbf0..9877d014 100644 --- a/openspec/specs/work-commands/spec.md +++ b/openspec/specs/work-commands/spec.md @@ -10,7 +10,7 @@ token, and the run's view of the items is the source of truth. ### Requirement: The work commands as work list clients `spinloop work` SHALL be a top-level command group with the subcommands add, -list, abort and remove, each a client of the orchestrator's work list API. Every +list, abort, remove and logs, each a client of the orchestrator's work list API. Every subcommand SHALL take a `--url` flag naming the API's base address, and SHALL present the API's token as a bearer on every request it makes — resolved from `--api-token`, else `--api-token-file`, else the `SPINLOOP_API_TOKEN` @@ -42,6 +42,65 @@ take any lock beside them. `SPINLOOP_API_TOKEN` environment is set - **THEN** the command presents that value as the bearer +### Requirement: Reading an item's log through the work list API + +`spinloop work logs ` SHALL call the API's `GET /v1/items/{id}/log` +path for the item's kept agent output and print it. An item the API +answers as carrying no output yet SHALL be printed as empty, not treated +as a fault. An id the run does not carry SHALL be refused, naming it, the +way the API states it. + +Given `-f`/`--follow`, the command SHALL instead poll the API for new +output and print it as it arrives, the way `tail -f` does: it SHALL keep +polling while the item's state is `backlog` or `running` — an item named +before, or just as, it starts is still followed — and SHALL stop once the +API reports the item `done` or `failed`, printing whatever output arrived +up to that point first. The operator's interrupt SHALL also end a follow, +cleanly. The command SHALL NOT itself read the items file, the state, or +the logs directly: the API's answers are the whole source of what it +prints and when it stops. + +#### Scenario: An item's kept output is printed + +- **WHEN** the operator runs `work logs` naming an item with kept output +- **THEN** the command prints it + +#### Scenario: An item with no output yet is empty, not a fault + +- **WHEN** the operator runs `work logs` naming an item the API answers +with no output yet +- **THEN** the command prints nothing, and does not fail + +#### Scenario: An id the run does not carry is refused + +- **WHEN** the operator runs `work logs` naming an id the API does not +carry +- **THEN** the command fails, naming the id, the way the API states it + +#### Scenario: A follow streams new output as it arrives + +- **WHEN** the operator runs `work logs -f` on a running item, and its +agent produces more output +- **THEN** the command prints the new output as later polls see it + +#### Scenario: A follow waits through the backlog + +- **WHEN** the operator runs `work logs -f` on an item the run has not yet +started +- **THEN** the command keeps polling rather than ending, and starts +printing output once the item runs + +#### Scenario: A follow ends once the item ends + +- **WHEN** a followed item's state becomes `done` or `failed` +- **THEN** the command prints whatever output arrived up to that point and +ends + +#### Scenario: A follow ends on the operator's interrupt + +- **WHEN** the operator interrupts a running follow +- **THEN** the command ends cleanly, without reporting a failure + ### Requirement: Adding an item through the work list API `spinloop work add` SHALL take the item's fields from flags — its id, its