Declare agent-host Session views in the Runtime–Harness protocol

apps/daemon/internal/agent/claudesdk/declaration.go

@@ -62,7 +62,7 @@ func discoverWithCheck(parent context.Context, options agent.DiscoveryOptions, d
 	}
 	out := &agent.Runtime{Info: descriptor, Session: func(context.Context, proto.PromptRequestPayload, chan<- proto.Envelope) (agent.Session, error) {
 		return nil, fmt.Errorf("claude_sdk: configured runtime is unavailable")
-	}}
+	}, View: nil}
 	var config Config
 
 	fail := func(err error) *agent.Runtime {

apps/daemon/internal/agent/clirunner/handle.go

@@ -0,0 +1,106 @@
+package clirunner
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+)
+
+// Handle is a running process that clirunner did not start, such as a Harness in an agent-host Session view. Its end also ends every descendant.
+type Handle interface {
+	// Signal delivers sig to the process and every descendant it still has. Once the process itself has exited it delivers nothing and returns an error that matches os.ErrProcessDone, even while its descendants are still ending.
+	Signal(syscall.Signal) error
+	// Wait returns once the process and its descendants have ended. The code is -1 when a signal ended the process. An error means the exit is unknown.
+	Wait() (int, error)
+	// Close kills whatever still runs and releases the handle. It never closes the stdio ends in HandleOptions, which the Process owns. It is safe to call more than once and after Wait.
+	Close() error
+}
+
+// HandleOptions are the stdio ends of a Handle's process and its cancellation grace. The Process owns the stdio ends and closes them in Wait.
+type HandleOptions struct {
+	Parent context.Context
+	// Stdin is nil when the process has no stdin pipe.
+	Stdin       io.WriteCloser
+	Stdout      io.ReadCloser
+	Stderr      io.ReadCloser
+	KillTimeout time.Duration
+}
+
+// FromHandle returns a Process that owns h. Cancel, or cancelling Parent, sends TERM and closes h after KillTimeout; cancelling a process that has ended does nothing. Wait closes the stdio ends and h, and returns the context error when the TERM reached the running process and it then exited 0, as exec.CommandContext does. A TERM that finds the process exited, which Signal reports with os.ErrProcessDone, leaves its result as it was.
+func FromHandle(h Handle, opts HandleOptions) (*Process, error) {
+	if h == nil || opts.Stdout == nil || opts.Stderr == nil {
+		return nil, errors.New("clirunner: handle, stdout and stderr required")
+	}
+	if opts.Parent == nil {
+		opts.Parent = context.Background()
+	}
+	if opts.KillTimeout <= 0 {
+		opts.KillTimeout = 3 * time.Second
+	}
+	ctx, cancel := context.WithCancel(opts.Parent)
+	p := &Process{Stdin: opts.Stdin, Stdout: opts.Stdout, Stderr: opts.Stderr, ctx: ctx, cancel: cancel, done: make(chan struct{}), killAfter: opts.KillTimeout}
+	var terminateOnce sync.Once
+	var interrupted atomic.Bool
+	p.cancelProcess = func() error {
+		terminateOnce.Do(func() {
+			err := h.Signal(syscall.SIGTERM)
+			interrupted.Store(err == nil)
+			go func() {
+				if err == nil || errors.Is(err, os.ErrProcessDone) {
+					timer := time.NewTimer(p.killAfter)
+					defer timer.Stop()
+					select {
+					case <-p.done:
+						return
+					case <-timer.C:
+					}
+				}
+				_ = h.Close()
+			}()
+		})
+		return nil
+	}
+	stop := context.AfterFunc(ctx, func() { _ = p.cancelProcess() })
+	var waitErr error
+	go func() {
+		code, err := h.Wait()
+		stop()
+		// A cancel in progress settles whether it interrupted the process, and none starts after the end.
+		terminateOnce.Do(func() {})
+		switch {
+		case err != nil:
+			waitErr = fmt.Errorf("clirunner: wait: %w", err)
+		case code != 0:
+			waitErr = fmt.Errorf("clirunner: exit code %d", code)
+		case interrupted.Load():
+			// Cancel runs before the context is cancelled.
+			if waitErr = ctx.Err(); waitErr == nil {
+				waitErr = context.Canceled
+			}
+		}
+		if err == nil {
+			p.exitCode, p.exited = code, true
+		}
+		close(p.done)
+	}()
+	var waitOnce sync.Once
+	p.waitProcess = func() error {
+		<-p.done
+		waitOnce.Do(func() {
+			closePipe(p.Stdin)
+			closePipe(p.Stdout)
+			closePipe(p.Stderr)
+			if err := h.Close(); err != nil {
+				waitErr = errors.Join(waitErr, fmt.Errorf("clirunner: close: %w", err))
+			}
+		})
+		return waitErr
+	}
+	return p, nil
+}

