diff --git a/tests/contrib/streams/test_live_links.py b/tests/contrib/streams/test_live_links.py new file mode 100644 index 000000000..f4920d6a2 --- /dev/null +++ b/tests/contrib/streams/test_live_links.py @@ -0,0 +1,375 @@ +"""Link records and followers, against a server with the stream notifier. + +An agent Workflow writes on its own stream and links a tool call that +streams on its own and a long-lived child agent. A client follower and a +Workflow follower read all of it in one order. The Workflow follower sleeps +while every source is idle and wakes through the notifier. The stock dev +server has no stream notifier, so every case here skips there. +``scripts/streams_live_lane.sh`` runs them on a server that has it, on +memory, standalone Redis and a Redis Cluster. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import os +import uuid +from datetime import timedelta +from typing import Any + +import pytest + +from temporalio import activity, workflow +from temporalio.client import Client +from temporalio.common import RetryPolicy +from temporalio.contrib.streams import ( + FollowedItem, + FollowPosition, + LinkStateKind, + RecordKind, + StreamRef, + StreamStorePlugin, + activity_handle, + start_activity_with_stream, + stream_handle, + stream_producer, + topic, + workflow_follower, +) +from temporalio.contrib.streams import _follow as follow_module +from temporalio.contrib.streams._record import Cursor +from temporalio.contrib.streams.memory import MemoryStreams +from temporalio.contrib.streams.redis import RedisStreams +from temporalio.exceptions import ApplicationError +from tests.contrib.streams._support import connect_with, new_workflow_id +from tests.contrib.streams.test_live_notifications import skip_without_notifier +from tests.helpers import new_worker + +EVENTS = topic("events", str) +TOKENS = topic("tokens", str) +# A link to a Workflow that never starts is unavailable after this grace. +GRACE = timedelta(seconds=3) + + +def _store(kind: str) -> StreamStorePlugin: + if kind == "memory": + return MemoryStreams(owner_start_grace=GRACE) + env = "STREAMS_REDIS_CLUSTER_URL" if kind == "cluster" else "STREAMS_REDIS_URL" + url = os.environ.get(env, "") + if not url: + pytest.skip(f"set {env} to run on Redis") + return RedisStreams( + url, + cluster=kind == "cluster", + key_prefix=f"live-{uuid.uuid4().hex}", + owner_start_grace=GRACE, + ) + + +@pytest.fixture(params=["memory", "redis", "cluster"]) +async def linked(request: pytest.FixtureRequest, client: Client) -> Client: + """A client whose store notifies, on a server with the notifier.""" + await skip_without_notifier(client) + return await connect_with(client, _store(request.param)) + + +def describe(item: FollowedItem[Any]) -> str: + """One item as ``:``.""" + if item.source.activity_id: + source = item.source.activity_id + else: + source = "agent" if item.depth == 0 else "child" + if item.link_state is not None: + state = item.link_state + what = f"{state.kind.name} {state.link.name}" + if state.reason: + what += f" {state.reason}" + return f"{source}:{what}" + record = item.record + assert record is not None + if record.kind is RecordKind.LINK: + assert record.link is not None + return f"{source}:LINK {record.link.event.name} {record.link.name}" + if record.kind is RecordKind.DATA: + return f"{source}:{record.value}" + return f"{source}:{record.kind.name}" + + +async def follow_all( + client: Client, workflow_id: str, after: Any = None, timeout: float = 60 +) -> list[FollowedItem[Any]]: + handle = stream_handle(client, workflow_id, topic=EVENTS) + items: list[FollowedItem[Any]] = [] + + async def pull() -> None: + kwargs = {} if after is None else {"after": after} + async for item in handle.follow(**kwargs): + items.append(item) + + await asyncio.wait_for(pull(), timeout) + return items + + +@workflow.defn +class Follows: + """Follows an agent's stream to its end and says what it saw.""" + + @workflow.run + async def run(self, agent_id: str) -> list[str]: + seen: list[str] = [] + async with workflow_follower(agent_id, topic=EVENTS) as follower: + async for batch in follower: + seen.extend(describe(i) for i in batch.items) + return seen + + +# Steps 1 and 2: the gates hold for a client follower and a Workflow follower. + + +@activity.defn +async def streams_tokens(tokens: list[str]) -> str: + out = activity_handle().producer(topic=TOKENS) + for token in tokens: + await out.append(token) + await asyncio.sleep(0.2) + return "ok" + + +@workflow.defn +class ChildAgent: + """A long-lived child that answers one turn per Signal.""" + + def __init__(self) -> None: + self.turns = 0 + + @workflow.run + async def run(self, turns: int) -> None: + events = stream_producer(EVENTS) + events.append("c0") + for turn in range(1, turns + 1): + await workflow.wait_condition(lambda: self.turns >= turn) + events.append(f"c{2 * turn - 1}") + events.append(f"c{2 * turn}") + + @workflow.signal + def turn(self) -> None: + self.turns += 1 + + +@workflow.defn +class Agent: + def __init__(self) -> None: + self.until: str | None = None + + @workflow.run + async def run(self, child_id: str, after: str) -> None: + events = stream_producer(EVENTS) + events.append("d1") + tool, tool_ref = start_activity_with_stream( + streams_tokens, + ["a1", "a2", "a3"], + activity_id="tool-a", + start_to_close_timeout=timedelta(seconds=30), + topic=TOKENS, + ) + events.link(tool_ref, name="tool-a") + events.link( + StreamRef.for_workflow(child_id, topic=EVENTS), + name="child", + after=Cursor(after), + ) + events.append("d2") + await tool + events.join(name="tool-a") + events.append("d3") + await workflow.wait_condition(lambda: self.until is not None) + assert self.until is not None + events.join(name="child", until=Cursor(self.until)) + events.append("d4") + + @workflow.signal + def child_turn_ended(self, cursor: str) -> None: + self.until = cursor + + +def check_gates(seen: list[str]) -> None: + at = {s: i for i, s in enumerate(seen)} + for token in ("a1", "a2", "a3"): + assert seen.count(f"tool-a:{token}") == 1 + assert at["agent:LINK OPEN tool-a"] < at[f"tool-a:{token}"] + for record in ("c1", "c2"): + assert seen.count(f"child:{record}") == 1 + assert at["agent:LINK OPEN child"] < at[f"child:{record}"] + # The link opened strictly after c0. + assert "child:c0" not in seen + assert at["tool-a:DONE tool-a"] < at["agent:d3"] + assert at["child:c2"] < at["agent:d4"] + assert [s for s in seen if s.startswith("agent:d")] == [ + "agent:d1", + "agent:d2", + "agent:d3", + "agent:d4", + ] + + +async def test_live_both_followers_keep_the_gates_of_an_agent(linked: Client): + async with new_worker( + linked, Agent, ChildAgent, Follows, activities=[streams_tokens] + ) as worker: + child_id = new_workflow_id() + child = await linked.start_workflow( + ChildAgent.run, 2, id=child_id, task_queue=worker.task_queue + ) + after = await _cursor_of_first(linked, child_id, "c0") + agent = await linked.start_workflow( + Agent.run, + args=[child_id, after], + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + follower = await linked.start_workflow( + Follows.run, agent.id, id=new_workflow_id(), task_queue=worker.task_queue + ) + client_items = asyncio.ensure_future(follow_all(linked, agent.id, timeout=90)) + await child.signal(ChildAgent.turn) + until = await _cursor_of_first(linked, child_id, "c2") + await agent.signal(Agent.child_turn_ended, until) + await asyncio.wait_for(agent.result(), 60) + seen = [describe(i) for i in await client_items] + check_gates(seen) + check_gates(await asyncio.wait_for(follower.result(), 60)) + await child.signal(ChildAgent.turn) + await asyncio.wait_for(child.result(), 30) + + +async def _cursor_of_first(client: Client, workflow_id: str, value: str) -> str: + handle = stream_handle(client, workflow_id, topic=EVENTS) + records = handle.read() + try: + async for record in records: + if record.value == value: + return record.cursor.token + finally: + await records.aclose() + raise AssertionError(f"{value} not found") + + +# Steps 3, 4 and 5: a failed tool, a missing child and a cycle never fail the +# follow. + + +@activity.defn +async def fails_with_boom() -> None: + await activity_handle().producer(topic=TOKENS).append("x1") + raise ApplicationError("boom", type="Boom", non_retryable=True) + + +@workflow.defn +class LinksTheHardCases: + @workflow.run + async def run(self, missing_id: str) -> None: + events = stream_producer(EVENTS) + tool, tool_ref = start_activity_with_stream( + fails_with_boom, + activity_id="tool-boom", + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy(maximum_attempts=1), + topic=TOKENS, + ) + events.link(tool_ref, name="tool-boom") + events.link(StreamRef.for_workflow(missing_id, topic=EVENTS), name="missing") + cycle_id = f"{workflow.info().workflow_id}-cycle" + await workflow.execute_child_workflow( + LinksBack.run, workflow.info().workflow_id, id=cycle_id + ) + events.link(StreamRef.for_workflow(cycle_id, topic=EVENTS), name="cycle") + with contextlib.suppress(Exception): + await tool + events.append("after") + + +@workflow.defn +class LinksBack: + @workflow.run + async def run(self, parent_id: str) -> None: + events = stream_producer(EVENTS) + events.link(StreamRef.for_workflow(parent_id, topic=EVENTS), name="parent") + events.append("cycled") + + +async def test_live_a_failed_tool_a_missing_child_and_a_cycle_go_on(linked: Client): + async with new_worker( + linked, LinksTheHardCases, LinksBack, activities=[fails_with_boom] + ) as worker: + agent = await linked.start_workflow( + LinksTheHardCases.run, + f"missing-{uuid.uuid4().hex}", + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + await asyncio.wait_for(agent.result(), 60) + items = await follow_all(linked, agent.id, timeout=60) + states = { + i.link_state.link.name: i.link_state + for i in items + if i.link_state is not None and i.link_state.kind is not LinkStateKind.OPENED + } + boom = states["tool-boom"] + assert boom.kind is LinkStateKind.DONE + error: BaseException | None = boom.error + while error is not None and not isinstance(error, ApplicationError): + error = error.__cause__ + assert isinstance(error, ApplicationError) and error.type == "Boom" + assert states["missing"].kind is LinkStateKind.UNAVAILABLE + # The child's link back to the agent is not opened again. + assert states["parent"].kind is LinkStateKind.SKIPPED + assert states["parent"].reason == "cycle" + seen = [describe(i) for i in items] + assert "tool-boom:x1" in seen and "child:cycled" in seen + # No JOIN holds the agent, so its last record need not wait for the children. + assert "agent:after" in seen + + +# Step 6: a position saved inside the tool's segment resumes the rest of it once. + + +async def test_live_a_follower_resumes_inside_a_segment( + linked: Client, monkeypatch: pytest.MonkeyPatch +): + async with new_worker( + linked, Agent, ChildAgent, activities=[streams_tokens] + ) as worker: + child_id = new_workflow_id() + child = await linked.start_workflow( + ChildAgent.run, 1, id=child_id, task_queue=worker.task_queue + ) + after = await _cursor_of_first(linked, child_id, "c0") + agent = await linked.start_workflow( + Agent.run, + args=[child_id, after], + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + await child.signal(ChildAgent.turn) + until = await _cursor_of_first(linked, child_id, "c2") + await agent.signal(Agent.child_turn_ended, until) + await asyncio.wait_for(agent.result(), 60) + monkeypatch.setattr(follow_module, "_MAX_RECORDS", 2) + items = await follow_all(linked, agent.id) + seen = [describe(i) for i in items] + inside = [ + (n, i.position) + for n, i in enumerate(items) + if i.position is not None + and "agent:LINK OPEN tool-a" in seen[: n + 1] + and "tool-a:DONE tool-a" not in seen[: n + 1] + ] + assert inside, seen + count, position = inside[0] + assert isinstance(position, FollowPosition) + rest = [describe(i) for i in await follow_all(linked, agent.id, after=position)] + assert seen[: count + 1] + rest == seen + tool_records = [s for s in seen[: count + 1] + rest if s.startswith("tool-a:a")] + assert tool_records == ["tool-a:a1", "tool-a:a2", "tool-a:a3"] + await asyncio.wait_for(child.result(), 30) diff --git a/tests/contrib/streams/test_live_wfa_streams.py b/tests/contrib/streams/test_live_wfa_streams.py new file mode 100644 index 000000000..2c8464df1 --- /dev/null +++ b/tests/contrib/streams/test_live_wfa_streams.py @@ -0,0 +1,371 @@ +"""Streams that a Workflow Activity owns, against a server that serves them. + +An agent Workflow starts a tool call with ``start_activity_with_stream``. The +tool writes its own stream, every attempt as a new producer attempt, and the +agent's Worker closes the stream when the agent sees the tool end. When that +close never comes, the server's notifier closes the stream after its graces. +The stock dev server has no stream notifier, so every case here skips +there. ``scripts/streams_live_lane.sh`` runs them on a server that has it, on +memory, standalone Redis and a Redis Cluster. +""" + +from __future__ import annotations + +import asyncio +import uuid +from datetime import timedelta +from typing import Any + +import pytest + +from temporalio import activity, workflow +from temporalio.api.common.v1 import WorkflowExecution +from temporalio.api.enums.v1 import EventType +from temporalio.api.workflowservice.v1 import ResetWorkflowExecutionRequest +from temporalio.client import Client, WorkflowFailureError +from temporalio.common import RetryPolicy +from temporalio.contrib.streams import ( + StreamCursorError, + StreamRef, + activity_handle, + start_activity_with_stream, + stream_handle, + topic, +) +from temporalio.exceptions import ( + ActivityError, + ApplicationError, + TimeoutError, +) +from temporalio.worker import Worker +from tests.contrib.streams import test_live_activity_owner as _saa +from tests.contrib.streams._support import new_workflow_id +from tests.contrib.streams.test_live_activity_owner import ( + OWNER_CHECK, + Consume, + _close_result, + _read, +) +from tests.helpers import new_worker + +# The same stores as the standalone Activity cases. +notifying = _saa.notifying + +TOKENS = topic("tokens", str) +TOOL_ID = "tool-1" +_RETRY = RetryPolicy(initial_interval=timedelta(seconds=1), maximum_attempts=2) + + +@workflow.defn +class Agent: + """Runs one tool call that streams, and returns its result.""" + + @workflow.run + async def run( + self, tool: str, timeout_ms: int, attempts: int, tool_queue: str = "" + ) -> Any: + handle, _ = start_activity_with_stream( + tool, + activity_id=TOOL_ID, + task_queue=tool_queue or None, + start_to_close_timeout=timedelta(milliseconds=timeout_ms), + retry_policy=RetryPolicy( + initial_interval=timedelta(seconds=1), maximum_attempts=attempts + ), + topic=TOKENS, + ) + return await handle + + +def _tool_ref(agent: Any) -> StreamRef: + return StreamRef.for_workflow_activity( + agent.id, run_id=agent.result_run_id, activity_id=TOOL_ID, topic=TOKENS + ) + + +async def _start( + client: Client, task_queue: str, tool: str, *, timeout_ms: int = 10_000 +) -> tuple[Any, Any]: + """Start the agent and a Workflow reader of its tool's stream.""" + agent = await client.start_workflow( + Agent.run, + args=[tool, timeout_ms, 2], + id=new_workflow_id(), + task_queue=task_queue, + ) + reader = await client.start_workflow( + Consume.run, _tool_ref(agent), id=new_workflow_id(), task_queue=task_queue + ) + return agent, reader + + +# Steps 1 to 3: a retry supersedes, and the close at resolution ends both +# readers with a null result. + + +@activity.defn +async def run_tool() -> str: + out = activity_handle().producer(topic=TOKENS) + if activity.info().attempt == 1: + await out.append("a1", "a2", "a3") + raise ApplicationError("retry me") + await out.append("b1", "b2") + return "ok" + + +async def test_live_both_readers_follow_a_tool_call_across_its_retry(notifying: Client): + async with new_worker(notifying, Agent, Consume, activities=[run_tool]) as worker: + agent, reader = await _start(notifying, worker.task_queue, "run_tool") + ref = _tool_ref(agent) + seen, withdrawn = await asyncio.wait_for(_read(notifying, ref), 60) + assert await agent.result() == "ok" + expected = ["a1", "a2", "a3", "SUPERSEDED", "b1", "b2"] + assert seen == expected + assert withdrawn == ["a1", "a2", "a3"] + assert await _close_result(notifying, ref) is None + assert await asyncio.wait_for(reader.result(), 60) == expected + + +# Step 4: a tool that fails for good. + + +@activity.defn +async def fails_hard() -> None: + await activity_handle().producer(topic=TOKENS).append("x1") + raise ApplicationError("boom", type="Boom", non_retryable=True) + + +def _application_cause(error: BaseException) -> ApplicationError: + while not isinstance(error, ApplicationError): + assert error.__cause__ is not None, error + error = error.__cause__ + return error + + +async def test_live_a_tool_that_fails_closes_its_stream_with_the_failure( + notifying: Client, +): + async with new_worker(notifying, Agent, Consume, activities=[fails_hard]) as worker: + agent, reader = await _start(notifying, worker.task_queue, "fails_hard") + ref = _tool_ref(agent) + with pytest.raises((ActivityError, ApplicationError)) as client_raised: + await asyncio.wait_for(_read(notifying, ref), 60) + assert _application_cause(client_raised.value).type == "Boom" + with pytest.raises(WorkflowFailureError): + await agent.result() + with pytest.raises(WorkflowFailureError) as raised: + await asyncio.wait_for(reader.result(), 60) + assert _application_cause(raised.value).type == "Boom" + + +# Step 5: a start-to-close timeout. + + +@activity.defn +async def hangs() -> None: + await activity_handle().producer(topic=TOKENS).append("h1") + await asyncio.sleep(30) + + +async def test_live_a_tool_that_times_out_closes_its_stream_with_the_timeout( + notifying: Client, +): + async with new_worker(notifying, Agent, Consume, activities=[hangs]) as worker: + agent = await notifying.start_workflow( + Agent.run, + args=["hangs", 2_000, 1], + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + reader = await notifying.start_workflow( + Consume.run, + _tool_ref(agent), + id=new_workflow_id(), + task_queue=worker.task_queue, + ) + with pytest.raises(WorkflowFailureError): + await asyncio.wait_for(agent.result(), 60) + with pytest.raises(WorkflowFailureError) as raised: + await asyncio.wait_for(reader.result(), 60) + error: BaseException | None = raised.value + while error is not None and not isinstance(error, TimeoutError): + error = error.__cause__ + assert isinstance(error, TimeoutError) + + +# Step 6: Continue-as-New gives the next run's Activity "1" a new stream. + + +@activity.defn +async def writes_its_run() -> str: + info = activity.info() + await activity_handle().producer(topic=TOKENS).append(info.workflow_run_id or "") + return info.workflow_run_id or "" + + +@workflow.defn +class ChainedAgent: + @workflow.run + async def run(self, runs_left: int) -> list[str]: + handle, ref = start_activity_with_stream( + writes_its_run, start_to_close_timeout=timedelta(seconds=10), topic=TOKENS + ) + await handle + assert ref.activity_id == "1" + if runs_left > 1: + workflow.continue_as_new(runs_left - 1) + return [workflow.info().run_id] + + +async def test_live_each_run_of_a_chain_gets_its_own_tool_stream(notifying: Client): + async with new_worker( + notifying, ChainedAgent, activities=[writes_its_run] + ) as worker: + agent = await notifying.start_workflow( + ChainedAgent.run, 2, id=new_workflow_id(), task_queue=worker.task_queue + ) + [last_run] = await asyncio.wait_for(agent.result(), 60) + first_run = agent.first_execution_run_id or "" + assert first_run != last_run + refs = [ + StreamRef.for_workflow_activity( + agent.id, run_id=run, activity_id="1", topic=TOKENS + ) + for run in (first_run, last_run) + ] + first = [r async for r in stream_handle(notifying, refs[0]).read()] + last = [r async for r in stream_handle(notifying, refs[1]).read()] + assert [r.value for r in first] == [first_run] + assert [r.value for r in last] == [last_run] + with pytest.raises(StreamCursorError): + async for _ in stream_handle(notifying, refs[1]).read( + after=first[-1].cursor + ): + pass + + +# Step 7: a Workflow reset reruns the tool in a new run, on a new stream. + +_release: dict[str, asyncio.Event] = {} +_started: dict[str, asyncio.Event] = {} + + +@activity.defn +async def waits_in_the_first_run() -> str: + info = activity.info() + workflow_id = info.workflow_id or "" + await activity_handle().producer(topic=TOKENS).append(info.workflow_run_id or "") + if info.workflow_run_id == _first_runs.get(workflow_id): + _started[workflow_id].set() + await _release[workflow_id].wait() + return "ok" + + +_first_runs: dict[str, str] = {} + + +async def test_live_a_reset_run_writes_a_new_tool_stream(notifying: Client): + workflow_id = new_workflow_id() + _release[workflow_id] = asyncio.Event() + _started[workflow_id] = asyncio.Event() + async with new_worker( + notifying, Agent, activities=[waits_in_the_first_run] + ) as worker: + agent = await notifying.start_workflow( + Agent.run, + args=["waits_in_the_first_run", 60_000, 1], + id=workflow_id, + task_queue=worker.task_queue, + ) + _first_runs[workflow_id] = agent.result_run_id or "" + await asyncio.wait_for(_started[workflow_id].wait(), 30) + history = await agent.fetch_history() + [first_task] = [ + e.event_id + for e in history.events + if e.event_type == EventType.EVENT_TYPE_WORKFLOW_TASK_COMPLETED + ][:1] + reset = await notifying.workflow_service.reset_workflow_execution( + ResetWorkflowExecutionRequest( + namespace=notifying.namespace, + workflow_execution=WorkflowExecution( + workflow_id=workflow_id, run_id=agent.result_run_id or "" + ), + reason="lane", + workflow_task_finish_event_id=first_task - 1, + request_id=str(uuid.uuid4()), + ) + ) + new_run = reset.run_id + assert ( + await asyncio.wait_for( + notifying.get_workflow_handle(workflow_id, run_id=new_run).result(), 60 + ) + == "ok" + ) + old = StreamRef.for_workflow_activity( + workflow_id, + run_id=_first_runs[workflow_id], + activity_id=TOOL_ID, + topic=TOKENS, + ) + new = StreamRef.for_workflow_activity( + workflow_id, run_id=new_run, activity_id=TOOL_ID, topic=TOKENS + ) + # The old run ended with the reset, so its read ends at once. + old_seen = [r.value async for r in stream_handle(notifying, old).read()] + assert old_seen == [_first_runs[workflow_id]] + new_seen = [r.value async for r in stream_handle(notifying, new).read()] + assert new_seen == [new_run] + _release[workflow_id].set() + + +# Step 8: the agent's Worker stops before it sees the tool end, so the +# notifier closes the stream after its graces. + +_tool_started: dict[str, asyncio.Event] = {} +_tool_go: dict[str, asyncio.Event] = {} + + +@activity.defn +async def outlives_its_agent() -> str: + workflow_id = activity.info().workflow_id or "" + await activity_handle().producer(topic=TOKENS).append("o1") + _tool_started[workflow_id].set() + await _tool_go[workflow_id].wait() + return "ok" + + +async def test_live_a_tool_whose_agent_never_closes_ends_by_the_owner_check( + notifying: Client, +): + workflow_id = new_workflow_id() + _tool_started[workflow_id] = asyncio.Event() + _tool_go[workflow_id] = asyncio.Event() + agent_queue = f"agent-{uuid.uuid4()}" + async with ( + new_worker(notifying, Consume) as readers, + new_worker(notifying, activities=[outlives_its_agent]) as tools, + ): + agent_worker = Worker(notifying, task_queue=agent_queue, workflows=[Agent]) + async with agent_worker: + agent = await notifying.start_workflow( + Agent.run, + args=["outlives_its_agent", 60_000, 1, tools.task_queue], + id=workflow_id, + task_queue=agent_queue, + ) + await asyncio.wait_for(_tool_started[workflow_id].wait(), 30) + # The tool runs on its own task queue, so it completes while the agent + # has no Worker to see it. + _tool_go[workflow_id].set() + reader = await notifying.start_workflow( + Consume.run, + _tool_ref(agent), + id=new_workflow_id(), + task_queue=readers.task_queue, + ) + with pytest.raises(WorkflowFailureError) as raised: + await asyncio.wait_for(reader.result(), 2 * OWNER_CHECK) + assert "without closing the stream" in str(_application_cause(raised.value)) + await agent.terminate(reason="lane")