diff --git a/src/google/adk/sessions/schemas/v0.py b/src/google/adk/sessions/schemas/v0.py
index 7679a56e5b..79e7bdc61b 100644
--- a/src/google/adk/sessions/schemas/v0.py
+++ b/src/google/adk/sessions/schemas/v0.py
@@ -29,10 +29,13 @@
 from datetime import datetime
 from datetime import timezone
 import json
+import logging
 import pickle
 from typing import Any
 from typing import Optional
 
+logger = logging.getLogger("google_adk." + __name__)
+
 from google.adk.platform import uuid as platform_uuid
 from google.genai import types
 from sqlalchemy import Boolean
@@ -61,6 +64,13 @@
 from .shared import DynamicJSON
 from .shared import PreciseTimestamp
 
+# Legacy v0 databases may retain error_message as VARCHAR(255) from earlier
+# schema versions. create_all() is additive only, so existing columns are never
+# altered. Truncate on write to avoid StringDataRightTruncationError on those
+# databases.
+_LEGACY_ERROR_MSG_MAX_LEN = 255
+_TRUNCATION_SUFFIX = "...[truncated]"
+
 
 class DynamicPickleType(TypeDecorator):
   """Represents a type that can be pickled."""
@@ -284,6 +294,19 @@ def long_running_tool_ids(self, value: set[str]):
 
   @classmethod
   def from_event(cls, session: Session, event: Event) -> StorageEvent:
+    error_message = event.error_message
+    if error_message and len(error_message) > _LEGACY_ERROR_MSG_MAX_LEN:
+      logger.warning(
+          "error_message for event %s exceeds %d characters and will be"
+          " truncated. Full message: %s",
+          event.id,
+          _LEGACY_ERROR_MSG_MAX_LEN,
+          error_message,
+      )
+      error_message = (
+          error_message[: _LEGACY_ERROR_MSG_MAX_LEN - len(_TRUNCATION_SUFFIX)]
+          + _TRUNCATION_SUFFIX
+      )
     storage_event = StorageEvent(
         id=event.id,
         invocation_id=event.invocation_id,
@@ -298,7 +321,7 @@ def from_event(cls, session: Session, event: Event) -> StorageEvent:
         partial=event.partial,
         turn_complete=event.turn_complete,
         error_code=event.error_code,
-        error_message=event.error_message,
+        error_message=error_message,
         interrupted=event.interrupted,
     )
     if event.content:
diff --git a/tests/unittests/sessions/test_v0_storage_event.py b/tests/unittests/sessions/test_v0_storage_event.py
index cc82272506..5889afd50b 100644
--- a/tests/unittests/sessions/test_v0_storage_event.py
+++ b/tests/unittests/sessions/test_v0_storage_event.py
@@ -15,9 +15,11 @@
 from datetime import datetime
 from datetime import timezone
 
+from google.adk.events.event import Event
 from google.adk.events.event_actions import EventActions
 from google.adk.events.event_actions import EventCompaction
 from google.adk.sessions.schemas.v0 import StorageEvent
+from google.adk.sessions.session import Session
 from google.genai import types
 
 
@@ -48,3 +50,55 @@ def test_storage_event_v0_to_event_rehydrates_compaction_model():
   assert isinstance(event.actions.compaction, EventCompaction)
   assert event.actions.compaction.start_timestamp == 1.0
   assert event.actions.compaction.end_timestamp == 2.0
+
+
+def test_from_event_truncates_error_message_exceeding_varchar255(caplog):
+  import logging
+
+  session = Session(app_name="app", user_id="user", id="session_id")
+  event = Event(
+      id="event_id",
+      invocation_id="invocation_id",
+      author="author",
+      timestamp=1.0,
+      error_code="MALFORMED_FUNCTION_CALL",
+      error_message="x" * 300,
+  )
+
+  with caplog.at_level(logging.WARNING):
+    storage_event = StorageEvent.from_event(session, event)
+
+  assert storage_event.error_message == "x" * 241 + "...[truncated]"
+  assert any("truncated" in r.message for r in caplog.records)
+  assert any("x" * 300 in r.message for r in caplog.records)
+
+
+def test_from_event_preserves_short_error_message():
+  session = Session(app_name="app", user_id="user", id="session_id")
+  short_message = "Malformed function call"
+  event = Event(
+      id="event_id",
+      invocation_id="invocation_id",
+      author="author",
+      timestamp=1.0,
+      error_code="MALFORMED_FUNCTION_CALL",
+      error_message=short_message,
+  )
+
+  storage_event = StorageEvent.from_event(session, event)
+
+  assert storage_event.error_message == short_message
+
+
+def test_from_event_with_none_error_message():
+  session = Session(app_name="app", user_id="user", id="session_id")
+  event = Event(
+      id="event_id",
+      invocation_id="invocation_id",
+      author="author",
+      timestamp=1.0,
+  )
+
+  storage_event = StorageEvent.from_event(session, event)
+
+  assert storage_event.error_message is None