apps/daemon/internal/agent/clirunner/handle_test.go

@@ -0,0 +1,177 @@
+package clirunner
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"slices"
+	"strings"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+)
+
+func TestHandleProcessCancellation(t *testing.T) {
+	const grace = 100 * time.Millisecond
+	for name, termExit := range map[string]int{"ignores TERM": -1, "exits 0 on TERM": 0, "exits 3 on TERM": 3} {
+		t.Run(name, func(t *testing.T) {
+			h := newFakeHandle(termExit)
+			p, err := FromHandle(h, HandleOptions{Stdout: emptyReader(), Stderr: emptyReader(), KillTimeout: grace})
+			if err != nil {
+				t.Fatal(err)
+			}
+			started := time.Now()
+			p.Cancel()
+			if signals := h.received(); !slices.Equal(signals, []syscall.Signal{syscall.SIGTERM}) {
+				t.Fatalf("signals = %v, want TERM", signals)
+			}
+			<-p.Done()
+			waitErr := p.Wait()
+			code, ok := p.ExitCode()
+			switch {
+			case termExit < 0:
+				if h.closedAt.Sub(started) < grace || waitErr == nil || ok {
+					t.Fatalf("closed after %v, Wait = %v, ExitCode ok = %v; want close after the grace and an unknown exit", h.closedAt.Sub(started), waitErr, ok)
+				}
+			case termExit == 0:
+				if !errors.Is(waitErr, context.Canceled) || !ok || code != 0 {
+					t.Fatalf("Wait = %v, ExitCode = %d, %v; want the context error and exit 0", waitErr, code, ok)
+				}
+			default:
+				if waitErr == nil || errors.Is(waitErr, context.Canceled) || !ok || code != termExit {
+					t.Fatalf("Wait = %v, ExitCode = %d, %v; want exit %d", waitErr, code, ok, termExit)
+				}
+			}
+			if !h.isClosed() {
+				t.Fatal("Wait did not close the handle")
+			}
+		})
+	}
+}
+
+func TestHandleProcessCancelAfterExit(t *testing.T) {
+	h := newFakeHandle(-1)
+	stdin, stdout, stderr := &closer{}, &closer{Reader: strings.NewReader("output")}, &closer{Reader: strings.NewReader("")}
+	p, err := FromHandle(h, HandleOptions{Stdin: stdin, Stdout: stdout, Stderr: stderr})
+	if err != nil {
+		t.Fatal(err)
+	}
+	h.exit <- 0
+	<-p.Done()
+	p.Cancel()
+	if signals := h.received(); len(signals) != 0 || h.isClosed() || stdout.isClosed() {
+		t.Fatalf("Cancel after exit sent %v and closed the handle %v, stdout %v; want nothing", signals, h.isClosed(), stdout.isClosed())
+	}
+	if out, err := io.ReadAll(p.Stdout); err != nil || string(out) != "output" {
+		t.Fatalf("read %q, %v; want the whole output", out, err)
+	}
+	for range 2 {
+		if err := p.Wait(); err != nil {
+			t.Fatalf("Wait = %v, want success", err)
+		}
+	}
+	if !stdin.isClosed() || !stdout.isClosed() || !stderr.isClosed() || !h.isClosed() {
+		t.Fatal("Wait did not close stdin, stdout, stderr and the handle")
+	}
+}
+
+// TestHandleProcessCancelAfterLeaderExit checks that a cancel that finds the process exited 0 while its descendants still end leaves the success.
+func TestHandleProcessCancelAfterLeaderExit(t *testing.T) {
+	h := newFakeHandle(-1)
+	h.leaderExited = true
+	p, err := FromHandle(h, HandleOptions{Stdout: emptyReader(), Stderr: emptyReader()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	p.Cancel()
+	h.exit <- 0
+	if err := p.Wait(); err != nil {
+		t.Fatalf("Wait = %v, want success", err)
+	}
+}
+
+type fakeHandle struct {
+	termExit     int
+	leaderExited bool
+	exit         chan int
+	closed       chan struct{}
+	closeOnce    sync.Once
+	closedAt     time.Time
+	mu           sync.Mutex
+	signals      []syscall.Signal
+}
+
+func newFakeHandle(termExit int) *fakeHandle {
+	return &fakeHandle{termExit: termExit, exit: make(chan int, 1), closed: make(chan struct{})}
+}
+
+func (h *fakeHandle) Signal(sig syscall.Signal) error {
+	if h.leaderExited {
+		return os.ErrProcessDone
+	}
+	h.mu.Lock()
+	h.signals = append(h.signals, sig)
+	h.mu.Unlock()
+	if sig == syscall.SIGTERM && h.termExit >= 0 {
+		h.exit <- h.termExit
+	}
+	return nil
+}
+
+func (h *fakeHandle) Wait() (int, error) {
+	select {
+	case code := <-h.exit:
+		return code, nil
+	case <-h.closed:
+		return 0, errors.New("closed")
+	}
+}
+
+func (h *fakeHandle) Close() error {
+	h.closeOnce.Do(func() {
+		h.closedAt = time.Now()
+		close(h.closed)
+	})
+	return nil
+}
+
+func (h *fakeHandle) isClosed() bool {
+	select {
+	case <-h.closed:
+		return true
+	default:
+		return false
+	}
+}
+
+func (h *fakeHandle) received() []syscall.Signal {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return slices.Clone(h.signals)
+}
+
+// closer is a stdio end that records Close.
+type closer struct {
+	io.Reader
+	mu     sync.Mutex
+	closed bool
+}
+
+func (c *closer) Write(b []byte) (int, error) { return len(b), nil }
+
+func (c *closer) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.closed = true
+	return nil
+}
+
+func (c *closer) isClosed() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.closed
+}
+
+func emptyReader() io.ReadCloser { return io.NopCloser(strings.NewReader("")) }

