Keep agent processes alive for one-hour idle window

CONTRIBUTING.md

@@ -148,10 +148,11 @@ description and keep ownership on the side listed here.
 - Every subprocess must be waited/reaped. Cancellation must close stdin when
   appropriate, send a graceful signal first, and escalate to kill after a
   bounded timeout.
-- A Parsar conversation is not a long-lived OS process. Each prompt turn may
-  start a CLI process, but the adapter must either resume the upstream engine
-  session on the next turn or explicitly document why that engine cannot
-  resume.
+- A completed prompt closes its protocol stream immediately, but daemon-side
+  CLI processes and their background children stay alive until the
+  conversation has received no new prompt for one hour. A new prompt for the
+  same `AgentStateKey` renews that idle window. Explicit cancellation, device
+  shutdown, and daemon shutdown still terminate processes immediately.
 - When an engine supports resume, persist the upstream session id through
   `agent_engine_sessions` and pass `AgentSessionID` plus `AgentStateKey` over
   the daemon protocol. Do not keep resume ids only in adapter memory, files

apps/parsar-daemon/internal/agent/claudecode/session.go

@@ -78,6 +78,8 @@ type Session struct {
 
 	out          chan<- proto.Envelope
 	closeOutOnce sync.Once
+	outMu        sync.RWMutex
+	outClosed    bool
 
 	// cancelCtx is a child of parent ctx so Session.Cancel can signal
 	// everyone without racing router shutdown.
@@ -523,7 +525,8 @@ func (s *Session) run(stdout io.Reader) {
 		}
 		if tx.Terminal {
 			terminal = true
-			s.finishAfterTerminal()
+			s.stopAllAskTimers()
+			s.closeOut()
 			break
 		}
 	}
