139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
|
|
from band_aider.judgement.human_feedback_reasonable_vibe_check import (
|
|
DEFAULT_FEEDBACK_DURATION_SECONDS,
|
|
MAX_FEEDBACK_DURATION_SECONDS,
|
|
HumanFeedbackReasonableVibeCheck,
|
|
)
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.ui.dto.human_feedback import HumanFeedbackKind
|
|
|
|
|
|
class _JsonClient(CompletionClient):
|
|
"""Serve explicit feedback-classification responses without transport."""
|
|
|
|
def __init__(self, response: str | list[str]) -> None:
|
|
"""Bind a fixed answer or a finite correction script."""
|
|
self.response = response
|
|
self.prompts: list[str] = []
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
return None
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
return ()
|
|
|
|
def complete(self, prompt: str, logger=None) -> str:
|
|
"""Retain every prompt before returning the next feedback response."""
|
|
self.prompts.append(prompt)
|
|
if isinstance(self.response, list):
|
|
return self.response.pop(0)
|
|
return self.response
|
|
|
|
|
|
def test_human_feedback_reasonable_vibe_check_normalizes_advice() -> None:
|
|
client = _JsonClient(
|
|
json.dumps(
|
|
{
|
|
"outcome": "advice",
|
|
"reason": "Reusable redirect.",
|
|
"additional": {
|
|
"advice": "do not add more files",
|
|
"duration_seconds": 7200,
|
|
},
|
|
}
|
|
)
|
|
)
|
|
check = HumanFeedbackReasonableVibeCheck(
|
|
client=client,
|
|
request_id="a:b",
|
|
request_md5="a",
|
|
response_md5="b",
|
|
original_prompt='Return strict JSON: {"needs_edit":false,"reason":"why"}.',
|
|
repeated_response_tail="same",
|
|
human_guidance="do not add more files",
|
|
source_action="resend",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
|
|
assert check() is True
|
|
feedback = check.to_feedback_dto(recorded_at=10.0)
|
|
|
|
assert feedback.kind is HumanFeedbackKind.Advice
|
|
assert feedback.advice == "do not add more files"
|
|
assert feedback.source_guidance == "do not add more files"
|
|
assert feedback.duration_seconds == 7200
|
|
assert feedback.expires_at == 7210.0
|
|
assert 'Return exactly this JSON shape:\n{"outcome":"ignore|ask|advice"' in client.prompts[0]
|
|
assert '{"needs_edit":false' not in client.prompts[0]
|
|
assert '"needs_edit":false,"reason":"why"' not in client.prompts[0]
|
|
assert 'Return strict JSON:' not in client.prompts[0]
|
|
assert '\\==== BEGIN ORIGINAL REQUEST ====\n---REDACTED---\n\\==== END ORIGINAL REQUEST ====' in client.prompts[0]
|
|
|
|
|
|
def test_human_feedback_corrects_duration_and_incoherent_ignore_advice() -> None:
|
|
client = _JsonClient(
|
|
[json.dumps(
|
|
{
|
|
"outcome": "ignore",
|
|
"reason": "Accept the repeated answer.",
|
|
"additional": {
|
|
"advice": "should be cleared",
|
|
"duration_seconds": MAX_FEEDBACK_DURATION_SECONDS + 999,
|
|
},
|
|
}
|
|
), json.dumps({"outcome": "ignore", "reason": "Accept the repeated answer.",
|
|
"additional": {"advice": "", "duration_seconds": MAX_FEEDBACK_DURATION_SECONDS}})]
|
|
)
|
|
check = HumanFeedbackReasonableVibeCheck(
|
|
client=client,
|
|
request_id="a:b",
|
|
request_md5="a",
|
|
response_md5="b",
|
|
original_prompt="prompt",
|
|
repeated_response_tail="same",
|
|
human_guidance="ignore this for a month",
|
|
source_action="resend",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
|
|
check()
|
|
feedback = check.to_feedback_dto(recorded_at=1.0)
|
|
|
|
assert feedback.kind is HumanFeedbackKind.Ignore
|
|
assert feedback.advice == ""
|
|
assert feedback.duration_seconds == MAX_FEEDBACK_DURATION_SECONDS
|
|
assert len(client.prompts) == 2
|
|
|
|
|
|
def test_human_feedback_reasonable_vibe_check_falls_back_to_ask_on_llm_failure() -> None:
|
|
class _ExplodingClient(_JsonClient):
|
|
def complete(self, prompt: str, logger=None) -> str:
|
|
raise RuntimeError("classification failed")
|
|
|
|
check = HumanFeedbackReasonableVibeCheck(
|
|
client=_ExplodingClient("{}"),
|
|
request_id="a:b",
|
|
request_md5="a",
|
|
response_md5="b",
|
|
original_prompt="prompt",
|
|
repeated_response_tail="same",
|
|
human_guidance="always ask me when this repeats",
|
|
source_action="resend",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
|
|
assert check() is False
|
|
feedback = check.to_feedback_dto(recorded_at=5.0)
|
|
|
|
assert feedback.kind is HumanFeedbackKind.Ask
|
|
assert feedback.duration_seconds == DEFAULT_FEEDBACK_DURATION_SECONDS
|