fix: improve opencode trial resume controls

This commit is contained in:
lda
2026-06-30 18:30:16 +07:00 Verified
parent 04b1cf30f9
commit 5e9b76eb94
5 changed files with 310 additions and 17 deletions
+3 -2
View File
@@ -12,8 +12,9 @@ Use this command prefix:
{{server_context}} {{server_context}}
Your writable trial workspace is `{{workspace_path}}`. Write attempt files only Your writable trial workspace is `{{workspace_path}}`. Write attempt files only
inside it. End with the challenge's requested YAML self-report. The self-report inside it. End with the challenge's requested YAML self-report inline in your
will be checked against observed tool calls and manually audited. final answer. A run without an inline self-report is invalid. The inline
self-report will be checked against observed tool calls and manually audited.
Do not read files under other `workspaces/*` trial directories. Prior trial Do not read files under other `workspaces/*` trial directories. Prior trial
workspaces may contain complete answers. If you do read another trial workspace, workspaces may contain complete answers. If you do read another trial workspace,
+87 -1
View File
@@ -3,7 +3,9 @@ from __future__ import annotations
import json import json
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any, Literal
PromptMode = Literal["auto", "continue", "final-report"]
FINAL_REPORT_PROMPT = ( FINAL_REPORT_PROMPT = (
"Your workflow attempt is over. Do not continue coding. Return only the " "Your workflow attempt is over. Do not continue coding. Return only the "
@@ -62,6 +64,15 @@ def resume_prompt_for_result(result: dict[str, object]) -> str:
return CONTINUE_PROMPT return CONTINUE_PROMPT
def resume_prompt_for_mode(result: dict[str, object], mode: PromptMode) -> str:
"""Return the operator-selected resume prompt, falling back to auto detection."""
if mode == "continue":
return CONTINUE_PROMPT
if mode == "final-report":
return FINAL_REPORT_PROMPT
return resume_prompt_for_result(result)
def build_resume_command( def build_resume_command(
*, *,
session_id: str, session_id: str,
@@ -77,6 +88,81 @@ def build_resume_command(
return command return command
def resume_command_from_result(
result: dict[str, Any],
*,
session_id: str | None = None,
attach_url: str | None = None,
model: str | None = None,
variant: str | None = None,
prompt_mode: PromptMode = "auto",
) -> list[str]:
"""Build a resume command from new metadata or recover it from old raw results."""
opencode = result.get("opencode")
if isinstance(opencode, dict):
command = opencode.get("resume_command")
has_override = (
any(value is not None for value in (session_id, attach_url, model, variant))
or prompt_mode != "auto"
)
if (
not has_override
and isinstance(command, list)
and all(isinstance(part, str) for part in command)
):
return command
stdout = result.get("stdout")
stdout_text = stdout if isinstance(stdout, str) else ""
if isinstance(opencode, dict):
metadata_session_id = opencode.get("session_id")
metadata_model = opencode.get("model")
metadata_variant = opencode.get("variant")
metadata_attach_url = opencode.get("attach_url")
else:
metadata_session_id = None
metadata_model = None
metadata_variant = None
metadata_attach_url = None
recovered_session_id = (
session_id
or (metadata_session_id if isinstance(metadata_session_id, str) else None)
or extract_session_id(stdout_text)
)
if not recovered_session_id:
raise ValueError("result has no opencode session id; pass --session")
result_model = result.get("model")
result_variant = result.get("variant")
resolved_model = (
model
or (metadata_model if isinstance(metadata_model, str) else None)
or (result_model if isinstance(result_model, str) else None)
)
resolved_variant = (
variant
or (metadata_variant if isinstance(metadata_variant, str) else None)
or (result_variant if isinstance(result_variant, str) else None)
)
resolved_attach_url = attach_url or (
metadata_attach_url if isinstance(metadata_attach_url, str) else None
)
if not resolved_model:
raise ValueError("result has no opencode model; pass --model")
if not resolved_variant:
raise ValueError("result has no opencode variant; pass --variant")
return build_resume_command(
session_id=recovered_session_id,
attach_url=resolved_attach_url,
model=resolved_model,
variant=resolved_variant,
prompt=resume_prompt_for_mode(result, prompt_mode),
)
def display_resume_command(command: list[str]) -> str: def display_resume_command(command: list[str]) -> str:
"""Render an argv list for copy-paste display without changing execution.""" """Render an argv list for copy-paste display without changing execution."""
return subprocess.list2cmdline(command) return subprocess.list2cmdline(command)
+66 -14
View File
@@ -11,11 +11,18 @@ from pathlib import Path
from typing import Any from typing import Any
try: try:
from .opencode_resume import display_resume_command, resume_result_path from .opencode_resume import (
PromptMode,
display_resume_command,
resume_command_from_result,
resume_result_path,
)
except ImportError: except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2])) sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from examples.agent_challenges.opencode_resume import ( from examples.agent_challenges.opencode_resume import (
PromptMode,
display_resume_command, display_resume_command,
resume_command_from_result,
resume_result_path, resume_result_path,
) )
@@ -30,16 +37,23 @@ def _utf8_subprocess_env() -> dict[str, str]:
return env return env
def _load_command(result: dict[str, Any]) -> list[str]: def _load_command(
opencode = result.get("opencode") result: dict[str, Any],
if not isinstance(opencode, dict): *,
raise ValueError("result has no opencode metadata") session_id: str | None = None,
command = opencode.get("resume_command") attach_url: str | None = None,
if not isinstance(command, list) or not all( model: str | None = None,
isinstance(part, str) for part in command variant: str | None = None,
): prompt_mode: PromptMode = "auto",
raise ValueError("result has no resume_command; session id may be missing") ) -> list[str]:
return command return resume_command_from_result(
result,
session_id=session_id,
attach_url=attach_url,
model=model,
variant=variant,
prompt_mode=prompt_mode,
)
def _display_command(command: list[str]) -> str: def _display_command(command: list[str]) -> str:
@@ -49,12 +63,24 @@ def _display_command(command: list[str]) -> str:
def resume_from_result( def resume_from_result(
result_path: Path, result_path: Path,
*, *,
session_id: str | None = None,
attach_url: str | None = None,
model: str | None = None,
variant: str | None = None,
prompt_mode: PromptMode = "auto",
run_fn: RunFn = subprocess.run, run_fn: RunFn = subprocess.run,
) -> Path: ) -> Path:
result = json.loads(result_path.read_text(encoding="utf-8")) result = json.loads(result_path.read_text(encoding="utf-8"))
if not isinstance(result, dict): if not isinstance(result, dict):
raise ValueError("result file must contain a JSON object") raise ValueError("result file must contain a JSON object")
command = _load_command(result) command = _load_command(
result,
session_id=session_id,
attach_url=attach_url,
model=model,
variant=variant,
prompt_mode=prompt_mode,
)
workspace_path = result.get("workspace_path") workspace_path = result.get("workspace_path")
cwd = str(workspace_path) if isinstance(workspace_path, str) else None cwd = str(workspace_path) if isinstance(workspace_path, str) else None
started = time.monotonic() started = time.monotonic()
@@ -89,6 +115,16 @@ def main(argv: list[str] | None = None) -> int:
description="Print or run an OpenCode trial resume command." description="Print or run an OpenCode trial resume command."
) )
parser.add_argument("--from-result", type=Path, required=True) parser.add_argument("--from-result", type=Path, required=True)
parser.add_argument("--session")
parser.add_argument("--attach", dest="attach_url")
parser.add_argument("--model")
parser.add_argument("--variant")
parser.add_argument(
"--prompt-mode",
choices=("auto", "continue", "final-report"),
default="auto",
help="Choose the resume prompt instead of relying on auto detection.",
)
parser.add_argument("--print-command", action="store_true") parser.add_argument("--print-command", action="store_true")
parser.add_argument("--run", action="store_true") parser.add_argument("--run", action="store_true")
args = parser.parse_args(argv) args = parser.parse_args(argv)
@@ -97,9 +133,25 @@ def main(argv: list[str] | None = None) -> int:
result = json.loads(args.from_result.read_text(encoding="utf-8")) result = json.loads(args.from_result.read_text(encoding="utf-8"))
if not isinstance(result, dict): if not isinstance(result, dict):
raise ValueError("result file must contain a JSON object") raise ValueError("result file must contain a JSON object")
command = _load_command(result) command = _load_command(
result,
session_id=args.session,
attach_url=args.attach_url,
model=args.model,
variant=args.variant,
prompt_mode=args.prompt_mode,
)
if args.run: if args.run:
print(resume_from_result(args.from_result).as_posix()) print(
resume_from_result(
args.from_result,
session_id=args.session,
attach_url=args.attach_url,
model=args.model,
variant=args.variant,
prompt_mode=args.prompt_mode,
).as_posix()
)
else: else:
print(_display_command(command)) print(_display_command(command))
except ValueError as exc: except ValueError as exc:
@@ -129,6 +129,9 @@ def test_challenge_prompt_is_identical_across_profiles(tmp_path: Path) -> None:
assert "genuinely blocked" in rendered[InstructionProfile.DEBUG].text assert "genuinely blocked" in rendered[InstructionProfile.DEBUG].text
assert "debug profile only" in rendered[InstructionProfile.DEBUG].text assert "debug profile only" in rendered[InstructionProfile.DEBUG].text
assert "ux_issues_found" in rendered[InstructionProfile.DEBUG].text assert "ux_issues_found" in rendered[InstructionProfile.DEBUG].text
none_prompt = rendered[InstructionProfile.NONE].text.replace("\n", " ")
assert "inline in your" in none_prompt
assert "without an inline self-report is invalid" in none_prompt
def test_skills_profile_copies_bundle_but_none_does_not(tmp_path: Path) -> None: def test_skills_profile_copies_bundle_but_none_does_not(tmp_path: Path) -> None:
@@ -9,6 +9,7 @@ from examples.agent_challenges.opencode_resume import (
build_resume_command, build_resume_command,
display_resume_command, display_resume_command,
extract_session_id, extract_session_id,
resume_command_from_result,
resume_prompt_for_result, resume_prompt_for_result,
resume_result_path, resume_result_path,
) )
@@ -105,6 +106,83 @@ def test_display_resume_command_quotes_prompt_with_spaces() -> None:
assert rendered == 'opencode run --session ses_cli "continue this trial"' assert rendered == 'opencode run --session ses_cli "continue this trial"'
def test_resume_command_recovers_old_raw_result_stdout_session() -> None:
command = resume_command_from_result(
{
"model": "opencode/mimo-v2.5-free",
"variant": "high",
"task_outcome": "failed",
"assertion_failures": [
"could not extract challenge report for required_fields evaluation"
],
"stdout": _event(type="step_start", sessionID="ses_old"),
},
attach_url="http://127.0.0.1:8192/",
)
assert command[0:4] == ["opencode", "run", "--session", "ses_old"]
assert "--attach" in command
assert "opencode/mimo-v2.5-free" in command
assert any("Do not continue coding" in part for part in command)
def test_resume_command_accepts_explicit_session_when_stdout_is_empty() -> None:
command = resume_command_from_result(
{
"model": "opencode/mimo-v2.5-free",
"variant": "high",
"task_outcome": "timeout",
"stdout": "",
},
session_id="ses_manual",
)
assert command[0:4] == ["opencode", "run", "--session", "ses_manual"]
def test_resume_command_prompt_mode_forces_continue_over_auto_final_report() -> None:
command = resume_command_from_result(
{
"model": "opencode/mimo-v2.5-free",
"variant": "high",
"task_outcome": "failed",
"assertion_failures": [
"could not extract challenge report for required_fields evaluation"
],
"stdout": _event(type="step_start", sessionID="ses_old"),
},
prompt_mode="continue",
)
assert any("Continue this same trial" in part for part in command)
assert not any("Do not continue coding" in part for part in command)
def test_resume_command_prompt_mode_overrides_stored_command() -> None:
command = resume_command_from_result(
{
"stdout": "",
"opencode": {
"model": "opencode/mimo-v2.5-free",
"variant": "high",
"session_id": "ses_metadata",
"resume_command": [
"opencode",
"run",
"--session",
"ses_metadata",
"old prompt",
],
},
},
prompt_mode="continue",
)
assert command[0:4] == ["opencode", "run", "--session", "ses_metadata"]
assert "old prompt" not in command
assert any("Continue this same trial" in part for part in command)
def test_resume_result_path_uses_next_resume_index(tmp_path: Path) -> None: def test_resume_result_path_uses_next_resume_index(tmp_path: Path) -> None:
original = tmp_path / "trial.json" original = tmp_path / "trial.json"
original.write_text("{}", encoding="utf-8") original.write_text("{}", encoding="utf-8")
@@ -154,6 +232,79 @@ def test_resume_trial_prints_resume_command(
assert "opencode run --session ses_cli" in output assert "opencode run --session ses_cli" in output
def test_resume_trial_prints_command_for_old_raw_result(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
from examples.agent_challenges.resume_trial import main
result_path = tmp_path / "trial.json"
result_path.write_text(
json.dumps(
{
"model": "opencode/mimo-v2.5-free",
"variant": "high",
"task_outcome": "failed",
"stdout": _event(type="step_start", sessionID="ses_old"),
}
),
encoding="utf-8",
)
assert (
main(
[
"--from-result",
str(result_path),
"--attach",
"http://127.0.0.1:8192/",
"--print-command",
]
)
== 0
)
output = capsys.readouterr().out
assert "opencode run --session ses_old" in output
assert "--attach http://127.0.0.1:8192/" in output
def test_resume_trial_prints_forced_continue_prompt(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
from examples.agent_challenges.resume_trial import main
result_path = tmp_path / "trial.json"
result_path.write_text(
json.dumps(
{
"model": "opencode/mimo-v2.5-free",
"variant": "high",
"task_outcome": "failed",
"assertion_failures": [
"could not extract challenge report for required_fields evaluation"
],
"stdout": _event(type="step_start", sessionID="ses_old"),
}
),
encoding="utf-8",
)
assert (
main(
[
"--from-result",
str(result_path),
"--prompt-mode",
"continue",
"--print-command",
]
)
== 0
)
output = capsys.readouterr().out
assert "Continue this same trial" in output
assert "Do not continue coding" not in output
def test_resume_trial_run_writes_resume_result(tmp_path: Path) -> None: def test_resume_trial_run_writes_resume_result(tmp_path: Path) -> None:
from subprocess import CompletedProcess from subprocess import CompletedProcess