From c8f01d8699136868e765e1d327612697eb4530a6 Mon Sep 17 00:00:00 2001 From: bekkilove <921693422@qq.com> Date: Sun, 11 Oct 2026 17:12:46 +0800 Subject: [PATCH 1/2] Fix release smoke receipt assertions --- .../e2e/test_public_release_empty_states.py | 5 +- .../e2e/test_public_release_worker_soak.py | 240 +++++++++++------- .../tests/test_public_release_worker_soak.py | 90 ++++++- 3 files changed, 243 insertions(+), 92 deletions(-) diff --git a/backend/scripts/e2e/test_public_release_empty_states.py b/backend/scripts/e2e/test_public_release_empty_states.py index ee929f5e..487c6971 100644 --- a/backend/scripts/e2e/test_public_release_empty_states.py +++ b/backend/scripts/e2e/test_public_release_empty_states.py @@ -88,9 +88,10 @@ def main() -> None: new_resume_button = page.get_by_role("button", name="新建简历", exact=True).last expect(new_resume_button).to_be_visible() new_resume_button.click() - create_dialog = page.get_by_role("dialog") + create_dialog = page.get_by_role("dialog", name="新建简历", exact=True) expect(create_dialog).to_be_visible() - expect(create_dialog.get_by_role("heading", name="新建简历", exact=True)).to_be_visible() + expect(create_dialog.get_by_text("新建简历", exact=True)).to_be_visible() + expect(create_dialog.get_by_label("简历命名")).to_be_visible() create_dialog.get_by_role("button", name="创建并编辑", exact=True).click() page.wait_for_function( "() => /^#\\/resume\\/\\d+/.test(window.location.hash)", diff --git a/backend/scripts/e2e/test_public_release_worker_soak.py b/backend/scripts/e2e/test_public_release_worker_soak.py index d2f4abb4..db672d03 100644 --- a/backend/scripts/e2e/test_public_release_worker_soak.py +++ b/backend/scripts/e2e/test_public_release_worker_soak.py @@ -164,6 +164,130 @@ def _wait_for_role_tasks( ) +def _has_director_event_receipt( + event: dict[str, Any], + director_task: dict[str, Any], +) -> bool: + result = event.get("result") if isinstance(event.get("result"), dict) else {} + return ( + bool(event.get("processed_at")) + and event.get("status") == director_task.get("status") + and result.get("task_id") == director_task.get("task_id") + and bool(result.get("inbox_item_id")) + ) + + +def _wait_for_job_saved_events( + client: httpx.Client, + job_target_ids: set[str], + director_tasks_by_job: dict[str, dict[str, Any]], +) -> list[dict[str, Any]]: + deadline = time.monotonic() + TASK_TIMEOUT_SECONDS + while time.monotonic() < deadline: + payload = _request( + client, + "GET", + "/api/agent/runtime/automation/events?event_type=JOB_SAVED&limit=200", + ) + events = payload.get("events", []) + if not isinstance(events, list): + raise AssertionError(f"AutomationEvent list is invalid: {payload}") + matching = [ + item + for item in events + if isinstance(item, dict) and str(item.get("target_id")) in job_target_ids + ] + events_by_job: dict[str, dict[str, Any]] = {} + for event in matching: + target_id = str(event.get("target_id") or "") + if target_id in events_by_job: + raise AssertionError(f"Job {target_id} has duplicate JOB_SAVED events") + events_by_job[target_id] = event + if set(events_by_job) == job_target_ids and all( + _has_director_event_receipt(event, director_tasks_by_job[target_id]) + for target_id, event in events_by_job.items() + ): + return matching + time.sleep(POLL_SECONDS) + + pending = [ + target_id + for target_id in sorted(job_target_ids) + if target_id not in events_by_job + or not _has_director_event_receipt( + events_by_job[target_id], director_tasks_by_job[target_id] + ) + ] + raise AssertionError( + f"JOB_SAVED terminal receipts were not materialized for {len(pending)} Jobs: " + f"{pending[:5]}" + ) + + +def _assert_job_saved_receipt( + event: dict[str, Any], + director_task: dict[str, Any], + role_task: dict[str, Any], +) -> None: + target_id = str(event.get("target_id") or "") + event_id = str(event.get("event_id") or "") + event_payload = event.get("payload") if isinstance(event.get("payload"), dict) else {} + event_result = event.get("result") if isinstance(event.get("result"), dict) else {} + director_input = ( + director_task.get("input") if isinstance(director_task.get("input"), dict) else {} + ) + role_input = role_task.get("input") if isinstance(role_task.get("input"), dict) else {} + if event.get("event_type") != "JOB_SAVED" or not event_id: + raise AssertionError(f"Job {target_id} lacks a persisted JOB_SAVED event ID") + if not _has_director_event_receipt(event, director_task): + raise AssertionError(f"JOB_SAVED event {event_id} lacks its Career Director receipt") + if ( + str(director_task.get("target_id") or "") != target_id + or director_input.get("automation_event_id") != event_id + or str(role_task.get("target_id") or "") != target_id + or str(role_input.get("job_id") or "") != target_id + or role_input.get("automation_event_id") != event_id + ): + raise AssertionError(f"CareerTasks for Job {target_id} are not bound to its event") + if event_result.get("role_intelligence_task_id") not in {None, role_task.get("task_id")}: + raise AssertionError(f"JOB_SAVED event {event_id} references a different Role task") + expected_provider = str(event_payload.get("runtime_provider") or "auto") + if role_task.get("runtime_provider") != expected_provider: + raise AssertionError( + f"Role task for Job {target_id} does not match its event provider" + ) + + +def _assert_role_task_terminal(task: dict[str, Any], job_target_id: str) -> None: + task_id = str(task.get("task_id") or "") + status = str(task.get("status") or "") + attempt_count = int(task.get("attempt_count") or 0) + max_attempts = int(task.get("max_attempts") or 0) + if max_attempts < 1 or attempt_count > max_attempts or not task.get("finished_at"): + raise AssertionError(f"Role task {task_id} lacks bounded terminal metadata") + if status == "completed": + result = task.get("result") if isinstance(task.get("result"), dict) else {} + benchmark = result.get("benchmark") if isinstance(result.get("benchmark"), dict) else {} + if ( + int(result.get("job_id") or 0) != int(job_target_id) + or not result.get("benchmark_run_id") + or benchmark.get("status") != "completed" + or task.get("result_ref") != f"career-task:{task_id}" + or attempt_count < 1 + or task.get("retryable") + ): + raise AssertionError(f"Role task {task_id} lacks a completed benchmark receipt") + elif status in {"blocked", "failed"}: + if ( + not task.get("retryable") + or not str(task.get("error") or "").strip() + and not str(task.get("error_id") or "").strip() + ): + raise AssertionError(f"Role task {task_id} failure is not visible and retryable") + else: + raise AssertionError(f"Role task {task_id} ended in unexpected state {status!r}") + + def main() -> None: run_key = str(int(time.time() * 1000)) started_at = time.perf_counter() @@ -258,112 +382,48 @@ def main() -> None: raise AssertionError( f"expected {CYCLES} unique persisted tasks, found {len(matching_tasks)}" ) - - automation = _request( - client, - "GET", - "/api/agent/runtime/automation/events?event_type=JOB_SAVED&limit=200", - ) - matching_events = [ - item - for item in automation.get("events", []) - if isinstance(item, dict) - and str(item.get("target_id")) in job_target_ids - ] - if len(matching_events) != CYCLES: - raise AssertionError( - f"expected {CYCLES} unique JOB_SAVED events, found {len(matching_events)}" - ) - events_by_job: dict[str, dict[str, Any]] = {} - for event in matching_events: - target_id = str(event.get("target_id") or "") - if target_id in events_by_job: - raise AssertionError(f"Job {target_id} has duplicate JOB_SAVED events") - if event.get("status") != "dispatched": - raise AssertionError( - f"JOB_SAVED event for Job {target_id} was not durably dispatched" - ) - events_by_job[target_id] = event - if set(events_by_job) != job_target_ids: - raise AssertionError("JOB_SAVED events do not cover every worker Job exactly once") + director_tasks_by_job: dict[str, dict[str, Any]] = {} + for task in matching_tasks: + target_id = str(task.get("target_id") or "") + if target_id in director_tasks_by_job: + raise AssertionError(f"Job {target_id} has duplicate Career Director tasks") + director_tasks_by_job[target_id] = task + if set(director_tasks_by_job) != job_target_ids: + raise AssertionError("Career Director tasks do not cover every worker Job exactly once") role_tasks = _wait_for_role_tasks(client, job_target_ids) if len(role_tasks) != CYCLES: raise AssertionError( f"expected one Role Intelligence task per Job, found {len(role_tasks)}" ) + matching_events = _wait_for_job_saved_events( + client, + job_target_ids, + director_tasks_by_job, + ) + events_by_job = {str(event["target_id"]): event for event in matching_events} + role_tasks_by_job = {str(task["target_id"]): task for task in role_tasks} + if set(role_tasks_by_job) != job_target_ids: + raise AssertionError("Role Intelligence tasks do not cover every worker Job exactly once") + role_task_ids: set[str] = set() role_task_statuses: list[str] = [] for task in role_tasks: target_id = str(task.get("target_id") or "") - event = events_by_job[target_id] - task_input = task.get("input") if isinstance(task.get("input"), dict) else {} task_id = str(task.get("task_id") or "") if not task_id or task_id in role_task_ids: raise AssertionError("Role Intelligence tasks did not have unique task IDs") role_task_ids.add(task_id) if task.get("task_type") != "role_intelligence" or task.get("source") != "automation": raise AssertionError(f"Job {target_id} has an unexpected Role Intelligence task: {task}") - if task.get("runtime_provider") != "replay": - raise AssertionError( - f"Role Intelligence task {task_id} used an unexpected provider" - ) - if ( - str(task_input.get("job_id") or "") != target_id - or task_input.get("automation_event_id") != event.get("event_id") - ): - raise AssertionError( - f"Role Intelligence task {task_id} is not bound to its JOB_SAVED event" - ) - event_result = event.get("result") if isinstance(event.get("result"), dict) else {} - event_role_task = event_result.get("role_intelligence_task") - if not isinstance(event_role_task, dict) or event_role_task.get("task_id") != task_id: - raise AssertionError( - f"JOB_SAVED event {event.get('event_id')} lacks the Role Intelligence task receipt" - ) - status = str(task.get("status") or "") role_task_statuses.append(status) - attempt_count = int(task.get("attempt_count") or 0) - max_attempts = int(task.get("max_attempts") or 0) - if max_attempts < 1 or attempt_count > max_attempts: - raise AssertionError(f"Role Intelligence task {task_id} exceeded its retry bound") - if status == "completed": - result = task.get("result") if isinstance(task.get("result"), dict) else {} - benchmark = result.get("benchmark") if isinstance(result.get("benchmark"), dict) else {} - if ( - int(result.get("job_id") or 0) != int(target_id) - or not result.get("benchmark_run_id") - or benchmark.get("status") != "completed" - or task.get("result_ref") != f"career-task:{task_id}" - or attempt_count < 1 - or task.get("retryable") - ): - raise AssertionError( - f"Role Intelligence task {task_id} lacks a completed benchmark receipt" - ) - elif status == "blocked": - error = str(task.get("error") or "").casefold() - unavailable_markers = ( - "provider", - "runtime", - "credential", - "unavailable", - "not configured", - "authentication", - ) - if ( - not task.get("retryable") - or not error - or not any(marker in error for marker in unavailable_markers) - ): - raise AssertionError( - f"Role Intelligence task {task_id} is blocked without a retryable provider reason" - ) - else: - raise AssertionError( - f"Role Intelligence task {task_id} ended in unexpected state {status!r}" - ) + _assert_job_saved_receipt( + events_by_job[target_id], + director_tasks_by_job[target_id], + role_tasks_by_job[target_id], + ) + _assert_role_task_terminal(task, target_id) integrity = _request(client, "GET", "/api/agent/data/safety/integrity") integrity_check = integrity.get("integrity_check") @@ -386,7 +446,9 @@ def main() -> None: "database_integrity": integrity_check, "foreign_key_violations": foreign_key_violations, "elapsed_seconds": round(time.perf_counter() - started_at, 3), - "role_intelligence_runtime_provider": "replay", + "role_intelligence_runtime_providers": sorted( + {str(task.get("runtime_provider") or "") for task in role_tasks} + ), "career_director_runtime_provider": "embedded", "role_intelligence_tasks": len(role_tasks), } diff --git a/backend/tests/test_public_release_worker_soak.py b/backend/tests/test_public_release_worker_soak.py index b6825b28..c9fe271d 100644 --- a/backend/tests/test_public_release_worker_soak.py +++ b/backend/tests/test_public_release_worker_soak.py @@ -1,12 +1,15 @@ from __future__ import annotations import importlib.util +from datetime import datetime, timezone from pathlib import Path from typing import Any def _load_worker_soak(monkeypatch: Any): - e2e_dir = Path(__file__).resolve().parents[1] / "scripts" / "e2e" + backend_dir = Path(__file__).resolve().parents[1] + e2e_dir = backend_dir / "scripts" / "e2e" + monkeypatch.syspath_prepend(str(backend_dir)) monkeypatch.syspath_prepend(str(e2e_dir)) script_path = e2e_dir / "test_public_release_worker_soak.py" spec = importlib.util.spec_from_file_location("offeru_worker_soak_script", script_path) @@ -37,3 +40,88 @@ def fake_request(_client: Any, method: str, path: str, **kwargs: Any) -> dict[st assert [request["params"]["page"] for request in requests] == [1, 2] assert all(request["method"] == "GET" and request["path"] == "/api/jobs/" for request in requests) assert all(request["params"]["keyword"] == f"Worker Orbit {run_key}-" for request in requests) + + +def test_job_saved_event_serializer_exposes_terminal_task_receipt( + monkeypatch: Any, +) -> None: + worker_soak = _load_worker_soak(monkeypatch) + from app.models.models import AutomationEvent, CareerTask + from app.services.automation import _event_view + from app.services.career_tasks import _task_view + + event_id = "automation_evt_release_test" + director_task_id = "career_task_release_director" + role_task_id = "career_task_release_role" + finished_at = datetime.now(timezone.utc).replace(tzinfo=None) + event = _event_view( + AutomationEvent( + event_id=event_id, + event_type="JOB_SAVED", + source="manual", + target_type="job", + target_id="101", + payload_json={"job_id": 101, "runtime_provider": "auto"}, + dedupe_key="job-saved:job:101", + status="blocked", + result_json={ + "task_id": director_task_id, + "inbox_item_id": f"automation_task_{director_task_id}", + "role_intelligence_task_id": None, + }, + error="", + created_at=finished_at, + processed_at=finished_at, + ) + ) + director_task = _task_view( + CareerTask( + task_id=director_task_id, + task_type="career_director", + source="automation", + target_type="job", + target_id="101", + runtime_provider="embedded", + input_json={ + "automation_event_id": event_id, + "event_type": "JOB_SAVED", + "job_id": 101, + }, + output_contract_json={}, + status="blocked", + progress_json={"stage": "blocked", "error_id": "err_director"}, + result_json={}, + checkpoint_json={}, + error="没有可用 Career Director Provider。", + retryable=True, + attempt_count=1, + max_attempts=3, + created_at=finished_at, + finished_at=finished_at, + ) + ) + role_task = _task_view( + CareerTask( + task_id=role_task_id, + task_type="role_intelligence", + source="automation", + target_type="job", + target_id="101", + runtime_provider="auto", + input_json={"automation_event_id": event_id, "job_id": 101}, + output_contract_json={}, + status="blocked", + progress_json={"stage": "blocked", "error_id": "err_role"}, + result_json={}, + checkpoint_json={}, + error="岗位情报 Provider 暂不可用，请配置后重试。", + retryable=True, + attempt_count=1, + max_attempts=3, + created_at=finished_at, + finished_at=finished_at, + ) + ) + + worker_soak._assert_job_saved_receipt(event, director_task, role_task) + worker_soak._assert_role_task_terminal(role_task, "101") From c23f7379c7d4164b507f9533e8ff32e00dd1f53e Mon Sep 17 00:00:00 2001 From: bekkilove <921693422@qq.com> Date: Sun, 11 Oct 2026 17:17:24 +0800 Subject: [PATCH 2/2] Handle either JOB_SAVED terminal receipt --- .../e2e/test_public_release_worker_soak.py | 61 +++++++--- .../tests/test_public_release_worker_soak.py | 107 +++++++++++------- 2 files changed, 115 insertions(+), 53 deletions(-) diff --git a/backend/scripts/e2e/test_public_release_worker_soak.py b/backend/scripts/e2e/test_public_release_worker_soak.py index db672d03..832613ac 100644 --- a/backend/scripts/e2e/test_public_release_worker_soak.py +++ b/backend/scripts/e2e/test_public_release_worker_soak.py @@ -164,23 +164,42 @@ def _wait_for_role_tasks( ) -def _has_director_event_receipt( +def _event_receipt_source( event: dict[str, Any], director_task: dict[str, Any], -) -> bool: + role_task: dict[str, Any], +) -> str | None: result = event.get("result") if isinstance(event.get("result"), dict) else {} - return ( - bool(event.get("processed_at")) - and event.get("status") == director_task.get("status") + if not event.get("processed_at"): + return None + if ( + event.get("status") == director_task.get("status") and result.get("task_id") == director_task.get("task_id") - and bool(result.get("inbox_item_id")) - ) + and result.get("inbox_item_id") == f"automation_task_{director_task.get('task_id')}" + ): + return "career_director" + if ( + role_task.get("status") == "completed" + and event.get("status") == "completed" + and result.get("task_id") == role_task.get("task_id") + ): + role_result = role_task.get("result") if isinstance(role_task.get("result"), dict) else {} + if result.get("benchmark_run_id") == role_result.get("benchmark_run_id"): + return "role_intelligence" + if ( + role_task.get("status") in {"blocked", "failed"} + and event.get("status") == role_task.get("status") + and result.get("task_id") == role_task.get("task_id") + ): + return "role_intelligence" + return None def _wait_for_job_saved_events( client: httpx.Client, job_target_ids: set[str], director_tasks_by_job: dict[str, dict[str, Any]], + role_tasks_by_job: dict[str, dict[str, Any]], ) -> list[dict[str, Any]]: deadline = time.monotonic() + TASK_TIMEOUT_SECONDS while time.monotonic() < deadline: @@ -204,7 +223,11 @@ def _wait_for_job_saved_events( raise AssertionError(f"Job {target_id} has duplicate JOB_SAVED events") events_by_job[target_id] = event if set(events_by_job) == job_target_ids and all( - _has_director_event_receipt(event, director_tasks_by_job[target_id]) + _event_receipt_source( + event, + director_tasks_by_job[target_id], + role_tasks_by_job[target_id], + ) for target_id, event in events_by_job.items() ): return matching @@ -214,8 +237,10 @@ def _wait_for_job_saved_events( target_id for target_id in sorted(job_target_ids) if target_id not in events_by_job - or not _has_director_event_receipt( - events_by_job[target_id], director_tasks_by_job[target_id] + or not _event_receipt_source( + events_by_job[target_id], + director_tasks_by_job[target_id], + role_tasks_by_job[target_id], ) ] raise AssertionError( @@ -239,8 +264,8 @@ def _assert_job_saved_receipt( role_input = role_task.get("input") if isinstance(role_task.get("input"), dict) else {} if event.get("event_type") != "JOB_SAVED" or not event_id: raise AssertionError(f"Job {target_id} lacks a persisted JOB_SAVED event ID") - if not _has_director_event_receipt(event, director_task): - raise AssertionError(f"JOB_SAVED event {event_id} lacks its Career Director receipt") + if not _event_receipt_source(event, director_task, role_task): + raise AssertionError(f"JOB_SAVED event {event_id} lacks a task projection receipt") if ( str(director_task.get("target_id") or "") != target_id or director_input.get("automation_event_id") != event_id @@ -400,6 +425,7 @@ def main() -> None: client, job_target_ids, director_tasks_by_job, + role_tasks_by_job, ) events_by_job = {str(event["target_id"]): event for event in matching_events} role_tasks_by_job = {str(task["target_id"]): task for task in role_tasks} @@ -408,6 +434,7 @@ def main() -> None: role_task_ids: set[str] = set() role_task_statuses: list[str] = [] + receipt_sources: list[str] = [] for task in role_tasks: target_id = str(task.get("target_id") or "") task_id = str(task.get("task_id") or "") @@ -418,11 +445,15 @@ def main() -> None: raise AssertionError(f"Job {target_id} has an unexpected Role Intelligence task: {task}") status = str(task.get("status") or "") role_task_statuses.append(status) - _assert_job_saved_receipt( + receipt_source = _event_receipt_source( events_by_job[target_id], director_tasks_by_job[target_id], role_tasks_by_job[target_id], ) + if receipt_source is None: + raise AssertionError(f"Job {target_id} lacks a terminal AutomationEvent receipt") + receipt_sources.append(receipt_source) + _assert_job_saved_receipt(events_by_job[target_id], director_tasks_by_job[target_id], task) _assert_role_task_terminal(task, target_id) integrity = _request(client, "GET", "/api/agent/data/safety/integrity") @@ -442,6 +473,10 @@ def main() -> None: "unique_automation_events": len(matching_events), "task_statuses": sorted(set(statuses)), "role_intelligence_statuses": sorted(set(role_task_statuses)), + "automation_event_receipt_sources": { + source: receipt_sources.count(source) + for source in ("career_director", "role_intelligence") + }, "task_attempts": 1, "database_integrity": integrity_check, "foreign_key_violations": foreign_key_violations, diff --git a/backend/tests/test_public_release_worker_soak.py b/backend/tests/test_public_release_worker_soak.py index c9fe271d..1712bb10 100644 --- a/backend/tests/test_public_release_worker_soak.py +++ b/backend/tests/test_public_release_worker_soak.py @@ -54,26 +54,25 @@ def test_job_saved_event_serializer_exposes_terminal_task_receipt( director_task_id = "career_task_release_director" role_task_id = "career_task_release_role" finished_at = datetime.now(timezone.utc).replace(tzinfo=None) - event = _event_view( - AutomationEvent( - event_id=event_id, - event_type="JOB_SAVED", - source="manual", - target_type="job", - target_id="101", - payload_json={"job_id": 101, "runtime_provider": "auto"}, - dedupe_key="job-saved:job:101", - status="blocked", - result_json={ - "task_id": director_task_id, - "inbox_item_id": f"automation_task_{director_task_id}", - "role_intelligence_task_id": None, - }, - error="", - created_at=finished_at, - processed_at=finished_at, - ) + event_row = AutomationEvent( + event_id=event_id, + event_type="JOB_SAVED", + source="manual", + target_type="job", + target_id="101", + payload_json={"job_id": 101, "runtime_provider": "auto"}, + dedupe_key="job-saved:job:101", + status="blocked", + result_json={ + "task_id": director_task_id, + "inbox_item_id": f"automation_task_{director_task_id}", + "role_intelligence_task_id": None, + }, + error="", + created_at=finished_at, + processed_at=finished_at, ) + event = _event_view(event_row) director_task = _task_view( CareerTask( task_id=director_task_id, @@ -100,28 +99,56 @@ def test_job_saved_event_serializer_exposes_terminal_task_receipt( finished_at=finished_at, ) ) - role_task = _task_view( - CareerTask( - task_id=role_task_id, - task_type="role_intelligence", - source="automation", - target_type="job", - target_id="101", - runtime_provider="auto", - input_json={"automation_event_id": event_id, "job_id": 101}, - output_contract_json={}, - status="blocked", - progress_json={"stage": "blocked", "error_id": "err_role"}, - result_json={}, - checkpoint_json={}, - error="岗位情报 Provider 暂不可用，请配置后重试。", - retryable=True, - attempt_count=1, - max_attempts=3, - created_at=finished_at, - finished_at=finished_at, - ) + role_row = CareerTask( + task_id=role_task_id, + task_type="role_intelligence", + source="automation", + target_type="job", + target_id="101", + runtime_provider="auto", + input_json={"automation_event_id": event_id, "job_id": 101}, + output_contract_json={}, + status="blocked", + progress_json={"stage": "blocked", "error_id": "err_role"}, + result_json={}, + checkpoint_json={}, + error="岗位情报 Provider 暂不可用，请配置后重试。", + retryable=True, + attempt_count=1, + max_attempts=3, + created_at=finished_at, + finished_at=finished_at, ) + role_task = _task_view(role_row) + assert worker_soak._event_receipt_source(event, director_task, role_task) == "career_director" worker_soak._assert_job_saved_receipt(event, director_task, role_task) worker_soak._assert_role_task_terminal(role_task, "101") + + benchmark_run_id = "benchmark_release_role" + role_row.status = "completed" + role_row.progress_json = {"stage": "completed", "percent": 100} + role_row.result_json = { + "job_id": 101, + "benchmark_run_id": benchmark_run_id, + "benchmark": {"status": "completed"}, + } + role_row.result_ref = f"career-task:{role_task_id}" + role_row.error = "" + role_row.retryable = False + completed_role_task = _task_view(role_row) + event_row.status = "completed" + event_row.result_json = { + "task_id": role_task_id, + "benchmark_run_id": benchmark_run_id, + "resume_candidate_status": None, + "application_packet_status": "partial", + } + role_first_event = _event_view(event_row) + + assert ( + worker_soak._event_receipt_source(role_first_event, director_task, completed_role_task) + == "role_intelligence" + ) + worker_soak._assert_job_saved_receipt(role_first_event, director_task, completed_role_task) + worker_soak._assert_role_task_terminal(completed_role_task, "101")