datamanager should not directly access the S3 DB with creating a session , change it to use wt-data-gateway instead - #39
Merged
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughThe PR adds evaluation-summary persistence, latest-checkout session reads, and job-partitioned cloud record operations. Cloud reward commits handle sessions without trainable trajectory rows. ChangesCloud evaluation storage
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant RewardCommitter
participant DataManager
participant CloudStrategy
participant LandingTable
RewardCommitter->>DataManager: record_evaluation_summary(session, step, reward, env_state)
DataManager->>CloudStrategy: record_evaluation_summary(...)
CloudStrategy->>LandingTable: persist terminal evaluation summary
LandingTable-->>CloudStrategy: persistence result
CloudStrategy-->>DataManager: record ID
DataManager-->>RewardCommitter: record ID
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (2)
core/data_manager/strategy/cloud_strategy_impl.py (2)
593-597: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd an explicit
strict=tozip().Ruff flags
B905on this call.recordsandrecord_idsare built in the same loop, sostrict=Truedocuments the invariant and satisfies the rule.♻️ Proposed change
- for record, record_id in zip(records, record_ids): + for record, record_id in zip(records, record_ids, strict=True):🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 593 - 597, Update the zip call in the record-processing loop to pass strict=True, documenting that records and record_ids must have equal lengths while satisfying Ruff B905. Keep the existing job_id mapping and return behavior unchanged.Source: Linters/SAST tools
847-849: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTrace label
filter_landingno longer matches the SDK method. Both call sites now invokeself.client.query_landingbut still reportsdk_operation="filter_landing". Traces and the DLDB metrics log therefore name a method that is no longer called._timed_db_callalso classifies the operation withsdk_operation.startswith("filter"), so renaming the label alone would flip these reads todb_write.
core/data_manager/strategy/cloud_strategy_impl.py#L847-L849: rename the label toquery_landinginlist_session_steps.core/data_manager/strategy/cloud_strategy_impl.py#L958-L960: rename the label toquery_landinginmark_latest_session_completed.Update the read/write classification in
_timed_db_callin the same change, for example by matching("filter", "query")prefixes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 847 - 849, Rename the _timed_db_call operation label from filter_landing to query_landing in list_session_steps at core/data_manager/strategy/cloud_strategy_impl.py#L847-L849 and mark_latest_session_completed at core/data_manager/strategy/cloud_strategy_impl.py#L958-L960. Update _timed_db_call’s read classification to recognize both filter and query prefixes so query_landing remains classified as a database read rather than db_write.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@core/data_manager/strategy/cloud_strategy_impl.py`:
- Around line 1139-1159: Make _load_existing_meta_json asynchronous and replace
its direct query_landing call with an awaited _timed_db_call so the database
operation runs off the event loop and is traced. Propagate the async change
through _normalize_session_step_updates_for_cloud, and await the normalizer from
update_session_step.
- Around line 942-950: Update the read that selects the latest persisted step
before sealing it, in the surrounding method containing _job_id_for_session and
the fallback warning, to pass checkout_latest=True. Ensure the same option is
applied to the corresponding reads at the additionally noted locations so the
method observes rows written by other processes and seals the actual newest
step.
- Around line 886-929: Update record_evaluation_summary to create and pass a
detached SessionContext copy to record_step instead of the cached session,
preserving the live session’s total_reward and message_history. Populate the
copy with the same session metadata and current job context, then ensure the
persisted summary row uses the evaluation score for both step_reward and reward,
matching the sqlite path.
- Around line 599-666: Update record_session_close handling around
mark_records_completed so the ValueError raised when job_id is unavailable is
treated as a no-op and still emits the completed status. Preserve existing
behavior for other exceptions, including their current reporting or propagation.
---
Nitpick comments:
In `@core/data_manager/strategy/cloud_strategy_impl.py`:
- Around line 593-597: Update the zip call in the record-processing loop to pass
strict=True, documenting that records and record_ids must have equal lengths
while satisfying Ruff B905. Keep the existing job_id mapping and return behavior
unchanged.
- Around line 847-849: Rename the _timed_db_call operation label from
filter_landing to query_landing in list_session_steps at
core/data_manager/strategy/cloud_strategy_impl.py#L847-L849 and
mark_latest_session_completed at
core/data_manager/strategy/cloud_strategy_impl.py#L958-L960. Update
_timed_db_call’s read classification to recognize both filter and query prefixes
so query_landing remains classified as a database read rather than db_write.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 968276cf-7f2b-4080-8adb-8cafaaeea94f
📒 Files selected for processing (5)
core/data_manager/manager.pycore/data_manager/strategy/base_strategy.pycore/data_manager/strategy/cloud_strategy_impl.pyevaluator/reward_committer.pyevaluator/trajectory_reader.py
Comment on lines
599
to
666
| async def mark_records_completed(self, record_ids: List[str]) -> int: | ||
| """Mark known landing record IDs completed without scanning session rows.""" | ||
| """Mark known landing record IDs completed in their associated HASH buckets.""" | ||
| await self.init() | ||
| unique_ids = list(dict.fromkeys(str(record_id) for record_id in record_ids if record_id)) | ||
| if not unique_ids: | ||
| return 0 | ||
| if self._enable_buffer: | ||
| await self._flush_records() | ||
|
|
||
| quoted_ids = ", ".join(f"'{_escape_sql_literal(record_id)}'" for record_id in unique_ids) | ||
| await self._timed_db_call( | ||
| "update_landing", | ||
| self.client.update_landing, | ||
| f"id IN ({quoted_ids})", | ||
| { | ||
| "is_session_completed": True, | ||
| "is_terminal": True, | ||
| }, | ||
| trace_context={"record_count": len(unique_ids)}, | ||
| ) | ||
| ids_by_job: Dict[str, List[str]] = {} | ||
| inferred_ids: List[str] = [] | ||
| missing_job_ids: List[str] = [] | ||
| for record_id in unique_ids: | ||
| job_id = self._record_job_ids.get(record_id) | ||
| if not job_id and self.job_id: | ||
| job_id = str(self.job_id) | ||
| inferred_ids.append(record_id) | ||
| if not job_id: | ||
| missing_job_ids.append(record_id) | ||
| continue | ||
| ids_by_job.setdefault(job_id, []).append(record_id) | ||
|
|
||
| if inferred_ids: | ||
| log.warning( | ||
| "Record-to-job association unavailable for %d landing records; " | ||
| "falling back to configured job_id=%s", | ||
| len(inferred_ids), | ||
| self.job_id, | ||
| ) | ||
| if missing_job_ids: | ||
| log.error( | ||
| "Cannot mark %d landing records completed without job_id; " | ||
| "refusing an all-bucket HASH update", | ||
| len(missing_job_ids), | ||
| ) | ||
| raise ValueError( | ||
| "job_id is required to mark landing records completed without " | ||
| "scanning all HASH buckets" | ||
| ) | ||
|
|
||
| for job_id, job_record_ids in ids_by_job.items(): | ||
| quoted_ids = ", ".join( | ||
| f"'{_escape_sql_literal(record_id)}'" | ||
| for record_id in job_record_ids | ||
| ) | ||
| filter_query = ( | ||
| f"job_id = '{_escape_sql_literal(job_id)}' " | ||
| f"AND id IN ({quoted_ids})" | ||
| ) | ||
| await self._timed_db_call( | ||
| "update_landing", | ||
| self.client.update_landing, | ||
| filter_query, | ||
| { | ||
| "is_session_completed": True, | ||
| "is_terminal": True, | ||
| }, | ||
| partition=job_id, | ||
| trace_context={ | ||
| "job_id": job_id, | ||
| "record_count": len(job_record_ids), | ||
| }, | ||
| ) | ||
| for record_id in job_record_ids: | ||
| self._record_job_ids.pop(record_id, None) | ||
|
|
||
| log.debug("Marked %d known cloud records completed", len(unique_ids)) | ||
| return len(unique_ids) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find callers of mark_records_completed and their error handling.
rg -nP -C 8 '\bmark_records_completed\s*\(' --type=pyRepository: AI45Lab/SAfactory
Length of output: 155
🏁 Script executed:
#!/bin/bash
set -e
echo "Repository files matching cloud_strategy_impl.py:"
fd -a 'cloud_strategy_impl\.py$' . || true
echo
echo "Git status/stat:"
git diff --stat || true
git status --short || true
echo
echo "Search mark_records_completed in tracked files:"
git ls-files | xargs grep -n "mark_records_completed\|ValueError" | sed -n '1,200p'
echo
echo "Inspect target file outline/section if present:"
target=$(fd 'cloud_strategy_impl\.py$' . | head -n1)
if [ -n "$target" ]; then
echo "target=$target"
wc -l "$target"
sed -n '560,680p' "$target" | nl -ba -v 560
fiRepository: AI45Lab/SAfactory
Length of output: 20020
🏁 Script executed:
#!/bin/bash
set -eu
python3 - <<'PY'
from pathlib import Path
target = Path("core/data_manager/strategy/cloud_strategy_impl.py")
print("mark_records_completed occurrences in cloud_strategy_impl.py:")
lines = target.read_text().splitlines()
for i,l in enumerate(lines, 1):
if "mark_records_completed" in l or "raise ValueError" in l or "'job_id is required' in" not in l:
if i == 634:
# do not print unrelated lines
pass
print(" (occurrences already listed above; no direct caller in cloud_strategy_impl.py except no-op context)")
for name, path in (
("data_manager_manager", Path("core/data_manager/manager.py")),
("gateway_storage", Path("gateway/storage.py")),
):
p = Path(path)
if p.exists():
print(f"\n--- {path} ---")
for i,l in enumerate(p.read_text().splitlines(), 1):
if "mark_records_completed" in l:
start=max(1,i-12); end=min(p.read_text().count("\n") + 1, i+20)
for j in range(start,end+1):
prefix=f"{j:5d} "
print(prefix + lines[j-1] if name=="cloud_strategy_impl" else p.read_text().splitlines()[j-1])
PY
echo
echo "All direct calls to mark_records_completed in tracked Python paths:"
python3 - <<'PY'
import sys, ast
from pathlib import Path
matches=[]
for p in Path(".").rglob("*.py"):
if p.parts[0] in {".git","env","evaluator"}:
continue
try:
s=p.read_text()
except Exception:
continue
try:
t=ast.parse(s)
except Exception:
continue
for node in ast.walk(t):
if isinstance(node, ast.Await) and isinstance(node.value, ast.Call):
call=node.value
elif isinstance(node, ast.Call):
call=node
else:
continue
func=call.func
if isinstance(func, ast.Attribute) and func.attr == "mark_records_completed":
line_no=node.lineno
context=max(0,line_no-10); context2=min(len(s.splitlines()),line_no+12)
matches.append((str(p), line_no, s.splitlines()[line_no-1], context, context2))
for m in matches:
file, line, stmt, context, context2 = m
print(f"\nFile: {file}:{line}")
print(" call:", stmt)
for j in range(context+1, context2+1):
print(f" {j:5d}: {s[j-1]}" if (s:=(Path(file).read_text().splitlines()) if False else None) is not None else)
PY
echo
echo "Inspect gateway/storage.py around direct call:"
sed -n '410,450p' gateway/storage.pyRepository: AI45Lab/SAfactory
Length of output: 4632
🏁 Script executed:
#!/bin/bash
set -eu
echo "Direct callers with surrounding control-flow:"
python3 - <<'PY'
from pathlib import Path
import ast
targets = []
for p in Path(".").rglob("*.py"):
if str(p).startswith((".git/", "env/", "evaluator/")):
continue
try:
s = p.read_text()
t = ast.parse(s)
except Exception:
continue
for i, node in enumerate(ast.iter_child_nodes(t)):
if isinstance(node, ast.Await) and isinstance(node.value, ast.Call):
if isinstance(node.value.func, ast.Attribute) and node.value.func.attr == "mark_records_completed":
targets.append((p, node.value.func.value.lineno, node.value.func.value.col_offset, node.value.func, node))
elif isinstance(node, ast.Call):
if isinstance(node.func, ast.Attribute) and node.func.attr == "mark_records_completed":
targets.append((p, node.func.value.lineno, node.func.value.col_offset, node.func, node))
lines_cache = {}
for p, outer_line, col, func, node in targets:
print(f"\nFile {p}:{outer_line}")
src = lines_cache.get(p)
if src is None:
src = p.read_text().splitlines()
lines_cache[p] = src
print(f" inner={func.lineno}:{func.col_offset} col={col}")
print(f" expr={src[func.lineno-1]}")
print(f" wait={src[node.lineno-1]}")
PY
echo
echo "Inspect storage.py session_close path:"
sed -n '380,470p' gateway/storage.py
echo
echo "Inspect manager.py callers around mark_records_completed:"
grep -n "mark_records_completed" core/data_manager/manager.py gateway/storage.pyRepository: AI45Lab/SAfactory
Length of output: 5078
🏁 Script executed:
#!/bin/bash
set -eu
echo "gateway/storage.py record_session_close completion path:"
sed -n '470,540p' gateway/storage.py
echo
echo "gateway/storage.py broader error handling for record_session_close callers:"
rg -n -C 6 "record_session_close|except|ValueError" gateway/app.py || true
echo
echo "Inspect cloud strategy init/job_id relevant section:"
sed -n '1,180p' core/data_manager/strategy/cloud_strategy_impl.py
sed -n '620,645p' core/data_manager/strategy/cloud_strategy_impl.py
sed -n '160,230p' core/data_manager/strategy/base_strategy.py
sed -n '130,150p' core/data_manager/manager.pyRepository: AI45Lab/SAfactory
Length of output: 21758
Handle ValueError from direct session-close callers.
gateway/storage.py calls self.data_manager.mark_records_completed(record_ids) under a generic except Exception block, but it immediately re-raises before reporting the completed status. If cloud storage uses this new ValueError, record_session_close currently propagates as a failure for the whole session-close call. Keep the existing path behavior or add a handling path that treats this as a no-op and emits completed status instead of raising.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 599 - 666,
Update record_session_close handling around mark_records_completed so the
ValueError raised when job_id is unavailable is treated as a no-op and still
emits the completed status. Preserve existing behavior for other exceptions,
including their current reporting or propagation.
Comment on lines
+1139
to
1159
| def _load_existing_meta_json( | ||
| self, | ||
| filter_query: str, | ||
| *, | ||
| job_id: Optional[str] = None, | ||
| ) -> Dict[str, Any]: | ||
| meta_json: Dict[str, Any] = {"source": "AIEvoBox"} | ||
| if not job_id: | ||
| log.warning( | ||
| "Loading landing meta_json without job_id; " | ||
| "falling back to an all-bucket HASH query" | ||
| ) | ||
| try: | ||
| df = self.client.session.filter( | ||
| self.client.config.tables.landing_table, | ||
| query=filter_query, | ||
| df = self.client.query_landing( | ||
| filter_query=filter_query, | ||
| limit=1, | ||
| columns=["meta_json"], | ||
| partition_cond=None, | ||
| partition=job_id or None, | ||
| checkout_latest=True, | ||
| as_dataframe=True, | ||
| ) |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
This blocking DB query runs on the event loop.
_load_existing_meta_json is synchronous and calls self.client.query_landing directly. The call chain is update_session_step (async, line 777) → _normalize_session_step_updates_for_cloud → this method. Every other DB call in this class goes through _timed_db_call, which offloads to a thread with asyncio.to_thread. This one does not, so it stalls the event loop for the query duration and produces no latency trace.
Make this helper async, await _timed_db_call, and await the normalizer from update_session_step.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 1139 - 1159,
Make _load_existing_meta_json asynchronous and replace its direct query_landing
call with an awaited _timed_db_call so the database operation runs off the event
loop and is traced. Propagate the async change through
_normalize_session_step_updates_for_cloud, and await the normalizer from
update_session_step.
hsballoon
reviewed
Aug 2, 2026
| columns=["step_id", "is_session_completed", "meta_json", "agent_model"], | ||
| partition_cond=None, | ||
| partition=job_id or None, | ||
| checkout_latest=False, |
Collaborator
There was a problem hiding this comment.
这个checkout_latest传false是否正确，需要确认一下。safactory其他地方都传的是true。
checkout_latest是lancedb提供的字段，他的含义可以简单理解为：
False：允许使用当前 dldb session 已经看到的表版本，开销较小。
True：查询前确保看到最新表版本，适合其他 worker/process 刚写完、当前进程马上要读的场景。
| self.client.session.filter, | ||
| self.client.config.tables.landing_table, | ||
| query=query, | ||
| self.client.query_landing, |
Collaborator
There was a problem hiding this comment.
query_landing这个方法在下午的讨论里改掉了(因为在这个pr之前没有人用这个方法T_T)，现在统一用query_data()，用法一样。调整一下
BinHuangPJLAB
force-pushed
the
gateway-bug-fix-1
branch
from
August 2, 2026 08:56
33ae7bf to
5bbe94d
Compare
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
core/data_manager/strategy/cloud_strategy_impl.py (1)
853-880: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
list_session_stepsdrops therequestfield on read; verify thequery_landingmethod name.Two concerns in this block:
meta_fieldsnow includes"request"(Line 1100) and_build_step_recordwrites it intometa_json(Line 727).fetch_done_steps_with_contextreads it back at Line 1479.list_session_steps, however, only extractsenv_state,group_id,dataset, andis_trainablefrommeta(Lines 871-878); it never setsrow["request"]. Any consumer oflist_session_stepsrows that expectsrequest(mirroring the other read path) gets nothing.- A previous human reviewer noted that
query_landingwas replaced withquery_data()during offline discussion, with an equivalent call signature. This call site (Line 853) still usesquery_landing, along with the other two call sites at Lines 964 and 1156.🐛 Proposed fix for issue 1
meta = _json_object(row.pop("meta_json", None)) row["llm_model"] = row.pop("agent_model", None) row["env_state"] = _json_object(meta.get("env_state")) row["group_id"] = meta.get("group_id") + row["request"] = meta.get("request") if "dataset" in meta: row["dataset"] = meta["dataset"]🤖 Verification script for the `query_landing`/`query_data` question
#!/bin/bash # Description: Confirm the current wt-data-gateway client method name for landing queries. rg -n 'def query_landing|def query_data' --type=py -g '!core/data_manager/strategy/cloud_strategy_impl.py' rg -n 'client\.query_landing|client\.query_data' --type=py🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 853 - 880, Update list_session_steps so it restores row["request"] from meta, matching fetch_done_steps_with_context and the metadata written by _build_step_record. Verify the cloud client’s supported landing-query method, then replace query_landing with query_data in this and the other relevant call sites if query_data is the current API.
♻️ Duplicate comments (4)
core/data_manager/strategy/cloud_strategy_impl.py (4)
946-971: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
checkout_latest=Falsecorrectness for latest-session reads is still unconfirmed.This method reads the latest step with
checkout_latest=False(Line 969) before sealing it as completed. Elsewhere in this PR, reads that may race with another writer passcheckout_latest=True(for example_load_existing_meta_json, and thelist_session_stepscallers inevaluator/reward_committer.pyandevaluator/trajectory_reader.py). A human reviewer also asked to confirm this choice, noting that other parts of the codebase consistently passTrue. WithFalse, this read can observe a stale table version and seal the wrong (older) step, or find nothing to complete. If the stale-read risk applies here too, passcheckout_latest=True.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 946 - 971, The latest-session read in the surrounding method uses checkout_latest=False, which can select stale data. Update the self.client.query_landing call in the latest-session completion flow to pass checkout_latest=True, matching the race-safe reads used by _load_existing_meta_json and related session-step callers.
1134-1163: 🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
_load_existing_meta_jsonstill runs a blocking DB call on the event loop.This helper is synchronous and calls
self.client.query_landingdirectly (Line 1156), while every other DB call in this class is routed through_timed_db_call, which offloads to a thread viaasyncio.to_thread. The call chain isupdate_session_step(async) →_normalize_session_step_updates_for_cloud→ this method, so the query stalls the event loop for its duration and produces no latency trace. This was raised previously; make the helper async, await_timed_db_call, and propagate theawaitthrough_normalize_session_step_updates_for_cloudand its caller.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 1134 - 1163, Make _load_existing_meta_json asynchronous and route query_landing through the existing _timed_db_call helper, awaiting its result to preserve latency tracing and thread offloading. Propagate await through _normalize_session_step_updates_for_cloud and its async caller update_session_step, while preserving the existing query arguments and metadata behavior.
601-668: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winAll-or-nothing failure blocks legitimate completions; downstream
ValueErrorhandling still unconfirmed.Two issues remain in
mark_records_completed:
- The loop at Lines 613-621 builds
ids_by_jobfor everyrecord_idthat resolves a job, and separately collectsmissing_job_ids. Line 630 checksif missing_job_ids:and raises before Line 641 processesids_by_job. If even onerecord_idlacks a resolvablejob_id, none of the other, already-resolved records get marked completed. The caller loses valid work due to one unrelated record.- This was flagged before:
gateway/storage.pycallsmark_records_completedand re-raises under a generic exception handler. That caller still needs to treat thisValueErroras a no-op that still reports completed status, not a hard failure.🐛 Proposed fix for issue 1: process resolvable buckets before raising
- if missing_job_ids: - log.error( - "Cannot mark %d landing records completed without job_id; " - "refusing an all-bucket HASH update", - len(missing_job_ids), - ) - raise ValueError( - "job_id is required to mark landing records completed without " - "scanning all HASH buckets" - ) - for job_id, job_record_ids in ids_by_job.items(): quoted_ids = ", ".join( f"'{_escape_sql_literal(record_id)}'" for record_id in job_record_ids ) filter_query = ( f"job_id = '{_escape_sql_literal(job_id)}' " f"AND id IN ({quoted_ids})" ) await self._timed_db_call( "update_landing", self.client.update_landing, filter_query, { "is_session_completed": True, "is_terminal": True, }, partition=job_id, trace_context={ "job_id": job_id, "record_count": len(job_record_ids), }, ) for record_id in job_record_ids: self._record_job_ids.pop(record_id, None) + + if missing_job_ids: + log.error( + "Cannot mark %d landing records completed without job_id; " + "refusing an all-bucket HASH update", + len(missing_job_ids), + ) + raise ValueError( + "job_id is required to mark landing records completed without " + "scanning all HASH buckets" + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 601 - 668, Update mark_records_completed to process all ids_by_job buckets before handling missing_job_ids, so resolvable records are completed even when unrelated records lack job associations; then raise ValueError only after those updates finish, preserving the existing refusal to perform an all-bucket update. Also update the caller handling mark_records_completed in the storage gateway so this ValueError is treated as a no-op that still reports completion rather than being re-raised by the generic exception path.
890-933: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftReused session still inflates the persisted reward for evaluation summaries.
record_evaluation_summarystill reuses the cachedSessionContextfromself._sessionsand passes it torecord_step._build_step_recordperformssession.total_reward += step_rewardand writesreward=session.total_reward. If the session already has trajectory steps, the persistedrewardbecomes the accumulated trajectory reward plus the evaluation score, not the evaluation score alone. This also overwritessession.message_historywith[], discarding the S3-URL prefix cache for a session that could still be active. This was raised previously with a detailed fix using a detachedSessionContextcopy; that fix is not yet applied here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@core/data_manager/strategy/cloud_strategy_impl.py` around lines 890 - 933, Update record_evaluation_summary to create and pass a detached SessionContext copy to record_step, preserving the cached session’s accumulated reward and message_history. Copy the required session metadata, including session_id, env_id, env_name, llm_model, and job_id, while using empty messages only on the detached context so the persisted evaluation reward remains evaluation-only and the active session remains unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@core/data_manager/strategy/cloud_strategy_impl.py`:
- Around line 853-880: Update list_session_steps so it restores row["request"]
from meta, matching fetch_done_steps_with_context and the metadata written by
_build_step_record. Verify the cloud client’s supported landing-query method,
then replace query_landing with query_data in this and the other relevant call
sites if query_data is the current API.
---
Duplicate comments:
In `@core/data_manager/strategy/cloud_strategy_impl.py`:
- Around line 946-971: The latest-session read in the surrounding method uses
checkout_latest=False, which can select stale data. Update the
self.client.query_landing call in the latest-session completion flow to pass
checkout_latest=True, matching the race-safe reads used by
_load_existing_meta_json and related session-step callers.
- Around line 1134-1163: Make _load_existing_meta_json asynchronous and route
query_landing through the existing _timed_db_call helper, awaiting its result to
preserve latency tracing and thread offloading. Propagate await through
_normalize_session_step_updates_for_cloud and its async caller
update_session_step, while preserving the existing query arguments and metadata
behavior.
- Around line 601-668: Update mark_records_completed to process all ids_by_job
buckets before handling missing_job_ids, so resolvable records are completed
even when unrelated records lack job associations; then raise ValueError only
after those updates finish, preserving the existing refusal to perform an
all-bucket update. Also update the caller handling mark_records_completed in the
storage gateway so this ValueError is treated as a no-op that still reports
completion rather than being re-raised by the generic exception path.
- Around line 890-933: Update record_evaluation_summary to create and pass a
detached SessionContext copy to record_step, preserving the cached session’s
accumulated reward and message_history. Copy the required session metadata,
including session_id, env_id, env_name, llm_model, and job_id, while using empty
messages only on the detached context so the persisted evaluation reward remains
evaluation-only and the active session remains unchanged.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: e8e3cc17-36d8-4c0b-acb4-d4f77b07507d
📒 Files selected for processing (5)
core/data_manager/manager.pycore/data_manager/strategy/base_strategy.pycore/data_manager/strategy/cloud_strategy_impl.pyevaluator/reward_committer.pyevaluator/trajectory_reader.py
🚧 Files skipped from review as they are similar to previous changes (4)
- evaluator/trajectory_reader.py
- core/data_manager/manager.py
- evaluator/reward_committer.py
- core/data_manager/strategy/base_strategy.py
hsballoon
approved these changes
Aug 2, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
New Features
Bug Fixes