tests/test_concurrency.py

import threading
import time

from agent import SolveResult
from cache import ResultStore
from evaluation import run_evaluation


class Client:
    def get_questions(self):
        return [
            {"task_id": str(index), "question": "q", "file_name": ""}
            for index in range(6)
        ]

    def download_attachment(self, task):
        return None


class Agent:
    class settings:
        worker_limit = 3

    def __init__(self):
        self.active = 0
        self.maximum = 0
        self.calls = 0
        self.lock = threading.Lock()

    def solve_detailed(self, task, attachment, *, force=False):
        with self.lock:
            self.calls += 1
            self.active += 1
            self.maximum = max(self.maximum, self.active)
        time.sleep(0.03)
        with self.lock:
            self.active -= 1
        return SolveResult(task["task_id"], "general_reasoning")


def test_concurrent_checkpointing_resume_and_force(tmp_path):
    agent = Agent()
    store = ResultStore(tmp_path / "results.json")
    first = run_evaluation(Client(), agent, store)
    assert len(first) == 6 and agent.maximum > 1 and agent.calls == 6
    run_evaluation(Client(), agent, store)
    assert agent.calls == 6
    run_evaluation(Client(), agent, store, force=True, task_ids={"2"})
    assert agent.calls == 7