diff --git a/server/pkg/agent/omp_session.go b/server/pkg/agent/omp_session.go new file mode 100644 index 00000000000..dacc9bbd38d --- /dev/null +++ b/server/pkg/agent/omp_session.go @@ -0,0 +1,63 @@ +package agent + +import ( + "fmt" + "log/slog" + "os" + "path/filepath" + "strings" +) + +// OMP aliases --session to --resume. All session selection and persistence +// flags must remain daemon-owned, including on fresh runs using --session-dir. +var ompSessionArgs = map[string]blockedArgMode{ + "--session": blockedOptionalValue, + "--session-dir": blockedWithValue, + "--no-session": blockedStandalone, + "--continue": blockedStandalone, + "-c": blockedStandalone, + "--resume": blockedOptionalValue, + "-r": blockedOptionalValue, + "--fork": blockedWithValue, + "--session-id": blockedWithValue, +} + +func buildOmpArgs(sessionPath, sessionDir string, opts ExecOptions, logger *slog.Logger) []string { + opts.CustomArgs = filterCustomArgs(opts.CustomArgs, ompSessionArgs, logger) + args := buildPiArgs(sessionPath, opts, logger) + if sessionDir != "" { + args = append(args, "--session-dir", sessionDir) + } + return args +} + +// Each fresh execution owns a private directory. OMP chooses the filename; +// discover it after exit instead of depending on its timestamp/ID naming +// convention or moving a transcript away from its companion artifacts. +func findOmpSessionFile(dir string) (string, error) { + entries, err := os.ReadDir(dir) + if err != nil { + return "", err + } + var path string + for _, entry := range entries { + if !entry.Type().IsRegular() || !strings.HasSuffix(entry.Name(), ".jsonl") { + continue + } + info, err := entry.Info() + if err != nil { + return "", err + } + if info.Size() == 0 { + continue + } + if path != "" { + return "", fmt.Errorf("multiple transcripts in fresh session directory %q", dir) + } + path = filepath.Join(dir, entry.Name()) + } + if path == "" { + return "", fmt.Errorf("no persisted transcript in fresh session directory %q", dir) + } + return path, nil +} diff --git a/server/pkg/agent/omp_session_integration_test.go b/server/pkg/agent/omp_session_integration_test.go new file mode 100644 index 00000000000..cb21198c3d0 --- /dev/null +++ b/server/pkg/agent/omp_session_integration_test.go @@ -0,0 +1,86 @@ +//go:build agentintegration + +package agent + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" +) + +// TestOmpSessionStartupWithoutModelCall checks the real CLI's session contract +// without sending a prompt. Use an explicitly selected OMP binary and an empty +// home/config directory; no user account or provider credential is consulted. +func TestOmpSessionStartupWithoutModelCall(t *testing.T) { + if os.Getenv("MULTICA_RUN_REAL_AGENT_SMOKE") != "1" { + t.Skip("set MULTICA_RUN_REAL_AGENT_SMOKE=1 to run real-agent smoke tests") + } + binary := os.Getenv("MULTICA_OMP_SMOKE_EXECUTABLE") + if binary == "" { + t.Skip("set MULTICA_OMP_SMOKE_EXECUTABLE to the OMP binary to verify") + } + home := t.TempDir() + cwd := t.TempDir() + env := []string{ + "PATH=" + os.Getenv("PATH"), "HOME=" + home, + "USERPROFILE=" + home, "TMPDIR=" + t.TempDir(), + "PI_CODING_AGENT_DIR=" + filepath.Join(home, "agent"), + // Select a catalog model without requiring a stored account. With no + // prompt, print mode never enters session.prompt or calls the provider. + "ANTHROPIC_API_KEY=unused-no-model-call", + } + opts := ExecOptions{Model: "anthropic/claude-sonnet-4-20250514", CustomArgs: []string{"--no-extensions", "--no-skills", "--no-title"}} + run := func(args []string) (string, string, error) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 45*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, binary, args...) + cmd.Dir, cmd.Env = cwd, env + cmd.Stdin = strings.NewReader("") + var stdout, stderr bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdout, &stderr + err := cmd.Run() + return stdout.String(), stderr.String(), err + } + empty := filepath.Join(cwd, "empty.jsonl") + if err := os.WriteFile(empty, nil, 0o600); err != nil { + t.Fatal(err) + } + _, stderr, err := run(buildOmpArgs(empty, "", opts, slog.Default())) + if err == nil || !strings.Contains(stderr, "session file holds no entries") { + t.Fatalf("expected OMP's empty-resume refusal: %v, %s", err, stderr) + } + stdout, stderr, err := run(buildOmpArgs("", t.TempDir(), opts, slog.Default())) + if err != nil { + t.Fatalf("fresh startup: %v, %s", err, stderr) + } + var header string + for _, line := range strings.Split(stdout, "\n") { + var event struct { + Type string `json:"type"` + } + if json.Unmarshal([]byte(line), &event) == nil && event.Type == "session" { + header = line + break + } + } + if header == "" { + t.Fatalf("fresh startup did not emit a session header: %s", stdout) + } + // A prompt-free invocation may discard its draft. Persist its native header + // explicitly to verify path-based resume without generating a model turn. + valid := filepath.Join(cwd, "valid.jsonl") + if err := os.WriteFile(valid, []byte(header+"\n"), 0o600); err != nil { + t.Fatal(err) + } + if _, stderr, err := run(buildOmpArgs(valid, "", opts, slog.Default())); err != nil { + t.Fatalf("valid resume startup: %v, %s", err, stderr) + } +} diff --git a/server/pkg/agent/omp_session_test.go b/server/pkg/agent/omp_session_test.go new file mode 100644 index 00000000000..c081e17c204 --- /dev/null +++ b/server/pkg/agent/omp_session_test.go @@ -0,0 +1,60 @@ +package agent + +import ( + "log/slog" + "os" + "path/filepath" + "slices" + "testing" +) + +func TestBuildOmpArgsOwnsSessionSelection(t *testing.T) { + for _, resume := range []string{"", "/saved.jsonl"} { + dir := "" + want := []string{"-p", "--mode", "json", "--tools", "read"} + if resume == "" { + dir = "/fresh" + want = append(want, "--session-dir", dir) + } else { + want = []string{"-p", "--mode", "json", "--session", resume, "--tools", "read"} + } + args := buildOmpArgs(resume, dir, ExecOptions{CustomArgs: []string{ + "--session-dir", "/other", "--session-dir=/other", "--no-session", + "--session", "--continue", "-c", "--resume=other", "-r", "other", + "--fork", "other", "--session-id", "other", "--tools", "read", + }}, slog.Default()) + if !slices.Equal(args, want) { + t.Fatalf("resume %q: args = %v, want %v", resume, args, want) + } + } +} + +func TestFindOmpSessionFile(t *testing.T) { + for _, tc := range []struct { + name string + files map[string]string + want string + }{ + {"missing", nil, ""}, + {"empty", map[string]string{"empty.jsonl": ""}, ""}, + {"persisted", map[string]string{"chosen.jsonl": "{}\n", "metadata.json": "{}", "empty.jsonl": ""}, "chosen.jsonl"}, + {"ambiguous", map[string]string{"a.jsonl": "{}\n", "b.jsonl": "{}\n"}, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + for name, data := range tc.files { + if err := os.WriteFile(filepath.Join(dir, name), []byte(data), 0o600); err != nil { + t.Fatal(err) + } + } + path, err := findOmpSessionFile(dir) + if tc.want == "" { + if err == nil || path != "" { + t.Fatalf("expected no resumable path, got %q, %v", path, err) + } + } else if err != nil || path != filepath.Join(dir, tc.want) { + t.Fatalf("got %q, %v", path, err) + } + }) + } +} diff --git a/server/pkg/agent/omp_session_unix_test.go b/server/pkg/agent/omp_session_unix_test.go new file mode 100644 index 00000000000..c2848820267 --- /dev/null +++ b/server/pkg/agent/omp_session_unix_test.go @@ -0,0 +1,106 @@ +//go:build unix + +package agent + +import ( + "log/slog" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestOmpFreshSessionThenResume(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + // Model-free reproduction of OMP 18.8.3's contract: --session rejects + // missing/empty files; --session-dir creates and persists a new transcript. + fake := filepath.Join(t.TempDir(), "omp") + writeTestExecutable(t, fake, []byte(`#!/bin/sh +while [ "$#" -gt 0 ]; do + case "$1" in + --session) session="$2"; shift ;; + --session-dir) dir="$2"; shift ;; + esac + shift +done +if [ -n "$session" ]; then + [ -s "$session" ] || exit 1 +else + [ -d "$dir" ] || exit 2 + session="$dir/runtime-chosen.jsonl" + printf '%s\n' '{"type":"session","id":"test"}' > "$session" +fi +cat >> "$session" +printf '\n' >> "$session" +printf '%s\n' '{"type":"agent_end"}' +`)) + backend, err := ResolveBackend("omp", Config{ExecutablePath: fake, Logger: slog.Default()}) + if err != nil { + t.Fatal(err) + } + execute := func(prompt, resume string) Result { + t.Helper() + session, err := backend.Execute(t.Context(), prompt, ExecOptions{ResumeSessionID: resume, Timeout: 5 * time.Second}) + if err != nil { + t.Fatal(err) + } + for range session.Messages { + } + return <-session.Result + } + var previous string + for _, prompt := range []string{"fresh chat", "quick-create task"} { + first := execute(prompt, "") + if first.Status != "completed" || first.SessionID == "" || first.SessionID == previous { + t.Fatalf("fresh result: %+v", first) + } + previous = first.SessionID + second := execute("follow-up", first.SessionID) + if second.Status != "completed" || second.SessionID != first.SessionID { + t.Fatalf("resume result: %+v", second) + } + data, err := os.ReadFile(second.SessionID) + if err != nil || !strings.Contains(string(data), prompt+"\nfollow-up\n") { + t.Fatalf("history not preserved: %q, %v", data, err) + } + } + // Old zero-byte transcripts must not be fabricated into valid resumes. + empty := filepath.Join(t.TempDir(), "empty.jsonl") + if err := os.WriteFile(empty, nil, 0o600); err != nil { + t.Fatal(err) + } + if result := execute("follow-up", empty); result.Status != "failed" { + t.Fatalf("empty resume unexpectedly succeeded: %+v", result) + } + if info, err := os.Stat(empty); err != nil || info.Size() != 0 { + t.Fatalf("empty resume was modified: %v, %v", info, err) + } + missing := filepath.Join(t.TempDir(), "missing.jsonl") + if _, err := backend.Execute(t.Context(), "follow-up", ExecOptions{ResumeSessionID: missing}); err == nil { + t.Fatal("missing resume unexpectedly started") + } + if _, err := os.Stat(missing); !os.IsNotExist(err) { + t.Fatalf("missing resume was created: %v", err) + } +} + +func TestOmpFreshSessionRequiresPersistedTranscript(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + fake := filepath.Join(t.TempDir(), "omp") + writeTestExecutable(t, fake, []byte("#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' '{\"type\":\"agent_end\"}'\n")) + backend, err := ResolveBackend("omp", Config{ExecutablePath: fake, Logger: slog.Default()}) + if err != nil { + t.Fatal(err) + } + session, err := backend.Execute(t.Context(), "prompt", ExecOptions{Timeout: 5 * time.Second}) + if err != nil { + t.Fatal(err) + } + for range session.Messages { + } + result := <-session.Result + if result.Status != "failed" || result.SessionID != "" || !strings.Contains(result.Error, "no persisted transcript") { + t.Fatalf("must not publish a fabricated resume path: %+v", result) + } +} diff --git a/server/pkg/agent/omp_test.go b/server/pkg/agent/omp_test.go index 48327b734e9..b31af38a8e0 100644 --- a/server/pkg/agent/omp_test.go +++ b/server/pkg/agent/omp_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "log/slog" + "os" "path/filepath" "runtime" "slices" @@ -110,6 +111,9 @@ func TestOmpExecuteDefaultsToOmpBinary(t *testing.T) { t.Fatalf("New(omp): %v", err) } sessionPath := filepath.Join(t.TempDir(), "session.jsonl") + if err := os.WriteFile(sessionPath, []byte("{}\n"), 0o600); err != nil { + t.Fatal(err) + } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() session, err := backend.Execute(ctx, "test prompt", ExecOptions{ @@ -201,7 +205,11 @@ func TestOmpExecuteCompletesFromEventStream(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - session, err := backend.Execute(ctx, "prompt-ignored", ExecOptions{Timeout: 5 * time.Second}) + sessionPath := filepath.Join(t.TempDir(), "session.jsonl") + if err := os.WriteFile(sessionPath, []byte("{}\n"), 0o600); err != nil { + t.Fatal(err) + } + session, err := backend.Execute(ctx, "prompt-ignored", ExecOptions{Timeout: 5 * time.Second, ResumeSessionID: sessionPath}) if err != nil { t.Fatalf("execute: %v", err) } diff --git a/server/pkg/agent/pi.go b/server/pkg/agent/pi.go index 2ec0e0a3197..4c00924d330 100644 --- a/server/pkg/agent/pi.go +++ b/server/pkg/agent/pi.go @@ -373,35 +373,58 @@ func (b *piBackend) Execute(ctx context.Context, prompt string, opts ExecOptions timeout := opts.Timeout - // Pi's --session flag expects a file path where events are appended. - // The path doubles as our opaque session identifier: we return it as - // SessionID and expect it back as ResumeSessionID on the next turn. + // The persisted JSONL path is our opaque session identifier. OMP's + // --session is strictly resume-only; let it create fresh sessions in a + // private directory, then return the transcript it actually persisted. sessionPath := opts.ResumeSessionID - if sessionPath == "" { - p, err := newPiSessionPath() + var sessionDir string + var sessionLock *os.File + if label == "omp" && sessionPath == "" { + dir, err := piSessionDir() if err != nil { - return nil, fmt.Errorf("%s session path: %w", label, err) + return nil, fmt.Errorf("%s session directory: %w", label, err) } - sessionPath = p - } - if err := ensurePiSessionFile(sessionPath); err != nil { - return nil, fmt.Errorf("%s session file: %w", label, err) - } - sessionLock, locked, err := tryLockPiSessionFile(sessionPath) - if err != nil { - return nil, fmt.Errorf("%s session lock: %w", label, err) - } - if !locked { - if opts.ResumeSessionID != "" { - return piSessionBusyResult(label, sessionPath), nil + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("%s session directory: %w", label, err) + } + sessionDir, err = os.MkdirTemp(dir, "omp-") + if err != nil { + return nil, fmt.Errorf("%s session directory: %w", label, err) + } + } else { + if sessionPath == "" { + sessionPath, err = newPiSessionPath() + if err != nil { + return nil, fmt.Errorf("%s session path: %w", label, err) + } + } + if label != "omp" { + if err := ensurePiSessionFile(sessionPath); err != nil { + return nil, fmt.Errorf("%s session file: %w", label, err) + } + } + var locked bool + sessionLock, locked, err = tryLockPiSessionFile(sessionPath) + if err != nil { + return nil, fmt.Errorf("%s session lock: %w", label, err) + } + if !locked { + if opts.ResumeSessionID != "" { + return piSessionBusyResult(label, sessionPath), nil + } + return nil, fmt.Errorf("%s session file %q is already in use", label, sessionPath) } - return nil, fmt.Errorf("%s session file %q is already in use", label, sessionPath) } runCtx, cancel := runContext(ctx, timeout) processCtx, cancelProcess := context.WithCancel(runCtx) - args := buildPiArgs(sessionPath, opts, b.cfg.Logger) + var args []string + if label == "omp" { + args = buildOmpArgs(sessionPath, sessionDir, opts, b.cfg.Logger) + } else { + args = buildPiArgs(sessionPath, opts, b.cfg.Logger) + } cmd, _, _ := b.cfg.commandAt(execName).execVia(processCtx, choosePiInvocation, lookedUp, args, b.cfg.Logger) hideAgentWindow(cmd) b.cfg.logAgentCommand(cmd, newAgentCommandLogArgs(args)) @@ -749,6 +772,15 @@ func (b *piBackend) Execute(ctx context.Context, prompt string, opts ExecOptions authoritativeTerminal = true } + if sessionDir != "" { + var sessionErr error + sessionPath, sessionErr = findOmpSessionFile(sessionDir) + if sessionErr != nil && finalStatus == "completed" { + finalStatus = "failed" + finalError = fmt.Sprintf("omp session file: %v", sessionErr) + } + } + b.cfg.Logger.Info(label+" finished", "pid", cmd.Process.Pid, "status", finalStatus, "duration", duration.Round(time.Millisecond).String()) // Publish the authoritative terminal boundary before Result. The daemon // uses this ordering to let an already-observed provider failure outrank