From 346c68eea5a4b202f3a14f43d6c61d50b4231ead Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Fri, 30 Oct 2026 20:52:52 -0700 Subject: [PATCH] fix(orphans): reclaim dead helper trees that leaked behind dead sessions Leaked Codex plugin broker trees (up to 300 processes each) were never offered for cleanup. Several rules each blocked them: - Any non-zero CPU sample counted as work, so idle MCP servers ticking timers and GC hid every tree. A member now counts as idle below 5% of one core and the whole tree below 25%, in both detection and the kill-time recheck. - A process from a live app's bundle protected its tree even when an agent helper only borrowed the bundle's runtime (ChatGPT.app's cua_node under codex app-server). It still protects the tree unless its parent is a known agent helper. - macOS lsof prints a Unix socket's own address in field d, not D, so no peer ever resolved and every connection, even to the tree's own children, read as external. The probe now requests and parses d. - Socket probes now report a PeerState: none, listening, system (peers only another user owns), or connected. A peer shown as (none) has closed, and a bound UDP socket on Linux counts as connected, as on Windows. - Signatures can set idle_without_clients. A flagged helper's own listeners no longer keep its tree, and under a flagged root neither do system-service sockets. Other processes' listeners, connections to the user's processes and network connections still do. The Codex broker and claude mcp serve set it, since the plugin starts a new broker when the old endpoint stops answering. - On macOS, PeerResolver takes the tree's Unix sockets from one lsof pass over all of the user's processes, so a listener's clients (lsof shows an accepted connection only on the client's side) and the owners of outside peers come from one snapshot. It runs only for trees with a flagged helper and fails closed on any lsof error. - The macOS probe deadline scales with the number of 200-pid batches. - The TUI kill path takes the configured signature set instead of a second copy of the closure. --- docs/usage.md | 19 ++ internal/actions/kill.go | 10 +- internal/cli/act_test.go | 8 +- internal/cli/top.go | 4 + .../orphan_socket_verification_test.go | 7 +- internal/collector/pass.go | 2 +- internal/collector/sources.go | 21 ++- internal/derive/orphans.go | 70 ++++++- internal/derive/orphans_test.go | 109 ++++++++++- internal/derive/windows_orphans_test.go | 4 +- internal/kill/kill.go | 43 +++-- internal/kill/kill_test.go | 8 +- internal/kill/verify.go | 67 +++++-- internal/kill/verify_test.go | 101 +++++++++- internal/kill/windows_plan_test.go | 2 +- internal/model/model.go | 10 + internal/platform/fake.go | 29 ++- internal/platform/peers.go | 176 ++++++++++++------ internal/platform/platform.go | 20 +- internal/platform/platform_darwin.go | 78 ++++++-- internal/platform/platform_darwin_test.go | 24 ++- internal/platform/platform_test.go | 108 ++++++++++- internal/platform/procfs.go | 20 +- internal/platform/procfs_test.go | 20 +- internal/platform/sockets_windows.go | 12 +- internal/platform/sockets_windows_test.go | 4 +- internal/platform/winsockets.go | 27 ++- internal/platform/winsockets_test.go | 18 +- internal/signatures/signatures.go | 5 + internal/signatures/signatures.toml | 5 + internal/tui/app.go | 22 ++- internal/viewer/startup_test.go | 8 +- 32 files changed, 872 insertions(+), 189 deletions(-) diff --git a/docs/usage.md b/docs/usage.md index b272e8f9..79fa5fd5 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -107,6 +107,23 @@ reported at the top level. A single-host response omits `machines`. Consumers should read the top-level primary report plus any additional `machines` entries. Summary reads only the latest health sample and does not build machine trends. +An orphan tree counts as idle when every process stays below 5% of one core, +the whole tree stays below 25%, and no process has a socket peer outside the +tree. Dead helpers still tick timers and garbage collection, so exact-zero CPU +is not required. Listening sockets and connections to other users' processes, +such as system services, also keep a tree. A helper whose signature sets +`idle_without_clients` is the exception: its own listeners do not count, and +when it is the tree root, neither do system-service connections in the tree. +A connection to another of your processes, any network connection, and any +other process's listener, such as a dev server a child started, still keep the +tree. The Codex plugin's `app-server-broker.mjs` and `claude mcp serve` set it, +because their clients start a replacement or use stdio. A peer that has already +closed does not count. On macOS, telling a listener's clients from system +services takes one extra `lsof` pass over all your processes, run only for +trees with such a helper; if it fails, every socket counts as in use. A process +that runs from a running app's bundle protects its tree, unless an agent helper +started it to borrow the bundle's runtime. + Remote orphans cannot be killed here; run the kill command on their machine. Remote inbox acknowledgments require the peer's `--allow-ack` grant; unreachable peers show `ack pending` and retry on reconnect. @@ -922,6 +939,8 @@ name = "example-helper" agent = "codex" role = "helper" match = ['^example-agent-helper([[:space:]]|$)'] +# Optional: when this helper roots a tree, only real clients keep the tree. +# idle_without_clients = true ``` Extra files are read in configuration order before built-in signatures; first diff --git a/internal/actions/kill.go b/internal/actions/kill.go index 5160b599..610739f5 100644 --- a/internal/actions/kill.go +++ b/internal/actions/kill.go @@ -12,12 +12,18 @@ import ( "github.com/labstack/driga/internal/derive" "github.com/labstack/driga/internal/kill" "github.com/labstack/driga/internal/model" + "github.com/labstack/driga/internal/signatures" ) type KillOptions struct{ All, Force bool } func (c KillOptions) Kill(ctx context.Context, g Environment, cfg config.Config, procs []model.Proc, candidates []model.Orphan) error { idx := derive.NewProcIndex(procs) + sigs, err := signatures.Load(cfg.ExtraSignatures) + if err != nil { + return err + } + classified := kill.WithClassifier(derive.NewClassifier(sigs, len(procs))) var errs []error for _, o := range candidates { if o.Young { @@ -50,7 +56,7 @@ func (c KillOptions) Kill(ctx context.Context, g Environment, cfg config.Config, if _, err := fmt.Fprintf(g.Stdout, "%s: checking it stays idle for %s…\n", o.ID, g.VerifyWindow); err != nil { return err } - ok, reason, err := kill.VerifyIdle(ctx, g.Platform(), o, expect, g.VerifyWindow) + ok, reason, err := kill.VerifyIdle(ctx, g.Platform(), o, expect, g.VerifyWindow, classified) if err == nil && !ok { err = errors.New("not idle: " + reason) } @@ -106,7 +112,7 @@ func (c KillOptions) Kill(ctx context.Context, g Environment, cfg config.Config, return err } } - result, killErr := kill.Tree(ctx, g.Platform(), o, expect, c.Force, 5*time.Second) + result, killErr := kill.Tree(ctx, g.Platform(), o, expect, c.Force, 5*time.Second, classified) if result.Aborted { killErr = fmt.Errorf("kill %s aborted: %s", o.ID, result.Reason) } diff --git a/internal/cli/act_test.go b/internal/cli/act_test.go index fe639d0c..6ae79ae2 100644 --- a/internal/cli/act_test.go +++ b/internal/cli/act_test.go @@ -53,10 +53,12 @@ func (p *cliSeqPlatform) Machine(context.Context) (model.MachineSample, error) { } func (p *cliSeqPlatform) Cwds(context.Context, []int32) (map[int32]string, error) { return nil, nil } func (p *cliSeqPlatform) Alive(int32, time.Time) bool { return false } -func (p *cliSeqPlatform) ExternalPeers(context.Context, []int32) (map[int32]bool, error) { - return p.peers, nil +func (p *cliSeqPlatform) ExternalPeers(context.Context, []int32) (map[int32]model.PeerState, error) { + return platform.FixturePeers(p.peers), nil } +var _ platform.SocketProber = (*cliSeqPlatform)(nil) + // Snapshot mode never accumulates two passes, so an unverified candidate is // verified on demand instead of being refused outright. func TestKillVerifiesUnverifiedOnDemand(t *testing.T) { @@ -74,7 +76,7 @@ func TestKillVerifiesUnverifiedOnDemand(t *testing.T) { prompt bool want string }{ - {"busy", &cliSeqPlatform{snaps: [][]model.Proc{tree(1), tree(2)}}, false, "not idle: used 1.00s CPU"}, + {"busy", &cliSeqPlatform{snaps: [][]model.Proc{tree(1), tree(2)}}, false, "not idle: pid 4242 used 1.00s CPU"}, {"connected", &cliSeqPlatform{snaps: [][]model.Proc{tree(1)}, peers: map[int32]bool{4242: true}}, false, "not idle: pid 4242 has a connection outside the tree"}, {"quiet", &cliSeqPlatform{snaps: [][]model.Proc{tree(1)}}, true, "verified idle"}, } { diff --git a/internal/cli/top.go b/internal/cli/top.go index 31d9183c..2353cc13 100644 --- a/internal/cli/top.go +++ b/internal/cli/top.go @@ -11,6 +11,7 @@ import ( "github.com/labstack/driga/internal/chat" "github.com/labstack/driga/internal/cleanup" "github.com/labstack/driga/internal/config" + "github.com/labstack/driga/internal/signatures" "github.com/labstack/driga/internal/tui" "github.com/labstack/driga/internal/view" ) @@ -66,6 +67,9 @@ func (c TopCmd) Run(g *Globals) error { opts.Tmux = os.Getenv("TMUX") opts.JumpTerminal = cfg.Inbox.JumpTerminal opts.MotionOff = cfg.Motion == "off" + if opts.Signatures, err = signatures.Load(cfg.ExtraSignatures); err != nil { + return err + } opts.CleanupList = func(ctx context.Context) ([]cleanup.Candidate, error) { envelope, err := cleanupRead(ctx, g, "", false, true) if err != nil { diff --git a/internal/collector/orphan_socket_verification_test.go b/internal/collector/orphan_socket_verification_test.go index f032b1f5..7f701a11 100644 --- a/internal/collector/orphan_socket_verification_test.go +++ b/internal/collector/orphan_socket_verification_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/labstack/driga/internal/model" + "github.com/labstack/driga/internal/platform" "github.com/labstack/driga/internal/spool" ) @@ -16,9 +17,11 @@ type socketPlatform struct { requested []int32 } -func (p *socketPlatform) ExternalPeers(_ context.Context, pids []int32) (map[int32]bool, error) { +var _ platform.SocketProber = (*socketPlatform)(nil) + +func (p *socketPlatform) ExternalPeers(_ context.Context, pids []int32) (map[int32]model.PeerState, error) { p.requested = append([]int32(nil), pids...) - return p.peers, nil + return platform.FixturePeers(p.peers), nil } func TestPassVerifiesOrphansViaSocketProber(t *testing.T) { diff --git a/internal/collector/pass.go b/internal/collector/pass.go index 7d6a72c5..ee49e622 100644 --- a/internal/collector/pass.go +++ b/internal/collector/pass.go @@ -481,7 +481,7 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err orphanOpts = append(orphanOpts, derive.WithSubreaper(collected.subreaper)) } orphans := derive.FindOrphans(idx, ss, d.Sigs, now, d.Cfg.OrphanMinAge, prev, orphanOpts...) - var peers map[int32]bool + var peers map[int32]model.PeerState if _, ok := d.Plat.(platform.SocketProber); ok && p.Sockets && len(orphans) > 0 { peers = collected.peers if e := source("sockets", collected.socketErr); e != nil { diff --git a/internal/collector/sources.go b/internal/collector/sources.go index 6cc77109..b92649bd 100644 --- a/internal/collector/sources.go +++ b/internal/collector/sources.go @@ -72,7 +72,7 @@ type sourceResults struct { gitErr error feedEnabled bool spool spoolResult - peers map[int32]bool + peers map[int32]model.PeerState socketErr error subreaper int32 paths map[string]string @@ -622,7 +622,7 @@ func spoolHooks(d Deps, events []model.HookEvent) []model.HookEvent { return hooks } -func collectSocketPeers(ctx context.Context, d Deps, st *store.Store, state *State, procs []model.Proc, ss []model.Session, procAt, now time.Time, subreaper int32, hooks []model.HookEvent, aliases map[int64]int) (map[int32]bool, error) { +func collectSocketPeers(ctx context.Context, d Deps, st *store.Store, state *State, procs []model.Proc, ss []model.Session, procAt, now time.Time, subreaper int32, hooks []model.HookEvent, aliases map[int64]int) (map[int32]model.PeerState, error) { probe, ok := d.Plat.(platform.SocketProber) if !ok { return nil, nil @@ -695,16 +695,29 @@ func collectSocketPeers(ctx context.Context, d Deps, st *store.Store, state *Sta memory[key] = id } } - candidates := derive.FindOrphans(idx, sessions, d.Sigs, now, d.Cfg.OrphanMinAge, prev, derive.WithSpawnMemory(memory), derive.WithSubreaper(subreaper), derive.WithClassifier(d.classifier)) + // FindOrphans waives listeners by signature, so the probe must resolve + // clients with the same classifier. + classifier := d.classifier + if classifier == nil && d.Sigs != nil { + classifier = derive.NewClassifier(d.Sigs, len(procs)) + } + candidates := derive.FindOrphans(idx, sessions, d.Sigs, now, d.Cfg.OrphanMinAge, prev, derive.WithSpawnMemory(memory), derive.WithSubreaper(subreaper), derive.WithClassifier(classifier)) var pids []int32 + var members []model.Proc for _, o := range candidates { pids = append(pids, o.PIDs...) + for _, pid := range o.PIDs { + if p, ok := idx.Get(pid); ok { + members = append(members, p) + } + } } + resolve := derive.NeedsResolvedPeers(classifier, members) if len(pids) == 0 { return nil, nil } sort.Slice(pids, func(i, j int) bool { return pids[i] < pids[j] }) - peers, err := probe.ExternalPeers(ctx, pids) + peers, err := platform.ProbePeers(ctx, probe, pids, resolve) if err != nil { return nil, err } diff --git a/internal/derive/orphans.go b/internal/derive/orphans.go index eeac4e4a..8c429590 100644 --- a/internal/derive/orphans.go +++ b/internal/derive/orphans.go @@ -11,11 +11,57 @@ import ( "github.com/labstack/driga/internal/signatures" ) +// BackgroundCPUPercent is the per-process share of one core below which an +// orphan tree member counts as idle, and BackgroundTreeCPUPercent bounds the +// whole tree. Dead helpers still tick timers and GC far below these; real work +// runs far above them. External peers stay the liveness proof for work that +// waits on the network. +const ( + BackgroundCPUPercent = 5.0 + BackgroundTreeCPUPercent = 25.0 +) + +// IdleWithoutClients reports whether p's signature sets IdleWithoutClients. +func IdleWithoutClients(c *Classifier, p model.Proc) bool { + if c == nil { + return false + } + sig, ok := c.Match(p) + return ok && sig.IdleWithoutClients +} + +// NeedsResolvedPeers reports whether InUse may waive a socket in tree, so its +// probe must tell a listener's clients and system services apart +// (platform.PeerResolver). +func NeedsResolvedPeers(c *Classifier, tree []model.Proc) bool { + for _, p := range tree { + if IdleWithoutClients(c, p) { + return true + } + } + return false +} + +// InUse reports whether member p's socket state shows its tree still serves +// someone. A listener counts unless p's own signature sets IdleWithoutClients; +// a system-service socket counts unless the tree root's does. +func InUse(c *Classifier, root, p model.Proc, state model.PeerState) bool { + switch state { + case model.PeerConnected: + return true + case model.PeerListening: + return !IdleWithoutClients(c, p) + case model.PeerSystem: + return !IdleWithoutClients(c, root) + } + return false +} + // OrphanOption supplies remembered observations from earlier process passes. type OrphanOption func(*orphanOptions) type orphanOptions struct { spawnMemory map[string]int64 - externalPeers map[int32]bool + externalPeers map[int32]model.PeerState subreaper int32 classifier *Classifier } @@ -38,7 +84,7 @@ func WithSubreaper(pid int32) OrphanOption { // Collectors must probe every tree pid each process pass, pass only successful // results here, and use the immediately preceding pass as prev. Missing results // leave candidates unverified. Verified also needs persistence in stored views. -func WithExternalPeers(peers map[int32]bool) OrphanOption { +func WithExternalPeers(peers map[int32]model.PeerState) OrphanOption { return func(o *orphanOptions) { o.externalPeers = peers } } @@ -147,13 +193,15 @@ func FindOrphans(idx ProcIndex, sessions []model.Session, sigs *signatures.Set, idle := true probed := true var matched string + var treeCPU float64 // visit applies the idle checks to one tree member; false stops the walk. visit := func(p model.Proc) bool { if protected[p.PID] { idle = false return false } - if p.CPUPercent != 0 || options.externalPeers[p.PID] { + treeCPU += p.CPUPercent + if p.CPUPercent >= BackgroundCPUPercent || treeCPU >= BackgroundTreeCPUPercent || InUse(classifier, r, p, options.externalPeers[p.PID]) { idle = false return false } @@ -164,7 +212,9 @@ func FindOrphans(idx ProcIndex, sessions []model.Session, sigs *signatures.Set, idle = false return false } - if bundle := orphanBundle(p); bundle != "" && apps[bundle] { + // A process from a running app's bundle belongs to that app, unless an + // agent helper started it to borrow the bundle's runtime. + if bundle := orphanBundle(p); bundle != "" && apps[bundle] && (p.PID == pid || !agentHelper(classifier, idx, p.PPID)) { idle = false return false } @@ -236,7 +286,7 @@ func FindOrphans(idx ProcIndex, sessions []model.Session, sigs *signatures.Set, } o.ProcCount = len(o.PIDs) if probed { - const idleEvidence = "no CPU or external peers" + const idleEvidence = "background CPU only, no external peers" o.Evidence = append(o.Evidence, idleEvidence) o.Verified = existed && old.StartTime.Equal(o.StartTime) && slices.Equal(old.PIDs, o.PIDs) && slices.Contains(old.Evidence, idleEvidence) @@ -280,6 +330,16 @@ func FindOrphans(idx ProcIndex, sessions []model.Session, sigs *signatures.Set, return result } +// agentHelper reports whether pid is a known agent helper, host or headless run. +func agentHelper(c *Classifier, idx ProcIndex, pid int32) bool { + parent, ok := idx.Get(pid) + if !ok || c == nil { + return false + } + sig, ok := c.Match(parent) + return ok && (sig.Role == signatures.RoleHelper || sig.Role == signatures.RoleHost || sig.Role == signatures.RoleHeadless) +} + func orphanBundle(p model.Proc) string { if bundle, ok := signatures.BundlePath(p.Cmdline); ok { return bundle diff --git a/internal/derive/orphans_test.go b/internal/derive/orphans_test.go index efa4092c..edcd17aa 100644 --- a/internal/derive/orphans_test.go +++ b/internal/derive/orphans_test.go @@ -33,7 +33,7 @@ func TestOrphanFirstSightingUnverified(t *testing.T) { func TestOrphanSecondIdlePassVerified(t *testing.T) { procs, sigs, now := orphanFixture(t) - opts := WithExternalPeers(map[int32]bool{10: false, 11: false, 12: false}) + opts := WithExternalPeers(map[int32]model.PeerState{10: model.PeerNone, 11: model.PeerNone, 12: model.PeerNone}) first := FindOrphans(unixOrphanIndex(procs), nil, sigs, now, 0, nil, opts) if len(first) != 1 || first[0].Verified { t.Fatal(first) @@ -49,7 +49,7 @@ func TestOrphanWithExternalPeersIsNotOrphan(t *testing.T) { procs, sigs, now := orphanFixture(t) for _, p := range procs { t.Run(fmt.Sprint(p.PID), func(t *testing.T) { - if got := FindOrphans(unixOrphanIndex(procs), nil, sigs, now, 0, nil, WithExternalPeers(map[int32]bool{p.PID: true})); len(got) != 0 { + if got := FindOrphans(unixOrphanIndex(procs), nil, sigs, now, 0, nil, WithExternalPeers(map[int32]model.PeerState{p.PID: model.PeerConnected})); len(got) != 0 { t.Fatal(got) } }) @@ -58,7 +58,7 @@ func TestOrphanWithExternalPeersIsNotOrphan(t *testing.T) { func TestOrphanVerificationRequiresConsecutiveProbedIdlePasses(t *testing.T) { procs, sigs, now := orphanFixture(t) - clear := WithExternalPeers(map[int32]bool{10: false, 11: false, 12: false}) + clear := WithExternalPeers(map[int32]model.PeerState{10: model.PeerNone, 11: model.PeerNone, 12: model.PeerNone}) first := FindOrphans(unixOrphanIndex(procs), nil, sigs, now, 0, nil, clear) prev := map[string]model.Orphan{first[0].ID: first[0]} // An unknown probe result invalidates the observation window, even if CPU @@ -73,13 +73,13 @@ func TestOrphanVerificationRequiresConsecutiveProbedIdlePasses(t *testing.T) { t.Fatalf("unprobed previous pass verified: %+v", again) } // A partially probed tree cannot establish an idle observation either. - partial := FindOrphans(unixOrphanIndex(procs), nil, sigs, now.Add(time.Minute), 0, map[string]model.Orphan{first[0].ID: first[0]}, WithExternalPeers(map[int32]bool{10: false})) + partial := FindOrphans(unixOrphanIndex(procs), nil, sigs, now.Add(time.Minute), 0, map[string]model.Orphan{first[0].ID: first[0]}, WithExternalPeers(map[int32]model.PeerState{10: model.PeerNone})) if len(partial) != 1 || partial[0].Verified { t.Fatal(partial) } // A newly added descendant has no preceding idle observation. procs = append(procs, model.Proc{PID: 13, PPID: 10, StartTime: now}) - changed := FindOrphans(unixOrphanIndex(procs), nil, sigs, now.Add(time.Minute), 0, map[string]model.Orphan{first[0].ID: first[0]}, WithExternalPeers(map[int32]bool{10: false, 11: false, 12: false, 13: false})) + changed := FindOrphans(unixOrphanIndex(procs), nil, sigs, now.Add(time.Minute), 0, map[string]model.Orphan{first[0].ID: first[0]}, WithExternalPeers(map[int32]model.PeerState{10: model.PeerNone, 11: model.PeerNone, 12: model.PeerNone, 13: model.PeerNone})) if len(changed) != 1 || changed[0].Verified { t.Fatal(changed) } @@ -157,13 +157,31 @@ func TestNotOrphanWhenBusy(t *testing.T) { procs, sigs, now := orphanFixture(t) for i := range procs { p := append([]model.Proc(nil), procs...) - p[i].CPUPercent = 0.1 + p[i].CPUPercent = BackgroundCPUPercent if got := FindOrphans(unixOrphanIndex(p), nil, sigs, now, 0, nil); len(got) != 0 { t.Fatal(got) } } } +// Idle MCP servers tick timers and GC; that noise must not hide a dead tree. +func TestOrphanToleratesBackgroundCPU(t *testing.T) { + procs, sigs, now := orphanFixture(t) + for i := range procs { + procs[i].CPUPercent = 0.05 + } + procs[2].CPUPercent = 1.8 + opts := WithExternalPeers(map[int32]model.PeerState{10: model.PeerNone, 11: model.PeerNone, 12: model.PeerNone}) + first := FindOrphans(unixOrphanIndex(procs), nil, sigs, now, 0, nil, opts) + if len(first) != 1 { + t.Fatal(first) + } + prev := map[string]model.Orphan{first[0].ID: first[0]} + if second := FindOrphans(unixOrphanIndex(procs), nil, sigs, now.Add(time.Minute), 0, prev, opts); len(second) != 1 || !second[0].Verified { + t.Fatalf("background CPU blocked verification: %+v", second) + } +} + func TestNotOrphanAppBundle(t *testing.T) { procs, sigs, now := orphanFixture(t) procs = procs[:1] @@ -177,6 +195,83 @@ func TestNotOrphanAppBundle(t *testing.T) { } } +// A helper that only borrows a running app's bundled runtime does not make the +// app own the tree; Codex app-server starts ChatGPT.app's cua_node this way. +func TestOrphanDescendantBorrowsAppBundle(t *testing.T) { + procs, sigs, now := orphanFixture(t) + const cua = "/Applications/ChatGPT.app/Contents/Resources/cua_node/bin/node_repl" + procs = append(procs, model.Proc{PID: 13, PPID: 11, Cmdline: cua, Argv: []string{cua}, Exe: cua, StartTime: now.Add(-time.Hour)}) + app := model.Proc{PID: 20, PPID: 1, Exe: "/Applications/ChatGPT.app/Contents/MacOS/ChatGPT", Cmdline: "/Applications/ChatGPT.app/Contents/MacOS/ChatGPT", Argv: []string{"/Applications/ChatGPT.app/Contents/MacOS/ChatGPT"}} + got := FindOrphans(unixOrphanIndex(append(procs, app)), nil, sigs, now, 0, nil) + if len(got) != 1 || got[0].RootPID != 10 || got[0].ProcCount != 4 { + t.Fatal(got) + } +} + +// The Codex plugin replaces a broker whose socket stops answering, so a +// flagged helper's own listener does not keep its tree, and under a flagged +// root neither do system-service sockets. Any other listener, such as a dev +// server a child started, and any real client still keep it. +func TestOrphanIdleWithoutClients(t *testing.T) { + procs, sigs, now := orphanFixture(t) + none, listening, system, connected := model.PeerNone, model.PeerListening, model.PeerSystem, model.PeerConnected + for _, tc := range []struct { + name string + root string + peers map[int32]model.PeerState + want int + }{ + {"broker listens", "node /plugin/app-server-broker.mjs", map[int32]model.PeerState{10: listening, 11: none, 12: none}, 1}, + {"child talks to a system service", "node /plugin/app-server-broker.mjs", map[int32]model.PeerState{10: listening, 11: system, 12: listening}, 1}, + {"broker has a client", "node /plugin/app-server-broker.mjs", map[int32]model.PeerState{10: connected, 11: none, 12: none}, 0}, + {"child has a client", "node /plugin/app-server-broker.mjs", map[int32]model.PeerState{10: listening, 11: connected, 12: none}, 0}, + {"unflagged child listens", "node /plugin/app-server-broker.mjs", map[int32]model.PeerState{10: none, 11: listening, 12: none}, 0}, + {"unmarked root listens", "codex app-server", map[int32]model.PeerState{10: listening, 11: none, 12: none}, 0}, + {"unmarked root's child uses a service", "codex app-server", map[int32]model.PeerState{10: none, 11: system, 12: none}, 0}, + {"flagged child listens under unmarked root", "codex app-server", map[int32]model.PeerState{10: none, 11: none, 12: listening}, 1}, + } { + t.Run(tc.name, func(t *testing.T) { + p := append([]model.Proc(nil), procs...) + p[0].Cmdline, p[0].Argv = tc.root, strings.Fields(tc.root) + if got := FindOrphans(unixOrphanIndex(p), nil, sigs, now, 0, nil, WithExternalPeers(tc.peers)); len(got) != tc.want { + t.Fatal(got) + } + }) + } + // Without signatures every listener keeps the tree. + if !InUse(nil, procs[0], procs[0], model.PeerListening) || !InUse(nil, procs[0], procs[0], model.PeerSystem) { + t.Fatal("unclassified socket treated as idle") + } + c := NewClassifier(sigs, len(procs)) + if !NeedsResolvedPeers(c, procs) || NeedsResolvedPeers(c, procs[1:2]) || NeedsResolvedPeers(nil, procs) { + t.Fatal("resolved probe requirement") + } +} + +// Many workers each under the per-process limit can still add up to work. +func TestNotOrphanWhenTreeBusy(t *testing.T) { + procs, sigs, now := orphanFixture(t) + for i := 0; i < 8; i++ { + procs = append(procs, model.Proc{PID: int32(20 + i), PPID: 11, StartTime: now.Add(-time.Hour), Cmdline: "worker", Argv: []string{"worker"}, CPUPercent: BackgroundCPUPercent - 1}) + } + if got := FindOrphans(unixOrphanIndex(procs), nil, sigs, now, 0, nil); len(got) != 0 { + t.Fatal(got) + } +} + +// A running app's bundle process protects the tree when a non-agent wrapper +// started it, but not when an agent helper borrowed the runtime. +func TestOrphanAppBundleUnderWrapper(t *testing.T) { + procs, sigs, now := orphanFixture(t) + const helper = "/Applications/Foo.app/Contents/Helpers/foo-helper" + app := model.Proc{PID: 20, PPID: 1, Exe: "/Applications/Foo.app/Contents/MacOS/Foo", Cmdline: "/Applications/Foo.app/Contents/MacOS/Foo", Argv: []string{"/Applications/Foo.app/Contents/MacOS/Foo"}} + wrapper := model.Proc{PID: 13, PPID: 11, Cmdline: "sh -c foo-helper", Argv: []string{"sh", "-c", "foo-helper"}, StartTime: now.Add(-time.Hour)} + child := model.Proc{PID: 14, PPID: 13, Exe: helper, Cmdline: helper, Argv: []string{helper}, StartTime: now.Add(-time.Hour)} + if got := FindOrphans(unixOrphanIndex(append(procs, app, wrapper, child)), nil, sigs, now, 0, nil); len(got) != 0 { + t.Fatal(got) + } +} + func TestNotOrphanUnmatched(t *testing.T) { procs, sigs, now := orphanFixture(t) procs = procs[:1] @@ -254,7 +349,7 @@ func TestOrphanReparentedToSystemdUser(t *testing.T) { t.Fatalf("unrelated subreaper %d: %+v", pid, got) } } - opts := []OrphanOption{WithSubreaper(500), WithExternalPeers(map[int32]bool{10: false, 11: false, 12: false})} + opts := []OrphanOption{WithSubreaper(500), WithExternalPeers(map[int32]model.PeerState{10: model.PeerNone, 11: model.PeerNone, 12: model.PeerNone})} got := FindOrphans(unixOrphanIndex(procs), nil, sigs, now, 10*time.Minute, nil, opts...) if len(got) != 1 || got[0].ProcCount != 3 || got[0].Verified || !strings.Contains(strings.Join(got[0].Evidence, ";"), "reparented to systemd --user") { t.Fatalf("systemd orphan: %+v", got) diff --git a/internal/derive/windows_orphans_test.go b/internal/derive/windows_orphans_test.go index 8ec9fdd3..bd6dca3a 100644 --- a/internal/derive/windows_orphans_test.go +++ b/internal/derive/windows_orphans_test.go @@ -29,7 +29,7 @@ func TestWindowsParentIdentity(t *testing.T) { ps = append(ps, parent, model.Proc{PID: 600, PPID: 600, StartTime: now.Add(-2 * time.Hour)}) } idx := newProcIndex(ps, true) - got := FindOrphans(idx, nil, signatures.Builtin(), now, 0, nil, WithExternalPeers(map[int32]bool{701: false}), WithSubreaper(700)) + got := FindOrphans(idx, nil, signatures.Builtin(), now, 0, nil, WithExternalPeers(map[int32]model.PeerState{701: model.PeerNone}), WithSubreaper(700)) if (len(got) == 1) != tc.orphan { t.Fatalf("orphans: %+v", got) } @@ -46,7 +46,7 @@ func TestWindowsParentIdentity(t *testing.T) { func TestWindowsOrphanPinsEveryMember(t *testing.T) { now := time.Now() ps := []model.Proc{{PID: 701, PPID: 700, StartTime: now.Add(-time.Hour), Cmdline: `"C:\Tools\claude.exe" mcp serve`, Argv: []string{"C:\\Tools\\claude.exe", "mcp", "serve"}}, {PID: 702, PPID: 701, StartTime: now.Add(-time.Minute)}} - peers := WithExternalPeers(map[int32]bool{701: false, 702: false}) + peers := WithExternalPeers(map[int32]model.PeerState{701: model.PeerNone, 702: model.PeerNone}) first := FindOrphans(newProcIndex(ps, true), nil, signatures.Builtin(), now, 0, nil, peers) if len(first) != 1 || len(first[0].ProcessStarts) != 2 || first[0].Verified { t.Fatalf("first: %+v", first) diff --git a/internal/kill/kill.go b/internal/kill/kill.go index 52f0d8d4..573ea268 100644 --- a/internal/kill/kill.go +++ b/internal/kill/kill.go @@ -24,17 +24,38 @@ type Result struct { Reason string } +// Option adjusts how a tree is revalidated. +type Option func(*options) + +type options struct{ classifier *derive.Classifier } + +// WithClassifier lets the root's signature set IdleWithoutClients. Without it, +// any listener or system-service socket keeps the tree. +func WithClassifier(c *derive.Classifier) Option { + return func(o *options) { o.classifier = c } +} + +func collectOptions(opts []Option) options { + var o options + for _, opt := range opts { + if opt != nil { + opt(&o) + } + } + return o +} + // Tree checks identities and expected descendants, then attempts graceful // shutdown deepest first. A nonpositive wait uses the five-second default. -func Tree(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, force bool, wait time.Duration) (Result, error) { - return runTree(ctx, plat, o, expect, force, wait, signalIdentity, runtime.GOOS == "windows") +func Tree(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, force bool, wait time.Duration, opts ...Option) (Result, error) { + return runTree(ctx, plat, o, expect, force, wait, signalIdentity, runtime.GOOS == "windows", collectOptions(opts)) } -func tree(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, force bool, wait time.Duration, signal func(int, syscall.Signal) error) (Result, error) { - return runTree(ctx, plat, o, expect, force, wait, func(p model.Proc, sig syscall.Signal) error { return signal(int(p.PID), sig) }, false) +func tree(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, force bool, wait time.Duration, signal func(int, syscall.Signal) error, opts ...Option) (Result, error) { + return runTree(ctx, plat, o, expect, force, wait, func(p model.Proc, sig syscall.Signal) error { return signal(int(p.PID), sig) }, false, collectOptions(opts)) } -func runTree(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, force bool, wait time.Duration, signal func(model.Proc, syscall.Signal) error, windows bool) (Result, error) { +func runTree(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, force bool, wait time.Duration, signal func(model.Proc, syscall.Signal) error, windows bool, opt options) (Result, error) { abort := func(reason string) (Result, error) { return Result{Aborted: true, Reason: reason}, nil } if err := ctx.Err(); err != nil { return Result{}, err @@ -105,15 +126,15 @@ func runTree(ctx context.Context, plat platform.Platform, o model.Orphan, expect for _, p := range targets { pids = append(pids, p.PID) } - peers, err := prober.ExternalPeers(ctx, pids) + peers, err := platform.ProbePeers(ctx, prober, pids, derive.NeedsResolvedPeers(opt.classifier, targets)) if err != nil { return Result{Aborted: true, Reason: "socket peer check failed"}, fmt.Errorf("check socket peers: %w", err) } - for _, pid := range pids { - if active, present := peers[pid]; !present { - return abort(fmt.Sprintf("socket peer check missing PID %d", pid)) - } else if active { - return abort(fmt.Sprintf("PID %d has an external peer or listening socket", pid)) + for _, p := range targets { + if state, present := peers[p.PID]; !present { + return abort(fmt.Sprintf("socket peer check missing PID %d", p.PID)) + } else if derive.InUse(opt.classifier, root, p, state) { + return abort(peerReason(p.PID, state)) } } } diff --git a/internal/kill/kill_test.go b/internal/kill/kill_test.go index 231b17b0..1ebe2860 100644 --- a/internal/kill/kill_test.go +++ b/internal/kill/kill_test.go @@ -29,9 +29,11 @@ type socketPlatform struct { probed []int32 } -func (p *socketPlatform) ExternalPeers(_ context.Context, pids []int32) (map[int32]bool, error) { +var _ platform.SocketProber = (*socketPlatform)(nil) + +func (p *socketPlatform) ExternalPeers(_ context.Context, pids []int32) (map[int32]model.PeerState, error) { p.probed = append([]int32(nil), pids...) - return p.peers, p.err + return platform.FixturePeers(p.peers), p.err } func TestKillAbortsOnExternalPeer(t *testing.T) { @@ -47,7 +49,7 @@ func TestKillAbortsOnExternalPeer(t *testing.T) { }, peers: map[int32]bool{100: active == 100, 200: active == 200}} r, err := Tree(context.Background(), p, model.Orphan{RootPID: 100, StartTime: now, Verified: true}, []int32{200}, true, time.Millisecond) slices.Sort(p.probed) - if err != nil || !r.Aborted || !strings.Contains(r.Reason, "external peer") || len(r.Signaled) != 0 || aliveCalls != 0 || !slices.Equal(p.probed, []int32{100, 200}) { + if err != nil || !r.Aborted || !strings.Contains(r.Reason, "connection outside the tree") || len(r.Signaled) != 0 || aliveCalls != 0 || !slices.Equal(p.probed, []int32{100, 200}) { t.Fatalf("%+v %v probed=%v alive=%d", r, err, p.probed, aliveCalls) } }) diff --git a/internal/kill/verify.go b/internal/kill/verify.go index dc3df67c..5341ed8d 100644 --- a/internal/kill/verify.go +++ b/internal/kill/verify.go @@ -13,13 +13,15 @@ import ( // VerifyIdle checks, on demand, what the collector otherwise establishes over // two passes (spec §7.2): across window the tree keeps its identity and -// membership, accumulates no CPU time, and has no socket peer outside itself. +// membership, no process uses more than background CPU, and no process has a +// socket peer outside the tree. // It lets snapshot mode offer a kill without the background service. -func VerifyIdle(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, window time.Duration) (bool, string, error) { - return verifyIdle(ctx, plat, o, expect, window, runtime.GOOS == "windows") +func VerifyIdle(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, window time.Duration, opts ...Option) (bool, string, error) { + return verifyIdle(ctx, plat, o, expect, window, runtime.GOOS == "windows", opts...) } -func verifyIdle(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, window time.Duration, windows bool) (bool, string, error) { +func verifyIdle(ctx context.Context, plat platform.Platform, o model.Orphan, expect []int32, window time.Duration, windows bool, opts ...Option) (bool, string, error) { + opt := collectOptions(opts) prober, ok := plat.(platform.SocketProber) if !ok { return false, "cannot check sockets on this platform", nil @@ -39,35 +41,62 @@ func verifyIdle(ctx context.Context, plat platform.Platform, o model.Orphan, exp if err != nil || reason != "" { return false, reason, err } - // CPUSeconds has tick granularity; 10ms over the window is noise, not work. if windows { if reason := windowsTreeChanged(first.tree, second.tree); reason != "" { return false, reason, nil } } - if grew := second.cpu - first.cpu; grew > 0.01 { - return false, fmt.Sprintf("used %.2fs CPU in %s", grew, window), nil + // CPUSeconds has tick granularity, so each process gets 10ms of slack on + // top of the background share of the window, and the tree as a whole gets + // 10ms on top of its larger share. + budget := 0.01 + window.Seconds()*derive.BackgroundCPUPercent/100 + var total float64 + for pid, after := range second.cpu { + grew := after - first.cpu[pid] + if grew > budget { + return false, fmt.Sprintf("pid %d used %.2fs CPU in %s", pid, grew, window), nil + } + total += grew + } + if total > 0.01+window.Seconds()*derive.BackgroundTreeCPUPercent/100 { + return false, fmt.Sprintf("tree used %.2fs CPU in %s", total, window), nil + } + pids := make([]int32, 0, len(second.tree)) + for _, p := range second.tree { + pids = append(pids, p.PID) } - pids := append([]int32{o.RootPID}, expect...) - peers, err := prober.ExternalPeers(ctx, pids) + root := second.tree[len(second.tree)-1] + peers, err := platform.ProbePeers(ctx, prober, pids, derive.NeedsResolvedPeers(opt.classifier, second.tree)) if err != nil { return false, "", err } - for _, pid := range pids { - if _, present := peers[pid]; windows && !present { - return false, fmt.Sprintf("socket peer check missing PID %d", pid), nil + for _, p := range second.tree { + state, present := peers[p.PID] + if windows && !present { + return false, fmt.Sprintf("socket peer check missing PID %d", p.PID), nil } - if peers[pid] { - return false, fmt.Sprintf("pid %d has a connection outside the tree", pid), nil + if derive.InUse(opt.classifier, root, p, state) { + return false, peerReason(p.PID, state), nil } } return true, "", nil } -// verifySample returns the tree's total CPU seconds, or a reason it no longer -// matches the candidate. +// peerReason says why a tree member's socket state keeps the tree. +func peerReason(pid int32, state model.PeerState) string { + switch state { + case model.PeerListening: + return fmt.Sprintf("pid %d has a listening socket", pid) + case model.PeerSystem: + return fmt.Sprintf("pid %d is connected to a system service", pid) + } + return fmt.Sprintf("pid %d has a connection outside the tree", pid) +} + +// verifySample returns each tree member's CPU seconds, or a reason the tree no +// longer matches the candidate. type idleSample struct { - cpu float64 + cpu map[int32]float64 tree []model.Proc } @@ -89,12 +118,12 @@ func verifySample(ctx context.Context, plat platform.Platform, o model.Orphan, e if len(desc) != len(want) { return idleSample{}, "orphan descendant set changed", nil } - cpu := root.CPUSeconds + cpu := map[int32]float64{root.PID: root.CPUSeconds} for _, p := range desc { if !want[p.PID] { return idleSample{}, "orphan descendant set changed", nil } - cpu += p.CPUSeconds + cpu[p.PID] = p.CPUSeconds } return idleSample{cpu: cpu, tree: append(desc, root)}, "", nil } diff --git a/internal/kill/verify_test.go b/internal/kill/verify_test.go index ab341c0c..94333a14 100644 --- a/internal/kill/verify_test.go +++ b/internal/kill/verify_test.go @@ -2,11 +2,14 @@ package kill import ( "context" + "strings" "testing" "time" + "github.com/labstack/driga/internal/derive" "github.com/labstack/driga/internal/model" "github.com/labstack/driga/internal/platform" + "github.com/labstack/driga/internal/signatures" ) // seqPlatform returns successive process snapshots, one per Procs call. @@ -30,15 +33,17 @@ func (p *seqPlatform) Alive(int32, time.Time) bool { type seqProber struct{ *seqPlatform } -func (p seqProber) ExternalPeers(_ context.Context, pids []int32) (map[int32]bool, error) { - result := make(map[int32]bool) +var _ platform.SocketProber = seqProber{} + +func (p seqProber) ExternalPeers(_ context.Context, pids []int32) (map[int32]model.PeerState, error) { + result := make(map[int32]model.PeerState) if !p.probe { for _, pid := range pids { - result[pid] = false + result[pid] = model.PeerNone } } - for pid, active := range p.peers { - result[pid] = active + for pid, state := range platform.FixturePeers(p.peers) { + result[pid] = state } return result, nil } @@ -68,6 +73,20 @@ func TestVerifyIdleRejectsCPUGrowth(t *testing.T) { } } +func TestVerifyIdleBudgetsBackgroundCPUPerProcess(t *testing.T) { + // Each process ticks just under the background budget over the window. + o, snaps := verifyFixture(1.02) + snaps[1][0].CPUSeconds += 0.02 + ok, reason, err := VerifyIdle(context.Background(), seqProber{&seqPlatform{snaps: snaps}}, o, []int32{101}, 400*time.Millisecond) + if err != nil || !ok { + t.Fatalf("background ticks rejected: %v %q", err, reason) + } + o, snaps = verifyFixture(1.1) + if ok, _, err := VerifyIdle(context.Background(), seqProber{&seqPlatform{snaps: snaps}}, o, []int32{101}, 400*time.Millisecond); err != nil || ok { + t.Fatalf("busy child accepted: %v %v", ok, err) + } +} + func TestVerifyIdleRejectsExternalPeers(t *testing.T) { o, snaps := verifyFixture(1) p := seqProber{&seqPlatform{snaps: snaps, peers: map[int32]bool{101: true}}} @@ -104,3 +123,75 @@ func TestWindowsVerifyIdleRejectsReusedChildAndMissingProbe(t *testing.T) { } } } + +type listenerProber struct { + *seqPlatform + states map[int32]model.PeerState +} + +func (p listenerProber) ExternalPeers(context.Context, []int32) (map[int32]model.PeerState, error) { + return p.states, nil +} + +// Under a broker root, listeners and system-service sockets pass once the +// classifier sees its IdleWithoutClients signature; without one they do not. +func TestVerifyIdleListenerNeedsSignature(t *testing.T) { + sigs, err := signatures.Load(nil) + if err != nil { + t.Fatal(err) + } + broker := func() (model.Orphan, [][]model.Proc) { + o, snaps := verifyFixture(1) + for i := range snaps { + snaps[i][0].Cmdline = "node /plugin/app-server-broker.mjs" + snaps[i][0].Argv = []string{"node", "/plugin/app-server-broker.mjs"} + } + return o, snaps + } + states := map[int32]model.PeerState{100: model.PeerListening, 101: model.PeerNone} + o, snaps := broker() + if ok, reason, err := VerifyIdle(context.Background(), listenerProber{&seqPlatform{snaps: snaps}, states}, o, []int32{101}, time.Millisecond); err != nil || ok || reason != "pid 100 has a listening socket" { + t.Fatalf("unclassified listener accepted: %v %q %v", ok, reason, err) + } + o, snaps = broker() + classified := WithClassifier(derive.NewClassifier(sigs, 2)) + if ok, reason, err := VerifyIdle(context.Background(), listenerProber{&seqPlatform{snaps: snaps}, states}, o, []int32{101}, time.Millisecond, classified); err != nil || !ok { + t.Fatalf("idle broker listener rejected: %v %q", err, reason) + } + o, snaps = broker() + states[101] = model.PeerSystem + if ok, reason, err := VerifyIdle(context.Background(), listenerProber{&seqPlatform{snaps: snaps}, states}, o, []int32{101}, time.Millisecond, classified); err != nil || !ok { + t.Fatalf("child's system-service socket rejected: %v %q", err, reason) + } + o, snaps = broker() + states[101] = model.PeerConnected + if ok, reason, err := VerifyIdle(context.Background(), listenerProber{&seqPlatform{snaps: snaps}, states}, o, []int32{101}, time.Millisecond, classified); err != nil || ok || reason != "pid 101 has a connection outside the tree" { + t.Fatalf("child client accepted: %v %q %v", ok, reason, err) + } + o, snaps = verifyFixture(1) + states[101] = model.PeerNone + if ok, _, err := VerifyIdle(context.Background(), listenerProber{&seqPlatform{snaps: snaps}, states}, o, []int32{101}, time.Millisecond, classified); err != nil || ok { + t.Fatalf("unmarked root listener accepted: %v %v", ok, err) + } +} + +// Each process stays under its own budget, but together they are working. +func TestVerifyIdleRejectsBusyTree(t *testing.T) { + start := time.Unix(1700000000, 0) + var before, after []model.Proc + var expect []int32 + before = append(before, model.Proc{PID: 100, PPID: 1, StartTime: start, CPUSeconds: 1}) + after = append(after, model.Proc{PID: 100, PPID: 1, StartTime: start, CPUSeconds: 1}) + for i := int32(101); i <= 110; i++ { + expect = append(expect, i) + before = append(before, model.Proc{PID: i, PPID: 100, StartTime: start, CPUSeconds: 1}) + // 25ms each is under the 30ms per-process budget; 250ms in all is + // over the tree's 110ms. + after = append(after, model.Proc{PID: i, PPID: 100, StartTime: start, CPUSeconds: 1.025}) + } + o := model.Orphan{ID: "100:1700000000", RootPID: 100, StartTime: start, PIDs: append([]int32{100}, expect...)} + ok, reason, err := VerifyIdle(context.Background(), seqProber{&seqPlatform{snaps: [][]model.Proc{before, after}}}, o, expect, 400*time.Millisecond) + if err != nil || ok || !strings.HasPrefix(reason, "tree used") { + t.Fatalf("busy tree accepted: %v %q %v", ok, reason, err) + } +} diff --git a/internal/kill/windows_plan_test.go b/internal/kill/windows_plan_test.go index fc45de8c..2f1f8d5a 100644 --- a/internal/kill/windows_plan_test.go +++ b/internal/kill/windows_plan_test.go @@ -55,7 +55,7 @@ func TestWindowsTreeAbortsBeforeSignalAndDuringGrace(t *testing.T) { forced++ } return nil - }, true) + }, true, options{}) want := 0 if duringGrace { want = 2 diff --git a/internal/model/model.go b/internal/model/model.go index b4af02ea..9b8fbf26 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -247,6 +247,16 @@ type InboxJump struct { Server string `json:"server,omitempty"` } +// PeerState is what a socket probe saw for one process, weakest first. +type PeerState uint8 + +const ( + PeerNone PeerState = iota // No listener and no peer outside the tree. + PeerListening // Listening sockets only: reachable, not connected. + PeerSystem // Connected only to peers no process of this user owns. + PeerConnected // Connected to one of this user's processes, or the network. +) + // Orphan is a process tree left behind by a dead agent session. type Orphan struct { MachineID string `json:"machine_id"` diff --git a/internal/platform/fake.go b/internal/platform/fake.go index f825c802..4ba35d72 100644 --- a/internal/platform/fake.go +++ b/internal/platform/fake.go @@ -111,8 +111,26 @@ func (f *Fake) OpenFiles(ctx context.Context, path string) (bool, error) { return false, ctx.Err() } +// FixturePeers converts a readable test map, where true is a connected outside +// peer, into probe states. A nil map stays nil. +func FixturePeers(peers map[int32]bool) map[int32]model.PeerState { + if peers == nil { + return nil + } + states := make(map[int32]model.PeerState, len(peers)) + for pid, connected := range peers { + if connected { + states[pid] = model.PeerConnected + } else { + states[pid] = model.PeerNone + } + } + return states +} + // ExternalPeers returns optional fixture sockets/listeners and peer pid links. -func (f *Fake) ExternalPeers(ctx context.Context, pids []int32) (map[int32]bool, error) { +// Fixture sockets are unclassified, so they report connected. +func (f *Fake) ExternalPeers(ctx context.Context, pids []int32) (map[int32]model.PeerState, error) { if err := ctx.Err(); err != nil { return nil, err } @@ -120,12 +138,15 @@ func (f *Fake) ExternalPeers(ctx context.Context, pids []int32) (map[int32]bool, for _, pid := range pids { wanted[pid] = true } - result := make(map[int32]bool, len(pids)) + result := make(map[int32]model.PeerState, len(pids)) for _, pid := range pids { - result[pid] = f.sockets[pid] + result[pid] = model.PeerNone + if f.sockets[pid] { + result[pid] = model.PeerConnected + } for _, peer := range f.peers[pid] { if !wanted[peer] { - result[pid] = true + result[pid] = model.PeerConnected } } } diff --git a/internal/platform/peers.go b/internal/platform/peers.go index c7e55851..5df9e674 100644 --- a/internal/platform/peers.go +++ b/internal/platform/peers.go @@ -3,69 +3,56 @@ package platform import ( "strconv" "strings" + + "github.com/labstack/driga/internal/model" ) -// ParseLsofPeers parses lsof -FpcnDtT socket records for the given tree. -// An unresolved Unix peer is external; listening sockets also count as serving. -func ParseLsofPeers(out []byte, pids []int32) map[int32]bool { - type socket struct { - pid int32 - device, kind, name, state string - } +// ParseLsofPeers parses lsof -FpcdnDtT socket records for the given tree. +// An unresolved Unix peer is external; a peer lsof shows as (none) has closed. +// Listening and bound Unix sockets report PeerListening. +func ParseLsofPeers(out []byte, pids []int32) map[int32]model.PeerState { + return ParseLsofPeersAll(out, pids, nil) +} + +// ParseLsofPeersAll is ParseLsofPeers with all, the same records for every +// process lsof can inspect. An outside socket whose peer is a tree socket makes +// that tree process connected: macOS lists a connection a server accepted only +// by its bound path, so only the client's side shows it. An unresolved peer no +// listed process owns belongs to another user, such as a system service, and +// reports PeerSystem. Without all, every unresolved peer stays connected. +func ParseLsofPeersAll(out []byte, pids []int32, all []byte) map[int32]model.PeerState { wanted := make(map[int32]bool, len(pids)) - result := make(map[int32]bool, len(pids)) + result := make(map[int32]model.PeerState, len(pids)) for _, pid := range pids { if pid > 0 { - wanted[pid], result[pid] = true, false - } - } - var sockets []socket - var pid int32 - var current socket - flush := func() { - if current.pid > 0 { - sockets = append(sockets, current) - } - current = socket{pid: pid} - } - for _, line := range strings.Split(string(out), "\n") { - if len(line) < 2 { - continue - } - switch line[0] { - case 'p': - flush() - pid = 0 - if n, err := strconv.ParseInt(line[1:], 10, 32); err == nil && n > 0 { - pid = int32(n) - } - current = socket{pid: pid} - case 'f': - flush() - case 'D': - current.device = line[1:] - case 't': - current.kind = line[1:] - case 'n': - current.name = line[1:] - case 'T': - if state, ok := strings.CutPrefix(line[1:], "ST="); ok { - current.state = state - } + wanted[pid], result[pid] = true, model.PeerNone } } - flush() + sockets := parseLsofSockets(out) // Collect endpoint ownership before resolving peers: the peer's process // record may occur after the connecting process in lsof's output. owners := make(map[string][]int32) for _, s := range sockets { - if s.kind != "unix" && !strings.Contains(s.name, "->0x") && !strings.HasPrefix(s.name, "/") { - continue + for _, endpoint := range s.endpoints() { + owners[endpoint] = append(owners[endpoint], s.pid) } - local, _, _ := strings.Cut(s.name, "->") - for _, endpoint := range []string{s.device, local} { - if endpoint = socketEndpoint(endpoint); endpoint != "" { - owners[endpoint] = append(owners[endpoint], s.pid) + } + var visible map[string]bool + if all != nil { + visible = make(map[string]bool) + for _, s := range parseLsofSockets(all) { + for _, endpoint := range s.endpoints() { + visible[endpoint] = true + } + if wanted[s.pid] { + continue + } + if _, peer, ok := strings.Cut(s.name, "->"); ok { + for _, owner := range owners[socketEndpoint(peer)] { + if wanted[owner] { + result[owner] = model.PeerConnected + } + } } } } @@ -73,11 +60,18 @@ func ParseLsofPeers(out []byte, pids []int32) map[int32]bool { if !wanted[s.pid] { continue } - if s.state == "ESTABLISHED" || s.state == "LISTEN" || strings.Contains(s.name, "(ESTABLISHED)") || strings.Contains(s.name, "(LISTEN)") { - result[s.pid] = true + if s.state == "ESTABLISHED" || strings.Contains(s.name, "(ESTABLISHED)") { + result[s.pid] = model.PeerConnected + continue + } + if s.state == "LISTEN" || strings.Contains(s.name, "(LISTEN)") { + result[s.pid] = max(result[s.pid], model.PeerListening) continue } _, peer, connected := strings.Cut(s.name, "->") + if connected && socketEndpoint(peer) == "(none)" { + continue + } if connected { peerOwners := owners[socketEndpoint(peer)] external := len(peerOwners) == 0 @@ -87,11 +81,19 @@ func ParseLsofPeers(out []byte, pids []int32) map[int32]bool { } } if external { - result[s.pid] = true + // The unrestricted pass lists Unix sockets only, so a network + // peer always stays connected. + unixPeer := s.kind == "unix" || strings.HasPrefix(socketEndpoint(peer), "0x") || strings.HasPrefix(peer, "/") + if unixPeer && len(peerOwners) == 0 && visible != nil && !visible[socketEndpoint(peer)] { + // Another user's process, such as a system service. + result[s.pid] = max(result[s.pid], model.PeerSystem) + } else { + result[s.pid] = model.PeerConnected + } } } else if (s.kind == "unix" || strings.HasPrefix(s.name, "/")) && socketEndpoint(s.name) != "" { // A bound Unix socket with no connected endpoint is serving too. - result[s.pid] = true + result[s.pid] = max(result[s.pid], model.PeerListening) } } return result @@ -106,3 +108,67 @@ func socketEndpoint(name string) string { } return name } + +type lsofSocket struct { + pid int32 + device, kind, name, state string +} + +// endpoints returns the addresses a Unix socket is known by: its kernel +// address and any bound path. +func (s lsofSocket) endpoints() []string { + if s.kind != "unix" && !strings.Contains(s.name, "->0x") && !strings.HasPrefix(s.name, "/") { + return nil + } + local, _, _ := strings.Cut(s.name, "->") + var endpoints []string + for _, endpoint := range []string{s.device, local} { + if endpoint = socketEndpoint(endpoint); endpoint != "" { + endpoints = append(endpoints, endpoint) + } + } + return endpoints +} + +func parseLsofSockets(out []byte) []lsofSocket { + var sockets []lsofSocket + var pid int32 + var current lsofSocket + flush := func() { + if current.pid > 0 { + sockets = append(sockets, current) + } + current = lsofSocket{pid: pid} + } + for _, line := range strings.Split(string(out), "\n") { + if len(line) < 2 { + continue + } + switch line[0] { + case 'p': + flush() + pid = 0 + if n, err := strconv.ParseInt(line[1:], 10, 32); err == nil && n > 0 { + pid = int32(n) + } + current = lsofSocket{pid: pid} + case 'f': + flush() + case 'd', 'D': + // A Unix socket's kernel address; macOS lsof prints it as d. + if strings.HasPrefix(line[1:], "0x") { + current.device = line[1:] + } + case 't': + current.kind = line[1:] + case 'n': + current.name = line[1:] + case 'T': + if state, ok := strings.CutPrefix(line[1:], "ST="); ok { + current.state = state + } + } + } + flush() + return sockets +} diff --git a/internal/platform/platform.go b/internal/platform/platform.go index 07f6b519..c50a5042 100644 --- a/internal/platform/platform.go +++ b/internal/platform/platform.go @@ -35,9 +35,25 @@ func ProcessOS(p Platform) string { } // SocketProber optionally checks socket activity for a candidate process tree. -// Results include every requested pid; true means an external peer or listener. +// Results include every requested pid, with the strongest state seen for it. type SocketProber interface { - ExternalPeers(ctx context.Context, pids []int32) (map[int32]bool, error) + ExternalPeers(ctx context.Context, pids []int32) (map[int32]model.PeerState, error) +} + +// PeerResolver is a SocketProber that also tells a listener's clients and +// system services apart: it finds outside processes connected to a tree's +// listeners, and reports peers no process of this user owns as PeerSystem. +// It scans every process, so use it only when the answer can change. +type PeerResolver interface { + ResolvedPeers(ctx context.Context, pids []int32) (map[int32]model.PeerState, error) +} + +// ProbePeers uses ResolvedPeers when resolve is set and p offers it. +func ProbePeers(ctx context.Context, p SocketProber, pids []int32, resolve bool) (map[int32]model.PeerState, error) { + if r, ok := p.(PeerResolver); ok && resolve { + return r.ResolvedPeers(ctx, pids) + } + return p.ExternalPeers(ctx, pids) } // FileProber fails closed if open file descriptors cannot be inspected. diff --git a/internal/platform/platform_darwin.go b/internal/platform/platform_darwin.go index 7f5eb628..025e0044 100644 --- a/internal/platform/platform_darwin.go +++ b/internal/platform/platform_darwin.go @@ -180,10 +180,67 @@ func (darwinPlatform) Cwds(ctx context.Context, pids []int32) (map[int32]string, return cwds, nil } -// ExternalPeers checks Unix and Internet sockets within a five-second deadline. -func (darwinPlatform) ExternalPeers(ctx context.Context, pids []int32) (map[int32]bool, error) { - ctx, cancel := context.WithTimeout(ctx, 5*time.Second) +// lsofBatch bounds one lsof argument list; each batch gets lsofBatchDeadline. +// lsofAllDeadline bounds the unrestricted pass that ResolvedPeers adds. +const ( + lsofBatch = 200 + lsofBatchDeadline = 5 * time.Second + lsofAllDeadline = 10 * time.Second +) + +// ExternalPeers checks the tree's own Unix and Internet sockets within five +// seconds per batch of 200 pids, so large trees are not cut off on a loaded +// machine. +func (darwinPlatform) ExternalPeers(ctx context.Context, pids []int32) (map[int32]model.PeerState, error) { + ctx, cancel := lsofTreeDeadline(ctx, pids) + defer cancel() + records, err := lsofTree(ctx, pids, "-U", "-i") + if err != nil { + return nil, err + } + return ParseLsofPeers(records, pids), nil +} + +// ResolvedPeers takes the tree's Unix sockets from one lsof pass over all of +// this user's processes, so a listener's clients and the owners of outside +// peers come from the same snapshot as the tree. If that pass fails or is +// incomplete, every pid reports PeerConnected. +func (darwinPlatform) ResolvedPeers(ctx context.Context, pids []int32) (map[int32]model.PeerState, error) { + treeCtx, cancel := lsofTreeDeadline(ctx, pids) defer cancel() + inet, err := lsofTree(treeCtx, pids, "-i") + if err != nil { + return nil, err + } + allCtx, cancelAll := context.WithTimeout(ctx, lsofAllDeadline) + defer cancelAll() + all, err := exec.CommandContext(allCtx, "lsof", "-U", "-n", "-P", "-F", "pcdnDtT").Output() + if ctx.Err() != nil { + return nil, ctx.Err() + } + if err != nil || allCtx.Err() != nil { + // Exit 1 means lsof skipped something; a missing client must not + // read as idle. + result := make(map[int32]model.PeerState, len(pids)) + for _, pid := range pids { + if pid > 0 { + result[pid] = model.PeerConnected + } + } + return result, nil + } + return ParseLsofPeersAll(append(inet, all...), pids, all), nil +} + +func lsofTreeDeadline(ctx context.Context, pids []int32) (context.Context, context.CancelFunc) { + batches := max(1, (len(pids)+lsofBatch-1)/lsofBatch) + return context.WithTimeout(ctx, time.Duration(batches)*lsofBatchDeadline) +} + +// lsofTree lists the given socket classes of pids. With -a, -U and -i +// intersect instead of forming a union, so each class runs separately while +// keeping the pid restriction. +func lsofTree(ctx context.Context, pids []int32, classes ...string) ([]byte, error) { ids := make([]string, 0, len(pids)) for _, pid := range pids { if pid > 0 { @@ -191,16 +248,15 @@ func (darwinPlatform) ExternalPeers(ctx context.Context, pids []int32) (map[int3 } } var records bytes.Buffer - for offset := 0; offset < len(ids); offset += 200 { - // With -a, -U and -i intersect instead of forming a union. Query - // each socket class separately while retaining the pid restriction. - for _, class := range []string{"-U", "-i"} { + for offset := 0; offset < len(ids); offset += lsofBatch { + for _, class := range classes { if err := ctx.Err(); err != nil { return nil, err } - // D identifies each Unix socket's local address; t and T identify - // socket type and TCP state, which -Fpcn alone does not expose. - out, err := exec.CommandContext(ctx, "lsof", "-a", class, "-n", "-P", "-F", "pcnDtT", "-p", strings.Join(ids[offset:min(offset+200, len(ids))], ",")).Output() + // macOS lsof reports a Unix socket's own kernel address in d (some + // builds use D); t and T identify socket type and TCP state, which + // -Fpcn alone does not expose. + out, err := exec.CommandContext(ctx, "lsof", "-a", class, "-n", "-P", "-F", "pcdnDtT", "-p", strings.Join(ids[offset:min(offset+lsofBatch, len(ids))], ",")).Output() records.Write(out) if ctx.Err() != nil { return nil, ctx.Err() @@ -214,7 +270,7 @@ func (darwinPlatform) ExternalPeers(ctx context.Context, pids []int32) (map[int3 if err := ctx.Err(); err != nil { return nil, err } - return ParseLsofPeers(records.Bytes(), pids), nil + return records.Bytes(), nil } func (darwinPlatform) Alive(pid int32, start time.Time) bool { diff --git a/internal/platform/platform_darwin_test.go b/internal/platform/platform_darwin_test.go index aff6dd72..085a98e8 100644 --- a/internal/platform/platform_darwin_test.go +++ b/internal/platform/platform_darwin_test.go @@ -13,6 +13,7 @@ import ( "testing" "time" + "github.com/labstack/driga/internal/model" "github.com/labstack/driga/internal/platform" "github.com/shirou/gopsutil/v4/process" ) @@ -126,7 +127,7 @@ func TestCwdsCanceled(t *testing.T) { } func TestExternalPeers(t *testing.T) { - fakeCommand(t, "lsof", `[ "$#" -eq 8 ] && [ "$1" = -a ] && [ "$3" = -n ] && [ "$4" = -P ] && [ "$5" = -F ] && [ "$6" = pcnDtT ] && [ "$7" = -p ] && [ "$8" = 10,11 ] || exit 2 + fakeCommand(t, "lsof", `[ "$#" -eq 8 ] && [ "$1" = -a ] && [ "$3" = -n ] && [ "$4" = -P ] && [ "$5" = -F ] && [ "$6" = pcdnDtT ] && [ "$7" = -p ] && [ "$8" = 10,11 ] || exit 2 [ "$2" = -i ] && exit 1 [ "$2" = -U ] || exit 2 printf 'p10\nf3\ntunix\nD0xa\nn->0xb\np11\nf3\ntunix\nD0xb\nn->0xa\n' @@ -137,7 +138,7 @@ exit 1 t.Fatal("Darwin platform does not implement SocketProber") } got, err := prober.ExternalPeers(context.Background(), []int32{10, 11}) - if err != nil || len(got) != 2 || got[10] || got[11] { + if err != nil || len(got) != 2 || got[10] != model.PeerNone || got[11] != model.PeerNone { t.Fatalf("peers = %v, %v", got, err) } } @@ -164,7 +165,7 @@ func TestExternalPeersListenerSelf(t *testing.T) { pid := int32(os.Getpid()) prober := platform.New().(platform.SocketProber) peers, err := prober.ExternalPeers(context.Background(), []int32{pid}) - if err != nil || !peers[pid] { + if err != nil || peers[pid] != model.PeerListening { t.Fatalf("own listener missing: %v, %v", peers, err) } } @@ -291,3 +292,20 @@ func TestCwdsAllOwnProcessesQuickly(t *testing.T) { t.Fatalf("own cwd = %q, want %q", cwds[int32(os.Getpid())], wd) } } + +// If the unrestricted pass is incomplete, a client may be missing, so every +// pid reports connected rather than idle. +func TestResolvedPeersFailsClosed(t *testing.T) { + fakeCommand(t, "lsof", `[ "$2" = -i ] && exit 1 +[ "$1" = -U ] && { printf 'p10\nf3\ntunix\nd0xa\nn/private/tmp/broker.sock\n'; exit 1; } +exit 2 +`) + resolver, ok := platform.New().(platform.PeerResolver) + if !ok { + t.Fatal("Darwin platform does not implement PeerResolver") + } + got, err := resolver.ResolvedPeers(context.Background(), []int32{10, 11}) + if err != nil || got[10] != model.PeerConnected || got[11] != model.PeerConnected { + t.Fatalf("peers = %v, %v", got, err) + } +} diff --git a/internal/platform/platform_test.go b/internal/platform/platform_test.go index b88599f2..c60596c4 100644 --- a/internal/platform/platform_test.go +++ b/internal/platform/platform_test.go @@ -76,16 +76,114 @@ p19 cclient f3 n127.0.0.1:4001->127.0.0.1:5001 (ESTABLISHED) +p20 +cstdio +f0 +tunix +D0x7a825c5d47b4858d +n->(none) +p21 +cbroker +f12 +tunix +D0x00f1 +n->0x00f2 +f17 +tunix +D0x00f3 +n/private/tmp/cxc/broker.sock +p22 +capp-server +f0 +tunix +D0x00f2 +n->0x00f1 `) - pids := []int32{10, 11, 12, 13, 14, 15, 16, 17, 18, 19} - want := map[int32]bool{10: true, 11: false, 12: false, 13: false, 14: true, 15: true, 16: true, 17: true, 18: false, 19: true} + none, listening, connected := model.PeerNone, model.PeerListening, model.PeerConnected + pids := []int32{10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22} + want := map[int32]model.PeerState{10: connected, 11: none, 12: none, 13: none, 14: connected, 15: listening, 16: listening, 17: connected, 18: none, 19: connected, + // A peer that closed shows as (none); a broker whose only client is its + // own child just listens. + 20: none, 21: listening, 22: none} if got := platform.ParseLsofPeers(out, pids); !reflect.DeepEqual(got, want) { t.Fatalf("peers = %v; want %v", got, want) } // A peer whose record is present but is outside the target set is external. - if got := platform.ParseLsofPeers(out, []int32{11}); !got[11] || len(got) != 1 { + if got := platform.ParseLsofPeers(out, []int32{11}); got[11] != connected || len(got) != 1 { t.Fatalf("peer outside pid set: %v", got) } + // The broker's child connection leaves the tree when the child is not in it. + if got := platform.ParseLsofPeers(out, []int32{21}); got[21] != connected { + t.Fatalf("broker client outside pid set: %v", got) + } +} + +// Records in the shape macOS lsof -FpcdnDtT prints, taken from a live Codex +// plugin broker: d carries each Unix socket's own address, and a connection the +// broker accepted shows only the bound path, so only the client's side names it. +func TestParseLsofPeersAllMacOSBroker(t *testing.T) { + tree := []byte(`p98021 +cnode +f12 +tunix +d0xe0d058e2c7c8edfd +n->0x46e036a0d862d8ac +f17 +tunix +d0x98e16ac011dfaa79 +n/private/tmp/cxc/broker.sock +f18 +tunix +d0xa2e57851b5fd884c +n/private/tmp/cxc/broker.sock +p98025 +ccodex +f0 +tunix +d0x46e036a0d862d8ac +n->0xe0d058e2c7c8edfd +f34 +tunix +d0x7a4edd3e6fd350f +n->0x4abc21701ce138cc +`) + client := []byte(`p99984 +cnc +f3 +tunix +d0x7633b400310f9ee +n->0xa2e57851b5fd884c +`) + pids := []int32{98021, 98025} + // Without the unrestricted pass, an outside peer of unknown owner counts. + strict := platform.ParseLsofPeers(tree, pids) + if want := map[int32]model.PeerState{98021: model.PeerListening, 98025: model.PeerConnected}; !reflect.DeepEqual(strict, want) { + t.Fatalf("strict = %v; want %v", strict, want) + } + // No other process names 0x4abc...: it belongs to another user's service. + idle := platform.ParseLsofPeersAll(tree, pids, tree) + if want := map[int32]model.PeerState{98021: model.PeerListening, 98025: model.PeerSystem}; !reflect.DeepEqual(idle, want) { + t.Fatalf("no client = %v; want %v", idle, want) + } + busy := platform.ParseLsofPeersAll(tree, pids, append(append([]byte(nil), tree...), client...)) + if busy[98021] != model.PeerConnected { + t.Fatalf("accepted client missed: %v", busy) + } + // The unrestricted pass covers Unix sockets only; network peers it cannot + // list stay connected, whatever their TCP state. + network := []byte(`p98025 +ccodex +f40 +tIPv4 +n10.0.0.2:5353->1.1.1.1:53 +f41 +tIPv4 +n10.0.0.2:50000->140.82.112.3:443 +TST=CLOSE_WAIT +`) + if got := platform.ParseLsofPeersAll(network, []int32{98025}, tree); got[98025] != model.PeerConnected { + t.Fatalf("network peer downgraded: %v", got) + } } func TestFakeExternalPeers(t *testing.T) { @@ -108,11 +206,11 @@ func TestFakeExternalPeers(t *testing.T) { } var prober platform.SocketProber = f got, err := prober.ExternalPeers(context.Background(), []int32{10, 11}) - if err != nil || !got[10] || got[11] { + if err != nil || got[10] != model.PeerConnected || got[11] != model.PeerNone { t.Fatalf("peers = %v, %v", got, err) } got, err = prober.ExternalPeers(context.Background(), []int32{11}) - if err != nil || !got[11] { + if err != nil || got[11] != model.PeerConnected { t.Fatalf("outside peer = %v, %v", got, err) } } diff --git a/internal/platform/procfs.go b/internal/platform/procfs.go index e0caae6b..9b220013 100644 --- a/internal/platform/procfs.go +++ b/internal/platform/procfs.go @@ -15,6 +15,8 @@ import ( "strings" "github.com/shirou/gopsutil/v4/common" + + "github.com/labstack/driga/internal/model" ) // ParsePSI extracts memory "some avg10"; an absent some line returns nil. @@ -225,15 +227,15 @@ func (p procFS) Cwds(ctx context.Context, pids []int32) (map[int32]string, error return result, ctx.Err() } -func (p procFS) ExternalPeers(ctx context.Context, pids []int32) (map[int32]bool, error) { - result := make(map[int32]bool, len(pids)) +func (p procFS) ExternalPeers(ctx context.Context, pids []int32) (map[int32]model.PeerState, error) { + result := make(map[int32]model.PeerState, len(pids)) owned := make(map[uint64]bool) inodes := make(map[int32][]uint64, len(pids)) for _, pid := range pids { if err := ctx.Err(); err != nil { return nil, err } - result[pid] = false + result[pid] = model.PeerNone inodes[pid] = SocketInodes(p.root, pid) for _, inode := range inodes[pid] { owned[inode] = true @@ -273,14 +275,18 @@ func (p procFS) ExternalPeers(ctx context.Context, pids []int32) (map[int32]bool if !ok { continue } - if s.Listen { - result[pid] = true + if s.Listen && strings.HasPrefix(kind, "udp") { + // A bound UDP endpoint can exchange datagrams with anyone, + // as on Windows, so it counts as connected. + result[pid] = model.PeerConnected + } else if s.Listen { + result[pid] = max(result[pid], model.PeerListening) } else if kind == "unix" { if s.State == "03" && !owned[peers[inode]] { - result[pid] = true + result[pid] = model.PeerConnected } } else if s.State == "01" && !internal[endpointPair{s.Remote, s.Local}] { - result[pid] = true + result[pid] = model.PeerConnected } } } diff --git a/internal/platform/procfs_test.go b/internal/platform/procfs_test.go index a138621e..ac02fdae 100644 --- a/internal/platform/procfs_test.go +++ b/internal/platform/procfs_test.go @@ -9,6 +9,8 @@ import ( "reflect" "runtime" "testing" + + "github.com/labstack/driga/internal/model" ) func procFixture(t *testing.T) string { @@ -105,17 +107,17 @@ func TestLinuxExternalPeers(t *testing.T) { return map[uint64]uint64{201: 202, 202: 201} }} got, err := fs.ExternalPeers(context.Background(), []int32{10, 11, 99}) - want := map[int32]bool{10: false, 11: true, 99: false} + want := map[int32]model.PeerState{10: model.PeerNone, 11: model.PeerConnected, 99: model.PeerNone} if err != nil || !reflect.DeepEqual(got, want) { t.Fatalf("peers = %v, %v; want %v", got, err, want) } got, err = fs.ExternalPeers(context.Background(), []int32{10}) - if err != nil || !got[10] { + if err != nil || got[10] != model.PeerConnected { t.Fatalf("peer outside tree = %v, %v", got, err) } fs.unixPeers = nil got, err = fs.ExternalPeers(context.Background(), []int32{10, 11}) - if err != nil || !got[10] { + if err != nil || got[10] != model.PeerConnected { t.Fatalf("unresolved Unix peer = %v, %v", got, err) } // Opposite TCP endpoints owned by the tree are internal; listeners always serve. @@ -128,16 +130,24 @@ func TestLinuxExternalPeers(t *testing.T) { t.Fatal(err) } got, err = fs.ExternalPeers(context.Background(), []int32{10, 11}) - if err != nil || got[10] { + if err != nil || got[10] != model.PeerNone { t.Fatalf("internal TCP = %v, %v", got, err) } if err := os.Symlink("socket:[100]", filepath.Join(root, "10/fd/8")); err != nil { t.Fatal(err) } got, err = fs.ExternalPeers(context.Background(), []int32{10, 11}) - if err != nil || !got[10] { + if err != nil || got[10] != model.PeerListening { t.Fatalf("listener = %v, %v", got, err) } + // A bound UDP socket exchanges datagrams with anyone, as on Windows. + if err := os.Symlink("socket:[105]", filepath.Join(root, "10/fd/9")); err != nil { + t.Fatal(err) + } + got, err = fs.ExternalPeers(context.Background(), []int32{10, 11}) + if err != nil || got[10] != model.PeerConnected { + t.Fatalf("bound UDP = %v, %v", got, err) + } ctx, cancel := context.WithCancel(context.Background()) cancel() if _, err := fs.ExternalPeers(ctx, []int32{10}); !errors.Is(err, context.Canceled) { diff --git a/internal/platform/sockets_windows.go b/internal/platform/sockets_windows.go index c45d066c..973a905e 100644 --- a/internal/platform/sockets_windows.go +++ b/internal/platform/sockets_windows.go @@ -6,6 +6,8 @@ import ( "unsafe" "golang.org/x/sys/windows" + + "github.com/labstack/driga/internal/model" ) var ipHelper = windows.NewLazySystemDLL("iphlpapi.dll") @@ -14,11 +16,11 @@ var extendedUDP = ipHelper.NewProc("GetExtendedUdpTable") var _ SocketProber = winPlatform{} -func (p winPlatform) ExternalPeers(ctx context.Context, pids []int32) (map[int32]bool, error) { - result := make(map[int32]bool, len(pids)) +func (p winPlatform) ExternalPeers(ctx context.Context, pids []int32) (map[int32]model.PeerState, error) { + result := make(map[int32]model.PeerState, len(pids)) for _, pid := range pids { if pid > 0 { - result[pid] = false + result[pid] = model.PeerNone } } for _, family := range []int{2, 23} { @@ -34,8 +36,8 @@ func (p winPlatform) ExternalPeers(ctx context.Context, pids []int32) (map[int32 if err != nil { return nil, err } - for pid, active := range peers { - result[pid] = result[pid] || active + for pid, state := range peers { + result[pid] = max(result[pid], state) } } } diff --git a/internal/platform/sockets_windows_test.go b/internal/platform/sockets_windows_test.go index 3004eba4..bf7b4edc 100644 --- a/internal/platform/sockets_windows_test.go +++ b/internal/platform/sockets_windows_test.go @@ -5,6 +5,8 @@ import ( "net" "os" "testing" + + "github.com/labstack/driga/internal/model" ) func TestWindowsSocketOwnerTables(t *testing.T) { @@ -29,7 +31,7 @@ func TestWindowsSocketOwnerTables(t *testing.T) { } pid := int32(os.Getpid()) peers, err := New().(SocketProber).ExternalPeers(context.Background(), []int32{pid}) - if err != nil || !peers[pid] || len(peers) != 1 { + if err != nil || peers[pid] == model.PeerNone || len(peers) != 1 { t.Fatalf("listener ownership: %v %v", peers, err) } }) diff --git a/internal/platform/winsockets.go b/internal/platform/winsockets.go index b2b483f3..692e01cb 100644 --- a/internal/platform/winsockets.go +++ b/internal/platform/winsockets.go @@ -3,6 +3,8 @@ package platform import ( "encoding/binary" "fmt" + + "github.com/labstack/driga/internal/model" ) // parseWindowsSocketTable reads the fixed DWORD layouts from IP Helper, never @@ -10,7 +12,7 @@ import ( // evidence (even a loopback connection); every bound UDP endpoint is a listener. // UDP owner tables have no peer information, so treating a binding as active // is deliberately conservative. -func parseWindowsSocketTable(b []byte, family int, udp bool, pids []int32) (map[int32]bool, error) { +func parseWindowsSocketTable(b []byte, family int, udp bool, pids []int32) (map[int32]model.PeerState, error) { row, pidOffset, stateOffset := 24, 20, 0 switch { case family == 2 && udp: @@ -29,10 +31,10 @@ func parseWindowsSocketTable(b []byte, family int, udp bool, pids []int32) (map[ if n > uint64((len(b)-4)/row) { return nil, fmt.Errorf("truncated socket table rows") } - result := make(map[int32]bool, len(pids)) + result := make(map[int32]model.PeerState, len(pids)) for _, pid := range pids { if pid > 0 { - result[pid] = false + result[pid] = model.PeerNone } } for i := 0; i < int(n); i++ { @@ -41,14 +43,21 @@ func parseWindowsSocketTable(b []byte, family int, udp bool, pids []int32) (map[ if _, wanted := result[pid]; !wanted { continue } - active := udp + // A bound UDP endpoint can exchange datagrams with anyone, so it + // counts as connected. + seen := model.PeerConnected if !udp { - state := binary.LittleEndian.Uint32(r[stateOffset:]) - // CLOSED, TIME_WAIT and DELETE_TCB hold no live connection. All - // other (including unrecognized future) states fail toward active. - active = state != 1 && state != 11 && state != 12 + // CLOSED, TIME_WAIT and DELETE_TCB hold no live connection; LISTEN + // waits for one. All other (including unrecognized future) states + // fail toward connected. + switch binary.LittleEndian.Uint32(r[stateOffset:]) { + case 1, 11, 12: + seen = model.PeerNone + case 2: + seen = model.PeerListening + } } - result[pid] = result[pid] || active + result[pid] = max(result[pid], seen) } return result, nil } diff --git a/internal/platform/winsockets_test.go b/internal/platform/winsockets_test.go index 7d4d836f..96b1dd50 100644 --- a/internal/platform/winsockets_test.go +++ b/internal/platform/winsockets_test.go @@ -3,6 +3,8 @@ package platform import ( "encoding/hex" "testing" + + "github.com/labstack/driga/internal/model" ) // Fixed DWORD-layout byte fixtures: count followed by MIB_*ROW_OWNER_PID. Addresses @@ -13,14 +15,14 @@ func TestWindowsSocketTableBuffers(t *testing.T) { family int udp bool pid int32 - active bool + state model.PeerState }{ - {"tcp4 listen", "01000000 02000000 00000000 1f900000 00000000 00000000 64000000", 2, false, 100, true}, - {"tcp4 established", "01000000 05000000 7f000001 c0000000 7f000001 1f900000 65000000", 2, false, 101, true}, - {"tcp4 closed", "01000000 01000000 00000000 00000000 00000000 00000000 64000000", 2, false, 100, false}, - {"tcp6 listen", "01000000 00000000000000000000000000000000 00000000 1f900000 00000000000000000000000000000000 00000000 00000000 02000000 66000000", 23, false, 102, true}, - {"udp4 bound", "01000000 00000000 00350000 67000000", 2, true, 103, true}, - {"udp6 bound", "01000000 00000000000000000000000000000000 00000000 00350000 68000000", 23, true, 104, true}, + {"tcp4 listen", "01000000 02000000 00000000 1f900000 00000000 00000000 64000000", 2, false, 100, model.PeerListening}, + {"tcp4 established", "01000000 05000000 7f000001 c0000000 7f000001 1f900000 65000000", 2, false, 101, model.PeerConnected}, + {"tcp4 closed", "01000000 01000000 00000000 00000000 00000000 00000000 64000000", 2, false, 100, model.PeerNone}, + {"tcp6 listen", "01000000 00000000000000000000000000000000 00000000 1f900000 00000000000000000000000000000000 00000000 00000000 02000000 66000000", 23, false, 102, model.PeerListening}, + {"udp4 bound", "01000000 00000000 00350000 67000000", 2, true, 103, model.PeerConnected}, + {"udp6 bound", "01000000 00000000000000000000000000000000 00000000 00350000 68000000", 23, true, 104, model.PeerConnected}, } { t.Run(tc.name, func(t *testing.T) { var compact string @@ -34,7 +36,7 @@ func TestWindowsSocketTableBuffers(t *testing.T) { t.Fatal(err) } got, err := parseWindowsSocketTable(b, tc.family, tc.udp, []int32{tc.pid, 999}) - if err != nil || got[tc.pid] != tc.active || got[999] || len(got) != 2 { + if err != nil || got[tc.pid] != tc.state || got[999] != model.PeerNone || len(got) != 2 { t.Fatalf("%v %v", got, err) } if _, err := parseWindowsSocketTable(b[:len(b)-1], tc.family, tc.udp, nil); err == nil { diff --git a/internal/signatures/signatures.go b/internal/signatures/signatures.go index 2b6fcdeb..85a2d69c 100644 --- a/internal/signatures/signatures.go +++ b/internal/signatures/signatures.go @@ -36,6 +36,11 @@ type Signature struct { HostName string `toml:"host_name"` // ProtectTree excludes this helper and its descendants from orphan cleanup. ProtectTree bool `toml:"protect_tree"` + // IdleWithoutClients applies when this helper roots an orphan tree: only a + // connection to another of the user's processes, or the network, shows the + // tree is in use. Listeners and system-service sockets do not. Set it only + // when clients start a replacement once the helper is gone. + IdleWithoutClients bool `toml:"idle_without_clients"` // CommandModes matches argv prefixes after the executable (Agent), skipping // leading Codex config overrides. These entries never inspect joined text. CommandModes [][]string `toml:"command_modes"` diff --git a/internal/signatures/signatures.toml b/internal/signatures/signatures.toml index e19a47ad..ece0ed39 100644 --- a/internal/signatures/signatures.toml +++ b/internal/signatures/signatures.toml @@ -39,6 +39,8 @@ name = "claude-mcp" agent = "claude" role = "helper" command_modes = [["mcp", "serve"]] +# Clients use stdio; the /tmp/cc-socks listener is Claude Code's own. +idle_without_clients = true [[signature]] name = "claude-chrome-native-host" @@ -105,6 +107,9 @@ command_modes = [["app-server"]] name = "codex-app-server-broker" agent = "codex" role = "helper" +# The Codex plugin reuses a workspace's broker while broker.sock answers and +# starts a new one when it does not, so a broker without clients is idle. +idle_without_clients = true match = [ '^(?:[^[:space:]]*/)?node[[:space:]]+(?:[^[:space:]]*/)?app-server-broker\.mjs([[:space:]]|$)', '^(?:[^[:space:]]*/)?app-server-broker\.mjs([[:space:]]|$)', diff --git a/internal/tui/app.go b/internal/tui/app.go index 9785b79a..07ee7f91 100644 --- a/internal/tui/app.go +++ b/internal/tui/app.go @@ -27,6 +27,7 @@ import ( "github.com/labstack/driga/internal/model" "github.com/labstack/driga/internal/pathx" "github.com/labstack/driga/internal/platform" + "github.com/labstack/driga/internal/signatures" "github.com/labstack/driga/internal/spool" "github.com/labstack/driga/internal/store" "github.com/labstack/driga/internal/view" @@ -51,12 +52,15 @@ type Options struct { Now func() time.Time Exec func([]string) error Kill func(context.Context, model.Orphan, []int32, bool) (kill.Result, error) - Spool func(model.HookEvent) error - Summarize func(context.Context, string) error - CleanupList func(context.Context) ([]cleanup.Candidate, error) - CleanupApply func(context.Context, cleanup.Candidate) (cleanup.Result, error) - DataDir string - Context context.Context + // Signatures classifies orphan trees for the default Kill; nil means the + // built-in set. + Signatures *signatures.Set + Spool func(model.HookEvent) error + Summarize func(context.Context, string) error + CleanupList func(context.Context) ([]cleanup.Candidate, error) + CleanupApply func(context.Context, cleanup.Candidate) (cleanup.Result, error) + DataDir string + Context context.Context // LookupTmux resolves an ancestor pane PID to its tmux session. LookupTmux func(context.Context, []int32) (string, error) } @@ -198,8 +202,12 @@ func New(src Source, opts Options) tea.Model { opts.Spool = func(e model.HookEvent) error { return spool.Append(config.SpoolDir(), e) } } if opts.Kill == nil { + sigs := opts.Signatures + if sigs == nil { + sigs = signatures.Builtin() + } opts.Kill = func(ctx context.Context, o model.Orphan, p []int32, force bool) (kill.Result, error) { - return kill.Tree(ctx, platform.New(), o, p, force, 5*time.Second) + return kill.Tree(ctx, platform.New(), o, p, force, 5*time.Second, kill.WithClassifier(derive.NewClassifier(sigs, len(p)+1))) } } if opts.LookupTmux == nil { diff --git a/internal/viewer/startup_test.go b/internal/viewer/startup_test.go index 199b7d9c..656fdcda 100644 --- a/internal/viewer/startup_test.go +++ b/internal/viewer/startup_test.go @@ -17,6 +17,7 @@ import ( "github.com/labstack/driga/internal/config" "github.com/labstack/driga/internal/gitscan" "github.com/labstack/driga/internal/model" + "github.com/labstack/driga/internal/platform" "github.com/labstack/driga/internal/signatures" "github.com/labstack/driga/internal/store" "github.com/labstack/driga/internal/view" @@ -218,11 +219,14 @@ type startupSocketPlatform struct{ startupPlatform } func (*startupSocketPlatform) Procs(context.Context) ([]model.Proc, error) { return []model.Proc{{PID: 200, PPID: 1, StartTime: time.Now().Add(-time.Hour), Cmdline: "claude mcp serve", Argv: []string{"claude", "mcp", "serve"}}}, nil } -func (p *startupSocketPlatform) ExternalPeers(ctx context.Context, _ []int32) (map[int32]bool, error) { + +var _ platform.SocketProber = (*startupSocketPlatform)(nil) + +func (p *startupSocketPlatform) ExternalPeers(ctx context.Context, _ []int32) (map[int32]model.PeerState, error) { close(p.started) select { case <-p.release: - return map[int32]bool{200: false}, nil + return map[int32]model.PeerState{200: model.PeerNone}, nil case <-ctx.Done(): return nil, ctx.Err() }