1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
"""Tests for cross-agent-status (TDD: tests written before implementation)."""
from __future__ import annotations
import json
import os
import subprocess
import textwrap
from pathlib import Path
import pytest
SCRIPT = Path(__file__).resolve().parent.parent / "cross-agent-comms" / "cross-agent-status"
def _make_msg(path: Path, *, conv_id: str, seq: int, msg_type: str = "request",
proto_version: str = "5", timestamp: str = "2026-04-27T05:00:00-05:00") -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(textwrap.dedent(f"""\
#+TITLE: T
#+CONVERSATION_ID: {conv_id}
#+MESSAGE_TYPE: {msg_type}
#+SEQUENCE: {seq}
#+TIMESTAMP: {timestamp}
#+PROTOCOL_VERSION: {proto_version}
Body.
"""))
return path
def _run(args: list[str], env: dict | None = None) -> subprocess.CompletedProcess:
return subprocess.run([str(SCRIPT), *args], capture_output=True, text=True, env=env)
@pytest.fixture
def fake_projects(tmp_path, monkeypatch):
"""Create a fake ~/projects/<name>/inbox/from-agents/ tree under tmp_path."""
home = tmp_path / "home"
home.mkdir()
monkeypatch.setenv("HOME", str(home))
return home
def test_status_help(fake_projects):
result = _run(["--help"], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
assert "snapshot" in result.stdout.lower() or "pending" in result.stdout.lower()
def test_status_no_projects_clean_output(fake_projects):
result = _run([], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
# Empty machine prints either header-only table or "no projects" — accept either.
# No crash, no pending claims.
assert "pending" in result.stdout.lower() or result.stdout.strip() == ""
def test_status_one_pending_shows_up(fake_projects):
inbox = fake_projects / "projects" / "homelab" / "inbox" / "from-agents"
_make_msg(inbox / "20260427T100000Z-from-career-fixup.org", conv_id="fixup", seq=1)
result = _run([], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
assert "homelab" in result.stdout
assert "1" in result.stdout # pending count
assert "20260427T100000Z-from-career-fixup.org" in result.stdout
def test_status_released_conversation_zero_pending(fake_projects):
"""A conversation with a release message in it counts as 0 pending."""
inbox = fake_projects / "projects" / "homelab" / "inbox" / "from-agents"
_make_msg(inbox / "20260427T100000Z-from-career-done.org", conv_id="done", seq=1)
_make_msg(inbox / "20260427T100100Z-from-homelab-done.org", conv_id="done", seq=2, msg_type="release")
result = _run([], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
# Check the homelab row shows 0 pending.
lines = [ln for ln in result.stdout.splitlines() if "homelab" in ln]
# At least one homelab line should show 0 pending or "—".
assert any("0" in ln or "—" in ln for ln in lines)
def test_status_partial_release(fake_projects):
"""Conversation with release + a later message → that later message counts as pending."""
inbox = fake_projects / "projects" / "homelab" / "inbox" / "from-agents"
_make_msg(inbox / "20260427T100000Z-from-career-x.org", conv_id="x", seq=1,
timestamp="2026-04-27T05:00:00-05:00")
_make_msg(inbox / "20260427T100100Z-from-homelab-x.org", conv_id="x", seq=2, msg_type="release",
timestamp="2026-04-27T05:01:00-05:00")
# New message AFTER release: starts a fresh thread that's pending.
_make_msg(inbox / "20260427T200000Z-from-career-x.org", conv_id="x", seq=3,
timestamp="2026-04-27T15:00:00-05:00")
result = _run([], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
homelab_line = next(ln for ln in result.stdout.splitlines() if "homelab" in ln)
assert "1" in homelab_line # the post-release message is pending
def test_status_multiple_projects(fake_projects):
inbox_a = fake_projects / "projects" / "homelab" / "inbox" / "from-agents"
inbox_b = fake_projects / "projects" / "career" / "inbox" / "from-agents"
_make_msg(inbox_a / "20260427T100000Z-from-x-a.org", conv_id="a", seq=1)
_make_msg(inbox_b / "20260427T100100Z-from-x-b.org", conv_id="b", seq=1)
_make_msg(inbox_b / "20260427T100200Z-from-x-c.org", conv_id="c", seq=1)
result = _run([], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
# career has 2 pending, homelab has 1.
career_line = next(ln for ln in result.stdout.splitlines() if "career" in ln)
homelab_line = next(ln for ln in result.stdout.splitlines() if "homelab" in ln)
assert "2" in career_line
assert "1" in homelab_line
def test_status_json_output(fake_projects):
inbox = fake_projects / "projects" / "homelab" / "inbox" / "from-agents"
_make_msg(inbox / "20260427T100000Z-from-career-test.org", conv_id="test", seq=1)
result = _run(["--json"], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
payload = json.loads(result.stdout)
assert "projects" in payload
assert isinstance(payload["projects"], list)
homelab = next((p for p in payload["projects"] if p["name"] == "homelab"), None)
assert homelab is not None
assert homelab["pending_count"] == 1
def test_status_sort_pending_first(fake_projects):
"""Projects with pending messages sort before projects with 0."""
(fake_projects / "projects" / "alpha" / "inbox" / "from-agents").mkdir(parents=True)
inbox_zeta = fake_projects / "projects" / "zeta" / "inbox" / "from-agents"
_make_msg(inbox_zeta / "20260427T100000Z-from-x-z.org", conv_id="z", seq=1)
result = _run([], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0
lines = result.stdout.splitlines()
zeta_idx = next(i for i, ln in enumerate(lines) if "zeta" in ln)
alpha_idx = next(i for i, ln in enumerate(lines) if "alpha" in ln)
assert zeta_idx < alpha_idx, "pending project should sort before zero-pending project"
def test_status_halt_shows_banner(fake_projects):
halt = fake_projects / ".config" / "cross-agent-comms" / "HALT"
halt.parent.mkdir(parents=True)
halt.write_text("halted for test")
inbox = fake_projects / "projects" / "homelab" / "inbox" / "from-agents"
_make_msg(inbox / "20260427T100000Z-from-x-x.org", conv_id="x", seq=1)
result = _run([], env={**os.environ, "HOME": str(fake_projects)})
assert result.returncode == 0 # status continues to print under HALT
assert "HALT" in result.stdout
# Banner should mention the reason.
assert "halted for test" in result.stdout
def test_status_projects_glob_override(fake_projects):
inbox = fake_projects / "projects" / "homelab" / "inbox" / "from-agents"
_make_msg(inbox / "20260427T100000Z-from-x-a.org", conv_id="a", seq=1)
other_inbox = fake_projects / "projects" / "career" / "inbox" / "from-agents"
_make_msg(other_inbox / "20260427T100100Z-from-x-b.org", conv_id="b", seq=1)
# Glob limits to homelab only.
result = _run(
["--projects-glob", str(fake_projects / "projects" / "homelab" / "inbox" / "from-agents") + "/"],
env={**os.environ, "HOME": str(fake_projects)},
)
assert result.returncode == 0
assert "homelab" in result.stdout
# career not in scope.
assert "career" not in result.stdout
|
