工具说报告写好了，助手可以直接交付吗？

"""Run only this bundled worker; demonstrate output acceptance, NOT a sandbox."""
import csv
import json
from pathlib import Path
import subprocess
import sys
import tempfile
from uuid import uuid4

SCRIPT = Path(__file__).resolve()
MODES = ("nonzero_with_file", "zero_no_file", "bad_json", "wrong_task", "wrong_scope", "wrong_rows", "valid")


def worker(mode, task_id):
    # The fixed worker reads only the synthetic request in its working directory.
    task = json.loads(Path("request.json").read_text(encoding="utf-8"))
    assert task["task_id"] == task_id
    artifact = {"task_id": task_id, "scope": task["scope"], "records": task["records"]}
    if mode == "wrong_task":
        artifact["task_id"] = "a-prior-task"
    if mode == "wrong_scope":
        artifact["scope"] = "another-project"
    if mode == "wrong_rows":
        artifact["records"] = artifact["records"][:1]
    if mode == "bad_json":
        Path("result.json").write_text('{"records":', encoding="utf-8")
    elif mode != "zero_no_file":
        Path("result.json").write_text(json.dumps(artifact), encoding="utf-8")
    print("Report ready.")  # Every mode says the same thing.
    return 7 if mode == "nonzero_with_file" else 0


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate_json_member")
        result[key] = value
    return result


def validate(artifact, task):
    if not isinstance(artifact, dict) or set(artifact) != {"task_id", "scope", "records"}:
        return "wrong_shape"
    if artifact["task_id"] != task["task_id"]:
        return "task_mismatch"
    if artifact["scope"] != task["scope"]:
        return "scope_mismatch"
    rows = artifact["records"]
    if not isinstance(rows, list) or any(not isinstance(r, dict) or
            set(r) != {"id", "text"} or not all(isinstance(v, str) for v in r.values()) for r in rows):
        return "invalid_records"
    ids = [r["id"] for r in rows]
    if len(ids) != len(set(ids)):
        return "duplicate_record_id"
    actual = {r["id"]: r["text"] for r in rows}
    expected = {r["id"]: r["text"] for r in task["records"]}
    if actual != expected:
        return "records_mismatch"
    return "accepted"


def inspect_artifact(path, task):
    if not path.is_file():
        return "missing_file", None
    try:
        artifact = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=unique_object)
    except (ValueError, UnicodeError):
        return "invalid_json", None
    return validate(artifact, task), artifact


def run_case(mode):
    task = {"task_id": uuid4().hex, "scope": "project-alpha-selected-notes",
            "records": [{"id": "N1", "text": "Read the input."}, {"id": "N2", "text": "Check the output."}]}
    # Fresh directory and task ID for EACH invocation prevent reuse of an old result.
    with tempfile.TemporaryDirectory(prefix="artifact-case-", dir=Path.cwd()) as temp:
        folder = Path(temp)
        path = folder / "result.json"
        assert not path.exists()
        (folder / "request.json").write_text(json.dumps(task), encoding="utf-8")
        result = subprocess.run([sys.executable, str(SCRIPT), "--worker", mode, task["task_id"]],
                                cwd=folder, capture_output=True, text=True, encoding="utf-8",
                                timeout=10, check=False)
        artifact_status, artifact = inspect_artifact(path, task)
        decision = "deliverable" if result.returncode == 0 and artifact_status == "accepted" else "rejected"
        return {"mode": mode, "task_id": task["task_id"], "returncode": result.returncode,
                "stdout": result.stdout.strip(), "stderr": result.stderr.strip(),
                "artifact_exists": path.is_file(), "artifact_status": artifact_status,
                "decision": decision, "artifact_snapshot": artifact}


def main():
    cases = [run_case(mode) for mode in MODES]
    assert len({r["task_id"] for r in cases}) == len(cases)
    assert all(r["stdout"] == "Report ready." for r in cases)
    assert [r["decision"] for r in cases] == ["rejected"] * 6 + ["deliverable"]
    assert [r["artifact_status"] for r in cases] == [
        "accepted", "missing_file", "invalid_json", "task_mismatch", "scope_mismatch", "records_mismatch", "accepted"]
    assert cases[0]["returncode"] == 7 and cases[0]["artifact_exists"]
    assert cases[1]["returncode"] == 0 and not cases[1]["artifact_exists"]
    assert all(r["stderr"] == "" for r in cases)
    try:
        json.loads('{"x":1,"x":2}', object_pairs_hook=unique_object)
    except ValueError:
        pass
    else:
        raise AssertionError("duplicate member was accepted")
    probe = {"task_id": "probe", "scope": "s", "records": [{"id": "1", "text": "x"}]}
    assert validate({**probe, "records": probe["records"] * 2}, probe) == "duplicate_record_id"
    assert validate({**probe, "records": [{"id": "1", "text": False}]}, probe) == "invalid_records"
    report = {"scope": "fixed bundled subprocess; exact export of two selected synthetic notes",
              "policy": "zero exit AND accepted artifact; stdout is diagnostic only",
              "cases": cases, "assertions": "passed"}
    Path("artifact-acceptance-report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    columns = ["mode", "task_id", "returncode", "artifact_exists", "artifact_status", "decision"]
    with Path("artifact-acceptance.csv").open("w", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=columns, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(cases)
    for row in cases:
        print(f"{row['mode']}: exit={row['returncode']}, artifact={row['artifact_status']}, {row['decision']}")
    print("PASS: 7 real child runs; one deliverable; independent directories and task IDs")


if __name__ == "__main__":
    if len(sys.argv) == 4 and sys.argv[1] == "--worker" and sys.argv[2] in MODES:
        raise SystemExit(worker(sys.argv[2], sys.argv[3]))
    main()