apps/daemon/internal/agent/clirunner/process.go

@@ -38,6 +38,10 @@ type Process struct {
 	waitOnce      sync.Once
 	cancelProcess func() error
 	waitProcess   func() error
+
+	// A handle-backed process records its exit here before closing done.
+	exitCode int
+	exited   bool
 }
 
 func Start(opts StartOptions) (*Process, error) {
@@ -144,17 +148,39 @@ func (p *Process) Cancel() {
 }
 
 func (p *Process) Wait() error {
-	if p == nil || p.Cmd == nil {
+	if p == nil {
 		return nil
 	}
 	if p.waitProcess != nil {
 		return p.waitProcess()
 	}
+	if p.Cmd == nil {
+		return nil
+	}
 	err := p.Cmd.Wait()
 	p.waitOnce.Do(func() { close(p.done) })
 	return err
 }
 
+// ExitCode returns the exit code once Done is closed, or -1 when a signal ended the process. ok is false before then and when the exit is unknown.
+func (p *Process) ExitCode() (code int, ok bool) {
+	if p == nil {
+		return 0, false
+	}
+	select {
+	case <-p.done:
+	default:
+		return 0, false
+	}
+	if p.Cmd == nil {
+		return p.exitCode, p.exited
+	}
+	if p.Cmd.ProcessState == nil {
+		return 0, false
+	}
+	return p.Cmd.ProcessState.ExitCode(), true
+}
+
 func closePipe(p io.Closer) {
 	if p != nil {
 		_ = p.Close()

apps/daemon/internal/agent/codex/declaration.go

@@ -45,13 +45,13 @@ var Declaration = agent.Declaration{Info: proto.SupportedAgentKind{
 		MCPHTTPRequired:                proto.CapabilityUnsupported,
 		MCPHTTPBearerAuth:              proto.CapabilitySupported,
 	},
-}, Configuration: configuration.Configuration(), Discover: discover}
+}, Configuration: configuration.Configuration(), ConnectionOptions: []string{"mcp_servers", "env"}, Discover: discover}
 
 func discover(ctx context.Context, options agent.DiscoveryOptions, info proto.SupportedAgentKind) *agent.Runtime {
 	return discoverWithCheck(ctx, options, info, CheckCLIAvailable)
 }
 func discoverWithCheck(parent context.Context, options agent.DiscoveryOptions, info proto.SupportedAgentKind, check func(context.Context, string) (string, error)) *agent.Runtime {
-	runtime := &agent.Runtime{Info: info, Session: Factory, SessionCapabilityContext: true, ExecutorCapabilityContext: true}
+	runtime := &agent.Runtime{Info: info, Session: Factory, SessionCapabilityContext: true, ExecutorCapabilityContext: true, View: nil}
 	ctx, cancel := context.WithTimeout(parent, 15*time.Second)
 	defer cancel()
 	version, err := check(ctx, "")

apps/daemon/internal/agent/codex/executor_native_test.go

@@ -69,7 +69,7 @@ func TestExecutorNativeReuse(t *testing.T) {
 		}
 	})
 	t.Logf("native_version=0.153.4 prepare_ms=%d", time.Since(began).Milliseconds())
