diff --git a/Makefile b/Makefile index dcf43bd..1f258b8 100644 --- a/Makefile +++ b/Makefile @@ -20,20 +20,26 @@ LDFLAGS := -s -w -X main.version=$(VERSION) IMAGE := $(REGISTRY):sandbox-$(VERSION) -.PHONY: all cli \ +.PHONY: all cli orchestrator \ vet lint test test-local test-kind test-remote test-all \ dev-sandbox dev-push tag clean help ## ── CLI ────────────────────────────────────────────────────────────── -## Build CLI + sandbox image for local dev -all: cli dev-sandbox +## Build CLI + orchestrator + sandbox image for local dev +all: cli orchestrator dev-sandbox ## Build the harness CLI binary cli: CGO_ENABLED=0 go build -ldflags '$(LDFLAGS)' -o harness . @echo "Built: ./harness ($(VERSION))" +## Build the in-sandbox orchestrator binary (linux/amd64) +orchestrator: + CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags '$(LDFLAGS)' \ + -o bin/harness-orchestrator ./cmd/orchestrator + @echo "Built: bin/harness-orchestrator (linux/amd64, $(VERSION))" + ## ── Lint targets ───────────────────────────────────────────────────── ## Run go vet @@ -90,8 +96,11 @@ test-all: test test-local test-kind test-remote ## ── Dev image builds ───────────────────────────────────────────────── ## Build dev sandbox image locally (native arch only) -dev-sandbox: +## Copies the orchestrator binary into the Docker context if available. +dev-sandbox: orchestrator + cp bin/harness-orchestrator profiles/images/sandbox-default/harness-orchestrator 2>/dev/null || true $(CONTAINER_CLI) build -t $(IMAGE) profiles/images/sandbox-default/ + rm -f profiles/images/sandbox-default/harness-orchestrator @echo "Built: $(IMAGE)" ## Build and push dev sandbox image (multi-arch) @@ -111,7 +120,7 @@ tag: ## Clean built binaries clean: - rm -f harness + rm -f harness bin/harness-orchestrator @echo "Cleaned binaries" ## Show available targets diff --git a/cmd/apply.go b/cmd/apply.go index 8a8ce4d..7c15af4 100644 --- a/cmd/apply.go +++ b/cmd/apply.go @@ -22,6 +22,8 @@ func NewApplyCmd(harnessDir, cli string) *cobra.Command { sandboxName string task string entrypoint string + mode string + watch bool attach bool providerRefresh bool dryRun bool @@ -50,6 +52,12 @@ then deploy a sandbox. Use --dry-run to validate without deploying, or agentPath := resolveAgentPath(harnessDir, agentName, file) // CLI overrides + if watch { + mode = "watch" + } + if mode != "" { + agentCfg.Mode = mode + } if entrypoint != "" { agentCfg.Entrypoint = entrypoint } @@ -143,6 +151,8 @@ then deploy a sandbox. Use --dry-run to validate without deploying, or cmd.Flags().StringVar(&sandboxName, "name", "", "Sandbox name (overrides agent config)") cmd.Flags().StringVar(&task, "task", "", "Task to pass to the agent (inline text or @filepath)") cmd.Flags().StringVar(&entrypoint, "entrypoint", "", "Override agent entrypoint (claude, opencode, bash)") + cmd.Flags().StringVar(&mode, "mode", "", "Orchestrator mode: once or watch") + cmd.Flags().BoolVar(&watch, "watch", false, "Shorthand for --mode=watch") cmd.Flags().BoolVar(&attach, "attach", false, "Attach TTY after creation (interactive mode)") cmd.Flags().BoolVar(&providerRefresh, "provider-refresh", false, "Delete and recreate all providers") cmd.Flags().BoolVar(&dryRun, "dry-run", false, "Validate configuration without deploying") diff --git a/cmd/executor.go b/cmd/executor.go index 5c8d265..a63f79e 100644 --- a/cmd/executor.go +++ b/cmd/executor.go @@ -121,7 +121,10 @@ func upLocal(opts upLocalOpts) error { if noTTY && agentCfg.Task == "" { sandboxCmd = []string{"true"} } else { - sandboxCmd = []string{"bash", "/sandbox/.config/openshell/run.sh"} + sandboxCmd = []string{ + "sh", "-c", + `if command -v harness-orchestrator >/dev/null 2>&1; then exec harness-orchestrator --config /sandbox/.config/openshell/orchestrator.yaml; else exec bash /sandbox/.config/openshell/run.sh; fi`, + } } err = createSandbox(sandboxOpts{ diff --git a/cmd/orchestrator/main.go b/cmd/orchestrator/main.go new file mode 100644 index 0000000..ec90463 --- /dev/null +++ b/cmd/orchestrator/main.go @@ -0,0 +1,51 @@ +package main + +import ( + "context" + "flag" + "fmt" + "os" + "os/signal" + "path/filepath" + "syscall" + + "github.com/stackrox/harness-openshell/internal/orchestrator" +) + +var version = "dev" + +func main() { + configPath := flag.String("config", "/sandbox/.config/openshell/orchestrator.yaml", "path to orchestrator config") + showVersion := flag.Bool("version", false, "print version and exit") + flag.Parse() + + if *showVersion { + fmt.Println("harness-orchestrator", version) + os.Exit(0) + } + + cfg, err := orchestrator.LoadConfig(*configPath) + if err != nil { + fmt.Fprintf(os.Stderr, "harness-orchestrator: %v\n", err) + os.Exit(1) + } + + configDir := filepath.Dir(*configPath) + if cfg.Task != "" && !filepath.IsAbs(cfg.Task) { + cfg.Task = filepath.Join(configDir, cfg.Task) + } + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + orch, err := orchestrator.New(cfg, configDir) + if err != nil { + fmt.Fprintf(os.Stderr, "harness-orchestrator: %v\n", err) + os.Exit(1) + } + + if err := orch.Run(ctx); err != nil { + fmt.Fprintf(os.Stderr, "harness-orchestrator: %v\n", err) + os.Exit(1) + } +} diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 579985b..660b4ba 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -37,6 +37,15 @@ type AgentConfig struct { Image string `yaml:"image,omitempty"` Include []string `yaml:"include,omitempty"` Payloads []PayloadEntry `yaml:"payloads,omitempty"` + + // Orchestrator fields — projected into orchestrator.yaml for in-sandbox use. + Mode string `yaml:"mode,omitempty"` + Sentinel *bool `yaml:"sentinel,omitempty"` + PollInterval int `yaml:"poll_interval,omitempty"` + MaxFailures int `yaml:"max_failures,omitempty"` + Heartbeat int `yaml:"heartbeat,omitempty"` + OnComplete []string `yaml:"on_complete,omitempty"` + OnPropose []string `yaml:"on_propose,omitempty"` } // MergeOver applies overlay fields on top of a base config. Non-empty @@ -106,6 +115,29 @@ func (base *AgentConfig) MergeOver(overlay *AgentConfig) *AgentConfig { merged.Include = append(merged.Include, overlay.Include...) } + // Orchestrator fields + if overlay.Mode != "" { + merged.Mode = overlay.Mode + } + if overlay.Sentinel != nil { + merged.Sentinel = overlay.Sentinel + } + if overlay.PollInterval > 0 { + merged.PollInterval = overlay.PollInterval + } + if overlay.MaxFailures > 0 { + merged.MaxFailures = overlay.MaxFailures + } + if overlay.Heartbeat > 0 { + merged.Heartbeat = overlay.Heartbeat + } + if len(overlay.OnComplete) > 0 { + merged.OnComplete = overlay.OnComplete + } + if len(overlay.OnPropose) > 0 { + merged.OnPropose = overlay.OnPropose + } + merged.BaseAgent = "" return &merged } @@ -389,6 +421,40 @@ func (c *AgentConfig) BuildRunSh() string { return b.String() } +// BuildOrchestratorConfig projects the host-side AgentConfig into the +// in-sandbox orchestrator config, emitting only non-default fields. +func (c *AgentConfig) BuildOrchestratorConfig() map[string]any { + cfg := map[string]any{ + "entrypoint": c.EffectiveEntrypoint(), + "tty": !c.NoTTY(), + } + if c.Mode != "" { + cfg["mode"] = c.Mode + } + if c.Task != "" { + cfg["task"] = "task.md" + } + if c.Sentinel != nil { + cfg["sentinel"] = *c.Sentinel + } + if c.PollInterval > 0 { + cfg["poll_interval"] = c.PollInterval + } + if c.MaxFailures > 0 { + cfg["max_failures"] = c.MaxFailures + } + if c.Heartbeat > 0 { + cfg["heartbeat"] = c.Heartbeat + } + if len(c.OnComplete) > 0 { + cfg["on_complete"] = c.OnComplete + } + if len(c.OnPropose) > 0 { + cfg["on_propose"] = c.OnPropose + } + return cfg +} + func RenderPayload(cfg *AgentConfig, baseDir, destDir string) error { if err := os.MkdirAll(destDir, 0o755); err != nil { return fmt.Errorf("creating payload dir: %w", err) @@ -403,6 +469,15 @@ func RenderPayload(cfg *AgentConfig, baseDir, destDir string) error { return fmt.Errorf("writing run.sh: %w", err) } + orchCfg := cfg.BuildOrchestratorConfig() + orchData, err := yaml.Marshal(orchCfg) + if err != nil { + return fmt.Errorf("marshaling orchestrator config: %w", err) + } + if err := os.WriteFile(filepath.Join(destDir, "orchestrator.yaml"), orchData, 0o644); err != nil { + return fmt.Errorf("writing orchestrator.yaml: %w", err) + } + if cfg.Task != "" { taskSrc := cfg.Task if !filepath.IsAbs(taskSrc) { diff --git a/internal/orchestrator/adapter.go b/internal/orchestrator/adapter.go new file mode 100644 index 0000000..ad652e5 --- /dev/null +++ b/internal/orchestrator/adapter.go @@ -0,0 +1,63 @@ +package orchestrator + +import ( + "fmt" + "os/exec" +) + +type HarnessAdapter interface { + Name() string + BuildCommand(task string, headless bool) *exec.Cmd +} + +func NewAdapter(name string) (HarnessAdapter, error) { + switch name { + case "claude": + return ClaudeAdapter{}, nil + case "codex": + return CodexAdapter{}, nil + case "opencode": + return OpenCodeAdapter{}, nil + default: + return nil, fmt.Errorf("unknown adapter %q", name) + } +} + +type ClaudeAdapter struct{} + +func (ClaudeAdapter) Name() string { return "claude" } + +func (ClaudeAdapter) BuildCommand(task string, headless bool) *exec.Cmd { + if task == "" { + return exec.Command("claude") + } + if headless { + return exec.Command("claude", "--print", task) + } + return exec.Command("claude", "-p", task) +} + +type CodexAdapter struct{} + +func (CodexAdapter) Name() string { return "codex" } + +func (CodexAdapter) BuildCommand(task string, headless bool) *exec.Cmd { + if task == "" { + return exec.Command("codex") + } + if headless { + return exec.Command("codex", "--print", task) + } + return exec.Command("codex", "-p", task) +} + +type OpenCodeAdapter struct{} + +func (OpenCodeAdapter) Name() string { return "opencode" } + +func (OpenCodeAdapter) BuildCommand(task string, headless bool) *exec.Cmd { + if task == "" { + return exec.Command("opencode") + } + return exec.Command("opencode", "run", task) +} diff --git a/internal/orchestrator/adapter_test.go b/internal/orchestrator/adapter_test.go new file mode 100644 index 0000000..769e76f --- /dev/null +++ b/internal/orchestrator/adapter_test.go @@ -0,0 +1,85 @@ +package orchestrator + +import ( + "strings" + "testing" +) + +func TestNewAdapterValid(t *testing.T) { + for _, name := range []string{"claude", "codex", "opencode"} { + a, err := NewAdapter(name) + if err != nil { + t.Errorf("NewAdapter(%q): %v", name, err) + } + if a.Name() != name { + t.Errorf("Name() = %q, want %q", a.Name(), name) + } + } +} + +func TestNewAdapterInvalid(t *testing.T) { + _, err := NewAdapter("unknown") + if err == nil { + t.Error("expected error for unknown adapter") + } +} + +func TestClaudeAdapterCommands(t *testing.T) { + a := ClaudeAdapter{} + + cmd := a.BuildCommand("do stuff", true) + args := strings.Join(cmd.Args, " ") + if args != "claude --print do stuff" { + t.Errorf("headless+task = %q", args) + } + + cmd = a.BuildCommand("do stuff", false) + args = strings.Join(cmd.Args, " ") + if args != "claude -p do stuff" { + t.Errorf("tty+task = %q", args) + } + + cmd = a.BuildCommand("", false) + args = strings.Join(cmd.Args, " ") + if args != "claude" { + t.Errorf("no task = %q", args) + } +} + +func TestCodexAdapterCommands(t *testing.T) { + a := CodexAdapter{} + + cmd := a.BuildCommand("review code", true) + args := strings.Join(cmd.Args, " ") + if args != "codex --print review code" { + t.Errorf("headless+task = %q", args) + } + + cmd = a.BuildCommand("review code", false) + args = strings.Join(cmd.Args, " ") + if args != "codex -p review code" { + t.Errorf("tty+task = %q", args) + } +} + +func TestOpenCodeAdapterCommands(t *testing.T) { + a := OpenCodeAdapter{} + + cmd := a.BuildCommand("fix bugs", true) + args := strings.Join(cmd.Args, " ") + if args != "opencode run fix bugs" { + t.Errorf("headless+task = %q", args) + } + + cmd = a.BuildCommand("fix bugs", false) + args = strings.Join(cmd.Args, " ") + if args != "opencode run fix bugs" { + t.Errorf("tty+task (opencode always uses run) = %q", args) + } + + cmd = a.BuildCommand("", false) + args = strings.Join(cmd.Args, " ") + if args != "opencode" { + t.Errorf("no task = %q", args) + } +} diff --git a/internal/orchestrator/config.go b/internal/orchestrator/config.go new file mode 100644 index 0000000..ead52d8 --- /dev/null +++ b/internal/orchestrator/config.go @@ -0,0 +1,73 @@ +package orchestrator + +import ( + "fmt" + "os" + + "gopkg.in/yaml.v3" +) + +type OrchestratorConfig struct { + Mode string `yaml:"mode"` + Entrypoint string `yaml:"entrypoint"` + Task string `yaml:"task"` + TTY bool `yaml:"tty"` + Sentinel bool `yaml:"sentinel"` + PollInterval int `yaml:"poll_interval"` + MaxFailures int `yaml:"max_failures"` + Heartbeat int `yaml:"heartbeat"` + OnComplete []string `yaml:"on_complete"` + OnPropose []string `yaml:"on_propose"` + SessionDir string `yaml:"session_dir"` +} + +func LoadConfig(path string) (*OrchestratorConfig, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("reading config: %w", err) + } + var cfg OrchestratorConfig + if err := yaml.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("parsing config: %w", err) + } + cfg.ApplyDefaults() + if err := cfg.Validate(); err != nil { + return nil, err + } + return &cfg, nil +} + +func (c *OrchestratorConfig) ApplyDefaults() { + if c.Mode == "" { + c.Mode = "once" + } + if c.Entrypoint == "" { + c.Entrypoint = "claude" + } + if c.PollInterval <= 0 { + c.PollInterval = 300 + } + if c.MaxFailures <= 0 { + c.MaxFailures = 5 + } + if c.Heartbeat == 0 { + c.Heartbeat = 60 + } + if c.SessionDir == "" { + c.SessionDir = "/sandbox/.harness" + } +} + +func (c *OrchestratorConfig) Validate() error { + switch c.Mode { + case "once", "watch": + default: + return fmt.Errorf("invalid mode %q: must be \"once\" or \"watch\"", c.Mode) + } + switch c.Entrypoint { + case "claude", "codex", "opencode": + default: + return fmt.Errorf("unsupported entrypoint %q: must be claude, codex, or opencode", c.Entrypoint) + } + return nil +} diff --git a/internal/orchestrator/config_test.go b/internal/orchestrator/config_test.go new file mode 100644 index 0000000..8110a27 --- /dev/null +++ b/internal/orchestrator/config_test.go @@ -0,0 +1,93 @@ +package orchestrator + +import ( + "os" + "path/filepath" + "testing" +) + +func TestLoadConfig(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "orchestrator.yaml") + os.WriteFile(path, []byte(` +mode: watch +entrypoint: claude +task: task.md +sentinel: true +poll_interval: 600 +max_failures: 3 +heartbeat: 30 +`), 0o644) + + cfg, err := LoadConfig(path) + if err != nil { + t.Fatal(err) + } + if cfg.Mode != "watch" { + t.Errorf("mode = %q, want watch", cfg.Mode) + } + if cfg.Entrypoint != "claude" { + t.Errorf("entrypoint = %q, want claude", cfg.Entrypoint) + } + if cfg.PollInterval != 600 { + t.Errorf("poll_interval = %d, want 600", cfg.PollInterval) + } + if cfg.MaxFailures != 3 { + t.Errorf("max_failures = %d, want 3", cfg.MaxFailures) + } + if cfg.Heartbeat != 30 { + t.Errorf("heartbeat = %d, want 30", cfg.Heartbeat) + } + if !cfg.Sentinel { + t.Error("sentinel = false, want true") + } +} + +func TestLoadConfigDefaults(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "orchestrator.yaml") + os.WriteFile(path, []byte("entrypoint: claude\n"), 0o644) + + cfg, err := LoadConfig(path) + if err != nil { + t.Fatal(err) + } + if cfg.Mode != "once" { + t.Errorf("default mode = %q, want once", cfg.Mode) + } + if cfg.PollInterval != 300 { + t.Errorf("default poll_interval = %d, want 300", cfg.PollInterval) + } + if cfg.MaxFailures != 5 { + t.Errorf("default max_failures = %d, want 5", cfg.MaxFailures) + } + if cfg.Heartbeat != 60 { + t.Errorf("default heartbeat = %d, want 60", cfg.Heartbeat) + } + if cfg.SessionDir != "/sandbox/.harness" { + t.Errorf("default session_dir = %q, want /sandbox/.harness", cfg.SessionDir) + } +} + +func TestValidateInvalidMode(t *testing.T) { + cfg := &OrchestratorConfig{Mode: "invalid", Entrypoint: "claude"} + if err := cfg.Validate(); err == nil { + t.Error("expected error for invalid mode") + } +} + +func TestValidateInvalidEntrypoint(t *testing.T) { + cfg := &OrchestratorConfig{Mode: "once", Entrypoint: "unknown"} + if err := cfg.Validate(); err == nil { + t.Error("expected error for invalid entrypoint") + } +} + +func TestValidateValidEntrypoints(t *testing.T) { + for _, ep := range []string{"claude", "codex", "opencode"} { + cfg := &OrchestratorConfig{Mode: "once", Entrypoint: ep} + if err := cfg.Validate(); err != nil { + t.Errorf("entrypoint %q: unexpected error: %v", ep, err) + } + } +} diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go new file mode 100644 index 0000000..2faa647 --- /dev/null +++ b/internal/orchestrator/orchestrator.go @@ -0,0 +1,226 @@ +package orchestrator + +import ( + "bytes" + "context" + "fmt" + "os" + "os/exec" + "time" +) + +type Orchestrator struct { + config *OrchestratorConfig + adapter HarnessAdapter + session *SessionWriter + configDir string + cycle int + failures int +} + +func New(cfg *OrchestratorConfig, configDir string) (*Orchestrator, error) { + adapter, err := NewAdapter(cfg.Entrypoint) + if err != nil { + return nil, err + } + session, err := NewSessionWriter(cfg.SessionDir) + if err != nil { + return nil, err + } + return &Orchestrator{ + config: cfg, + adapter: adapter, + session: session, + configDir: configDir, + }, nil +} + +func (o *Orchestrator) Run(ctx context.Context) error { + defer o.session.Close() + + switch o.config.Mode { + case "once": + result, err := o.runCycle(ctx) + if err != nil { + return err + } + if result.ExitCode != 0 { + return fmt.Errorf("agent exited with code %d", result.ExitCode) + } + return nil + case "watch": + return o.runWatch(ctx) + default: + return fmt.Errorf("unknown mode %q", o.config.Mode) + } +} + +func (o *Orchestrator) runWatch(ctx context.Context) error { + for { + result, err := o.runCycle(ctx) + if err != nil { + if ctx.Err() != nil { + return ctx.Err() + } + o.failures++ + logf("cycle %d failed: %v (failure %d/%d)", o.cycle, err, o.failures, o.config.MaxFailures) + if o.failures >= o.config.MaxFailures { + return fmt.Errorf("max transient failures (%d) exceeded", o.config.MaxFailures) + } + o.sleepWithHeartbeat(ctx, o.config.PollInterval) + if ctx.Err() != nil { + return ctx.Err() + } + continue + } + + o.runHooks(result) + + if result.IsTerminal() { + logf("cycle %d: terminal status %q (%s)", o.cycle, result.Status, result.Reason) + return nil + } + + switch result.Status { + case "transient_failure": + o.failures++ + logf("cycle %d: transient failure (%s), failure %d/%d", o.cycle, result.Reason, o.failures, o.config.MaxFailures) + if o.failures >= o.config.MaxFailures { + return fmt.Errorf("max transient failures (%d) exceeded", o.config.MaxFailures) + } + default: + o.failures = 0 + } + + poll := o.config.PollInterval + if result.PollSeconds > 0 { + poll = result.PollSeconds + } + + logf("cycle %d: %s (%s), next cycle in %ds", o.cycle, result.Status, result.Reason, poll) + o.sleepWithHeartbeat(ctx, poll) + if ctx.Err() != nil { + return ctx.Err() + } + } +} + +func (o *Orchestrator) runCycle(ctx context.Context) (*CycleResult, error) { + o.cycle++ + start := time.Now() + + task := o.resolveTask() + headless := !o.config.TTY + cmd := o.adapter.BuildCommand(task, headless) + cmd.Stderr = os.Stderr + + var stdout bytes.Buffer + if headless { + cmd.Stdout = &stdout + } else { + cmd.Stdin = os.Stdin + cmd.Stdout = os.Stdout + } + + logf("cycle %d: running %s", o.cycle, o.adapter.Name()) + err := cmd.Run() + duration := time.Since(start) + + var result *CycleResult + exitCode := 0 + if err != nil { + if exitErr, ok := err.(*exec.ExitError); ok { + exitCode = exitErr.ExitCode() + } else { + return nil, fmt.Errorf("running %s: %w", o.adapter.Name(), err) + } + } + + if o.config.Sentinel && headless { + parsed, parseErr := ParseSentinel(stdout.Bytes()) + if parseErr != nil { + logf("cycle %d: sentinel parse error: %v", o.cycle, parseErr) + result = &CycleResult{Status: "transient_failure", Reason: "malformed_sentinel", ExitCode: exitCode} + } else if parsed != nil { + result = parsed + result.ExitCode = exitCode + } + } + + if result == nil { + result = SyntheticResult(exitCode) + } + + record := SessionRecord{ + Timestamp: start, + Cycle: o.cycle, + Entrypoint: o.adapter.Name(), + Mode: o.config.Mode, + DurationSec: duration.Seconds(), + Result: result, + } + if writeErr := o.session.Write(record); writeErr != nil { + logf("warning: failed to write session record: %v", writeErr) + } + + return result, nil +} + +func (o *Orchestrator) resolveTask() string { + if o.config.Task == "" { + return "" + } + data, err := os.ReadFile(o.config.Task) + if err != nil { + logf("warning: could not read task file %s: %v", o.config.Task, err) + return "" + } + return string(data) +} + +func (o *Orchestrator) sleepWithHeartbeat(ctx context.Context, seconds int) { + if o.config.Heartbeat <= 0 { + select { + case <-time.After(time.Duration(seconds) * time.Second): + case <-ctx.Done(): + } + return + } + + deadline := time.After(time.Duration(seconds) * time.Second) + tick := time.NewTicker(time.Duration(o.config.Heartbeat) * time.Second) + defer tick.Stop() + + for { + select { + case <-deadline: + return + case <-tick.C: + logf("heartbeat: cycle %d, sleeping", o.cycle) + case <-ctx.Done(): + return + } + } +} + +func (o *Orchestrator) runHooks(result *CycleResult) { + var hooks []string + switch result.Status { + case "complete": + hooks = o.config.OnComplete + case "propose": + hooks = o.config.OnPropose + } + for _, h := range hooks { + cmd := exec.Command("sh", "-c", h) + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + if err := cmd.Run(); err != nil { + logf("hook failed: %v", err) + } + } +} + +func logf(format string, args ...any) { + fmt.Fprintf(os.Stderr, "harness-orchestrator: "+format+"\n", args...) +} diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go new file mode 100644 index 0000000..4a2d7ab --- /dev/null +++ b/internal/orchestrator/orchestrator_test.go @@ -0,0 +1,218 @@ +package orchestrator + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "testing" + "time" +) + +func TestRunOnceSuccess(t *testing.T) { + dir := t.TempDir() + cfg := &OrchestratorConfig{ + Mode: "once", + Entrypoint: "claude", + SessionDir: dir, + } + cfg.ApplyDefaults() + + orch := &Orchestrator{ + config: cfg, + adapter: &mockAdapter{script: "true"}, + configDir: dir, + } + sw, _ := NewSessionWriter(dir) + orch.session = sw + + err := orch.Run(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + records := readRecords(t, dir) + if len(records) != 1 { + t.Fatalf("records = %d, want 1", len(records)) + } + if records[0].Result.Status != "complete" { + t.Errorf("status = %q, want complete", records[0].Result.Status) + } +} + +func TestRunOnceFailure(t *testing.T) { + dir := t.TempDir() + cfg := &OrchestratorConfig{ + Mode: "once", + Entrypoint: "claude", + SessionDir: dir, + } + cfg.ApplyDefaults() + + orch := &Orchestrator{ + config: cfg, + adapter: &mockAdapter{script: "exit 1"}, + configDir: dir, + } + sw, _ := NewSessionWriter(dir) + orch.session = sw + + err := orch.Run(context.Background()) + if err == nil { + t.Fatal("expected error for exit code 1") + } +} + +func TestRunOnceSentinel(t *testing.T) { + dir := t.TempDir() + cfg := &OrchestratorConfig{ + Mode: "once", + Entrypoint: "claude", + Sentinel: true, + SessionDir: dir, + } + cfg.ApplyDefaults() + + sentinel := `OPENSHELL_AGENT_RESULT {"status":"complete","reason":"merged"}` + orch := &Orchestrator{ + config: cfg, + adapter: &mockAdapter{script: fmt.Sprintf("echo 'working...'; echo '%s'", sentinel)}, + configDir: dir, + } + sw, _ := NewSessionWriter(dir) + orch.session = sw + + err := orch.Run(context.Background()) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + records := readRecords(t, dir) + if records[0].Result.Reason != "merged" { + t.Errorf("reason = %q, want merged", records[0].Result.Reason) + } +} + +func TestRunWatchStopsOnComplete(t *testing.T) { + dir := t.TempDir() + cfg := &OrchestratorConfig{ + Mode: "watch", + Entrypoint: "claude", + Sentinel: true, + PollInterval: 1, + Heartbeat: -1, + SessionDir: dir, + } + + sentinels := []string{ + `OPENSHELL_AGENT_RESULT {"status":"waiting","reason":"pending","next_poll_seconds":1}`, + `OPENSHELL_AGENT_RESULT {"status":"complete","reason":"done"}`, + } + + orch := &Orchestrator{ + config: cfg, + adapter: &multiMockAdapter{sentinels: sentinels}, + configDir: dir, + } + sw, _ := NewSessionWriter(dir) + orch.session = sw + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + err := orch.Run(ctx) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + records := readRecords(t, dir) + if len(records) != 2 { + t.Fatalf("records = %d, want 2", len(records)) + } + if records[0].Result.Status != "waiting" { + t.Errorf("cycle 1 status = %q, want waiting", records[0].Result.Status) + } + if records[1].Result.Status != "complete" { + t.Errorf("cycle 2 status = %q, want complete", records[1].Result.Status) + } +} + +func TestRunWatchContextCancel(t *testing.T) { + dir := t.TempDir() + cfg := &OrchestratorConfig{ + Mode: "watch", + Entrypoint: "claude", + Sentinel: true, + PollInterval: 300, + Heartbeat: -1, + SessionDir: dir, + } + + sentinel := `OPENSHELL_AGENT_RESULT {"status":"waiting","reason":"forever","next_poll_seconds":300}` + orch := &Orchestrator{ + config: cfg, + adapter: &mockAdapter{script: fmt.Sprintf("echo '%s'", sentinel)}, + configDir: dir, + } + sw, _ := NewSessionWriter(dir) + orch.session = sw + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + err := orch.Run(ctx) + if err == nil { + t.Fatal("expected context cancellation error") + } +} + +type mockAdapter struct { + script string +} + +func (m *mockAdapter) Name() string { return "mock" } + +func (m *mockAdapter) BuildCommand(_ string, _ bool) *exec.Cmd { + return exec.Command("sh", "-c", m.script) +} + +type multiMockAdapter struct { + sentinels []string + idx int +} + +func (m *multiMockAdapter) Name() string { return "mock" } + +func (m *multiMockAdapter) BuildCommand(_ string, _ bool) *exec.Cmd { + var script string + if m.idx < len(m.sentinels) { + script = fmt.Sprintf("echo '%s'", m.sentinels[m.idx]) + m.idx++ + } else { + script = `echo 'OPENSHELL_AGENT_RESULT {"status":"terminal_failure","reason":"out_of_results"}'` + } + return exec.Command("sh", "-c", script) +} + +func readRecords(t *testing.T, dir string) []SessionRecord { + t.Helper() + f, err := os.Open(filepath.Join(dir, "sessions.jsonl")) + if err != nil { + t.Fatal(err) + } + defer f.Close() + + var records []SessionRecord + scanner := bufio.NewScanner(f) + for scanner.Scan() { + var r SessionRecord + if err := json.Unmarshal(scanner.Bytes(), &r); err != nil { + t.Fatal(err) + } + records = append(records, r) + } + return records +} diff --git a/internal/orchestrator/sentinel.go b/internal/orchestrator/sentinel.go new file mode 100644 index 0000000..71fb3cf --- /dev/null +++ b/internal/orchestrator/sentinel.go @@ -0,0 +1,49 @@ +package orchestrator + +import ( + "bytes" + "encoding/json" + "fmt" +) + +const SentinelPrefix = "OPENSHELL_AGENT_RESULT " + +type CycleResult struct { + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + PollSeconds int `json:"next_poll_seconds,omitempty"` + Artifacts []string `json:"artifacts,omitempty"` + ExitCode int `json:"exit_code,omitempty"` +} + +func (r *CycleResult) IsTerminal() bool { + return r.Status == "complete" || r.Status == "terminal_failure" +} + +// ParseSentinel scans agent stdout for the last OPENSHELL_AGENT_RESULT line +// and parses the JSON payload. Returns nil if no sentinel is found. +func ParseSentinel(output []byte) (*CycleResult, error) { + lines := bytes.Split(bytes.TrimRight(output, "\n"), []byte("\n")) + for i := len(lines) - 1; i >= 0; i-- { + line := bytes.TrimSpace(lines[i]) + if !bytes.HasPrefix(line, []byte(SentinelPrefix)) { + continue + } + payload := line[len(SentinelPrefix):] + var result CycleResult + if err := json.Unmarshal(payload, &result); err != nil { + return nil, fmt.Errorf("malformed sentinel JSON: %w", err) + } + return &result, nil + } + return nil, nil +} + +// SyntheticResult creates a CycleResult from a process exit code when no +// sentinel protocol is used. +func SyntheticResult(exitCode int) *CycleResult { + if exitCode == 0 { + return &CycleResult{Status: "complete", Reason: "exit_0", ExitCode: 0} + } + return &CycleResult{Status: "terminal_failure", Reason: fmt.Sprintf("exit_%d", exitCode), ExitCode: exitCode} +} diff --git a/internal/orchestrator/sentinel_test.go b/internal/orchestrator/sentinel_test.go new file mode 100644 index 0000000..9dfe0d6 --- /dev/null +++ b/internal/orchestrator/sentinel_test.go @@ -0,0 +1,120 @@ +package orchestrator + +import "testing" + +func TestParseSentinelValid(t *testing.T) { + output := []byte(`some agent output +working on stuff +OPENSHELL_AGENT_RESULT {"status":"complete","reason":"done"} +`) + result, err := ParseSentinel(output) + if err != nil { + t.Fatal(err) + } + if result == nil { + t.Fatal("expected non-nil result") + } + if result.Status != "complete" { + t.Errorf("status = %q, want complete", result.Status) + } + if result.Reason != "done" { + t.Errorf("reason = %q, want done", result.Reason) + } +} + +func TestParseSentinelWithPollSeconds(t *testing.T) { + output := []byte(`OPENSHELL_AGENT_RESULT {"status":"waiting","reason":"checks_pending","next_poll_seconds":120}`) + result, err := ParseSentinel(output) + if err != nil { + t.Fatal(err) + } + if result.Status != "waiting" { + t.Errorf("status = %q, want waiting", result.Status) + } + if result.PollSeconds != 120 { + t.Errorf("poll_seconds = %d, want 120", result.PollSeconds) + } +} + +func TestParseSentinelMultipleLines(t *testing.T) { + output := []byte(`OPENSHELL_AGENT_RESULT {"status":"waiting","reason":"first"} +some output +OPENSHELL_AGENT_RESULT {"status":"complete","reason":"last"} +`) + result, err := ParseSentinel(output) + if err != nil { + t.Fatal(err) + } + if result.Status != "complete" { + t.Errorf("should return last sentinel, got status %q", result.Status) + } + if result.Reason != "last" { + t.Errorf("reason = %q, want last", result.Reason) + } +} + +func TestParseSentinelNoSentinel(t *testing.T) { + output := []byte("just regular output\nno sentinel here\n") + result, err := ParseSentinel(output) + if err != nil { + t.Fatal(err) + } + if result != nil { + t.Errorf("expected nil result for no sentinel, got %+v", result) + } +} + +func TestParseSentinelMalformedJSON(t *testing.T) { + output := []byte(`OPENSHELL_AGENT_RESULT {not valid json}`) + _, err := ParseSentinel(output) + if err == nil { + t.Error("expected error for malformed JSON") + } +} + +func TestParseSentinelEmptyOutput(t *testing.T) { + result, err := ParseSentinel([]byte("")) + if err != nil { + t.Fatal(err) + } + if result != nil { + t.Error("expected nil for empty output") + } +} + +func TestSyntheticResultSuccess(t *testing.T) { + r := SyntheticResult(0) + if r.Status != "complete" { + t.Errorf("status = %q, want complete", r.Status) + } +} + +func TestSyntheticResultFailure(t *testing.T) { + r := SyntheticResult(1) + if r.Status != "terminal_failure" { + t.Errorf("status = %q, want terminal_failure", r.Status) + } + if r.ExitCode != 1 { + t.Errorf("exit_code = %d, want 1", r.ExitCode) + } +} + +func TestCycleResultIsTerminal(t *testing.T) { + cases := []struct { + status string + terminal bool + }{ + {"complete", true}, + {"terminal_failure", true}, + {"waiting", false}, + {"blocked", false}, + {"transient_failure", false}, + {"propose", false}, + } + for _, tc := range cases { + r := CycleResult{Status: tc.status} + if r.IsTerminal() != tc.terminal { + t.Errorf("IsTerminal(%q) = %v, want %v", tc.status, r.IsTerminal(), tc.terminal) + } + } +} diff --git a/internal/orchestrator/session.go b/internal/orchestrator/session.go new file mode 100644 index 0000000..5a0d7fa --- /dev/null +++ b/internal/orchestrator/session.go @@ -0,0 +1,51 @@ +package orchestrator + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "time" +) + +type SessionRecord struct { + Timestamp time.Time `json:"timestamp"` + Cycle int `json:"cycle"` + Entrypoint string `json:"entrypoint"` + Mode string `json:"mode"` + DurationSec float64 `json:"duration_sec"` + Result *CycleResult `json:"result"` +} + +type SessionWriter struct { + path string + file *os.File +} + +func NewSessionWriter(dir string) (*SessionWriter, error) { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("creating session dir: %w", err) + } + path := filepath.Join(dir, "sessions.jsonl") + f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + return nil, fmt.Errorf("opening session file: %w", err) + } + return &SessionWriter{path: path, file: f}, nil +} + +func (w *SessionWriter) Write(record SessionRecord) error { + data, err := json.Marshal(record) + if err != nil { + return fmt.Errorf("marshaling session record: %w", err) + } + data = append(data, '\n') + if _, err := w.file.Write(data); err != nil { + return fmt.Errorf("writing session record: %w", err) + } + return nil +} + +func (w *SessionWriter) Close() error { + return w.file.Close() +} diff --git a/internal/orchestrator/session_test.go b/internal/orchestrator/session_test.go new file mode 100644 index 0000000..e22ff68 --- /dev/null +++ b/internal/orchestrator/session_test.go @@ -0,0 +1,117 @@ +package orchestrator + +import ( + "bufio" + "encoding/json" + "os" + "path/filepath" + "testing" + "time" +) + +func TestSessionWriterCreatesFile(t *testing.T) { + dir := t.TempDir() + w, err := NewSessionWriter(dir) + if err != nil { + t.Fatal(err) + } + defer w.Close() + + path := filepath.Join(dir, "sessions.jsonl") + if _, err := os.Stat(path); err != nil { + t.Errorf("session file not created: %v", err) + } +} + +func TestSessionWriterWriteAndRead(t *testing.T) { + dir := t.TempDir() + w, err := NewSessionWriter(dir) + if err != nil { + t.Fatal(err) + } + + ts := time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) + record := SessionRecord{ + Timestamp: ts, + Cycle: 1, + Entrypoint: "claude", + Mode: "once", + DurationSec: 42.5, + Result: &CycleResult{Status: "complete", Reason: "done"}, + } + if err := w.Write(record); err != nil { + t.Fatal(err) + } + w.Close() + + path := filepath.Join(dir, "sessions.jsonl") + f, err := os.Open(path) + if err != nil { + t.Fatal(err) + } + defer f.Close() + + scanner := bufio.NewScanner(f) + if !scanner.Scan() { + t.Fatal("no line in JSONL") + } + + var got SessionRecord + if err := json.Unmarshal(scanner.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.Cycle != 1 { + t.Errorf("cycle = %d, want 1", got.Cycle) + } + if got.Entrypoint != "claude" { + t.Errorf("entrypoint = %q, want claude", got.Entrypoint) + } + if got.Result.Status != "complete" { + t.Errorf("result.status = %q, want complete", got.Result.Status) + } +} + +func TestSessionWriterAppends(t *testing.T) { + dir := t.TempDir() + w, err := NewSessionWriter(dir) + if err != nil { + t.Fatal(err) + } + + for i := 1; i <= 3; i++ { + w.Write(SessionRecord{ + Timestamp: time.Now(), + Cycle: i, + Entrypoint: "claude", + Mode: "watch", + Result: &CycleResult{Status: "waiting"}, + }) + } + w.Close() + + path := filepath.Join(dir, "sessions.jsonl") + f, _ := os.Open(path) + defer f.Close() + + count := 0 + scanner := bufio.NewScanner(f) + for scanner.Scan() { + count++ + } + if count != 3 { + t.Errorf("lines = %d, want 3", count) + } +} + +func TestSessionWriterCreatesSubdirs(t *testing.T) { + dir := filepath.Join(t.TempDir(), "nested", "dir") + w, err := NewSessionWriter(dir) + if err != nil { + t.Fatal(err) + } + w.Close() + + if _, err := os.Stat(filepath.Join(dir, "sessions.jsonl")); err != nil { + t.Errorf("file not created in nested dir: %v", err) + } +} diff --git a/profiles/images/sandbox-default/Dockerfile b/profiles/images/sandbox-default/Dockerfile index 04608e0..a3a7fa9 100644 --- a/profiles/images/sandbox-default/Dockerfile +++ b/profiles/images/sandbox-default/Dockerfile @@ -73,6 +73,11 @@ COPY claude.json /sandbox/.claude.json COPY mcp.json /sandbox/.mcp.json COPY opencode.json /sandbox/opencode.json +# In-sandbox orchestrator binary (manages agent lifecycle, watch mode, sentinel parsing) +# Built by: make orchestrator +# Falls back to run.sh if not present in the image. +COPY harness-orchestrator /usr/local/bin/harness-orchestrator + # Harness config directory (populated at runtime via upload) RUN mkdir -p /sandbox/.config/openshell && chown sandbox:sandbox /sandbox/.config/openshell