DasPoschi/hermes-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: address turn journal lifecycle review

Browse Source
This commit is contained in:
ai-ag2026
2026-05-11 17:16:43 +02:00
parent 4b486f2860
commit c864ad47af
6 changed files with 63 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
api/routes.py
View File

@@ -6675,7 +6675,9 @@ def _start_chat_stream_for_session(
model_provider=model_provider,
stream_id=stream_id,
)
diag.stage("turn_journal_submitted") if diag else None
diag.stage("turn_journal_submitted") if diag else None
journal_event = {}
try:
from api.turn_journal import append_turn_journal_event
journal_event = append_turn_journal_event(
s.session_id,
@@ -6691,6 +6693,8 @@ def _start_chat_stream_for_session(
"created_at": s.pending_started_at,
},
)
except Exception:
logger.warning("Failed to append submitted turn journal event", exc_info=True)
diag.stage("set_last_workspace") if diag else None
set_last_workspace(workspace)
diag.stage("stream_registration") if diag else None

34
api/streaming.py
View File

@@ -3570,21 +3570,25 @@ def _run_agent_streaming(
# Better to leave context_length=0 than crash the save.
pass
if not ephemeral and s.messages:
for _idx, _msg in enumerate(s.messages):
if isinstance(_msg, dict) and _msg.get('role') == 'assistant':
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "assistant_started",
"created_at": float(_msg.get('timestamp') or time.time()),
"assistant_message_index": _idx,
},
)
except Exception:
logger.debug("Failed to append assistant_started turn journal event", exc_info=True)
break
_latest_assistant_idx = next(
(idx for idx in range(len(s.messages) - 1, -1, -1)
if isinstance(s.messages[idx], dict) and s.messages[idx].get('role') == 'assistant'),
None,
)
if _latest_assistant_idx is not None:
_latest_assistant = s.messages[_latest_assistant_idx]
try:
append_turn_journal_event_for_stream(
s.session_id,
stream_id,
{
"event": "assistant_started",
"created_at": float(_latest_assistant.get('timestamp') or time.time()),
"assistant_message_index": _latest_assistant_idx,
},
)
except Exception:
logger.debug("Failed to append assistant_started turn journal event", exc_info=True)
s.save()
if not ephemeral:
try:

16
api/turn_journal.py
View File

@@ -63,14 +63,10 @@ def append_turn_journal_event(
path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n"
fd = os.open(path, os.O_CREAT | os.O_APPEND | os.O_WRONLY, 0o600)
try:
with os.fdopen(fd, "a", encoding="utf-8") as fh:
fh.write(line)
fh.flush()
os.fsync(fh.fileno())
finally:
# fd ownership moves to fdopen(); this finally exists only for clarity.
pass
with os.fdopen(fd, "a", encoding="utf-8") as fh:
fh.write(line)
fh.flush()
os.fsync(fh.fileno())
try:
dir_fd = os.open(path.parent, os.O_DIRECTORY)
try:
@@ -115,7 +111,9 @@ def derive_turn_journal_states(events: Iterable[dict]) -> dict[str, dict]:
turn_id = str(event.get("turn_id") or "").strip()
if not turn_id:
continue
states[turn_id] = event
previous = states.get(turn_id)
if previous is None or float(event.get("created_at") or 0) >= float(previous.get("created_at") or 0):
states[turn_id] = event
return states

9
tests/test_turn_journal.py
View File

@@ -69,6 +69,15 @@ def test_derive_turn_journal_states_keeps_latest_event_per_turn():
assert states["turn-2"]["event"] == "submitted"
def test_derive_turn_journal_states_uses_created_at_not_file_order():
states = derive_turn_journal_states([
{"event": "completed", "turn_id": "turn-1", "created_at": 20},
{"event": "submitted", "turn_id": "turn-1", "created_at": 10},
])
assert states["turn-1"]["event"] == "completed"
def test_audit_reports_pending_turn_journal_entry_when_user_message_absent(tmp_path):
_write_session(tmp_path, "sid-1", messages=[])
append_turn_journal_event(

13
tests/test_turn_journal_callsite.py
View File

@@ -10,3 +10,16 @@ def test_chat_start_appends_submitted_turn_journal_before_worker_thread_start():
assert save_idx < append_idx < thread_idx
assert '"event": "submitted"' in src[append_idx:thread_idx]
assert '"role": "user"' in src[append_idx:thread_idx]
def test_chat_start_writes_turn_journal_after_session_lock_and_handles_failure():
src = Path("api/routes.py").read_text(encoding="utf-8")
lock_idx = src.index("with session_lock:")
append_idx = src.index("append_turn_journal_event(", lock_idx)
stream_registration_idx = src.index("STREAMS[stream_id] = stream", append_idx)
lock_block = src[lock_idx:append_idx]
append_block = src[append_idx:stream_registration_idx]
assert "append_turn_journal_event(" not in lock_block
assert "except Exception:" in append_block
assert "Failed to append submitted turn journal event" in append_block

10
tests/test_turn_journal_lifecycle_callsite.py
View File

@@ -19,6 +19,16 @@ def test_streaming_appends_assistant_started_before_final_save():
assert block_idx < assistant_idx < save_idx
def test_streaming_assistant_started_uses_latest_assistant_message():
src = Path("api/streaming.py").read_text(encoding="utf-8")
block_idx = src.index("if not ephemeral and s.messages:")
assistant_idx = src.index('"event": "assistant_started"', block_idx)
block = src[block_idx:assistant_idx]
assert "range(len(s.messages) - 1, -1, -1)" in block
assert '"assistant_message_index": _latest_assistant_idx' in src[assistant_idx:src.index("s.save()", assistant_idx)]
def test_streaming_appends_completed_after_final_save():
src = Path("api/streaming.py").read_text(encoding="utf-8")
assistant_idx = src.index('"event": "assistant_started"')