174 lines
9.5 KiB
Python
174 lines
9.5 KiB
Python
"""Native goal rounds and state changes through the Python SDK profile."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
import threading
|
|
|
|
import pytest
|
|
from pathlib import Path
|
|
|
|
from deepseek_harness.models import CommandExecution
|
|
from deepseek_harness import RunResult, Session
|
|
from deepseek_harness.client import NotificationSubscription
|
|
from test_commands_runtime import (
|
|
_Completion, _Runtime, _SESSION_ID, _WAIT_SECONDS, _WaitResult, _assert_request_permission,
|
|
_command_pair, _of_type, _permission_facts, _read_events,
|
|
_request_texts, _runtime, _text_blocks, _wait_activity,
|
|
)
|
|
|
|
|
|
def _command(runtime: _Runtime, session: Session, text: str, *, kind: str = "success") -> CommandExecution:
|
|
result = runtime.call(runtime.harness.client._query, session.id, text)
|
|
assert isinstance(result, CommandExecution)
|
|
assert result.result.kind == kind, result.result.text
|
|
return result
|
|
|
|
|
|
def _until(subscription: NotificationSubscription, event_type: str, operation: str | None = None) -> dict:
|
|
deadline = time.monotonic() + _WAIT_SECONDS
|
|
while True:
|
|
notification = subscription.next(timeout_seconds=max(0, deadline - time.monotonic()))
|
|
if notification.method != "session.event":
|
|
continue
|
|
event = notification.payload["event"]
|
|
if event["type"] == event_type and (operation is None or event["data"].get("operation") == operation):
|
|
return event["data"]
|
|
|
|
|
|
def _complete_goal(request: dict) -> tuple[str, dict]:
|
|
results = [block for message in request["messages"] for block in message["content"]
|
|
if block["type"] == "tool_result"]
|
|
current = json.loads("\n".join(_text_blocks(results[-1]["content"])))
|
|
goal = current["goal"]
|
|
assert goal["phase"] == "active" and current["activation"] == "armed"
|
|
return "update_goal", {"goal_id": goal["id"], "revision": goal["revision"], "action": "complete"}
|
|
|
|
|
|
def _completion_steps(final: _Completion | None = None) -> list[_Completion]:
|
|
return [_Completion("", tool_call=("get_goal", {})), _Completion("", tool_call=_complete_goal),
|
|
final if final is not None else _Completion("GOAL_DONE")]
|
|
|
|
|
|
def test_goal_reads_and_invalid_controls_do_not_start_model_work(tmp_path: Path) -> None:
|
|
with _runtime(tmp_path, []) as runtime:
|
|
session = runtime.harness.start_session(_SESSION_ID)
|
|
executions = [_command(runtime, session, "/目标")]
|
|
for line in ["/goal edit", "/goal edit Missing goal", "/goal pause", "/goal resume"]:
|
|
executions.append(_command(runtime, session, line, kind="error"))
|
|
executions.append(_command(runtime, session, "/goal clear"))
|
|
assert runtime.model.captured() == []
|
|
events = _read_events(runtime.logs)
|
|
assert not _of_type(events, "goal/change") and not _of_type(events, "turn/start")
|
|
for execution in executions:
|
|
_command_pair(events, execution, "goal")
|
|
|
|
|
|
@pytest.mark.parametrize("clear_before_reply", [False, True])
|
|
def test_goal_continues_rounds_and_finishes_wrapup_after_completion_event(
|
|
tmp_path: Path, clear_before_reply: bool,
|
|
) -> None:
|
|
final = _Completion("GOAL_DONE", blocked=True)
|
|
with _runtime(tmp_path, [_Completion("PROGRESS"), *_completion_steps(final)],
|
|
initial_permission="workspace-write", max_goal_rounds=3) as runtime:
|
|
session = runtime.harness.start_session(_SESSION_ID)
|
|
with runtime.harness.client.subscribe_session_notifications(session.id) as events:
|
|
running = runtime.workers.submit(session.run, "/目标 Verify the small change.")
|
|
completed = _until(events, "goal/change", "complete")
|
|
assert completed["goal"]["phase"] == "complete" and completed["roundsStarted"] == 2
|
|
assert final.entered.wait(_WAIT_SECONDS)
|
|
assert not running.done()
|
|
if clear_before_reply:
|
|
cleared = runtime.call(session.run, "/goal clear")
|
|
assert cleared.final_response == "Goal cleared." and cleared.finish_reason == "completed"
|
|
assert not running.done()
|
|
final.release.set()
|
|
result = running.result(timeout=_WAIT_SECONDS)
|
|
assert isinstance(result, RunResult)
|
|
assert result.final_response == "GOAL_DONE" and result.finish_reason == "completed"
|
|
operations = ["create", "complete", *(["clear"] if clear_before_reply else [])]
|
|
assert [event["data"]["operation"] for event in result.events if event["type"] == "goal/change"] == operations
|
|
_command_pair(_read_events(runtime.logs), result, "goal")
|
|
|
|
|
|
def test_pause_aborts_then_edit_and_resume_use_revised_goal_without_changing_other_sessions(tmp_path: Path) -> None:
|
|
first = _Completion("CANCELLED_RESPONSE", blocked=True, allow_disconnect=True)
|
|
with _runtime(tmp_path, [first, *_completion_steps(), _Completion("RAW_GOAL_TEXT")]) as runtime:
|
|
session = runtime.harness.start_session(_SESSION_ID)
|
|
with runtime.harness.client.subscribe_session_notifications(session.id) as events:
|
|
_command(runtime, session, "/goal Original objective")
|
|
assert first.entered.wait(_WAIT_SECONDS)
|
|
_command(runtime, session, "/goal Duplicate objective", kind="error")
|
|
_command(runtime, session, "/goal edit", kind="error")
|
|
_command(runtime, session, "/goal resume", kind="error")
|
|
_command(runtime, session, "/goal PAUSE")
|
|
assert _until(events, "turn/end")["reason"]["kind"] == "aborted"
|
|
assert not first.release.is_set()
|
|
_command(runtime, session, "/goal edit Revised objective")
|
|
assert len(runtime.model.captured()) == 1
|
|
other = runtime.harness.start_session("goal-other")
|
|
assert "No goal is currently set" in _command(runtime, other, "/goal").result.text
|
|
_command(runtime, session, "/goal resume")
|
|
_until(events, "goal/change", "complete")
|
|
_wait_activity(runtime, session)
|
|
assert 'Objective: "Revised objective"' in "\n".join(_request_texts(runtime.model.captured()[1]))
|
|
_command(runtime, session, "/goal clear")
|
|
_command(runtime, session, "/goal clear")
|
|
runtime.call(runtime.harness.client.session_prompt, session.id, [{"type": "text", "text": "/goal Plain text objective"}])
|
|
_wait_activity(runtime, session)
|
|
assert "/goal Plain text objective" in _request_texts(runtime.model.captured()[-1])
|
|
changes = _of_type(_read_events(runtime.logs), "goal/change")
|
|
assert [e["data"]["operation"] for e in changes] == ["create", "pause", "edit", "resume", "complete", "clear"]
|
|
assert not _of_type(_read_events(runtime.logs, session_id="goal-other"), "goal/change")
|
|
|
|
|
|
def test_clear_keeps_current_request_and_stops_further_goal_rounds(tmp_path: Path) -> None:
|
|
first = _Completion("CURRENT_ROUND_FINISHED", blocked=True)
|
|
with _runtime(tmp_path, [first]) as runtime:
|
|
session = runtime.harness.start_session(_SESSION_ID)
|
|
with runtime.harness.client.subscribe_session_notifications(session.id) as events:
|
|
_command(runtime, session, "/goal Work until cleared")
|
|
assert first.entered.wait(_WAIT_SECONDS)
|
|
_command(runtime, session, "/goal clear")
|
|
first.release.set()
|
|
_wait_activity(runtime, session)
|
|
assert _until(events, "turn/end")["reason"]["kind"] == "completed"
|
|
assert len(runtime.model.captured()) == 1
|
|
events = _read_events(runtime.logs)
|
|
assert len(_of_type(events, "assistant/message")) == 1
|
|
assert [e["data"]["operation"] for e in _of_type(events, "goal/change")] == ["create", "clear"]
|
|
|
|
|
|
def test_round_limit_blocks_and_rejects_resume_without_another_request(tmp_path: Path) -> None:
|
|
with _runtime(tmp_path, [_Completion("ONE"), _Completion("TWO")], max_goal_rounds=2) as runtime:
|
|
session = runtime.harness.start_session(_SESSION_ID)
|
|
with runtime.harness.client.subscribe_session_notifications(session.id) as events:
|
|
_command(runtime, session, "/goal An unfinished objective")
|
|
blocked = _until(events, "goal/change", "block")
|
|
assert blocked["goal"]["blockedReason"]["code"] == "round-limit"
|
|
assert blocked["roundsStarted"] == 2
|
|
_command(runtime, session, "/goal resume", kind="error")
|
|
_wait_activity(runtime, session)
|
|
assert len(runtime.model.captured()) == 2
|
|
|
|
|
|
def test_restored_active_goal_waits_for_explicit_resume(tmp_path: Path) -> None:
|
|
first = _Completion("INTERRUPTED", blocked=True, allow_disconnect=True)
|
|
with _runtime(tmp_path, [first], max_goal_rounds=3) as runtime:
|
|
session = runtime.harness.start_session(_SESSION_ID)
|
|
runtime.call(runtime.harness.client.session_prompt, session.id, [{"type": "text", "text": "Start an ordinary task"}])
|
|
assert first.entered.wait(_WAIT_SECONDS)
|
|
_command(runtime, session, "/goal Persisted objective")
|
|
runtime.harness.close()
|
|
with _runtime(tmp_path, _completion_steps()) as runtime:
|
|
session = runtime.call(runtime.harness.resume_session, _SESSION_ID)
|
|
shown = _command(runtime, session, "/goal")
|
|
assert "Status: active" in shown.result.text and "Activation: disarmed" in shown.result.text
|
|
assert runtime.model.captured() == []
|
|
with runtime.harness.client.subscribe_session_notifications(session.id) as events:
|
|
_command(runtime, session, "/goal resume")
|
|
_until(events, "goal/change", "complete")
|
|
_wait_activity(runtime, session)
|
|
assert 'Objective: "Persisted objective"\nRound: 1/3' in "\n".join(_request_texts(runtime.model.captured()[0]))
|