fix: improve terminal log diagnostics and HTTP timeout handling

swanlab/sdk/internal/core_python/core.py

@@ -341,7 +341,17 @@ def _upsert_logs_when_offline(self, logs: List[LogRecord]) -> None:
     def _upsert_logs_when_online(self, logs: List[LogRecord]) -> None:
         records = [builder.build_log_record(self._counter, self._epoch, c) for c in logs]
         self._store_records(records)
+        if records:
+            console.debug(
+                f"Stored log records locally: count={len(records)}, nums={records[0].num}..{records[-1].num}",
+                write_to_tty=False,
+            )
         self._transport_put(records)
+        if records:
+            console.debug(
+                f"Queued log records for upload: count={len(records)}, nums={records[0].num}..{records[-1].num}",
+                write_to_tty=False,
+            )
 
     # ---- upsert_saves ----
 

swanlab/sdk/internal/core_python/store/__init__.py

@@ -88,14 +88,16 @@ def write(self, data: bytes) -> None:
                 data_used += LEVELDBLOG_DATA_LEN
                 data_left -= LEVELDBLOG_DATA_LEN
             self._write_record(data[data_used:], LEVELDBLOG_LAST)
-        # FIX: 【会导致磁盘高频 IO】每次 write 后统一 fsync，保证落盘
-        try:
-            self._fp.flush()
-            # 指标量在较大的情况下高频 fsync 会影响上传效率
-            # os.fsync(self._fp.fileno())
-        except OSError:
-            pass
-        self._flush_offset = self._index
+        # 不在每次 write 后调用 flush/fsync：高频指标会产生大量磁盘 IO；文件位于 NAS 等网络文件系统时，
+        # 网络重试还可能使写入线程长时间阻塞。正常关闭时由 close() 统一 flush；代价是进程异常退出时可能
+        # 丢失尚在 Python 缓冲区中的尾部数据。若需要增强耐久性，应采用按批次或定时刷新，而非逐条刷新。
+        # 后面用Go重写core时会考虑将本地持久化存储和远程上传分离在不同的channel中，避免阻塞写入线程，目前暂时注释掉flush/fsync
+        # try:
+        #     self._fp.flush()
+        #     os.fsync(self._fp.fileno())
+        # except OSError:
+        #     pass
+        # self._flush_offset = self._index
 
     def ensure_flushed(self) -> None:
         assert self._fp is not None, "writer is not open"

swanlab/sdk/internal/core_python/transport/dispatch.py

@@ -68,9 +68,19 @@ def _upload_chunk(self, record_type: str, chunk: Sequence[Record]) -> bool:
         """单 chunk 上传。返回 True 表示成功，False 表示失败。"""
         if self._sender is None:
             raise RuntimeError("sender not set")
+        if record_type == "log":
+            console.debug(
+                f"Dispatching log records: count={len(chunk)}, nums={chunk[0].num}..{chunk[-1].num}",
+                write_to_tty=False,
+            )
         with safe.block(message=f"record chunk upload failed, record_type={record_type!r}", write_to_tty=False):
             # ⚠️：在持续断网的状态下，上传失败可能会重复打印，破坏掉 transport 层的 UploadWarningThrottle
             self._sender.upload(record_type, chunk)
+            if record_type == "log":
+                console.debug(
+                    f"Log sender completed: count={len(chunk)}, nums={chunk[0].num}..{chunk[-1].num}",
+                    write_to_tty=False,
+                )
             return True
         return False
 

swanlab/sdk/internal/core_python/transport/sender.py

@@ -244,6 +244,7 @@ def upload_media(self, records: Sequence[Record]) -> None:
 
     def upload_log(self, records: Sequence[Record]) -> None:
         with safe.block(message="Failed to upload terminal logs, skipping"):
+            console.debug(f"Preparing log HTTP request: records={len(records)}", write_to_tty=False)
             metrics: UploadLogBatch = []
             for record in records:
                 if not record.HasField("log"):
@@ -258,7 +259,9 @@ def upload_log(self, records: Sequence[Record]) -> None:
                         "create_time": create_time,
                     }
                     metrics.append(metric)
