24802435632

Committed 22 Apr 2026 08:56AM UTC coverage: 76.474% (-2.0%) from 78.441%

Build # 24802435632

Build Type

push

github

Committed by

christophevg

Commit Message

docs: replace ASCII/mermaid diagrams with SVG

Remove Mermaid diagram from README.md and replace ASCII diagrams
with SVG references using consistent approved style.

Changes:
- Remove Mermaid diagram, use architecture-diagram.svg
- Create workflow-model.svg for Interactive ↔ Autonomous diagram
- Create architecture-diagram.svg for library-first design
- Update docs with Sphinx {image} directive syntax
- Copy SVG files to docs/_static for Sphinx builds
- Remove "Positioning in the Ecosystem" section (to be revamped)

🤖 Implemented together with a coding agent.

Coverage Stats

336 of 430 branches covered (78.14%)

Branch coverage included in aggregate %.

2232 of 2928 relevant lines covered (76.23%)

0.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.01

/src/yoker/context/basic.py

"""Basic persistence context manager implementation.

Provides JSONL-based context persistence with secure file handling.

Note:
  File locking uses fcntl (Unix-only). On Windows, file writes
  are still atomic but lack inter-process locking protection.
  For production use on Windows, consider adding a cross-platform
  file locking library.
"""

import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any

from yoker.context.interface import ContextStatistics
from yoker.context.validator import is_safe_path, validate_session_id, validate_storage_path
from yoker.exceptions import ContextCorruptionError, SessionNotFoundError
from yoker.logging import get_logger

log = get_logger(__name__)

# File permissions
DIR_MODE = 0o700  # Owner-only for directories
FILE_MODE = 0o600  # Owner-only for files


