Skip to content

Commit a8d9b8f

Browse files
authored
feat(presence): distinguish notifier liveness from activity (#230)
## Summary - classify a verified wake process as `notifier_live` and fresh `last_seen` as `recent_activity` - expose the source in `who` and `doctor --ops` without implying message consumption - document daemon-free systemd and launchd supervision for separate wake and monitor processes ## Verification - `make ci` - `go test -race ./...` - `GOOS=windows go build ./...` - `GOOS=freebsd go build ./...` - `gitleaks dir --no-banner --redact --exit-code 1 .` Closes #206
1 parent db99e78 commit a8d9b8f

9 files changed

Lines changed: 321 additions & 3 deletions

File tree

CLAUDE.md

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -369,6 +369,14 @@ pipes open after its JSON/text output exits. `doctor --ops` may report
369369
`target_present` and `repair_available`, but must remain diagnostic-only and
370370
never spawn a wake process.
371371

372+
`who` and `doctor --ops` presence sources have narrow semantics:
373+
`notifier_live` requires a valid wake lock with process identity confirmed by
374+
the existing wake-lock inspector; it proves prompt notification only, never
375+
consumption. `recent_activity` means a fresh `last_seen` without that proof.
376+
Do not introduce `consumer_live` wording without a separate monitor heartbeat
377+
or lock. Long-running wake/monitor supervision belongs to launchd, systemd, or
378+
another layer above daemon-free AMQ.
379+
372380
Git worktree diagnostics stay in `doctor --ops`, never in the send routing
373381
path. Relative project roots and auto-detected roots are per-worktree; sharing
374382
requires the same absolute `.amqrc` root or `AMQ_GLOBAL_ROOT`. The deep check is

COOP.md

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -280,6 +280,56 @@ code and may itself inject terminal input, `none` rejects `--inject-via`,
280280
`--inject-arg`, and `--inject-cmd`. Stderr output shares the TUI terminal by
281281
default and may remain visible until the TUI redraws.
282282

283+
### Supervisor recipes
284+
285+
AMQ remains daemon-free. If `wake` or `monitor` must stay attached across
286+
terminal restarts, let the operating system supervise the CLI process. Keep
287+
notification and consumption separate: `wake` only scans and notifies;
288+
`monitor` is the path that drains messages and emits receipts.
289+
290+
For systemd, use separate services with an absolute root. A notifier unit can
291+
use:
292+
293+
```ini
294+
[Service]
295+
Environment=AM_ROOT=/absolute/path/to/shared/.agent-mail/collab
296+
ExecStart=/usr/local/bin/amq wake --me claude --inject-mode none --bell
297+
Restart=always
298+
RestartSec=2
299+
```
300+
301+
A consumer unit uses the same environment but a different `ExecStart`; it
302+
restarts after each emitted batch:
303+
304+
```ini
305+
[Service]
306+
Environment=AM_ROOT=/absolute/path/to/shared/.agent-mail/collab
307+
ExecStart=/usr/local/bin/amq monitor --me claude --timeout 0 --include-body --json
308+
Restart=always
309+
RestartSec=2
310+
```
311+
312+
For launchd, put the equivalent absolute arguments in `ProgramArguments` and
313+
enable `KeepAlive`. Use one plist for `wake` and another for `monitor`:
314+
315+
```xml
316+
<key>ProgramArguments</key>
317+
<array>
318+
<string>/usr/local/bin/amq</string>
319+
<string>wake</string>
320+
<string>--root</string><string>/absolute/path/to/shared/.agent-mail/collab</string>
321+
<string>--me</string><string>claude</string>
322+
<string>--inject-mode</string><string>none</string>
323+
</array>
324+
<key>KeepAlive</key><true/>
325+
<key>ThrottleInterval</key><integer>2</integer>
326+
```
327+
328+
For the consumer plist, replace the command arguments after the executable with
329+
`monitor --root <absolute-root> --me claude --timeout 0 --include-body --json`.
330+
Route stdout/stderr to supervisor-managed logs and secure the unit/plist for the
331+
local user who owns the mailbox.
332+
283333
**Options:**
284334
- `--inject-mode auto|raw|paste|none` - Injection strategy; `none` enforces zero terminal input
285335
- `--wake-inject-mode auto|raw|paste|none` - `coop exec` pass-through for its managed wake

README.md

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -217,6 +217,18 @@ or injecting into the wrong terminal. Repaired wake output goes to
217217
`agents/<agent>/.wake.repair.log`; `doctor --ops` can report whether repair is
218218
available, but it never starts a wake process.
219219

220+
`amq who` and `amq doctor --ops` distinguish two activity sources:
221+
222+
- `notifier_live` means AMQ verified the process identity behind a valid
223+
`amq wake` lock. It proves prompt notification is attached; it does **not**
224+
prove messages are consumed.
225+
- `recent_activity` means `last_seen` was refreshed in the last 10 minutes,
226+
without a verified live notifier.
227+
228+
Consumption remains the job of `drain` or `monitor` and is evidenced by
229+
receipts. For long-running `wake` and `monitor` examples under systemd or
230+
launchd, see [Supervisor recipes](COOP.md#supervisor-recipes).
231+
220232
## Message Kinds & Priority
221233

222234
AMQ messages support kinds (`review_request`, `question`, `todo`, etc.) and priority levels (`urgent`, `normal`, `low`). See [COOP.md](COOP.md) for the full protocol.

internal/cli/doctor.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -176,6 +176,9 @@ func runDoctor(args []string) error {
176176
line += fmt.Sprintf(", %d DLQ", a.DLQCount)
177177
}
178178
line += fmt.Sprintf(", presence %s (%.0fs ago)", a.PresenceStatus, a.PresenceAgeSeconds)
179+
if a.PresenceSource != "" {
180+
line += ", source " + a.PresenceSource
181+
}
179182
if err := writeStdoutLine(line); err != nil {
180183
return err
181184
}

internal/cli/doctor_ops.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ type opsAgent struct {
3535
OldestDLQAgeSeconds float64 `json:"oldest_dlq_age_seconds"`
3636
PresenceStatus string `json:"presence_status"`
3737
PresenceAgeSeconds float64 `json:"presence_age_seconds"`
38+
PresenceSource string `json:"presence_source,omitempty"`
3839
}
3940

4041
type opsOperatorGate struct {
@@ -126,15 +127,18 @@ func runOpsChecks(root string, rootSource string, fixWakeLocks bool) *doctorOpsR
126127
}
127128

128129
// Presence
130+
recentActivity := false
129131
p, err := presence.Read(root, handle)
130132
if err == nil {
131133
agent.PresenceStatus = p.Status
132134
if t, err := time.Parse(time.RFC3339Nano, p.LastSeen); err == nil {
133135
agent.PresenceAgeSeconds = now.Sub(t).Seconds()
136+
recentActivity = agent.PresenceAgeSeconds < (10 * time.Minute).Seconds()
134137
}
135138
} else {
136139
agent.PresenceStatus = "unknown"
137140
}
141+
agent.PresenceSource = resolvePresenceSource(root, handle, recentActivity)
138142

139143
// Round to reasonable precision
140144
agent.OldestUnreadAgeSeconds = math.Round(agent.OldestUnreadAgeSeconds)

internal/cli/presence_source.go

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
package cli
2+
3+
const (
4+
presenceSourceNotifierLive = "notifier_live"
5+
presenceSourceRecentActivity = "recent_activity"
6+
)
7+
8+
func resolvePresenceSource(root, agent string, recentActivity bool) string {
9+
inspection := inspectWakeLock(root, agent)
10+
if inspection.Status == wakeLockValid && inspection.IdentityConfirmed {
11+
return presenceSourceNotifierLive
12+
}
13+
if recentActivity {
14+
return presenceSourceRecentActivity
15+
}
16+
return ""
17+
}
Lines changed: 209 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,209 @@
1+
package cli
2+
3+
import (
4+
"encoding/json"
5+
"path/filepath"
6+
"strings"
7+
"testing"
8+
"time"
9+
10+
"github.com/avivsinai/agent-message-queue/internal/config"
11+
"github.com/avivsinai/agent-message-queue/internal/fsq"
12+
"github.com/avivsinai/agent-message-queue/internal/presence"
13+
)
14+
15+
func TestRunWhoDistinguishesNotifierLiveFromRecentActivity(t *testing.T) {
16+
root := setupPresenceSourceFixture(t)
17+
18+
output, err := captureEnvStdout(t, func() error {
19+
return runWho([]string{"--root", root, "--json"})
20+
})
21+
if err != nil {
22+
t.Fatalf("runWho json: %v", err)
23+
}
24+
var sessions []struct {
25+
Agents []struct {
26+
Handle string `json:"handle"`
27+
Active bool `json:"active"`
28+
PresenceSource string `json:"presence_source"`
29+
} `json:"agents"`
30+
}
31+
if err := json.Unmarshal([]byte(output), &sessions); err != nil {
32+
t.Fatalf("unmarshal who json: %v\n%s", err, output)
33+
}
34+
if len(sessions) != 1 {
35+
t.Fatalf("sessions=%#v, want one", sessions)
36+
}
37+
agents := make(map[string]struct {
38+
Active bool
39+
Source string
40+
})
41+
for _, agent := range sessions[0].Agents {
42+
agents[agent.Handle] = struct {
43+
Active bool
44+
Source string
45+
}{Active: agent.Active, Source: agent.PresenceSource}
46+
}
47+
if got := agents["notifier"]; !got.Active || got.Source != "notifier_live" {
48+
t.Fatalf("notifier=%#v, want active notifier_live", got)
49+
}
50+
if got := agents["recent"]; !got.Active || got.Source != "recent_activity" {
51+
t.Fatalf("recent=%#v, want active recent_activity", got)
52+
}
53+
if got := agents["unverified"]; !got.Active || got.Source != "recent_activity" {
54+
t.Fatalf("unverified=%#v, want recent_activity without notifier claim", got)
55+
}
56+
if got := agents["stale"]; got.Active || got.Source != "" {
57+
t.Fatalf("stale=%#v, want inactive without source", got)
58+
}
59+
60+
text, err := captureEnvStdout(t, func() error {
61+
return runWho([]string{"--root", root})
62+
})
63+
if err != nil {
64+
t.Fatalf("runWho text: %v", err)
65+
}
66+
for _, want := range []string{
67+
"notifier active (notifier_live)",
68+
"recent active (recent_activity)",
69+
"unverified active (recent_activity)",
70+
} {
71+
if !strings.Contains(text, want) {
72+
t.Fatalf("who text missing %q:\n%s", want, text)
73+
}
74+
}
75+
for _, forbidden := range []string{"consumer_live", "consumption"} {
76+
if strings.Contains(text, forbidden) {
77+
t.Fatalf("who text must not claim %q:\n%s", forbidden, text)
78+
}
79+
}
80+
}
81+
82+
func TestDoctorOpsDistinguishesNotifierLiveFromRecentActivity(t *testing.T) {
83+
root := setupPresenceSourceFixture(t)
84+
result := runOpsChecks(root, "test", false)
85+
86+
agents := make(map[string]opsAgent)
87+
for _, agent := range result.Agents {
88+
agents[agent.Handle] = agent
89+
}
90+
if got := agents["notifier"].PresenceSource; got != "notifier_live" {
91+
t.Fatalf("notifier presence_source=%q, want notifier_live", got)
92+
}
93+
if got := agents["recent"].PresenceSource; got != "recent_activity" {
94+
t.Fatalf("recent presence_source=%q, want recent_activity", got)
95+
}
96+
if got := agents["unverified"].PresenceSource; got != "recent_activity" {
97+
t.Fatalf("unverified presence_source=%q, want recent_activity", got)
98+
}
99+
if got := agents["stale"].PresenceSource; got != "" {
100+
t.Fatalf("stale presence_source=%q, want empty", got)
101+
}
102+
103+
data, err := json.Marshal(result)
104+
if err != nil {
105+
t.Fatalf("marshal ops result: %v", err)
106+
}
107+
if !strings.Contains(string(data), `"presence_source":"notifier_live"`) ||
108+
!strings.Contains(string(data), `"presence_source":"recent_activity"`) {
109+
t.Fatalf("doctor JSON missing presence sources: %s", data)
110+
}
111+
112+
t.Setenv("AM_ROOT", root)
113+
t.Setenv("AM_BASE_ROOT", "")
114+
t.Setenv("AM_SESSION", "")
115+
text, err := captureEnvStdout(t, func() error {
116+
return runDoctor([]string{"--ops"})
117+
})
118+
if err != nil {
119+
t.Fatalf("runDoctor text: %v", err)
120+
}
121+
for agent, source := range map[string]string{
122+
"notifier": "notifier_live",
123+
"recent": "recent_activity",
124+
} {
125+
line := lineWithPrefix(text, " "+agent+":")
126+
if !strings.Contains(line, ", source "+source) {
127+
t.Fatalf("doctor line for %s missing source %q: %q\n%s", agent, source, line, text)
128+
}
129+
}
130+
}
131+
132+
func lineWithPrefix(text, prefix string) string {
133+
for _, line := range strings.Split(text, "\n") {
134+
if strings.HasPrefix(line, prefix) {
135+
return line
136+
}
137+
}
138+
return ""
139+
}
140+
141+
func setupPresenceSourceFixture(t *testing.T) string {
142+
t.Helper()
143+
root := filepath.Join(secureTempDirForTest(t), "collab")
144+
agents := []string{"notifier", "recent", "stale", "unverified"}
145+
for _, agent := range agents {
146+
if err := fsq.EnsureAgentDirs(root, agent); err != nil {
147+
t.Fatalf("EnsureAgentDirs(%s): %v", agent, err)
148+
}
149+
}
150+
if err := config.WriteConfig(filepath.Join(root, "meta", "config.json"), config.Config{
151+
Version: 1,
152+
Agents: agents,
153+
}, true); err != nil {
154+
t.Fatalf("write config: %v", err)
155+
}
156+
157+
now := time.Now()
158+
for _, item := range []struct {
159+
agent string
160+
seenAt time.Time
161+
}{
162+
{agent: "notifier", seenAt: now.Add(-time.Hour)},
163+
{agent: "recent", seenAt: now},
164+
{agent: "stale", seenAt: now.Add(-time.Hour)},
165+
{agent: "unverified", seenAt: now},
166+
} {
167+
if err := presence.Write(root, presence.New(item.agent, "active", "", item.seenAt)); err != nil {
168+
t.Fatalf("write %s presence: %v", item.agent, err)
169+
}
170+
}
171+
172+
const pid = 4242
173+
const processStart = "verified-start"
174+
const bootID = "verified-boot"
175+
args := []string{"amq", "wake", "--root", root, "--me", "notifier"}
176+
writeWakeLockForTest(t, root, "notifier", wakeLock{
177+
PID: pid,
178+
ProcessStart: processStart,
179+
BootID: bootID,
180+
Executable: "amq",
181+
Args: args,
182+
})
183+
const unverifiedPID = 4343
184+
writeWakeLockForTest(t, root, "unverified", wakeLock{
185+
PID: unverifiedPID,
186+
Executable: "amq",
187+
})
188+
stubInspectWakeProcess(t, func(gotPID int) wakeProcessInfo {
189+
if gotPID == unverifiedPID {
190+
return wakeProcessInfo{
191+
PID: gotPID,
192+
Running: true,
193+
Executable: "/usr/local/bin/amq",
194+
}
195+
}
196+
if gotPID != pid {
197+
return wakeProcessInfo{PID: gotPID}
198+
}
199+
return wakeProcessInfo{
200+
PID: gotPID,
201+
Running: true,
202+
StartToken: processStart,
203+
BootID: bootID,
204+
Executable: "/usr/local/bin/amq",
205+
Args: args,
206+
}
207+
})
208+
return root
209+
}

0 commit comments

Comments
 (0)