When two AI jobs overlap, a shared “current run” variable can put the wrong ID on a log event. You can investigate the wrong request even though the business calculation itself was correct. Keep correlation state with the executing task, then clear it when that job’s scope ends.
This guide reproduces the mix-up with a forced asyncio interleaving and replaces the shared slot with Python ContextVar. It supplies a small JSON event formatter, ten runnable tests and an explicit child-task inheritance check. It is a local engineering example, not a test of an AI model or an existing agent platform.
One shared slot, two fictional jobs
Signal Dock is a fictional business preparing two AI-assisted briefing jobs. The authored test calls them run-A and run-B and uses no customer data. Job A starts, waits while B starts and finishes, then A finishes. A single dictionary slot now contains B’s ID, so the last event is labeled B too.
The demonstration forces that order with asyncio.Event objects. It does not depend on a lucky sleep duration or claim a measured production incident. The in-memory event list below is the observed result of executing those two example jobs.
| Forced step | Task actually emitting | Shared slot records | Scoped context records |
|---|---|---|---|
| 1 · started | A | run-A | run-A |
| 2 · started | B | run-B | run-B |
| 3 · finished | B | run-B | run-B |
| 4 · finished | A resumes after B | run-B · wrong job | run-A |
Bind at the job boundary, format inside the scope
The contextvars documentation describes context-local values and reset tokens. The example declares one ContextVar at module level, sets an immutable run ID on entry and resets its token in finally. Nested scopes restore the previous value rather than replacing it with a guessed default.
The asyncio task documentation explains that a task created without an explicit context receives a copy of the current context. Creating the A and B tasks, then binding inside each job, gives the two jobs separate values during the forced interleaving.
The example’s formatter reads the current ID when creating a record, accepts only started, finished or failed, and returns a JSON string with exactly two fields. It does not accept arbitrary prompt text or exception messages. This narrow interface reduces accidental logging in this example; it is not a general redaction system. A name-shaped ID can still contain private information if you put it there. Use opaque IDs generated by your application.
Run the complete reproduction
Save this as run_context.py in a new disposable directory. The code requires no third-party package. Execution used Windows and CPython 3.14.3 with its native _asyncio.Task implementation. The current Python 3.14 documentation identifies itself as 3.14.10; that newer runtime was not tested.
The versioned contextvars wrapper and native asyncio source were inspected for the corresponding import and task-context behavior. The behavioral claims below are also checked by executed tests of the installed runtime.
"""In-memory correlation demo; no model, network, files, or log collector."""
import asyncio
from contextlib import contextmanager
from contextvars import ContextVar
import json
import re
import unittest
RUN = ContextVar('run_id', default=None)
EVENTS = {'started', 'finished', 'failed'}
@contextmanager
def bind_run(run_id):
if not isinstance(run_id, str) or re.fullmatch(r'run-[A-Za-z0-9-]{1,32}', run_id) is None:
raise ValueError('Use an opaque run- ID with 1 to 32 ASCII letters/digits/hyphens')
token = RUN.set(run_id)
try:
yield
finally:
RUN.reset(token)
def event_json(event):
run_id = RUN.get()
if run_id is None:
raise RuntimeError('No run context is bound')
if not isinstance(event, str) or event not in EVENTS:
raise ValueError('Unsupported event')
return json.dumps({'run_id': run_id, 'event': event}, sort_keys=True)
async def scoped_demo():
"""Force A-start, B-start, B-finish, A-finish without timing guesses."""
a_started, b_finished = asyncio.Event(), asyncio.Event()
records = []
async def job_a():
with bind_run('run-A'):
records.append(json.loads(event_json('started')))
a_started.set()
await b_finished.wait()
records.append(json.loads(event_json('finished')))
async def job_b():
await a_started.wait()
with bind_run('run-B'):
records.append(json.loads(event_json('started')))
records.append(json.loads(event_json('finished')))
b_finished.set()
async with asyncio.TaskGroup() as group:
group.create_task(job_a())
group.create_task(job_b())
return records
async def shared_demo():
"""The same forced ordering, with one deliberately unsafe shared slot."""
state, records = {}, []
a_started, b_finished = asyncio.Event(), asyncio.Event()
async def job_a():
state['run_id'] = 'run-A'
records.append(dict(run_id=state['run_id'], event='started'))
a_started.set()
await b_finished.wait()
records.append(dict(run_id=state['run_id'], event='finished'))
async def job_b():
await a_started.wait()
state['run_id'] = 'run-B'
records.append(dict(run_id=state['run_id'], event='started'))
records.append(dict(run_id=state['run_id'], event='finished'))
b_finished.set()
async with asyncio.TaskGroup() as group:
group.create_task(job_a())
group.create_task(job_b())
return records
class ContextTests(unittest.TestCase):
def test_missing_context_is_rejected(self):
with self.assertRaises(RuntimeError): event_json('started')
def test_event_allowlist(self):
with bind_run('run-A'):
for value in ['customer prompt text', None, 7, ['started']]:
with self.subTest(value=value), self.assertRaises(ValueError): event_json(value)
def test_invalid_ids(self):
for value in [None, 7, '', 'A', 'run-', 'run-a\nb', 'run-' + 'a'*33]:
with self.subTest(value=value), self.assertRaises(ValueError):
with bind_run(value): pass
def test_nested_scope_restores_outer(self):
with bind_run('run-A'):
with bind_run('run-B'): self.assertEqual(RUN.get(), 'run-B')
self.assertEqual(RUN.get(), 'run-A')
self.assertIsNone(RUN.get())
def test_exception_restores_context(self):
with self.assertRaises(ValueError):
with bind_run('run-A'): raise ValueError('synthetic failure')
self.assertIsNone(RUN.get())
def test_exact_record_fields(self):
with bind_run('run-A'):
self.assertEqual(json.loads(event_json('failed')), {'run_id':'run-A','event':'failed'})
def test_shared_slot_mislabels_final_event(self):
self.assertEqual(asyncio.run(shared_demo())[-1], {'run_id':'run-B','event':'finished'})
def test_concurrent_scopes_keep_ids(self):
rows = asyncio.run(scoped_demo())
self.assertEqual([r['run_id'] for r in rows], ['run-A','run-B','run-B','run-A'])
self.assertEqual([r['event'] for r in rows], ['started','started','finished','finished'])
self.assertIsNone(RUN.get())
def test_child_task_inherits_creation_context(self):
async def scenario():
ready = asyncio.Event()
async def child():
await ready.wait()
return RUN.get()
with bind_run('run-A'):
task = asyncio.create_task(child())
self.assertIsNone(RUN.get())
ready.set()
return await task
self.assertEqual(asyncio.run(scenario()), 'run-A')
def test_cancellation_runs_finally(self):
async def scenario():
entered, never = asyncio.Event(), asyncio.Event()
cleanup = []
async def worker():
try:
with bind_run('run-A'):
entered.set()
await never.wait()
finally:
cleanup.append(RUN.get())
task = asyncio.create_task(worker())
await entered.wait()
task.cancel()
with self.assertRaises(asyncio.CancelledError): await task
return cleanup
self.assertEqual(asyncio.run(scenario()), [None])
if __name__ == '__main__':
print(json.dumps({'shared_slot': asyncio.run(shared_demo()),
'scoped_context': asyncio.run(scoped_demo())}, indent=2))
Run both commands. The first reports ten passing test methods. The second prints the shared-slot and scoped-context event lists.
python -X utf8 -m unittest -v run_context.py
python -X utf8 run_context.py{
"shared_slot": [
{
"run_id": "run-A",
"event": "started"
},
{
"run_id": "run-B",
"event": "started"
},
{
"run_id": "run-B",
"event": "finished"
},
{
"run_id": "run-B",
"event": "finished"
}
],
"scoped_context": [
{
"event": "started",
"run_id": "run-A"
},
{
"event": "started",
"run_id": "run-B"
},
{
"event": "finished",
"run_id": "run-B"
},
{
"event": "finished",
"run_id": "run-A"
}
]
}The child-task trap still matters
A context copy is not a lifecycle boundary. A child created while run-A is bound can keep run-A after the parent leaves that scope. The inheritance test deliberately holds the child until after the parent resets, then confirms the child still sees run-A. Resetting the parent is not a way to erase a running child’s label.
For work that belongs to the same run, that inheritance can be useful. For a separate job, bind its own ID at its entry point. For a long-lived worker reading multiple queue items, bind and reset separately around every item. Do not bind once when the worker starts and assume all future work will identify itself. These are adaptation guidelines; no queue or worker host was integrated in this example.
The cancellation test confirms that finally resets this example’s context when the worker is cancelled and awaited. It does not claim that abrupt process termination runs cleanup or that every future modification handles cancellation correctly. The demo does not catch and suppress CancelledError in its worker.
Turn the formatter into a real logging boundary carefully
In a real application, create the complete structured record while the run context is active, then pass that record to your chosen sink. If a queue listener later reads its own current context instead of the captured record, it can lose the producer’s identity. This article returns JSON strings and stores parsed records in memory; it does not implement a logging handler, background writer or collector.
Keep run IDs in retry and handoff records so an operator can follow one attempt without logging its full prompt. Decide separately whether a retry gets a new attempt ID under the same run. The example has one ID only and does not implement distributed tracing, uniqueness enforcement, retry policy or retention. See the separate retryable CRM handoff and local run review console for adjacent workflow ideas.
A correlation ID is a lookup label, not an authorization check. ContextVar does not isolate tenants, restrict tool access, secure log storage or stop a developer from putting data into an ID. Keep those controls in the application and storage boundary. Prefer an immutable scalar here; binding a shared mutable object can reintroduce shared-state problems.
What was actually confirmed
Ten runtime test methods cover missing context, allowed events, invalid IDs, nested scopes, exception cleanup, exact record fields, the deliberately broken shared-slot result, corrected concurrent labels, child inheritance and cancellation cleanup. Both complete demonstrations ran, using events to force the stated order. Final article code is extracted into a disposable directory and executed; the published code is compared with independently preserved hashes and executed again.
No model SDK, Codex or Claude host, web framework, queue, thread pool, separate process, logging collector or production business application was exercised. The example itself makes no network calls and does not write application data. The result establishes the selected in-process asyncio behavior on the recorded runtime; it does not certify an agent integration or prove that real logs are complete, durable or private.
Jason Ead publishes AlgoThoughts AI News and connects practical engineering with My Biz Heroes: websites, automation, workflows, AI tools and training. Explore Jason’s work. The jellyfish pinball-conductor cover is generated conceptual artwork. The interior timeline reports the executed example, not a screenshot of an agent product.