Run sequential appends

For the complete documentation index see: llms.txt

All documentation pages available in markdown.

The demo script is adapted from the atomic session append example in the adk-aerospike repository. Open that source if you want to see how each append_event call builds events and state_delta values before you run it for yourself.

The script walks through sequential append_event calls with scoped state, then runs concurrent writers. Here you run and review the sequential phase only.

The script also defines WRITERS, WRITES_PER_WRITER, and MAX_IN_FLIGHT for the concurrent phase. MAX_IN_FLIGHT uses a semaphore (a concurrency limiter) to cap how many append_event calls run at once. See Run concurrent writers for more information on the constants and the semaphore.

Save and run the demo script

Create atomic-session-append-demo.py with the script in this section (or copy from the adk-aerospike example):

import asyncio
import time

import aerospike
from google.adk.events import Event, EventActions
from google.genai import types as genai_types

from adk_aerospike import AerospikeSessionService

APP, USER, SID = "demo-app", "user-1", "session-1"
WRITERS = 64
WRITES_PER_WRITER = 1000
MAX_IN_FLIGHT = 4

sem = asyncio.Semaphore(MAX_IN_FLIGHT)


def print_session_record(client: aerospike.Client, label: str) -> None:
    pk = ("test", "adk_sessions", f"{APP}:{USER}:{SID}")
    _, _, bins = client.select(pk, ["cur", "state"])
    print(
        f"{label} — cur_segment={bins.get('cur')}, "
        f"session_state_keys={len(bins.get('state') or {})}"
    )


async def main() -> None:
    client = aerospike.client(
        {
            "hosts": [("127.0.0.1", 3000)],
            "max_error_rate": 100,
            "error_rate_window": 1,
        }
    ).connect()
    session_service = AerospikeSessionService(client, "test")

    session = await session_service.create_session(
        app_name=APP, user_id=USER, session_id=SID
    )

    # --- sequential appends: scoped state deltas accumulate ---
    for i in range(5):
        await session_service.append_event(
            session,
            Event(
                invocation_id=f"seq-{i}",
                author="assistant",
                content=genai_types.Content(
                    role="model",
                    parts=[genai_types.Part(text=f"assistant reply {i}")],
                ),
                actions=EventActions(
                    state_delta={
                        "turn": i,
                        "app:tenant": "acme",
                        "user:locale": "en-US",
                    }
                ),
            ),
        )
        snap = await session_service.get_session(
            app_name=APP, user_id=USER, session_id=SID
        )
        print(
            f"after append {i}: events={len(snap.events)}, "
            f"state={snap.state}"
        )

    print_session_record(client, "after 5 sequential appends")

    # --- concurrent load: 64 writers × 1000 appends each ---
    async def writer(writer_id: int) -> None:
        for i in range(WRITES_PER_WRITER):
            async with sem:
                await session_service.append_event(
                    session,
                    Event(
                        invocation_id=f"w{writer_id:02d}-{i:04d}",
                        author=f"worker-{writer_id}",
                        timestamp=time.time(),
                        content=genai_types.Content(
                            role="user",
                            parts=[
                                genai_types.Part(
                                    text=f"worker {worker_id} write {i}"
                                )
                            ],
                        ),
                    ),
                )
            session.events.clear()

    started = time.perf_counter()
    await asyncio.gather(*(writer(w) for w in range(WRITERS)))
    elapsed = time.perf_counter() - started

    expected = 5 + WRITERS * WRITES_PER_WRITER
    final = await session_service.get_session(
        app_name=APP, user_id=USER, session_id=SID
    )
    print(
        f"\n{WRITERS} writers × {WRITES_PER_WRITER} appends "
        f"(max {MAX_IN_FLIGHT} in flight) in {elapsed:.1f}s — "
        f"stored {len(final.events)} events (expected {expected})"
    )
    print_session_record(client, "after concurrent appends")

    await session_service.delete_session(app_name=APP, user_id=USER, session_id=SID)
    session_service.close()
    print("Connection closed.")


if __name__ == "__main__":
    asyncio.run(main())

Run the script:

python atomic-session-append-demo.py

Expected result

after append 0: events=1, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 0}
after append 1: events=2, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 1}
after append 2: events=3, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 2}
after append 3: events=4, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 3}
after append 4: events=5, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 4}
after 5 sequential appends — cur_segment=0, session_state_keys=1

The script continues into the concurrent phase after these lines. Keep the terminal open. On the Run concurrent writers page, you review the concurrent output from this run.

The sequential phase starts after the script connects and creates one session:

session = await session_service.create_session(
    app_name=APP, user_id=USER, session_id=SID
)

Review the sequential phase output

Read the per-append lines:

The script loops five times. Each iteration appends one event with a multi-scope state_delta, then reads the session back and prints a snapshot:

for i in range(5):
    await session_service.append_event(
        session,
        Event(
            invocation_id=f"seq-{i}",
            author="assistant",
            content=genai_types.Content(
                role="model",
                parts=[genai_types.Part(text=f"assistant reply {i}")],
            ),
            actions=EventActions(
                state_delta={
                    "turn": i,                  # session-scoped
                    "app:tenant": "acme",       # app-scoped
                    "user:locale": "en-US",     # user-scoped
                }
            ),
        ),
    )
    snap = await session_service.get_session(
        app_name=APP, user_id=USER, session_id=SID
    )
    print(
        f"after append {i}: events={len(snap.events)}, "
        f"state={snap.state}"
    )

after append 0: events=1, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 0}
after append 1: events=2, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 1}
after append 2: events=3, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 2}
after append 3: events=4, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 3}
after append 4: events=5, state={'app:tenant': 'acme', 'user:locale': 'en-US', 'turn': 4}

events= grows by one on each append because each append_event adds one event to the session.
state= shows merged session state after each append. The app: and user: prefixes route values to app-wide and user-wide scopes. turn stays session-scoped.

Read the session record line:

After the loop, the script reads the session row directly from Aerospike:
atomic-session-append-demo.py
```
def print_session_record(client: aerospike.Client, label: str) -> None:
    pk = ("test", "adk_sessions", f"{APP}:{USER}:{SID}")
    _, _, bins = client.select(pk, ["cur", "state"])
    print(
        f"{label} — cur_segment={bins.get('cur')}, "
        f"session_state_keys={len(bins.get('state') or {})}"
    )

# ... sequential loop ...

print_session_record(client, "after 5 sequential appends")
```
```
after 5 sequential appends — cur_segment=0, session_state_keys=1
```
- cur_segment=0 is the active segment index. Segment 0 is the first segment, so all five events fit there with no rollover. The demo prints this label for readability. In Aerospike, the value is stored in the cur bin on the session row.
- session_state_keys=1 counts session-scoped keys in the state bin (here, only turn). App- and user-scoped keys are stored on separate Aerospike records and are not counted here. The count stays at 1 in the concurrent phase too (see Session and segment layout).
Confirm the sequential phase:

The script ran five append_event calls in order. Each call carried a state_delta spanning three scopes. AerospikeSessionService merged each event with its state updates in one round trip per append.
- Event count grows from one to five across the five lines.
- Merged state in each line includes app:tenant, user:locale, and turn.

Each append with a state_delta coalesces the event write and session, app, and user state updates into one round trip.