mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-22 08:25:21 +08:00
Hyu
4482e8d876
Core no longer assembles a whole trace locally. Every workflow node is uploaded
the moment it completes as its own `execution_node` record, and each execution
closes with exactly one `execution_chain` record; all records of one execution
carry the same `event_id`, so Space groups them back into a single chain.
- every execution starts at an inbound event: event-less Pipeline lanes and
WebUI/API debug runs synthesize their virtual inbound event at the boundary
and are marked debug=true, so a Runner is never a trace origin
- sampling is decided once per event from a stable hash of the event id (the
previous `int(id[:8], 16)` raised on every `platform:`-prefixed id and so
dropped every successful inbound-event chain) and is reported on each record
as sample{mode,denominator}; the default mode is now `all`
- the local trace buffer, the 120s TTL, the per-process trace cap and the shared
open-node mirror are gone, which also removes the truncated-chain upload and
the re-buffered-closed-trace leak they caused
- a broken chain is always uploaded, and its reason falls back to the first
failed node so the chain explains itself
- flush keeps records until the sender accepts them and counts what it drops;
telemetry health replaces the unused send-task queue
- heartbeat reports bound Agent / plugin-processor / event-binding counts
- bound Pipeline and Agent resources (plugins, MCP servers, knowledge bases,
tools with their Host source, skills, models and the Runner schema selectors
that picked them) are reported on the chain record
- failure text is bounded and redacted instead of passing exception text through
- the dead per-query feature counters and their call sites are removed
Wire contract: schema 2 (`execution_node` / `execution_chain`).
289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""The Pipeline lane must land its nodes on one v2 execution chain."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import types
|
|
from importlib import import_module
|
|
|
|
import pytest
|
|
|
|
|
|
def get_modules():
|
|
# Import the application package first; the pipeline package is wired into it.
|
|
import_module('langbot.pkg.core.app')
|
|
return (
|
|
import_module('langbot.pkg.pipeline.pipelinemgr'),
|
|
import_module('langbot.pkg.telemetry.execution'),
|
|
)
|
|
|
|
|
|
def get_trace():
|
|
import_module('langbot.pkg.core.app')
|
|
return import_module('langbot.pkg.telemetry.trace')
|
|
|
|
|
|
class FakeManager:
|
|
def __init__(self):
|
|
self.telemetry_config = {'url': 'https://space.example.test', 'execution_trace': 'all'}
|
|
self.sent: list[dict] = []
|
|
|
|
async def send(self, payload: dict) -> bool:
|
|
self.sent.append(payload)
|
|
return True
|
|
|
|
|
|
CONTEXT = types.SimpleNamespace(instance_uuid='instance-1', workspace_uuid='workspace-1')
|
|
|
|
|
|
def build_pipeline(pipelinemgr, ap):
|
|
pipeline = object.__new__(pipelinemgr.RuntimePipeline)
|
|
pipeline.ap = ap
|
|
pipeline.execution_context = CONTEXT
|
|
return pipeline
|
|
|
|
|
|
def test_lane_failure_marks_the_owned_chain_without_any_prior_stage():
|
|
pipelinemgr, execution = get_modules()
|
|
manager = FakeManager()
|
|
counters = execution.ExecutionCounters(manager)
|
|
ap = types.SimpleNamespace(telemetry=types.SimpleNamespace(execution=counters))
|
|
pipeline = build_pipeline(pipelinemgr, ap)
|
|
query = types.SimpleNamespace(pipeline_config={})
|
|
|
|
with execution.ingress(ap, 'pipeline_done', CONTEXT, execution_id='query-7'):
|
|
pipeline._record_lane_failure(query, 'sandbox exited 127')
|
|
|
|
rows = [record for record in counters.records if record['event_type'] == 'execution_node']
|
|
assert len(rows) == 1
|
|
node = rows[0]
|
|
assert node['event_id'] == 'query-7'
|
|
assert node['error'] == 'sandbox exited 127'
|
|
assert (node['family'], node['operation'], node['outcome']) == ('runner', 'execute', 'failed')
|
|
chain = [record for record in counters.records if record['event_type'] == 'execution_chain'][0]
|
|
assert chain['event_id'] == 'query-7'
|
|
assert chain['outcome'] == 'failed'
|
|
assert chain['error'] == 'sandbox exited 127'
|
|
|
|
|
|
def test_lane_failure_does_not_duplicate_the_runner_node_the_orchestrator_recorded():
|
|
"""A break inside a runner is already a node; the lane only carries the reason over."""
|
|
pipelinemgr, execution = get_modules()
|
|
manager = FakeManager()
|
|
counters = execution.ExecutionCounters(manager)
|
|
ap = types.SimpleNamespace(telemetry=types.SimpleNamespace(execution=counters))
|
|
pipeline = build_pipeline(pipelinemgr, ap)
|
|
query = types.SimpleNamespace(pipeline_config={})
|
|
|
|
with execution.ingress(ap, 'pipeline_done', CONTEXT, execution_id='query-8'):
|
|
execution.record(
|
|
ap,
|
|
CONTEXT,
|
|
family='runner',
|
|
operation='execute',
|
|
mode='pipeline',
|
|
runner='plugin:langbot-team/LocalAgent/default',
|
|
outcome='failed',
|
|
error='No authorized model for local-agent',
|
|
)
|
|
pipeline._record_lane_failure(query, 'No authorized model for local-agent')
|
|
|
|
rows = [record for record in counters.records if record['event_type'] == 'execution_node']
|
|
assert [(record['family'], record['operation'], record['outcome']) for record in rows] == [
|
|
('runner', 'execute', 'failed')
|
|
]
|
|
chain = [record for record in counters.records if record['event_type'] == 'execution_chain'][-1]
|
|
assert chain['outcome'] == 'failed'
|
|
assert chain['error'] == 'No authorized model for local-agent'
|
|
|
|
|
|
def build_counters():
|
|
manager = FakeManager()
|
|
counters = import_module('langbot.pkg.telemetry.execution').ExecutionCounters(manager)
|
|
ap = types.SimpleNamespace(telemetry=types.SimpleNamespace(execution=counters))
|
|
return counters, ap
|
|
|
|
|
|
def build_lane(pipelinemgr, ap):
|
|
# ``process_query`` lives on the RuntimePipeline lane, not on PipelineManager.
|
|
lane = object.__new__(pipelinemgr.RuntimePipeline)
|
|
lane.ap = ap
|
|
lane.execution_context = CONTEXT
|
|
return lane
|
|
|
|
|
|
def build_query(**overrides):
|
|
query = types.SimpleNamespace(query_uuid='exec-lane', query_id=1, pipeline_config={}, variables={})
|
|
for key, value in overrides.items():
|
|
setattr(query, key, value)
|
|
return query
|
|
|
|
|
|
def lane_nodes(counters):
|
|
return [record for record in counters.records if record['event_type'] == 'execution_node']
|
|
|
|
|
|
def lane_shapes(counters):
|
|
return [
|
|
(record['family'], record['operation'], record['node_id'], record['parent_node_id'])
|
|
for record in lane_nodes(counters)
|
|
]
|
|
|
|
|
|
async def stop_counters(counters):
|
|
"""Cancel the telemetry flush task so no loop is left with pending work."""
|
|
task = counters.task
|
|
if task is not None and not task.done():
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
counters.task = None
|
|
|
|
|
|
async def test_lane_step_owns_the_apis_and_runner_it_calls():
|
|
pipelinemgr, execution = get_modules()
|
|
trace = get_trace()
|
|
counters, ap = build_counters()
|
|
lane = build_lane(pipelinemgr, ap)
|
|
query = build_query()
|
|
seen = {}
|
|
|
|
async def fake_process(_query):
|
|
execution.record(
|
|
ap, CONTEXT, family='platform_api', operation='send_message',
|
|
adapter='AiocqhttpAdapter', mode='pipeline', outcome='success',
|
|
)
|
|
with trace.stage_scope() as run_node:
|
|
seen['run'] = run_node
|
|
execution.record(
|
|
ap, CONTEXT, family='runner', operation='execute', mode='pipeline', outcome='success', node=run_node
|
|
)
|
|
# A reply issued after the runner still belongs to the lane.
|
|
execution.record(
|
|
ap, CONTEXT, family='platform_api', operation='reply_message',
|
|
adapter='AiocqhttpAdapter', mode='pipeline', outcome='success',
|
|
)
|
|
|
|
lane._process_query = fake_process
|
|
|
|
async def scenario():
|
|
await lane.process_query(query)
|
|
await stop_counters(counters)
|
|
|
|
await scenario()
|
|
|
|
shapes = lane_shapes(counters)
|
|
# The synthetic inbound event roots the chain; the lane step hangs under it.
|
|
assert [shape[0:2] for shape in shapes] == [
|
|
('platform_event', 'message.received'),
|
|
('platform_api', 'send_message'),
|
|
('runner', 'execute'),
|
|
('platform_api', 'reply_message'),
|
|
('pipeline', 'run'),
|
|
]
|
|
event_node = shapes[0][2]
|
|
lane_node = shapes[-1][2]
|
|
assert shapes[0][3] == ''
|
|
assert shapes[1][3] == lane_node
|
|
assert shapes[2] == ('runner', 'execute', seen['run'], lane_node)
|
|
assert shapes[3][3] == lane_node
|
|
# The lane step is recorded last and nests under the synthetic event node.
|
|
assert shapes[4] == ('pipeline', 'run', lane_node, event_node)
|
|
|
|
|
|
async def test_lane_node_nests_under_the_platform_route_that_called_it():
|
|
pipelinemgr, execution = get_modules()
|
|
trace = get_trace()
|
|
counters, ap = build_counters()
|
|
lane = build_lane(pipelinemgr, ap)
|
|
query = build_query(query_uuid='exec-routed')
|
|
|
|
async def fake_process(_query):
|
|
execution.record(
|
|
ap, CONTEXT, family='platform_api', operation='send_message', mode='pipeline', outcome='success'
|
|
)
|
|
|
|
lane._process_query = fake_process
|
|
|
|
async def scenario():
|
|
# A platform event owns the chain; the route step wraps the Pipeline lane.
|
|
with execution.ingress(ap, 'event_done', CONTEXT, execution_id='exec-routed'):
|
|
with trace.stage_scope() as event_node:
|
|
execution.record(
|
|
ap, CONTEXT, family='platform_event', operation='message.received',
|
|
adapter='AiocqhttpAdapter', outcome='success', node=event_node,
|
|
)
|
|
with trace.stage_scope() as route_node:
|
|
execution.record(
|
|
ap, CONTEXT, family='event_route', operation='message.received',
|
|
adapter='AiocqhttpAdapter', outcome='success', node=route_node,
|
|
)
|
|
await lane.process_query(query)
|
|
await stop_counters(counters)
|
|
return event_node, route_node
|
|
|
|
event_node, route_node = await scenario()
|
|
|
|
assert event_node != route_node
|
|
shapes = lane_shapes(counters)
|
|
assert shapes[0] == ('platform_event', 'message.received', event_node, '')
|
|
assert shapes[1] == ('event_route', 'message.received', route_node, event_node)
|
|
lane_node = shapes[-1][2]
|
|
# The API is called by the lane, so it hangs on the lane node.
|
|
assert shapes[2][0:2] == ('platform_api', 'send_message')
|
|
assert shapes[2][3] == lane_node and shapes[2][2] != lane_node
|
|
# The lane nests under the route step that called it.
|
|
assert shapes[3] == ('pipeline', 'run', lane_node, route_node)
|
|
|
|
|
|
async def test_broken_lane_records_a_failed_step_with_its_reason():
|
|
pipelinemgr, execution = get_modules()
|
|
counters, ap = build_counters()
|
|
lane = build_lane(pipelinemgr, ap)
|
|
|
|
async def fake_process(_query):
|
|
raise RuntimeError('lane exploded')
|
|
|
|
lane._process_query = fake_process
|
|
|
|
async def scenario():
|
|
try:
|
|
await lane.process_query(build_query())
|
|
finally:
|
|
await stop_counters(counters)
|
|
|
|
with pytest.raises(RuntimeError, match='lane exploded'):
|
|
await scenario()
|
|
|
|
shapes = lane_shapes(counters)
|
|
family, operation, node, parent = shapes[-1]
|
|
row = lane_nodes(counters)[-1]
|
|
# The lane step nests under the synthetic inbound event that roots the chain.
|
|
assert (family, operation) == ('pipeline', 'run')
|
|
assert shapes[0] == ('platform_event', 'message.received', parent, '')
|
|
assert row['outcome'] == 'failed'
|
|
assert row['error'] == 'lane exploded'
|
|
|
|
|
|
async def test_lane_reported_failure_marks_the_step_failed():
|
|
pipelinemgr, execution = get_modules()
|
|
counters, ap = build_counters()
|
|
lane = build_lane(pipelinemgr, ap)
|
|
query = build_query(variables={'_monitoring_has_error': True})
|
|
|
|
async def fake_process(_query):
|
|
return None
|
|
|
|
lane._process_query = fake_process
|
|
|
|
async def scenario():
|
|
await lane.process_query(query)
|
|
await stop_counters(counters)
|
|
|
|
await scenario()
|
|
|
|
row = lane_nodes(counters)[-1]
|
|
assert (row['family'], row['operation'], row['outcome']) == ('pipeline', 'run', 'failed')
|