[Python] Action context leaks sequentially and nested in Action/BidiAction execution

[huangjeff5]

Description & Root Cause

Genkit Python SDK action executions (including flows, tools, models, etc.) use a module-level contextvars.ContextVar named _action_context inside py/packages/genkit/src/genkit/_core/_action.py to manage request-scoped context (e.g. auth data, request metadata).

When an action is executed via Action.run (which underlies __call__ and stream), or BidiAction.stream_bidi, the runner sets the context variable:

if context:
    _ = _action_context.set(context)

However, the runner never resets the token returned by ContextVar.set().

Because ContextVars are bound to the active asyncio Task, this results in two main failure modes:

1. Sequential Bleeding: If two actions are run sequentially inside the same asyncio Task, the second action will inherit the context of the first action if the second action does not pass its own context:

   # Sets _action_context to {'auth': 'user1'}
   await action1(input, context={'auth': 'user1'})
   
   # No context passed, but still retrieves {'auth': 'user1'}!
   await action2(input) 

2. Nested Permanent Override: If a parent action actionA calls a child action actionB with a different context, the child action will permanently override the context for the parent action's remaining execution, because the parent's original context token is never restored when actionB completes.

---

Proposed Fix

Wrap the execution blocks of Action.run() and BidiAction.stream_bidi() in try...finally statements that capture the token from _action_context.set() and reset it when done.

This has been implemented and verified locally with tests:

1. Action.run():

        token = None
        if context:
            token = _action_context.set(context)

        try:
            return await self._run_with_telemetry(...)
        finally:
            if token is not None:
                _action_context.reset(token)

2. BidiAction.stream_bidi():

        token = None
        if context:
            token = _action_context.set(context)
        ...
        try:
            asyncio.create_task(_run())
            return conn
        finally:
            if token is not None:
                _action_context.reset(token)