diff --git a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py b/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py index 513519e..9010aa1 100644 --- a/engineering/persona-history-runtime/runtime/guanghu_history_runtime.py +++ b/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: diff --git a/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py b/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py index c0b435f..88707c2 100644 --- a/engineering/persona-history-runtime/runtime/test_guanghu_history_runtime.py +++ b/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)