fix(kimi): capture replies from kimi-code native wire logs; gate pane fallback

lib/provider_backends/kimi/execution.py

@@ -263,9 +263,11 @@ def _poll_submission(submission: ProviderSubmission, *, now: str) -> ProviderPol
     pane_observation = _observe_kimi_pane_turn(backend, pane_id, req_id)
     if pane_observation is not None:
         pane_observation = _stabilize_pane_observation(state, pane_observation, now)
-    if pane_observation is not None and (
-        observation is None or (pane_observation.completed and not observation.completed)
-    ):
+    if pane_observation is not None and observation is None:
+        # Pane scraping is a rescue path only: once the native wire log has
+        # recorded the req_id anchor, wait for the native turn to complete
+        # instead of letting a prematurely-stable pane snapshot (input box
+        # visible during answer generation) truncate the reply.
         observation = pane_observation
         state["pane_fallback_observed"] = True
     if observation is None:

lib/provider_backends/kimi/native_log.py

@@ -32,37 +32,56 @@ def observe_kimi_turn(
 ) -> KimiTurnObservation | None:
     if not req_id:
         return None
-    observations: list[KimiTurnObservation] = []
-    for wire_path in _wire_paths(work_dir, home_candidates=home_candidates):
-        observed = _observe_wire_file(wire_path, req_id=req_id)
-        if observed is not None:
-            observations.append(observed)
-    if not observations:
-        return None
-    completed = [item for item in observations if item.completed]
-    if completed:
-        return max(completed, key=_observation_sort_key)
-    return max(observations, key=_observation_sort_key)
+    wire_paths = _wire_paths(work_dir, home_candidates=home_candidates)
+    # Multiple kimi agents can share one work_dir, so one sessions root holds
+    # many agents' wire logs and a req_id can be *mentioned* in other agents'
+    # prompts. Prefer turns whose prompt carries the `CCB_REQ_ID: <id>`
+    # header (strict); fall back to plain substring matching (loose) so
+    # legacy prompt formats without the header still resolve.
+    for strict in (True, False):
+        observations: list[KimiTurnObservation] = []
+        for wire_path in wire_paths:
+            observed = _observe_wire_file(wire_path, req_id=req_id, strict=strict)
+            if observed is not None:
+                observations.append(observed)
+        if observations:
+            completed = [item for item in observations if item.completed]
+            if completed:
+                return max(completed, key=_observation_sort_key)
+            return max(observations, key=_observation_sort_key)
+    return None
 
 
 def kimi_project_hash(work_dir: Path) -> str:
     normalized = str(Path(work_dir).expanduser().resolve(strict=False))
     return hashlib.md5(normalized.encode("utf-8", "surrogateescape")).hexdigest()
 
 
+def kimi_code_project_dirname(work_dir: Path) -> str:
+    normalized = str(Path(work_dir).expanduser().resolve(strict=False))
+    digest = hashlib.sha256(normalized.encode("utf-8", "surrogateescape")).hexdigest()[:12]
+    basename = Path(normalized).name[:40]
+    return f"wd_{basename}_{digest}"
+
+
 def kimi_sessions_root(work_dir: Path, *, home: Path | None = None) -> Path:
     base = _kimi_home(home)
     return base / "sessions" / kimi_project_hash(work_dir)
 
 
+def kimi_code_sessions_root(work_dir: Path, *, home: Path | None = None) -> Path:
+    base = _kimi_code_home(home)
+    return base / "sessions" / kimi_code_project_dirname(work_dir)
+
+
 def _wire_paths(work_dir: Path, *, home_candidates: Iterable[Path] | None) -> list[Path]:
     paths: list[Path] = []
     seen: set[Path] = set()
-    for home in _candidate_homes(home_candidates):
-        root = kimi_sessions_root(work_dir, home=home)
-        if not root.is_dir():
-            continue
-        for path in root.glob("*/wire.jsonl"):
+
+    def _add(pattern_root: Path, pattern: str) -> None:
+        if not pattern_root.is_dir():
+            return
+        for path in pattern_root.glob(pattern):
             try:
                 resolved = path.resolve(strict=False)
             except Exception:
@@ -71,6 +90,12 @@ def _wire_paths(work_dir: Path, *, home_candidates: Iterable[Path] | None) -> li
                 continue
             seen.add(resolved)
             paths.append(path)
+
+    for home in _candidate_homes(home_candidates):
+        # Legacy kimi layout: ~/.kimi/sessions/<md5(work_dir)>/<session>/wire.jsonl
+        _add(kimi_sessions_root(work_dir, home=home), "*/wire.jsonl")
+        # kimi-code layout: ~/.kimi-code/sessions/wd_<base>_<sha256[:12]>/<session>/agents/<agent>/wire.jsonl
+        _add(kimi_code_sessions_root(work_dir, home=home), "*/agents/*/wire.jsonl")
     return sorted(paths, key=_path_mtime)
 
 
@@ -102,14 +127,42 @@ def _kimi_home(home: Path | None) -> Path:
     return home / ".kimi"
 
 
-def _observe_wire_file(path: Path, *, req_id: str) -> KimiTurnObservation | None:
+def _kimi_code_home(home: Path | None) -> Path:
+    if home is None:
+        return current_provider_source_home() / ".kimi-code"
+    if home.name == ".kimi-code":
+        return home
+    return home / ".kimi-code"
+
+
+def _observe_wire_file(path: Path, *, req_id: str, strict: bool = False) -> KimiTurnObservation | None:
     try:
         lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
     except OSError:
         return None
 
     current: dict[str, object] | None = None
     latest: KimiTurnObservation | None = None
