From b88a038ffeac0a6a9dda475d1926fd76739b9224 Mon Sep 17 00:00:00 2001 From: Tsuyoshi Ushio Date: Thu, 3 Sep 2026 12:31:46 -0700 Subject: [PATCH] Add DTS workflow display names Use durabletask.displayName tags for workflow orchestrations and activities while preserving registered function names. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 184fe41a-172f-4767-942d-c006ec943d9e --- docs/architecture.md | 2 +- docs/frds/0004-dynamic-workflows.md | 59 ++++++++++++- docs/workflows.md | 5 +- .../workflows/engine.py | 49 ++++++----- src/azure_functions_agents/workflows/tools.py | 10 ++- tests/test_workflow_engine.py | 82 +++++++++++++------ tests/test_workflow_registry.py | 15 ++-- 7 files changed, 163 insertions(+), 59 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 217b016d..bc96a51a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -77,7 +77,7 @@ A few boundaries are worth calling out explicitly: | `azure_functions_agents/runner.py` | Executes prompts through the Microsoft Agent Framework, managing sessions, tools, and streaming; builds per-request `delegate_` tools and fresh stateless workflow leaf agents; attempts one internal token-usage record through the shared runtime logger for each actual MAF invocation attempt. | `run_agent()`, `run_agent_stream()`, `build_subagent_tools()`, `run_leaf_agent_task()` | | `azure_functions_agents/client_manager.py` | Defines the pluggable inference-client abstraction, immutable inference-target metadata, and the default MAF-backed implementation. | `ClientManager`, `InferenceTarget`, `get_client_manager()`, `set_client_manager()` | | `azure_functions_agents/workflows/integration.py` | Builds the complete immutable handler catalog, immutable slug-keyed workflow-agent policy catalog, per-agent management tools/addenda, validates declared trigger support for workflow-enabled agents, and performs the one app-wide Durable registration. It also resolves the packaged `data-driven-workflows` skill used for progressive authoring guidance. | `build_workflow_handler_catalog()`, `build_workflow_agent_policy_catalog()`, `build_workflow_agent_integration()`, `data_driven_workflows_skill_path()`, `validate_workflow_agent_trigger()`, `register_workflow_runtime()` | -| `azure_functions_agents/workflows/engine.py` | Registers one Durable blueprint per app and executes the orchestrator, workflow-tool Activity, and Workflow Sub Agent Activity. Capability-bearing Activities reauthorize against the current workflow-agent policy before complete-catalog dispatch. Data-driven execution uses typed persisted-task/state contracts and deterministic phase helpers for `when` evaluation, bounded `for_each` materialization, runnable selection, ordered aggregation, result application, cancellation restoration, structured (`schema_version: 2`) status, and controlled-failure normalization. Durable `yield` boundaries remain in the top-level orchestrator generator. | `register_workflows()` | +| `azure_functions_agents/workflows/engine.py` | Registers one Durable blueprint per app and executes the native two-argument Durable Task orchestrator, workflow-tool Activity, and Workflow Sub Agent Activity. Orchestration and Activity schedules attach `durabletask.displayName` tags for readable DTS dashboard timelines without changing registered function names. Capability-bearing Activities reauthorize against the current workflow-agent policy before complete-catalog dispatch. Data-driven execution uses typed persisted-task/state contracts and deterministic phase helpers for `when` evaluation, bounded `for_each` materialization, runnable selection, ordered aggregation, result application, cancellation restoration, structured (`schema_version: 2`) status, and controlled-failure normalization. Durable `yield` boundaries remain in the top-level orchestrator generator. | `register_workflows()` | | `azure_functions_agents/workflows/context.py` | Tracks invocation context by `(workflow_agent_slug, session_id)` and derives non-revealing 128-bit agent/session prefixes for Durable instance IDs. | `session_instance_prefix()`, `new_workflow_instance_id()`, `workflow_matches_agent_session()` | | `azure_functions_agents/workflows/registry.py` | Defines immutable workflow handler entries/catalogs; production app composition passes this complete catalog explicitly rather than using the compatibility singleton allowlist as authorization. | `WorkflowHandlerCatalog`, `build_handler_catalog()` | | `azure_functions_agents/workflows/schema.py`, `workflows/tools.py` | Define workflow plans/policies — including the data-driven `when` predicate and bounded `for_each` fields — and build agent-scoped management tools. Start-time validation and list/status/cancel/terminate operations use the captured workflow-agent policy and agent/session identity. | `WorkflowPlanPolicy`, `WorkflowCondition`, `validate_plan()`, `evaluate_condition()`, `build_workflow_tools()` | diff --git a/docs/frds/0004-dynamic-workflows.md b/docs/frds/0004-dynamic-workflows.md index 431361bf..bf7e8ff6 100644 --- a/docs/frds/0004-dynamic-workflows.md +++ b/docs/frds/0004-dynamic-workflows.md @@ -4,7 +4,7 @@ title: Dynamic workflows status: Finalized author: TsuyoshiUshio created: 2026-07-06 -updated: 2026-08-14 +updated: 2026-09-03 issues: [https://github.com/Azure/azure-functions-agents-runtime/issues/108, https://github.com/Azure/azure-functions-agents-runtime/issues/109, https://github.com/Azure/azure-functions-bucees-planning/issues/1274, https://github.com/Azure/azure-functions-bucees-planning/issues/1275, https://github.com/Azure/azure-functions-bucees-planning/issues/1276] pull_requests: [https://github.com/Azure/azure-functions-agents-runtime/pull/77, https://github.com/Azure/azure-functions-agents-runtime/pull/112, https://github.com/Azure/azure-functions-agents-runtime/pull/117, https://github.com/Azure/azure-functions-agents-runtime/pull/151, https://github.com/Azure/azure-functions-agents-runtime/pull/163] --- @@ -39,6 +39,12 @@ workflow-enabled agent with agent/session isolation. The records only that extension's behavioral and architectural delta instead of repeating the base workflow design. +The next operational evolution adds human-readable DTS dashboard labels without +changing workflow plans or registered Azure Function names. Orchestrations use +the existing workflow session `agent_name` plus the fixed `-orchestration` +suffix; tool and Sub Agent Activities use their concrete tool name or agent +slug. + ## 2. Motivation / problem Today agents can call tools directly through the Microsoft Agent Framework (MAF) @@ -938,6 +944,39 @@ The Dynamic Workflow sample for this extension must demonstrate: 4. status output showing expanded, running, skipped, and aggregated states; and 5. deterministic completion on both Azure Storage and DTS Durable backends. +### Durable Task Scheduler display names + +The runtime attaches the well-known `durabletask.displayName` tag when it starts +an orchestration and schedules a tool or Workflow Sub Agent Activity. DTS uses +that tag as the primary label in orchestration lists, sequence/flow views, and +detail panels while retaining the registered function name in metadata. + +The orchestration label is derived deterministically as +`-orchestration`. Here, `agent_name` is the existing workflow +session value currently populated from the resolved agent slug on production +invocation paths. The runtime does not ask the LLM to invent a workflow title +and does not add a field to `start_workflow`. + +Tool Activities use the workflow tool name. Workflow Sub Agent Activities use +the authorized agent slug. Expanded `for_each` instances intentionally share +the same display name; their distinct runtime task IDs remain in the Activity +input and details. Timer tasks are unchanged. + +The pinned Durable Functions client exposes orchestration tags through +`schedule_new_orchestration`, so workflow startup moves from deprecated +`start_new(..., client_input=...)` to +`schedule_new_orchestration(..., input=..., tags=...)`. + +The Azure Functions one-argument compatibility orchestration context does not +expose Activity tags. The engine therefore registers its orchestrator using the +supported native two-argument Durable Task contract. The orchestration payload +becomes the second argument; native `call_activity(..., input=..., tags=...)` +and module-level task combinators replace compatibility-only helpers. The +engine explicitly normalizes the native replay-safe UTC timestamp before +comparing it with timezone-aware absolute wait deadlines. Activity names, +inputs, authorization, scheduling order, status payloads, workflow IDs, and +results remain unchanged. + ## 5. Decisions log | # | Decision | Options considered | Choice | Decided by | Date | @@ -1007,6 +1046,10 @@ The Dynamic Workflow sample for this extension must demonstrate: | 63 | Built-in skill packaging | Generate content in code / external public file / packaged workflow asset | Store `SKILL.md` under `workflows/skills/data-driven-workflows`, resolve relative to `integration.py`, add `workflows/skills/**` package data, and verify a built wheel | Agent, architecture review | 2026-08-19 | | 64 | Legacy Durable payload decoding | Require all new keys / revalidate via Pydantic / optional typed keys with one compatibility boundary | Mark dynamic keys `NotRequired`, apply defaults once at the persisted JSON boundary, and trust the previously validated payload internally | Agent, architecture review | 2026-08-19 | | 65 | Progressive-disclosure selection pointer | Keep a qualified shared-addendum pointer / rely on public docs / use only MAF skill metadata | Keep the narrow load condition in the Skill description and remove the shared-addendum pointer after E2E showed that mentioning the Skill there caused fixed DAGs to load it speculatively | Agent, E2E evidence | 2026-08-19 | +| 66 | Record DTS display names | Create a new FRD / evolve FRD 0004 | Evolve FRD 0004 because display tags are a small operational improvement to the existing Dynamic Workflows engine with no new authoring contract | Human (TsuyoshiUshio) | 2026-09-03 | +| 67 | Orchestration display label | LLM-authored title / agent display name / existing workflow session `agent_name` | Use `-orchestration` so labels are deterministic and require no LLM-facing schema change | Human (TsuyoshiUshio) | 2026-09-03 | +| 68 | Activity display label | Runtime Activity name / task id / execution target | Use the workflow tool name for tool Activities and the authorized agent slug for Workflow Sub Agent Activities | Human + Agent | 2026-09-03 | +| 69 | Durable Activity-tag integration | Access compatibility-context internals / wait for wrapper support / native two-argument orchestrator | Use the public native Durable Task orchestrator contract exposed by the pinned package; preserve existing execution semantics and cover the context migration in regression tests | Human + Agent | 2026-09-03 | ## 6. Test plan @@ -1140,6 +1183,17 @@ The Dynamic Workflow sample for this extension must demonstrate: - a sample discovers a collection, dynamically fans out, skips one item, and aggregates results; - the scenario completes with deterministic output on Azure Storage and DTS. +- [ ] Evolution #DTS display names: + - `tests/test_workflow_registry.py` verifies + `schedule_new_orchestration(..., input=..., tags=...)` and the deterministic + `-orchestration` label; + - `tests/test_workflow_engine.py` drives the native two-argument orchestrator + contract and verifies tool/Sub Agent tags on static, dynamic, and expanded + Activities; + - cancellation, timers, custom status, authorization, deterministic ordering, + and absolute/relative wait deadlines retain their existing behavior; + - emulator and Azure DTS runs show display tags in place of shared registered + function names. ## 7. Docs impact @@ -1172,6 +1226,9 @@ The Dynamic Workflow sample for this extension must demonstrate: project-discovered skills, and direct/delegated capability paths. - [ ] Evolution #1276: update the selected workflow sample and its README with a collection-driven fan-out/fan-in scenario. +- [ ] Evolution #DTS display names: update `docs/workflows.md` and + `docs/architecture.md` with derived orchestration/Activity labels and the + native orchestrator execution contract. ## 8. Status & sign-off diff --git a/docs/workflows.md b/docs/workflows.md index daefe8b5..2cc9a86c 100644 --- a/docs/workflows.md +++ b/docs/workflows.md @@ -802,7 +802,10 @@ agent-wide throttle. - **Durable Task Scheduler portal** — when the app's `host.json` is configured with the DTS `storageProvider`, each workflow appears as a queryable instance with per-task state and retry - history. + history. The runtime labels each orchestration + `-orchestration`, each tool Activity with its workflow tool name, + and each Workflow Sub Agent Activity with its agent slug. DTS retains the + shared registered Function name in the item's details. - **`custom_status`** — the orchestration emits a low-cost polling summary. Static plans return a concise string (`"3/7 tasks done, current=summarize"`); dynamically controlled plans return the structured `schema_version: 2` diff --git a/src/azure_functions_agents/workflows/engine.py b/src/azure_functions_agents/workflows/engine.py index a4fa67dc..a7d2b180 100644 --- a/src/azure_functions_agents/workflows/engine.py +++ b/src/azure_functions_agents/workflows/engine.py @@ -8,7 +8,7 @@ refs see something useful. Cooperative cancel is implemented as a single ``wait_for_external_event`` -("cancel") task that races the wave via ``context.task_any``. When the +("cancel") task that races the wave via ``when_any``. When the event fires we return a ``canceled=True`` envelope and stop scheduling. The Durable runtime_status remains ``Completed`` (Durable doesn't have a first-class cooperative-cancel terminal state); the tool-facing @@ -24,11 +24,12 @@ import json from collections.abc import Generator, Mapping from dataclasses import dataclass +from datetime import UTC from typing import Any, Literal, NotRequired, TypedDict, cast import azure.durable_functions as df import azure.functions as func -from durabletask.task import CancellableTask, Task +from durabletask.task import CancellableTask, OrchestrationContext, Task, when_any from azure_functions_agents._logger import logger from azure_functions_agents.registration.catalog import AgentCatalog @@ -103,7 +104,7 @@ def _run_echo(args: dict[str, Any]) -> dict[str, Any]: ) -def _wait_deadline(context: df.DurableOrchestrationContext, task: Mapping[str, Any]) -> Any: +def _wait_deadline(context: OrchestrationContext, task: Mapping[str, Any]) -> Any: """Compute the absolute UTC deadline for a wait task. Validation already enforced exactly one of ``duration`` / ``until`` @@ -114,6 +115,8 @@ def _wait_deadline(context: df.DurableOrchestrationContext, task: Mapping[str, A clock at submit time, which can drift between submit and execution. """ now = context.current_utc_datetime + if now.tzinfo is None: + now = now.replace(tzinfo=UTC) if task.get("duration") is not None: delta = parse_iso8601_duration(task["duration"]) deadline = now + delta @@ -128,7 +131,7 @@ def _wait_deadline(context: df.DurableOrchestrationContext, task: Mapping[str, A def _await_wave( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, cancel_task: Task[Any], wave_tasks: list[Task[Any]], ) -> Generator[Task[Any], Task[Any], list[Any] | None]: @@ -140,7 +143,7 @@ def _await_wave( outcomes: dict[int, Any] = {} pending = list(range(len(wave_tasks))) while pending: - winner = yield context.task_any( + winner = yield when_any( [cancel_task, *(wave_tasks[index] for index in pending)] ) if winner is cancel_task: @@ -177,7 +180,7 @@ def _plan_is_dynamic(tasks: list[WorkflowTaskInput]) -> bool: def _run_static_workflow( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, payload: WorkflowPayload, tasks: list[WorkflowTaskInput], ) -> Any: @@ -230,13 +233,14 @@ def _run_static_workflow( wave_tasks.append( context.call_activity( _ACTIVITY_NAME, - { + input={ "id": tid, "tool": task["tool"], "args": resolved_args, "workflow_agent_slug": workflow_agent_slug, "workflow_id": context.instance_id, }, + tags={"durabletask.displayName": task["tool"]}, ) ) wave_specs.append({"id": tid, "type": TOOL_TASK_TYPE}) @@ -254,13 +258,14 @@ def _run_static_workflow( wave_tasks.append( context.call_activity( SUB_AGENT_ACTIVITY_NAME, - { + input={ "id": tid, "agent": task["agent"], "task": resolved_task, "workflow_id": context.instance_id, "workflow_agent_slug": workflow_agent_slug, }, + tags={"durabletask.displayName": task["agent"]}, ) ) wave_specs.append({"id": tid, "type": SUB_AGENT_TASK_TYPE}) @@ -516,14 +521,14 @@ def _resolve_dynamic_args( def _publish_dynamic_status( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, state: _DynamicWorkflowState, ) -> None: context.set_custom_status(_dynamic_status(state)) def _dynamic_failure( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, state: _DynamicWorkflowState, *, error: str, @@ -566,7 +571,7 @@ def _aggregate_dynamic_node(state: _DynamicWorkflowState, logical_id: str) -> No def _materialize_for_each_node( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, state: _DynamicWorkflowState, logical_id: str, task: WorkflowTaskInput, @@ -684,7 +689,7 @@ def _materialize_for_each_node( def _materialize_normal_node( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, state: _DynamicWorkflowState, logical_id: str, task: WorkflowTaskInput, @@ -744,7 +749,7 @@ def _materialize_normal_node( def _materialize_ready_nodes( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, state: _DynamicWorkflowState, ) -> dict[str, Any] | None: progressed = True @@ -821,7 +826,7 @@ def _collect_runnable_instances( def _dispatch_dynamic_wave( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, state: _DynamicWorkflowState, wave: list[_MaterializedInstance], ) -> list[Task[Any]]: @@ -838,13 +843,14 @@ def _dispatch_dynamic_wave( wave_tasks.append( context.call_activity( _ACTIVITY_NAME, - { + input={ "id": instance["instance_id"], "tool": task["tool"], "args": instance["resolved"], "workflow_agent_slug": state.workflow_agent_slug, "workflow_id": context.instance_id, }, + tags={"durabletask.displayName": task["tool"]}, ) ) instance["kind"] = "activity" @@ -857,13 +863,14 @@ def _dispatch_dynamic_wave( wave_tasks.append( context.call_activity( SUB_AGENT_ACTIVITY_NAME, - { + input={ "id": instance["instance_id"], "agent": task["agent"], "task": instance["resolved"], "workflow_id": context.instance_id, "workflow_agent_slug": state.workflow_agent_slug, }, + tags={"durabletask.displayName": task["agent"]}, ) ) instance["kind"] = "activity" @@ -925,7 +932,7 @@ def _apply_dynamic_wave_results( def _run_dynamic_workflow( - context: df.DurableOrchestrationContext, + context: OrchestrationContext, payload: WorkflowPayload, tasks: list[WorkflowTaskInput], ) -> Any: @@ -1164,8 +1171,11 @@ async def agents_workflow_run_sub_agent( json.dumps(result) return result - @bp.orchestration_trigger(context_name="context") # type: ignore[arg-type] - def agents_workflow_orchestrator(context: df.DurableOrchestrationContext) -> Any: + @bp.orchestration_trigger(context_name="context") + def agents_workflow_orchestrator( + context: OrchestrationContext, + raw_payload: WorkflowPayload | None, + ) -> Any: """Execute a workflow plan, selecting the static or dynamic scheduler. A plan is *static* when no task carries a ``when`` predicate or a @@ -1182,7 +1192,6 @@ def agents_workflow_orchestrator(context: df.DurableOrchestrationContext) -> Any - No I/O outside ``call_activity`` / ``create_timer`` / ``wait_for_external_event``. """ - raw_payload: Any = context.get_input() if raw_payload is None: payload: WorkflowPayload = { "tasks": [], diff --git a/src/azure_functions_agents/workflows/tools.py b/src/azure_functions_agents/workflows/tools.py index 743be3f6..fe40d1ff 100644 --- a/src/azure_functions_agents/workflows/tools.py +++ b/src/azure_functions_agents/workflows/tools.py @@ -457,10 +457,10 @@ async def start_workflow( ) try: - returned_id = await session.durable_client.start_new( + returned_id = await session.durable_client.schedule_new_orchestration( ORCHESTRATOR_NAME, instance_id=instance_id, - client_input={ + input={ "tasks": plan_to_activity_inputs(plan), "workflow_agent_slug": session.workflow_agent_slug, "workflow_agent": workflow_agent, @@ -469,10 +469,14 @@ async def start_workflow( "allowed_subagents": sorted(policy.allowed_subagents), }, }, + tags={ + "durabletask.displayName": f"{session.agent_name}-orchestration" + }, ) except Exception: logger.exception( - "start_workflow: client.start_new failed workflow_agent=%s session=%s", + "start_workflow: client.schedule_new_orchestration failed " + "workflow_agent=%s session=%s", session.workflow_agent_slug, session.session_id, ) diff --git a/tests/test_workflow_engine.py b/tests/test_workflow_engine.py index d656e500..21d1004a 100644 --- a/tests/test_workflow_engine.py +++ b/tests/test_workflow_engine.py @@ -267,6 +267,7 @@ def __init__(self, result: Any = None) -> None: self._result = result self.is_complete = True self.cancelled = False + self._parent = None @property def result(self) -> Any: @@ -293,41 +294,36 @@ def __init__( self._input = {"workflow_agent_slug": "coordinator", "tasks": tasks} self._result_for = result_for self.calls: list[tuple[str, dict[str, Any]]] = [] + self.activity_tags: list[tuple[str, dict[str, str]]] = [] self.last_wave = _Task([]) self.cancel_task = _Task() self.statuses: list[str] = [] self.selections = 0 - def get_input(self) -> dict[str, Any]: - return self._input - def wait_for_external_event(self, name: str) -> _Task: assert name == engine.CANCEL_EVENT_NAME return self.cancel_task - def call_activity(self, name: str, payload: dict[str, Any]) -> _Task: - self.calls.append((name, payload)) - return _Task(self._result_for(name, payload)) - - def task_all(self, tasks: list[_Task]) -> _Task: - self.last_wave = _Task([task.result for task in tasks]) - return self.last_wave - - def task_any(self, tasks: list[_Task]) -> _Task: - self.selections += 1 - selection = _Task() - selection.candidates = list(tasks) - self.last_wave = selection - return selection + def call_activity( + self, + name: str, + *, + input: dict[str, Any], + tags: dict[str, str], + ) -> _Task: + self.calls.append((name, input)) + self.activity_tags.append((name, tags)) + return _Task(self._result_for(name, input)) def set_custom_status(self, status: str) -> None: self.statuses.append(status) def _drive(context: _FakeOrchestrationContext, selection: Any) -> Any: - candidates = getattr(selection, "candidates", None) + candidates = getattr(selection, "_tasks", None) if candidates is None: return selection + context.selections += 1 if getattr(context, "cancel_next", False): context.cancel_next = False return context.cancel_task @@ -338,10 +334,10 @@ def _drive(context: _FakeOrchestrationContext, selection: Any) -> Any: def _run_orchestrator( - orchestrator: Callable[[Any], Any], + orchestrator: Callable[[Any, Any], Any], context: _FakeOrchestrationContext, ) -> dict[str, Any]: - generator = orchestrator(context) + generator = orchestrator(context, context._input) try: selection = next(generator) while True: @@ -351,10 +347,10 @@ def _run_orchestrator( def _drive_one_wave(generator: Any, context: _FakeOrchestrationContext, selection: Any) -> Any: - remaining = [task for task in selection.candidates if task is not context.cancel_task] + remaining = [task for task in selection._tasks if task is not context.cancel_task] while remaining: selection = generator.send(remaining.pop(0)) - candidates = getattr(selection, "candidates", []) + candidates = getattr(selection, "_tasks", []) remaining = [task for task in remaining if task in candidates] return selection @@ -448,6 +444,20 @@ def result_for(name: str, payload: dict[str, Any]) -> dict[str, Any]: payload["workflow_agent_slug"] == "coordinator" for _, payload in context.calls ) + assert context.activity_tags == [ + ( + engine.SUB_AGENT_ACTIVITY_NAME, + {"durabletask.displayName": "pr_status_analyst"}, + ), + ( + engine.SUB_AGENT_ACTIVITY_NAME, + {"durabletask.displayName": "pr_status_analyst"}, + ), + ( + engine.SUB_AGENT_ACTIVITY_NAME, + {"durabletask.displayName": "report_writer"}, + ), + ] assert context.statuses == [ "0/3 tasks done, running=analyze_117,analyze_118", "2/3 tasks done, next=report", @@ -486,6 +496,12 @@ def test_orchestrator_threads_workflow_agent_slug_to_tool_activity() -> None: }, ) ] + assert context.activity_tags == [ + ( + "agents_workflow_run_tool", + {"durabletask.displayName": "publish"}, + ) + ] def test_tool_activity_reauthorizes_current_agent_policy() -> None: @@ -640,12 +656,19 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self.activity_calls = 0 - def call_activity(self, name: str, payload: dict[str, Any]) -> _Task: + def call_activity( + self, + name: str, + *, + input: dict[str, Any], + tags: dict[str, str], + ) -> _Task: self.activity_calls += 1 if self.activity_calls == 2: - self.calls.append((name, payload)) + self.calls.append((name, input)) + self.activity_tags.append((name, tags)) return _Task(RuntimeError("dynamic activity failed")) - return super().call_activity(name, payload) + return super().call_activity(name, input=input, tags=tags) tasks = [ { @@ -1001,6 +1024,11 @@ def result_for(name: str, payload: dict[str, Any]) -> dict[str, Any]: "analyze[0]", "analyze[2]", ] + assert context.activity_tags == [ + (engine._ACTIVITY_NAME, {"durabletask.displayName": "collect"}), + (engine._ACTIVITY_NAME, {"durabletask.displayName": "at"}), + (engine._ACTIVITY_NAME, {"durabletask.displayName": "at"}), + ] def test_empty_expansion_aggregates_immediately() -> None: @@ -1643,7 +1671,7 @@ def result_for(name: str, payload: dict[str, Any]) -> dict[str, Any]: context.cancel_task.result = "user-request" orchestrator = _registered_function(engine.ORCHESTRATOR_NAME) - gen = orchestrator(context) + gen = orchestrator(context, context._input) selection = next(gen) # yields the t1 wave _drive_one_wave(gen, context, selection) # completes t1, expands w1, dispatches its timer result: dict[str, Any] = {} @@ -1701,7 +1729,7 @@ def result_for(name: str, payload: dict[str, Any]) -> dict[str, Any]: context.cancel_task.result = "user-request" orchestrator = _registered_function(engine.ORCHESTRATOR_NAME) - gen = orchestrator(context) + gen = orchestrator(context, context._input) selection = next(gen) # discover selection = _drive_one_wave(gen, context, selection) # first inspect wave _drive_one_wave(gen, context, selection) # final inspect instance diff --git a/tests/test_workflow_registry.py b/tests/test_workflow_registry.py index 9ca58afa..5a3a5d25 100644 --- a/tests/test_workflow_registry.py +++ b/tests/test_workflow_registry.py @@ -73,7 +73,7 @@ def __init__( class _FailingDurableClient: secret = "durable storage account internal details" - async def start_new(self, *args, **kwargs): + async def schedule_new_orchestration(self, *args, **kwargs): raise RuntimeError(self.secret) async def get_status(self, *args, **kwargs): @@ -98,7 +98,7 @@ def __init__(self, statuses): async def get_status_all(self, *args, **kwargs): return self.statuses - async def start_new(self, *args, **kwargs): + async def schedule_new_orchestration(self, *args, **kwargs): self.started = True self.start_kwargs = kwargs return kwargs["instance_id"] @@ -801,12 +801,15 @@ async def test_start_workflow_threads_workflow_agent_slug_into_durable_input() - ) assert "workflow_id" in json.loads(result) - assert client.start_kwargs["client_input"]["workflow_agent_slug"] == "incident" - assert client.start_kwargs["client_input"]["workflow_agent"] == { + assert client.start_kwargs["input"]["workflow_agent_slug"] == "incident" + assert client.start_kwargs["input"]["workflow_agent"] == { "workflow_agent_slug": "incident", "session_id": "session-1", "agent_name": "Incident", } + assert client.start_kwargs["tags"] == { + "durabletask.displayName": "Incident-orchestration" + } def test_start_workflow_params_survive_framework_default_materialization() -> None: @@ -1017,8 +1020,8 @@ def __init__(self): async def get_status_all(self, *args, **kwargs): return [] - async def start_new(self, *args, **kwargs): - self.client_input = kwargs["client_input"] + async def schedule_new_orchestration(self, *args, **kwargs): + self.client_input = kwargs["input"] return kwargs["instance_id"]