From a9c2b5d0c412e4557d31e9f675e069fc9f7a619c Mon Sep 17 00:00:00 2001 From: Vic G Date: Sat, 24 Oct 2026 18:11:19 -0400 Subject: [PATCH] Fix #42: Resume unfinished tasks from saved trajectory checkpoints Closes #42 --- README.md | 28 +- cmd/tinyagent/main.go | 193 +++++++++- cmd/tinyagent/main_test.go | 5 +- cmd/tinyagent/resume_contract_test.go | 347 ++++++++++++++++++ internal/agent/compact.go | 2 + internal/agent/compact_test.go | 3 +- internal/agent/loop.go | 164 +++++++-- internal/agent/loop_test.go | 3 +- internal/agent/resume.go | 58 +++ internal/agent/resume_contract_test.go | 200 ++++++++++ internal/agent/resume_wire_test.go | 140 +++++++ internal/tools/process_test.go | 16 +- .../trajectory/checkpoint_contract_test.go | 221 +++++++++++ internal/trajectory/resume.go | 187 ++++++++++ internal/trajectory/trajectory.go | 60 ++- 15 files changed, 1566 insertions(+), 61 deletions(-) create mode 100644 cmd/tinyagent/resume_contract_test.go create mode 100644 internal/agent/resume.go create mode 100644 internal/agent/resume_contract_test.go create mode 100644 internal/agent/resume_wire_test.go create mode 100644 internal/trajectory/checkpoint_contract_test.go create mode 100644 internal/trajectory/resume.go diff --git a/README.md b/README.md index 423a190..bc7acaf 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,8 @@ The `codex` provider uses your ChatGPT subscription instead of an API key. Sign | Flag | Default | | |---|---|---| | `-h, --help` | | Print usage to stdout and exit | -| `-t, --task` | — | Task prompt (required); `-` reads it from stdin | +| `-t, --task` | — | Task prompt (required unless `--resume`); `-` reads it from stdin | +| `--resume` | — | Continue a validated unfinished trajectory; requires explicit matching `--cwd` and a new `-o` path | | `-m, --model` | `$AGENT_MODEL`, else config `model`, else `openai/gpt-4o` | `/`; no `/` means openai. Providers: openai, openrouter, anthropic, ollama, codex | | `-c, --config` | user config, else built-in | YAML config file; see [Config](#config) | | `--cwd` | `.` | Root directory; tools can't reach outside it | @@ -97,6 +98,31 @@ The limit flags must be >= 0, and `--cost-limit` must also be finite. A typo suc Environment: `OPENAI_API_KEY`, `OPENAI_BASE_URL` (any OpenAI-compatible endpoint), `ANTHROPIC_API_KEY`, `ANTHROPIC_BASE_URL`, `OPENROUTER_API_KEY`, `OLLAMA_HOST`, `CODEX_HOME`. +### Resume an unfinished task + +```sh +# Save complete tool turns; this run stops at 20 cumulative model steps. +tinyagent -t "fix the failing test" --cwd ~/src/project --step-limit 20 --no-compact -o run.json + +# Inspect the workspace (for example, git diff) and rerun relevant checks. +# Continue the same task with a ceiling of 40 total steps. +tinyagent --resume run.json --cwd ~/src/project --step-limit 40 --no-compact -o continued.json +``` + +`--resume ` is noninteractive and uses the saved task and model. Do not pass `--task`; an explicitly different `--model` is rejected. The saved model wins over `$AGENT_MODEL` and the config's default model. Credentials and provider endpoints come from the current normal setup, never the checkpoint. Paths to checkpoints are relative to the invoking shell's directory, independently of `--cwd`. + +An explicit `--cwd` must resolve to the recorded canonical root (directory symlinks are resolved). The effective system prompt must match the saved prompt, including `{{cwd}}`/`{{os}}` substitutions and the recorded AGENTS.md loading mode. Current config is used to reconstruct it without appending AGENTS.md twice. Changed instructions, a conflicting `--no-agents-md`, or changed tool definitions/compatibility require a fresh task. In particular, a task that edits its effective AGENTS.md cannot resume under changed instructions. + +Only new-format, unfinished checkpoints at a complete model-query boundary are eligible: an in-progress save or a stop caused by a step/cost/time limit, interruption, or error. Successful `Submitted`/`Completed` runs cannot be extended. Legacy logs, unsupported versions/state, malformed or partial tool turns, and **compacted checkpoints** are refused before provider setup. This release does not restore the separate compacted model view; use `--no-compact` when the ability to resume later is needed. Checkpoint loading is limited to regular files of at most **64 MiB**. Larger trajectories remain audit logs. + +A distinct `-o`/`--output` is required, and any existing destination is refused, including symlink and hard-link aliases of the source. The input stays unchanged. The validated starting state is saved to the new destination before the first request; completed tool turns and termination then replace it using a private temporary file. Initial publication requires filesystem hard-link support to publish atomically without replacing another file. A save failure in resume mode stops further work, prints an error, and exits nonzero, preserving the last complete checkpoint. Ordinary fresh-run save failures retain their existing warning behavior. Rename/link publication does not promise power-loss durability. + +Steps, cost, token counts, and cache totals remain cumulative. Step/cost limits are cumulative ceilings: if already exhausted, no model request is made; increase the ceiling explicitly (or set it to 0 for unlimited). Runtime limits, tool timeouts, output settings, and compaction settings use normal config/flag precedence. Wall time and the stderr elapsed duration restart for each invocation; stopped time is not charged. Normal streaming, summaries, and success/failure exit codes still apply. The next action is a new model query, which incurs ordinary context/token costs; loading a checkpoint never dispatches historical tool calls. + +New JSON keeps the existing top-level trajectory fields and adds versioned metadata: `schema_version` (currently 1), `root`, `system_prompt`, `skip_agents_md`, `tools`, and `tool_signature`. `resume` records the input path, previous status/submission, cumulative step, and continuation-message index; nested `previous` entries retain earlier resume boundaries. These audit paths are not read as configuration. Metadata excludes credentials and environment dumps. Checkpoints still contain prompts, code, and tool outputs: treat them as private, trusted local execution inputs. New checkpoint and temporary files use owner-only permissions (0600) where supported. + +A checkpoint restores **conversation state**, not files, Git state, shell variables, live processes, or external services. The model receives a continuation notice and stderr warns to inspect and revalidate the current workspace. Commands may have produced unrecorded side effects after the last save, especially during a hard crash in a tool batch. Historical process IDs cannot reattach to children from another invocation. There is no rollback or exactly-once execution guarantee; inspect current state before resuming. + ### Project instructions (AGENTS.md) Unless `--no-agents-md` is set, tinyagent looks for `AGENTS.md` starting at `--cwd` and walking up through its parents, stopping at the repository root (the nearest directory containing `.git`). The nearest file wins, and only that one is added to the system prompt. Without a `.git` above `--cwd`, only `/AGENTS.md` is checked. The lookup never goes above the repository root, and `-v` prints which file was loaded. This only changes the prompt: tools are still confined to `--cwd`. diff --git a/cmd/tinyagent/main.go b/cmd/tinyagent/main.go index c9e8ca0..7506263 100644 --- a/cmd/tinyagent/main.go +++ b/cmd/tinyagent/main.go @@ -35,11 +35,11 @@ var ( // options holds the parsed command line. type options struct { - task, modelName, configPath, cwd, output string - stepLimit, timeLimit int - costLimit float64 - verbose, noStream, noAgentsMD, noCompact bool - showHelp, showVersion, doUpdate bool + task, modelName, configPath, cwd, output, resume string + stepLimit, timeLimit int + costLimit float64 + verbose, noStream, noAgentsMD, noCompact bool + showHelp, showVersion, doUpdate bool } // newFlagSet defines the command-line flags, storing their values in o. Each @@ -50,7 +50,7 @@ func newFlagSet(o *options) *flag.FlagSet { fs.Usage = func() { printUsage(fs.Output(), fs) } fs.BoolVar(&o.showHelp, "help", false, "show this help and exit") fs.BoolVar(&o.showHelp, "h", false, "shorthand for --help") - fs.StringVar(&o.task, "task", "", `task/prompt (required); "-" reads it from stdin`) + fs.StringVar(&o.task, "task", "", `task/prompt (required unless --resume); "-" reads it from stdin`) fs.StringVar(&o.task, "t", "", "shorthand for --task") fs.StringVar(&o.modelName, "model", defaultModel(), "provider/model; if unset, $AGENT_MODEL, then the config's model") fs.StringVar(&o.modelName, "m", defaultModel(), "shorthand for --model") @@ -62,6 +62,7 @@ func newFlagSet(o *options) *flag.FlagSet { fs.IntVar(&o.timeLimit, "time-limit", 0, "max wall-clock seconds, 0 = unlimited") fs.StringVar(&o.output, "output", "", "write trajectory JSON") fs.StringVar(&o.output, "o", "", "shorthand for --output") + fs.StringVar(&o.resume, "resume", "", "continue an unfinished run from a trajectory JSON checkpoint") fs.BoolVar(&o.verbose, "verbose", false, "print tool outputs") fs.BoolVar(&o.verbose, "v", false, "shorthand for --verbose") fs.BoolVar(&o.noStream, "no-stream", false, "wait for full responses instead of streaming") @@ -114,6 +115,9 @@ func run() int { fmt.Println(msg) return 0 } + if o.resume != "" { + return runResume(&o, fs) + } if o.task == "" { fmt.Fprintln(os.Stderr, "error: --task is required") fs.Usage() @@ -132,7 +136,7 @@ func run() int { o.modelName = resolveModel(fs, o.modelName, cfg.Model) model.PriceOverride = [2]float64{cfg.InputPricePerMTok, cfg.OutputPricePerMTok} - root, err := filepath.Abs(o.cwd) + root, err := canonicalRoot(o.cwd) if err != nil { return fail(err) } @@ -174,10 +178,11 @@ func run() int { NoCompact: o.noCompact, } if o.output != "" { - a.OnStep = func(t *trajectory.Trajectory) { + a.OnStep = func(t *trajectory.Trajectory) error { if err := t.Save(o.output); err != nil { fmt.Fprintln(os.Stderr, "warning: saving trajectory:", err) } + return nil } } @@ -190,6 +195,171 @@ func run() int { return 1 } +// runResume continues an unfinished run from a validated checkpoint. It reuses +// the saved task and model, requires a matching working directory and a +// distinct output destination, and refuses to continue if the effective +// instructions or tool set have changed. +func runResume(o *options, fs *flag.FlagSet) int { + if flagWasSet(fs, "task", "t") { + fmt.Fprintln(os.Stderr, "error: --resume and --task are mutually exclusive") + return 2 + } + + if !flagWasSet(fs, "cwd") { + return fail(errors.New("resume: an explicit --cwd matching the checkpoint root is required")) + } + + src, err := trajectory.Load(o.resume) + if err != nil { + return fail(fmt.Errorf("resume: %w", err)) + } + + // The saved model wins unless an explicit, different model is given. + if flagWasSet(fs, "model", "m") && o.modelName != src.Model { + return fail(fmt.Errorf("resume: --model %q conflicts with the checkpoint's model %q", o.modelName, src.Model)) + } + modelName := src.Model + + cfg, err := config.Load(config.Find(o.configPath)) + if err != nil { + return fail(err) + } + applyFlagOverrides(fs, &cfg, o.stepLimit, o.costLimit, o.timeLimit) + model.PriceOverride = [2]float64{cfg.InputPricePerMTok, cfg.OutputPricePerMTok} + + root, err := canonicalRoot(o.cwd) + if err != nil { + return fail(err) + } + if fi, err := os.Stat(root); err != nil || !fi.IsDir() { + return fail(fmt.Errorf("--cwd %s is not a directory", o.cwd)) + } + if root != src.Root { + return fail(fmt.Errorf("resume: --cwd %s does not match the checkpoint's root %s", root, src.Root)) + } + + if o.output == "" { + return fail(errors.New("resume: --output is required to write the continued trajectory")) + } + if sameFile(o.resume, o.output) { + return fail(errors.New("resume: --output must be distinct from the source checkpoint")) + } + if _, err := os.Lstat(o.output); err == nil { + return fail(errors.New("resume: --output already exists; choose a new destination")) + } else if !errors.Is(err, os.ErrNotExist) { + return fail(fmt.Errorf("resume: checking --output: %w", err)) + } + if flagWasSet(fs, "no-agents-md") && o.noAgentsMD != src.SkipAgentsMD { + return fail(errors.New("resume: --no-agents-md conflicts with the recorded loading mode; start a fresh task")) + } + if err := agent.ValidateResume(src, cfg, root); err != nil { + return fail(fmt.Errorf("resume: %w", err)) + } + + var onText func(string) + if !o.noStream { + onText = func(s string) { os.Stdout.WriteString(s) } + } + m, err := model.New(modelName, onText) + if err != nil { + code := fail(err) + printHint(os.Stderr, modelName, err.Error()) + return code + } + warnUnpriced(os.Stderr, modelName, cfg.CostLimit) + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + + // The agent records this boundary and clears the active termination before + // persisting the new starting state. + prevStatus := src.ExitStatus + if prevStatus == "" { + prevStatus = "in progress" + } + + a := &agent.Agent{ + Model: m, + ModelName: modelName, + Tools: &tools.Executor{ + Root: root, + BashTimeout: time.Duration(cfg.BashTimeoutSec) * time.Second, + MaxOutputChars: cfg.MaxOutputChars, + MaxProcesses: cfg.MaxBackgroundProcesses, + Stream: os.Stdout, + }, + Config: cfg, + Out: os.Stdout, + Err: os.Stderr, + Verbose: o.verbose, + Streamed: !o.noStream, + SkipAgentsMD: src.SkipAgentsMD, + NoCompact: o.noCompact, + Resume: src, + ResumeSource: o.resume, + } + initial := true + a.OnStep = func(t *trajectory.Trajectory) error { + if initial { + if err := t.SaveNew(o.output); err != nil { + return err + } + initial = false + return nil + } + return t.Save(o.output) + } + + fmt.Fprintf(os.Stderr, "resuming %s from %s (previous status %s after %d steps)\n", modelName, o.resume, prevStatus, src.Steps) + fmt.Fprintln(os.Stderr, "note: inspect and revalidate the workspace; it may have changed and commands may have produced unrecorded side effects after the last save. Live processes are not restored") + + start := time.Now() + t := a.Run(ctx, src.Task) + reportSummary(os.Stderr, modelName, t, time.Since(start)) + if t.ExitStatus == agent.StatusSubmitted || t.ExitStatus == agent.StatusCompleted { + return 0 + } + return 1 +} + +// sameFile reports whether two paths refer to the same file, following +// symlinks and detecting hard links. +func sameFile(a, b string) bool { + ai, errA := os.Stat(a) + bi, errB := os.Stat(b) + if errA == nil && errB == nil && os.SameFile(ai, bi) { + return true + } + ra, _ := filepath.EvalSymlinks(a) + rb, _ := filepath.EvalSymlinks(b) + if ra != "" && rb != "" && ra == rb { + return true + } + aa, _ := filepath.Abs(a) + bb, _ := filepath.Abs(b) + return filepath.Clean(aa) == filepath.Clean(bb) +} + +func flagWasSet(fs *flag.FlagSet, names ...string) bool { + found := false + fs.Visit(func(f *flag.Flag) { + for _, name := range names { + if f.Name == name { + found = true + } + } + }) + return found +} + +func canonicalRoot(path string) (string, error) { + root, err := filepath.Abs(path) + if err != nil { + return "", err + } + return filepath.EvalSymlinks(root) +} + // usageGroups lays out --help: each flag's shorthand, long name and value // placeholder, by group. Help text and defaults come from the FlagSet. var usageGroups = []struct { @@ -197,7 +367,7 @@ var usageGroups = []struct { flags [][3]string }{ {"", [][3]string{{"h", "help", ""}}}, - {"Task", [][3]string{{"t", "task", "string"}, {"m", "model", "string"}, {"c", "config", "path"}, {"", "cwd", "dir"}}}, + {"Task", [][3]string{{"t", "task", "string"}, {"m", "model", "string"}, {"c", "config", "path"}, {"", "cwd", "dir"}, {"", "resume", "path"}}}, {"Limits", [][3]string{{"", "step-limit", "n"}, {"", "cost-limit", "usd"}, {"", "time-limit", "sec"}, {"", "no-compact", ""}}}, {"Output", [][3]string{{"o", "output", "path"}, {"v", "verbose", ""}, {"", "no-stream", ""}, {"", "no-agents-md", ""}}}, {"Other", [][3]string{{"V", "version", ""}, {"", "update", ""}}}, @@ -220,7 +390,7 @@ func printUsage(w io.Writer, fs *flag.FlagSet) { width = max(width, len(left(f))) } } - fmt.Fprint(w, "Usage: tinyagent -t \"\" [flags] (or --task \"\")\n echo \"\" | tinyagent -t -\n\n") + fmt.Fprint(w, "Usage: tinyagent -t \"\" [flags] (or --task \"\")\n echo \"\" | tinyagent -t -\n tinyagent --resume --cwd -o [flags]\n\n") for _, g := range usageGroups { if g.name != "" { fmt.Fprintf(w, "%s:\n", g.name) @@ -236,7 +406,8 @@ func printUsage(w io.Writer, fs *flag.FlagSet) { } fmt.Fprint(w, "\nExamples:\n"+ " tinyagent -t \"fix the failing test\" -m anthropic/claude-sonnet-4\n"+ - " cat task.md | tinyagent -t - --step-limit 30 -o run.json\n") + " cat task.md | tinyagent -t - --step-limit 30 -o run.json\n"+ + " tinyagent --resume run.json --cwd . --step-limit 60 -o continued.json\n") } // reportSummary prints one line on how the run ended: status, steps, elapsed diff --git a/cmd/tinyagent/main_test.go b/cmd/tinyagent/main_test.go index 0ef6d46..ac2695a 100644 --- a/cmd/tinyagent/main_test.go +++ b/cmd/tinyagent/main_test.go @@ -136,13 +136,15 @@ func TestSummaryLine(t *testing.T) { // first and -V for --version. const wantUsage = `Usage: tinyagent -t "" [flags] (or --task "") echo "" | tinyagent -t - + tinyagent --resume --cwd -o [flags] -h, --help show this help and exit Task: - -t, --task string task/prompt (required); "-" reads it from stdin + -t, --task string task/prompt (required unless --resume); "-" reads it from stdin -m, --model string provider/model; if unset, $AGENT_MODEL, then the config's model (default "openai/gpt-4o") -c, --config path YAML config (default: user config if present, else built-in) --cwd dir working directory (default ".") + --resume path continue an unfinished run from a trajectory JSON checkpoint Limits: --step-limit n max model calls, 0 = unlimited --cost-limit usd max USD, 0 = unlimited @@ -160,6 +162,7 @@ Other: Examples: tinyagent -t "fix the failing test" -m anthropic/claude-sonnet-4 cat task.md | tinyagent -t - --step-limit 30 -o run.json + tinyagent --resume run.json --cwd . --step-limit 60 -o continued.json ` // runWith runs the CLI with args, returning the exit code and what went to diff --git a/cmd/tinyagent/resume_contract_test.go b/cmd/tinyagent/resume_contract_test.go new file mode 100644 index 0000000..04030da --- /dev/null +++ b/cmd/tinyagent/resume_contract_test.go @@ -0,0 +1,347 @@ +package main + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/vicgupta/tinyagent/internal/agent" + "github.com/vicgupta/tinyagent/internal/config" + "github.com/vicgupta/tinyagent/internal/model" + "github.com/vicgupta/tinyagent/internal/tools" + "github.com/vicgupta/tinyagent/internal/trajectory" +) + +func resumeFixture(t *testing.T) (*trajectory.Trajectory, string, string, string) { + t.Helper() + root, err := canonicalRoot(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := os.Mkdir(filepath.Join(root, ".git"), 0700); err != nil { + t.Fatal(err) + } + cfg := filepath.Join(root, "config.yaml") + if err := os.WriteFile(cfg, []byte("model: codex/ignored-config-default\n"), 0600); err != nil { + t.Fatal(err) + } + system := agent.SystemPrompt(config.Default(), root, false) + tr := &trajectory.Trajectory{ + Task: "saved task", Model: "openai/gpt-4o", SchemaVersion: trajectory.SchemaVersion, Root: root, + SystemPrompt: system, ToolSignature: agent.ToolSignature(), ExitStatus: agent.StatusError, Submission: "previous provider failure", + Steps: 1, Cost: 0.01, InputTokens: 100, OutputTokens: 20, CacheReadTokens: 9, CacheWriteTokens: 7, + Messages: []model.Message{ + {Role: "system", Content: system}, {Role: "user", Content: "saved task"}, + {Role: "assistant", ToolCalls: []model.ToolCall{{ID: "historic", Type: "function", Function: model.FunctionCall{Name: "write_file", Arguments: `{"path":"must-not-replay","content":"bad"}`}}}}, + {Role: "tool", ToolCallID: "historic", Content: "ok"}, + }, + } + for _, spec := range tools.Specs() { + tr.Tools = append(tr.Tools, spec.Function.Name) + } + source := filepath.Join(root, "source.json") + if err := tr.Save(source); err != nil { + t.Fatal(err) + } + return tr, source, filepath.Join(root, "continued.json"), cfg +} + +func TestCLIResumePersistsBeforeQueryAndPreservesSource(t *testing.T) { + tr, source, output, cfg := resumeFixture(t) + if err := os.WriteFile(filepath.Join(tr.Root, "AGENTS.md"), []byte("Keep this project instruction."), 0o600); err != nil { + t.Fatal(err) + } + tr.SystemPrompt = agent.SystemPrompt(config.Default(), tr.Root, false) + tr.Messages[0].Content = tr.SystemPrompt + if err := tr.Save(source); err != nil { + t.Fatal(err) + } + before, _ := os.ReadFile(source) + t.Setenv("AGENT_MODEL", "codex/ignored-env-default") + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + starting, err := trajectory.Load(output) + if err != nil { + t.Errorf("starting checkpoint not readable before query: %v", err) + } else if starting.ExitStatus != "" || starting.Submission != "" || starting.Resume == nil || starting.Resume.ExitStatus != agent.StatusError || starting.Resume.Submission != "previous provider failure" { + t.Error("starting state/provenance incorrect") + } + var body struct { + Model string + Messages []model.Message + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + } + if body.Model != "gpt-4o" || len(body.Messages) != 5 || body.Messages[4].Content != trajectory.ContinuationNotice { + t.Error("saved model/history not restored") + } + if strings.Count(body.Messages[0].Content, "Keep this project instruction.") != 1 { + t.Error("AGENTS.md instructions missing or duplicated") + } + w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"done"},"finish_reason":"stop"}],"usage":{"prompt_tokens":100,"completion_tokens":20}}`)) + })) + defer srv.Close() + t.Setenv("OPENAI_BASE_URL", srv.URL) + code := 0 + stderr := captureStderr(t, func() { + code, _, _ = runWith(t, "--resume", source, "--cwd", tr.Root, "-c", cfg, "-o", output, "--no-stream", "--step-limit", "40") + }) + if code != 0 || calls != 1 { + t.Fatalf("exit %d, calls %d: %s", code, calls, stderr) + } + if !strings.Contains(stderr, "previous status Error after 1 steps") || !strings.Contains(stderr, "unrecorded side effects") || !strings.Contains(stderr, "200 input / 40 output tokens") { + t.Fatalf("missing feedback: %s", stderr) + } + after, _ := os.ReadFile(source) + if string(before) != string(after) { + t.Fatal("source changed") + } + if _, err := os.Stat(filepath.Join(tr.Root, "must-not-replay")); !os.IsNotExist(err) { + t.Fatal("replayed historical write") + } + data, err := os.ReadFile(output) + if err != nil { + t.Fatal(err) + } + var got trajectory.Trajectory + if err := json.Unmarshal(data, &got); err != nil { + t.Fatal(err) + } + if got.Steps != 2 || got.InputTokens != 200 || got.OutputTokens != 40 || got.Cost <= tr.Cost || got.CacheReadTokens != 9 || got.CacheWriteTokens != 7 { + t.Fatalf("accounting: %+v", got) + } +} + +func TestCLIResumeRefusalsBeforeProviderSetup(t *testing.T) { + for _, which := range []string{"task", "empty task", "model", "missing root", "wrong root", "instructions", "mode", "tool signature", "tools", "existing", "source", "symlink", "hardlink", "dangling symlink", "bad history", "legacy", "compacted"} { + t.Run(which, func(t *testing.T) { + tr, source, output, cfg := resumeFixture(t) + // A Codex model would read local auth during setup. Every refusal must + // happen before that, even if the provider cannot initialize. + tr.Model = "codex/gpt-5" + t.Setenv("CODEX_HOME", t.TempDir()) + args := []string{"--resume", source, "--cwd", tr.Root, "-c", cfg, "-o", output, "--no-stream"} + want := "" + switch which { + case "task": + args = append(args, "--task", "other") + want = "mutually exclusive" + case "empty task": + args = append(args, "-t", "") + want = "mutually exclusive" + case "model": + args = append(args, "-m", "openai/different") + want = "conflicts" + case "missing root": + args = append(args[:2], args[4:]...) + want = "explicit --cwd" + case "wrong root": + args[3] = t.TempDir() + want = "root" + case "instructions": + if err := os.WriteFile(filepath.Join(tr.Root, "AGENTS.md"), []byte("new instructions"), 0600); err != nil { + t.Fatal(err) + } + want = "prompt" + case "mode": + args = append(args, "--no-agents-md") + want = "loading mode" + case "tool signature": + tr.ToolSignature = "other" + want = "compatibility" + case "tools": + tr.Tools = tr.Tools[:1] + want = "tool" + case "existing": + if err := os.WriteFile(output, []byte("preserve me"), 0600); err != nil { + t.Fatal(err) + } + want = "exists" + case "source": + args[7] = source + want = "distinct" + case "symlink": + if err := os.Symlink(source, output); err != nil { + t.Skip(err) + } + want = "distinct" + case "hardlink": + if err := os.Link(source, output); err != nil { + t.Skip(err) + } + want = "distinct" + case "dangling symlink": + if err := os.Symlink(source+".missing", output); err != nil { + t.Skip(err) + } + want = "exists" + case "bad history": + tr.Messages = tr.Messages[:3] + want = "result" + case "legacy": + tr.SchemaVersion = 0 + want = "metadata" + case "compacted": + tr.Compactions = []trajectory.Compaction{{Step: 1}} + want = "compacted" + } + if err := tr.Save(source); err != nil { + t.Fatal(err) + } + if which == "hardlink" { + if err := os.Remove(output); err != nil { + t.Fatal(err) + } + if err := os.Link(source, output); err != nil { + t.Fatal(err) + } + } + before, _ := os.ReadFile(source) + code := 0 + stderr := captureStderr(t, func() { code, _, _ = runWith(t, args...) }) + if code == 0 || !strings.Contains(stderr, want) || strings.Contains(stderr, "no login") { + t.Fatalf("exit %d, want %s: %s", code, want, stderr) + } + after, _ := os.ReadFile(source) + if string(before) != string(after) { + t.Fatal("source modified on refusal") + } + if which == "existing" { + b, _ := os.ReadFile(output) + if string(b) != "preserve me" { + t.Fatal("existing destination modified") + } + } + }) + } +} + +func TestCLIResumeExhaustedLimitCanResumeAgain(t *testing.T) { + tr, source, output, cfg := resumeFixture(t) + // Point at a local server to prove no request is made with exhausted totals. + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"done"}}]}`)) + })) + defer srv.Close() + t.Setenv("OPENAI_BASE_URL", srv.URL) + for _, which := range []string{"step", "cost"} { + t.Run(which, func(t *testing.T) { + dest := output + which + limit := []string{"--step-limit", "1"} + if which == "cost" { + limit = []string{"--cost-limit", "0.01"} + } + args := append([]string{"--resume", source, "--cwd", tr.Root, "-c", cfg, "-o", dest, "--no-stream"}, limit...) + stderr := captureStderr(t, func() { + code, _, _ := runWith(t, args...) + if code != 1 { + t.Errorf("exit %d", code) + } + }) + if calls != 0 { + t.Fatal("request with exhausted ceiling") + } + if !strings.Contains(stderr, "after 1 steps in 0s") { + t.Fatalf("elapsed/cumulative summary: %s", stderr) + } + c, err := trajectory.Load(dest) + if err != nil || c.Resume == nil || c.Resume.ExitStatus != agent.StatusError { + t.Fatalf("new interrupted boundary: %v", err) + } + next := dest + "-next" + captureStderr(t, func() { + code, _, _ := runWith(t, "--resume", dest, "--cwd", tr.Root, "-c", cfg, "-o", next, "--no-stream", "--step-limit", "2", "--cost-limit", "0") + if code != 0 { + t.Errorf("second resume exit %d", code) + } + }) + if calls != 1 { + t.Fatalf("queries after increased limit: %d", calls) + } + calls = 0 + }) + } +} + +func TestCLIResumeCanonicalRootAlias(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlink creation may require privileges") + } + tr, source, output, cfg := resumeFixture(t) + alias := filepath.Join(t.TempDir(), "root-alias") + if err := os.Symlink(tr.Root, alias); err != nil { + t.Skip(err) + } + stderr := captureStderr(t, func() { + code, _, _ := runWith(t, "--resume", source, "--cwd", alias, "-c", cfg, "-o", output, "--step-limit", "1") + if code != 1 { + t.Errorf("exit %d", code) + } + }) + if strings.Contains(stderr, "does not match") { + t.Fatal(stderr) + } + if _, err := trajectory.Load(output); err != nil { + t.Fatal(err) + } +} + +func TestCLIResumeInitialSaveFailure(t *testing.T) { + tr, source, output, cfg := resumeFixture(t) + blocker := filepath.Join(tr.Root, "not-a-directory") + if err := os.WriteFile(blocker, []byte("file"), 0600); err != nil { + t.Fatal(err) + } + output = filepath.Join(blocker, "out.json") + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { calls++; w.Write([]byte(`{}`)) })) + defer srv.Close() + t.Setenv("OPENAI_BASE_URL", srv.URL) + stderr := captureStderr(t, func() { + code, _, _ := runWith(t, "--resume", source, "--cwd", tr.Root, "-c", cfg, "-o", output) + if code != 1 { + t.Errorf("exit %d", code) + } + }) + if calls != 0 || !strings.Contains(stderr, "error:") && !strings.Contains(stderr, "saving trajectory") { + t.Fatalf("calls %d: %s", calls, stderr) + } + if _, err := trajectory.Load(source); err != nil { + t.Fatal(err) + } +} + +func TestCLIResumeRetainsRecordedSkipAgentsMode(t *testing.T) { + tr, source, output, cfg := resumeFixture(t) + tr.SkipAgentsMD = true + if err := tr.Save(source); err != nil { + t.Fatal(err) + } + // The saved invocation excluded AGENTS.md. Its current contents must not + // silently change that mode when the flag is absent on the next invocation. + if err := os.WriteFile(filepath.Join(tr.Root, "AGENTS.md"), []byte("new instructions excluded by recorded mode"), 0600); err != nil { + t.Fatal(err) + } + stderr := captureStderr(t, func() { + code, _, _ := runWith(t, "--resume", source, "--cwd", tr.Root, "-c", cfg, "-o", output, "--step-limit", "1") + if code != 1 { + t.Errorf("exit %d", code) + } + }) + loaded, err := trajectory.Load(output) + if err != nil { + t.Fatalf("recorded loading mode not retained: %v; %s", err, stderr) + } + if !loaded.SkipAgentsMD || loaded.Messages[0].Content != tr.SystemPrompt { + t.Fatal("saved instructions changed") + } +} diff --git a/internal/agent/compact.go b/internal/agent/compact.go index a5007b2..d04a619 100644 --- a/internal/agent/compact.go +++ b/internal/agent/compact.go @@ -237,6 +237,8 @@ func (a *Agent) summarize(ctx context.Context, t *trajectory.Trajectory, task st t.Cost += cost.USD t.InputTokens += cost.InputTokens t.OutputTokens += cost.OutputTokens + t.CacheReadTokens += cost.CacheReadTokens + t.CacheWriteTokens += cost.CacheWriteTokens s := strings.TrimSpace(resp.Content) if s == "" { return "", errors.New("summarizing context: the model returned an empty summary") diff --git a/internal/agent/compact_test.go b/internal/agent/compact_test.go index 23d813a..dc5c2a2 100644 --- a/internal/agent/compact_test.go +++ b/internal/agent/compact_test.go @@ -184,10 +184,11 @@ func TestCompactSummarizes(t *testing.T) { }} a, stderr := compactAgent(t, f, 30000) out := filepath.Join(t.TempDir(), "traj.json") - a.OnStep = func(tr *trajectory.Trajectory) { + a.OnStep = func(tr *trajectory.Trajectory) error { if err := tr.Save(out); err != nil { t.Fatal(err) } + return nil } tr := a.Run(context.Background(), "the task") diff --git a/internal/agent/loop.go b/internal/agent/loop.go index fac6d31..d9dee03 100644 --- a/internal/agent/loop.go +++ b/internal/agent/loop.go @@ -50,10 +50,20 @@ type Agent struct { // NoCompact turns off context compaction: the full history is always // sent, and a context-overflow error ends the run. NoCompact bool - // OnStep is called after every step, e.g. to save the trajectory. - OnStep func(*trajectory.Trajectory) + // OnStep is called after every step, e.g. to save the trajectory. A + // non-nil error stops the run. + OnStep func(*trajectory.Trajectory) error + // Resume, when non-nil, continues that trajectory instead of starting a + // fresh run: its messages, accounting totals, and steps are the starting + // state, and a continuation notice is appended before the first request. + Resume *trajectory.Trajectory + ResumeSource string // audit-only input path for resume provenance } +// resumeNotice is appended to the model's history when a run is resumed, so +// the model knows the workspace may have changed since the last save. +const resumeNotice = trajectory.ContinuationNotice + // Run executes task until submission, a limit, or an error. It always returns // a trajectory describing how the run ended. func (a *Agent) Run(ctx context.Context, task string) *trajectory.Trajectory { @@ -69,17 +79,47 @@ func (a *Agent) Run(ctx context.Context, task string) *trajectory.Trajectory { defer cancel() } - system := strings.NewReplacer("{{cwd}}", a.Tools.Root, "{{os}}", runtime.GOOS).Replace(a.Config.SystemPrompt) - if !a.SkipAgentsMD { - system += a.agentsMD() - } - t := &trajectory.Trajectory{ - Task: task, - Model: a.ModelName, - Messages: []model.Message{ - {Role: "system", Content: system}, - {Role: "user", Content: task}, - }, + var t *trajectory.Trajectory + if a.Resume != nil { + t = a.Resume + if a.ModelName != t.Model { + t.ExitStatus, t.Submission = StatusError, "resume model differs from checkpoint" + return t + } + if err := ValidateResume(t, a.Config, a.Tools.Root); err != nil { + t.ExitStatus, t.Submission = StatusError, err.Error() + return t + } + source := a.ResumeSource + if source == "" { + source = "memory" + } + t.Resume = &trajectory.ResumeInfo{ + Source: source, ExitStatus: t.ExitStatus, Submission: t.Submission, + Steps: t.Steps, Message: len(t.Messages), Previous: t.Resume, + } + t.ExitStatus, t.Submission = "", "" + t.Messages = append(t.Messages, model.Message{Role: "user", Content: resumeNotice}) + // Persist the validated starting state before the first request. + if err := a.step(t); err != nil { + return a.saveFailed(t, err) + } + } else { + system := a.systemPrompt() + t = &trajectory.Trajectory{ + Task: task, + Model: a.ModelName, + SchemaVersion: trajectory.SchemaVersion, + Root: a.Tools.Root, + SystemPrompt: system, + SkipAgentsMD: a.SkipAgentsMD, + Tools: toolNames(tools.Specs()), + ToolSignature: ToolSignature(), + Messages: []model.Message{ + {Role: "system", Content: system}, + {Role: "user", Content: task}, + }, + } } specs := tools.Specs() v := newView(t.Messages) @@ -144,7 +184,9 @@ func (a *Agent) Run(ctx context.Context, task string) *trajectory.Trajectory { fmt.Fprintf(a.Out, " -> %s\n", indent(out)) } } - a.step(t) + if err := a.step(t); err != nil { + return a.saveFailed(t, err) + } } } @@ -157,22 +199,47 @@ func (a *Agent) compactFailed(ctx context.Context, t *trajectory.Trajectory, err return a.finish(t, StatusError, err.Error()) } +// systemPrompt returns the effective system prompt for this agent's run. +func (a *Agent) systemPrompt() string { + system := strings.NewReplacer("{{cwd}}", a.Tools.Root, "{{os}}", runtime.GOOS).Replace(a.Config.SystemPrompt) + if !a.SkipAgentsMD { + system += agentsMD(a.Tools.Root, a.Err, a.Out, a.Verbose) + } + return system +} + +// SystemPrompt returns the effective system prompt a run rooted at root would +// use, given cfg and whether AGENTS.md is skipped. It is exported so callers +// can validate a checkpoint against the current configuration. +func SystemPrompt(cfg config.AgentConfig, root string, skipAgentsMD bool) string { + system := strings.NewReplacer("{{cwd}}", root, "{{os}}", runtime.GOOS).Replace(cfg.SystemPrompt) + if !skipAgentsMD { + system += agentsMD(root, io.Discard, io.Discard, false) + } + return system +} + // agentsMD returns the project's AGENTS.md as a system prompt section, or "" -// when there is none. -func (a *Agent) agentsMD() string { - path := findAgentsMD(a.Tools.Root) - if path == "" { - return "" +// when there is none. Warnings go to errW and verbose notices to outW. +func agentsMD(root string, errW, outW io.Writer, verbose bool) string { + section, err := loadAgentsMD(root, outW, verbose) + if err != nil { + fmt.Fprintln(errW, "warning: reading AGENTS.md:", err) + } + return section +} + +func loadAgentsMD(root string, outW io.Writer, verbose bool) (string, error) { + path, err := findAgentsMD(root) + if err != nil || path == "" { + return "", err } b, err := os.ReadFile(path) if err != nil { - if !errors.Is(err, os.ErrNotExist) { - fmt.Fprintln(a.Err, "warning: reading AGENTS.md:", err) - } - return "" + return "", err } - if a.Verbose { - fmt.Fprintf(a.Out, "loaded AGENTS.md from %s (%d bytes)\n", path, len(b)) + if verbose { + fmt.Fprintf(outW, "loaded AGENTS.md from %s (%d bytes)\n", path, len(b)) } s := string(b) if len(s) > maxAgentsMD { @@ -183,13 +250,22 @@ func (a *Agent) agentsMD() string { } s = s[:n] + "\n[AGENTS.md truncated]" } - return "\n\n## Project instructions (AGENTS.md)\n\n" + s + return "\n\n## Project instructions (AGENTS.md)\n\n" + s, nil +} + +// toolNames returns the tool names from specs, in order. +func toolNames(specs []model.ToolSpec) []string { + names := make([]string, len(specs)) + for i, s := range specs { + names[i] = s.Function.Name + } + return names } // findAgentsMD returns the nearest AGENTS.md at or above root, stopping at the // repository root: the nearest directory containing .git. Without a .git above // root, only root itself is checked. It returns "" when there is no such file. -func findAgentsMD(root string) string { +func findAgentsMD(root string) (string, error) { stop := root for dir := root; ; { if _, err := os.Stat(filepath.Join(dir, ".git")); err == nil { @@ -206,10 +282,12 @@ func findAgentsMD(root string) string { for dir := root; ; { path := filepath.Join(dir, "AGENTS.md") if _, err := os.Stat(path); err == nil { - return path + return path, nil + } else if !errors.Is(err, os.ErrNotExist) { + return "", err } if dir == stop { - return "" + return "", nil } dir = filepath.Dir(dir) } @@ -238,25 +316,37 @@ func ctxStatus(ctx context.Context) (string, bool) { return "", false } -func (a *Agent) finish(t *trajectory.Trajectory, status, submission string) *trajectory.Trajectory { - // Kill any background processes the model left running, and record - // their final output. +func (a *Agent) closeTools(t *trajectory.Trajectory) { if summary := a.Tools.Close(); summary != "" { t.BackgroundProcesses = summary if a.Verbose { fmt.Fprintf(a.Out, "stopped background processes:\n%s", summary) } } - t.ExitStatus = status - t.Submission = submission - a.step(t) +} + +func (a *Agent) finish(t *trajectory.Trajectory, status, submission string) *trajectory.Trajectory { + a.closeTools(t) + t.ExitStatus, t.Submission = status, submission + if err := a.step(t); err != nil { + return a.saveFailed(t, err) + } + return t +} + +// Never retry persistence after failure: the previous complete checkpoint is +// the recovery boundary, even if an in-memory turn has since completed. +func (a *Agent) saveFailed(t *trajectory.Trajectory, err error) *trajectory.Trajectory { + a.closeTools(t) + t.ExitStatus, t.Submission = StatusError, "saving trajectory: "+err.Error() return t } -func (a *Agent) step(t *trajectory.Trajectory) { +func (a *Agent) step(t *trajectory.Trajectory) error { if a.OnStep != nil { - a.OnStep(t) + return a.OnStep(t) } + return nil } func (a *Agent) logAssistant(msg model.Message) { diff --git a/internal/agent/loop_test.go b/internal/agent/loop_test.go index 23c572c..3ffd0e3 100644 --- a/internal/agent/loop_test.go +++ b/internal/agent/loop_test.go @@ -58,11 +58,12 @@ func TestRunSubmits(t *testing.T) { a := newAgent(t, m) out := filepath.Join(t.TempDir(), "traj.json") saves := 0 - a.OnStep = func(tr *trajectory.Trajectory) { + a.OnStep = func(tr *trajectory.Trajectory) error { saves++ if err := tr.Save(out); err != nil { t.Fatal(err) } + return nil } tr := a.Run(context.Background(), "write hi") diff --git a/internal/agent/resume.go b/internal/agent/resume.go new file mode 100644 index 0000000..6830c51 --- /dev/null +++ b/internal/agent/resume.go @@ -0,0 +1,58 @@ +package agent + +import ( + "crypto/sha256" + "encoding/json" + "errors" + "fmt" + "io" + "runtime" + "strings" + + "github.com/vicgupta/tinyagent/internal/config" + "github.com/vicgupta/tinyagent/internal/tools" + "github.com/vicgupta/tinyagent/internal/trajectory" +) + +// ToolSignature fingerprints the definitions plus an execution contract +// version. Bump the prefix when tool semantics change without a spec change. +func ToolSignature() string { + data, _ := json.Marshal(tools.Specs()) + return fmt.Sprintf("tools-v1:%x", sha256.Sum256(data)) +} + +// ValidateResume runs before provider setup, using only the explicit execution +// root and current config. Checkpoint paths and credentials are never executed +// or loaded from metadata. +func ValidateResume(t *trajectory.Trajectory, cfg config.AgentConfig, root string) error { + if err := t.Validate(); err != nil { + return err + } + if root != t.Root { + return errors.New("--cwd does not match the checkpoint's canonical root") + } + if t.ToolSignature != ToolSignature() { + return errors.New("tool compatibility has changed; start a fresh task") + } + names := toolNames(tools.Specs()) + if len(names) != len(t.Tools) { + return errors.New("tool set has changed; start a fresh task") + } + for i, name := range names { + if t.Tools[i] != name { + return errors.New("tool set has changed; start a fresh task") + } + } + system := strings.NewReplacer("{{cwd}}", root, "{{os}}", runtime.GOOS).Replace(cfg.SystemPrompt) + if !t.SkipAgentsMD { + section, err := loadAgentsMD(root, io.Discard, false) + if err != nil { + return errors.New("cannot reconstruct AGENTS.md instructions; inspect permissions and start a fresh task if instructions changed") + } + system += section + } + if system != t.SystemPrompt { + return errors.New("effective system prompt or AGENTS.md has changed; start a fresh task") + } + return nil +} diff --git a/internal/agent/resume_contract_test.go b/internal/agent/resume_contract_test.go new file mode 100644 index 0000000..9744c96 --- /dev/null +++ b/internal/agent/resume_contract_test.go @@ -0,0 +1,200 @@ +package agent + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "github.com/vicgupta/tinyagent/internal/model" + "github.com/vicgupta/tinyagent/internal/trajectory" +) + +func unfinishedRun(t *testing.T, steps int) (*Agent, *trajectory.Trajectory) { + t.Helper() + replies := make([]model.Message, steps) + for i := range replies { + replies[i] = model.Message{Role: "assistant", ToolCalls: []model.ToolCall{ + call("same", "write_file", `{"path":"edited","content":"historical"}`), + call("read", "read_file", `{"path":"edited"}`), + }} + } + a := newAgent(t, &scripted{replies: replies}) + a.Config.StepLimit = steps + tr := a.Run(context.Background(), "finish the work") + if tr.ExitStatus != StatusStepLimit { + t.Fatalf("setup: %s", tr.ExitStatus) + } + return a, tr +} + +type resumeUsageScript struct{ scripted } + +func (s *resumeUsageScript) Query(ctx context.Context, msgs []model.Message, specs []model.ToolSpec) (model.Message, model.Cost, error) { + msg, cost, err := s.scripted.Query(ctx, msgs, specs) + cost.CacheReadTokens, cost.CacheWriteTokens = 5, 3 + return msg, cost, err +} + +func TestResumeCheckpointContextAndAccounting(t *testing.T) { + a, tr := unfinishedRun(t, 20) + before := append([]model.Message(nil), tr.Messages...) + tr.CacheReadTokens, tr.CacheWriteTokens = 9, 7 + baseCost := tr.Cost + path := filepath.Join(a.Tools.Root, "edited") + if err := os.WriteFile(path, []byte("inspected workspace"), 0o600); err != nil { + t.Fatal(err) + } + m := &resumeUsageScript{scripted{replies: []model.Message{{Role: "assistant", Content: "done"}}}} + a.Model, a.Resume, a.Config.StepLimit = m, tr, 40 + saves := 0 + a.OnStep = func(c *trajectory.Trajectory) error { + saves++ + if saves == 1 && len(m.seen) != 0 { + t.Fatal("queried before initial persistence") + } + return nil + } + got := a.Run(context.Background(), "") + if len(m.seen) != 1 { + t.Fatalf("queries: %d", len(m.seen)) + } + msgs := m.seen[0] + if len(msgs) != len(before)+1 || !reflect.DeepEqual(msgs[:len(before)], before) { + t.Fatal("saved history not sent intact") + } + notice := msgs[len(msgs)-1] + if notice.Role != "user" || !strings.Contains(notice.Content, "workspace") || !strings.Contains(notice.Content, "side effects") { + t.Fatalf("missing recovery notice: %+v", notice) + } + b, _ := os.ReadFile(path) + if string(b) != "inspected workspace" { + t.Fatal("historical tool executed again") + } + if got.Steps != 21 || got.InputTokens != 2100 || got.OutputTokens != 420 || got.Cost != baseCost+0.01 || got.CacheReadTokens != 14 || got.CacheWriteTokens != 10 { + t.Fatalf("accounting changed: %+v", got) + } + if got.ExitStatus != StatusCompleted || saves != 2 { + t.Fatalf("status %s, saves %d", got.ExitStatus, saves) + } +} + +func TestResumeExhaustedCeilingsAndCancellation(t *testing.T) { + for _, which := range []string{"step", "cost", "cancelled"} { + t.Run(which, func(t *testing.T) { + a, tr := unfinishedRun(t, 1) + m := &scripted{} + a.Model, a.Resume = m, tr + a.Config.StepLimit = 0 + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + want := StatusStepLimit + switch which { + case "step": + a.Config.StepLimit = 1 + case "cost": + a.Config.CostLimit, want = tr.Cost, StatusCostLimit + case "cancelled": + cancel() + want = StatusInterrupted + } + got := a.Run(ctx, "") + if len(m.seen) != 0 || got.ExitStatus != want { + t.Fatalf("queries %d, status %s", len(m.seen), got.ExitStatus) + } + }) + } +} + +func TestResumeSaveFailureStopsWork(t *testing.T) { + for _, failAt := range []int{1, 2} { + t.Run(string(rune('0'+failAt)), func(t *testing.T) { + a, tr := unfinishedRun(t, 1) + m := &scripted{replies: []model.Message{ + {Role: "assistant", ToolCalls: []model.ToolCall{call("new", "read_file", `{"path":"edited"}`)}}, + {Role: "assistant", ToolCalls: []model.ToolCall{call("forbidden", "write_file", `{"path":"must-not-exist","content":"bad"}`)}}, + }} + a.Model, a.Resume, a.Config.StepLimit = m, tr, 5 + out := filepath.Join(t.TempDir(), "out.json") + saves := 0 + a.OnStep = func(c *trajectory.Trajectory) error { + saves++ + if saves == failAt { + return errors.New("injected disk failure") + } + return c.Save(out) + } + got := a.Run(context.Background(), "") + if got.ExitStatus != StatusError || len(m.seen) != failAt-1 || saves != failAt { + t.Fatalf("status %s, queries %d, saves %d", got.ExitStatus, len(m.seen), saves) + } + if _, err := os.Stat(filepath.Join(a.Tools.Root, "must-not-exist")); !os.IsNotExist(err) { + t.Fatal("continued after save failure") + } + if failAt == 2 { + prev, err := trajectory.Load(out) + if err != nil || prev.Steps != 1 { + t.Fatalf("last valid checkpoint lost: %v", err) + } + } + }) + } +} + +type resumeDeadlineModel struct{ started time.Time } + +func (m *resumeDeadlineModel) Query(ctx context.Context, _ []model.Message, _ []model.ToolSpec) (model.Message, model.Cost, error) { + m.started = time.Now() + <-ctx.Done() + return model.Message{}, model.Cost{}, ctx.Err() +} + +func TestResumeStartsANewWallTimeBudget(t *testing.T) { + a, tr := unfinishedRun(t, 1) + m := &resumeDeadlineModel{} + a.Model, a.Resume = m, tr + a.Config.StepLimit, a.Config.WallTimeSec = 2, 1 + start := time.Now() + got := a.Run(context.Background(), "") + if m.started.IsZero() || got.ExitStatus != StatusTimeLimit || time.Since(start) < 900*time.Millisecond { + t.Fatalf("new wall-time budget was not used: started=%v, status=%s", m.started, got.ExitStatus) + } +} + +func TestResumeSummaryUsageRetainsCacheAccounting(t *testing.T) { + a, tr := unfinishedRun(t, 1) + a.Model = &resumeUsageScript{scripted{replies: []model.Message{{Role: "assistant", Content: "summary"}}}} + tr.CacheReadTokens, tr.CacheWriteTokens = 9, 7 + if _, err := a.summarize(context.Background(), tr, tr.Task, tr.Messages[2:]); err != nil { + t.Fatal(err) + } + if tr.InputTokens != 200 || tr.OutputTokens != 40 || tr.CacheReadTokens != 14 || tr.CacheWriteTokens != 10 { + t.Fatalf("summary usage not accumulated: %+v", tr) + } +} + +func TestResumeFinalSaveFailureIsAnError(t *testing.T) { + a, tr := unfinishedRun(t, 1) + m := &scripted{replies: []model.Message{{Role: "assistant", Content: "done"}}} + a.Model, a.Resume, a.Config.StepLimit = m, tr, 2 + path := filepath.Join(t.TempDir(), "continued.json") + saves := 0 + a.OnStep = func(c *trajectory.Trajectory) error { + saves++ + if saves == 2 { + return errors.New("injected final save failure") + } + return c.Save(path) + } + got := a.Run(context.Background(), "") + if got.ExitStatus != StatusError || saves != 2 || len(m.seen) != 1 { + t.Fatalf("status %s, saves %d, requests %d", got.ExitStatus, saves, len(m.seen)) + } + if last, err := trajectory.Load(path); err != nil || last.Steps != 1 { + t.Fatalf("last checkpoint was lost: %v", err) + } +} diff --git a/internal/agent/resume_wire_test.go b/internal/agent/resume_wire_test.go new file mode 100644 index 0000000..086dd90 --- /dev/null +++ b/internal/agent/resume_wire_test.go @@ -0,0 +1,140 @@ +package agent + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + + "github.com/vicgupta/tinyagent/internal/model" + "github.com/vicgupta/tinyagent/internal/trajectory" +) + +// Exercise the real wire adapters with a saved, loaded multi-tool history, +// including IDs reused in a later turn. No credentials or live requests. +func TestResumeProviderWireHistory(t *testing.T) { + for _, provider := range []string{"openai", "anthropic", "codex"} { + for _, streamed := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/stream=%v", provider, streamed), func(t *testing.T) { + a, tr := unfinishedRun(t, 2) + p := filepath.Join(t.TempDir(), "checkpoint.json") + if err := tr.Save(p); err != nil { + t.Fatal(err) + } + loaded, err := trajectory.Load(p) + if err != nil { + t.Fatal(err) + } + requests := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + var body map[string]json.RawMessage + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Error(err) + return + } + switch provider { + case "openai": + var msgs []model.Message + if err := json.Unmarshal(body["messages"], &msgs); err != nil { + t.Error(err) + return + } + if len(msgs) != len(tr.Messages)+1 || !reflect.DeepEqual(msgs[:len(tr.Messages)], tr.Messages) || msgs[len(msgs)-1].Content != resumeNotice { + t.Error("OpenAI history or continuation notice changed") + } + if streamed { + io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"done\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n") + } else { + io.WriteString(w, `{"choices":[{"message":{"role":"assistant","content":"done"},"finish_reason":"stop"}]}`) + } + case "anthropic": + var system string + _ = json.Unmarshal(body["system"], &system) + var wire []struct { + Role string + Content []map[string]any + } + if err := json.Unmarshal(body["messages"], &wire); err != nil { + t.Error(err) + return + } + if system != tr.Messages[0].Content || len(wire) != 5 { + t.Error("Anthropic system or turns changed") + return + } + for _, i := range []int{1, 3} { + if wire[i].Role != "assistant" || len(wire[i].Content) != 2 || wire[i].Content[0]["id"] != "same" || wire[i].Content[1]["id"] != "read" { + t.Error("Anthropic tool calls changed") + } + results := wire[i+1].Content + if wire[i+1].Role != "user" || len(results) < 2 || results[0]["tool_use_id"] != "same" || results[0]["content"] != "ok" || results[1]["tool_use_id"] != "read" || !strings.Contains(fmt.Sprint(results[1]["content"]), "historical") { + t.Error("Anthropic results changed") + } + } + last := wire[4].Content + if len(last) != 3 || last[2]["text"] != resumeNotice { + t.Error("Anthropic notice missing") + } + if streamed { + io.WriteString(w, "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"done\"}}\n\ndata: {\"type\":\"message_stop\"}\n\n") + } else { + io.WriteString(w, `{"content":[{"type":"text","text":"done"}],"stop_reason":"end_turn"}`) + } + case "codex": + var instructions string + _ = json.Unmarshal(body["instructions"], &instructions) + var input []map[string]any + if err := json.Unmarshal(body["input"], &input); err != nil { + t.Error(err) + return + } + if instructions != tr.Messages[0].Content || len(input) != 10 { + t.Error("Codex instructions or input changed") + return + } + for _, i := range []int{1, 5} { + if input[i]["type"] != "function_call" || input[i]["call_id"] != "same" || input[i+1]["call_id"] != "read" || input[i+2]["type"] != "function_call_output" || input[i+2]["call_id"] != "same" || input[i+2]["output"] != "ok" || input[i+3]["call_id"] != "read" { + t.Error("Codex calls/results changed") + } + } + content, _ := input[9]["content"].([]any) + if len(content) != 1 || content[0].(map[string]any)["text"] != resumeNotice { + t.Error("Codex notice missing") + } + io.WriteString(w, "data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"message\",\"content\":[{\"type\":\"output_text\",\"text\":\"done\"}]}}\n\ndata: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n") + } + })) + defer srv.Close() + var onText func(string) + if streamed { + onText = func(string) {} + } + switch provider { + case "openai": + a.Model = &model.OpenAI{BaseURL: srv.URL, Model: "gpt-4o", OnText: onText} + case "anthropic": + a.Model = &model.Anthropic{BaseURL: srv.URL, Model: "claude-sonnet-4", OnText: onText} + case "codex": + auth := filepath.Join(t.TempDir(), "auth.json") + if err := os.WriteFile(auth, []byte(`{"tokens":{"access_token":"fake","refresh_token":"fake","account_id":"fake"}}`), 0o600); err != nil { + t.Fatal(err) + } + a.Model = &model.Codex{BaseURL: srv.URL, Model: "gpt-5", AuthFile: auth, OnText: onText} + } + a.Resume, a.Config.StepLimit = loaded, 3 + got := a.Run(context.Background(), "") + if requests != 1 || got.ExitStatus != StatusCompleted { + t.Fatalf("requests %d, status %s: %s", requests, got.ExitStatus, got.Submission) + } + }) + } + } +} diff --git a/internal/tools/process_test.go b/internal/tools/process_test.go index eac618b..7df823f 100644 --- a/internal/tools/process_test.go +++ b/internal/tools/process_test.go @@ -165,12 +165,20 @@ func TestProcessLimit(t *testing.T) { func TestProcessBufferCap(t *testing.T) { e := &Executor{Root: t.TempDir(), ProcessBufferBytes: 100} defer e.Close() - id, _ := startID(t, e, `sleep 0.6; for i in $(seq 1 200); do echo line$i; done; sleep 300`) - var out string + id, _ := startID(t, e, `while [ ! -f emit ]; do sleep 0.01; done; for i in $(seq 1 200); do echo line$i; done; sleep 300`) + // Wait for the entire burst without consuming it. Reading concurrently + // can legitimately return early lines before the buffer reaches its cap. + if err := os.WriteFile(filepath.Join(e.Root, "emit"), nil, 0o600); err != nil { + t.Fatal(err) + } + e.procs.mu.Lock() + p := e.procs.procs[id] + e.procs.mu.Unlock() waitFor(t, func() bool { - out += runTool(t, e, "read_process", map[string]any{"id": id, "timeout": 1000}) - return strings.Contains(out, "line200") + buffered, _, _ := p.out.since(0) + return strings.Contains(buffered, "line200\n") }) + out := runTool(t, e, "read_process", map[string]any{"id": id}) if !strings.Contains(out, "dropped: buffer cap") { t.Fatalf("no drop note in %q", out) } diff --git a/internal/trajectory/checkpoint_contract_test.go b/internal/trajectory/checkpoint_contract_test.go new file mode 100644 index 0000000..e8209db --- /dev/null +++ b/internal/trajectory/checkpoint_contract_test.go @@ -0,0 +1,221 @@ +package trajectory + +import ( + "encoding/json" + "math" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/vicgupta/tinyagent/internal/model" +) + +func checkpoint() *Trajectory { + calls := []model.ToolCall{ + {ID: "a", Type: "function", Function: model.FunctionCall{Name: "list_files", Arguments: `{}`}}, + {ID: "b", Type: "function", Function: model.FunctionCall{Name: "read_file", Arguments: `{"path":"x"}`}}, + } + return &Trajectory{ + Task: "task", Model: "openai/gpt-4o", Steps: 2, Cost: 0.5, InputTokens: 100, + SchemaVersion: 1, Root: func() string { root, _ := filepath.Abs("project"); return root }(), SystemPrompt: "instructions", ToolSignature: "hash", Tools: []string{"list_files", "read_file"}, + ExitStatus: "StepLimitExceeded", + Messages: []model.Message{ + {Role: "system", Content: "instructions"}, {Role: "user", Content: "task"}, + {Role: "assistant", ToolCalls: calls}, + {Role: "tool", ToolCallID: "a", Content: "first"}, + {Role: "tool", ToolCallID: "b", Content: "second"}, + {Role: "assistant", ToolCalls: calls}, // IDs may repeat in another turn. + {Role: "tool", ToolCallID: "a", Content: "third"}, + {Role: "tool", ToolCallID: "b", Content: "fourth"}, + }, + } +} + +func TestCheckpointValidationContract(t *testing.T) { + cases := []struct { + name string + edit func(*Trajectory) + want string + }{ + {"valid repeated IDs", func(*Trajectory) {}, ""}, + {"legacy", func(c *Trajectory) { c.SchemaVersion = 0 }, "metadata"}, + {"version", func(c *Trajectory) { c.SchemaVersion++ }, "version"}, + {"mode", func(c *Trajectory) { c.SystemPrompt = "" }, "metadata"}, + {"completed", func(c *Trajectory) { c.ExitStatus = "Completed" }, "finished"}, + {"submitted", func(c *Trajectory) { c.ExitStatus = "Submitted" }, "finished"}, + {"status", func(c *Trajectory) { c.ExitStatus = "unknown" }, "status"}, + {"negative", func(c *Trajectory) { c.InputTokens = -1 }, "accounting"}, + {"nan", func(c *Trajectory) { c.Cost = math.NaN() }, "accounting"}, + {"infinity", func(c *Trajectory) { c.Cost = math.Inf(1) }, "accounting"}, + {"cache", func(c *Trajectory) { c.CacheReadTokens = 101 }, "accounting"}, + {"steps", func(c *Trajectory) { c.Steps++ }, "steps"}, + {"system", func(c *Trajectory) { c.Messages[0].Role = "user" }, "system"}, + {"task", func(c *Trajectory) { c.Messages[1].Content = "different" }, "task"}, + {"role", func(c *Trajectory) { c.Messages[2].Role = "developer" }, "history"}, + {"missing", func(c *Trajectory) { c.Messages = c.Messages[:7] }, "result"}, + {"duplicate result", func(c *Trajectory) { c.Messages[4].ToolCallID = "a" }, "result"}, + {"out of order", func(c *Trajectory) { c.Messages[3], c.Messages[4] = c.Messages[4], c.Messages[3] }, "result"}, + {"orphan", func(c *Trajectory) { c.Messages = append(c.Messages, c.Messages[7]) }, "history"}, + {"duplicate call", func(c *Trajectory) { c.Messages[2].ToolCalls[1].ID = "a" }, "call"}, + {"arguments", func(c *Trajectory) { c.Messages[2].ToolCalls[0].Function.Arguments = "bad" }, "call"}, + {"unknown call", func(c *Trajectory) { c.Messages[2].ToolCalls[0].Function.Name = "other" }, "call"}, + {"empty call ID", func(c *Trajectory) { c.Messages[2].ToolCalls[0].ID = "" }, "call"}, + {"call type", func(c *Trajectory) { c.Messages[2].ToolCalls[0].Type = "other" }, "call"}, + {"array args", func(c *Trajectory) { c.Messages[2].ToolCalls[0].Function.Arguments = "[]" }, "call"}, + {"null args", func(c *Trajectory) { c.Messages[2].ToolCalls[0].Function.Arguments = "null" }, "call"}, + {"system mismatch", func(c *Trajectory) { c.Messages[0].Content = "changed" }, "system"}, + {"tool with calls", func(c *Trajectory) { c.Messages[3].ToolCalls = c.Messages[2].ToolCalls }, "result"}, + {"terminal assistant", func(c *Trajectory) { c.Messages[5].ToolCalls = nil; c.Messages = c.Messages[:6] }, "history"}, + {"submit marker", func(c *Trajectory) { c.Messages[2].Content = "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT" }, "history"}, + {"extra instruction", func(c *Trajectory) { + c.Messages = append(c.Messages, model.Message{Role: "user", Content: ContinuationNotice}) + }, "history"}, + {"compacted", func(c *Trajectory) { c.Compactions = []Compaction{{Step: 1}} }, "compacted"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c := checkpoint() + tc.edit(c) + err := c.Validate() + if tc.want == "" && err != nil || tc.want != "" && (err == nil || !strings.Contains(err.Error(), tc.want)) { + t.Fatalf("got %v, want %q", err, tc.want) + } + }) + } +} + +func TestCheckpointLoadRefusals(t *testing.T) { + p := filepath.Join(t.TempDir(), "run.json") + for _, data := range []string{"", "null", "{secret transcript", `{}`, `{"resume":{"version":999}}`, `{"model_view":[]}`} { + if err := os.WriteFile(p, []byte(data), 0o600); err != nil { + t.Fatal(err) + } + if _, err := Load(p); err == nil || strings.Contains(err.Error(), "secret transcript") { + t.Fatalf("invalid input: %v", err) + } + } + if _, err := Load(p + ".missing"); err == nil { + t.Fatal("missing accepted") + } + c := checkpoint() + if runtime.GOOS == "windows" { + c.Root = `C:\project` + } + if err := c.Save(p); err != nil { + t.Fatal(err) + } + if _, err := Load(p); err != nil { + t.Fatal(err) + } +} + +func TestCheckpointPrivateAtomicSave(t *testing.T) { + dir := t.TempDir() + p := filepath.Join(dir, "run.json") + victim := filepath.Join(dir, "victim") + if err := os.WriteFile(victim, []byte("untouched"), 0o600); err != nil { + t.Fatal(err) + } + // The former predictable temporary path must never be followed. + _ = os.Symlink(victim, p+".tmp") + c := checkpoint() + if err := c.Save(p); err != nil { + t.Fatal(err) + } + fi, err := os.Stat(p) + if err != nil { + t.Fatal(err) + } + if runtime.GOOS != "windows" && fi.Mode().Perm() != 0o600 { + t.Fatalf("mode %v", fi.Mode()) + } + c.Cost = math.NaN() + if err := c.Save(p); err == nil { + t.Fatal("invalid JSON saved") + } + b, err := os.ReadFile(p) + if err != nil || !json.Valid(b) { + t.Fatal("last valid checkpoint lost") + } + b, _ = os.ReadFile(victim) + if string(b) != "untouched" { + t.Fatal("followed temporary symlink") + } +} + +func TestCheckpointExclusivePublication(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "checkpoint.json") + c := checkpoint() + if err := c.SaveNew(path); err != nil { + t.Fatal(err) + } + before, _ := os.ReadFile(path) + if err := c.SaveNew(path); err == nil { + t.Fatal("overwrote an existing destination") + } + alias := filepath.Join(dir, "alias.json") + if err := os.Symlink(path, alias); err == nil { + if err := c.SaveNew(alias); err == nil { + t.Fatal("overwrote a symlink destination") + } + } + after, _ := os.ReadFile(path) + if string(before) != string(after) { + t.Fatal("exclusive publication changed the prior checkpoint") + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + for _, e := range entries { + if strings.HasPrefix(e.Name(), ".tinyagent-") { + t.Fatal("temporary file leaked") + } + } +} + +func TestCheckpointSizeBoundAndUnknownState(t *testing.T) { + path := filepath.Join(t.TempDir(), "large.json") + f, err := os.Create(path) + if err != nil { + t.Fatal(err) + } + if err := f.Truncate(maxCheckpointBytes + 1); err != nil { + f.Close() + t.Fatal(err) + } + f.Close() + if _, err := Load(path); err == nil || !strings.Contains(err.Error(), "64 MiB") { + t.Fatalf("size bound: %v", err) + } + c := checkpoint() + data, _ := json.Marshal(c) + data = append(data[:len(data)-1], []byte(`,"model_view":[]}`)...) + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + if _, err := Load(path); err == nil || !strings.Contains(err.Error(), "unsupported checkpoint fields") { + t.Fatalf("unknown state: %v", err) + } +} + +func TestCheckpointContinuationProvenance(t *testing.T) { + c := checkpoint() + c.Resume = &ResumeInfo{Source: "first.json", ExitStatus: "Error", Steps: 2, Message: len(c.Messages)} + c.Messages = append(c.Messages, model.Message{Role: "user", Content: ContinuationNotice}) + if err := c.Validate(); err != nil { + t.Fatal(err) + } + c.Resume = &ResumeInfo{Source: "second.json", ExitStatus: "Interrupted", Steps: 2, Message: len(c.Messages), Previous: c.Resume} + c.Messages = append(c.Messages, model.Message{Role: "user", Content: ContinuationNotice}) + if err := c.Validate(); err != nil { + t.Fatal(err) + } + c.Resume.Previous.Message = 3 + if err := c.Validate(); err == nil { + t.Fatal("accepted provenance inside a tool turn") + } +} diff --git a/internal/trajectory/resume.go b/internal/trajectory/resume.go new file mode 100644 index 0000000..93dedca --- /dev/null +++ b/internal/trajectory/resume.go @@ -0,0 +1,187 @@ +package trajectory + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "math" + "os" + "path/filepath" + "strings" +) + +// maxCheckpointBytes bounds parsing and memory use. Larger trajectories remain +// audit logs but cannot be resumed by this format. +const maxCheckpointBytes = 64 << 20 // 64 MiB + +// ContinuationNotice is part of the versioned checkpoint contract. It is a +// user message at each recorded resume boundary, never a new system prompt. +const ContinuationNotice = "The previous invocation stopped before this task finished. " + + "The conversation above is saved history up to the last checkpoint. " + + "The workspace may have changed since that save, and commands may have produced " + + "side effects that are not reflected in the transcript. Inspect and revalidate " + + "the current workspace before continuing. Live processes and shell state are not " + + "restored; historical process IDs cannot be used to reattach to old processes." + +// Load reads a trusted local checkpoint without modifying it. Parse diagnostics +// intentionally omit decoder errors, which can contain transcript values. +func Load(path string) (*Trajectory, error) { + fi, err := os.Stat(path) + if err != nil { + return nil, fmt.Errorf("reading checkpoint: %w", err) + } + if !fi.Mode().IsRegular() { + return nil, errors.New("checkpoint must be a regular file") + } + if fi.Size() > maxCheckpointBytes { + return nil, errors.New("checkpoint exceeds the 64 MiB resume size limit") + } + f, err := os.Open(path) + if err != nil { + return nil, fmt.Errorf("reading checkpoint: %w", err) + } + defer f.Close() + fi, err = f.Stat() + if err != nil { + return nil, fmt.Errorf("reading checkpoint: %w", err) + } + if !fi.Mode().IsRegular() { + return nil, errors.New("checkpoint must be a regular file") + } + data, err := io.ReadAll(io.LimitReader(f, maxCheckpointBytes+1)) + if err != nil { + return nil, fmt.Errorf("reading checkpoint: %w", err) + } + if len(data) > maxCheckpointBytes { + return nil, errors.New("checkpoint exceeds the 64 MiB resume size limit") + } + if len(bytes.TrimSpace(data)) == 0 { + return nil, errors.New("checkpoint is empty") + } + var t *Trajectory + dec := json.NewDecoder(bytes.NewReader(data)) + dec.DisallowUnknownFields() + if err := dec.Decode(&t); err != nil || t == nil { + return nil, errors.New("invalid checkpoint JSON or unsupported checkpoint fields") + } + if err := dec.Decode(new(any)); err != io.EOF { + return nil, errors.New("invalid checkpoint JSON: trailing data") + } + if t.SchemaVersion != SchemaVersion { + return nil, t.Validate() + } + // A false loading mode must be explicit, rather than guessed from a + // missing field. All new files write this field even when it is false. + var fields map[string]json.RawMessage + _ = json.Unmarshal(data, &fields) + if _, ok := fields["skip_agents_md"]; !ok || bytes.Equal(fields["skip_agents_md"], []byte("null")) { + return nil, errors.New("checkpoint is missing AGENTS.md loading metadata; start a fresh task") + } + if err := t.Validate(); err != nil { + return nil, err + } + return t, nil +} + +func unfinished(status string) bool { + switch status { + case "", "StepLimitExceeded", "CostLimitExceeded", "TimeLimitExceeded", "Interrupted", "Error": + return true + } + return false +} + +// Validate checks a complete model-query boundary, matching IDs within each +// assistant turn. It never repairs or dispatches historical calls. +func (t *Trajectory) Validate() error { + if t.SchemaVersion == 0 { + return errors.New("checkpoint has no resume metadata (it predates --resume); start a fresh task") + } + if t.SchemaVersion != SchemaVersion { + return fmt.Errorf("checkpoint schema version %d is unsupported (want %d)", t.SchemaVersion, SchemaVersion) + } + if !filepath.IsAbs(t.Root) || filepath.Clean(t.Root) != t.Root || t.SystemPrompt == "" || len(t.Tools) == 0 || t.ToolSignature == "" { + return errors.New("checkpoint is missing required root, instruction, or tool compatibility metadata") + } + if t.Task == "" || t.Model == "" { + return errors.New("checkpoint is missing its task or model") + } + if t.ExitStatus == "Submitted" || t.ExitStatus == "Completed" { + return errors.New("checkpoint already finished; start a fresh task") + } + if !unfinished(t.ExitStatus) { + return errors.New("checkpoint has an unsupported exit status") + } + if t.Steps < 0 || math.IsNaN(t.Cost) || math.IsInf(t.Cost, 0) || t.Cost < 0 || t.InputTokens < 0 || t.OutputTokens < 0 || t.CacheReadTokens < 0 || t.CacheWriteTokens < 0 || t.CacheReadTokens > t.InputTokens || t.CacheWriteTokens > t.InputTokens-t.CacheReadTokens { + return errors.New("checkpoint has invalid accounting counters") + } + if len(t.Compactions) > 0 { + return errors.New("checkpoint was compacted; resuming compacted model views is not yet supported; start a fresh task") + } + if len(t.Messages) < 2 || t.Messages[0].Role != "system" || t.Messages[0].Content != t.SystemPrompt || len(t.Messages[0].ToolCalls) != 0 || t.Messages[0].ToolCallID != "" { + return errors.New("checkpoint has an invalid system prompt") + } + if t.Messages[1].Role != "user" || t.Messages[1].Content != t.Task || len(t.Messages[1].ToolCalls) != 0 || t.Messages[1].ToolCallID != "" { + return errors.New("checkpoint has an invalid task message") + } + boundaries := map[int]*ResumeInfo{} + end, steps := len(t.Messages), t.Steps + for b := t.Resume; b != nil; b = b.Previous { + if b.Message < 2 || b.Message >= end || b.Steps < 0 || b.Steps > steps || !unfinished(b.ExitStatus) || b.Source == "" { + return errors.New("checkpoint has invalid resume provenance") + } + boundaries[b.Message] = b + end, steps = b.Message, b.Steps + } + known := map[string]bool{} + for _, name := range t.Tools { + if name == "" || known[name] { + return errors.New("checkpoint has invalid tool metadata") + } + known[name] = true + } + steps = 0 + for i := 2; i < len(t.Messages); { + m := t.Messages[i] + if b, ok := boundaries[i]; ok { + if b.Steps != steps || m.Role != "user" || m.Content != ContinuationNotice || len(m.ToolCalls) != 0 || m.ToolCallID != "" { + return errors.New("checkpoint has an invalid continuation notice") + } + delete(boundaries, i) + i++ + continue + } + if m.Role != "assistant" || m.ToolCallID != "" || len(m.ToolCalls) == 0 || strings.Contains(m.Content, "COMPLETE_TASK_AND_SUBMIT_FINAL_OUTPUT") { + return errors.New("checkpoint history is not an unfinished assistant/tool boundary") + } + ids := map[string]bool{} + for _, c := range m.ToolCalls { + var args map[string]json.RawMessage + if c.ID == "" || ids[c.ID] || c.Type != "function" || !known[c.Function.Name] || json.Unmarshal([]byte(c.Function.Arguments), &args) != nil || args == nil { + return errors.New("checkpoint has an invalid or duplicate tool call") + } + ids[c.ID] = true + } + i++ + for _, c := range m.ToolCalls { + if i >= len(t.Messages) { + return errors.New("checkpoint ends mid-turn: missing tool result") + } + r := t.Messages[i] + if r.Role != "tool" || r.ToolCallID != c.ID || len(r.ToolCalls) != 0 { + return errors.New("checkpoint has a missing, duplicate, orphaned, or out-of-order tool result") + } + i++ + } + steps++ + } + if len(boundaries) != 0 { + return errors.New("checkpoint resume boundary is inside a tool turn") + } + if steps != t.Steps { + return errors.New("checkpoint steps do not match the recorded history") + } + return nil +} diff --git a/internal/trajectory/trajectory.go b/internal/trajectory/trajectory.go index 69ac1d1..dd9de13 100644 --- a/internal/trajectory/trajectory.go +++ b/internal/trajectory/trajectory.go @@ -28,8 +28,36 @@ type Trajectory struct { // BackgroundProcesses summarizes background processes still registered // when the run ended (all are killed): status and output tail of each. BackgroundProcesses string `json:"background_processes,omitempty"` + + // Resume metadata, present on trajectories written by this version. It + // lets a later invocation validate and continue an unfinished run. + SchemaVersion int `json:"schema_version,omitempty"` + Root string `json:"root,omitempty"` // canonical working directory + // SystemPrompt is the effective system prompt (with {{cwd}}/{{os}} + // substituted and AGENTS.md appended) the run used. + SystemPrompt string `json:"system_prompt,omitempty"` + // SkipAgentsMD records whether AGENTS.md was excluded from the prompt. + SkipAgentsMD bool `json:"skip_agents_md"` + Tools []string `json:"tools,omitempty"` // tool names, in spec order + ToolSignature string `json:"tool_signature,omitempty"` // versioned spec/semantics fingerprint + // Resume records the checkpoint this run continued, when it did. + Resume *ResumeInfo `json:"resume,omitempty"` +} + +// ResumeInfo records the checkpoint a run continued, so the previous +// termination stays distinguishable from the active result. +type ResumeInfo struct { + Source string `json:"source"` // path of the input checkpoint + ExitStatus string `json:"exit_status"` // previous termination status + Steps int `json:"steps"` // steps at the previous termination + Submission string `json:"submission,omitempty"` + Message int `json:"message"` // continuation notice index in Messages + Previous *ResumeInfo `json:"previous,omitempty"` } +// SchemaVersion is the resume-metadata version written by this build. +const SchemaVersion = 1 + // Compaction describes one compaction of the model's view of the history. // Indexes refer to Trajectory.Messages. type Compaction struct { @@ -49,9 +77,20 @@ type Compaction struct { View []int `json:"view"` } -// Save writes the trajectory to path atomically, so a crash mid-write never -// leaves a truncated file. +// Save replaces path with a complete private file using a same-directory +// temporary file and rename. This does not guarantee power-loss durability. func (t *Trajectory) Save(path string) error { + return t.save(path, false) +} + +// SaveNew publishes the starting checkpoint without replacing any existing +// destination, including symlinks. Linking the completed temporary file makes +// publication atomic and exclusive on filesystems supporting hard links. +func (t *Trajectory) SaveNew(path string) error { + return t.save(path, true) +} + +func (t *Trajectory) save(path string, exclusive bool) error { data, err := json.MarshalIndent(t, "", " ") if err != nil { return err @@ -59,9 +98,20 @@ func (t *Trajectory) Save(path string) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return err } - tmp := path + ".tmp" - if err := os.WriteFile(tmp, data, 0o644); err != nil { + tmp, err := os.CreateTemp(filepath.Dir(path), ".tinyagent-*.json") + if err != nil { + return err + } + defer os.Remove(tmp.Name()) + if _, err := tmp.Write(data); err != nil { + tmp.Close() return err } - return os.Rename(tmp, path) + if err := tmp.Close(); err != nil { + return err + } + if exclusive { + return os.Link(tmp.Name(), path) + } + return os.Rename(tmp.Name(), path) }