fix(history): preserve chronological replay ownership

engineering/persona-history-runtime/runtime/guanghu_history_runtime.py

@@ -30,6 +30,11 @@ PERSONAS = {
 PRIVATE_MARKERS = ("email", "token", "password", "secret", "api_key", "private_key")
 STREAM_BUFFER_BYTES = 1024 * 1024
 GPT_METADATA_SAMPLE_BYTES = 128 * 1024
+SOURCE_WAITING_STATUSES = {
+    "PENDING",
+    "WAITING_FOR_SOURCE",
+    "WAITING_FOR_SOURCE_ACCEPTANCE",
+}
 
 
 def now_iso() -> str:
@@ -62,6 +67,11 @@ def classify_personas(text: str) -> list[str]:
     ]
 
 
+def blocks_later_history(status: str) -> bool:
+    """An available active/error source owns the chronological replay lane."""
+    return status != "COMPLETE" and status not in SOURCE_WAITING_STATUSES
+
+
 def iter_top_level_json_objects(
     handle: BinaryIO, start_offset: int = 0
 ) -> Iterator[tuple[bytes, int]]:
@@ -753,6 +763,8 @@ class Runtime:
             self.process_source(source)
             if self.stop.is_set():
                 break
+            if blocks_later_history(self.store.state(source["id"])["status"]):
+                break
         return self.write_public()
 
     def run(self) -> None:

engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py

@@ -41,6 +41,12 @@ class RuntimeTests(unittest.TestCase):
         labels = runtime.classify_personas("曜冥宝宝和霜砚不是铸渊，也不是凝渊")
         self.assertEqual(labels, ["YAOMING-BABY", "SHUANGYAN", "ZHUYUAN", "NINGYUAN"])
 
+    def test_available_active_source_blocks_later_history(self):
+        self.assertTrue(runtime.blocks_later_history("ACTIVE"))
+        self.assertTrue(runtime.blocks_later_history("ERROR_RETRYABLE"))
+        self.assertFalse(runtime.blocks_later_history("COMPLETE"))
+        self.assertFalse(runtime.blocks_later_history("WAITING_FOR_SOURCE_ACCEPTANCE"))
+
     def test_public_snapshot_excludes_private_locators(self):
         with tempfile.TemporaryDirectory() as directory:
             root = pathlib.Path(directory)