tmux-amq/tests/test_broker.py

from tamq.broker import BrokerIdentity, InputBroker
from tamq.store import Store


def test_broker_preserves_operator_identity_and_provenance(tmp_path, monkeypatch):
    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
    store = Store(tmp_path / "queue.sqlite3")
    broker = InputBroker(store, BrokerIdentity("tmux-amq-42", "net-kingdom"))
    assert broker.inspect_operator_line("To:railiance-platform: hello") is not None
    row = store.list()[0]
    assert row["sender_repo"] == "net-kingdom"
    assert row["endpoint_id"] == "tmux-amq-42"
    assert row["provenance"] == "operator_input"


def test_broker_accepts_worker_output_with_distinct_provenance(tmp_path, monkeypatch):
    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
    store = Store(tmp_path / "queue.sqlite3")
    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
    assert broker.inspect_worker_line("To:railiance-platform: hello") is None
    assert broker.inspect_worker_line("") is not None
    assert store.list()[0]["provenance"] == "worker_output"


def test_broker_accepts_case_insensitive_framed_worker_output(tmp_path, monkeypatch):
    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
    store = Store(tmp_path / "queue.sqlite3")
    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
    assert broker.inspect_worker_line("• tO:railiance-platform: hello") is None
    assert broker.inspect_worker_line("") is not None
    row = store.list()[0]
    assert row["body"] == "hello"
    assert row["provenance"] == "worker_output"


def test_broker_collects_worker_followup_lines_until_empty(tmp_path, monkeypatch):
    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
    store = Store(tmp_path / "queue.sqlite3")
    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))

    assert broker.inspect_worker_line("• To:railiance-platform: Review this") is None
    assert broker.inspect_worker_line("  Context: AUTH-WP-4.") is None
    assert broker.inspect_worker_line("  Reply with accepted or blocked.") is None
    assert broker.inspect_worker_line("") is not None

    row = store.list()[0]
    assert row["body"] == (
        "Review this\nContext: AUTH-WP-4.\nReply with accepted or blocked."
    )
    events = store.protocol_events()
    assert [event["event_type"] for event in events] == [
        "worker.block_started",
        "message.accepted",
        "worker.block_closed",
    ]
    assert events[-1]["message_id"] == row["message_id"]
    assert events[-1]["outcome"] == "accepted"
    assert events[-1]["detail"] == '{"line_count":3,"reason":"empty_line"}'


def test_new_worker_address_flushes_previous_block(tmp_path, monkeypatch):
    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
    store = Store(tmp_path / "queue.sqlite3")
    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
    broker.inspect_worker_line("To:first: one")
    assert broker.inspect_worker_line("To:second: two") is not None
    assert broker.flush_worker_message() is not None
    assert [(row["target_repo"], row["body"]) for row in store.list()] == [
        ("first", "one"),
        ("second", "two"),
    ]


def test_case_insensitive_delivery_envelopes_are_not_rescanned(tmp_path):
    store = Store(tmp_path / "queue.sqlite3")
    broker = InputBroker(store, BrokerIdentity("ep", "net-kingdom"))
    assert broker.inspect_worker_line("fRoM:railiance-platform: hello") is None
    assert store.line_state("ep", "net-kingdom")["output"] == 0


def test_worker_cmd_is_inert_but_operator_cmd_changes_mode(tmp_path):
    store = Store(tmp_path / "queue.sqlite3")
    store.register_endpoint("ep", 9, "tamq", ["repo"], "output")
    notices = []
    broker = InputBroker(
        store, BrokerIdentity("ep", "repo"), notify=notices.append
    )
    broker.inspect_worker_line("Cmd: mode=trigger")
    assert store.endpoint("ep")["delivery_mode"] == "output"
    broker.inspect_operator_line("cMD: MODE=TRIGGER")
    assert store.endpoint("ep")["delivery_mode"] == "trigger"
    assert notices[-1] == "From:tamq: Mode set to trigger."
    command_events = store.protocol_events(event_type="command.applied")
    assert len(command_events) == 1
    assert command_events[0]["delivery_mode"] == "trigger"
    assert command_events[0]["provenance"] == "operator_input"


def test_operator_command_names_are_case_insensitive(tmp_path):
    store = Store(tmp_path / "queue.sqlite3")
    store.register_endpoint("ep", 9, "tamq", ["repo"], "output")
    notices = []
    broker = InputBroker(
        store, BrokerIdentity("ep", "repo"), notify=notices.append
    )
    broker.inspect_operator_line("CMD: MAXMSG=12")
    assert store.line_state("ep", "repo")["maxmsg"] == 12
    broker.inspect_operator_line("cmd: RESET-LIMITS")
    assert notices[-1] == "From:tamq: Limits reset for this terminal."


def test_limits_block_at_equality_and_reset_current_window(tmp_path, monkeypatch):
    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
    store = Store(tmp_path / "queue.sqlite3")
    notices = []
    broker = InputBroker(
        store,
        BrokerIdentity("ep", "repo"),
        limits=(1, 10, 10),
        notify=notices.append,
    )
    assert broker.inspect_operator_line("To:target: first") is not None
    assert broker.inspect_operator_line("To:target: blocked") is None
    assert "Running linecount 1 limited by 1 lines of messages" in notices[-1]
    assert len(store.list()) == 1
    broker.inspect_operator_line("Cmd: reset-limits")
    assert broker.inspect_operator_line("To:target: after reset") is not None


def test_input_limit_includes_the_current_operator_line(tmp_path, monkeypatch):
    monkeypatch.setattr("tamq.broker.validate_targets", lambda targets: None)
    store = Store(tmp_path / "queue.sqlite3")
    notices = []
    broker = InputBroker(
        store,
        BrokerIdentity("ep", "repo"),
        limits=(8, 1, 32768),
        notify=notices.append,
    )
    assert broker.inspect_operator_line("To:target: blocked") is None
    assert notices[-1] == (
        "From:tamq: Messaging blocked! Running linecount 1 limited by 1 lines "
        "of input in this terminal. Use 'Cmd: reset-limits' to unblock."
    )