@@ -30,6 +30,7 @@
 from opentelemetry.sdk.trace.export import SpanExporter
 from opentelemetry.sdk.trace.export import SpanExportResult
 from opentelemetry.trace import SpanContext
+from opentelemetry.trace import SpanKind
 from opentelemetry.trace import TraceFlags
 from opentelemetry.trace import TraceState
 from opentelemetry.util.types import AttributeValue
@@ -58,6 +59,15 @@
 CREATE INDEX IF NOT EXISTS spans_trace_id_idx ON spans(trace_id);
 """
 
+# Ordered schema migrations: entry 0 creates the original schema and each later
+# entry upgrades the one before it. `PRAGMA user_version` records how many have
+# been applied, so opening a database runs only the ones it lacks. Append new
+# entries; never edit one that has shipped.
+_MIGRATIONS: tuple[tuple[str, ...], ...] = (
+    (_CREATE_SPANS_TABLE, _CREATE_SESSION_INDEX, _CREATE_TRACE_INDEX),
+    ("ALTER TABLE spans ADD COLUMN kind TEXT NOT NULL DEFAULT 'INTERNAL'",),
+)
+
 _INSERT_SPAN = """
 INSERT OR REPLACE INTO spans (
   span_id,
@@ -68,18 +78,45 @@
   end_time_unix_nano,
   session_id,
   invocation_id,
-  attributes_json
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?);
+  attributes_json,
+  kind
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
 """
 
 _DEFAULT_TIMEOUT_SECONDS = 30.0
 
 
+def _apply_migrations(
+    conn: sqlite3.Connection, migrations: Sequence[Sequence[str]]
+) -> None:
+  """Applies the entries of `migrations` the database has not run yet."""
+  # Checked outside any transaction first, so opening a current database never
+  # waits on another connection's write lock.
+  if conn.execute("PRAGMA user_version").fetchone()[0] >= len(migrations):
+    return
+  with conn:
+    # CPython opens no transaction for DDL, so begin one explicitly. It
+    # takes the write lock before the version is read, so two processes
+    # opening the same file cannot both apply a migration.
+    conn.execute("BEGIN IMMEDIATE")
+    version = conn.execute("PRAGMA user_version").fetchone()[0]
+    if version >= len(migrations):
+      return
+    for statements in migrations[version:]:
+      for statement in statements:
+        conn.execute(statement)
+    # PRAGMA takes no bind parameters; the value is an int length.
+    conn.execute(f"PRAGMA user_version = {len(migrations)}")
+
+
 class SqliteSpanExporter(SpanExporter):
   """Exports spans to a local SQLite database.
 
   This is intended for local development (e.g. `adk web`) to allow reloading
   traces for older sessions after process restart.
