chore(benchmark): remove longmemeval final answer review file (#366)
Some checks are pending
Pre-commit / run (ubuntu-latest) (push) Waiting to run
Tests ReMe / Unit Tests - py3.11 (push) Waiting to run
Tests ReMe / Unit Tests - py3.12 (push) Waiting to run
Tests ReMe / Unit Tests - py3.13 (push) Waiting to run
Windows Smoke / CLI smoke - py3.11 (push) Waiting to run

* feat(benchmark): add final answer review step for evaluation

- Introduce FinalAnswerReviewStep to handle answer validation
- Add final_answer_review.jsonl dataset with 24 evaluation cases
- Include detailed reasoning and golden check results for each case
- Support various question types including temporal reasoning and preferences
- Implement time consistency checks for session references
- Add comprehensive test coverage for different evaluation scenarios

* chore(benchmark): remove longmemeval final answer review file

- Removed final_answer_review.jsonl containing 23 evaluation records
- Deleted question_id mappings with detailed reasoning for golden answers
- Removed answer correctness assessments and session time validation checks
- Cleaned up benchmark dataset used for memory evaluation testing
- Eliminated JSONL format evaluation results for temporal reasoning tasks
- Removed references to various session IDs and time-based validations

* config(default): disable shell step configuration by commenting out

- Commented out the shell step configuration in default.yaml
- Disabled asynchronous shell command execution capability
- Removed shell step from available backend operations
- Preserved traverse backend configuration unchanged

* refactor(tests): remove unused shell job test from config parser tests

- Removed test_default_config_registers_shell_job function that was no longer needed
- Kept existing test for frontmatter chunk metadata configuration
- Cleaned up test suite by removing obsolete test case
This commit is contained in:
jinliyl 2026-07-16 20:32:28 +08:00 • committed by GitHub
parent c3b1e93918
commit 2eb05392c6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1285 additions and 62 deletions

View file

@ -0,0 +1,469 @@
#!/usr/bin/env python3
"""Review every LongMemEval golden answer with the configured Claude Code job.
Every numeric ``datasets/longmemeval/<idx>`` workspace is processed sequentially.
The reference JSONL files are merged by ``question_id`` and supplied only when
they contain an alternative answer for that sample:
reme start config=jinli_lme job=final_answer_review
The job returns a plain four-field JSON object with ``reason``,
``golden_answer_correct``, ``answer``, and ``is_session_time_wrong``. After
every new success, this driver atomically rewrites the complete accumulated
output JSONL so an interrupted run can safely resume.
Examples:
python benchmark/longmemeval/run_final_answer_review.py
python benchmark/longmemeval/run_final_answer_review.py --exclude-reference-question-ids
python benchmark/longmemeval/run_final_answer_review.py --only-reference-question-ids --rerun-selected
python benchmark/longmemeval/run_final_answer_review.py --concurrency 2 --submit-interval-seconds 6
python benchmark/longmemeval/run_final_answer_review.py --question-id e47becba
python benchmark/longmemeval/run_final_answer_review.py --reference path/to/results.jsonl
python benchmark/longmemeval/run_final_answer_review.py --limit 3
python benchmark/longmemeval/run_final_answer_review.py --no-resume
python benchmark/longmemeval/run_final_answer_review.py --dry-run
"""
import argparse
import concurrent.futures
import json
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from typing import Any
REPO = Path(__file__).resolve().parents[2]
DATA = REPO / "datasets" / "longmemeval"
DEFAULT_REFERENCES = (
REPO / "benchmark" / "longmemeval" / "golden_check_list_false.jsonl",
REPO / "benchmark" / "longmemeval" / "merge_confirm_jinli_false.jsonl",
)
DEFAULT_OUTPUT = REPO / "benchmark" / "longmemeval" / "final_answer_review.jsonl"
DEFAULT_LOG_DIR = REPO / "logs" / "final_answer_review"
REFERENCE_PATHS_ENV = "LME_FINAL_ANSWER_REFERENCE_PATHS"
MAX_CONCURRENCY = 3
MIN_SUBMIT_INTERVAL_SECONDS = 5.0
DEFAULT_SUBMIT_INTERVAL_SECONDS = 5.1
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
parser = argparse.ArgumentParser(
description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"--question-id",
dest="question_ids",
action="append",
help="process only this dataset question ID; repeat for multiple IDs (default: all)",
)
reference_selection = parser.add_mutually_exclusive_group()
reference_selection.add_argument(
"--exclude-reference-question-ids",
action="store_true",
help="skip question IDs found in the selected reference-answer JSONL files",
)
reference_selection.add_argument(
"--only-reference-question-ids",
action="store_true",
help="process only question IDs found in the selected reference-answer JSONL files",
)
parser.add_argument(
"--reference",
dest="references",
action="append",
type=Path,
help="reference-answer JSONL; repeat for multiple files (default: built-in disputed results)",
)
parser.add_argument(
"--output",
type=Path,
default=DEFAULT_OUTPUT,
help=f"output JSONL (default: {DEFAULT_OUTPUT})",
)
parser.add_argument(
"--log-dir",
type=Path,
default=DEFAULT_LOG_DIR,
help="directory for per-question logs",
)
parser.add_argument(
"--concurrency",
type=int,
default=MAX_CONCURRENCY,
help=f"maximum concurrent jobs, from 1 to {MAX_CONCURRENCY} (default: {MAX_CONCURRENCY})",
)
parser.add_argument(
"--submit-interval-seconds",
type=float,
default=DEFAULT_SUBMIT_INTERVAL_SECONDS,
help=f"minimum time between job submissions; must be > {MIN_SUBMIT_INTERVAL_SECONDS:g} "
f"(default: {DEFAULT_SUBMIT_INTERVAL_SECONDS:g})",
)
parser.add_argument(
"--limit",
type=int,
default=0,
help="process only the first N pending questions (0 = all)",
)
resume_mode = parser.add_mutually_exclusive_group()
resume_mode.add_argument(
"--no-resume",
action="store_true",
help="ignore existing output and rerun every selected question",
)
resume_mode.add_argument(
"--rerun-selected",
action="store_true",
help="rerun every selected question while preserving existing results until replacements finish",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="show the selected cases without invoking ReMe",
)
return parser.parse_args()
def _read_jsonl(path: Path) -> list[dict[str, Any]]:
"""Read a JSONL file and reject malformed or non-object rows."""
rows: list[dict[str, Any]] = []
try:
with path.open(encoding="utf-8") as file:
for line_number, line in enumerate(file, start=1):
if not line.strip():
continue
try:
row = json.loads(line)
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSON at {path}:{line_number}") from exc
if not isinstance(row, dict):
raise ValueError(f"Expected a JSON object at {path}:{line_number}")
rows.append(row)
except OSError as exc:
raise FileNotFoundError(f"Cannot read JSONL file: {path}") from exc
return rows
def merge_references(paths: list[Path]) -> dict[str, list[dict[str, Any]]]:
"""Merge reference rows by question ID, preserving file and row order."""
merged: dict[str, list[dict[str, Any]]] = {}
seen_sources: set[tuple[str, str]] = set()
for path in paths:
for row in _read_jsonl(path):
question_id = str(row.get("question_id") or "").strip()
if not question_id:
raise ValueError(f"Reference row in {path} has no question_id")
source_key = (question_id, str(path.resolve()))
if source_key in seen_sources:
raise ValueError(f"Duplicate question_id={question_id!r} within {path}")
seen_sources.add(source_key)
merged.setdefault(question_id, []).append({"source": path.name, **row})
if not merged:
raise ValueError("No reference answers found")
return merged
def workspace_map() -> dict[str, Path]:
"""Map every dataset question ID to its numeric sample workspace."""
mapping: dict[str, Path] = {}
for workspace in sorted(
(path for path in DATA.iterdir() if path.is_dir() and path.name.isdigit()),
key=lambda p: int(p.name),
):
query_path = workspace / "query.json"
if not query_path.is_file():
continue
try:
with query_path.open(encoding="utf-8") as file:
query = json.load(file)
except (OSError, json.JSONDecodeError) as exc:
raise ValueError(f"Cannot parse {query_path}") from exc
if not isinstance(query, dict):
raise ValueError(f"Expected a JSON object in {query_path}")
question_id = str(query.get("question_id") or "").strip()
if not question_id:
raise ValueError(f"Missing question_id in {query_path}")
if question_id in mapping:
raise ValueError(
f"Duplicate dataset question_id={question_id!r}: {mapping[question_id]} and {workspace}",
)
mapping[question_id] = workspace
return mapping
def select_question_ids(
mapping: dict[str, Path],
requested: list[str] | None,
excluded: set[str] | None = None,
) -> list[str]:
"""Return all dataset IDs or validate an explicitly requested subset."""
excluded = excluded or set()
if not requested:
return [question_id for question_id in mapping if question_id not in excluded]
selected: list[str] = []
seen: set[str] = set()
for raw_question_id in requested:
question_id = raw_question_id.strip()
if not question_id:
raise ValueError("--question-id must not be empty")
if question_id in seen:
raise ValueError(f"Duplicate --question-id: {question_id}")
if question_id not in mapping:
raise ValueError(f"No dataset workspace for question ID: {question_id}")
if question_id not in excluded:
selected.append(question_id)
seen.add(question_id)
return selected
def _validate_result(value: Any, *, source: str) -> dict[str, Any]:
"""Validate the final four-field answer contract."""
expected_keys = {"reason", "golden_answer_correct", "answer", "is_session_time_wrong"}
if not isinstance(value, dict) or set(value) != expected_keys:
raise ValueError(
f"{source} must contain exactly 'reason', 'golden_answer_correct', 'answer', "
"and 'is_session_time_wrong'",
)
if not isinstance(value["reason"], str) or not value["reason"].strip():
raise ValueError(f"{source} has an invalid reason")
if not isinstance(value["golden_answer_correct"], bool):
raise ValueError(f"{source} has an invalid golden_answer_correct")
if not isinstance(value["answer"], str):
raise ValueError(f"{source} has an invalid answer")
answer = value["answer"].strip()
if value["golden_answer_correct"] and answer:
raise ValueError(f"{source} answer must be empty when golden_answer_correct is true")
if not value["golden_answer_correct"] and not answer:
raise ValueError(f"{source} answer must be non-empty when golden_answer_correct is false")
if not isinstance(value["is_session_time_wrong"], bool):
raise ValueError(f"{source} has an invalid is_session_time_wrong")
return {
"reason": value["reason"].strip(),
"golden_answer_correct": value["golden_answer_correct"],
"answer": answer,
"is_session_time_wrong": False,
}
def load_existing(path: Path) -> dict[str, dict[str, Any]]:
"""Load resumable output, rejecting duplicate or malformed rows."""
if not path.exists():
return {}
results: dict[str, dict[str, Any]] = {}
for row in _read_jsonl(path):
question_id = str(row.get("question_id") or "").strip()
if not question_id:
raise ValueError(f"Existing output row in {path} has no question_id")
if question_id in results:
raise ValueError(
f"Duplicate question_id={question_id!r} in existing output {path}",
)
results[question_id] = _validate_result(
{key: value for key, value in row.items() if key != "question_id"},
source=f"existing result for {question_id}",
)
return results
def atomic_write_results(
path: Path,
order: list[str],
results: dict[str, dict[str, Any]],
) -> None:
"""Atomically rewrite all accumulated rows in stable merged-input order."""
path.parent.mkdir(parents=True, exist_ok=True)
temp_path: Path | None = None
try:
with tempfile.NamedTemporaryFile(
"w",
encoding="utf-8",
dir=path.parent,
prefix=f".{path.name}.",
delete=False,
) as file:
temp_path = Path(file.name)
for question_id in order:
if question_id not in results:
continue
row = {"question_id": question_id, **results[question_id]}
file.write(
json.dumps(row, ensure_ascii=False, separators=(",", ":")) + "\n",
)
file.flush()
os.fsync(file.fileno())
os.replace(temp_path, path)
finally:
if temp_path is not None and temp_path.exists():
temp_path.unlink()
def run_one(
question_id: str,
workspace: Path,
log_dir: Path,
reference_paths: list[Path],
) -> dict[str, Any]:
"""Run the configured one-shot job and validate its stdout JSON."""
env = dict(os.environ, LME_WORKSPACE_DIR=str(workspace.relative_to(REPO)))
env[REFERENCE_PATHS_ENV] = json.dumps(
[str(path.resolve()) for path in reference_paths],
ensure_ascii=False,
)
completed = subprocess.run(
[
sys.executable,
"-c",
"from reme.reme import main; main()",
"start",
"config=jinli_lme",
"job=final_answer_review",
],
cwd=REPO,
env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
check=False,
)
log_dir.mkdir(parents=True, exist_ok=True)
log_path = log_dir / f"{question_id}.log"
log_text = (
f"workspace={workspace}\nreturncode={completed.returncode}\n\n"
f"[stdout]\n{completed.stdout}\n[stderr]\n{completed.stderr}"
)
log_path.write_text(
log_text,
encoding="utf-8",
)
if completed.returncode != 0:
raise RuntimeError(
f"Job failed for {question_id} with rc={completed.returncode}; see {log_path}",
)
try:
value = json.loads(completed.stdout.strip())
except json.JSONDecodeError as exc:
raise ValueError(
f"Job stdout is not JSON for {question_id}; see {log_path}",
) from exc
return _validate_result(value, source=f"job result for {question_id}")
def main() -> int:
"""Review and checkpoint the selected dataset cases sequentially."""
args = parse_args()
if args.limit < 0:
raise ValueError("--limit must be >= 0")
if not 1 <= args.concurrency <= MAX_CONCURRENCY:
raise ValueError(f"--concurrency must be between 1 and {MAX_CONCURRENCY}")
if args.submit_interval_seconds <= MIN_SUBMIT_INTERVAL_SECONDS:
raise ValueError(
f"--submit-interval-seconds must be > {MIN_SUBMIT_INTERVAL_SECONDS:g}",
)
reference_paths = [path.resolve() for path in (args.references or DEFAULT_REFERENCES)]
mapping = workspace_map()
references = merge_references(reference_paths)
missing = [question_id for question_id in references if question_id not in mapping]
if missing:
raise ValueError(f"No dataset workspace for question IDs: {', '.join(missing)}")
full_order = list(mapping)
excluded = set(references) if args.exclude_reference_question_ids else set()
order = select_question_ids(mapping, args.question_ids, excluded)
if args.only_reference_question_ids:
order = [question_id for question_id in order if question_id in references]
results = {} if args.no_resume else load_existing(args.output.resolve())
pending = (
list(order) if args.rerun_selected else [question_id for question_id in order if question_id not in results]
)
if args.limit:
pending = pending[: args.limit]
no_reference = sum(question_id not in references for question_id in order)
one_reference = sum(len(references.get(question_id, [])) == 1 for question_id in order)
multiple_references = sum(len(references.get(question_id, [])) > 1 for question_id in order)
print(
f"total={len(order)} no_reference={no_reference} one_reference={one_reference} "
f"multiple_references={multiple_references} "
f"excluded={len(excluded)} "
f"only_reference_questions={args.only_reference_question_ids} "
f"concurrency={args.concurrency} submit_interval={args.submit_interval_seconds:g}s "
f"existing={len(results)} pending={len(pending)} output={args.output.resolve()}",
flush=True,
)
if args.dry_run:
for question_id in pending:
print(
f"[would-run] question_id={question_id} workspace={mapping[question_id].name} "
f"references={len(references.get(question_id, []))}",
)
return 0
executor = concurrent.futures.ThreadPoolExecutor(max_workers=args.concurrency)
active: dict[concurrent.futures.Future[dict[str, Any]], tuple[int, str]] = {}
next_position = 0
saved_count = 0
next_submit_at = 0.0
try:
while next_position < len(pending) or active:
can_submit = next_position < len(pending) and len(active) < args.concurrency
if can_submit and time.monotonic() >= next_submit_at:
question_id = pending[next_position]
position = next_position + 1
workspace = mapping[question_id]
print(
f"[submit {position}/{len(pending)}] question_id={question_id} "
f"workspace={workspace.name} references={len(references.get(question_id, []))}",
flush=True,
)
future = executor.submit(
run_one,
question_id,
workspace,
args.log_dir.resolve(),
reference_paths,
)
active[future] = (position, question_id)
next_position += 1
next_submit_at = time.monotonic() + args.submit_interval_seconds
continue
if not active:
time.sleep(max(0.0, next_submit_at - time.monotonic()))
continue
timeout = None
if can_submit:
timeout = max(0.0, next_submit_at - time.monotonic())
done, _ = concurrent.futures.wait(
active,
timeout=timeout,
return_when=concurrent.futures.FIRST_COMPLETED,
)
for future in done:
position, question_id = active.pop(future)
results[question_id] = future.result()
atomic_write_results(args.output.resolve(), full_order, results)
saved_count += 1
print(
f"[saved {saved_count}/{len(pending)}] submitted_position={position} " f"question_id={question_id}",
flush=True,
)
finally:
executor.shutdown(wait=True, cancel_futures=True)
print(
f"ALL FINISHED total_saved={sum(question_id in results for question_id in order)}",
flush=True,
)
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -237,23 +237,23 @@ jobs:
steps:
- backend: help_step
shell:
backend: base
description: "execute a shell command asynchronously in the workspace"
parameters:
type: object
properties:
cmd:
type: string
description: "shell command to execute"
shell_timeout:
type: number
description: "maximum execution time in seconds"
default: 86400
required:
- cmd
steps:
- backend: shell_step
# shell:
# backend: base
# description: "execute a shell command asynchronously in the workspace"
# parameters:
# type: object
# properties:
# cmd:
# type: string
# description: "shell command to execute"
# shell_timeout:
# type: number
# description: "maximum execution time in seconds"
# default: 86400
# required:
# - cmd
# steps:
# - backend: shell_step
traverse:
backend: base

View file

@ -199,6 +199,21 @@ jobs:
- backend: lme_golden_check_step
agent_wrapper: lme_judge
final_answer_review:
backend: base
description: "Review one LongMemEval golden answer from all sessions available by question_date."
parameters:
type: object
properties: { }
steps:
- backend: lme_final_answer_review_step
agent_wrapper: lme_final_answer_review
reference_paths:
- benchmark/longmemeval/golden_check_list_false.jsonl
- benchmark/longmemeval/merge_confirm_jinli_false.jsonl
retry_initial_seconds: 5
retry_max_seconds: 300
components:
tokenizer:
default:
@ -335,6 +350,14 @@ components:
base_url: ${CLAUDE_CODE_BASE_URL:-https://dashscope.aliyuncs.com/apps/anthropic}
permission_mode: bypassPermissions
lme_final_answer_review:
backend: claude_code
model: ${CLAUDE_CODE_MODEL_NAME:-claude-opus-4-8}
api_key: ${CLAUDE_CODE_API_KEY:-}
base_url: ${CLAUDE_CODE_BASE_URL:-}
cwd: session
permission_mode: bypassPermissions
lme_review:
backend: agentscope
as_llm: plus

View file

@ -4,12 +4,14 @@ from .agentic_answer import LmeAgenticAnswerStep
from .auto_memory import LmeAutoMemoryStep
from .context_answer import ContextAnswerStep
from .extract_session import LmeExtractSessionStep
from .final_answer_review import FinalAnswerReviewStep
from .golden_check import GoldenCheckStep
from .lme_llm_judge import LmeLlmJudgeStep
from .session_review import SessionReviewStep
__all__ = [
"ContextAnswerStep",
"FinalAnswerReviewStep",
"GoldenCheckStep",
"LmeAgenticAnswerStep",
"LmeAutoMemoryStep",

View file

@ -0,0 +1,280 @@
"""Produce a final, evidence-backed answer for a LongMemEval case.
The step puts the complete query, golden-answer object, and any available
disputed reference answers directly into the prompt. Raw session content stays out of the model
context: Claude Code starts in the sample's ``session`` directory and uses its
normal file tools to inspect whichever sessions it needs. Session timestamps
are scanned only to identify evidence that did not exist at question time;
``answer_session_ids`` are not evaluated.
Claude Code is intentionally used without an output schema. Its ordinary text
reply may contain narration but must include exactly one fenced ``json`` block
whose object contains ``reason``, ``golden_answer_correct``, ``answer``, and
``is_session_time_wrong``. API errors and invalid replies are retried with
capped exponential backoff.
"""
import asyncio
import json
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any
from ....components import R
from ...base_step import BaseStep
DEFAULT_REFERENCE_PATHS = (
"benchmark/longmemeval/golden_check_list_false.jsonl",
"benchmark/longmemeval/merge_confirm_jinli_false.jsonl",
)
REFERENCE_PATHS_ENV = "LME_FINAL_ANSWER_REFERENCE_PATHS"
RETRY_INITIAL_SECONDS = 5.0
RETRY_MAX_SECONDS = 300.0
_LME_DATETIME_RE = re.compile(r"(\d{4})/(\d{2})/(\d{2}).*?(\d{2}):(\d{2})")
_FENCED_JSON_RE = re.compile(r"```json\s*(.*?)\s*```", re.IGNORECASE | re.DOTALL)
@R.register("lme_final_answer_review_step")
class FinalAnswerReviewStep(BaseStep):
"""Ask a Claude Code agent to review one golden answer."""
@staticmethod
def _load_json(path: Path) -> dict[str, Any]:
try:
with path.open(encoding="utf-8") as file:
value = json.load(file)
except OSError as exc:
raise FileNotFoundError(f"Cannot read LongMemEval file: {path}") from exc
except json.JSONDecodeError as exc:
raise ValueError(f"Invalid JSON in LongMemEval file: {path}") from exc
if not isinstance(value, dict):
raise ValueError(f"Expected a JSON object in {path}")
return value
@staticmethod
def _parse_datetime(raw_date: Any, *, source: str) -> datetime:
text = str(raw_date or "").strip()
match = _LME_DATETIME_RE.search(text)
if match is None:
raise ValueError(f"Invalid LongMemEval datetime in {source}: {text!r}")
try:
return datetime(*(int(part) for part in match.groups()))
except ValueError as exc:
raise ValueError(
f"Invalid LongMemEval datetime in {source}: {text!r}",
) from exc
def _resolve_reference_path(self, raw_path: str) -> Path:
path = Path(raw_path).expanduser()
if path.is_absolute():
return path
# The configured defaults are repository-relative. Tests and custom
# jobs may instead provide workspace-relative fixture paths.
repository_path = Path.cwd() / path
if repository_path.is_file():
return repository_path
return self.workspace_path / path
def _load_references(self, question_id: str) -> list[dict[str, Any]]:
raw_paths: Any
serialized_paths = os.environ.get(REFERENCE_PATHS_ENV)
if serialized_paths:
try:
raw_paths = json.loads(serialized_paths)
except json.JSONDecodeError as exc:
raise ValueError(f"{REFERENCE_PATHS_ENV} must be a JSON array of paths") from exc
else:
raw_paths = self.kwargs.get("reference_paths") or DEFAULT_REFERENCE_PATHS
if isinstance(raw_paths, str):
raw_paths = [raw_paths]
if not isinstance(raw_paths, (list, tuple)) or not raw_paths:
raise ValueError("reference_paths must contain at least one JSONL path")
references: list[dict[str, Any]] = []
for raw_path in raw_paths:
path = self._resolve_reference_path(str(raw_path))
try:
with path.open(encoding="utf-8") as file:
for line_number, line in enumerate(file, start=1):
if not line.strip():
continue
try:
item = json.loads(line)
except json.JSONDecodeError as exc:
raise ValueError(
f"Invalid JSONL at {path}:{line_number}",
) from exc
if not isinstance(item, dict):
raise ValueError(
f"Expected a JSON object at {path}:{line_number}",
)
if str(item.get("question_id") or "") == question_id:
references.append({"source": path.name, **item})
except OSError as exc:
raise FileNotFoundError(
f"Cannot read reference-answer file: {path}",
) from exc
return references
def _inspect_session_times(self, question_dt: datetime) -> tuple[int, list[dict[str, str]]]:
"""Return the session count and timestamp-only metadata for future sessions."""
resource_dir = self.app_context.app_config.resource_dir if self.app_context is not None else "session"
session_dir = self.workspace_path / resource_dir
if not session_dir.is_dir():
raise FileNotFoundError(f"Session directory not found: {session_dir}")
session_paths = sorted(session_dir.glob("*.json"))
future_sessions: list[dict[str, str]] = []
for path in session_paths:
session = self._load_json(path)
session_id = str(session.get("haystack_session_id") or path.stem)
session_date = str(session.get("haystack_date") or "").strip()
session_dt = self._parse_datetime(
session_date,
source=f"{path}:haystack_date",
)
if session_dt > question_dt:
future_sessions.append(
{
"session_id": session_id,
"session_date": session_date,
"session_file": path.name,
},
)
return len(session_paths), future_sessions
@staticmethod
def _parse_reply(raw_reply: Any) -> dict[str, Any]:
if not isinstance(raw_reply, str) or not raw_reply.strip():
raise ValueError("Agent returned an empty reply")
json_blocks = _FENCED_JSON_RE.findall(raw_reply)
if len(json_blocks) != 1:
raise ValueError("Agent reply must contain exactly one fenced ```json``` block")
try:
value = json.loads(json_blocks[0].strip())
except json.JSONDecodeError as exc:
raise ValueError("Agent's fenced json block is not valid JSON") from exc
if not isinstance(value, dict):
raise ValueError("Agent reply must be a JSON object")
if set(value) != {"reason", "golden_answer_correct", "answer", "is_session_time_wrong"}:
raise ValueError(
"Agent reply must contain exactly 'reason', 'golden_answer_correct', 'answer', "
"and 'is_session_time_wrong'",
)
answer = value["answer"]
reason = value["reason"]
golden_answer_correct = value["golden_answer_correct"]
is_session_time_wrong = value["is_session_time_wrong"]
if not isinstance(reason, str) or not reason.strip():
raise ValueError("Agent reply 'reason' must be a non-empty string")
if "answer_session_ids" in reason.casefold():
raise ValueError("Agent reply 'reason' must not evaluate answer_session_ids")
if not isinstance(golden_answer_correct, bool):
raise ValueError("Agent reply 'golden_answer_correct' must be a boolean")
if not isinstance(answer, str):
raise ValueError("Agent reply 'answer' must be a string")
answer = answer.strip()
if golden_answer_correct and answer:
raise ValueError("Agent reply 'answer' must be empty when golden_answer_correct is true")
if not golden_answer_correct and not answer:
raise ValueError("Agent reply 'answer' must be non-empty when golden_answer_correct is false")
if not isinstance(is_session_time_wrong, bool):
raise ValueError("Agent reply 'is_session_time_wrong' must be a boolean")
if is_session_time_wrong:
raise ValueError("Agent reply 'is_session_time_wrong' is deprecated and must be false")
return {
"reason": reason.strip(),
"golden_answer_correct": golden_answer_correct,
"answer": answer,
"is_session_time_wrong": is_session_time_wrong,
}
async def execute(self):
assert self.context is not None
if self.agent_wrapper is None:
raise ValueError("lme_final_answer_review_step requires agent_wrapper")
query = self._load_json(self.workspace_path / "query.json")
golden = self._load_json(self.workspace_path / "answer.json")
question_id = str(query.get("question_id") or "").strip()
if not question_id:
raise ValueError("query.json requires a non-empty 'question_id'")
question_dt = self._parse_datetime(
query.get("question_date"),
source="query.json:question_date",
)
references = self._load_references(question_id)
num_sessions, future_sessions = self._inspect_session_times(question_dt)
payload = {
"query": query,
"answer_json": golden,
"reference_answers": references,
"session_time_check": {
"sessions_after_question_date": future_sessions,
},
}
user_prompt = self.prompt_format(
"user_message",
question_id=question_id,
question_date=str(query.get("question_date") or ""),
num_sessions=num_sessions,
num_future_sessions=len(future_sessions),
num_references=len(references),
payload_json=json.dumps(payload, ensure_ascii=False, indent=2),
)
retry_initial_seconds = float(
self.kwargs.get("retry_initial_seconds", RETRY_INITIAL_SECONDS),
)
retry_max_seconds = float(
self.kwargs.get("retry_max_seconds", RETRY_MAX_SECONDS),
)
if retry_initial_seconds <= 0:
retry_initial_seconds = RETRY_INITIAL_SECONDS
retry_max_seconds = max(retry_max_seconds, retry_initial_seconds)
attempt = 1
sleep_seconds = retry_initial_seconds
while True:
try:
# Deliberately do not pass output_schema: this case evaluates an
# ordinary Claude Code response and validates it afterward.
result = await self.agent_wrapper.reply(
user_prompt,
system_prompt=self.get_prompt("system_prompt"),
)
final_answer = self._parse_reply(result.get("result"))
if attempt > 1:
self.logger.info(
f"[{self.name}] recovered after {attempt} attempts",
)
break
except Exception as exc: # noqa: BLE001 - agent/API/format failures share the retry contract
delay = min(sleep_seconds, retry_max_seconds)
self.logger.warning(
f"[{self.name}] attempt {attempt} failed for {question_id}: {exc}; retrying in {delay:.1f}s",
)
await asyncio.sleep(delay)
sleep_seconds = min(sleep_seconds * 2, retry_max_seconds)
attempt += 1
self.context.response.success = True
self.context.response.answer = json.dumps(final_answer, ensure_ascii=False)
self.context.response.metadata.update(
{
"question_id": question_id,
"num_sessions": num_sessions,
"num_future_sessions": len(future_sessions),
"future_sessions": future_sessions,
"num_reference_answers": len(references),
"is_session_time_wrong": False,
"attempts": attempt,
"agent_session_id": result.get("session_id"),
},
)
return self.context.response

View file

@ -0,0 +1,53 @@
system_prompt: |
你是 LongMemEval 答案的最终审核员。完整的 query.json、answer.json,以及零个或多个可能正确、
也可能错误的参考答案已经放在用户消息的 input JSON 中,不需要去其他目录寻找这些输入。没有参考
答案时,应直接根据原始 session 独立审核 answer.json,不能因为缺少争议记录就假定 golden 答案正确。
你的当前工作目录就是该问题的 session 目录。目录中的每个 JSON 文件都是一个完整原始聊天 session。
原始 session 内容没有预先放进上下文;请主动使用 Read、Glob、Grep、Bash 等工具在当前目录自由检索,
并阅读所有与问题可能相关的 session。不要修改或删除这些文件。
你的任务是独立判断最合理的答案。answer.json 和 reference_answers 都只是待核对的线索,不是事实,
不能因为多个参考答案一致就直接采纳。必须综合全部聊天记录,仔细区分用户与 assistant 的陈述,处理
时间、更新、冲突、计数、偏好和指代关系。
检索时必须始终检查每个文件中的 haystack_date:发生在 question_date 之后的 session 属于未来
信息,绝对不能用其聊天内容推导正确答案或判断 answer.json 正确。即使未来 session 给出了非常直接、
看似正确或与参考答案一致的信息,也必须忽略其内容,避免时间穿越。必须先仅根据 question_date 当时
已经存在的 session 独立得出正确答案,再与 answer.json 比较;合法证据不足时,正确答案为 unknown。
`answer_session_ids` 不属于本次审核对象。不要检查其是否完整、相关、存在或晚于 question_date,也
不得因其包含未来、无关或错误的 session ID 而把 golden answer 判错。`golden_answer_correct` 只由
`answer.json` 中 `answer` 的内容是否完整、正确决定。
input JSON 中的 session_time_check 只用于指出哪些 session 内容晚于 question_date、不能作为答题
证据;它不用于检查 `answer_session_ids`。reason 中不需要评价 `answer_session_ids`。
你可以在最终回复中补充必要的分析文字,但必须包含且只能包含一个 ```json 代码块。程序只解析这个
代码块;没有代码块、存在多个 json 代码块或块内 JSON 无效都会触发重试。代码块内必须是一个对象,
且只能包含四个字段:
- reason:中文详细推理。说明如何处理不同线索和参考答案,尽量逐条引用有证据作用的 session id、
session 时间与具体事实,使后续人工 reviewer 可以复核。
- golden_answer_correct:JSON boolean。仅根据 question_date 之前(含同一时刻)的 session 判断
answer.json 中的 answer 是否完整且正确;不要考虑 answer_session_ids。
- answer:仅当 golden_answer_correct 为 false 时,填写合法证据支持的正确答案(证据不足填
unknown);为 true 时必须填空字符串。
- is_session_time_wrong:为兼容现有输出结构保留的弃用字段,始终填 false。
不要在 reason 或任何字段中评价 answer_session_ids。
输出格式示例仅用于说明 JSON 外形,不是内容 few-shot:
```json
{"reason":"详细推理与 session 证据","golden_answer_correct":false,"answer":"修正答案","is_session_time_wrong":false}
```
user_message: |
请审核 question_id={question_id}。
Question date: {question_date}
Session files in current working directory: {num_sessions}
Sessions after question_date: {num_future_sessions}
Reference answer count: {num_references}
以下 input JSON 包含完整 query.json、answer.json、参考答案和 session 时间检查结果。请先读完,再使用当前
session 目录中的原始文件查找证据,独立推理后严格按 system prompt 要求输出带 ```json 代码块的结果:
{payload_json}

View file

@ -3,8 +3,8 @@
Consumes ``session_review.json`` produced by ``lme_session_review_step`` and
hands its extracted session information to an agent that is equipped with the
``python_execute`` tool. The agent uses ``python_execute`` only as a scratchpad
for the hard reasoning (checking the golden answer and cross-checking the
filtered ``answer_session_ids``); the final verdict is not the
for checking the golden answer; ``answer_session_ids`` are outside the audit
scope. The final verdict is not the
raw Python stdout but a *structured* object extracted from the whole conversation
via ``output_schema``. Sessions dated after ``question_date`` are filtered
upstream by ``lme_session_review_step`` and are not included in this
@ -36,8 +36,7 @@ _VERDICT_SCHEMA = {
"properties": {
"reasoning": {
"type": "string",
"description": "用中文写出详细的推理过程:先说明证据支持的答案,再逐步判断 golden_answer "
"是否正确,以及 answer_session_ids 是否恰好正确。",
"description": "用中文写出详细的推理过程:先说明证据支持的答案,再判断 " "golden_answer 是否正确。",
},
"golden_answer_correct": {
"type": "boolean",
@ -45,26 +44,14 @@ _VERDICT_SCHEMA = {
},
"true_answer": {
"type": "string",
"description": "仅当 golden_answer_correct 为 false 时填写:证据支持的正确答案(证据不足时填 "
"'unknown')。golden_answer_correct 为 true 时填空字符串。",
},
"answer_session_ids_correct": {
"type": "boolean",
"description": "answer_session_ids 是否恰好是支持答案所需的会话(多、少、无关的 id 都算错误)。",
},
"true_answer_session_ids": {
"type": "array",
"items": {"type": "string"},
"description": "仅当 answer_session_ids_correct 为 false 时填写:真正支持答案的 session id 列表。"
"answer_session_ids_correct 为 true 时填空列表。",
"description": "仅当 golden_answer_correct 为 false 时填写:证据支持的正确答案"
"(证据不足时填 'unknown')。golden_answer_correct 为 true 时填空字符串。",
},
},
"required": [
"reasoning",
"golden_answer_correct",
"true_answer",
"answer_session_ids_correct",
"true_answer_session_ids",
],
"additionalProperties": False,
}
@ -110,8 +97,6 @@ class GoldenCheckStep(BaseStep):
question_type = str(query.get("question_type") or "").strip()
question_date = str(query.get("question_date") or "").strip()
golden_answer = str(golden.get("answer") or "").strip()
answer_session_ids = golden.get("answer_session_ids_filter_illegal") or []
if not question:
raise ValueError(f"{review_path} does not contain a question")
@ -120,7 +105,6 @@ class GoldenCheckStep(BaseStep):
"question_type": question_type,
"question_date": question_date,
"golden_answer": golden_answer,
"answer_session_ids": answer_session_ids,
"session_summaries": session_summaries,
}
user_prompt = self.prompt_format(
@ -129,7 +113,6 @@ class GoldenCheckStep(BaseStep):
question_type=question_type,
question_date=question_date,
golden_answer=golden_answer,
answer_session_ids=", ".join(str(s) for s in answer_session_ids) or "(none)",
num_session_summaries=len(session_summaries),
payload_json=json.dumps(prompt_input, ensure_ascii=False, indent=2),
)
@ -173,6 +156,10 @@ class GoldenCheckStep(BaseStep):
if not isinstance(verdict, dict):
self.logger.warning(f"[{self.name}] no structured verdict; falling back to free text")
verdict = {"reasoning": (result.get("result") or "").strip()}
# Retain the legacy fields for readers of existing check_golden.json
# artifacts. They are compatibility placeholders, not audit results.
verdict["answer_session_ids_correct"] = True
verdict["true_answer_session_ids"] = []
# Slim output: do NOT duplicate session_review.json (referenced by path);
# keep only the compact session_summaries and the verdict.

View file

@ -1,10 +1,9 @@
system_prompt: |
你是 LongMemEval 基准测试的审核员。你要根据从用户聊天记录中提取的证据,判断某个问题的
golden_answer 是否正确,以及 answer_session_ids 是否恰好正确。
golden_answer 是否正确。只审核答案内容,不检查或评价 answer_session_ids。
你可以使用 python_execute 工具作为推理草稿本:统计相关会话、抽取答案、交叉核对
answer_session_ids。把给定的数据以字面量形式直接嵌入 Python 代码,使计算可复现;把中间结果
以 JSON 打印出来,便于审计。
你可以使用 python_execute 工具作为推理草稿本:统计相关会话、抽取答案。把给定的数据以字面量
形式直接嵌入 Python 代码,使计算可复现;把中间结果以 JSON 打印出来,便于审计。
python 的 stdout 不是你的最终答案,只是草稿。计算充分、确信之后,停止调用 python,用中文
给出结论。最终的结构化结果会从整段对话中自动抽取,所以务必把推理和结论清楚表达。
@ -12,20 +11,17 @@ system_prompt: |
结构化输出要求:
- reasoning 必须是详细的中文推理过程。
- true_answer 仅在 golden_answer_correct 为 false 时填写,为空字符串否则。
- true_answer_session_ids 仅在 answer_session_ids_correct 为 false 时填写,为空列表否则。
user_message: |
Question: {question}
Question type: {question_type}
Question date: {question_date}
Golden answer: {golden_answer}
Answer session ids: {answer_session_ids}
Number of session extractions included: {num_session_summaries}
证据(JSON)。下方 answer_session_ids 已是 session_review.json 中的
answer_session_ids_filter_illegal;session_summaries 含上游审核过的会话,每条只有 session_id、
session_date、extracted_info:
证据(JSON)。session_summaries 含上游审核过的会话,每条只有 session_id、session_date、
extracted_info:
{payload_json}
用 python_execute 统计和推理,过程中打印中间 JSON。然后用中文给出最终结论:golden_answer
是否正确(不正确时给出 true_answer),answer_session_ids 是否正确(不正确时给出真正的 id)。
是否正确(不正确时给出 true_answer)。不要评价 answer_session_ids。

View file

@ -10,8 +10,7 @@ system_prompt: |
- Keep time expressions inline with the fact they modify, including dates, weekdays, relative
times such as "last week" or "since January 15th", durations, and frequencies.
- Do not invent facts. Only extract what is actually present in the session.
- Do not judge whether answer_session_ids are correct. That is handled by the downstream
golden check.
- Do not judge whether answer_session_ids are correct. They are outside the answer audit scope.
- Output only the extracted information as plain text. Do not output JSON, markdown fences, or
relevance labels.

View file

@ -49,17 +49,6 @@ def test_default_config_registers_daily_write_job():
assert job["parameters"]["required"] == ["name", "description", "session_id", "content"]
def test_default_config_registers_shell_job():
"""``shell`` exposes command execution through ``shell_step``."""
cfg = _load_config("default.yaml")
job = cfg["jobs"]["shell"]
assert job["backend"] == "base"
assert job["steps"] == [{"backend": "shell_step"}]
assert job["parameters"]["required"] == ["cmd"]
assert job["parameters"]["properties"]["shell_timeout"]["default"] == 86400
def test_default_config_keeps_frontmatter_chunk_metadata_opt_in():
"""Markdown frontmatter-to-chunk metadata is disabled by default for compatibility."""
cfg = _load_config("default.yaml")

View file

@ -0,0 +1,425 @@
"""Focused tests for the disputed LongMemEval final-answer workflow."""
import asyncio
import json
import sys
import threading
import time
from pathlib import Path
from unittest.mock import AsyncMock
import pytest
from benchmark.longmemeval import run_final_answer_review as driver_module
from benchmark.longmemeval.run_final_answer_review import (
REFERENCE_PATHS_ENV,
atomic_write_results,
merge_references,
select_question_ids,
)
from reme.components.agent_wrapper.base_agent_wrapper import BaseAgentWrapper
from reme.components.agent_wrapper.cc_agent_wrapper import CcAgentWrapper
from reme.components.application_context import ApplicationContext
from reme.config import resolve_app_config
from reme.steps.benchmark.lme import final_answer_review as review_module
from reme.steps.benchmark.lme.final_answer_review import FinalAnswerReviewStep
class _FakeAgentWrapper(BaseAgentWrapper):
"""Return queued ordinary text replies and retain every prompt call."""
def __init__(self, replies: list[str]):
super().__init__()
self.replies = list(replies)
self.calls: list[tuple[str, dict]] = []
async def reply(self, inputs, **kwargs) -> dict:
"""Return the next queued agent response."""
self.calls.append((inputs, kwargs))
return {
"session_id": f"attempt-{len(self.calls)}",
"result": self.replies.pop(0),
}
def _write_json(path: Path, value: object) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
def _write_jsonl(path: Path, rows: list[dict]) -> None:
path.write_text(
"".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
encoding="utf-8",
)
def _session(session_id: str, date: str, marker: str) -> dict:
return {
"haystack_session_id": session_id,
"haystack_date": date,
"messages": [{"role": "user", "content": marker}],
"other_session_field": f"full-{marker}",
}
def test_final_answer_review_keeps_raw_sessions_out_of_prompt_and_retries_plain_json(
tmp_path,
monkeypatch,
):
"""Raw session messages stay on disk, and invalid ordinary replies are retried."""
query = {
"question_id": "question-1",
"question": "What happened?",
"question_type": "single-session-user",
"question_date": "2024/01/02 (Tue) 10:00",
"extra_query_field": "keep-me",
}
golden = {
"answer": "old answer",
"answer_session_ids": ["past", "future"],
"extra_answer_field": "keep-me-too",
}
_write_json(tmp_path / "query.json", query)
_write_json(tmp_path / "answer.json", golden)
_write_json(
tmp_path / "session" / "past.json",
_session("past", "2024/01/02 (Tue) 09:59", "past-evidence"),
)
_write_json(
tmp_path / "session" / "equal.json",
_session("equal", "2024/01/02 (Tue) 10:00", "equal-evidence"),
)
_write_json(
tmp_path / "session" / "future.json",
_session("future", "2024/01/02 (Tue) 10:01", "future-secret"),
)
_write_jsonl(
tmp_path / "first.jsonl",
[
{
"question_id": "question-1",
"answer": "reference one",
"reason": "first reason",
},
],
)
_write_jsonl(
tmp_path / "second.jsonl",
[
{
"question_id": "question-1",
"answer": "reference two",
"reason": "second reason",
},
],
)
wrapper = _FakeAgentWrapper(
[
'{"reason":"missing fence","golden_answer_correct":false,"answer":"invalid",'
'"is_session_time_wrong":false}',
'```json\n{"reason":"deprecated timestamp verdict","golden_answer_correct":false,'
'"answer":"still invalid","is_session_time_wrong":true}\n```',
"补充分析可以放在代码块外。\n"
'```json\n{"reason":"由 past 和 equal 两个 session 支持 golden answer。",'
'"golden_answer_correct":true,"answer":"","is_session_time_wrong":false}\n```\n'
"审核完成。",
],
)
sleep = AsyncMock()
monkeypatch.setattr(review_module.asyncio, "sleep", sleep)
app_context = ApplicationContext(
workspace_dir=str(tmp_path),
resource_dir="session",
)
step = FinalAnswerReviewStep(
app_context=app_context,
agent_wrapper=wrapper,
reference_paths=["first.jsonl", "second.jsonl"],
retry_initial_seconds=0.01,
retry_max_seconds=0.02,
)
response = asyncio.run(step())
assert response.success is True
assert json.loads(response.answer) == {
"reason": "由 past 和 equal 两个 session 支持 golden answer。",
"golden_answer_correct": True,
"answer": "",
"is_session_time_wrong": False,
}
assert response.metadata["attempts"] == 3
assert response.metadata["num_sessions"] == 3
assert response.metadata["num_future_sessions"] == 1
assert response.metadata["future_sessions"] == [
{
"session_id": "future",
"session_date": "2024/01/02 (Tue) 10:01",
"session_file": "future.json",
},
]
assert len(wrapper.calls) == 3
prompt, reply_kwargs = wrapper.calls[0]
assert "past-evidence" not in prompt
assert "equal-evidence" not in prompt
assert "full-past-evidence" not in prompt
assert "future-secret" not in prompt
assert "extra_query_field" in prompt
assert "extra_answer_field" in prompt
assert "reference one" in prompt and "reference two" in prompt
assert '"session_time_check"' in prompt
assert '"sessions_after_question_date": [' in prompt
assert '"answer_session_ids_after_question_date"' not in prompt
assert '"future"' in prompt
assert "output_schema" not in reply_kwargs
assert [call.args for call in sleep.await_args_list] == [(0.01,), (0.02,)]
# pylint: disable=protected-access
def test_final_answer_review_reference_paths_env_overrides_config(tmp_path, monkeypatch):
"""The batch driver can pass its selected reference files into the job process."""
configured = tmp_path / "configured.jsonl"
selected = tmp_path / "selected.jsonl"
_write_jsonl(
configured,
[{"question_id": "question-1", "answer": "configured", "reason": "configured reason"}],
)
_write_jsonl(
selected,
[{"question_id": "question-1", "answer": "selected", "reason": "selected reason"}],
)
monkeypatch.setenv(REFERENCE_PATHS_ENV, json.dumps([str(selected)]))
step = FinalAnswerReviewStep(
app_context=ApplicationContext(workspace_dir=str(tmp_path)),
reference_paths=[str(configured)],
)
references = step._load_references("question-1")
assert len(references) == 1
assert references[0]["answer"] == "selected"
assert references[0]["source"] == selected.name
def test_final_answer_review_allows_question_without_reference_answer(tmp_path):
"""Samples outside the disputed lists are reviewed from answer.json alone."""
references_path = tmp_path / "references.jsonl"
_write_jsonl(
references_path,
[{"question_id": "another-question", "answer": "other", "reason": "other reason"}],
)
step = FinalAnswerReviewStep(
app_context=ApplicationContext(workspace_dir=str(tmp_path)),
reference_paths=[str(references_path)],
)
assert not step._load_references("question-without-reference")
# pylint: enable=protected-access
def test_final_answer_review_agent_cwd_is_sample_session_directory(tmp_path):
"""The configured relative cwd resolves inside each selected LME workspace."""
config = resolve_app_config(config="jinli_lme", log_config=False)
agent_config = config["components"]["agent_wrapper"]["lme_final_answer_review"]
assert agent_config["cwd"] == "session"
wrapper = CcAgentWrapper(
app_context=ApplicationContext(workspace_dir=str(tmp_path)),
cwd=agent_config["cwd"],
)
assert wrapper.cwd == tmp_path / "session"
# pylint: disable=protected-access
def test_final_answer_review_requires_empty_answer_when_golden_is_correct():
"""Correct golden answers are collected without duplicating their answer text."""
parsed = FinalAnswerReviewStep._parse_reply(
'```json\n{"reason":"golden is supported","golden_answer_correct":true,"answer":"",'
'"is_session_time_wrong":false}\n```',
)
assert parsed == {
"reason": "golden is supported",
"golden_answer_correct": True,
"answer": "",
"is_session_time_wrong": False,
}
with pytest.raises(ValueError, match="answer.*must be empty"):
FinalAnswerReviewStep._parse_reply(
'```json\n{"reason":"bad duplicate","golden_answer_correct":true,"answer":"duplicate",'
'"is_session_time_wrong":false}\n```',
)
with pytest.raises(ValueError, match="deprecated and must be false"):
FinalAnswerReviewStep._parse_reply(
'```json\n{"reason":"legacy session id verdict",'
'"golden_answer_correct":false,"answer":"corrected","is_session_time_wrong":true}\n```',
)
with pytest.raises(ValueError, match="must not evaluate answer_session_ids"):
FinalAnswerReviewStep._parse_reply(
'```json\n{"reason":"answer_session_ids contains a future session",'
'"golden_answer_correct":false,"answer":"corrected","is_session_time_wrong":false}\n```',
)
# pylint: enable=protected-access
def test_final_answer_review_rejects_unparseable_session_time_before_agent(tmp_path):
"""An unknown session time is never silently admitted across the time boundary."""
_write_json(
tmp_path / "query.json",
{
"question_id": "question-1",
"question": "Q",
"question_date": "2024/01/02 (Tue) 10:00",
},
)
_write_json(tmp_path / "answer.json", {"answer": "A"})
_write_json(
tmp_path / "session" / "bad.json",
_session("bad", "unknown", "must-not-reach-agent"),
)
_write_jsonl(
tmp_path / "refs.jsonl",
[{"question_id": "question-1", "answer": "reference", "reason": "reason"}],
)
valid_reply = "".join(
[
'```json\n{"reason":"y","golden_answer_correct":false,',
'"answer":"x","is_session_time_wrong":false}\n```',
],
)
wrapper = _FakeAgentWrapper([valid_reply])
step = FinalAnswerReviewStep(
app_context=ApplicationContext(
workspace_dir=str(tmp_path),
resource_dir="session",
),
agent_wrapper=wrapper,
reference_paths=["refs.jsonl"],
)
with pytest.raises(ValueError, match="Invalid LongMemEval datetime"):
asyncio.run(step())
assert not wrapper.calls
def test_driver_merges_references_and_atomically_rewrites_in_input_order(tmp_path):
"""The batch checkpoint contains one stable row per completed question."""
first = tmp_path / "first.jsonl"
second = tmp_path / "second.jsonl"
_write_jsonl(
first,
[
{"question_id": "q2", "answer": "a2", "reason": "r2"},
{"question_id": "q1", "answer": "a1", "reason": "r1"},
],
)
_write_jsonl(second, [{"question_id": "q1", "answer": "a1b", "reason": "r1b"}])
merged = merge_references([first, second])
assert list(merged) == ["q2", "q1"]
assert len(merged["q2"]) == 1
assert len(merged["q1"]) == 2
output = tmp_path / "result.jsonl"
atomic_write_results(
output,
list(merged),
{
"q1": {
"reason": "reason-1",
"golden_answer_correct": False,
"answer": "final-1",
"is_session_time_wrong": False,
},
"q2": {
"reason": "reason-2",
"golden_answer_correct": False,
"answer": "final-2",
"is_session_time_wrong": True,
},
},
)
rows = _read_output(output)
assert [row["question_id"] for row in rows] == ["q2", "q1"]
assert driver_module.load_existing(output)["q2"]["is_session_time_wrong"] is False
def test_driver_selects_all_or_explicit_question_ids(tmp_path):
"""Explicit IDs may select samples that have no reference-answer row."""
mapping = {
"q1": tmp_path / "0",
"q2": tmp_path / "1",
"q3": tmp_path / "2",
}
assert select_question_ids(mapping, None) == ["q1", "q2", "q3"]
assert select_question_ids(mapping, ["q3", "q1"]) == ["q3", "q1"]
assert select_question_ids(mapping, None, {"q1", "q3"}) == ["q2"]
assert select_question_ids(mapping, ["q3", "q2"], {"q3"}) == ["q2"]
with pytest.raises(ValueError, match="No dataset workspace"):
select_question_ids(mapping, ["unknown"])
with pytest.raises(ValueError, match="Duplicate"):
select_question_ids(mapping, ["q1", "q1"])
def test_driver_limits_concurrency_and_spaces_submissions(tmp_path, monkeypatch):
"""Concurrent jobs never exceed the cap and are not submitted in a burst."""
mapping = {f"q{index}": tmp_path / str(index) for index in range(4)}
starts: list[float] = []
active = 0
max_active = 0
lock = threading.Lock()
def fake_run_one(question_id, workspace, log_dir, reference_paths):
del question_id, workspace, log_dir, reference_paths
nonlocal active, max_active
with lock:
starts.append(time.monotonic())
active += 1
max_active = max(max_active, active)
time.sleep(0.055)
with lock:
active -= 1
return {
"reason": "reviewed",
"golden_answer_correct": True,
"answer": "",
"is_session_time_wrong": False,
}
monkeypatch.setattr(driver_module, "workspace_map", lambda: mapping)
monkeypatch.setattr(driver_module, "merge_references", lambda paths: {})
monkeypatch.setattr(driver_module, "load_existing", lambda path: {})
monkeypatch.setattr(driver_module, "atomic_write_results", lambda *args: None)
monkeypatch.setattr(driver_module, "run_one", fake_run_one)
monkeypatch.setattr(driver_module, "MIN_SUBMIT_INTERVAL_SECONDS", 0.0)
monkeypatch.setattr(
sys,
"argv",
[
"run_final_answer_review.py",
"--concurrency",
"3",
"--submit-interval-seconds",
"0.02",
"--output",
str(tmp_path / "output.jsonl"),
],
)
assert driver_module.main() == 0
assert max_active == 3
assert len(starts) == 4
assert all(later - earlier >= 0.015 for earlier, later in zip(starts, starts[1:]))
def _read_output(path: Path) -> list[dict]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]