class BasicPersistenceContextManager:
  """Context manager with JSONL persistence.

  Stores conversation history in JSONL (JSON Lines) format with:
  - Atomic writes for crash safety
  - Secure file permissions
  - Session lifecycle tracking

  Record types:
  - session_start: Session metadata
  - message: User/assistant/system message
  - tool_result: Tool execution result
  - turn: Turn boundary marker
  - session_end: Session termination marker
  """

  def __init__(
    self,
    storage_path: Path | str,
    session_id: str = "auto",
  ) -> None:
    """Initialize context manager.

    Args:
      storage_path: Directory for storing context files.
      session_id: Session ID or "auto" to generate.

    Raises:
      ValidationError: If storage_path or session_id is invalid.
    """
    # Validate and resolve storage path
    self._storage_path = validate_storage_path(
      Path(storage_path), "context.storage_path"
    )

    # Validate session ID
    self._session_id = validate_session_id(session_id, "context.session_id")

    # In-memory context: ordered sequence of all items
    # Each item is {"type": "message"|"tool_result", "data": {...}}
    self._sequence: list[dict[str, Any]] = []

    # Statistics
    self._start_time = datetime.now()
    self._last_turn_time: datetime | None = None
    self._tool_call_count = 0

    # File path
    self._file_path = self._storage_path / f"{self._session_id}.jsonl"

    log.debug(
      "context_initialized",
      session_id=self._session_id,
      storage_path=str(self._storage_path),
    )

  def get_session_id(self) -> str:
    """Get the unique session identifier."""
    return self._session_id

  def add_message(
    self,
    role: str,
    content: str,
    metadata: dict[str, Any] | None = None,
  ) -> None:
    """Add a message to the context."""
    message: dict[str, Any] = {
      "role": role,
      "content": content,
    }
    if metadata:
      message["metadata"] = metadata

    self._sequence.append({"type": "message", "data": message})
    self._append_record("message", message)

  def add_tool_result(
    self,
    tool_name: str,
    tool_id: str,
    result: str,
    success: bool = True,
  ) -> None:
    """Add a tool execution result to the context."""
    tool_result: dict[str, Any] = {
      "tool_name": tool_name,
      "tool_id": tool_id,
      "result": result,
      "success": success,
    }

    self._sequence.append({"type": "tool_result", "data": tool_result})
    self._tool_call_count += 1
    self._append_record("tool_result", tool_result)

  def get_context(self) -> list[dict[str, Any]]:
    """Get the full context for backend submission.

    Returns messages and tool results in the correct order for the LLM API:
    - User messages
    - Assistant messages (with tool_calls)
    - Tool result messages (after the assistant message)
    """
    context: list[dict[str, Any]] = []

    for item in self._sequence:
      if item["type"] == "message":
        msg = item["data"]
        context.append({
          "role": msg["role"],
          "content": msg["content"],
        })
      elif item["type"] == "tool_result":
        tr = item["data"]
        context.append({
          "role": "tool",
          "name": tr["tool_name"],
          "content": tr["result"],
        })

    return context

  def get_messages(self) -> list[dict[str, Any]]:
    """Get all recorded messages (excludes tool results)."""
    messages: list[dict[str, Any]] = []
    for item in self._sequence:
      if item["type"] == "message":
        messages.append(item["data"])
    return messages

  def start_turn(self, user_message: str) -> None:
    """Start a new conversation turn."""
    turn_record: dict[str, Any] = {
      "user_message": user_message,
      "start_time": datetime.now().isoformat(),
    }
    self._sequence.append({"type": "turn", "data": turn_record})
    self._append_record("turn_start", turn_record)

    # Add user message
    self.add_message("user", user_message)

  def end_turn(self, assistant_message: str) -> None:
    """End the current conversation turn."""
    self._last_turn_time = datetime.now()

    # Add assistant message
    self.add_message("assistant", assistant_message)

    # Append turn end record
    self._append_record("turn_end", {
      "assistant_message": assistant_message,
    })

  def save(self) -> None:
    """Persist context to storage.

    Creates storage directory if needed and writes all records.
    """
    self._ensure_storage_directory()

    # Write session_start if file doesn't exist
    if not self._file_path.exists():
      self._write_session_start()

    # Flush any buffered writes
    self._flush_pending_records()

  def load(self) -> bool:
    """Load context from storage.

    Returns:
      True if context was loaded, False if no stored context exists.

    Raises:
      ContextCorruptionError: If stored context is corrupted.
    """
    if not self._file_path.exists():
      return False

    # Check if path is safe
    if not is_safe_path(self._storage_path, self._file_path):
      raise SessionNotFoundError(self._session_id)

    try:
      self._load_from_file()
      return True
    except json.JSONDecodeError as e:
      raise ContextCorruptionError(
        str(self._file_path),
        e.lineno or 0,
        f"Invalid JSON: {e.msg}",
      ) from None

  def clear(self) -> None:
    """Clear in-memory context (does not delete from storage)."""
    self._sequence.clear()
    self._tool_call_count = 0
    self._last_turn_time = None

  def delete(self) -> None:
    """Delete stored context from disk.

    Raises:
      SessionNotFoundError: If session doesn't exist.
    """
    if not self._file_path.exists():
      raise SessionNotFoundError(self._session_id)

    try:
      self._file_path.unlink()
      log.debug("context_deleted", path=str(self._file_path))
    except OSError as e:
      log.error("context_delete_failed", path=str(self._file_path), error=str(e))
      raise

  def get_statistics(self) -> ContextStatistics:
    """Get statistics about context usage."""
    # Count items in sequence
    message_count = sum(1 for item in self._sequence if item["type"] == "message")
    turn_count = sum(1 for item in self._sequence if item["type"] == "turn")

    return ContextStatistics(
      message_count=message_count,
      turn_count=turn_count,
      tool_call_count=self._tool_call_count,
      start_time=self._start_time,
      last_turn_time=self._last_turn_time,
    )

  def close(self) -> None:
    """Release resources and flush any pending writes."""
    self._append_record("session_end", {
      "end_time": datetime.now().isoformat(),
    })

  # Private methods

  def _ensure_storage_directory(self) -> None:
    """Create storage directory with secure permissions if it doesn't exist."""
    if not self._storage_path.exists():
      self._storage_path.mkdir(parents=True, mode=DIR_MODE)
      log.debug("storage_created", path=str(self._storage_path))
    else:
      # Ensure correct permissions
      try:
        self._storage_path.chmod(DIR_MODE)
      except OSError:
        pass  # Ignore permission errors on existing directories

  def _append_record(self, record_type: str, data: dict[str, Any]) -> None:
    """Append a record to the JSONL file.

    Uses atomic write for crash safety.

    Args:
      record_type: Type of record (session_start, message, etc.).
      data: Record data dictionary.
    """
    self._ensure_storage_directory()

    record = {
      "type": record_type,
      "timestamp": datetime.now().isoformat(),
      "data": data,
    }

    self._atomic_write_jsonl(record)

  def _atomic_write_jsonl(self, record: dict[str, Any]) -> None:
    """Write a record atomically to the JSONL file.

    Uses file locking for atomic appends with secure permissions.

    Args:
      record: The record dictionary to write.
    """
    import fcntl

    # Ensure storage directory exists
    self._ensure_storage_directory()

    # Create file if needed with secure permissions
    if not self._file_path.exists():
      self._file_path.touch(mode=FILE_MODE)
    else:
      # Ensure permissions on existing file
      try:
        self._file_path.chmod(FILE_MODE)
      except OSError:
        pass

    # Write with file locking for atomic append
    with open(self._file_path, "a") as f:
      # Acquire exclusive lock
      fcntl.flock(f.fileno(), fcntl.LOCK_EX)
      try:
        json.dump(record, f)
        f.write("\n")
        f.flush()
        os.fsync(f.fileno())
      finally:
        fcntl.flock(f.fileno(), fcntl.LOCK_UN)

  def _write_session_start(self) -> None:
    """Write session_start record."""
    record = {
      "type": "session_start",
      "timestamp": self._start_time.isoformat(),
      "data": {
        "session_id": self._session_id,
        "start_time": self._start_time.isoformat(),
      },
    }
    self._atomic_write_jsonl(record)

  def _flush_pending_records(self) -> None:
    """Flush any pending writes to disk.

    For JSONL files, this is a no-op since records are written immediately.
    """
    # Sync file to disk
    if self._file_path.exists():
      with open(self._file_path, "a") as f:
        os.fsync(f.fileno())

  def _load_from_file(self) -> None:
    """Load context from JSONL file.

    Parses all records and reconstructs in-memory state.

    Raises:
      ContextCorruptionError: If file is corrupted.
    """
    self.clear()

    line_num = 0
    try:
      with open(self._file_path) as f:
        for line_num, line in enumerate(f, start=1):
          line = line.strip()
          if not line:
            continue

          record = json.loads(line)
          self._process_record(record, line_num)

    except json.JSONDecodeError as e:
      raise ContextCorruptionError(
        str(self._file_path),
        line_num,
        f"Invalid JSON: {e.msg}",
      ) from None

  def _process_record(self, record: dict[str, Any], line_num: int) -> None:
    """Process a single JSONL record.

    Args:
      record: The parsed record dictionary.
      line_num: Line number for error messages.

    Raises:
      ContextCorruptionError: If record is malformed.
    """
    if "type" not in record:
      raise ContextCorruptionError(
        str(self._file_path),
        line_num,
        "Missing record type",
      )

    record_type = record.get("type")
    data = record.get("data", {})

    if record_type == "session_start":
      # Already handled during initialization
      pass

    elif record_type == "message":
      if "role" not in data or "content" not in data:
        raise ContextCorruptionError(
          str(self._file_path),
          line_num,
          "Missing message fields",
        )
      self._sequence.append({"type": "message", "data": data})

    elif record_type == "tool_result":
      if "tool_id" not in data:
        raise ContextCorruptionError(
          str(self._file_path),
          line_num,
          "Missing tool_id",
        )
      self._sequence.append({"type": "tool_result", "data": data})
      self._tool_call_count += 1

    elif record_type == "turn_start":
      self._sequence.append({"type": "turn", "data": data})

    elif record_type == "turn_end":
      # Turn end - nothing special to do, statistics will be computed
      pass

    elif record_type == "session_end":
      # Session ended, nothing special to do
      pass

    else:
      log.warning(
        "unknown_record_type",
        record_type=record_type,
        line=line_num,
      )


