Godopu
ec61511699
- Introduce finalize_unfinished in multi-agent-mux-delegate-job for fail-closed atomic transitions - Enforce CAS dispatch_token and attempt_id pinning across engine, CLI, and lifecycle scripts - Integrate Herdr wait_for_agent_idle turn handshake to prevent prompt loss and race conditions - Resolve bash 3.2 set -u empty array expansion compatibility in lib.sh - Preserve Issue #5 R2 contract in CAS transition engine (engine.py, publish_event.py, service.py) - Add comprehensive regression suites (test_issue6_fail_closed.py, test_issue6_turn_handshake.py) - Bump lockstep framework and skills to v4.4.1 across lib.sh, eager.py, VERSIONS.md, 8 SKILL.md - Synchronize documentation (IMPROVEMENTS.md, MESSAGING.md, state-transitions.md, task.md, phase5-e2e.md) - Unanimous PASS achieved across 4 reviewer team leaders (Claude, Codex, OpenCode, Grok) Closes #6
36 lines
1.4 KiB
Python
36 lines
1.4 KiB
Python
import os
|
|
import subprocess
|
|
import pytest
|
|
|
|
def test_send_keys_safe_b8_verification(tmp_path):
|
|
"""Verify that send_keys_safe for agy sessions goes through submission verification and does not return 0 blindly on failure."""
|
|
# Test script that mocks herdr commands and tests send_keys_safe behavior for agy
|
|
test_script = f"""#!/usr/bin/env bash
|
|
set -euo pipefail
|
|
SKILL_DIR="{os.path.abspath('.agents/skills')}"
|
|
source "$SKILL_DIR/lib.sh"
|
|
|
|
# Mock _pane_quiescent to succeed
|
|
_pane_quiescent() {{ return 0; }}
|
|
_pane_dialog_open() {{ return 1; }}
|
|
|
|
# Mock _sks_herdr commands
|
|
_sks_herdr() {{
|
|
if [ "${{1:-}}" = "agent" ] && [ "${{2:-}}" = "prompt" ]; then
|
|
echo "unknown command: agent prompt" >&2
|
|
return 1
|
|
fi
|
|
return 0
|
|
}}
|
|
|
|
# Mock _pane_capture to simulate fixed pane content (no execution token, no pane change)
|
|
_pane_capture() {{ echo "static content"; }}
|
|
|
|
# Running send_keys_safe for agy session when submission is NOT accepted
|
|
send_keys_safe "test-workspace-creator-agy" "test prompt" "job-123"
|
|
"""
|
|
res = subprocess.run(["bash", "-c", test_script], capture_output=True, text=True)
|
|
# Since submission failed after 3 tries (pane static, no token), it must return exit code 4 (not 0)
|
|
assert res.returncode == 4, f"Expected returncode 4 on submission failure for agy, got {res.returncode}. Output: {res.stdout} {res.stderr}"
|
|
assert "Enter not accepted after 3 tries" in res.stderr
|