context: thread daemon ctx through sessions to agents
Daemon creates a root context, cancelled on SIGTERM. Sessions derive their context from it. Agent turns derive from session ctx. Tools get the turn ctx. Signal propagation: - SIGTERM → daemonCancel() → all session ctxs cancelled → agent turns exit - User interrupt → agent.Interrupt() (SIGINT equivalent, turn-level only) - Session kill → sess.Cancel() → that session's agents stop Previously sessions used context.Background() — daemon shutdown had no automatic propagation to running agents.
This commit is contained in:
parent
bd30678ea0
commit
d17b5df7dc
5
main.go
5
main.go
|
|
@ -178,7 +178,11 @@ func runServer(sockPath string) {
|
||||||
// Elevate broker (initialized after manager; closures capture the pointer).
|
// Elevate broker (initialized after manager; closures capture the pointer).
|
||||||
var elevateBroker *elevate.Broker
|
var elevateBroker *elevate.Broker
|
||||||
|
|
||||||
|
daemonCtx, daemonCancel := context.WithCancel(context.Background())
|
||||||
|
defer daemonCancel()
|
||||||
|
|
||||||
mgr := mgr.NewManager(mgr.ManagerConfig{
|
mgr := mgr.NewManager(mgr.ManagerConfig{
|
||||||
|
Ctx: daemonCtx,
|
||||||
ToolRegistry: toolRegistry,
|
ToolRegistry: toolRegistry,
|
||||||
SkillsRegistry: skillsRegistry,
|
SkillsRegistry: skillsRegistry,
|
||||||
AgentsDir: agentsDirs[0],
|
AgentsDir: agentsDirs[0],
|
||||||
|
|
@ -332,6 +336,7 @@ func runServer(sockPath string) {
|
||||||
<-sigChan
|
<-sigChan
|
||||||
|
|
||||||
fmt.Println("shutting down")
|
fmt.Println("shutting down")
|
||||||
|
daemonCancel() // signal all sessions via context propagation
|
||||||
if dbusAdapter != nil {
|
if dbusAdapter != nil {
|
||||||
dbusAdapter.Close()
|
dbusAdapter.Close()
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -101,6 +101,7 @@ func newTestSessionManager(t *testing.T) *mgr.Manager {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
sink := testSink()
|
sink := testSink()
|
||||||
return mgr.NewManager(mgr.ManagerConfig{
|
return mgr.NewManager(mgr.ManagerConfig{
|
||||||
|
Ctx: context.Background(),
|
||||||
Log: sink.NewLogger("test"),
|
Log: sink.NewLogger("test"),
|
||||||
Sink: sink,
|
Sink: sink,
|
||||||
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
||||||
|
|
@ -112,6 +113,7 @@ func newTestSessionManagerWithCore(t *testing.T) *mgr.Manager {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
sink := testSink()
|
sink := testSink()
|
||||||
return mgr.NewManager(mgr.ManagerConfig{
|
return mgr.NewManager(mgr.ManagerConfig{
|
||||||
|
Ctx: context.Background(),
|
||||||
Log: sink.NewLogger("test"),
|
Log: sink.NewLogger("test"),
|
||||||
Sink: sink,
|
Sink: sink,
|
||||||
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
||||||
|
|
@ -347,6 +349,7 @@ func TestSessionManagerShutdown(t *testing.T) {
|
||||||
func TestSessionManagerRename(t *testing.T) {
|
func TestSessionManagerRename(t *testing.T) {
|
||||||
sink := testSink()
|
sink := testSink()
|
||||||
s := mgr.NewManager(mgr.ManagerConfig{
|
s := mgr.NewManager(mgr.ManagerConfig{
|
||||||
|
Ctx: context.Background(),
|
||||||
Log: sink.NewLogger("test"),
|
Log: sink.NewLogger("test"),
|
||||||
Sink: sink,
|
Sink: sink,
|
||||||
ReadFile: func(string) ([]byte, error) { return nil, nil },
|
ReadFile: func(string) ([]byte, error) { return nil, nil },
|
||||||
|
|
@ -1145,6 +1148,7 @@ func TestSessionManagerCreateSessionEnvExpansion(t *testing.T) {
|
||||||
t.Setenv("TEST_CWD", "/expanded/path")
|
t.Setenv("TEST_CWD", "/expanded/path")
|
||||||
sink := testSink()
|
sink := testSink()
|
||||||
s := mgr.NewManager(mgr.ManagerConfig{
|
s := mgr.NewManager(mgr.ManagerConfig{
|
||||||
|
Ctx: context.Background(),
|
||||||
Log: sink.NewLogger("test"),
|
Log: sink.NewLogger("test"),
|
||||||
Sink: sink,
|
Sink: sink,
|
||||||
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
||||||
|
|
@ -1169,6 +1173,7 @@ func TestSessionManagerCreateSessionEnvExpansion(t *testing.T) {
|
||||||
func TestSessionManagerCreateSessionTildeExpansion(t *testing.T) {
|
func TestSessionManagerCreateSessionTildeExpansion(t *testing.T) {
|
||||||
sink := testSink()
|
sink := testSink()
|
||||||
s := mgr.NewManager(mgr.ManagerConfig{
|
s := mgr.NewManager(mgr.ManagerConfig{
|
||||||
|
Ctx: context.Background(),
|
||||||
Log: sink.NewLogger("test"),
|
Log: sink.NewLogger("test"),
|
||||||
Sink: sink,
|
Sink: sink,
|
||||||
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil },
|
||||||
|
|
|
||||||
|
|
@ -137,7 +137,7 @@ func (s *Manager) CreateSession(args []string) (string, error) {
|
||||||
// FIXME: Eager dial means session creation blocks on SSH connect.
|
// FIXME: Eager dial means session creation blocks on SSH connect.
|
||||||
// Eventually make this async or move host info into the prompt
|
// Eventually make this async or move host info into the prompt
|
||||||
// scripts themselves (running remotely via the RPC).
|
// scripts themselves (running remotely via the RPC).
|
||||||
rsrv, dialErr := execute.RemoteDial(context.Background(), execute.RemoteConfig{
|
rsrv, dialErr := execute.RemoteDial(s.cfg.Ctx, execute.RemoteConfig{
|
||||||
SSHTarget: remoteTarget,
|
SSHTarget: remoteTarget,
|
||||||
CWD: cwd,
|
CWD: cwd,
|
||||||
})
|
})
|
||||||
|
|
@ -228,7 +228,7 @@ func (s *Manager) CreateSession(args []string) (string, error) {
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(s.cfg.Ctx)
|
||||||
sess := NewSession(sessID, core, ctx, cancel)
|
sess := NewSession(sessID, core, ctx, cancel)
|
||||||
sessPtr = sess
|
sessPtr = sess
|
||||||
sess.remote = remoteTarget
|
sess.remote = remoteTarget
|
||||||
|
|
|
||||||
|
|
@ -28,6 +28,7 @@ func FileMode(name string) (os.FileMode, bool) {
|
||||||
|
|
||||||
// ManagerConfig holds the dependencies for a Manager.
|
// ManagerConfig holds the dependencies for a Manager.
|
||||||
type ManagerConfig struct {
|
type ManagerConfig struct {
|
||||||
|
Ctx context.Context // daemon context — sessions derive from this
|
||||||
AgentsDir string
|
AgentsDir string
|
||||||
SessionsDir string
|
SessionsDir string
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -137,7 +137,7 @@ func (s *Manager) restoreSession(ps *agent.PersistedAgent) error {
|
||||||
remoteTarget := ps.Remote
|
remoteTarget := ps.Remote
|
||||||
|
|
||||||
if remoteTarget != "" {
|
if remoteTarget != "" {
|
||||||
rsrv, dialErr := execute.RemoteDial(context.Background(), execute.RemoteConfig{
|
rsrv, dialErr := execute.RemoteDial(s.cfg.Ctx, execute.RemoteConfig{
|
||||||
SSHTarget: remoteTarget,
|
SSHTarget: remoteTarget,
|
||||||
CWD: cwd,
|
CWD: cwd,
|
||||||
})
|
})
|
||||||
|
|
@ -237,7 +237,7 @@ func (s *Manager) restoreSession(ps *agent.PersistedAgent) error {
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(s.cfg.Ctx)
|
||||||
sess := NewSession(sessID, core, ctx, cancel)
|
sess := NewSession(sessID, core, ctx, cancel)
|
||||||
sessPtr = sess
|
sessPtr = sess
|
||||||
sess.uname = uname
|
sess.uname = uname
|
||||||
|
|
|
||||||
Reference in New Issue