Evan Mattson
4aa737eee5
* Harden functional workflow continuation authority Use a versioned opaque single-use token on WorkflowRunResult, validate it before request correlation, consume it immediately before replayed user code, and rotate it on each pause. Carry the same explicit authority through streaming and non-streaming FunctionalWorkflowAgent responses. Files changed: functional workflow/runtime result APIs, functional HITL regression tests, core agent guidance, and the functional HITL sample. Next iteration: enforce pending-state overlap and token-authorized abandonment, then document and test checkpoint authorization boundaries. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Enforce one pending functional continuation Reject fresh messages and checkpoint restores while an in-memory continuation is pending. Add token-authorized abandonment on FunctionalWorkflow and FunctionalWorkflowAgent, and clear retained replay state atomically when authority is consumed while preserving the active message for token rotation and checkpoints. Files changed: functional workflow runtime and agent adapter, functional lifecycle regression tests, and core workflow guidance. Next iteration: preserve and document authorized checkpoint continuation boundaries. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Preserve authorized functional checkpoint continuation Treat checkpoint restore as a host- and storage-authorized path independent of process-local continuation tokens, and issue fresh authority whenever restored execution pauses again. Cover default and per-run storage, deterministic and custom request IDs, token rotation, and checkpoint-plus-response restore. Files changed: functional workflow and checkpoint interface guidance, functional checkpoint lifecycle tests, the functional HITL sample, and core workflow guidance. Next iteration: run the final repository-wide Python validation gates. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Validate Python continuation hardening Run the complete Python workspace checks, aggregate coverage suite, repository hooks, and core package build from the final combined worktree. Keep the validation iteration code-neutral because all gates pass without corrective changes. Files changed: none; this commit records the final validation gate. Blockers: none. Next iteration: no remaining AFK tasks. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Handle functional checkpoint continuation failures Publish retained continuation state only after checkpoint persistence succeeds, and cover reuse after a transient save failure. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Address functional continuation review findings Add owner recovery for lost tokens, harden malformed token validation, preserve consistent failure surfaces, and keep agent pending state aligned with resumable workflow state. Document process-local single-use continuation semantics and extend regression coverage across direct, streaming, checkpoint, and agent paths. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Handle functional continuation cancellation Release the workflow run guard when cancellation interrupts resumed user code while keeping the single-use continuation token consumed. Replace sample assertions with explicit runtime checks and add cancellation regression coverage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Simplify functional workflow instance isolation Remove continuation-token handling and align functional workflows with the graph workflow ownership model: one stateful instance per logical caller or session. Add create_instance() for independent callers, document the ownership contract, and cover pending-state isolation between instances. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Scope functional workflow checkpoint storage Do not inherit checkpoint storage when creating an independent workflow instance. Allow hosts to provide an explicitly caller-scoped storage adapter and document that shared checkpoint access requires host authorization and tenant isolation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 * Require building functional workflow instances Make @workflow return a stateless FunctionalWorkflowDefinition and require build() before run() or as_agent(). This aligns functional workflows with the graph definition/build lifecycle and prevents module-level decorated definitions from retaining caller state. Move checkpoint configuration to build(), export the definition type, migrate samples, and cover isolated built instances. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78 --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: a8f47743-1cdc-4924-8e1b-667d0d790b78
59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
# Copyright (c) Microsoft. All rights reserved.
|
|
|
|
"""Basic sequential pipeline using the functional workflow API.
|
|
|
|
The simplest possible workflow: plain async functions orchestrated by @workflow.
|
|
No @step decorator needed — just write Python.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from agent_framework import workflow
|
|
|
|
|
|
# These are plain async functions — no decorators needed.
|
|
# They run normally inside the workflow, just like any other Python function.
|
|
async def fetch_data(url: str) -> dict[str, str | int]:
|
|
"""Simulate fetching data from a URL."""
|
|
return {"url": url, "content": f"Data from {url}", "status": 200}
|
|
|
|
|
|
async def transform_data(data: dict[str, str | int]) -> str:
|
|
"""Transform raw data into a summary string."""
|
|
return f"[{data['status']}] {data['content']}"
|
|
|
|
|
|
# @workflow turns this async function into a stateless workflow definition.
|
|
# Call .build() to create a stateful FunctionalWorkflow with:
|
|
# - .run() that returns a WorkflowRunResult with events and outputs
|
|
# - .run(stream=True) for streaming events in real time
|
|
# - .as_agent() to use this workflow anywhere an agent is expected
|
|
#
|
|
# The function's first parameter receives the input from .run("...").
|
|
# Add a `ctx: RunContext` parameter only if you need HITL, state, or custom events.
|
|
@workflow
|
|
async def data_pipeline(url: str) -> str:
|
|
"""A simple sequential data pipeline."""
|
|
raw = await fetch_data(url)
|
|
summary = await transform_data(raw)
|
|
|
|
# This is just a function — plain Python works between calls.
|
|
# No need to wrap every operation in a separate async function.
|
|
is_valid = len(summary) > 0 and "[200]" in summary
|
|
tag = "VALID" if is_valid else "INVALID"
|
|
|
|
# Returning a value automatically emits it as an output.
|
|
# Callers retrieve it via result.get_outputs().
|
|
return f"[{tag}] {summary}"
|
|
|
|
|
|
async def main():
|
|
workflow_instance = data_pipeline.build()
|
|
result = await workflow_instance.run("https://example.com/api/data")
|
|
print("Output:", result.get_outputs()[0])
|
|
print("State:", result.get_final_state())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|