From 777c3b0079fc3f94cc4e8303e02791363ae266b1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= <xsji@kland.com.cn>
Date: Tue, 19 May 2026 18:18:14 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1?=
 =?UTF-8?q?=E6=89=A7=E8=A1=8C=E8=BF=87=E7=A8=8B=E5=AE=9E=E6=97=B6=E5=88=B7?=
 =?UTF-8?q?=E6=96=B0=E2=80=94=E2=80=94RunJournal=20=E5=A2=9E=E5=8A=A0?=
 =?UTF-8?q?=E6=8C=89=E6=97=B6=E9=97=B4=20flush?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

RunJournal 原本只在缓冲满 20 条或运行结束时才把事件落库，短运行会在结束前一直查不到消息，导致定时任务执行详情页的「执行过程」一直显示「暂无执行消息」。

新增 _FLUSH_INTERVAL_SECONDS=3s 的按时间 flush（由 _put 驱动）；flush 失败时退回缓冲并重新抛出以便区分成败；flush 成功后若缓冲仍有残留则补一次 flush，失败则不重试避免空转。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---
 .../harness/deerflow/runtime/journal.py       | 30 ++++++++++++++++++-
 1 file changed, 29 insertions(+), 1 deletion(-)

diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/journal.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/journal.py
index a27ba0ec..70d81454 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/journal.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/journal.py
@@ -33,6 +33,13 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# Flush the buffer at least this often (seconds) even when it never fills up.
+# Live consumers — notably the scheduled-task run-detail page, which polls
+# ``GET /runs/{rid}/messages`` every 4s — would otherwise see nothing until a
+# run ends, because a short run can finish before the buffer reaches
+# ``flush_threshold``.
+_FLUSH_INTERVAL_SECONDS = 3.0
+
 
 class RunJournal(BaseCallbackHandler):
     """LangChain callback handler that captures events to RunEventStore."""
@@ -56,6 +63,9 @@ class RunJournal(BaseCallbackHandler):
         # Write buffer
         self._buffer: list[dict] = []
         self._pending_flush_tasks: set[asyncio.Task[None]] = set()
+        # Monotonic timestamp of the last flush attempt — drives time-based
+        # flushing so incremental progress is visible to live pollers.
+        self._last_flush_at: float = time.monotonic()
 
         # Token accumulators
         self._total_input_tokens = 0
@@ -269,8 +279,14 @@ class RunJournal(BaseCallbackHandler):
                 "created_at": datetime.now(UTC).isoformat(),
             }
         )
+        # Flush on a full buffer, or once enough time has elapsed since the
+        # last flush. The time-based path lets live consumers (the
+        # scheduled-task run-detail page polls every 4s) see incremental
+        # progress even when a run ends before the buffer fills up.
         if len(self._buffer) >= self._flush_threshold:
             self._flush_sync()
+        elif time.monotonic() - self._last_flush_at >= _FLUSH_INTERVAL_SECONDS:
+            self._flush_sync()
 
     def _flush_sync(self) -> None:
         """Best-effort flush of buffer to RunEventStore.
@@ -293,6 +309,7 @@ class RunJournal(BaseCallbackHandler):
             return
         batch = self._buffer.copy()
         self._buffer.clear()
+        self._last_flush_at = time.monotonic()
         task = loop.create_task(self._flush_async(batch))
         self._pending_flush_tasks.add(task)
         task.add_done_callback(self._on_flush_done)
@@ -307,8 +324,10 @@ class RunJournal(BaseCallbackHandler):
                 self.run_id,
                 exc_info=True,
             )
-            # Return failed events to buffer for retry on next flush
+            # Return failed events to buffer for retry on next flush, then
+            # re-raise so _on_flush_done can tell success from failure.
             self._buffer = batch + self._buffer
+            raise
 
     def _on_flush_done(self, task: asyncio.Task) -> None:
         self._pending_flush_tasks.discard(task)
@@ -316,7 +335,16 @@ class RunJournal(BaseCallbackHandler):
             return
         exc = task.exception()
         if exc:
+            # The batch was already returned to the buffer by _flush_async.
+            # Don't retry here — that would busy-loop on a persistent DB
+            # outage; the next _put (or the final flush()) retries instead.
             logger.warning("Journal flush task failed: %s", exc)
+            return
+        # Events may have accumulated while this flush was in flight — drain
+        # them now so incremental progress keeps reaching the store without
+        # waiting for the next _put to cross a threshold.
+        if self._buffer:
+            self._flush_sync()
 
     def _identify_caller(self, tags: list[str] | None) -> str:
         _tags = tags or []