+
+    def _finalize_current(index: int, timestamp: object | None) -> None:
+        # A turn that produced reply parts and then gave way to a new turn is
+        # complete even if the wire log has no explicit TurnEnd event
+        # (kimi-code wire logs have none).
+        nonlocal current, latest
+        if current is None:
+            return
+        parts = current.get("parts")
+        if isinstance(parts, list) and parts:
+            latest = _observation_from_state(
+                path,
+                current,
+                req_id=req_id,
+                completed=True,
+                completed_at=timestamp,
+                line_count=index,
+            )
+        current = None
+
     for index, line in enumerate(lines, 1):
         try:
             event = json.loads(line)
@@ -120,7 +173,7 @@ def _observe_wire_file(path: Path, *, req_id: str) -> KimiTurnObservation | None
         event_type, payload, timestamp = _normalize_event(event)
 
         if event_type == "TurnBegin":
-            if _payload_has_req_id(payload, req_id):
+            if _payload_has_req_id(payload, req_id, strict=strict):
                 current = {
                     "parts": [],
                     "started_at": timestamp,
@@ -140,7 +193,8 @@ def _observe_wire_file(path: Path, *, req_id: str) -> KimiTurnObservation | None
             continue
 
         if event_type in {"turn.prompt", "turn.started"}:
-            if _value_has_req_id(payload, req_id):
+            if _value_has_req_id(payload, req_id, strict=strict):
+                _finalize_current(index, timestamp)
                 current = {
                     "parts": [],
                     "started_at": timestamp,
@@ -155,6 +209,8 @@ def _observe_wire_file(path: Path, *, req_id: str) -> KimiTurnObservation | None
                     completed_at=None,
                     line_count=index,
                 )
+            else:
+                _finalize_current(index, timestamp)
             continue
 
         if event_type == "context.append_message":
@@ -163,7 +219,7 @@ def _observe_wire_file(path: Path, *, req_id: str) -> KimiTurnObservation | None
                 continue
             role = str(message.get("role") or "").strip().lower()
             content = _text_from_value(message.get("content"))
-            if role == "user" and req_id in content:
+            if role == "user" and _req_id_in_text(content, req_id, strict=strict):
                 current = {
                     "parts": [],
                     "started_at": timestamp,
@@ -180,7 +236,7 @@ def _observe_wire_file(path: Path, *, req_id: str) -> KimiTurnObservation | None
                 )
                 continue
             if role == "user" and current is not None:
-                current = None
+                _finalize_current(index, timestamp)
                 continue
             if current is None or role != "assistant":
                 continue
@@ -247,6 +303,14 @@ def _observe_wire_file(path: Path, *, req_id: str) -> KimiTurnObservation | None
                         completed_at=None,
                         line_count=index,
                     )
+            elif nested_type == "step.end":
+                # kimi-code wire logs have no TurnEnd; a step that ends for
+                # any reason other than tool_use is the final step of a turn.
+                finish = str(
+                    nested.get("finishReason") or nested.get("finish_reason") or ""
+                ).strip().lower()
+                if finish and finish != "tool_use":
+                    _finalize_current(index, timestamp)
             continue
 
         if event_type == "StatusUpdate":
@@ -295,15 +359,21 @@ def _normalize_event(event: dict[str, object]) -> tuple[str, dict[str, object],
     return event_type, event, event.get("timestamp") or event.get("time")
 
 
-def _payload_has_req_id(payload: dict[str, object], req_id: str) -> bool:
+def _req_id_in_text(text: str, req_id: str, *, strict: bool) -> bool:
+    if not strict:
+        return req_id in text
+    return f"CCB_REQ_ID: {req_id}" in text
+
+
+def _payload_has_req_id(payload: dict[str, object], req_id: str, *, strict: bool = False) -> bool:
     user_input = payload.get("user_input")
     if not isinstance(user_input, list):
         return False
     for part in user_input:
         if not isinstance(part, dict):
             continue
         text = part.get("text")
-        if isinstance(text, str) and req_id in text:
+        if isinstance(text, str) and _req_id_in_text(text, req_id, strict=strict):
             return True
     return False
 
@@ -318,8 +388,8 @@ def _append_part(state: dict[str, object], text: str, *, continuous: bool = Fals
     parts.append(text)
 
 
-def _value_has_req_id(value: object, req_id: str) -> bool:
-    return req_id in _text_from_value(value)
+def _value_has_req_id(value: object, req_id: str, *, strict: bool = False) -> bool:
+    return _req_id_in_text(_text_from_value(value), req_id, strict=strict)
 
 
 def _text_from_value(value: object) -> str:
@@ -354,6 +424,9 @@ def _observation_from_state(
     parts = state.get("parts")
     reply = clean_native_reply("\n".join(str(part) for part in parts), req_id) if isinstance(parts, list) else ""
     session_id = path.parent.name if path.parent.name else None
+    if path.parent.parent.name == "agents" and path.parent.parent.parent.name:
+        # kimi-code layout: <session>/agents/<agent>/wire.jsonl
+        session_id = path.parent.parent.parent.name
     message_id = state.get("message_id")
     provider_turn_ref = str(message_id).strip() if message_id else session_id
     return KimiTurnObservation(
@@ -383,6 +456,8 @@ def _path_mtime(path: Path) -> float:
 
 __all__ = [
     "KimiTurnObservation",
+    "kimi_code_project_dirname",
+    "kimi_code_sessions_root",
     "kimi_project_hash",
     "kimi_sessions_root",
     "observe_kimi_turn",