Skip to content

Commit af88bd9

Browse files
kyle-computekyle-compute
committed
Scope the audit admission gate to real QA work; savepoint the healer
Full-suite feedback on the audit gate. A task whose eligible set is empty (every agent trial cancelled by the baseline gate) creates no QA trial -- it just completes -- so deferring it on a live audit only delayed completion for nothing: the gate now defers only when a QA trial would actually be created, and the audit's import still lands on the version regardless of task status. The verdict-pending healer also ran each task's repair unprotected in the sweep's shared transaction: one unrepairable task (its experiment memberships gone, so QA creation raises) aborted the whole step and, being first in the updated_at ordering, starved every task behind it on every sweep. Each task now repairs inside its own savepoint, matching the stale-job reaper's isolation. Claude-Session: https://claude.ai/code/session_01ACF6SUXdbLFarpj3qwF1Ki
1 parent cd1c5bb commit af88bd9

3 files changed

Lines changed: 100 additions & 69 deletions

File tree

‎oddish/src/oddish/queue.py‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1632,10 +1632,17 @@ async def maybe_start_task_qa_stage(
16321632
# created QA trial is never rebuilt when the audit lands later: starting
16331633
# now would permanently bake "(none recorded)" into the brief. Defer
16341634
# while an audit trial is live -- the same gate the manual QA endpoint
1635-
# applies. The audit's own settlement re-enters this admission
1635+
# applies -- but only when a QA trial would actually be created: with
1636+
# nothing eligible the task just completes (no brief exists to poison),
1637+
# and the audit's later import writes onto the version regardless of
1638+
# task status. The audit's own settlement re-enters this admission
16361639
# (handle_analysis_trial_settled), so deferring cannot strand the task.
16371640
if await live_analysis_trial_id(session, task_id, kind="audit") is not None:
1638-
return TaskQAStageAdmission()
1641+
eligible = await qa_eligible_trial_ids(
1642+
session, task_id, task_version_id=task.current_version_id
1643+
)
1644+
if eligible:
1645+
return TaskQAStageAdmission()
16391646

16401647
await start_qa_for_task(session, task)
16411648
await session.flush()

‎oddish/src/oddish/workers/queue/cleanup.py‎

Lines changed: 85 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -1385,74 +1385,92 @@ async def _heal_stale_verdict_pending(session) -> int:
13851385
verdict_pending_completed = 0
13861386
reimport_trial_ids: list[str] = []
13871387
for (task_id,) in stale_verdict_pending:
1388-
task = (
1389-
await session.execute(
1390-
select(TaskModel).where(TaskModel.id == str(task_id)).with_for_update()
1391-
)
1392-
).scalar_one_or_none()
1393-
if not task or task.status != TaskStatus.VERDICT_PENDING:
1394-
continue
1395-
# The candidate scan precedes the row lock. A trial settlement may
1396-
# have created a fresh QA trial while cleanup waited, so recheck
1397-
# after locking before repairing state or creating a duplicate.
1398-
active_qa = await session.scalar(
1399-
text(
1400-
"""
1401-
SELECT 1 FROM trials
1402-
WHERE task_id = :task_id AND kind = 'qa'
1403-
AND deleted_at IS NULL
1404-
AND superseded_by_trial_id IS NULL
1405-
AND status::text NOT IN ('SUCCESS', 'FAILED', 'SKIPPED')
1406-
LIMIT 1
1407-
"""
1408-
),
1409-
{"task_id": task.id},
1410-
)
1411-
if active_qa is not None:
1412-
continue
1413-
if task.verdict_status in (VerdictStatus.SUCCESS, VerdictStatus.FAILED):
1414-
task.status = TaskStatus.COMPLETED
1415-
task.finished_at = task.finished_at or utcnow()
1416-
verdict_pending_completed += 1
1417-
continue
1418-
# A terminal QA trial with a non-terminal verdict means the import
1419-
# never landed (worker died between settle and import). Re-import
1420-
# after this transaction; only create a fresh QA trial when none
1421-
# exists.
1422-
settled_qa = await session.scalar(
1423-
text(
1424-
"""
1425-
SELECT tr.id FROM trials tr
1426-
WHERE tr.task_id = :task_id AND tr.kind = 'qa'
1427-
AND tr.deleted_at IS NULL
1428-
AND tr.superseded_by_trial_id IS NULL
1429-
AND tr.status::text IN ('SUCCESS', 'FAILED')
1430-
ORDER BY tr.created_at DESC LIMIT 1
1431-
"""
1432-
),
1433-
{"task_id": task.id},
1434-
)
1435-
if settled_qa is not None:
1436-
logger.info(
1437-
"healer: task %s has settled qa trial %s with no verdict, re-importing",
1438-
task.id,
1439-
settled_qa,
1440-
)
1441-
reimport_trial_ids.append(str(settled_qa))
1442-
continue
1443-
# start_qa_for_task itself has no audit gate; creating a QA trial
1444-
# while an audit is live would bake "(none recorded)" findings into
1445-
# its brief. Skip for now: the audit's settlement re-enters
1446-
# admission, and the next sweep retries regardless.
1447-
if await live_analysis_trial_id(session, task.id, kind="audit") is not None:
1448-
continue
1449-
if await start_qa_for_task(session, task):
1450-
logger.info(
1451-
"healer: task %s was wedged in VERDICT_PENDING with no qa trial",
1452-
task.id,
1388+
# Savepoint per task: one unrepairable task (e.g. its experiment
1389+
# memberships are gone, so QA creation raises) must not abort the
1390+
# step and starve every task behind it in the updated_at ordering.
1391+
try:
1392+
async with session.begin_nested():
1393+
task = (
1394+
await session.execute(
1395+
select(TaskModel)
1396+
.where(TaskModel.id == str(task_id))
1397+
.with_for_update()
1398+
)
1399+
).scalar_one_or_none()
1400+
if not task or task.status != TaskStatus.VERDICT_PENDING:
1401+
continue
1402+
# The candidate scan precedes the row lock. A trial
1403+
# settlement may have created a fresh QA trial while cleanup
1404+
# waited, so recheck after locking before repairing state or
1405+
# creating a duplicate.
1406+
active_qa = await session.scalar(
1407+
text(
1408+
"""
1409+
SELECT 1 FROM trials
1410+
WHERE task_id = :task_id AND kind = 'qa'
1411+
AND deleted_at IS NULL
1412+
AND superseded_by_trial_id IS NULL
1413+
AND status::text NOT IN ('SUCCESS', 'FAILED', 'SKIPPED')
1414+
LIMIT 1
1415+
"""
1416+
),
1417+
{"task_id": task.id},
1418+
)
1419+
if active_qa is not None:
1420+
continue
1421+
if task.verdict_status in (VerdictStatus.SUCCESS, VerdictStatus.FAILED):
1422+
task.status = TaskStatus.COMPLETED
1423+
task.finished_at = task.finished_at or utcnow()
1424+
verdict_pending_completed += 1
1425+
continue
1426+
# A terminal QA trial with a non-terminal verdict means the
1427+
# import never landed (worker died between settle and
1428+
# import). Re-import after this transaction; only create a
1429+
# fresh QA trial when none exists.
1430+
settled_qa = await session.scalar(
1431+
text(
1432+
"""
1433+
SELECT tr.id FROM trials tr
1434+
WHERE tr.task_id = :task_id AND tr.kind = 'qa'
1435+
AND tr.deleted_at IS NULL
1436+
AND tr.superseded_by_trial_id IS NULL
1437+
AND tr.status::text IN ('SUCCESS', 'FAILED')
1438+
ORDER BY tr.created_at DESC LIMIT 1
1439+
"""
1440+
),
1441+
{"task_id": task.id},
1442+
)
1443+
if settled_qa is not None:
1444+
logger.info(
1445+
"healer: task %s has settled qa trial %s with no "
1446+
"verdict, re-importing",
1447+
task.id,
1448+
settled_qa,
1449+
)
1450+
reimport_trial_ids.append(str(settled_qa))
1451+
continue
1452+
# start_qa_for_task itself has no audit gate; creating a QA
1453+
# trial while an audit is live would bake "(none recorded)"
1454+
# findings into its brief. Skip for now: the audit's
1455+
# settlement re-enters admission, and the next sweep retries
1456+
# regardless.
1457+
if (
1458+
await live_analysis_trial_id(session, task.id, kind="audit")
1459+
is not None
1460+
):
1461+
continue
1462+
if await start_qa_for_task(session, task):
1463+
logger.info(
1464+
"healer: task %s was wedged in VERDICT_PENDING "
1465+
"with no qa trial",
1466+
task.id,
1467+
)
1468+
else:
1469+
verdict_pending_completed += 1
1470+
except Exception: # noqa: BLE001 -- log and move to the next task
1471+
logger.exception(
1472+
"healer: verdict-pending repair failed for task %s", task_id
14531473
)
1454-
else:
1455-
verdict_pending_completed += 1
14561474

14571475
for trial_id in reimport_trial_ids:
14581476
try:

‎oddish/tests/test_analysis_trials.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -520,6 +520,12 @@ async def test_historical_trials_do_not_block_the_qa_import():
520520
trial.status = TrialStatus.RUNNING
521521
await session.flush()
522522
assert await _qa_import_still_current(session, task_id, v2) is False
523+
# Leave nothing for the sweep healer: this task has no experiment
524+
# membership, so a later test running the real cleanup sweep would
525+
# otherwise try (and fail) to create a QA trial for it.
526+
task = await session.get(TaskModel, task_id)
527+
task.status = TaskStatus.COMPLETED
528+
await session.commit()
523529

524530

525531
@pytest.mark.asyncio

0 commit comments

Comments
 (0)