Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions docs/development/testing-and-quality.md
Original file line number Diff line number Diff line change
Expand Up @@ -1009,10 +1009,13 @@ For focused thin/brief prompt-decision regression, use
explicit release qualification. It defaults to no calls; missing credentials
report `skipped`, not a live pass. With securely injected `ARK_API_KEY`, it uses
Doubao evolving for two independent repetitions of quiet-work, notifying-wait,
quiet-wait and required-vision-replan cases in each mode. Expected decisions
remain outside model input. All attempts must pass; no answer correction or
retry-until-pass is used. Ordinary pytest only checks the probe and negative
oracles with scripted responses, without provider calls.
quiet-wait, required-vision-replan and typed external-wait fallback cases in
each mode. The fallback case uses the real compact quota projection: its wait
transition already exists, so the host must advance the selected independent
successor and notify rather than authoring another transition. Expected
decisions remain outside model input. All attempts must pass; no answer
correction or retry-until-pass is used. Ordinary pytest only checks the probe
and negative oracles with scripted responses, without provider calls.

This is a synthetic decision-level probe using current generated prompts,
not proof of tool execution, host scheduling, upgrade delivery or full-Goal
Expand Down
33 changes: 17 additions & 16 deletions loopx/control_plane/testing/host_prompt_behavior.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,22 +6,24 @@
from pathlib import Path

from ...heartbeat_prompt import build_heartbeat_prompt
from ..quota.cli_projection import compact_quota_should_run_cli_payload
from ..quota.effective_action import EffectiveAction
from .action_portfolio_scenarios import external_wait_fallback_scenario_source
from .model_tool_behavior import DoubaoExecToolClient


def cases() -> list[dict]:
# Independent semantic oracle: silence does not cancel work; a gate does;
# required vision replan is not terminal closure. Never send expected to
# the model, or derive it from the renderer being qualified.
# required vision replan is not terminal closure. An already-transitioned
# external wait selects its independent fallback for work. Never send
# expected to the model, or derive it from the renderer being qualified.
rows = (
("quiet_work", True, False, False, False, "work"),
("notifying_wait", False, True, False, False, "wait"),
("quiet_wait", False, False, False, False, "wait"),
("vision_replan", True, False, True, False, "replan"),
("external_wait_transition", True, False, False, True, "external_wait"),
("quiet_work", True, False, False, "work"),
("notifying_wait", False, True, False, "wait"),
("quiet_wait", False, False, False, "wait"),
("vision_replan", True, False, True, "replan"),
)
return [{
probes = [{
"id": name,
"packet": {
"ok": True,
Expand All @@ -33,13 +35,6 @@ def cases() -> list[dict]:
"execution_obligation": {"must_attempt_work": work},
"heartbeat_recommendation": {"agent_must_attempt": work},
"autonomous_replan_obligation": {"required": replan},
"external_wait_observation": ({
"schema_version": "typed_external_wait_observation_v0",
"selected_todo_id": "todo_waiting",
"monitor_todo_id": "todo_monitor",
"independent_successor_todo_id": "todo_successor",
"state": "external_review_pending",
} if external_wait else None),
"interaction_contract": {
"user_channel": {"notify": "NOTIFY" if notify else "DONT_NOTIFY"},
"agent_channel": {"delivery_allowed": work and not replan},
Expand All @@ -49,7 +44,13 @@ def cases() -> list[dict]:
"run_history": {"latest_runs": [{"delivery_outcome": "outcome_progress"}]},
},
"expected": {"action": action, "notify": notify, "finish_goal": False},
} for name, work, notify, replan, external_wait, action in rows]
} for name, work, notify, replan, action in rows]
probes.append({
"id": "wait_fallback_work",
"packet": compact_quota_should_run_cli_payload(external_wait_fallback_scenario_source()),
"expected": {"action": "work", "notify": True, "finish_goal": False},
})
return probes


def probe_messages(mode: str, packet: dict) -> list[dict]:
Expand Down
7 changes: 6 additions & 1 deletion tests/control_plane/test_host_prompt_behavior.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
{"action": "wait", "notify": True, "finish_goal": False},
{"action": "wait", "notify": False, "finish_goal": False},
{"action": "replan", "notify": False, "finish_goal": False},
{"action": "external_wait", "notify": False, "finish_goal": False},
{"action": "work", "notify": True, "finish_goal": False},
]


Expand All @@ -35,6 +35,11 @@ def next_final_content(self, messages):


def test_probe_uses_current_production_prompts_and_hidden_independent_oracle():
fallback = cases()[-1]
assert fallback["packet"]["selected_todo"]["todo_id"] == "todo_external_wait_fallback"
assert fallback["packet"]["execution_obligation"]["must_attempt_work"] is True
assert fallback["packet"]["interaction_contract"]["user_channel"]["notify"] == "NOTIFY"
assert "external_wait_observation" not in fallback["packet"]
for mode in ("thin", "brief"):
for case in cases():
messages = probe_messages(mode, case["packet"])
Expand Down
Loading