@@ -594,6 +597,11 @@ func (s *Session) doneMetaForCancel() map[string]any {
 }
 
 func (s *Session) trySend(env proto.Envelope) {
+	s.outMu.RLock()
+	defer s.outMu.RUnlock()
+	if s.outClosed {
+		return
+	}
 	select {
 	case s.out <- env:
 	case <-time.After(2 * time.Second):
@@ -603,19 +611,12 @@ func (s *Session) trySend(env proto.Envelope) {
 }
 
 func (s *Session) closeOut() {
-	s.closeOutOnce.Do(func() { close(s.out) })
-}
-
-func (s *Session) finishAfterTerminal() {
-	s.stdinMu.Lock()
-	if s.stdin != nil {
-		_ = s.stdin.Close()
-		s.stdin = nil
-	}
-	s.stdinMu.Unlock()
-	if s.proc != nil {
-		s.proc.Cancel()
-	}
+	s.closeOutOnce.Do(func() {
+		s.outMu.Lock()
+		s.outClosed = true
+		close(s.out)
+		s.outMu.Unlock()
+	})
 }
 
 // resolveSessionWorkDir returns the directory that BOTH plugin installs

apps/parsar-daemon/internal/agent/claudecode/session_export_test.go

@@ -32,3 +32,7 @@ func NewSessionForTest(ctx context.Context, req proto.PromptRequestPayload, out
 func (s *Session) SubmitPromptForUserChoiceForTest(askID string, decision proto.PromptForUserChoiceDecisionPayload) error {
 	return s.SubmitPromptForUserChoice(context.Background(), askID, decision)
 }
+
+func (s *Session) ProcessDoneForTest() <-chan struct{} {
+	return s.proc.Done()
+}

apps/parsar-daemon/internal/agent/claudecode/session_test.go

@@ -62,6 +62,19 @@ func runFakeClaude(role string) {
 			"usage":      map[string]int{"input_tokens": 5, "output_tokens": 2},
 		})
 
+	case "terminal-wait":
+		_ = enc.Encode(map[string]any{
+			"type": "system", "subtype": "init",
+			"session_id": "sess_terminal_wait",
+		})
+		_ = enc.Encode(map[string]any{
+			"type": "result", "subtype": "success",
+			"result":     "background work started",
+			"session_id": "sess_terminal_wait",
+		})
+		for stdin.Scan() {
+		}
+
 	case "echo-error":
 		_ = enc.Encode(map[string]any{
 			"type": "result", "subtype": "error_during_execution",
@@ -242,6 +255,36 @@ func TestSessionEndToEndSuccess(t *testing.T) {
 	}
 }
 
+func TestTerminalResultKeepsProcessAliveUntilCancel(t *testing.T) {
+	out := make(chan proto.Envelope, 16)
+	sess, err := claudecode.NewSessionForTest(context.Background(),
+		helperReq("run_terminal_wait", "start background work", "terminal-wait"), out, helperConfig())
+	if err != nil {
+		t.Fatalf("NewSessionForTest: %v", err)
+	}
+
+	got, closed := drain(t, out, 5*time.Second)
+	if !closed {
+		t.Fatalf("out did not close, drained %d envelopes", len(got))
+	}
+	mustContain(t, envTypes(got), proto.TypeDone)
+
+	select {
+	case <-sess.ProcessDoneForTest():
+		t.Fatal("terminal result killed the CLI process before the idle timeout")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	if err := sess.Cancel(context.Background()); err != nil {
+		t.Fatalf("Cancel: %v", err)
+	}
+	select {
+	case <-sess.ProcessDoneForTest():
+	case <-time.After(2 * time.Second):
+		t.Fatal("CLI process did not exit after explicit cancel")
+	}
+}
+
 func TestSessionEndToEndError(t *testing.T) {
 	out := make(chan proto.Envelope, 16)
 	sess, err := claudecode.NewSessionForTest(context.Background(),

apps/parsar-daemon/internal/agent/codex/session.go

@@ -39,8 +39,9 @@ func defaultSessionConfig() sessionConfig {
 }
 
 // Factory implements agent.Factory for agent_kind="codex". Spawns one
-// codex app-server child per prompt; the child is torn down when the
-// turn completes or the parent context is cancelled.
+// codex app-server child per prompt. The run stream closes when the turn
+// completes; the router retains the child until the conversation's idle
+// window expires or cancellation shuts it down sooner.
 func Factory(ctx context.Context, req proto.PromptRequestPayload, out chan<- proto.Envelope) (agent.Session, error) {
 	return newSession(ctx, req, out, defaultSessionConfig())
 }
@@ -65,6 +66,8 @@ type Session struct {
 
 	cancelOnce   sync.Once
 	closeOutOnce sync.Once
+	outMu        sync.RWMutex
+	outClosed    bool
 	waitDone     chan struct{}
 	cleanup      func()
 
@@ -192,7 +195,6 @@ func (s *Session) SubmitPromptForUserChoice(_ context.Context, _ string, _ proto
 
 func (s *Session) run(plan SessionPlan, req proto.PromptRequestPayload) {
 	defer close(s.waitDone)
-	defer func() { _ = s.rpc.Close() }()
 	defer s.cleanup()
 	defer s.closeOut()
 
@@ -655,6 +657,11 @@ func (s *Session) emitTerminal(message string, asError bool) {
 }
 
 func (s *Session) trySend(env proto.Envelope) {
+	s.outMu.RLock()
+	defer s.outMu.RUnlock()
+	if s.outClosed {
+		return
+	}
 	select {
 	case s.out <- env:
 	case <-s.cancelCtx.Done():
@@ -664,16 +671,16 @@ func (s *Session) trySend(env proto.Envelope) {
 }
 
 func (s *Session) closeOut() {
-	s.closeOutOnce.Do(func() { close(s.out) })
+	s.closeOutOnce.Do(func() {
+		s.outMu.Lock()
+		s.outClosed = true
+		close(s.out)
+		s.outMu.Unlock()
+	})
 }
 
 func (s *Session) finishAfterTerminal() {
-	if s.cancelFn != nil {
-		s.cancelFn()
-	}
-	if s.rpc != nil {
-		_ = s.rpc.Close()
-	}
+	s.closeOut()
 }
 
 // ---------------------------------------------------------------------------

apps/parsar-daemon/internal/agent/codex/session_log_test.go

@@ -173,7 +173,7 @@ func TestOnTurnFailed_LogsBufferedError(t *testing.T) {
 	}
 }
 
-func TestTerminalTurnClosesRPCAndCancelsSession(t *testing.T) {
+func TestTerminalTurnKeepsRPCAliveUntilSessionCancel(t *testing.T) {
 	tests := []struct {
 		name string
 		run  func(*Session)
@@ -219,14 +219,15 @@ func TestTerminalTurnClosesRPCAndCancelsSession(t *testing.T) {
 
 			tt.run(s)
 
-			if rpc.Alive() {
-				t.Fatal("terminal turn must close the codex RPC client")
+			if !rpc.Alive() {
+				t.Fatal("terminal turn must keep the codex RPC client alive during the idle window")
 			}
 			select {
 			case <-ctx.Done():
+				t.Fatal("terminal turn must not cancel the session context")
 			default:
-				t.Fatal("terminal turn must cancel the session context")
 			}
+
 		})
 	}
 }

apps/parsar-daemon/internal/agent/registry.go

@@ -6,9 +6,10 @@
 //
 //   - Factory takes an out chan<- proto.Envelope owned by the dispatch
 //     router. The agent SENDS upstream events on it and OWNS the
-//     close: it MUST close(out) exactly once when the session is
-//     fully done, AFTER emitting a terminal "done" or "error" frame.
-//     The router uses the close as the "session terminated" signal.
+//     close: it MUST close(out) exactly once after emitting the current
+//     run's terminal "done" or "error" frame. The underlying CLI may
+//     remain alive during the router's idle window and is terminated
+//     through Session.Cancel.
 //
 //   - Session.Cancel is best-effort and idempotent: a session that
 //     already finished naturally must accept a Cancel call without
@@ -36,8 +37,9 @@ import (
 // doc). ctx is cancelled by the router to wind the session down.
 type Factory func(ctx context.Context, req proto.PromptRequestPayload, out chan<- proto.Envelope) (Session, error)
 
-// Session is one in-flight prompt run. Completion is signalled by
-// closing the out channel passed to its Factory.
+// Session owns one prompt run and any CLI process retained after that
+// run. Run completion is signalled by closing out; process teardown is
+// signalled separately through Cancel.
 type Session interface {
 	// Cancel signals the session to abort. Idempotent. Actual teardown
 	// happens asynchronously and is signalled via the out channel close.