-	pid := e.prepared.session.rpc.cmd.Process.Pid
+	pid := e.prepared.session.rpc.process.Cmd.Process.Pid
 	var thread string
 	run := func(id, prompt string, old agent.Turn, interrupt bool) (agent.Turn, string) {
 		t.Helper()
@@ -135,7 +135,7 @@ func TestExecutorNativeReuse(t *testing.T) {
 		if thread == "" {
 			thread = s.currentThreadID()
 		}
-		if thread == "" || s.currentThreadID() != thread || !s.rpc.Alive() || s.rpc.cmd.Process.Pid != pid {
+		if thread == "" || s.currentThreadID() != thread || !s.rpc.Alive() || s.rpc.process.Cmd.Process.Pid != pid {
 			t.Fatal("native owner/thread changed")
 		}
 		t.Logf("turn=%s first_event_ms=%d settled_ms=%d same_process=true same_thread=true", id, first.Milliseconds(), time.Since(started).Milliseconds())

apps/daemon/internal/agent/codex/executor_test.go

@@ -170,7 +170,7 @@ func TestExecutorCloseRetainsPlanUntilReaped(t *testing.T) {
 		t.Fatal(err)
 	}
 	rpc := NewJSONRPCClient(JSONRPCConfig{})
-	rpc.process, rpc.cmd, rpc.stdin, rpc.alive = process, process.Cmd, process.Stdin, true
+	rpc.process, rpc.stdin, rpc.alive = process, process.Stdin, true
 
 	var reap sync.Once
 	t.Cleanup(func() { process.Cancel(); reap.Do(rpc.waitChild) })

apps/daemon/internal/agent/codex/preparation_router_test.go

@@ -119,7 +119,7 @@ func TestPreparationRouterRetainsActualNativeChild(t *testing.T) {
 			ready := await("ready")
 			p := <-prepared
 			assertPreparationOnly(t, root)
-			pid := p.session.rpc.cmd.Process.Pid
+			pid := p.session.rpc.process.Cmd.Process.Pid
 			if r.ActiveRuns() != 0 {
 				t.Fatal("preparation became a Run")
 			}