+
+  Span kinds survive export and reload. Existing databases are upgraded in
+  place; spans saved before kind persistence retain the INTERNAL default.
   """
 
   def __init__(self, *, db_path: str):
@@ -100,11 +137,7 @@ def _get_connection(self) -> sqlite3.Connection:
 
   def _ensure_schema(self) -> None:
     with self._lock:
-      conn = self._get_connection()
-      conn.execute(_CREATE_SPANS_TABLE)
-      conn.execute(_CREATE_SESSION_INDEX)
-      conn.execute(_CREATE_TRACE_INDEX)
-      conn.commit()
+      _apply_migrations(self._get_connection(), _MIGRATIONS)
 
   def _serialize_attributes(
       self, attributes: Mapping[str, AttributeValue]
@@ -167,6 +200,7 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
               session_id,
               invocation_id,
               self._serialize_attributes(attributes),
+              span.kind.name,
           ))
         conn.executemany(_INSERT_SPAN, rows)
         conn.commit()
@@ -221,6 +255,7 @@ def _row_to_readable_span(self, row: sqlite3.Row) -> ReadableSpan:
         name=row["name"] or "",
         context=context,
         parent=parent,
+        kind=SpanKind[row["kind"]],
         attributes=attributes,
         start_time=row["start_time_unix_nano"],
         end_time=row["end_time_unix_nano"],
 
@@ -16,13 +16,17 @@
 
 import json
 from pathlib import Path
+import sqlite3
 
+from google.adk.telemetry.sqlite_span_exporter import _apply_migrations
 from google.adk.telemetry.sqlite_span_exporter import SqliteSpanExporter
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExportResult
 from opentelemetry.trace import SpanContext
+from opentelemetry.trace import SpanKind
 from opentelemetry.trace import TraceFlags
 from opentelemetry.trace import TraceState
+import pytest
 
 
 def _create_span(
@@ -32,6 +36,7 @@ def _create_span(
     parent_span_id: int | None = None,
     name: str = "test_span",
     attributes: dict | None = None,
+    kind: SpanKind = SpanKind.INTERNAL,
     start_time: int = 1000,
     end_time: int = 2000,
 ) -> ReadableSpan:
@@ -58,6 +63,7 @@ def _create_span(
       name=name,
       context=context,
       parent=parent,
+      kind=kind,
       attributes=attributes or {},
       start_time=start_time,
       end_time=end_time,
@@ -227,6 +233,185 @@ def test_round_trip_preserves_span_attributes(tmp_path):
   assert retrieved.attributes == original_attributes
 
 
+def test_span_kinds_survive_exporter_restart(tmp_path: Path) -> None:
+  db_path = tmp_path / "test.db"
+  kinds = [
+      SpanKind.CLIENT,
+      SpanKind.SERVER,
+      SpanKind.PRODUCER,
+      SpanKind.CONSUMER,
+  ]
+  exporter = SqliteSpanExporter(db_path=str(db_path))
+  spans = [
+      _create_span(
+          span_id=index + 1,
+          kind=kind,
+          attributes={"gen_ai.conversation.id": "session-kind"},
+          start_time=index,
+      )
+      for index, kind in enumerate(kinds)
+  ]
+  assert exporter.export(spans) == SpanExportResult.SUCCESS
+  exporter.shutdown()
+
+  reopened = SqliteSpanExporter(db_path=str(db_path))
+  try:
+    restored = reopened.get_all_spans_for_session("session-kind")
+    assert [(span.context.span_id, span.kind) for span in restored] == [
+        (index + 1, kind) for index, kind in enumerate(kinds)
+    ]
+  finally:
+    reopened.shutdown()
+
+
+def test_legacy_database_retains_spans_and_accepts_new_kinds(
+    tmp_path: Path,
+) -> None:
+  db_path = tmp_path / "legacy.db"
+  with sqlite3.connect(db_path) as conn:
+    conn.execute("""
+        CREATE TABLE spans (
+          span_id TEXT PRIMARY KEY, trace_id TEXT NOT NULL,
+          parent_span_id TEXT, name TEXT NOT NULL,
+          start_time_unix_nano INTEGER, end_time_unix_nano INTEGER,
+          session_id TEXT, invocation_id TEXT, attributes_json TEXT
+        )
+    """)
+    conn.execute(
+        "INSERT INTO spans VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            "0000000000000001",
+            "000000000000000000000000000def45",
+            None,
+            "legacy_operation",
+            1000,
+            2000,
+            "session-legacy",
+            None,
+            '{"gen_ai.conversation.id": "session-legacy"}',
+        ),
+    )
+
+  exporter = SqliteSpanExporter(db_path=str(db_path))
+  try:
+    assert (
+        exporter.export([
+            _create_span(
+                span_id=2,
+                name="new_operation",
+                kind=SpanKind.CLIENT,
+                attributes={"gen_ai.conversation.id": "session-legacy"},
+                start_time=3000,
+            )
+        ])
+        == SpanExportResult.SUCCESS
+    )
+    restored = exporter.get_all_spans_for_session("session-legacy")
+    assert [(span.name, span.kind) for span in restored] == [
+        ("legacy_operation", SpanKind.INTERNAL),
+        ("new_operation", SpanKind.CLIENT),
+    ]
+  finally:
+    exporter.shutdown()
+
+
+def test_database_from_newer_schema_version_is_not_downgraded(
+    tmp_path: Path,
+) -> None:
+  db_path = tmp_path / "newer.db"
+  SqliteSpanExporter(db_path=str(db_path)).shutdown()
+  with sqlite3.connect(db_path) as conn:
+    conn.execute("PRAGMA user_version = 99")
+
+  SqliteSpanExporter(db_path=str(db_path)).shutdown()
+
+  with sqlite3.connect(db_path) as conn:
+    assert conn.execute("PRAGMA user_version").fetchone()[0] == 99
+
+
+def _column_names(conn: sqlite3.Connection, table: str) -> list[str]:
+  return [row[1] for row in conn.execute(f"PRAGMA table_info({table})")]
+
+
+def test_failing_migration_rolls_back_and_can_be_retried(
+    tmp_path: Path,
+) -> None:
+  conn = sqlite3.connect(tmp_path / "migrations.db")
+  try:
+    create = ("CREATE TABLE t (a TEXT)",)
+    _apply_migrations(conn, [create])
+
+    with pytest.raises(sqlite3.OperationalError):
+      _apply_migrations(
+          conn,
+          [
+              create,
+              ("ALTER TABLE t ADD COLUMN b TEXT", "SELECT * FROM missing"),
+          ],
+      )
+
+    assert conn.execute("PRAGMA user_version").fetchone()[0] == 1
+    assert _column_names(conn, "t") == ["a"]
+
+    _apply_migrations(conn, [create, ("ALTER TABLE t ADD COLUMN b TEXT",)])
+
+    assert conn.execute("PRAGMA user_version").fetchone()[0] == 2
+    assert _column_names(conn, "t") == ["a", "b"]
+  finally:
+    conn.close()
+
+
+def test_apply_migrations_skips_entries_already_applied(tmp_path: Path) -> None:
+  conn = sqlite3.connect(tmp_path / "migrations.db")
+  try:
+    conn.execute("CREATE TABLE t (a TEXT)")
+    conn.execute("PRAGMA user_version = 1")
+
+    # Entry 0 would fail if re-run because the table already exists.
+    _apply_migrations(
+        conn,
+        [("CREATE TABLE t (a TEXT)",), ("ALTER TABLE t ADD COLUMN b TEXT",)],
+    )
+
+    assert conn.execute("PRAGMA user_version").fetchone()[0] == 2
+    assert _column_names(conn, "t") == ["a", "b"]
+  finally:
+    conn.close()
+
+
+def test_current_database_opens_while_another_writer_holds_lock(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+  db_path = tmp_path / "locked.db"
+  exporter = SqliteSpanExporter(db_path=str(db_path))
+  assert (
+      exporter.export(
+          [_create_span(attributes={"gen_ai.conversation.id": "session-lock"})]
+      )
+      == SpanExportResult.SUCCESS
+  )
+  exporter.shutdown()
+  monkeypatch.setattr(
+      "google.adk.telemetry.sqlite_span_exporter._DEFAULT_TIMEOUT_SECONDS",
+      0.1,
+  )
+
+  writer = sqlite3.connect(db_path, isolation_level=None)
+  try:
+    writer.execute("BEGIN IMMEDIATE")
+    writer.execute("UPDATE spans SET name = 'uncommitted'")
+
+    reopened = SqliteSpanExporter(db_path=str(db_path))
+    try:
+      restored = reopened.get_all_spans_for_session("session-lock")
+      assert [span.name for span in restored] == ["test_span"]
+    finally:
+      reopened.shutdown()
+  finally:
+    writer.rollback()
+    writer.close()
+
+
 def test_spans_with_parent_context_exported_correctly(tmp_path):
   db_path = tmp_path / "test.db"
   exporter = SqliteSpanExporter(db_path=str(db_path))