[improvement](ci) Strengthen Codex review context and Litefuse observability (#65178)

The Codex GitHub Actions review pipeline needed stronger guarantees that
agents review the intended PR snapshot, use one authoritative diff,
pursue high-risk cross-module mechanisms, and expose the work of spawned
subagents in Litefuse.

This PR:

- verifies the checked-out PR head and rechecks the live base/head after
context preparation;
- prepares an authoritative aggregate PR patch and changed-file list,
with atomic bounded retry (`2s`, `4s`, `8s`) for transient GitHub API
failures;
- directs the main agent to perform an initial risk scan, dispatch
focused subagents for suspicious mechanisms, maintain a shared
append-only ledger, validate/deduplicate every candidate, and converge
for at most three rounds;
- requires paths outside the authoritative diff/changed-file context to
be confirmed with `rg --files` before reading;
- records separate Litefuse traces for each Codex subagent session,
including command/tool I/O and non-empty encrypted-reasoning status,
instead of mixing subagent activity into the main trace;
- makes Litefuse ingestion more resilient to transport timeouts while
keeping telemetry upload non-blocking for the review outcome.

The repository review skill is kept focused on review invariants;
orchestration and completion policy live in the workflow prompt, where
they are available to the GitHub Actions invocation.

### Validation and experiment material

- Static checks: `python3 -m py_compile
.github/scripts/emit_litefuse_otel_io.py` and `git diff --check`.
- Litefuse uploader validation: real/synthetic subagent session
conversion, payload-limit coverage, and readback of independent subagent
traces.
- End-to-end replayed review experiments against fresh fork PRs built
from the original PR base/head commits:

| Replay PR | GitHub Actions run | Result |
| --- | --- | --- |
| [#136](https://github.com/zclllyybb/doris/pull/136) |
[29112823062](https://github.com/zclllyybb/doris/actions/runs/29112823062)
| Success; 2 inline comments |
| [#137](https://github.com/zclllyybb/doris/pull/137) |
[29114260631](https://github.com/zclllyybb/doris/actions/runs/29114260631)
| Success; 3 inline comments |
| [#138](https://github.com/zclllyybb/doris/pull/138) |
[29115946034](https://github.com/zclllyybb/doris/actions/runs/29115946034)
| Success; 1 inline comment |
| [#139](https://github.com/zclllyybb/doris/pull/139) |
[29150372758](https://github.com/zclllyybb/doris/actions/runs/29150372758)
| Success; 1 inline comment |
| [#140](https://github.com/zclllyybb/doris/pull/140) |
[29151057520](https://github.com/zclllyybb/doris/actions/runs/29151057520)
| Success; 1 inline comment |

The latest replay trace, `86be226391a40e5992bb8d3b4e125eca`, confirms
one authoritative `pr.diff` read, two changed-file-list reads, no `gh pr
diff` or `git diff` calls from the reviewing agent, and separate
subagent sessions. It also exposed one remaining prompt-compliance gap:
a guessed adjacent `.cpp` path was read without a preceding `rg --files`
lookup; this is documented test evidence rather than hidden by the
validation.

.claude/skills/code-review/SKILL.md

@@ -14,13 +14,9 @@ Use this when you need to review code, whether it is code you just completed or
 
 ## How to use me
 
-0. **MANDATORY GOAL COMPLETION REQUIREMENT:** When the review is running in Codex goal mode, the goal is complete only after every changed file and relevant surrounding code path has been examined, every suspicious point has been accepted as an inline issue or dismissed with evidence, and every accepted issue has been submitted and verified on GitHub.
-1. **MANDATORY GOAL PROCESS REQUIREMENT:** The goal's progress tracking must cover instruction loading, subagent spawning, shared-ledger maintenance, candidate verification/deduplication, final subagent convergence, GitHub review submission, and GitHub API verification. The goal is not complete until every live subagent has said `NO_NEW_VALUABLE_FINDINGS` for the same current ledger/comment set after the last candidate update.
-2. **MANDATORY SUBAGENT REVIEW REQUIREMENT:** Use the available subagent or multi-agent spawn tool for focused review passes; do not merely simulate subagent output. The main agent must read the subagent results, independently verify or dismiss every candidate with concrete code evidence, deduplicate against existing review threads, submit the final GitHub review itself, and summarize the subagent conclusions.
-3. **MANDATORY SHARED LEDGER REQUIREMENT:** When a shared subagent review ledger is provided, every subagent must read the whole ledger and append findings only to its assigned subagent section. The main agent must use the ledger as the source of truth for merging, status updates, duplicate suppression, proposed final comments, and the final convergence round. Subagents must not edit another subagent section or any main-owned section; this section-owned append-only rule avoids concurrent patch conflicts while keeping all findings visible in one document.
-4. **Always read and respond to Part 1** (General Principles) — it applies to all code.
-5. For module-specific review, **read the `AGENTS.md` in the corresponding source directory** listed in Part 2. Those files contain non-obvious conventions and traps specific to each subsystem.
-6. Parts 3–7 cover cross-module concerns, testing, high-risk patterns, functions, and standards — refer as needed.
+1. **Always read and respond to Part 1** (General Principles) — it applies to all code.
+2. For module-specific review, **read the `AGENTS.md` in the corresponding source directory** listed in Part 2. Those files contain non-obvious conventions and traps specific to each subsystem.
+3. Parts 3–7 cover cross-module concerns, testing, high-risk patterns, functions, and standards — refer as needed.
 
 ---
 
@@ -29,6 +25,7 @@ Use this when you need to review code, whether it is code you just completed or
 ### 1.1 Core Invariants
 
 Always focus on the following core invariants during review:
+
 1. **Data Correctness**: Data from any committed transaction must be visible to subsequent queries and not lost; data from uncommitted transactions must not be visible
 2. **Version Consistency**: Partition visible version is the sole standard for data visibility; BE must strictly read data not exceeding this version
 3. **Delete Bitmap Consistency** (MoW tables): delete bitmap must be strictly aligned with rowset versions; sentinel marks and `TEMP_VERSION`-style placeholders must be replaced with actual versions before use
@@ -138,7 +135,7 @@ If it involves the judgment of concurrent scenarios, it is necessary to find the
 
 - [ ] **Reservation**: `try_reserve()` before large allocations, with guaranteed release on every exit path
 - [ ] **Allocator-aware ownership**: In BE code, for memory-owning containers or buffers that should be tracked by Doris memory accounting, prefer allocator-aware types from `be/src/core/custom_allocator.h` such as `DorisVector`, `DorisMap`, and `DorisUniqueBufferPtr` instead of `std::vector`, `std::map`, and `std::unique_ptr`
-- [ ] **Ownership**: See `be/src/runtime/AGENTS.md` for cache-handle, shared_ptr-cycle, and factory-creator rules
+- [ ] **Ownership**: See `be/src/runtime/AGENTS.md` for cache-handle, shared_ptr-cycle, and factory-creator rules. See `be/src/core/AGENTS.md` for Proper application of the cow mechanism related to `IColumn`, `Block`...
 
 #### 1.3.4 Data Correctness (High Priority)
 

.github/scripts/emit_litefuse_otel_io.py

@@ -488,6 +488,513 @@ def build_ingestion_payload(args, input_text, output_text, events):
     return trace_id, {"batch": batch}, len(batch) - 1
 
 
+def main_trace_metadata(args):
+    trace_metadata = {
+        "repository": args.repository,
+        "workflow": args.workflow,
+        "run_id": args.run_id,
+        "pr_number": args.pr_number,
+        "head_sha": args.head_sha,
+        "base_sha": args.base_sha,
+        "model_reasoning_effort": args.reasoning_effort,
+    }
+    return {
+        key: value for key, value in trace_metadata.items() if value not in (None, "")
+    }
+
+
+def receiver_thread_ids(events):
+    thread_ids = set()
+    for event in events:
+        item = event.get("item") if isinstance(event.get("item"), dict) else {}
+        if item.get("type") != "collab_tool_call":
+            continue
+        for thread_id in item.get("receiver_thread_ids") or []:
+            if thread_id:
+                thread_ids.add(str(thread_id))
+    return thread_ids
+
+
+def session_jsonl_files(root):
+    if not root or not os.path.isdir(root):
+        return []
+    paths = []
+    for dirpath, _dirnames, filenames in os.walk(root):
+        for filename in filenames:
+            if filename.endswith(".jsonl"):
+                paths.append(os.path.join(dirpath, filename))
+    return sorted(paths)
+
+
+def session_meta(events):
+    for event in events:
+        if event.get("type") != "session_meta":
+            continue
+        payload = event.get("payload")
+        if isinstance(payload, dict):
+            return payload
+    return {}
+
+
+def subagent_spawn(meta):
+    source = meta.get("source") if isinstance(meta, dict) else {}
+    source = source if isinstance(source, dict) else {}
+    subagent = source.get("subagent")
+    subagent = subagent if isinstance(subagent, dict) else {}
+    spawn = subagent.get("thread_spawn")
+    return spawn if isinstance(spawn, dict) else {}
+
+
+def is_subagent_meta(meta, expected_thread_ids):
+    if not isinstance(meta, dict):
+        return False
+    thread_id = str(meta.get("id") or "")
+    if expected_thread_ids:
+        return thread_id in expected_thread_ids
+    if meta.get("thread_source") == "subagent":
+        return True
+    source = meta.get("source")
+    return isinstance(source, dict) and bool(source.get("subagent"))
+
+
+def compact_session_meta(meta, max_json_chars):
+    compact = {}
+    for key in (
+        "id",
+        "parent_thread_id",
+        "timestamp",
+        "cwd",
+        "originator",
+        "cli_version",
+        "thread_source",
+        "agent_nickname",
+        "agent_role",
+        "model_provider",
+        "model",
+    ):
+        if meta.get(key) not in (None, ""):
+            compact[key] = meta.get(key)
+    for key in ("base_instructions", "user_instructions"):
+        if key in meta:
+            compact[f"{key}_present"] = True
+    source = meta.get("source")
+    if isinstance(source, dict):
+        compact["source"] = truncate_json(source, max_json_chars)
+    return compact
+
+
+def session_content_text(content, max_chars):
+    if isinstance(content, str):
+        return truncate_text(content, max_chars)
+    if not isinstance(content, list):
+        return truncate_json(content, max_chars)
+    parts = []
+    for item in content:
+        if isinstance(item, str):
+            parts.append(item)
+            continue
+        if not isinstance(item, dict):
+            parts.append(json_attr(item))
+            continue
+        for key in ("text", "input_text", "output_text"):
+            if isinstance(item.get(key), str):
+                parts.append(item[key])
+                break
+        else:
+            parts.append(json_attr(item))
+    return truncate_text("\n".join(parts), max_chars)
+
+
+def session_first_user_message(events, max_chars):
+    for event in events:
+        if event.get("type") != "response_item":
+            continue
+        payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
+        if payload.get("type") == "message" and payload.get("role") == "user":
+            text = session_content_text(payload.get("content"), max_chars)
+            if isinstance(text, str) and text.strip():
+                return text
+    return ""
+
+
+def session_final_assistant_message(events, max_chars):
+    for event in reversed(events):
+        payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
+        if event.get("type") == "response_item":
+            if payload.get("type") == "message" and payload.get("role") == "assistant":
+                text = session_content_text(payload.get("content"), max_chars)
+                if isinstance(text, str) and text.strip():
+                    return text
+        elif event.get("type") == "event_msg" and payload.get("type") == "agent_message":
+            text = payload.get("message") or ""
+            if text.strip():
+                return truncate_text(text, max_chars)
+    return ""
+
+
+def safe_name_component(value):
+    text = str(value or "unknown")
+    safe = "".join(
+        char if char.isalnum() or char in "._-" else "_" for char in text
+    ).strip("._-")
+    return safe[:120] or "unknown"
+
+
+def jsonish(value):
+    if not isinstance(value, str):
+        return value
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        return value
+
+
+def session_call_outputs(events):
+    outputs = {}
+    for event in events:
+        payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
+        if payload.get("type") != "function_call_output":
+            continue
+        call_id = payload.get("call_id")
+        if call_id and call_id not in outputs:
+            outputs[call_id] = (payload, event)
+    return outputs
+
+
+def session_event_timestamp(event, fallback_ns):
+    timestamp = event.get("timestamp")
+    if isinstance(timestamp, str) and timestamp:
+        return timestamp
+    return iso_from_ns(fallback_ns)
+
+
+def ns_from_iso(timestamp):
+    if not isinstance(timestamp, str) or not timestamp:
+        return None
+    try:
+        normalized = timestamp.replace("Z", "+00:00")
+        return int(datetime.fromisoformat(normalized).timestamp() * 1_000_000_000)
+    except ValueError:
+        return None
+
+
+def session_observation_shape(event, call_outputs, max_json_chars):
+    event_type = event.get("type") or "unknown"
+    payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
+
+    if event_type == "session_meta":
+        compact_meta = compact_session_meta(payload, max_json_chars)
+        return {
+            "name": "codex.subagent.session_meta",
+            "type": "span",
+            "input": compact_meta,
+            "output": {
+                "thread_source": compact_meta.get("thread_source"),
+                "status": "recorded",
+            },
+            "metadata": {"session_event_type": event_type},
+        }
+
+    if event_type == "turn_context":
+        return {
+            "name": "codex.subagent.turn_context",
+            "type": "span",
+            "input": truncate_json(payload, max_json_chars),
+            "output": {"status": "recorded"},
+            "metadata": {"session_event_type": event_type},
+        }
+
+    if event_type == "event_msg":
+        message_type = payload.get("type") or "unknown"
+        message = payload.get("message") or ""
+        body = {
+            key: value
+            for key, value in payload.items()
+            if key not in ("message", "text_elements", "images", "local_images")
+        }
+        if payload.get("text_elements"):
+            body["text_elements"] = truncate_json(
+                payload.get("text_elements"), max_json_chars
+            )
+        if payload.get("images"):
+            body["images"] = truncate_json(payload.get("images"), max_json_chars)
+        return {
+            "name": f"codex.subagent.event.{safe_name_component(message_type)}",
+            "type": "generation" if message_type == "agent_message" else "span",
+            "input": {
+                "session_event_type": event_type,
+                "message_type": message_type,
+                "payload": truncate_json(body, max_json_chars),
+            },
+            "output": {
+                "message": truncate_text(message, max_json_chars),
+                "status": "recorded",
+            },
+            "metadata": {
+                "session_event_type": event_type,
+                "message_type": message_type,
+            },
+        }
+
+    if event_type != "response_item":
+        return {
+            "name": f"codex.subagent.{safe_name_component(event_type)}",
+            "type": "span",
+            "input": {"session_event_type": event_type},
+            "output": truncate_json(payload, max_json_chars),
+            "metadata": {"session_event_type": event_type},
+        }
+
+    item_type = payload.get("type") or "unknown"
+    if item_type == "message":
+        role = payload.get("role") or "unknown"
+        text = session_content_text(payload.get("content"), max_json_chars)
+        is_assistant = role == "assistant"
+        return {
+            "name": f"codex.subagent.message.{safe_name_component(role)}",
+            "type": "generation" if is_assistant else "span",
+            "input": {
+                "session_event_type": event_type,
+                "item_type": item_type,
+                "role": role,
+                "phase": payload.get("phase"),
+                **({} if is_assistant else {"content": text}),
+            },
+            "output": (
+                {"text": text}
+                if is_assistant
+                else {"status": "recorded", "role": role}
+            ),
+            "metadata": {
+                "session_event_type": event_type,
+                "item_type": item_type,
+                "role": role,
+                "phase": payload.get("phase"),
+            },
+        }
+
+    if item_type == "function_call":
+        call_id = payload.get("call_id")
+        output_payload, output_event = call_outputs.get(call_id, ({}, {}))
+        return {
+            "name": f"codex.subagent.tool.{safe_name_component(payload.get('name'))}",
+            "type": "span",
+            "input": {
+                "name": payload.get("name"),
+                "call_id": call_id,
+                "arguments": truncate_json(
+                    jsonish(payload.get("arguments")), max_json_chars
+                ),
+            },
+            "output": {
+                "call_id": call_id,
+                "output": truncate_json(
+                    jsonish(output_payload.get("output")), max_json_chars
+                ),
+                "status": "completed" if output_payload else "unknown",
+            },
+            "metadata": {
+                "session_event_type": event_type,
+                "item_type": item_type,
+                "tool_name": payload.get("name"),
+                "call_id": call_id,
+                "output_line": output_event.get("_line_number"),
+            },
+        }
+
+    if item_type == "function_call_output":
+        return {
+            "name": "codex.subagent.tool_output",
+            "type": "span",
+            "input": {"call_id": payload.get("call_id")},
+            "output": truncate_json(jsonish(payload.get("output")), max_json_chars),
+            "metadata": {
+                "session_event_type": event_type,
+                "item_type": item_type,
+                "call_id": payload.get("call_id"),
+            },
+        }
+
+    if item_type == "reasoning":
+        encrypted_content_present = bool(payload.get("encrypted_content"))
+        return {
+            "name": "codex.subagent.reasoning",
+            "type": "span",
+            "input": {
+                "session_event_type": event_type,
+                "item_type": item_type,
+                "encrypted_content_present": encrypted_content_present,
+            },
+            "output": {
+                "summary": truncate_json(payload.get("summary"), max_json_chars),
+                "encrypted_content_present": encrypted_content_present,
+                "status": "recorded",
+            },
+            "metadata": {"session_event_type": event_type, "item_type": item_type},
+        }
+
+    return {
+        "name": f"codex.subagent.response_item.{safe_name_component(item_type)}",
+        "type": "span",
+        "input": {"session_event_type": event_type, "item_type": item_type},
+        "output": truncate_json(payload, max_json_chars),
+        "metadata": {"session_event_type": event_type, "item_type": item_type},
+    }
+
+
+def build_subagent_session_payload(args, session_path, session_events):
+    now = time.time_ns()
+    meta = session_meta(session_events)
+    spawn = subagent_spawn(meta)
+    thread_id = str(meta.get("id") or os.path.basename(session_path))
+    parent_thread_id = (
+        spawn.get("parent_thread_id") or meta.get("parent_thread_id") or ""
+    )
+    agent_nickname = meta.get("agent_nickname") or spawn.get("agent_nickname") or ""
+    agent_role = meta.get("agent_role") or spawn.get("agent_role") or ""
+    trace_id = secrets.token_hex(16)
+    root_observation_id = secrets.token_hex(16)
+    trace_input = session_first_user_message(session_events, args.max_input_chars)
+    if not trace_input:
+        trace_input = json_attr(compact_session_meta(meta, args.max_json_chars))
+    trace_output = session_final_assistant_message(session_events, args.max_output_chars)
+    if not trace_output:
+        trace_output = json_attr({"session_status": "recorded"})
+
+    trace_metadata = {
+        **main_trace_metadata(args),
+        "codex_session_jsonl": True,
+        "subagent_session": True,
+        "main_session_id": args.session_id,
+        "thread_id": thread_id,
+        "parent_thread_id": parent_thread_id,
+        "agent_nickname": agent_nickname,
+        "agent_role": agent_role,
+        "session_file": session_path,
+        "session_event_count": len(session_events),
+    }
+    trace_metadata = {
+        key: value for key, value in trace_metadata.items() if value not in (None, "")
+    }
+    trace_session_id = f"{args.session_id}:subagent:{thread_id}"
+    first_timestamp = (
+        session_event_timestamp(session_events[0], now)
+        if session_events
+        else iso_from_ns(now)
+    )
+    base_ns = ns_from_iso(first_timestamp) or now
+    event_start_nses = [
+        ns_from_iso(event.get("timestamp")) or base_ns + offset * 1_000_000
+        for offset, event in enumerate(session_events, start=2)
+    ]
+    latest_event_start_ns = max(event_start_nses) if event_start_nses else base_ns
+    root_end = iso_from_ns(latest_event_start_ns + 1_000_000)
+    call_outputs = session_call_outputs(session_events)
+    function_call_ids = {
+        (event.get("payload") or {}).get("call_id")
+        for event in session_events
+        if event.get("type") == "response_item"
+        and isinstance(event.get("payload"), dict)
+        and event["payload"].get("type") == "function_call"
+        and event["payload"].get("call_id")
+    }
+
+    batch = [
+        ingestion_event(
+            "trace-create",
+            first_timestamp,
+            {
+                "id": trace_id,
+                "timestamp": first_timestamp,
+                "name": args.subagent_trace_name,
+                "input": trace_input,
+                "output": trace_output,
+                "sessionId": trace_session_id,
+                "environment": args.environment,
+                "metadata": trace_metadata,
+                "tags": ["doris-ai-review-subagent", "codex-session-jsonl"],
+            },
+        ),
+        ingestion_event(
+            "span-create",
+            first_timestamp,
+            {
+                "id": root_observation_id,
+                "traceId": trace_id,
+                "name": "codex.subagent.review",
+                "startTime": first_timestamp,
+                "endTime": root_end,
+                "input": {"session_file": session_path, "thread_id": thread_id},
+                "output": {"final_message": trace_output},
+                "environment": args.environment,
+                "metadata": trace_metadata,
+            },
+        ),
+    ]
+
+    observation_count = 1
+    for offset, event in enumerate(session_events, start=2):
+        payload = event.get("payload") if isinstance(event.get("payload"), dict) else {}
+        if (
+            event.get("type") == "response_item"
+            and payload.get("type") == "function_call_output"
+            and payload.get("call_id") in function_call_ids
+        ):
+            continue
+        shape = session_observation_shape(event, call_outputs, args.max_json_chars)
+        fallback_start_ns = base_ns + offset * 1_000_000
+        start_time = session_event_timestamp(event, fallback_start_ns)
+        end_ns = (ns_from_iso(start_time) or fallback_start_ns) + 750_000
+        body = {
+            "id": secrets.token_hex(16),
+            "traceId": trace_id,
+            "parentObservationId": root_observation_id,
+            "name": shape["name"],
+            "startTime": start_time,
+            "endTime": iso_from_ns(end_ns),
+            "input": shape["input"],
+            "output": shape["output"],
+            "environment": args.environment,
+            "metadata": {
+                **trace_metadata,
+                "session_event_type": event.get("type"),
+                "session_line": event.get("_line_number"),
+            },
+        }
+        for key, value in shape.get("metadata", {}).items():
+            if value not in (None, ""):
+                body["metadata"][key] = value
+        event_type = "generation-create" if shape["type"] == "generation" else "span-create"
+        if shape["type"] == "generation":
+            body["model"] = args.model
+        batch.append(ingestion_event(event_type, start_time, body))
+        observation_count += 1
+
+    return {
+        "trace_id": trace_id,
+        "session_id": trace_session_id,
+        "thread_id": thread_id,
+        "path": session_path,
+        "event_count": len(session_events),
+        "observation_count": observation_count,
+        "payload": {"batch": batch},
+    }
+
+
+def build_subagent_session_payloads(args, main_events):
+    expected_thread_ids = receiver_thread_ids(main_events)
+    payloads = []
+    for path in session_jsonl_files(args.subagent_sessions_dir):
+        events = load_jsonl(path)
+        meta = session_meta(events)
+        if not is_subagent_meta(meta, expected_thread_ids):
+            continue
+        payloads.append(build_subagent_session_payload(args, path, events))
+        if len(payloads) >= args.max_subagent_sessions:
+            break
+    return payloads
+
+
 def compact_json_bytes(payload):
     return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
 
@@ -619,7 +1126,7 @@ def chunk_payload(payload, max_payload_bytes):
     return chunks
 
 
-def post_payload_once(endpoint, public_key, secret_key, payload):
+def post_payload_once(endpoint, public_key, secret_key, payload, timeout_seconds):
     auth = base64.b64encode(f"{public_key}:{secret_key}".encode()).decode()
     request = urllib.request.Request(
         endpoint,
@@ -630,7 +1137,7 @@ def post_payload_once(endpoint, public_key, secret_key, payload):
         },
         method="POST",
     )
-    with urllib.request.urlopen(request, timeout=30) as response:
+    with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
         body = response.read().decode()
         detail = json.loads(body) if body else {}
         errors = detail.get("errors") if isinstance(detail, dict) else None
@@ -659,20 +1166,40 @@ def retry_payload_chunks_after_413(payload, request_size, max_payload_bytes):
     return chunk_payload(payload, next_limit)
 
 
-def post_payload(endpoint, public_key, secret_key, payload, max_payload_bytes):
+def retry_payload_chunks_after_transport_error(payload, request_size, max_payload_bytes):
+    batch = payload.get("batch") or []
+    if len(batch) <= 1:
+        return [(payload, request_size)]
+    next_limit = max(1_000, min(max_payload_bytes - 1, request_size // 2))
+    return chunk_payload(payload, next_limit)
+
+
+def post_payload(
+    endpoint,
+    public_key,
+    secret_key,
+    payload,
+    max_payload_bytes,
+    timeout_seconds,
+    retry_attempts,
+    retry_sleep_seconds,
+):
     statuses = []
     success_count = 0
     request_sizes = []
-    retry_count = 0
+    payload_too_large_retry_count = 0
+    transport_retry_count = 0
     chunks = chunk_payload(payload, max_payload_bytes)
     while chunks:
         chunk, request_size = chunks.pop(0)
         try:
-            status = post_payload_once(endpoint, public_key, secret_key, chunk)
+            status = post_payload_once(
+                endpoint, public_key, secret_key, chunk, timeout_seconds
+            )
         except urllib.error.HTTPError as exc:
             if exc.code != 413:
                 raise
-            retry_count += 1
+            payload_too_large_retry_count += 1
             chunks = (
                 retry_payload_chunks_after_413(
                     chunk, request_size, max_payload_bytes
@@ -680,6 +1207,21 @@ def post_payload(endpoint, public_key, secret_key, payload, max_payload_bytes):
                 + chunks
             )
             continue
+        except (TimeoutError, urllib.error.URLError) as exc:
+            transport_retry_count += 1
+            if transport_retry_count > retry_attempts:
+                raise RuntimeError(
+                    "Litefuse ingestion failed after transport retries: "
+                    f"{type(exc).__name__}: {exc}"
+                ) from exc
+            time.sleep(retry_sleep_seconds)
+            chunks = (
+                retry_payload_chunks_after_transport_error(
+                    chunk, request_size, max_payload_bytes
+                )
+                + chunks
+            )
+            continue
         statuses.append(status["status"])
         success_count += int(status.get("success_count") or 0)
         request_sizes.append(request_size)
@@ -688,7 +1230,8 @@ def post_payload(endpoint, public_key, secret_key, payload, max_payload_bytes):
         "request_count": len(statuses),
         "request_sizes": request_sizes,
         "max_request_size": max(request_sizes) if request_sizes else 0,
-        "payload_too_large_retries": retry_count,
+        "payload_too_large_retries": payload_too_large_retry_count,
+        "transport_retries": transport_retry_count,
         "success_count": success_count,
     }
 
@@ -961,6 +1504,8 @@ def parse_args():
     parser.add_argument("--events-file", required=True)
     parser.add_argument("--output-file", default="")
     parser.add_argument("--trace-name", default="doris-ai-review")
+    parser.add_argument("--subagent-trace-name", default="doris-ai-review-subagent")
+    parser.add_argument("--subagent-sessions-dir", default="")
     parser.add_argument("--session-id", required=True)
     parser.add_argument("--repository", default="")
     parser.add_argument("--workflow", default="")
@@ -976,6 +1521,10 @@ def parse_args():
     parser.add_argument("--max-json-chars", type=int, default=40_000)
     parser.add_argument("--max-context-json-chars", type=int, default=0)
     parser.add_argument("--max-payload-bytes", type=int, default=4_000_000)
+    parser.add_argument("--max-subagent-sessions", type=int, default=100)
+    parser.add_argument("--post-timeout-seconds", type=int, default=120)
+    parser.add_argument("--post-retry-attempts", type=int, default=5)
+    parser.add_argument("--post-retry-sleep-seconds", type=int, default=5)
     parser.add_argument("--verify", action="store_true")
     parser.add_argument("--dry-run", action="store_true")
     parser.add_argument("--verify-attempts", type=int, default=24)
@@ -1001,6 +1550,11 @@ def main():
     trace_id, payload, observation_count = build_ingestion_payload(
         args, input_text, output_text, events
     )
+    subagent_payloads = (
+        build_subagent_session_payloads(args, events)
+        if args.subagent_sessions_dir
+        else []
+    )
 
     result = {
         "dry_run": args.dry_run,
@@ -1009,6 +1563,7 @@ def main():
         "trace_id": trace_id,
         "session_id": args.session_id,
         "trace_name": args.trace_name,
+        "subagent_trace_count": len(subagent_payloads),
         "model": args.model,
         "reasoning_effort": args.reasoning_effort,
     }
@@ -1021,14 +1576,62 @@ def main():
         result["max_request_size"] = (
             max(result["request_sizes"]) if result["request_sizes"] else 0
         )
+        result["subagent_traces"] = []
+        for subagent_payload in subagent_payloads:
+            chunks = chunk_payload(subagent_payload["payload"], args.max_payload_bytes)
+            request_sizes = [request_size for _chunk, request_size in chunks]
+            result["subagent_traces"].append(
+                {
+                    "trace_id": subagent_payload["trace_id"],
+                    "session_id": subagent_payload["session_id"],
+                    "thread_id": subagent_payload["thread_id"],
+                    "path": subagent_payload["path"],
+                    "event_count": subagent_payload["event_count"],
+                    "observation_count": subagent_payload["observation_count"],
+                    "batch_count": len(subagent_payload["payload"]["batch"]),
+                    "request_count": len(chunks),
+                    "request_sizes": request_sizes,
+                    "max_request_size": max(request_sizes) if request_sizes else 0,
+                }
+            )
         print(json.dumps(result, sort_keys=True))
         return
 
     public_key = os.environ["LANGFUSE_PUBLIC_KEY"]
     secret_key = os.environ["LANGFUSE_SECRET_KEY"]
     result["status"] = post_payload(
-        endpoint, public_key, secret_key, payload, args.max_payload_bytes
+        endpoint,
+        public_key,
+        secret_key,
+        payload,
+        args.max_payload_bytes,
+        args.post_timeout_seconds,
+        args.post_retry_attempts,
+        args.post_retry_sleep_seconds,
     )
+    result["subagent_traces"] = []
+    for subagent_payload in subagent_payloads:
+        status = post_payload(
+            endpoint,
+            public_key,
+            secret_key,
+            subagent_payload["payload"],
+            args.max_payload_bytes,
+            args.post_timeout_seconds,
+            args.post_retry_attempts,
+            args.post_retry_sleep_seconds,
+        )
+        result["subagent_traces"].append(
+            {
+                "trace_id": subagent_payload["trace_id"],
+                "session_id": subagent_payload["session_id"],
+                "thread_id": subagent_payload["thread_id"],
+                "path": subagent_payload["path"],
+                "event_count": subagent_payload["event_count"],
+                "observation_count": subagent_payload["observation_count"],
+                "status": status,
+            }
+        )
 
     if args.verify:
         try:

.github/workflows/code-review-runner.yml

@@ -250,14 +250,59 @@ jobs:
             printf 'No additional user-provided review focus.\n' > "$REVIEW_CONTEXT_DIR/review_focus.txt"
           fi
 
-      - name: Prepare required AGENTS guides
+      - name: Prepare authoritative PR context and required AGENTS guides
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           REPO: ${{ github.repository }}
           PR_NUMBER: ${{ inputs.pr_number }}
+          HEAD_SHA: ${{ inputs.head_sha }}
+          BASE_SHA: ${{ inputs.base_sha }}
           HELPER_REF: ${{ github.workflow_sha || github.sha }}
         run: |
-          gh api --paginate "repos/${REPO}/pulls/${PR_NUMBER}/files" --jq '.[].filename' > "$REVIEW_CONTEXT_DIR/pr_changed_files.txt"
+          checkout_head_sha="$(git rev-parse HEAD)"
+          if [ "$checkout_head_sha" != "$HEAD_SHA" ]; then
+            echo "Checked-out HEAD $checkout_head_sha does not match expected PR head $HEAD_SHA"
+            exit 1
+          fi
+
+          retry_to_file() {
+            local output_file="$1"
+            shift
+            local attempt delay=2 tmp_file
+            tmp_file="$(mktemp "${output_file}.tmp.XXXXXX")"
+
+            for attempt in 1 2 3 4; do
+              if "$@" > "$tmp_file"; then
+                mv "$tmp_file" "$output_file"
+                return 0
+              fi
+
+              if [ "$attempt" -lt 4 ]; then
+                echo "Attempt ${attempt}/4 failed while preparing ${output_file}; retrying in ${delay}s..."
+                sleep "$delay"
+                delay=$((delay * 2))
+              fi
+            done
+
+            rm -f "$tmp_file"
+            return 1
+          }
+
+          retry_to_file "$REVIEW_CONTEXT_DIR/pr_changed_files.txt" \
+            gh api --paginate "repos/${REPO}/pulls/${PR_NUMBER}/files?per_page=100" \
+            --jq '.[].filename'
+          retry_to_file "$REVIEW_CONTEXT_DIR/pr.diff" \
+            gh pr diff "$PR_NUMBER" --repo "$REPO" --color=never
+
+          live_pr="$(gh api "repos/${REPO}/pulls/${PR_NUMBER}")"
+          live_head_sha="$(jq -r '.head.sha' <<<"$live_pr")"
+          live_base_sha="$(jq -r '.base.sha' <<<"$live_pr")"
+          if [ "$live_head_sha" != "$HEAD_SHA" ] || [ "$live_base_sha" != "$BASE_SHA" ]; then
+            echo "PR changed while review context was being prepared; restart the review"
+            echo "Expected base/head: $BASE_SHA $HEAD_SHA"
+            echo "Current base/head:  $live_base_sha $live_head_sha"
+            exit 1
+          fi
 
           helper="$RUNNER_TEMP/prepare_review_agents.py"
           gh api \
@@ -277,13 +322,13 @@ jobs:
       - name: Prepare review prompt
         run: |
           cat > "$REVIEW_CONTEXT_DIR/review_prompt.txt" <<'PROMPT'
-          This review task is executed in Codex goal mode. The review goal is complete only after every changed file and relevant surrounding code path has been examined, every suspicious point has a clear conclusion, and there is no remaining possibility of producing any new valuable review comment.
-
           You are performing an automated code review inside a GitHub Actions runner. The gh CLI is available and authenticated via GH_TOKEN.
           The current directory is the code repository for the PR to be reviewed.
+          This environment is only used for review operations, so do not attempt any builds or code modifications. However, temporary scripts you use for review purposes are exceptions.
           You MUST NOT attempt to access any files outside the current directory. and you DO NOT need to. But this does not prevent you from normally using any skill or web fetch tools.
           You can comment on the pull request.
-          Proceed with all subsequent research at the HIGHEST level of thought, aiming to identify all issues and submit all comments in JSON format.
+          This review task is executed in Codex goal mode. Subsequent prompts and skill indicate all the review and commenting actions you need to perform. The current round can only be concluded after completing the main agent risk scan, the normal full-review subagent review for each round, and the risk-focused subagent scan (if applicable). The final GitHub review can only be submitted after all subagents return NO_NEW_VALUABLE_FINDINGS, and all candidate points from the subagents have been validated, deduplicated, accepted, or rejected by the main agent.
+          In addition, the number of iterations shall not exceed 3 rounds. If new subagents identify valuable candidate points after 3 rounds, the main agent must indicate in the final summary that the review is incomplete.
 
           Context:
           - Repository: PLACEHOLDER_REPO
@@ -294,38 +339,47 @@ jobs:
           - Raw inline review comments JSON: PLACEHOLDER_CONTEXT_DIR/pr_review_comments.json
           - User review focus: PLACEHOLDER_CONTEXT_DIR/review_focus.txt
           - PR changed files: PLACEHOLDER_CONTEXT_DIR/pr_changed_files.txt
+          - Authoritative aggregate PR diff: PLACEHOLDER_CONTEXT_DIR/pr.diff
           - Required AGENTS.md files: PLACEHOLDER_CONTEXT_DIR/required_agents.txt
           - Shared subagent review ledger: PLACEHOLDER_CONTEXT_DIR/subagent_review_findings.md
 
+          PR diff and path-reading (for all subagents and the main agent):
+          - PLACEHOLDER_CONTEXT_DIR/pr.diff is the authoritative aggregate diff for the whole PR, and PLACEHOLDER_CONTEXT_DIR/pr_changed_files.txt is the authoritative changed-path list. Do not obtain the PR diff or changed-path list through other means.
+          - Before reading any file whose exact path is not already confirmed by PLACEHOLDER_CONTEXT_DIR/pr_changed_files.txt, PLACEHOLDER_CONTEXT_DIR/pr.diff, or a previous successful command output, you MUST first run `rg --files` to confirm the actual path of the target file.
+
           Before reviewing any code, you MUST read and follow the code review skill in this repository. During review, you must strictly follow those instructions.
           The active review goal's progress tracking MUST include, and must stay current throughout the review:
-          - Read the review prompt, code-review skill, required AGENTS.md files, existing review threads, user focus, changed-file list, and the shared subagent review ledger.
-          - Spawn the required focused subagents and assign each subagent a dedicated section in the shared ledger before it reviews.
-          - Require every subagent to append only to its assigned ledger section, while reading the whole ledger to avoid duplicate candidates.
-          - Read the shared ledger after every subagent result, then independently verify, deduplicate, accept, or dismiss every candidate in the main merged section.
-          - Run at least one final convergence round where every live subagent reviews the current ledger and proposed final comment set.
-          - Continue reviewing if any subagent reports a new valuable candidate during convergence; repeat verification and convergence until every subagent says `NO_NEW_VALUABLE_FINDINGS` for the same current ledger/comment set.
-          - Submit the final GitHub review and verify that all accepted comments landed before marking the goal complete.
-          The active review goal MUST remain incomplete until every suspicious point found during review has a clear conclusion: submitted as an inline issue, dismissed as already covered by existing review context, or dismissed with concrete code evidence explaining why it is not a bug.
-          You MUST use the available subagent or multi-agent spawn tool for this review; do not merely simulate subagent output in your own text. If tool discovery is needed, search for the subagent/multi-agent tool first. Spawn at least two subagents before final submission:
-          - one subagent assigned to the `optimizer-rewrite` ledger section and focused on optimizer/rewrite correctness, semantic equivalence, and parallel join/aggregate paths.
-          - one subagent assigned to the `tests-session-config` ledger section and focused on regression tests, expected outputs, session/config propagation, compatibility, and basic CI/style failures.
-          The shared subagent review ledger is PLACEHOLDER_CONTEXT_DIR/subagent_review_findings.md. Before spawning any subagent, the main agent MUST read this ledger. Every subagent prompt MUST include this ledger path and the exact ledger section assigned to that subagent. Subagents must read the whole ledger before reviewing, avoid duplicating existing candidates, and append their findings only under their assigned subagent section. They must not rewrite the whole ledger, edit another subagent section, edit the main merged sections, edit repository source files, or submit GitHub comments.
-          Each subagent must record candidate findings in its own ledger section with stable IDs, path/line, claim, evidence, duplicate relationship if any, and recommendation. If a candidate overlaps an existing ledger item, the subagent should add a duplicate note in its own section that references the existing candidate ID instead of modifying the existing candidate. This section-owned append-only rule is mandatory to avoid concurrent patch conflicts.
-          The main agent must read the shared ledger after each subagent result, merge duplicate candidates into the main merged section, update candidate statuses, and keep a proposed final comment set in the main-owned ledger sections. Statuses must distinguish at least: proposed_by_subagent, accepted_for_inline_comment, dismissed_with_evidence, duplicate_of_existing_thread, duplicate_of_candidate, and needs_more_evidence.
-          Before final submission, the main agent MUST send the current ledger and proposed final comment set to every live subagent and require a convergence response. A subagent may either append a new valuable candidate under its assigned section or reply exactly `NO_NEW_VALUABLE_FINDINGS` for the current ledger/comment set. The main agent MUST NOT complete the goal, final summary, or GitHub review until every live subagent has returned `NO_NEW_VALUABLE_FINDINGS` in the same convergence round after the last accepted or dismissed candidate update.
-          Subagents must only report candidate findings with evidence. The main agent must read their results and the shared ledger, independently verify or dismiss every candidate with concrete code evidence, deduplicate against existing review threads, and submit the final GitHub review itself. The final summary must include a short "Subagent conclusions" section describing which subagent candidates became inline comments, which were dismissed, which duplicates were merged, and which convergence round ended with all subagents reporting `NO_NEW_VALUABLE_FINDINGS`.
-          If and only if the runtime exposes no subagent or multi-agent tool after explicit discovery, state that limitation in the final summary and continue the review manually.
-          You MUST NOT stop after finding the first blocking issue. Keep reviewing changed files, related control flow, tests, and parallel/special-case paths until all plausible correctness, lifecycle, configuration, compatibility, performance, and coverage bugs have been investigated and every bug you can substantiate has been reported.
-          Before submitting the final review, do one explicit final sweep over the changed-file list and your unresolved candidate list. Only finish when there are no unresolved suspicious points and all possible substantiated bugs have been pointed out in the GitHub review.
+          1. Read the review prompt, code-review skill, required AGENTS.md files, existing review threads, user focus, changed-file list, and the shared subagent review ledger.
+          2. Perform a main-agent initial risk scan before spawning review subagents. You must thoroughly read all the changes in this PR and understand all the involved mechanisms, pointing out: if there is a problem with this PR, where is the problem most likely to be? Or are there any points you suspect to be risky? write the result under `## Main Initial Risk Scan` in the shared ledger.
+          3. Based on the requirements of the code-review skill regarding the coverage points of the review spawn 1-3 subagents (depends on the complexity of the PR), each focusing on certain aspects. The sum of their focus points must fully cover the entire content of PR, as well as the points required in skill and any additional review points you consider necessary. They must complete the review of their own aspects and identify all possible issues before they can finish their work. You need assign each subagent a dedicated section in the shared ledger.
+          4. If the main initial risk scan identifies one or more suspicious mechanisms, spawn additional risk-focused subagent(s), separate from the normal complete-review subagents, to specially investigate those specific mechanisms and their upstream/downstream interactions.
+          5. Read the shared ledger after every subagent result, then independently verify, deduplicate, accept, or dismiss every candidate in the main merged section. The status of each candidate must be clarified after this stage.
+          6. If all subagents return `NO_NEW_VALUABLE_FINDINGS`, or if this loop has already executed 3 rounds, conduct the necessary review to ensure that all current candidate points have been verified, deduplicated, accepted, or rejected, and then submit the final GitHub comment. Otherwise, continue with the next round of the same review process from the beginning.
+          7. Submit the final GitHub review and verify that all accepted comments landed before marking the goal complete. You must address all checkpoints required by the skill, user concerns (if any), and indicate the status of the review completion.
+
+          Subagent Constraints:
+          - Every subagent appends only to its assigned ledger section, while reading the whole ledger to avoid duplicate candidates.
+          - Follow the principles indicated in the code-review skill.
+          - The shared subagent review ledger is PLACEHOLDER_CONTEXT_DIR/subagent_review_findings.md. Subagents must read the whole ledger before reviewing, avoid duplicating existing candidates, and append their findings only under their assigned subagent section. They must not rewrite the whole ledger, edit another subagent section, edit the main merged sections, edit repository source files, or submit GitHub comments.
+          - Each subagent must record candidate findings in its own ledger section with stable IDs, path/line, claim, evidence, duplicate relationship if any, and recommendation. Do not add duplicates of existing items. This section-owned append-only rule is mandatory to avoid concurrent patch conflicts.
+          - Before returning, a further thought must be made—what problem, if any, might I have missed just now, and where? Then a thorough recheck should be conducted to confirm whether this potential issue is real and whether it needs to be reported.
+
+          Main-agent Constraints:
+          - The active review goal MUST remain incomplete until every suspicious point found during review has a clear conclusion: submitted as an inline issue, dismissed as already covered by existing review context, or dismissed with concrete code evidence explaining why it is not a bug.
+          - The main initial risk scan MUST be written under `## Main Initial Risk Scan` in the shared ledger before any subagent is spawned. For every risk focus item, include: ID, changed files/lines involved, related mechanisms to inspect, why it is suspicious, required upstream/downstream files or paths, and the exact question the risk-focused subagent must answer.
+          - The shared subagent review ledger is PLACEHOLDER_CONTEXT_DIR/subagent_review_findings.md. Before spawning any subagent, the main agent MUST read this ledger. Every subagent prompt MUST include this ledger path and the exact ledger section assigned to that subagent.
+          - The main agent must read the shared ledger after each subagent result, merge duplicate candidates into the main merged section, update candidate statuses, and keep a proposed final comment set in the main-owned ledger sections. It must also update the status of every main risk focus item.
+          - Before submitting the final review, do one explicit final sweep over the changed-file list and your unresolved candidate list. Only finish when there are no unresolved suspicious points and all possible substantiated bugs have been pointed out in the GitHub review. At this stage, you can conduct final research on any part you still have doubts about or which may not have been fully investigated, ensuring that all potential issues have been thoroughly investigated and reported.
+
+          Any agent MUST NOT stop after finding the first blocking issue. Keep reviewing changed files, related control flow, tests, and parallel/special-case paths until all plausible correctness, lifecycle, configuration, compatibility, performance, and coverage bugs have been investigated and every bug you can substantiate has been reported.
+
           Before inspecting the PR diff or related code, you MUST read the contents of every AGENTS.md file listed below. These paths are computed from the PR changed file ancestors in this checkout. Searching for or listing paths is not sufficient; read each listed file directly.
           Required AGENTS.md files for this PR:
           PLACEHOLDER_REQUIRED_AGENTS_BLOCK
+
           Before proposing any new issue, you MUST read PLACEHOLDER_CONTEXT_DIR/pr_review_threads.md and treat every existing inline comment thread and reply as already-known review context.
-          Do NOT submit the same or substantially similar issue again if it has already been raised in the existing review threads, even if you would phrase it differently.
-          Only raise a similar concern when the PR introduces a genuinely different instance in another location that is not already covered by the existing thread, and explain why it is distinct.
-          You MUST also read PLACEHOLDER_CONTEXT_DIR/review_focus.txt. Perform a complete review of the whole PR as usual, and additionally pay special attention to the user-provided focus points from that file.
-          In the final summary, include a short response to the user focus points, including when no additional issue was found for them.
+          Do NOT submit the same or substantially similar issue again if it has already been raised in the existing review threads, even if you would phrase it differently. Only raise a similar concern when the PR introduces a genuinely different instance in another location that is not already covered by the existing thread, and explain why it is distinct.
+          You MUST also read PLACEHOLDER_CONTEXT_DIR/review_focus.txt. Perform a complete review of the whole PR as usual, and additionally pay special attention to the user-provided focus points from that file. In the final summary, include a short response to the user focus points, including when no additional issue was found for them.
           In addition, you can perform any desired review operations to observe suspicious code and details in order to identify issues as much as possible.
 
           ## Final response format
@@ -339,13 +393,14 @@ jobs:
               - Build a JSON array of comments like: [{ "path": "<file>", "position": <diff_position>, "body": "..." }]
               - Submit via: gh api repos/PLACEHOLDER_REPO/pulls/PLACEHOLDER_PR_NUMBER/reviews --input <json_file>
               - The JSON file should contain: {"event":"REQUEST_CHANGES","body":"<summary>","comments":[...]}
-          - A complete and detailed review of the entire PR must be conducted, identifying all potential issues and submitting corresponding comments. Only then can the review work be considered finished. If this standard is not strictly met, the review of the remaining parts must continue.
           PROMPT
+
           sed -i "s|PLACEHOLDER_REPO|${REPO}|g" "$REVIEW_CONTEXT_DIR/review_prompt.txt"
           sed -i "s|PLACEHOLDER_PR_NUMBER|${PR_NUMBER}|g" "$REVIEW_CONTEXT_DIR/review_prompt.txt"
           sed -i "s|PLACEHOLDER_HEAD_SHA|${HEAD_SHA}|g" "$REVIEW_CONTEXT_DIR/review_prompt.txt"
           sed -i "s|PLACEHOLDER_BASE_SHA|${BASE_SHA}|g" "$REVIEW_CONTEXT_DIR/review_prompt.txt"
           sed -i "s|PLACEHOLDER_CONTEXT_DIR|${REVIEW_CONTEXT_REL}|g" "$REVIEW_CONTEXT_DIR/review_prompt.txt"
+
           python3 - "$REVIEW_CONTEXT_DIR/review_prompt.txt" "$REVIEW_CONTEXT_DIR/required_agents_prompt.txt" <<'PY'
           import sys
           from pathlib import Path
@@ -368,15 +423,25 @@ jobs:
           - Subagents must not rewrite this file, edit another subagent section, edit main-owned sections, edit repository source files, or submit GitHub comments.
           - Avoid duplicates. If a candidate overlaps an existing candidate, add a duplicate note in your own section that references the existing candidate ID.
           - The main agent owns final status, final deduplication, GitHub review submission, and GitHub API verification.
-          - The goal is not complete until all live subagents have reviewed the current ledger and proposed final comment set in the same convergence round and replied `NO_NEW_VALUABLE_FINDINGS`.
+
+          ## Main Initial Risk Scan
+
+          Main-owned format:
+
+          - ID:
+            Status:
+            Changed files/lines:
+            Related mechanisms to inspect:
+            Why suspicious:
+            Required upstream/downstream files or paths:
+            Question for risk-focused subagent:
+            Final conclusion:
 
           Candidate statuses:
           - proposed_by_subagent
           - accepted_for_inline_comment
           - dismissed_with_evidence
-          - duplicate_of_existing_thread
-          - duplicate_of_candidate
-          - needs_more_evidence
+          - duplicated
 
           Candidate format:
 
@@ -390,18 +455,6 @@ jobs:
             Duplicate relationship:
             Recommendation:
 
-          ## Subagent Candidate Sections
-
-          ### optimizer-rewrite
-
-          Subagent assignment: optimizer/rewrite correctness, semantic equivalence, and parallel join/aggregate paths.
-          Append only below this line and above the next subagent heading.
-
-          ### tests-session-config
-
-          Subagent assignment: regression tests, expected outputs, session/config propagation, compatibility, and basic CI/style failures.
-          Append only below this line and above the main-owned sections.
-
           ## Main Merged Findings
 
           Main-owned format:
@@ -428,9 +481,7 @@ jobs:
           cat > "$REVIEW_CONTEXT_DIR/codex_goal_prompt.txt" <<EOF
           You are performing an automated code review inside a GitHub Actions runner.
           This invocation is already running in Codex goal mode. Before inspecting the PR diff or related code, read ${REVIEW_CONTEXT_REL}/review_prompt.txt verbatim and follow that file as the complete review instruction set.
-          This review must use the available subagent or multi-agent spawn tool for focused review passes. All subagents and the main agent must maintain ${REVIEW_CONTEXT_REL}/subagent_review_findings.md as the shared source of truth. Each subagent may append only to its assigned ledger section; the main agent owns merged statuses, duplicate suppression, proposed final comments, GitHub review submission, and GitHub API verification.
-          The review goal's completion process must include shared-ledger review, subagent spawning with assigned ledger sections, candidate verification, duplicate merge, a final subagent convergence round, GitHub review submission, and GitHub API verification. Do not complete the goal unless every live subagent has replied NO_NEW_VALUABLE_FINDINGS for the same current ledger/comment set after the last candidate update.
-          Keep reviewing until every suspicious point has a clear conclusion, every substantiated issue has been submitted as a GitHub inline review comment, and no new valuable review comment remains possible.
+          This document outlines the complete process, cyclic procedures, and exit agreements you need to follow. Please read it seriously and reconfirm the requirements at the end of each cycle. Complete the review according to the stipulations in this document.
           EOF
         env:
           REPO: ${{ github.repository }}
@@ -444,7 +495,6 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          REASONING_EFFORT: xhigh
           REPO: ${{ github.repository }}
           PR_NUMBER: ${{ inputs.pr_number }}
           HEAD_SHA: ${{ inputs.head_sha }}
@@ -458,7 +508,7 @@ jobs:
           codex exec --goal "$GOAL_PROMPT" \
             --cd "$GITHUB_WORKSPACE" \
             --model "gpt-5.5" \
-            --config "model_reasoning_effort=\"${REASONING_EFFORT}\"" \
+            --config "model_reasoning_effort=xhigh" \
             --sandbox danger-full-access \
             --color never \
             --json \
@@ -531,7 +581,6 @@ jobs:
           HEAD_SHA: ${{ inputs.head_sha }}
           BASE_SHA: ${{ inputs.base_sha }}
           HELPER_REF: ${{ github.workflow_sha || github.sha }}
-          REASONING_EFFORT: xhigh
         run: |
           if [ ! -s "$REVIEW_CONTEXT_DIR/codex_goal_prompt.txt" ] || [ ! -s "$REVIEW_CONTEXT_DIR/codex-events.jsonl" ]; then
             echo "Goal prompt or Codex JSONL event stream is missing; skipping Litefuse I/O recording."
@@ -550,6 +599,8 @@ jobs:
             --events-file "$REVIEW_CONTEXT_DIR/codex-events.jsonl" \
             --output-file "$REVIEW_CONTEXT_DIR/codex-final-message.txt" \
             --trace-name "doris-ai-review" \
+            --subagent-trace-name "doris-ai-review-subagent" \
+            --subagent-sessions-dir "$CODEX_HOME/sessions" \
             --session-id "$GITHUB_RUN_ID" \
             --repository "$REPO" \
             --workflow "$GITHUB_WORKFLOW" \
@@ -558,7 +609,7 @@ jobs:
             --head-sha "$HEAD_SHA" \
             --base-sha "$BASE_SHA" \
             --model "gpt-5.5" \
-            --reasoning-effort "$REASONING_EFFORT" \
+            --reasoning-effort "xhigh" \
             --environment "github-actions" \
             --max-payload-bytes 4000000 \
             --min-observations 4 \

be/src/core/AGENTS.md

@@ -12,12 +12,30 @@ Types in `be/src/core/custom_allocator.h` exist to route owned memory through Do
 
 Vectorized columns (`IColumn`) use intrusive-reference-counted copy-on-write.
 
+### Standard Use Pattern
+
+1. For newly created columns locally: Prefer to directly retain the original `MutableColumnPtr` if possible. Use `assume_mutable()` if necessary.
+2. For shared or unknown ownership `ColumnPtr`: Use `IColumn::mutate(...)` and write back the owner.
+3. When modifying the entire `Block` or `Column`: Prefer using `mutate_columns_scoped()` or `mutate_column_scoped(pos)`.
+4. When applying an algorithm to a `Block` using `MutableBlock`: Understand the potential for detachment, then use `ScopedMutableBlock` or `VectorizedUtils::build_scoped_mutable_mem_reuse_block(...)`.
+5. Row-level loops: Do not mutate the block or columns row by row. Obtain the mutable owners all at once.
+6. Modifying columns of complex types: First mutate the parent owner. Do not force clone to satisfy mutable access. Only detach when it is definitely necessary to modify.
+
+Avoid anti-patterns:
+
+1. Do not use `IColumn::mutate` to modify and write back a column in a `Block`; instead, use `mutate_column_scoped`.
+2. Do not call `Block::get_columns()` directly on a `Block` before mutation.
+3. Do not use `assert_mutable()` if ownership is not clearly defined; instead, use `mutate` and understand the possibility of detach.
+4. Do not `mutate()` row by row.
+5. Do not call `ScopedMutableColumns::release()` unless it is a special scenario requiring temporary transfer of ownership.
+
 ### Checkpoints
 
 - [ ] Exclusive ownership guaranteed before `mutate()` on hot paths? Shared ownership triggers deep copy
 - [ ] `assert_mutable()` used only when exclusive ownership is already guaranteed?
 - [ ] If you need to modify the data within a `Block`, have you correctly used `ScopedMutableBlock`?
 - [ ] `convert_to_full_column_if_const()` materializes only `ColumnConst`; ordinary columns may return shared storage?
+- [ ] If a `Block` is still going to be used later, and we temporarily need to use its column externally，should we prefer to copy rather than `std::move` its `ColumnPtr`? If moving is necessary, are they all put back before all exits?
 
 ## Type System and Serialization
 

be/src/format/table/jdbc_jni_reader.cpp

@@ -190,7 +190,7 @@ Status JdbcJniReader::_cast_string_to_special_type(const SlotDescriptor* slot_de
 
     ColumnsWithTypeAndName argument_template;
     argument_template.reserve(2);
-    argument_template.emplace_back(std::move(input_col), input_string_type, "java.sql.String");
+    argument_template.emplace_back(input_col, input_string_type, "java.sql.String");
     argument_template.emplace_back(cast_param, cast_param_data_type, target_data_type_name);
 
     FunctionBasePtr func_cast = SimpleFunctionFactory::instance().get_function(