+            console.debug(f"Sending log HTTP request: metrics={len(metrics)}", write_to_tty=False)
             upload_log(self._project_id, self._experiment_id, metrics=metrics)
+            console.debug(f"Log HTTP request completed: metrics={len(metrics)}", write_to_tty=False)
 
     # ── 文件保存上传 ──
 

swanlab/sdk/internal/pkg/client/session.py

@@ -60,8 +60,8 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def send(self, request, *args, **kwargs):
-        if self.timeout is not None:
-            kwargs.setdefault("timeout", self.timeout)
+        if self.timeout is not None and kwargs.get("timeout") is None:
+            kwargs["timeout"] = self.timeout
 
         # 2. 直接从上下文中读取 retries，无需触碰 request.headers
         retries = request_retries_ctx.get()

swanlab/sdk/internal/run/components/consumer/__init__.py

@@ -179,6 +179,7 @@ def _handle_event(self, event) -> None:
             self._save_batch.append(self._builder.build_config(event))
         elif isinstance(event, LogEvent):
             self._log_batch.append(self._builder.build_log(event))
+            console.debug(f"Buffered terminal log record: pending={len(self._log_batch)}", write_to_tty=False)
         elif isinstance(event, FileSaveEvent):
             self._save_batch.append(self._builder.build_save(event))
 
@@ -210,7 +211,9 @@ def _flush(self) -> None:
             on_error=lambda _: self._restore_batches(log_batch, column_batch, scalar_batch, media_batch, save_batch),
         ):
             if log_batch:
+                console.debug(f"Flushing log records to core: count={len(log_batch)}", write_to_tty=False)
                 self._core.upsert_logs(log_batch)
+                console.debug(f"Core log call completed: count={len(log_batch)}", write_to_tty=False)
                 self._callbacker.on_log_flush(log_batch)
                 log_batch = []
 

swanlab/sdk/internal/run/components/terminal/__init__.py

@@ -22,7 +22,7 @@
 
 from swanlab.proto.swanlab.terminal.v1.log_pb2 import LogLevel
 from swanlab.sdk.internal.bus import EmitterProtocol, LogEvent
-from swanlab.sdk.internal.pkg import safe
+from swanlab.sdk.internal.pkg import console, safe
 
 from .capture import StreamCapture
 from .emulator import TerminalEmulator
@@ -232,4 +232,6 @@ def _emit_line(self, line: str, log_level: LogLevel = LogLevel.LOG_LEVEL_INFO) -
 
         ts = Timestamp()
         ts.GetCurrentTime()
+        console.debug("Emitting terminal log event", write_to_tty=False)
         self._emitter.emit(LogEvent(line=line, level=log_level, timestamp=ts))
+        console.debug("Terminal log event emitted", write_to_tty=False)

tests/unit/sdk/internal/pkg/client/test_client_session.py

@@ -7,6 +7,8 @@
 
 import pytest
 import responses
+from requests import Request, Response
+from requests.adapters import HTTPAdapter
 
 from swanlab.exceptions import ApiError
 from swanlab.sdk.internal.pkg.client.session import (
@@ -54,6 +56,19 @@ def test_session_send_success(session):
     assert resp.json() == {"ok": True}
 
 
+def test_timeout_adapter_applies_default_when_requests_passes_none(mocker):
+    """requests 会显式传 timeout=None，此时 adapter 应替换为配置的默认超时。"""
+    adapter = TimeoutHTTPAdapter(timeout=30)
+    request = Request("GET", "http://api.test/data").prepare()
+    response = Response()
+    response.status_code = 200
+    mock_send = mocker.patch.object(HTTPAdapter, "send", return_value=response)
+
+    adapter.send(request, timeout=None)
+
+    assert mock_send.call_args.kwargs["timeout"] == 30
+
+
 @responses.activate()
 def test_session_send_api_error_custom_parsed(session):
     """测试 4xx/5xx 时，能否正确提取 JSON 中的错误信息并抛出 ApiError"""