mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-08-28 05:15:00 +00:00
874 lines
34 KiB
Python
874 lines
34 KiB
Python
"""Session transcript and storage evidence adapter."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
|
|
from .memory_refs import canonical_path_hash, memory_ref_id
|
|
from .store import EvidenceStore
|
|
from .tool_adapter import ToolEvidenceAdapter
|
|
from .types import EvidenceEvent, ResourceRef
|
|
|
|
|
|
class SessionEvidenceAdapter:
|
|
"""Translate durable SessionStorage entries into evidence refs."""
|
|
|
|
def __init__(self, store: EvidenceStore) -> None:
|
|
self._store = store
|
|
self._tool_adapter = ToolEvidenceAdapter(store)
|
|
|
|
async def on_session_entry(
|
|
self,
|
|
entry_type: str,
|
|
data: dict[str, Any],
|
|
) -> None:
|
|
event = self.build_event(entry_type, data)
|
|
if event is not None:
|
|
self._store.ingest_event(event)
|
|
|
|
def build_event(
|
|
self,
|
|
entry_type: str,
|
|
data: Mapping[str, Any],
|
|
) -> EvidenceEvent | None:
|
|
normalized = str(entry_type or data.get("entry_type") or "")
|
|
if normalized == "message":
|
|
return self._message_event(data)
|
|
if normalized == "transcript-segment":
|
|
return self._segment_event(data)
|
|
if normalized == "transcript-rewrite":
|
|
return self._rewrite_event(data)
|
|
if normalized in {"compact-summary", "compact_summary"}:
|
|
return self._compact_summary_event(data)
|
|
if normalized == "file-history-snapshot":
|
|
return self._file_history_event(data)
|
|
if normalized == "content-replacement":
|
|
return self._content_replacement_event(data)
|
|
return None
|
|
|
|
async def ingest_storage_delta(
|
|
self,
|
|
storage: Any,
|
|
*,
|
|
task_id: str | None = None,
|
|
parent_task_id: str | None = None,
|
|
agent_id: str | None = None,
|
|
) -> None:
|
|
"""Best-effort checkpoint scan for transcript entries missed by callback."""
|
|
|
|
transcript_path = Path(getattr(storage, "transcript_path", "") or "")
|
|
if not transcript_path.is_file():
|
|
return
|
|
session_id = _none_or_str(getattr(storage, "session_id", None))
|
|
if not session_id:
|
|
return
|
|
session_dir = _none_or_str(getattr(storage, "session_dir", None))
|
|
tool_results_dir = _none_or_str(getattr(storage, "tool_results_dir", None))
|
|
file_history_dir = _none_or_str(getattr(storage, "file_history_dir", None))
|
|
generation = int(getattr(storage, "current_generation", 0) or 0)
|
|
scoped_task_id = _none_or_str(task_id)
|
|
scoped_parent_task_id = _none_or_str(parent_task_id)
|
|
scoped_agent_id = _none_or_str(agent_id) or "primary"
|
|
message_index = 0
|
|
|
|
for entry in _iter_jsonl(transcript_path):
|
|
entry_type = str(entry.get("type") or entry.get("entry_type") or "")
|
|
data = entry.get("data") if isinstance(entry.get("data"), Mapping) else {}
|
|
if entry_type == "session-metadata":
|
|
scoped_task_id = (
|
|
_metadata_task_id(data)
|
|
or scoped_task_id
|
|
or _none_or_str(task_id)
|
|
)
|
|
scoped_parent_task_id = (
|
|
_metadata_parent_task_id(data)
|
|
or scoped_parent_task_id
|
|
or _none_or_str(parent_task_id)
|
|
)
|
|
scoped_agent_id = (
|
|
_metadata_agent_id(data)
|
|
or scoped_agent_id
|
|
or _none_or_str(agent_id)
|
|
or "primary"
|
|
)
|
|
generation = _metadata_generation(data, generation)
|
|
continue
|
|
if entry_type == "transcript-rewrite":
|
|
generation = _safe_int(data.get("new_generation"), generation)
|
|
|
|
payload = _payload_from_storage_entry(
|
|
entry,
|
|
session_id=session_id,
|
|
session_dir=session_dir,
|
|
transcript_path=str(transcript_path),
|
|
tool_results_dir=tool_results_dir,
|
|
file_history_dir=file_history_dir,
|
|
transcript_generation=generation,
|
|
task_id=scoped_task_id,
|
|
parent_task_id=scoped_parent_task_id,
|
|
agent_id=scoped_agent_id,
|
|
message_index=message_index,
|
|
)
|
|
if payload is None:
|
|
continue
|
|
if entry_type == "message":
|
|
message_index += 1
|
|
await self.on_session_entry(entry_type, payload)
|
|
|
|
def _message_event(self, data: Mapping[str, Any]) -> EvidenceEvent | None:
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
message_uuid = _none_or_str(data.get("message_uuid"))
|
|
if not session_id or not message_uuid:
|
|
return None
|
|
generation = int(data.get("transcript_generation") or 0)
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
parent_task_id = _none_or_str(data.get("parent_task_id"))
|
|
agent_id = _none_or_str(data.get("agent_id"))
|
|
created_at = _utc_now()
|
|
ref_id = (
|
|
_none_or_str(data.get("ref_id"))
|
|
or f"transcript_message:{session_id}:g{generation}:{message_uuid}"
|
|
)
|
|
transcript_path = _none_or_str(data.get("transcript_path"))
|
|
uri = (
|
|
f"{transcript_path}#generation={generation}&uuid={message_uuid}"
|
|
if transcript_path
|
|
else None
|
|
)
|
|
metadata = {
|
|
"transcript_generation": generation,
|
|
"logical_message_uuid": message_uuid,
|
|
"rewrite_marker": data.get("rewrite_marker"),
|
|
"parent_uuid": data.get("parent_uuid"),
|
|
"logical_parent_uuid": data.get("logical_parent_uuid"),
|
|
"role": data.get("role"),
|
|
"message_index": data.get("message_index"),
|
|
"has_tool_result_metadata": bool(data.get("has_tool_result_metadata")),
|
|
"attachment_type": data.get("attachment_type"),
|
|
"content_shape": data.get("content_shape"),
|
|
"tool_call_id": data.get("tool_call_id"),
|
|
"tool_name": data.get("tool_name"),
|
|
}
|
|
refs = [
|
|
ResourceRef(
|
|
ref_id=ref_id,
|
|
ref_type="transcript_message",
|
|
uri=uri,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=parent_task_id,
|
|
agent_id=agent_id,
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="persisted",
|
|
role="primary",
|
|
hash=_none_or_str(data.get("message_hash")),
|
|
preview=str(data.get("content_preview") or "")[:500],
|
|
metadata=metadata,
|
|
)
|
|
]
|
|
tool_result_ref = self._tool_result_from_message(data, created_at=created_at)
|
|
if tool_result_ref is not None:
|
|
refs.append(tool_result_ref)
|
|
refs.extend(self._memory_refs_from_attachment(data, created_at=created_at))
|
|
primary_refs, supporting_refs, derived_refs = _refs_by_event_role(refs)
|
|
|
|
digest = _digest({"session_id": session_id, "ref_id": ref_id})
|
|
return EvidenceEvent.create(
|
|
event_id=f"evt_session_msg_{digest}",
|
|
event_type="session_transcript_message",
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=parent_task_id,
|
|
agent_id=agent_id,
|
|
idempotency_key=f"session:message:{ref_id}",
|
|
primary_refs=primary_refs,
|
|
supporting_refs=supporting_refs,
|
|
derived_refs=derived_refs,
|
|
metadata={"entry_type": "message", "role": data.get("role")},
|
|
)
|
|
|
|
def _segment_event(self, data: Mapping[str, Any]) -> EvidenceEvent | None:
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
segment_id = _none_or_str(data.get("segment_id"))
|
|
path = _none_or_str(data.get("path"))
|
|
if not session_id or not (segment_id or path):
|
|
return None
|
|
created_at = _utc_now()
|
|
ref_id = f"transcript_segment:{session_id}:{segment_id or _digest(path)}"
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
missing = bool(path and not Path(path).expanduser().is_file())
|
|
ref = ResourceRef(
|
|
ref_id=ref_id,
|
|
ref_type="transcript_segment",
|
|
uri=path,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=_none_or_str(data.get("parent_task_id")),
|
|
agent_id=_none_or_str(data.get("agent_id")),
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="fallback" if missing else "persisted",
|
|
role="supporting" if missing else "primary",
|
|
hash=_file_hash(path),
|
|
preview=(
|
|
f"{data.get('message_count')} messages; "
|
|
f"{data.get('head_uuid')}..{data.get('tail_uuid')}"
|
|
),
|
|
metadata={
|
|
"segment_id": segment_id,
|
|
"reason": data.get("reason"),
|
|
"head_uuid": data.get("head_uuid"),
|
|
"tail_uuid": data.get("tail_uuid"),
|
|
"message_count": data.get("message_count"),
|
|
"missing": missing,
|
|
},
|
|
)
|
|
primary_refs, supporting_refs, derived_refs = _refs_by_event_role([ref])
|
|
return EvidenceEvent.create(
|
|
event_id=f"evt_session_segment_{_digest(ref_id)}",
|
|
event_type="session_transcript_segment",
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
idempotency_key=f"session:segment:{ref_id}",
|
|
primary_refs=primary_refs,
|
|
supporting_refs=supporting_refs,
|
|
derived_refs=derived_refs,
|
|
metadata={"entry_type": "transcript-segment"},
|
|
)
|
|
|
|
def _rewrite_event(self, data: Mapping[str, Any]) -> EvidenceEvent | None:
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
if not session_id:
|
|
return None
|
|
old_generation = data.get("old_generation")
|
|
new_generation = data.get("new_generation")
|
|
created_at = _utc_now()
|
|
ref_id = f"transcript_rewrite:{session_id}:g{old_generation}:g{new_generation}"
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
parent_task_id = _none_or_str(data.get("parent_task_id"))
|
|
agent_id = _none_or_str(data.get("agent_id"))
|
|
preserved = _none_or_str(data.get("preserved_segment_path"))
|
|
preserved_ref_id = (
|
|
"transcript_segment:"
|
|
f"{session_id}:preserved-g{old_generation}:{_digest(preserved)[:12]}"
|
|
if preserved
|
|
else None
|
|
)
|
|
rewrite_ref = ResourceRef(
|
|
ref_id=ref_id,
|
|
ref_type="transcript_rewrite",
|
|
uri=_none_or_str(data.get("transcript_path")),
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=parent_task_id,
|
|
agent_id=agent_id,
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="persisted",
|
|
role="supporting",
|
|
preview=f"transcript generation {old_generation} -> {new_generation}",
|
|
metadata={
|
|
"old_generation": old_generation,
|
|
"new_generation": new_generation,
|
|
"message_count": data.get("message_count"),
|
|
"preserved_segment_path": data.get("preserved_segment_path"),
|
|
},
|
|
raw_backrefs=[preserved_ref_id] if preserved_ref_id else [],
|
|
)
|
|
refs = [rewrite_ref]
|
|
if preserved and preserved_ref_id:
|
|
refs.append(
|
|
ResourceRef(
|
|
ref_id=preserved_ref_id,
|
|
ref_type="transcript_segment",
|
|
uri=preserved,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=parent_task_id,
|
|
agent_id=agent_id,
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="persisted",
|
|
role="supporting",
|
|
hash=_file_hash(preserved),
|
|
preview=f"preserved generation {old_generation}",
|
|
metadata={
|
|
"reason": "transcript_rewrite_preserve",
|
|
"old_generation": old_generation,
|
|
},
|
|
)
|
|
)
|
|
primary_refs, supporting_refs, derived_refs = _refs_by_event_role(refs)
|
|
return EvidenceEvent.create(
|
|
event_id=f"evt_session_rewrite_{_digest(ref_id)}",
|
|
event_type="session_transcript_rewrite",
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
idempotency_key=f"session:rewrite:{ref_id}",
|
|
primary_refs=primary_refs,
|
|
supporting_refs=supporting_refs,
|
|
derived_refs=derived_refs,
|
|
metadata={"entry_type": "transcript-rewrite"},
|
|
)
|
|
|
|
def _compact_summary_event(self, data: Mapping[str, Any]) -> EvidenceEvent | None:
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
summary_uuid = _none_or_str(data.get("summary_message_uuid"))
|
|
if not session_id:
|
|
return None
|
|
created_at = _utc_now()
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
segment_ref_id = _none_or_str(data.get("segment_ref_id"))
|
|
raw_backrefs = [segment_ref_id] if segment_ref_id else []
|
|
digest = _digest(
|
|
{
|
|
"session_id": session_id,
|
|
"summary_uuid": summary_uuid,
|
|
"segment_ref_id": segment_ref_id,
|
|
"source": data.get("compact_source"),
|
|
}
|
|
)
|
|
ref_id = f"compact_summary:{session_id}:{summary_uuid or digest}"
|
|
transcript_path = _none_or_str(data.get("transcript_path"))
|
|
ref = ResourceRef(
|
|
ref_id=ref_id,
|
|
ref_type="compact_summary",
|
|
uri=(
|
|
f"{transcript_path}#compact-summary:{summary_uuid}"
|
|
if transcript_path and summary_uuid
|
|
else transcript_path
|
|
),
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=_none_or_str(data.get("parent_task_id")),
|
|
agent_id=_none_or_str(data.get("agent_id")),
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="derived" if raw_backrefs else "summary_only",
|
|
role="supporting",
|
|
preview=f"compact summary from {data.get('compact_source') or 'unknown'}",
|
|
metadata={
|
|
"compact_source": data.get("compact_source"),
|
|
"summary_message_uuid": summary_uuid,
|
|
"segment_ref_id": segment_ref_id,
|
|
"memory_path": data.get("memory_path"),
|
|
"was_truncated": bool(data.get("was_truncated")),
|
|
"missing_raw_backrefs": not bool(raw_backrefs),
|
|
},
|
|
raw_backrefs=raw_backrefs,
|
|
)
|
|
return EvidenceEvent.create(
|
|
event_id=f"evt_compact_summary_{_digest(ref_id)}",
|
|
event_type="session_compact_summary",
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
idempotency_key=f"session:compact_summary:{ref_id}",
|
|
supporting_refs=[ref],
|
|
metadata={"entry_type": "compact-summary"},
|
|
)
|
|
|
|
def _file_history_event(self, data: Mapping[str, Any]) -> EvidenceEvent | None:
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
message_uuid = _none_or_str(data.get("message_uuid"))
|
|
if not session_id or not message_uuid:
|
|
return None
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
created_at = _utc_now()
|
|
snapshot = data.get("snapshot")
|
|
touched_files, read_files, written_files = _file_history_paths(snapshot)
|
|
changed_paths: list[str] = []
|
|
if isinstance(snapshot, Mapping):
|
|
raw_paths = snapshot.get("paths") or snapshot.get("files") or []
|
|
if isinstance(raw_paths, (list, tuple)):
|
|
changed_paths = [str(path) for path in raw_paths if path]
|
|
if not changed_paths:
|
|
changed_paths = list(touched_files)
|
|
ref_id = (
|
|
f"file_history:{session_id}:{message_uuid}:"
|
|
f"{_digest({'snapshot': snapshot, 'update': data.get('is_snapshot_update')})[:12]}"
|
|
)
|
|
ref = ResourceRef(
|
|
ref_id=ref_id,
|
|
ref_type="file_history",
|
|
uri=(
|
|
f"{_none_or_str(data.get('transcript_path'))}#file-history:{message_uuid}"
|
|
if _none_or_str(data.get("transcript_path"))
|
|
else _none_or_str(data.get("file_history_dir"))
|
|
),
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=_none_or_str(data.get("parent_task_id")),
|
|
agent_id=_none_or_str(data.get("agent_id")),
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="persisted",
|
|
role="supporting",
|
|
preview=f"file history snapshot for {message_uuid}",
|
|
metadata={
|
|
"message_uuid": message_uuid,
|
|
"snapshot_id": (
|
|
snapshot.get("snapshot_id")
|
|
if isinstance(snapshot, Mapping)
|
|
else None
|
|
),
|
|
"is_snapshot_update": bool(data.get("is_snapshot_update")),
|
|
"changed_paths": changed_paths,
|
|
"touched_files": touched_files,
|
|
"read_files": read_files,
|
|
"written_files": written_files,
|
|
"file_history_dir": data.get("file_history_dir"),
|
|
"original_session_id": data.get("original_session_id"),
|
|
"original_task_id": data.get("original_task_id"),
|
|
"original_parent_task_id": data.get("original_parent_task_id"),
|
|
"original_agent_id": data.get("original_agent_id"),
|
|
},
|
|
)
|
|
return EvidenceEvent.create(
|
|
event_id=f"evt_file_history_{_digest(ref_id)}",
|
|
event_type="session_file_history_snapshot",
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
idempotency_key=f"session:file_history:{ref_id}",
|
|
supporting_refs=[ref],
|
|
metadata={"entry_type": "file-history-snapshot"},
|
|
)
|
|
|
|
def _content_replacement_event(self, data: Mapping[str, Any]) -> EvidenceEvent | None:
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
if not session_id:
|
|
return None
|
|
created_at = _utc_now()
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
ref_id = f"content_replacement:{session_id}:{_digest(data)}"
|
|
ref = ResourceRef(
|
|
ref_id=ref_id,
|
|
ref_type="content_replacement",
|
|
uri=_none_or_str(data.get("path")),
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="persisted",
|
|
role="supporting",
|
|
preview=str(data.get("reason") or "content replacement")[:200],
|
|
metadata={
|
|
key: value
|
|
for key, value in data.items()
|
|
if key not in {"content", "message"}
|
|
},
|
|
)
|
|
return EvidenceEvent.create(
|
|
event_id=f"evt_content_repl_{_digest(ref_id)}",
|
|
event_type="session_content_replacement",
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
idempotency_key=f"session:content_replacement:{ref_id}",
|
|
supporting_refs=[ref],
|
|
metadata={"entry_type": "content-replacement"},
|
|
)
|
|
|
|
def _tool_result_from_message(
|
|
self,
|
|
data: Mapping[str, Any],
|
|
*,
|
|
created_at: str,
|
|
) -> ResourceRef | None:
|
|
tool_result_metadata = data.get("tool_result_metadata")
|
|
if not isinstance(tool_result_metadata, Mapping):
|
|
return None
|
|
tool_call_id = _none_or_str(data.get("tool_call_id"))
|
|
tool_name = _none_or_str(data.get("tool_name"))
|
|
if not tool_call_id or not tool_name:
|
|
return None
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
agent_id = _none_or_str(data.get("agent_id")) or "primary"
|
|
backend = _none_or_str(data.get("backend")) or "unknown"
|
|
server_name = _none_or_str(data.get("server_name")) or "default"
|
|
raw_backref = f"tool_event:{session_id or 'none'}:{task_id or 'none'}:{agent_id}:{tool_call_id}"
|
|
return self._tool_adapter.persisted_tool_result_ref(
|
|
{
|
|
"tool_result_metadata": tool_result_metadata,
|
|
"current_iteration": data.get("current_iteration"),
|
|
},
|
|
created_at=created_at,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=_none_or_str(data.get("parent_task_id")),
|
|
agent_id=agent_id,
|
|
tool_use_id=tool_call_id,
|
|
tool_name=tool_name,
|
|
tool_key=f"{backend}:{server_name}:{tool_name}",
|
|
raw_backref=raw_backref,
|
|
)
|
|
|
|
def _memory_refs_from_attachment(
|
|
self,
|
|
data: Mapping[str, Any],
|
|
*,
|
|
created_at: str,
|
|
) -> list[ResourceRef]:
|
|
attachment_type = _none_or_str(data.get("attachment_type"))
|
|
if attachment_type not in {"relevant_memories", "nested_memory"}:
|
|
return []
|
|
memories = data.get("memories")
|
|
if not isinstance(memories, list):
|
|
return []
|
|
refs: list[ResourceRef] = []
|
|
session_id = _none_or_str(data.get("session_id"))
|
|
task_id = _none_or_str(data.get("task_id"))
|
|
parent_task_id = _none_or_str(data.get("parent_task_id"))
|
|
agent_id = _none_or_str(data.get("agent_id"))
|
|
for memory in memories:
|
|
if not isinstance(memory, Mapping):
|
|
continue
|
|
path = _none_or_str(memory.get("path"))
|
|
if not path:
|
|
continue
|
|
source_kind = f"transcript_attachment:{attachment_type}"
|
|
refs.append(
|
|
ResourceRef(
|
|
ref_id=memory_ref_id(
|
|
source_kind,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
agent_id=agent_id,
|
|
path=path,
|
|
),
|
|
ref_type="memory_ref",
|
|
uri=path,
|
|
session_id=session_id,
|
|
task_id=task_id,
|
|
parent_task_id=parent_task_id,
|
|
agent_id=agent_id,
|
|
producer="session_storage",
|
|
created_at=created_at,
|
|
reliability="persisted",
|
|
role="supporting",
|
|
hash=_file_hash(path),
|
|
preview=str(memory.get("header") or "")[:300],
|
|
metadata={
|
|
"attachment_type": attachment_type,
|
|
"path": path,
|
|
"mtime_ms": memory.get("mtimeMs"),
|
|
"limit": memory.get("limit"),
|
|
"memory_kind": _attachment_memory_kind(attachment_type),
|
|
"canonical_path_hash": canonical_path_hash(path),
|
|
"loaded_in_context": True,
|
|
"read_or_written_by_tool": False,
|
|
"source_event": "transcript_attachment",
|
|
"source_kind": source_kind,
|
|
},
|
|
)
|
|
)
|
|
return refs
|
|
|
|
|
|
def _file_hash(path_text: str | None) -> str | None:
|
|
if not path_text:
|
|
return None
|
|
try:
|
|
path = Path(path_text).expanduser()
|
|
if not path.is_file():
|
|
return None
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _attachment_memory_kind(attachment_type: str | None) -> str:
|
|
if attachment_type == "nested_memory":
|
|
return "nested"
|
|
if attachment_type == "relevant_memories":
|
|
return "relevant"
|
|
return "model_visible"
|
|
|
|
|
|
def _refs_by_event_role(
|
|
refs: list[ResourceRef],
|
|
) -> tuple[list[ResourceRef], list[ResourceRef], list[ResourceRef]]:
|
|
primary: list[ResourceRef] = []
|
|
supporting: list[ResourceRef] = []
|
|
derived: list[ResourceRef] = []
|
|
for ref in refs:
|
|
if ref.role == "primary":
|
|
primary.append(ref)
|
|
elif ref.role == "derived":
|
|
derived.append(ref)
|
|
else:
|
|
supporting.append(ref)
|
|
return primary, supporting, derived
|
|
|
|
|
|
def _iter_jsonl(path: Path) -> list[dict[str, Any]]:
|
|
entries: list[dict[str, Any]] = []
|
|
try:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
for line in handle:
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
value = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
if isinstance(value, dict):
|
|
entries.append(value)
|
|
except OSError:
|
|
return []
|
|
return entries
|
|
|
|
|
|
def _payload_from_storage_entry(
|
|
entry: Mapping[str, Any],
|
|
*,
|
|
session_id: str,
|
|
session_dir: str | None,
|
|
transcript_path: str,
|
|
tool_results_dir: str | None,
|
|
file_history_dir: str | None,
|
|
transcript_generation: int,
|
|
task_id: str | None,
|
|
parent_task_id: str | None,
|
|
agent_id: str | None,
|
|
message_index: int,
|
|
) -> dict[str, Any] | None:
|
|
entry_type = str(entry.get("type") or entry.get("entry_type") or "")
|
|
data = entry.get("data") if isinstance(entry.get("data"), Mapping) else {}
|
|
base = {
|
|
"entry_type": entry_type,
|
|
"session_id": session_id,
|
|
"session_dir": session_dir,
|
|
"transcript_path": transcript_path,
|
|
"tool_results_dir": tool_results_dir,
|
|
"file_history_dir": file_history_dir,
|
|
"transcript_generation": _safe_int(
|
|
data.get("transcript_generation"),
|
|
transcript_generation,
|
|
),
|
|
"task_id": _none_or_str(data.get("task_id")) or task_id,
|
|
"parent_task_id": _none_or_str(data.get("parent_task_id")) or parent_task_id,
|
|
"agent_id": _none_or_str(data.get("agent_id")) or agent_id or "primary",
|
|
}
|
|
if entry_type == "message":
|
|
message = entry.get("message")
|
|
if not isinstance(message, Mapping):
|
|
return None
|
|
message_uuid = _none_or_str(entry.get("uuid")) or _message_uuid(message)
|
|
if not message_uuid:
|
|
return None
|
|
meta = message.get("_meta") if isinstance(message.get("_meta"), Mapping) else {}
|
|
meta_map = meta if isinstance(meta, Mapping) else {}
|
|
tool_result_metadata = meta_map.get("tool_result_metadata")
|
|
attachment = meta_map.get("attachment")
|
|
attachment_map = attachment if isinstance(attachment, Mapping) else {}
|
|
generation = int(base["transcript_generation"] or 0)
|
|
return {
|
|
**base,
|
|
"ref_id": (
|
|
_none_or_str(data.get("ref_id"))
|
|
or f"transcript_message:{session_id}:g{generation}:{message_uuid}"
|
|
),
|
|
"message_uuid": message_uuid,
|
|
"parent_uuid": _none_or_str(entry.get("parent_uuid")),
|
|
"logical_parent_uuid": _none_or_str(entry.get("logical_parent_uuid")),
|
|
"role": message.get("role"),
|
|
"message_index": message_index,
|
|
"content_preview": _message_content_preview(message),
|
|
"content_shape": _message_content_shape(message),
|
|
"message_hash": _sha256_json(message),
|
|
"has_tool_result_metadata": isinstance(tool_result_metadata, Mapping),
|
|
"tool_result_metadata": (
|
|
dict(tool_result_metadata)
|
|
if isinstance(tool_result_metadata, Mapping)
|
|
else {}
|
|
),
|
|
"tool_call_id": message.get("tool_call_id") or meta_map.get("tool_call_id"),
|
|
"tool_name": message.get("name") or meta_map.get("tool_name"),
|
|
"backend": meta_map.get("backend"),
|
|
"server_name": meta_map.get("server_name"),
|
|
"attachment_type": attachment_map.get("type"),
|
|
"memories": _attachment_memory_summaries(attachment_map),
|
|
}
|
|
if entry_type in {
|
|
"transcript-segment",
|
|
"transcript-rewrite",
|
|
"file-history-snapshot",
|
|
"content-replacement",
|
|
"compact-summary",
|
|
}:
|
|
return {**base, **dict(data)}
|
|
return None
|
|
|
|
|
|
def _message_uuid(message: Mapping[str, Any]) -> str | None:
|
|
meta = message.get("_meta")
|
|
if isinstance(meta, Mapping) and meta.get("uuid"):
|
|
return str(meta["uuid"])
|
|
value = message.get("uuid")
|
|
return str(value) if value else None
|
|
|
|
|
|
def _message_content_preview(message: Mapping[str, Any], max_chars: int = 500) -> str:
|
|
content = message.get("content")
|
|
if isinstance(content, str):
|
|
return content[:max_chars]
|
|
if isinstance(content, list):
|
|
parts: list[str] = []
|
|
for block in content:
|
|
if isinstance(block, Mapping):
|
|
if block.get("type") == "text":
|
|
parts.append(str(block.get("text") or ""))
|
|
elif block.get("type"):
|
|
parts.append(f"[{block.get('type')}]")
|
|
elif isinstance(block, str):
|
|
parts.append(block)
|
|
return "\n".join(parts)[:max_chars]
|
|
return "" if content is None else str(content)[:max_chars]
|
|
|
|
|
|
def _message_content_shape(message: Mapping[str, Any]) -> str:
|
|
meta = message.get("_meta")
|
|
if isinstance(meta, Mapping) and meta.get("type") == "tool_result":
|
|
return "tool_result"
|
|
content = message.get("content")
|
|
if isinstance(content, str):
|
|
return "text"
|
|
if isinstance(content, list):
|
|
if any(
|
|
isinstance(block, Mapping)
|
|
and block.get("type") in {"image", "image_url", "document"}
|
|
for block in content
|
|
):
|
|
return "attachment"
|
|
return "blocks"
|
|
return "text" if content is not None else "empty"
|
|
|
|
|
|
def _attachment_memory_summaries(attachment: Mapping[str, Any]) -> list[dict[str, Any]]:
|
|
memories = attachment.get("memories")
|
|
if not isinstance(memories, list):
|
|
if attachment.get("path"):
|
|
memories = [attachment]
|
|
else:
|
|
return []
|
|
result: list[dict[str, Any]] = []
|
|
for memory in memories:
|
|
if not isinstance(memory, Mapping):
|
|
continue
|
|
path = _none_or_str(memory.get("path"))
|
|
if not path:
|
|
continue
|
|
result.append(
|
|
{
|
|
"path": path,
|
|
"mtimeMs": memory.get("mtimeMs"),
|
|
"header": memory.get("header"),
|
|
"limit": memory.get("limit"),
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def _file_history_paths(snapshot: Any) -> tuple[list[str], list[str], list[str]]:
|
|
if not isinstance(snapshot, Mapping):
|
|
return [], [], []
|
|
touched: set[str] = set()
|
|
read_files: set[str] = set()
|
|
written: set[str] = set()
|
|
for key in ("paths", "files", "touched_files", "changed_paths"):
|
|
raw = snapshot.get(key)
|
|
if isinstance(raw, (list, tuple, set)):
|
|
touched.update(str(path) for path in raw if path)
|
|
for key in ("read_files", "readFiles"):
|
|
raw = snapshot.get(key)
|
|
if isinstance(raw, (list, tuple, set)):
|
|
read_files.update(str(path) for path in raw if path)
|
|
for key in ("written_files", "writtenFiles"):
|
|
raw = snapshot.get(key)
|
|
if isinstance(raw, (list, tuple, set)):
|
|
written.update(str(path) for path in raw if path)
|
|
backups = snapshot.get("tracked_file_backups")
|
|
if isinstance(backups, Mapping):
|
|
touched.update(str(path) for path in backups.keys() if path)
|
|
written.update(str(path) for path in backups.keys() if path)
|
|
return sorted(touched), sorted(read_files), sorted(written)
|
|
|
|
|
|
def _metadata_task_id(data: Mapping[str, Any]) -> str | None:
|
|
runtime = data.get("runtime") if isinstance(data.get("runtime"), Mapping) else {}
|
|
runtime_map = runtime if isinstance(runtime, Mapping) else {}
|
|
return _none_or_str(runtime_map.get("active_task_id")) or _none_or_str(
|
|
data.get("last_task_id")
|
|
)
|
|
|
|
|
|
def _metadata_parent_task_id(data: Mapping[str, Any]) -> str | None:
|
|
runtime = data.get("runtime") if isinstance(data.get("runtime"), Mapping) else {}
|
|
runtime_map = runtime if isinstance(runtime, Mapping) else {}
|
|
return _none_or_str(runtime_map.get("parent_task_id")) or _none_or_str(
|
|
data.get("parent_task_id")
|
|
)
|
|
|
|
|
|
def _metadata_agent_id(data: Mapping[str, Any]) -> str | None:
|
|
runtime = data.get("runtime") if isinstance(data.get("runtime"), Mapping) else {}
|
|
runtime_map = runtime if isinstance(runtime, Mapping) else {}
|
|
return _none_or_str(runtime_map.get("agent_id")) or _none_or_str(data.get("agent_id"))
|
|
|
|
|
|
def _metadata_generation(data: Mapping[str, Any], default: int) -> int:
|
|
runtime = data.get("runtime") if isinstance(data.get("runtime"), Mapping) else {}
|
|
runtime_map = runtime if isinstance(runtime, Mapping) else {}
|
|
return _safe_int(
|
|
data.get("transcript_generation"),
|
|
_safe_int(runtime_map.get("transcript_generation"), default),
|
|
)
|
|
|
|
|
|
def _safe_int(value: Any, default: int = 0) -> int:
|
|
try:
|
|
if value is not None:
|
|
return max(0, int(value))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return max(0, int(default))
|
|
|
|
|
|
def _sha256_json(value: Any) -> str:
|
|
return hashlib.sha256(
|
|
json.dumps(value, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def _digest(payload: Any) -> str:
|
|
return hashlib.sha256(
|
|
json.dumps(payload, sort_keys=True, default=str).encode("utf-8")
|
|
).hexdigest()[:24]
|
|
|
|
|
|
def _none_or_str(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value)
|
|
return text if text else None
|
|
|
|
|
|
def _utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|