632 lines
25 KiB
Python
632 lines
25 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from multiprocessing.queues import Queue
|
|
from queue import Empty
|
|
from typing import cast
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
|
|
from band_aider.ai_client import escalation_utils
|
|
from band_aider.ai_client.cycle_jacker import (
|
|
CycleJacker,
|
|
CycleJackerAborted,
|
|
cycle,
|
|
)
|
|
from band_aider.ai_client.human_feedback_store import HumanFeedbackStore
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.ui.dto.human_feedback import HumanFeedbackDTO, HumanFeedbackKind
|
|
from band_aider.ui.dto.human_intervention import CycleInterruptCommandDTO, HumanInterventionRequestDTO, HumanInterventionResponseDTO, InterventionSuppressionCommandDTO
|
|
from band_aider.ui.dto.model_escalation import ModelEscalationDTO
|
|
from band_aider.ui.queue_status_publisher import QueueHumanCommandReceiver
|
|
from band_aider.ui.status_publisher import HumanCommandReceiver, StatusPublisher
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _RecordingClient(CompletionClient):
|
|
def __init__(
|
|
self,
|
|
responses: list[str],
|
|
configs: tuple[OpenAIConfig, ...] = (),
|
|
guided_response: str | None = None,
|
|
classifier_advice: str | None = None,
|
|
) -> None:
|
|
self.responses = responses
|
|
self.prompts: list[str] = []
|
|
self._configs = configs
|
|
self.guided_response = guided_response
|
|
self.classifier_advice = classifier_advice
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
return None
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
return self._configs
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return a stationary response without invoking an external provider."""
|
|
|
|
del logger
|
|
self.prompts.append(prompt)
|
|
if "You are classifying human feedback" in prompt:
|
|
guidance = prompt.rsplit("Human feedback:\n", maxsplit=1)[-1].strip()
|
|
return json.dumps(
|
|
{
|
|
"outcome": "advice",
|
|
"reason": "Reusable human redirect.",
|
|
"additional": {
|
|
"advice": self.classifier_advice or guidance,
|
|
"duration_seconds": 3600,
|
|
},
|
|
}
|
|
)
|
|
if "Human guidance" in prompt and self.guided_response is not None:
|
|
return self.guided_response
|
|
if len(self.responses) == 1:
|
|
return self.responses[0]
|
|
return self.responses.pop(0)
|
|
|
|
|
|
class _Publisher(StatusPublisher):
|
|
def __init__(self) -> None:
|
|
self.events: list[object] = []
|
|
|
|
def publish(self, event: object) -> None:
|
|
self.events.append(event)
|
|
|
|
|
|
class _Receiver(HumanCommandReceiver):
|
|
def __init__(self, action: str = "resend", guidance: str = "fix the stale import") -> None:
|
|
self.action = action
|
|
self.guidance = guidance
|
|
|
|
def consume_pending(self) -> None:
|
|
return None
|
|
|
|
def wait_for(self, request_id: str) -> HumanInterventionResponseDTO:
|
|
return HumanInterventionResponseDTO(
|
|
request_id=request_id,
|
|
guidance=self.guidance,
|
|
action=self.action,
|
|
)
|
|
|
|
|
|
class _MemoryCommandQueue(Queue):
|
|
"""Nominal multiprocessing queue with preloaded in-memory commands."""
|
|
|
|
def __init__(self, items: list[object]) -> None:
|
|
self._items = list(items)
|
|
|
|
def get_nowait(self) -> object:
|
|
"""Return the next command without blocking or allocating OS primitives."""
|
|
|
|
if not self._items:
|
|
raise Empty
|
|
return self._items.pop(0)
|
|
|
|
def get(self, block: bool = True, timeout: float | None = None) -> object:
|
|
"""Return the next expected command for the correlated intervention."""
|
|
|
|
del block, timeout
|
|
if not self._items:
|
|
raise AssertionError("queue unexpectedly empty")
|
|
return self._items.pop(0)
|
|
|
|
|
|
def _configs() -> tuple[OpenAIConfig, ...]:
|
|
return (
|
|
OpenAIConfig(model="local", api_base="http://local"),
|
|
OpenAIConfig(model="cloud-1", api_base="http://cloud-1"),
|
|
OpenAIConfig(model="cloud-2", api_base="http://cloud-2"),
|
|
)
|
|
|
|
|
|
def _store() -> HumanFeedbackStore:
|
|
"""Return a fresh human-feedback store over an explicit in-memory repository."""
|
|
|
|
adapter = MemoryRepoExec()
|
|
return HumanFeedbackStore(adapter.root, repo_exec=adapter)
|
|
|
|
|
|
def _formatter_builder() -> MarkdownPromptFormatterBuilder:
|
|
"""Return the explicit prompt formatter required by each test client."""
|
|
|
|
return MarkdownPromptFormatterBuilder()
|
|
|
|
|
|
def test_first_response_does_not_escalate() -> None:
|
|
inner = _RecordingClient(["ok"], configs=_configs())
|
|
publisher = _Publisher()
|
|
client = CycleJacker(inner=inner, event_publisher=publisher, feedback_store=_store(), repeat_threshold=1, prompt_formatter_builder=_formatter_builder())
|
|
|
|
assert client.complete("prompt") == "ok"
|
|
assert publisher.events == []
|
|
|
|
|
|
def test_second_exact_pair_escalates() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs())
|
|
cloud = _RecordingClient(["fixed"])
|
|
publisher = _Publisher()
|
|
client = CycleJacker(inner=inner, event_publisher=publisher, feedback_store=_store(), repeat_threshold=1, escalation_clients=(cloud,), prompt_formatter_builder=_formatter_builder())
|
|
|
|
assert client.complete("prompt") == "same"
|
|
assert client.complete("prompt") == "fixed"
|
|
assert cloud.prompts == ["prompt"]
|
|
escalations = [event for event in publisher.events if isinstance(event, ModelEscalationDTO)]
|
|
assert [event.state for event in escalations] == [
|
|
"detected",
|
|
"escalating",
|
|
"resolved",
|
|
]
|
|
assert escalations[0].trigger_name == "CycleJacker"
|
|
assert escalations[0].prompt_preview == "prompt"
|
|
assert escalations[0].occurred_at > 0
|
|
|
|
|
|
def test_changed_response_resets_cycle_count() -> None:
|
|
inner = _RecordingClient(["a", "b", "b"], configs=())
|
|
publisher = _Publisher()
|
|
client = CycleJacker(
|
|
inner=inner,
|
|
event_publisher=publisher,
|
|
command_receiver=_Receiver(action="defer"),
|
|
feedback_store=_store(),
|
|
repeat_threshold=1,
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
|
|
assert client.complete("prompt") == "a"
|
|
assert client.complete("prompt") == "b"
|
|
assert client.complete("prompt") == "b"
|
|
record = client.feedback_store.latest_cycle_record(hashlib.md5(b"prompt").hexdigest())
|
|
assert record is not None
|
|
assert record.repeat_count == 1
|
|
assert [event.state for event in publisher.events if isinstance(event, ModelEscalationDTO)] == [
|
|
"detected",
|
|
"human_required",
|
|
]
|
|
|
|
|
|
def test_same_response_for_different_prompt_does_not_escalate() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs())
|
|
publisher = _Publisher()
|
|
client = CycleJacker(inner=inner, event_publisher=publisher, feedback_store=_store(), repeat_threshold=1, prompt_formatter_builder=_formatter_builder())
|
|
|
|
assert client.complete("one") == "same"
|
|
assert client.complete("two") == "same"
|
|
assert publisher.events == []
|
|
|
|
|
|
def test_response_tail_is_bounded_and_md5_is_stable() -> None:
|
|
inner = _RecordingClient(["x" * 1100], configs=_configs())
|
|
client = CycleJacker(inner=inner, feedback_store=_store(), repeat_threshold=1, prompt_formatter_builder=_formatter_builder())
|
|
|
|
record = client._record("prompt", "x" * 1100, "local")
|
|
|
|
assert record.response_tail == "x" * 1000
|
|
assert record.request_md5 == hashlib.md5(b"prompt").hexdigest()
|
|
assert record.response_md5 == hashlib.md5(("x" * 1100).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def test_human_interrupt_flag_forces_immediate_human_guidance() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs(), guided_response="guided")
|
|
publisher = _Publisher()
|
|
client = CycleJacker(
|
|
inner=inner,
|
|
event_publisher=publisher,
|
|
command_receiver=_Receiver(guidance="prefer concise fix"),
|
|
feedback_store=_store(),
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
cycle.human_interrupt = False
|
|
cycle.human_interrupt = True
|
|
|
|
assert client.complete("prompt") == "guided"
|
|
assert cycle.human_interrupt is False
|
|
assert any(isinstance(event, ModelEscalationDTO) and event.state == "human_required" for event in publisher.events)
|
|
|
|
|
|
def test_queue_command_receiver_consumes_cycle_interrupt_before_completion() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs(), guided_response="guided")
|
|
publisher = _Publisher()
|
|
|
|
receiver = QueueHumanCommandReceiver(
|
|
_MemoryCommandQueue(
|
|
[
|
|
CycleInterruptCommandDTO(),
|
|
HumanInterventionResponseDTO(
|
|
request_id="request-1",
|
|
guidance="prefer concise fix",
|
|
action="resend",
|
|
),
|
|
]
|
|
)
|
|
)
|
|
client = CycleJacker(
|
|
inner=inner,
|
|
event_publisher=publisher,
|
|
command_receiver=receiver,
|
|
feedback_store=_store(),
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
cycle.human_interrupt = False
|
|
|
|
with patch("band_aider.ai_client.escalation_utils.uuid4", return_value="request-1"):
|
|
assert client.complete("prompt") == "guided"
|
|
assert cycle.human_interrupt is False
|
|
assert any(isinstance(event, ModelEscalationDTO) and event.state == "human_required" for event in publisher.events)
|
|
|
|
|
|
def test_prompt_with_guidance_declares_human_priority() -> None:
|
|
guided = escalation_utils.guided_prompt("prompt", "do X", _formatter_builder())
|
|
|
|
assert "highest priority" in guided
|
|
assert "authoritative instruction" in guided
|
|
|
|
|
|
def test_model_name_comes_from_openai_configs() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs())
|
|
cloud = _RecordingClient(["fixed"])
|
|
publisher = _Publisher()
|
|
client = CycleJacker(inner=inner, event_publisher=publisher, feedback_store=_store(), repeat_threshold=1, escalation_clients=(cloud,), prompt_formatter_builder=_formatter_builder())
|
|
|
|
client.complete("prompt")
|
|
client.complete("prompt")
|
|
|
|
models = [event.active_model for event in publisher.events if isinstance(event, ModelEscalationDTO)]
|
|
assert models == ["local", "cloud-1", "cloud-1"]
|
|
|
|
|
|
def test_identical_cloud_answer_advances_to_next_model() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs())
|
|
cloud_1 = _RecordingClient(["same"])
|
|
cloud_2 = _RecordingClient(["fixed"])
|
|
client = CycleJacker(inner=inner, feedback_store=_store(), repeat_threshold=1, escalation_clients=(cloud_1, cloud_2), prompt_formatter_builder=_formatter_builder())
|
|
|
|
client.complete("prompt")
|
|
|
|
assert client.complete("prompt") == "same"
|
|
assert client.complete("prompt") == "fixed"
|
|
assert cloud_1.prompts == ["prompt"]
|
|
assert cloud_2.prompts == ["prompt"]
|
|
|
|
|
|
def test_each_cycle_model_gets_twelve_responses_before_human_intervention() -> None:
|
|
"""An unchanged last-cloud response retains its full allowance before human help is requested."""
|
|
|
|
local = _RecordingClient(["same"], configs=_configs())
|
|
cloud_1 = _RecordingClient(["same"])
|
|
cloud_2 = _RecordingClient(["same"], guided_response="hinted response")
|
|
publisher = _Publisher()
|
|
client = CycleJacker(
|
|
local, escalation_clients=(cloud_1, cloud_2), feedback_store=_store(),
|
|
event_publisher=publisher, command_receiver=_Receiver(), prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
|
|
assert tuple(map(lambda _: client.complete("prompt"), range(36))) == ("same",) * 36
|
|
assert cloud_1.prompts == cloud_2.prompts == ["prompt"] * 12
|
|
assert not any(map(lambda event: isinstance(event, HumanInterventionRequestDTO), publisher.events))
|
|
|
|
assert client.complete("prompt") == "hinted response"
|
|
assert len(local.prompts) == 37
|
|
assert len(cloud_2.prompts) == 13
|
|
assert len(tuple(filter(lambda event: isinstance(event, HumanInterventionRequestDTO), publisher.events))) == 1
|
|
|
|
|
|
def test_cycle_threshold_applies_to_every_rung_and_wraps_after_human_responses() -> None:
|
|
"""A two-response allowance produces local, cloud, and human blocks, then returns to local."""
|
|
|
|
local = _RecordingClient(["same"], configs=_configs())
|
|
cloud_1 = _RecordingClient(["same"])
|
|
cloud_2 = _RecordingClient(["same"])
|
|
publisher = _Publisher()
|
|
client = CycleJacker(
|
|
local, escalation_clients=(cloud_1, cloud_2), repeat_threshold=2, feedback_store=_store(),
|
|
event_publisher=publisher, command_receiver=_Receiver(guidance="same"), prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
|
|
replies = tuple(map(lambda _: client.complete("prompt"), range(9)))
|
|
|
|
assert replies == ("same",) * 9
|
|
assert len(local.prompts) == 9
|
|
assert len(cloud_1.prompts) == 2
|
|
assert len(cloud_2.prompts) == 3
|
|
assert tuple(map(lambda event: cast(HumanInterventionRequestDTO, event).mode, filter(lambda event: isinstance(event, HumanInterventionRequestDTO), publisher.events))) == ("hint", "direct_response")
|
|
assert tuple(map(lambda event: cast(ModelEscalationDTO, event).config_index, filter(lambda event: isinstance(event, ModelEscalationDTO) and event.state == "escalating", publisher.events))) == (1, 1, 2, 2, 2)
|
|
|
|
|
|
def test_suppressed_cycle_repeats_full_twelve_response_model_ladders() -> None:
|
|
"""Skipping human intervention must not shorten a cloud block or add a thirteenth local slot."""
|
|
|
|
local = _RecordingClient(["same"], configs=_configs())
|
|
cloud_1 = _RecordingClient(["same"])
|
|
cloud_2 = _RecordingClient(["same"])
|
|
publisher = _Publisher()
|
|
client = CycleJacker(
|
|
local, escalation_clients=(cloud_1, cloud_2), feedback_store=_store(),
|
|
event_publisher=publisher, prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
with patch.object(cycle, "intervention_suppressed", True):
|
|
replies = tuple(map(lambda _: client.complete("prompt"), range(72)))
|
|
|
|
assert replies == ("same",) * 72
|
|
assert len(local.prompts) == 72
|
|
assert cloud_1.prompts == cloud_2.prompts == ["prompt"] * 24
|
|
assert not any(map(lambda event: isinstance(event, HumanInterventionRequestDTO), publisher.events))
|
|
|
|
|
|
def test_cycle_resume_preserves_remaining_cloud_allowance() -> None:
|
|
"""A new decorator instance resumes the same exact-pair count at a cloud boundary."""
|
|
|
|
local = _RecordingClient(["same"], configs=_configs())
|
|
cloud_1 = _RecordingClient(["same"])
|
|
cloud_2 = _RecordingClient(["same"])
|
|
store = _store()
|
|
client = CycleJacker(local, escalation_clients=(cloud_1, cloud_2), feedback_store=store, prompt_formatter_builder=_formatter_builder())
|
|
tuple(map(lambda _: client.complete("prompt"), range(23)))
|
|
resumed = CycleJacker(local, escalation_clients=(cloud_1, cloud_2), feedback_store=store, prompt_formatter_builder=_formatter_builder())
|
|
|
|
assert resumed.complete("prompt") == "same"
|
|
assert len(cloud_1.prompts) == 12
|
|
assert cloud_2.prompts == []
|
|
assert resumed.complete("prompt") == "same"
|
|
assert len(cloud_1.prompts) == 12
|
|
assert cloud_2.prompts == ["prompt"]
|
|
|
|
|
|
@pytest.mark.parametrize("repeat_threshold", (0, -1))
|
|
def test_cycle_requires_a_positive_tier_allowance(repeat_threshold: int) -> None:
|
|
"""A configured allowance must admit at least one response per tier."""
|
|
|
|
with pytest.raises(ValueError, match="repeat_threshold must be positive"):
|
|
CycleJacker(_RecordingClient(["same"]), repeat_threshold=repeat_threshold, feedback_store=_store(), prompt_formatter_builder=_formatter_builder())
|
|
|
|
|
|
def test_exhausted_models_publish_human_intervention_and_guidance_modifies_prompt() -> None:
|
|
guidance = (
|
|
"Create the FileManager first.\n"
|
|
"Use file_manager.repo_exec.\n"
|
|
"Keep mementos=[memento].\n"
|
|
"Do not use memento=.\n"
|
|
"Do not call memento.enter()."
|
|
)
|
|
inner = _RecordingClient(
|
|
["same"],
|
|
configs=_configs(),
|
|
guided_response="guided",
|
|
classifier_advice="Ensure the Memento in _check is created using file_manager.repo_exec.",
|
|
)
|
|
cloud_1 = _RecordingClient(["same"])
|
|
cloud_2 = _RecordingClient(["same"], guided_response="guided")
|
|
|
|
publisher = _Publisher()
|
|
client = CycleJacker(
|
|
inner=inner,
|
|
event_publisher=publisher,
|
|
command_receiver=_Receiver(guidance=guidance),
|
|
feedback_store=_store(),
|
|
repeat_threshold=1,
|
|
escalation_clients=(cloud_1, cloud_2),
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
|
|
client.complete("prompt")
|
|
|
|
assert client.complete("prompt") == "same"
|
|
assert client.complete("prompt") == "same"
|
|
assert client.complete("prompt") == "guided"
|
|
assert "== BEGIN HUMAN GUIDANCE ==" in cloud_2.prompts[-1]
|
|
assert "==== BEGIN ORIGINAL REQUEST ====" in cloud_2.prompts[-1]
|
|
assert "==== BEGIN AUTHORITATIVE GUIDANCE ====" in cloud_2.prompts[-1]
|
|
assert "prompt" in cloud_2.prompts[-1]
|
|
assert guidance in cloud_2.prompts[-1]
|
|
assert cloud_2.prompts[-1].count("file_manager.repo_exec") == 1
|
|
assert "Ensure the Memento in _check is created using file_manager.repo_exec." not in cloud_2.prompts[-1]
|
|
assert len(inner.prompts) == 4
|
|
assert len(cloud_2.prompts) == 2
|
|
assert not any("classifying human feedback" in prompt for prompt in inner.prompts)
|
|
assert any(event.__class__.__name__ == "HumanInterventionRequestDTO" for event in publisher.events)
|
|
|
|
|
|
def test_retry_without_hint_reuses_original_prompt() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs())
|
|
cloud_1 = _RecordingClient(["same"])
|
|
cloud_2 = _RecordingClient(["same"])
|
|
retried = _RecordingClient(["fresh"])
|
|
|
|
client = CycleJacker(
|
|
inner=inner,
|
|
command_receiver=_Receiver(action="retry", guidance="ignored"),
|
|
feedback_store=_store(),
|
|
repeat_threshold=1,
|
|
escalation_clients=(cloud_1, retried),
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
|
|
client.complete("prompt")
|
|
|
|
assert client.complete("prompt") == "same"
|
|
assert client.complete("prompt") == "fresh"
|
|
assert retried.prompts == ["prompt"]
|
|
|
|
|
|
def test_deferred_intervention_restarts_cycles_and_explicit_abort_stops() -> None:
|
|
"""Ordinary rendering also survives deferral while preserving the explicit abort command."""
|
|
overflow_clients = (_RecordingClient(["same"]), _RecordingClient(["same"]))
|
|
deferred = CycleJacker(
|
|
inner=_RecordingClient(["same"], configs=_configs()),
|
|
command_receiver=_Receiver(action="defer"),
|
|
feedback_store=_store(),
|
|
repeat_threshold=1,
|
|
escalation_clients=overflow_clients,
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
deferred.complete("prompt")
|
|
assert deferred.complete("prompt") == "same"
|
|
assert deferred.complete("prompt") == "same"
|
|
assert deferred.complete("prompt") == "same"
|
|
|
|
aborted = CycleJacker(
|
|
inner=_RecordingClient(["same"], configs=_configs()),
|
|
command_receiver=_Receiver(action="abort"),
|
|
feedback_store=_store(),
|
|
repeat_threshold=1,
|
|
escalation_clients=overflow_clients,
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
aborted.complete("prompt")
|
|
aborted.complete("prompt")
|
|
aborted.complete("prompt")
|
|
try:
|
|
aborted.complete("prompt")
|
|
except CycleJackerAborted:
|
|
pass
|
|
else:
|
|
raise AssertionError("abort should raise")
|
|
|
|
|
|
def test_suppression_command_releases_wait_and_f5_reenables_intervention() -> None:
|
|
"""The UI command works even while the worker is waiting for a correlated response."""
|
|
|
|
with patch.object(cycle, "intervention_suppressed", False), patch.object(cycle, "human_interrupt", False):
|
|
receiver = QueueHumanCommandReceiver(_MemoryCommandQueue([InterventionSuppressionCommandDTO(True)]))
|
|
|
|
response = receiver.wait_for("pending-human-request")
|
|
|
|
assert response.action == "suppress"
|
|
assert response.request_id == "pending-human-request"
|
|
assert cycle.intervention_suppressed is True
|
|
resumed = QueueHumanCommandReceiver(_MemoryCommandQueue([CycleInterruptCommandDTO()]))
|
|
resumed.consume_pending()
|
|
assert cycle.intervention_suppressed is False
|
|
assert cycle.human_interrupt is True
|
|
|
|
|
|
def test_suppressed_cycle_recovery_returns_to_models_without_publishing_help() -> None:
|
|
"""The global suppression choice covers renderer helpers without borrowing stabilization counters."""
|
|
|
|
publisher = _Publisher()
|
|
client = CycleJacker(
|
|
inner=_RecordingClient(["same"]), event_publisher=publisher,
|
|
feedback_store=_store(), repeat_threshold=1, prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
with patch.object(cycle, "intervention_suppressed", True):
|
|
assert client.complete("render") == "same"
|
|
assert client.complete("render") == "same"
|
|
assert client.complete("render") == "same"
|
|
assert not any(map(lambda event: isinstance(event, HumanInterventionRequestDTO), publisher.events))
|
|
|
|
|
|
def test_stored_ignore_suppresses_escalation() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs())
|
|
publisher = _Publisher()
|
|
store = _store()
|
|
client = CycleJacker(inner=inner, event_publisher=publisher, feedback_store=store, prompt_formatter_builder=_formatter_builder())
|
|
|
|
first = client.complete("prompt")
|
|
record = store.latest_cycle_record(hashlib.md5(b"prompt").hexdigest())
|
|
assert first == "same"
|
|
assert record is not None
|
|
store.record_feedback(
|
|
HumanFeedbackDTO(
|
|
request_id=f"{record.request_md5}:{record.response_md5}",
|
|
request_md5=record.request_md5,
|
|
response_md5=record.response_md5,
|
|
kind=HumanFeedbackKind.Ignore,
|
|
reason="This repeated answer is acceptable.",
|
|
source_guidance="ignore this for now",
|
|
source_action="resend",
|
|
recorded_at=1.0,
|
|
expires_at=9999999999.0,
|
|
)
|
|
)
|
|
|
|
assert client.complete("prompt") == "same"
|
|
assert publisher.events == []
|
|
|
|
|
|
def test_stored_advice_cannot_replay_consumed_human_guidance() -> None:
|
|
"""Historical advice remains evidence and cannot steer another completion."""
|
|
inner = _RecordingClient(["same"], configs=_configs(), guided_response="guided-fix")
|
|
cloud = _RecordingClient(["unused"], guided_response="guided-fix")
|
|
store = _store()
|
|
client = CycleJacker(inner=inner, feedback_store=store, repeat_threshold=1, escalation_clients=(cloud,), prompt_formatter_builder=_formatter_builder())
|
|
|
|
client.complete("prompt")
|
|
record = store.latest_cycle_record(hashlib.md5(b"prompt").hexdigest())
|
|
assert record is not None
|
|
store.record_feedback(
|
|
HumanFeedbackDTO(
|
|
request_id=f"{record.request_md5}:{record.response_md5}",
|
|
request_md5=record.request_md5,
|
|
response_md5=record.response_md5,
|
|
kind=HumanFeedbackKind.Advice,
|
|
reason="Reusable redirect.",
|
|
source_guidance="Use X. Keep Y. Do not introduce Z.",
|
|
source_action="resend",
|
|
advice="Use X.",
|
|
recorded_at=1.0,
|
|
expires_at=9999999999.0,
|
|
)
|
|
)
|
|
|
|
assert client.complete("prompt") == "unused"
|
|
assert cloud.prompts == ["prompt"]
|
|
latest = store.latest_feedback(f"{record.request_md5}:{record.response_md5}")
|
|
assert latest is not None
|
|
assert latest.application_count == 0
|
|
assert latest.last_applied_request_md5 == ""
|
|
|
|
|
|
def test_stored_ask_skips_model_escalation_and_goes_straight_to_human() -> None:
|
|
inner = _RecordingClient(["same"], configs=_configs())
|
|
cloud = _RecordingClient(["cloud-should-not-run"])
|
|
publisher = _Publisher()
|
|
store = _store()
|
|
client = CycleJacker(
|
|
inner=inner,
|
|
event_publisher=publisher,
|
|
command_receiver=_Receiver(action="abort"),
|
|
feedback_store=store,
|
|
repeat_threshold=1,
|
|
escalation_clients=(cloud,),
|
|
prompt_formatter_builder=_formatter_builder(),
|
|
)
|
|
|
|
client.complete("prompt")
|
|
record = store.latest_cycle_record(hashlib.md5(b"prompt").hexdigest())
|
|
assert record is not None
|
|
store.record_feedback(
|
|
HumanFeedbackDTO(
|
|
request_id=f"{record.request_md5}:{record.response_md5}",
|
|
request_md5=record.request_md5,
|
|
response_md5=record.response_md5,
|
|
kind=HumanFeedbackKind.Ask,
|
|
reason="Always interrupt a human for this case.",
|
|
source_guidance="always ask me",
|
|
source_action="resend",
|
|
recorded_at=1.0,
|
|
expires_at=9999999999.0,
|
|
)
|
|
)
|
|
|
|
try:
|
|
client.complete("prompt")
|
|
except CycleJackerAborted:
|
|
pass
|
|
else:
|
|
raise AssertionError("stored ask should interrupt a human before any model escalation")
|
|
assert cloud.prompts == []
|
|
assert [event.state for event in publisher.events if isinstance(event, ModelEscalationDTO)] == [
|
|
"human_required",
|
|
]
|