__all__ = [
  "BasicPersistenceContextManager",
]

1	"""Basic persistence context manager implementation.
2
3	Provides JSONL-based context persistence with secure file handling.
4
5	Note:
6	File locking uses fcntl (Unix-only). On Windows, file writes
7	are still atomic but lack inter-process locking protection.
8	For production use on Windows, consider adding a cross-platform
9	file locking library.
10	"""
11
12	import json	1✔
13	import os	1✔
14	from datetime import datetime	1✔
15	from pathlib import Path	1✔
16	from typing import Any	1✔
17
18	from yoker.context.interface import ContextStatistics	1✔
19	from yoker.context.validator import is_safe_path, validate_session_id, validate_storage_path	1✔
20	from yoker.exceptions import ContextCorruptionError, SessionNotFoundError	1✔
21	from yoker.logging import get_logger	1✔
22
23	log = get_logger(__name__)	1✔
24
25	# File permissions
26	DIR_MODE = 0o700 # Owner-only for directories	1✔
27	FILE_MODE = 0o600 # Owner-only for files	1✔
28
29
30	class BasicPersistenceContextManager:	1✔
31	"""Context manager with JSONL persistence.
32
33	Stores conversation history in JSONL (JSON Lines) format with:
34	- Atomic writes for crash safety
35	- Secure file permissions
36	- Session lifecycle tracking
37
38	Record types:
39	- session_start: Session metadata
40	- message: User/assistant/system message
41	- tool_result: Tool execution result
42	- turn: Turn boundary marker
43	- session_end: Session termination marker
44	"""
45
46	def __init__(	1✔
47	self,
48	storage_path: Path \| str,
49	session_id: str = "auto",
50	) -> None:
51	"""Initialize context manager.
52
53	Args:
54	storage_path: Directory for storing context files.
55	session_id: Session ID or "auto" to generate.
56
57	Raises:
58	ValidationError: If storage_path or session_id is invalid.
59	"""
60	# Validate and resolve storage path
61	self._storage_path = validate_storage_path(	1✔
62	Path(storage_path), "context.storage_path"
63	)
64
65	# Validate session ID
66	self._session_id = validate_session_id(session_id, "context.session_id")	1✔
67
68	# In-memory context: ordered sequence of all items
69	# Each item is {"type": "message"\|"tool_result", "data": {...}}
70	self._sequence: list[dict[str, Any]] = []	1✔
71
72	# Statistics
73	self._start_time = datetime.now()	1✔
74	self._last_turn_time: datetime \| None = None	1✔
75	self._tool_call_count = 0	1✔
76
77	# File path
78	self._file_path = self._storage_path / f"{self._session_id}.jsonl"	1✔
79
80	log.debug(	1✔
81	"context_initialized",
82	session_id=self._session_id,
83	storage_path=str(self._storage_path),
84	)
85
86	def get_session_id(self) -> str:	1✔
87	"""Get the unique session identifier."""
88	return self._session_id	1✔
89
90	def add_message(	1✔
91	self,
92	role: str,
93	content: str,
94	metadata: dict[str, Any] \| None = None,
95	) -> None:
96	"""Add a message to the context."""
97	message: dict[str, Any] = {	1✔
98	"role": role,
99	"content": content,
100	}
101	if metadata:	1✔
102	message["metadata"] = metadata	×
103
104	self._sequence.append({"type": "message", "data": message})	1✔
105	self._append_record("message", message)	1✔
106
107	def add_tool_result(	1✔
108	self,
109	tool_name: str,
110	tool_id: str,
111	result: str,
112	success: bool = True,
113	) -> None:
114	"""Add a tool execution result to the context."""
115	tool_result: dict[str, Any] = {	1✔
116	"tool_name": tool_name,
117	"tool_id": tool_id,
118	"result": result,
119	"success": success,
120	}
121
122	self._sequence.append({"type": "tool_result", "data": tool_result})	1✔
123	self._tool_call_count += 1	1✔
124	self._append_record("tool_result", tool_result)	1✔
125
126	def get_context(self) -> list[dict[str, Any]]:	1✔
127	"""Get the full context for backend submission.
128
129	Returns messages and tool results in the correct order for the LLM API:
130	- User messages
131	- Assistant messages (with tool_calls)
132	- Tool result messages (after the assistant message)
133	"""
134	context: list[dict[str, Any]] = []	1✔
135
136	for item in self._sequence:	1✔
137	if item["type"] == "message":	1✔
138	msg = item["data"]	1✔
139	context.append({	1✔
140	"role": msg["role"],
141	"content": msg["content"],
142	})
143	elif item["type"] == "tool_result":	1✔
144	tr = item["data"]	1✔
145	context.append({	1✔
146	"role": "tool",
147	"name": tr["tool_name"],
148	"content": tr["result"],
149	})
150
151	return context	1✔
152
153	def get_messages(self) -> list[dict[str, Any]]:	1✔
154	"""Get all recorded messages (excludes tool results)."""
155	messages: list[dict[str, Any]] = []	1✔
156	for item in self._sequence:	1✔
157	if item["type"] == "message":	1✔
158	messages.append(item["data"])	1✔
159	return messages	1✔
160
161	def start_turn(self, user_message: str) -> None:	1✔
162	"""Start a new conversation turn."""
163	turn_record: dict[str, Any] = {	1✔
164	"user_message": user_message,
165	"start_time": datetime.now().isoformat(),
166	}
167	self._sequence.append({"type": "turn", "data": turn_record})	1✔
168	self._append_record("turn_start", turn_record)	1✔
169
170	# Add user message
171	self.add_message("user", user_message)	1✔
172
173	def end_turn(self, assistant_message: str) -> None:	1✔
174	"""End the current conversation turn."""
175	self._last_turn_time = datetime.now()	1✔
176
177	# Add assistant message
178	self.add_message("assistant", assistant_message)	1✔
179
180	# Append turn end record
181	self._append_record("turn_end", {	1✔
182	"assistant_message": assistant_message,
183	})
184
185	def save(self) -> None:	1✔
186	"""Persist context to storage.
187
188	Creates storage directory if needed and writes all records.
189	"""
190	self._ensure_storage_directory()	1✔
191
192	# Write session_start if file doesn't exist
193	if not self._file_path.exists():	1✔
194	self._write_session_start()	1✔
195
196	# Flush any buffered writes
197	self._flush_pending_records()	1✔
198
199	def load(self) -> bool:	1✔
200	"""Load context from storage.
201
202	Returns:
203	True if context was loaded, False if no stored context exists.
204
205	Raises:
206	ContextCorruptionError: If stored context is corrupted.
207	"""
208	if not self._file_path.exists():	1✔
209	return False	1✔
210
211	# Check if path is safe
212	if not is_safe_path(self._storage_path, self._file_path):	1✔
213	raise SessionNotFoundError(self._session_id)	×
214
215	try:	1✔
216	self._load_from_file()	1✔
217	return True	1✔
218	except json.JSONDecodeError as e:	1✔
219	raise ContextCorruptionError(	×
220	str(self._file_path),
221	e.lineno or 0,
222	f"Invalid JSON: {e.msg}",
223	) from None
224
225	def clear(self) -> None:	1✔
226	"""Clear in-memory context (does not delete from storage)."""
227	self._sequence.clear()	1✔
228	self._tool_call_count = 0	1✔
229	self._last_turn_time = None	1✔
230
231	def delete(self) -> None:	1✔
232	"""Delete stored context from disk.
233
234	Raises:
235	SessionNotFoundError: If session doesn't exist.
236	"""
237	if not self._file_path.exists():	1✔
238	raise SessionNotFoundError(self._session_id)	1✔
239
240	try:	1✔
241	self._file_path.unlink()	1✔
242	log.debug("context_deleted", path=str(self._file_path))	1✔
243	except OSError as e:	×
244	log.error("context_delete_failed", path=str(self._file_path), error=str(e))	×
245	raise	×
246
247	def get_statistics(self) -> ContextStatistics:	1✔
248	"""Get statistics about context usage."""
249	# Count items in sequence
250	message_count = sum(1 for item in self._sequence if item["type"] == "message")	1✔
251	turn_count = sum(1 for item in self._sequence if item["type"] == "turn")	1✔
252
253	return ContextStatistics(	1✔
254	message_count=message_count,
255	turn_count=turn_count,
256	tool_call_count=self._tool_call_count,
257	start_time=self._start_time,
258	last_turn_time=self._last_turn_time,
259	)
260
261	def close(self) -> None:	1✔
262	"""Release resources and flush any pending writes."""
263	self._append_record("session_end", {	1✔
264	"end_time": datetime.now().isoformat(),
265	})
266
267	# Private methods
268
269	def _ensure_storage_directory(self) -> None:	1✔
270	"""Create storage directory with secure permissions if it doesn't exist."""
271	if not self._storage_path.exists():	1✔
272	self._storage_path.mkdir(parents=True, mode=DIR_MODE)	1✔
273	log.debug("storage_created", path=str(self._storage_path))	1✔
274	else:
275	# Ensure correct permissions
276	try:	1✔
277	self._storage_path.chmod(DIR_MODE)	1✔
278	except OSError:	×
279	pass # Ignore permission errors on existing directories	×
280
281	def _append_record(self, record_type: str, data: dict[str, Any]) -> None:	1✔
282	"""Append a record to the JSONL file.
283
284	Uses atomic write for crash safety.
285
286	Args:
287	record_type: Type of record (session_start, message, etc.).
288	data: Record data dictionary.
289	"""
290	self._ensure_storage_directory()	1✔
291
292	record = {	1✔
293	"type": record_type,
294	"timestamp": datetime.now().isoformat(),
295	"data": data,
296	}
297
298	self._atomic_write_jsonl(record)	1✔
299
300	def _atomic_write_jsonl(self, record: dict[str, Any]) -> None:	1✔
301	"""Write a record atomically to the JSONL file.
302
303	Uses file locking for atomic appends with secure permissions.
304
305	Args:
306	record: The record dictionary to write.
307	"""
308	import fcntl	1✔
309
310	# Ensure storage directory exists
311	self._ensure_storage_directory()	1✔
312
313	# Create file if needed with secure permissions
314	if not self._file_path.exists():	1✔
315	self._file_path.touch(mode=FILE_MODE)	1✔
316	else:
317	# Ensure permissions on existing file
318	try:	1✔
319	self._file_path.chmod(FILE_MODE)	1✔
320	except OSError:	×
321	pass	×
322
323	# Write with file locking for atomic append
324	with open(self._file_path, "a") as f:	1✔
325	# Acquire exclusive lock
326	fcntl.flock(f.fileno(), fcntl.LOCK_EX)	1✔
327	try:	1✔
328	json.dump(record, f)	1✔
329	f.write("\n")	1✔
330	f.flush()	1✔
331	os.fsync(f.fileno())	1✔
332	finally:
333	fcntl.flock(f.fileno(), fcntl.LOCK_UN)	1✔
334
335	def _write_session_start(self) -> None:	1✔
336	"""Write session_start record."""
337	record = {	1✔
338	"type": "session_start",
339	"timestamp": self._start_time.isoformat(),
340	"data": {
341	"session_id": self._session_id,
342	"start_time": self._start_time.isoformat(),
343	},
344	}
345	self._atomic_write_jsonl(record)	1✔
346
347	def _flush_pending_records(self) -> None:	1✔
348	"""Flush any pending writes to disk.
349
350	For JSONL files, this is a no-op since records are written immediately.
351	"""
352	# Sync file to disk
353	if self._file_path.exists():	1✔
354	with open(self._file_path, "a") as f:	1✔
355	os.fsync(f.fileno())	1✔
356
357	def _load_from_file(self) -> None:	1✔
358	"""Load context from JSONL file.
359
360	Parses all records and reconstructs in-memory state.
361
362	Raises:
363	ContextCorruptionError: If file is corrupted.
364	"""
365	self.clear()	1✔
366
367	line_num = 0	1✔
368	try:	1✔
369	with open(self._file_path) as f:	1✔
370	for line_num, line in enumerate(f, start=1):	1✔
371	line = line.strip()	1✔
372	if not line:	1✔
373	continue	×
374
375	record = json.loads(line)	1✔
376	self._process_record(record, line_num)	1✔
377
378	except json.JSONDecodeError as e:	1✔
379	raise ContextCorruptionError(	1✔
380	str(self._file_path),
381	line_num,
382	f"Invalid JSON: {e.msg}",
383	) from None
384
385	def _process_record(self, record: dict[str, Any], line_num: int) -> None:	1✔
386	"""Process a single JSONL record.
387
388	Args:
389	record: The parsed record dictionary.
390	line_num: Line number for error messages.
391
392	Raises:
393	ContextCorruptionError: If record is malformed.
394	"""
395	if "type" not in record:	1✔
396	raise ContextCorruptionError(	×
397	str(self._file_path),
398	line_num,
399	"Missing record type",
400	)
401
402	record_type = record.get("type")	1✔
403	data = record.get("data", {})	1✔
404
405	if record_type == "session_start":	1✔
406	# Already handled during initialization
407	pass	1✔
408
409	elif record_type == "message":	1✔
410	if "role" not in data or "content" not in data:	1✔
411	raise ContextCorruptionError(	×
412	str(self._file_path),
413	line_num,
414	"Missing message fields",
415	)
416	self._sequence.append({"type": "message", "data": data})	1✔
417
418	elif record_type == "tool_result":	1✔
419	if "tool_id" not in data:	×
420	raise ContextCorruptionError(	×
421	str(self._file_path),
422	line_num,
423	"Missing tool_id",
424	)
425	self._sequence.append({"type": "tool_result", "data": data})	×
426	self._tool_call_count += 1	×
427
428	elif record_type == "turn_start":	1✔
429	self._sequence.append({"type": "turn", "data": data})	×
430
431	elif record_type == "turn_end":	1✔
432	# Turn end - nothing special to do, statistics will be computed
433	pass	×
434
435	elif record_type == "session_end":	1✔
436	# Session ended, nothing special to do
437	pass	1✔
438
439	else:
440	log.warning(	×
441	"unknown_record_type",
442	record_type=record_type,
443	line=line_num,
444	)
445
446
447	__all__ = [	1✔
448	"BasicPersistenceContextManager",
449	]

christophevg / yoker / 24802435632

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous