"""Comprehensive validation harness for native Go port of Albedo CLI.
Scenarios tested:
1. Noninteractive stdout commands (sessions, sessions --json, send, stop, daemon, login non-tty rejection, unknown command)
2. Interactive PTY Resume (120x40): live SSE stream rendering, assistant text, thinking, tool traces, status footer, clean terminal exit
3. Interactive PTY Session Picker (120x40): bare 'albedo' with multiple sessions launches Session Picker
4. Interactive PTY Login (120x40): unconfigured profile launches Login form
5. Interactive PTY Tiny Terminal (40x10): constrained viewport layout; asserts line widths <= 40 cols
6. SSE Memory Retention: 1,200+ uninterrupted assistant EventText chunks (>300KB) + canonical EventMessage with verification token; asserts complete consumption, peak RSS < 150 MiB, exit 0, and token display in UI.
Durable evidence is saved to /tmp/albedo-smoke-evidence/.
"""
import argparse
import fcntl
import http.server
import json
import os
from pathlib import Path
import pty
import re
import selectors
import struct
import subprocess
import termios
import threading
import time
ROOT = Path(__file__).resolve().parents[2]
EVIDENCE_DIR = Path("/tmp/albedo-smoke-evidence")
COMPLETION_TOKEN = "FINAL_ASSISTANT_STREAM_CONFIRMED_PARITY_CHECK_TOKEN_774910"
class StrictMockDaemon:
def __init__(
self,
home_dir,
token="fixture-token",
version=2,
sessions=None,
configured=True,
is_stress=False,
):
self.home_dir = Path(home_dir)
self.token = token
self.version = version
self.sessions = sessions or [
{
"id": "deadbeef12345678",
"title": "Albedo Refactor Session",
"last_assistant_at": int(time.time()) - 300,
"workspace": "/tmp/workspace1",
"model": "gpt-4o",
"protocol": "responses",
"provider": "openai",
},
{
"id": "cafebabe87654321",
"title": "Documentation Updates",
"last_assistant_at": int(time.time()) - 7200,
"workspace": "/tmp/workspace2",
"model": "claude-3-5-sonnet",
"protocol": "responses",
"provider": "anthropic",
},
]
self.configured = configured
self.is_stress = is_stress
self.received_events = []
self.received_interrupts = []
self.total_events_sent = 0
self.ready_event = threading.Event()
self.ended_event = threading.Event()
self.stop_stream = threading.Event()
self.server = None
self.thread = None
def start(self):
parent = self
class Handler(http.server.BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass
def check_auth(self):
auth = self.headers.get("Authorization")
if auth != f"Bearer {parent.token}":
self.send_response(401)
self.end_headers()
self.wfile.write(b'{"error":"unauthorized"}')
return False
return True
def do_GET(self):
if not self.check_auth():
return
if self.path == "/health":
self.send_response(200)
self.send_header("content-type", "application/json")
self.end_headers()
payload = {
"ok": True,
"version": parent.version,
"capabilities": [
"session_provider",
"session_workspace",
"session_extensions",
"session_tree",
"session_context",
"session_commands",
],
}
self.wfile.write(json.dumps(payload).encode())
return
if self.path == "/sessions":
self.send_response(200)
self.send_header("content-type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(parent.sessions).encode())
return
if "/status" in self.path:
self.send_response(200)
self.send_header("content-type", "application/json")
self.end_headers()
self.wfile.write(
json.dumps(
{"running": False, "idle": True, "phase": "resting"}
).encode()
)
return
if "/commands" in self.path:
self.send_response(200)
self.send_header("content-type", "application/json")
self.end_headers()
catalog = [
{
"name": "/review",
"description": "review recent changes",
"method": "review",
"arguments": [],
"modelCallable": True,
"userTurn": True,
},
{
"name": "/compact",
"description": "compact history",
"method": "compact",
"arguments": [],
"modelCallable": False,
"userTurn": True,
},
]
self.wfile.write(json.dumps(catalog).encode())
return
if "/stream" in self.path:
self.send_response(200)
self.send_header("content-type", "text/event-stream")
self.send_header("cache-control", "no-cache")
self.end_headers()
def send_chunk(events, cursor):
payload = (
"data: "
+ json.dumps(
{
"generation": "smoke-generation",
"cursor": cursor,
"events": events,
}
)
+ "\n\n"
)
self.wfile.write(payload.encode())
self.wfile.flush()
parent.total_events_sent += len(events)
try:
cursor = 0
send_chunk([{"type": "reset"}], cursor)
cursor += 1
send_chunk(
[
{
"type": "user",
"text": "streaming benchmark test turn",
"source": "chat",
"triggeredAt": "smoke",
}
],
cursor,
)
cursor += 1
parent.ready_event.set()
if parent.is_stress:
# Massive uninterrupted assistant EventText stream (>300KB, >1200 chunks)
full_text_acc = []
for i in range(1200):
if parent.stop_stream.is_set():
break
chunk = f"Paragraph {i}: Validating bounded stream retention and memory virtualization under uninterrupted EventText loads at step {i}.\n"
full_text_acc.append(chunk)
send_chunk([{"type": "text", "text": chunk}], cursor)
cursor += 1
if i % 100 == 0:
time.sleep(0.002)
# Append ending token
final_marker = f"\n\n{COMPLETION_TOKEN}\n"
full_text_acc.append(final_marker)
send_chunk([{"type": "text", "text": final_marker}], cursor)
cursor += 1
# Canonical EventMessage
assembled = "".join(full_text_acc)
send_chunk(
[
{
"type": "message",
"role": "assistant",
"text": assembled,
"timestamp": int(time.time() * 1000),
}
],
cursor,
)
cursor += 1
send_chunk(
[
{
"type": "usage",
"completionTokens": cursor,
"totalTokens": cursor + 500,
}
],
cursor,
)
cursor += 1
parent.ended_event.set()
else:
# Standard test stream with thinking, text, and tools
send_chunk(
[
{
"type": "thinking",
"text": "analyzing workspace structure\n",
}
],
cursor,
)
cursor += 1
text_parts = [
"I have verified the project configuration.\n\n",
'```go\nfunc main() {\n fmt.Println("Albedo Native Go CLI")\n}\n```\n',
f"Everything is verified. {COMPLETION_TOKEN}\n",
]
for t in text_parts:
send_chunk([{"type": "text", "text": t}], cursor)
cursor += 1
time.sleep(0.01)
send_chunk(
[
{
"type": "tool",
"name": "edit",
"args": {"path": "cli/cmd/albedo/main.go"},
"result": "success",
"trace": {
"activities": [
{"kind": "read", "target": "main.go"}
],
"changes": [
{
"path": "main.go",
"kind": "diff",
"diff": "@@ -1,3 +1,3 @@\n-old\n+new",
"added": 1,
"removed": 1,
}
],
},
}
],
cursor,
)
cursor += 1
send_chunk(
[
{
"type": "usage",
"completionTokens": 50,
"totalTokens": 150,
}
],
cursor,
)
cursor += 1
parent.ended_event.set()
while not parent.stop_stream.wait(0.2):
self.wfile.write(b": keepalive\n\n")
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError):
pass
return
# Strict: Unknown GET returns 404
self.send_response(404)
self.end_headers()
self.wfile.write(b'{"error":"not found"}')
def do_POST(self):
if not self.check_auth():
return
content_len = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_len).decode() if content_len > 0 else ""
parsed = json.loads(body) if body else {}
if "/events" in self.path:
parent.received_events.append(parsed)
self.send_response(200)
self.send_header("content-type", "application/json")
self.end_headers()
self.wfile.write(b'{"ok":true,"queued":false}')
return
if "/interrupt" in self.path:
parent.received_interrupts.append(parsed)
self.send_response(200)
self.send_header("content-type", "application/json")
self.end_headers()
self.wfile.write(b'{"ok":true,"interrupted":true}')
return
if self.path == "/sessions":
new_sess = {
"id": "newsession12345678",
"title": "Fresh Created Session",
"workspace": parsed.get("workspace", "/tmp"),
"model": "gpt-4o",
"protocol": "responses",
"provider": "openai",
}
self.send_response(201)
self.send_header("content-type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(new_sess).encode())
return
if "/shutdown" in self.path:
self.send_response(200)
self.send_header("content-type", "application/json")
self.end_headers()
self.wfile.write(b'{"ok":true}')
return
# Strict: Unknown POST returns 404
self.send_response(404)
self.end_headers()
self.wfile.write(b'{"error":"not found"}')
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
self.thread.start()
self.home_dir.mkdir(parents=True, exist_ok=True)
(self.home_dir / "daemon.json").write_text(
json.dumps(
{
"port": self.server.server_port,
"token": self.token,
"pid": os.getpid(),
"version": self.version,
}
)
)
if self.configured:
(self.home_dir / "config.json").write_text(
json.dumps(
{
"active": "fixture",
"providers": {
"fixture": {
"extension": "openai",
"baseUrl": f"http://127.0.0.1:{self.server.server_port}",
"apiKey": "fixture-key",
"model": "gpt-4o",
"protocol": "responses",
}
},
}
)
)
else:
(self.home_dir / "config.json").write_text(json.dumps({"providers": {}}))
def stop(self):
self.stop_stream.set()
if self.server:
self.server.shutdown()
self.server.server_close()
def clean_ansi(raw_bytes):
text = raw_bytes.decode("utf-8", errors="replace")
# Replace cursor movement/addressing ([H, [9;H, etc.) with newline
text = re.sub(r"\x1b\[[0-9;]*[Hf]", "\n", text)
# Strip remaining CSI escape sequences
text = re.sub(r"\x1b\[[0-9;?]*[a-zA-Z]", "", text)
# Strip OSC sequences
text = re.sub(r"\x1b\][^\x1b]*(\x07|\x1b\\)", "", text)
# Normalize carriage returns and line endings
text = text.replace("\r\n", "\n").replace("\r", "\n")
return text
def run_pty_scenario(binary, home_dir, args, rows, cols, duration=2.5, input_keys=None):
raw_output = bytearray()
master, slave = pty.openpty()
fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
env = {
**os.environ,
"ALBEDO_HOME": str(home_dir),
"TERM": "xterm-256color",
"ALBEDO_NO_BROWSER": "1",
}
child = subprocess.Popen(
[str(binary)] + args,
env=env,
stdin=slave,
stdout=slave,
stderr=slave,
start_new_session=True,
)
os.close(slave)
os.set_blocking(master, False)
selector = selectors.DefaultSelector()
selector.register(master, selectors.EVENT_READ)
started = time.monotonic()
input_sent = False
exit_sent = False
try:
while child.poll() is None:
for key, _ in selector.select(0.05):
try:
data = os.read(key.fd, 4096)
if data:
raw_output.extend(data)
except (BlockingIOError, OSError):
pass
elapsed = time.monotonic() - started
if input_keys and not input_sent and elapsed > 0.8:
for k in input_keys:
os.write(master, k)
time.sleep(0.05)
input_sent = True
if not exit_sent and elapsed > duration:
os.write(master, b"\x03")
exit_sent = True
if elapsed > duration + 4.0:
child.terminate()
break
child.wait(timeout=3.0)
# Drain trailing bytes
while True:
try:
data = os.read(master, 4096)
if not data:
break
raw_output.extend(data)
except (BlockingIOError, OSError):
break
finally:
if child.poll() is None:
child.terminate()
child.wait(timeout=2.0)
os.close(master)
selector.close()
cleaned = clean_ansi(raw_output)
return raw_output, cleaned, child.returncode
def test_noninteractive(binary, out_dir):
print("=== 1. Noninteractive CLI Commands ===")
home = out_dir / "home_noninteractive"
mock = StrictMockDaemon(home)
mock.start()
env = {**os.environ, "ALBEDO_HOME": str(home)}
try:
p = subprocess.run(
[str(binary), "sessions"],
env=env,
capture_output=True,
text=True,
timeout=10,
)
assert p.returncode == 0, f"sessions error: {p.stderr}"
assert "Albedo Refactor Session" in p.stdout
assert "[deadbeef]" in p.stdout
print(" [PASS] albedo sessions")
p = subprocess.run(
[str(binary), "sessions", "--json"],
env=env,
capture_output=True,
text=True,
timeout=10,
)
assert p.returncode == 0, f"sessions --json error: {p.stderr}"
data = json.loads(p.stdout)
assert len(data) == 2
assert data[0]["id"] == "deadbeef12345678"
print(" [PASS] albedo sessions --json")
p = subprocess.run(
[str(binary), "send", "deadbeef12345678", "hello albedo"],
env=env,
capture_output=True,
text=True,
timeout=10,
)
assert p.returncode == 0, f"send error: {p.stderr}"
assert len(mock.received_events) == 1
assert mock.received_events[0]["content"] == "hello albedo"
print(" [PASS] albedo send")
p = subprocess.run(
[str(binary), "stop", "deadbeef12345678"],
env=env,
capture_output=True,
text=True,
timeout=10,
)
assert p.returncode == 0, f"stop error: {p.stderr}"
assert len(mock.received_interrupts) == 1
print(" [PASS] albedo stop")
p = subprocess.run(
[str(binary), "daemon"], env=env, capture_output=True, text=True, timeout=10
)
assert p.returncode == 0, f"daemon error: {p.stderr}"
assert mock.server is not None
assert f"127.0.0.1:{mock.server.server_port}" in p.stdout
print(" [PASS] albedo daemon")
p = subprocess.run(
[str(binary), "login"], env=env, capture_output=True, text=True, timeout=10
)
assert p.returncode != 0
assert "login requires a terminal" in (p.stderr + p.stdout)
print(" [PASS] albedo login (non-TTY rejection)")
p = subprocess.run(
[str(binary), "foobar"], env=env, capture_output=True, text=True, timeout=10
)
assert p.returncode != 0
assert "unknown command" in (p.stderr + p.stdout)
print(" [PASS] unknown command rejection")
finally:
mock.stop()
def test_pty_resume(binary, out_dir):
print("=== 2. Interactive PTY Resume (120x40 Standard) ===")
home = out_dir / "home_resume"
mock = StrictMockDaemon(home)
mock.start()
try:
raw, cleaned, code = run_pty_scenario(
binary,
home,
["resume", "deadbeef12345678"],
rows=40,
cols=120,
duration=2.5,
)
(out_dir / "pty_resume_120x40.raw").write_bytes(raw)
(out_dir / "pty_resume_120x40.txt").write_text(cleaned)
assert code == 0, f"Expected clean exit 0, got {code}"
assert b"\x1b[?1049h" not in raw, (
"Inline TUI unexpectedly entered the alternate screen"
)
assert b"\x1b[?25h" in raw, "Cursor show sequence not emitted"
for mode in (1002, 1006):
enabled = f"\x1b[?{mode}h".encode()
disabled = f"\x1b[?{mode}l".encode()
assert enabled in raw, f"Chat mouse mode {mode} was not enabled"
assert raw.rfind(disabled) > raw.rfind(enabled), (
f"Chat mouse mode {mode} was not disabled on exit"
)
lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
assert any("streaming benchmark test turn" in line for line in lines), (
"User prompt not rendered"
)
assert any("gpt-4o" in line for line in lines), (
"Status footer model not rendered"
)
assert any(COMPLETION_TOKEN in line for line in lines), (
"Completion token not rendered in viewport"
)
print(
" [PASS] Rendered transcript, model status, and exit code 0. Terminal restored."
)
finally:
mock.stop()
def test_pty_session_picker(binary, out_dir):
print("=== 3. Interactive PTY Bare Launch -> Session Picker ===")
home = out_dir / "home_picker"
mock = StrictMockDaemon(home)
mock.start()
try:
raw, cleaned, code = run_pty_scenario(
binary, home, [], rows=30, cols=100, duration=2.0, input_keys=[b"\x1b"]
)
(out_dir / "pty_bare_picker.raw").write_bytes(raw)
(out_dir / "pty_bare_picker.txt").write_text(cleaned)
assert code == 0, f"Expected clean exit 0, got {code}"
lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
has_picker = any(
"sessions" in line or "Albedo Refactor Session" in line for line in lines
)
assert has_picker, f"Session picker not rendered. Output: {cleaned}"
print(f" [PASS] Session Picker rendered cleanly. Code: {code}")
finally:
mock.stop()
def test_pty_unconfigured_login(binary, out_dir):
print("=== 4. Interactive PTY Unconfigured -> Login View ===")
home = out_dir / "home_login"
mock = StrictMockDaemon(home, configured=False)
mock.start()
try:
raw, cleaned, code = run_pty_scenario(
binary, home, [], rows=30, cols=100, duration=2.0, input_keys=[b"\x1b"]
)
(out_dir / "pty_login_unconfigured.raw").write_bytes(raw)
(out_dir / "pty_login_unconfigured.txt").write_text(cleaned)
assert code == 0, f"Expected clean exit 0, got {code}"
lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
has_login = any("login" in line or "provider" in line for line in lines)
assert has_login, f"Login screen not rendered. Output: {cleaned}"
print(f" [PASS] Login screen rendered cleanly. Code: {code}")
finally:
mock.stop()
def test_pty_standalone_login_cancel(binary, out_dir):
print("=== Standalone albedo login [Cancel -> Exit 0 to Shell] ===")
home = out_dir / "home_login_cancel"
mock = StrictMockDaemon(home)
mock.start()
try:
raw, cleaned, code = run_pty_scenario(
binary,
home,
["login"],
rows=30,
cols=100,
duration=1.5,
input_keys=[b"\x1b"],
)
(out_dir / "pty_standalone_login_cancel.txt").write_text(cleaned)
assert code == 0, (
f"Expected clean exit 0 on standalone login cancel, got {code}"
)
lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
assert any("login" in line for line in lines), "Login view not rendered"
# Must not enter session picker or chat
assert not any("albedo sessions" in line for line in lines), (
"Standalone login cancel opened session picker!"
)
assert not any(
"type a message" in line or "› type" in line for line in lines
), "Standalone login cancel opened chat!"
print(" [PASS] Standalone login cancel exited cleanly to shell (code 0)")
finally:
mock.stop()
def test_pty_standalone_login_save_flow(binary, out_dir):
print("=== Standalone albedo login [Provider Save -> Exit 0 to Shell] ===")
home = out_dir / "home_login_save"
home.mkdir(parents=True, exist_ok=True)
mock = StrictMockDaemon(home, configured=False)
mock.start()
assert mock.server is not None
# Pre-populate an inactive provider in config.json
cfg = {
"providers": {
"testprov": {
"extension": "openai",
"baseUrl": f"http://127.0.0.1:{mock.server.server_port}",
"apiKey": "test-key",
"model": "gpt-4o",
"protocol": "responses",
}
}
}
(home / "config.json").write_text(json.dumps(cfg))
try:
# Run albedo login testprov: matches existing inactive provider, saves as active, and quits directly to shell
raw, cleaned, code = run_pty_scenario(
binary, home, ["login", "testprov"], rows=30, cols=100, duration=1.5
)
(out_dir / "pty_standalone_login_save.txt").write_text(cleaned)
assert code == 0, f"Expected exit code 0 on standalone login save, got {code}"
# Verify config.json was updated with active = testprov
saved_cfg = json.loads((home / "config.json").read_text())
assert saved_cfg.get("active") == "testprov", (
f"Expected active: testprov, got: {saved_cfg}"
)
# Verify it did not launch chat or session picker
lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
assert not any("albedo sessions" in line for line in lines), (
"Standalone login save opened session picker!"
)
assert not any("type a message" in line for line in lines), (
"Standalone login save opened chat!"
)
print(
" [PASS] Standalone login saved active provider and exited directly to shell (code 0)"
)
finally:
mock.stop()
def test_pty_tiny_terminal(binary, out_dir):
print("=== 5. Interactive PTY Tiny Terminal (40x10) ===")
home = out_dir / "home_tiny"
mock = StrictMockDaemon(home)
mock.start()
try:
raw, cleaned, code = run_pty_scenario(
binary, home, ["resume", "deadbeef12345678"], rows=10, cols=40, duration=2.0
)
(out_dir / "pty_tiny_40x10.raw").write_bytes(raw)
(out_dir / "pty_tiny_40x10.txt").write_text(cleaned)
assert code == 0, f"Expected clean exit 0, got {code}"
lines = cleaned.splitlines()
# Behavioral layout check: every rendered line must fit within 40 columns
for line in lines:
assert len(line) <= 40, (
f"Line exceeded tiny terminal width 40: {len(line)} chars: {line!r}"
)
nonempty = [line.strip() for line in lines if line.strip()]
assert len(nonempty) > 0, "No content rendered in tiny terminal"
assert any(
"gpt-4o" in line or "reasoning" in line or "streaming" in line
for line in nonempty
), "No expected session content in tiny view"
print(
f" [PASS] Tiny terminal (40x10) respected max column bounds on all {len(lines)} lines without panic. Code: {code}"
)
finally:
mock.stop()
def test_sse_retention_and_memory(binary, out_dir):
print("=== 6. SSE Stream Retention & Peak Memory (>1,200 EventText chunks) ===")
home = out_dir / "home_memory"
mock = StrictMockDaemon(home, is_stress=True)
mock.start()
raw_output = bytearray()
master, slave = pty.openpty()
fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 40, 120, 0, 0))
env = {
**os.environ,
"ALBEDO_HOME": str(home),
"TERM": "xterm-256color",
"ALBEDO_NO_BROWSER": "1",
}
child = subprocess.Popen(
[str(binary), "resume", "deadbeef12345678"],
env=env,
stdin=slave,
stdout=slave,
stderr=slave,
start_new_session=True,
)
os.close(slave)
os.set_blocking(master, False)
selector = selectors.DefaultSelector()
selector.register(master, selectors.EVENT_READ)
started = time.monotonic()
rss_kb_samples = []
token_seen_in_ui = False
sig_sent = False
try:
while child.poll() is None:
for key, _ in selector.select(0.05):
try:
data = os.read(key.fd, 8192)
if data:
raw_output.extend(data)
except (BlockingIOError, OSError):
pass
try:
out = (
subprocess.check_output(["ps", "-o", "rss=", "-p", str(child.pid)])
.decode()
.strip()
)
if out:
rss_kb_samples.append(int(out))
except Exception:
pass
# Strict verification: wait until the completion token is actually received in the PTY output
if COMPLETION_TOKEN.encode() in raw_output and not sig_sent:
token_seen_in_ui = True
time.sleep(0.2)
os.write(master, b"\x03")
sig_sent = True
# Timeout safety
if time.monotonic() - started > 25.0:
os.write(master, b"\x03")
time.sleep(0.2)
child.terminate()
break
child.wait(timeout=4.0)
# Drain trailing bytes
while True:
try:
data = os.read(master, 4096)
if not data:
break
raw_output.extend(data)
except (BlockingIOError, OSError):
break
finally:
if child.poll() is None:
child.terminate()
child.wait(timeout=2.0)
os.close(master)
selector.close()
mock.stop()
cleaned = clean_ansi(raw_output)
(out_dir / "pty_stream_stress.raw").write_bytes(raw_output)
(out_dir / "pty_stream_stress.txt").write_text(cleaned)
# 1. Assertions on Stream Completion
assert mock.ended_event.is_set(), "Server stream did not complete"
assert mock.total_events_sent >= 1200, (
f"Expected >=1200 events, sent: {mock.total_events_sent}"
)
assert child.returncode == 0, f"Expected clean child exit 0, got {child.returncode}"
# 2. Strict UI Consumption Assertion: token must be present in captured UI output
assert token_seen_in_ui, (
f"UI did not consume/display the ending marker {COMPLETION_TOKEN} before exit!"
)
assert COMPLETION_TOKEN in cleaned, f"Cleaned output missing {COMPLETION_TOKEN}!"
# 3. Assertions on Memory
assert len(rss_kb_samples) > 0, "No memory samples were captured during streaming"
peak_rss_mib = max(rss_kb_samples) / 1024.0
assert peak_rss_mib > 0.0, "Peak RSS recorded was 0"
assert peak_rss_mib < 150.0, f"Peak memory exceeded budget: {peak_rss_mib:.2f} MiB"
(out_dir / "memory_samples.json").write_text(
json.dumps(
{
"samples_count": len(rss_kb_samples),
"peak_rss_mib": round(peak_rss_mib, 2),
"total_events": mock.total_events_sent,
"token_confirmed": True,
"all_rss_kb": rss_kb_samples,
},
indent=2,
)
)
print(f" [PASS] Streamed {mock.total_events_sent} uninterrupted events (>300KB).")
print(
f" [PASS] Confirmed UI consumed and displayed ending token: {COMPLETION_TOKEN}."
)
print(
f" [PASS] Peak RSS: {peak_rss_mib:.2f} MiB across {len(rss_kb_samples)} samples. Exit code: {child.returncode}."
)
def main():
parser = argparse.ArgumentParser(
description="Albedo Go Port Comprehensive Smoke Suite"
)
parser.add_argument(
"--binary",
default=str(ROOT / "cli/bin/albedo"),
help="Path to albedo executable",
)
parser.add_argument(
"--outdir",
default=str(EVIDENCE_DIR),
help="Directory for durable evidence files",
)
args = parser.parse_args()
binary = Path(args.binary).resolve()
if not binary.exists():
print(
f"Error: binary {binary} not found. Run go -C cli build -o bin/albedo ./cmd/albedo first."
)
return 1
out_dir = Path(args.outdir).resolve()
out_dir.mkdir(parents=True, exist_ok=True)
print(f"Starting Albedo Go Port Validation. Binary: {binary}, Evidence: {out_dir}")
test_noninteractive(binary, out_dir)
test_pty_resume(binary, out_dir)
test_pty_session_picker(binary, out_dir)
test_pty_unconfigured_login(binary, out_dir)
test_pty_standalone_login_cancel(binary, out_dir)
test_pty_standalone_login_save_flow(binary, out_dir)
test_pty_tiny_terminal(binary, out_dir)
test_sse_retention_and_memory(binary, out_dir)
summary = {
"status": "PASS",
"timestamp": time.time(),
"binary": str(binary),
"evidence_dir": str(out_dir),
"completion_token": COMPLETION_TOKEN,
"checks": [
"noninteractive_sessions",
"noninteractive_sessions_json",
"noninteractive_send",
"noninteractive_stop",
"noninteractive_daemon",
"noninteractive_login_rejection",
"noninteractive_unknown_command",
"pty_resume_120x40",
"pty_bare_picker",
"pty_unconfigured_login",
"pty_tiny_40x10_width_bounds",
"sse_stream_retention_memory_completion_token",
],
}
(out_dir / "summary.json").write_text(json.dumps(summary, indent=2))
print("\n=======================================================")
print(
"ALL COMPREHENSIVE INTEGRATION & SMOKE CHECKS PASSED! (\u3063\u02d8\u25e1\u02d8)\u3063"
)
print(f"Durable evidence captured in: {out_dir}")
print("=======================================================\n")
return 0
if __name__ == "__main__":
import sys
sys.exit(main())