From 1fb62f9d45ebb781f8569d46334a60398e7717af Mon Sep 17 00:00:00 2001 From: Jerry Yan <792602257@qq.com> Date: Mon, 6 Jul 2026 23:15:30 +0800 Subject: [PATCH] workflow --- .../workflow-orchestration-runtime/tasks.md | 88 ++-- pyproject.toml | 1 + tests/test_smoke.py | 1 + tests/test_workflow_conditions.py | 83 ++++ tests/test_workflow_models.py | 84 ++++ tests/test_workflow_runner.py | 387 ++++++++++++++++++ tests/test_workflow_skill_exec.py | 79 ++++ tests/test_workflow_store.py | 67 +++ workflow/__init__.py | 27 ++ workflow/conditions.py | 128 ++++++ workflow/config.py | 45 ++ workflow/models.py | 277 +++++++++++++ workflow/runner.py | 321 +++++++++++++++ workflow/skill_exec.py | 77 ++++ workflow/store.py | 244 +++++++++++ 15 files changed, 1865 insertions(+), 44 deletions(-) create mode 100644 tests/test_workflow_conditions.py create mode 100644 tests/test_workflow_models.py create mode 100644 tests/test_workflow_runner.py create mode 100644 tests/test_workflow_skill_exec.py create mode 100644 tests/test_workflow_store.py create mode 100644 workflow/__init__.py create mode 100644 workflow/conditions.py create mode 100644 workflow/config.py create mode 100644 workflow/models.py create mode 100644 workflow/runner.py create mode 100644 workflow/skill_exec.py create mode 100644 workflow/store.py diff --git a/openspec/changes/workflow-orchestration-runtime/tasks.md b/openspec/changes/workflow-orchestration-runtime/tasks.md index d912654..17ada2c 100644 --- a/openspec/changes/workflow-orchestration-runtime/tasks.md +++ b/openspec/changes/workflow-orchestration-runtime/tasks.md @@ -1,67 +1,67 @@ ## 1. Package scaffolding -- [ ] 1.1 Create the `workflow/` package (`__init__.py`, `models.py`, `conditions.py`, `skill_exec.py`, `runner.py`, `store.py`, `config.py`) -- [ ] 1.2 Add `workflow*` to `[tool.setuptools.packages.find].include` in `pyproject.toml` (no new third-party dependency; `sqlite3` is stdlib, already used by `storage/task_metadata.py`) -- [ ] 1.3 Add Workflow Runtime configuration in `workflow/config.py`: default poll interval, default wait-step timeout, default `WorkflowStore` db path (`workflows/workflows.sqlite3`) -- [ ] 1.4 Extend the project's smoke test (that imports every package) to import `workflow` +- [x] 1.1 Create the `workflow/` package (`__init__.py`, `models.py`, `conditions.py`, `skill_exec.py`, `runner.py`, `store.py`, `config.py`) +- [x] 1.2 Add `workflow*` to `[tool.setuptools.packages.find].include` in `pyproject.toml` (no new third-party dependency; `sqlite3` is stdlib, already used by `storage/task_metadata.py`) +- [x] 1.3 Add Workflow Runtime configuration in `workflow/config.py`: default poll interval, default wait-step timeout, default `WorkflowStore` db path (`workflows/workflows.sqlite3`) +- [x] 1.4 Extend the project's smoke test (that imports every package) to import `workflow` ## 2. Workflow and step data model (capability: workflow-orchestration) -- [ ] 2.1 Implement `workflow/models.py`: `ConditionSpec{kind: str, params: dict}`, the four step dataclasses (`PlannedGoalStep{step_id, goal, next_step_id}`, `SkillInvocationStep{step_id, skill_id, args, next_step_id}`, `WaitForConditionStep{step_id, condition, timeout_seconds, poll_interval_seconds, next_step_id}`, `BranchStep{step_id, condition, on_true, on_false}`), and the `WorkflowStep` union type -- [ ] 2.2 Implement `WorkflowDefinition{id, name, steps: list[WorkflowStep], entry_step_id}` with a constructor-time validation pass rejecting duplicate `step_id`s and any `next_step_id`/`on_true`/`on_false`/`entry_step_id` that does not reference an existing step in the same definition -- [ ] 2.3 Implement `WorkflowRun{id, definition_id, status, current_step_id, variables: dict, step_results: list[WorkflowStepResult], created_at, updated_at}` and `WorkflowStepResult{step_id, kind, success, detail, task_id: str | None, timestamp}`, with `to_dict()`/`from_dict()` helpers mirroring `core/models.py`'s style -- [ ] 2.4 Write unit tests for `WorkflowDefinition` construction: valid heterogeneous-step definition accepted; duplicate `step_id` rejected; dangling `next_step_id`/`on_true`/`on_false`/`entry_step_id` reference rejected +- [x] 2.1 Implement `workflow/models.py`: `ConditionSpec{kind: str, params: dict}`, the four step dataclasses (`PlannedGoalStep{step_id, goal, next_step_id}`, `SkillInvocationStep{step_id, skill_id, args, next_step_id}`, `WaitForConditionStep{step_id, condition, timeout_seconds, poll_interval_seconds, next_step_id}`, `BranchStep{step_id, condition, on_true, on_false}`), and the `WorkflowStep` union type +- [x] 2.2 Implement `WorkflowDefinition{id, name, steps: list[WorkflowStep], entry_step_id}` with a constructor-time validation pass rejecting duplicate `step_id`s and any `next_step_id`/`on_true`/`on_false`/`entry_step_id` that does not reference an existing step in the same definition +- [x] 2.3 Implement `WorkflowRun{id, definition_id, status, current_step_id, variables: dict, step_results: list[WorkflowStepResult], created_at, updated_at}` and `WorkflowStepResult{step_id, kind, success, detail, task_id: str | None, timestamp}`, with `to_dict()`/`from_dict()` helpers mirroring `core/models.py`'s style +- [x] 2.4 Write unit tests for `WorkflowDefinition` construction: valid heterogeneous-step definition accepted; duplicate `step_id` rejected; dangling `next_step_id`/`on_true`/`on_false`/`entry_step_id` reference rejected ## 3. WorkflowStore persistence (capability: workflow-orchestration) -- [ ] 3.1 Implement `workflow/store.py`: `WorkflowStore(db_path)` with schema creation for `workflow_definitions`, `workflow_runs`, `workflow_step_results` tables in its own SQLite file, following `storage/task_metadata.py`'s connect-per-call pattern -- [ ] 3.2 Implement `WorkflowStore.save_definition(definition)` / `get_definition(definition_id)` -- [ ] 3.3 Implement `WorkflowStore.create_run(definition_id, initial_variables) -> WorkflowRun` (status `pending`/`running`, `current_step_id` set to the definition's `entry_step_id`) -- [ ] 3.4 Implement `WorkflowStore.append_step_result(run_id, step_result)` and `WorkflowStore.update_run(run_id, *, status=None, current_step_id=None, variables=None)`, both persisting immediately (no in-memory-only buffering) -- [ ] 3.5 Implement `WorkflowStore.get_run(run_id) -> WorkflowRun` reconstructing the full run (status, current step, variables, ordered step results) from persisted rows -- [ ] 3.6 Write unit tests for `WorkflowStore`: create/get round-trip, step-result append ordering, run survives being re-opened via a fresh `WorkflowStore(same db_path)` instance (simulating a process restart) +- [x] 3.1 Implement `workflow/store.py`: `WorkflowStore(db_path)` with schema creation for `workflow_definitions`, `workflow_runs`, `workflow_step_results` tables in its own SQLite file, following `storage/task_metadata.py`'s connect-per-call pattern +- [x] 3.2 Implement `WorkflowStore.save_definition(definition)` / `get_definition(definition_id)` +- [x] 3.3 Implement `WorkflowStore.create_run(definition_id, initial_variables) -> WorkflowRun` (status `pending`/`running`, `current_step_id` set to the definition's `entry_step_id`) +- [x] 3.4 Implement `WorkflowStore.append_step_result(run_id, step_result)` and `WorkflowStore.update_run(run_id, *, status=None, current_step_id=None, variables=None)`, both persisting immediately (no in-memory-only buffering) +- [x] 3.5 Implement `WorkflowStore.get_run(run_id) -> WorkflowRun` reconstructing the full run (status, current step, variables, ordered step results) from persisted rows +- [x] 3.6 Write unit tests for `WorkflowStore`: create/get round-trip, step-result append ordering, run survives being re-opened via a fresh `WorkflowStore(same db_path)` instance (simulating a process restart) ## 4. ConditionEvaluator registry (capability: workflow-orchestration) -- [ ] 4.1 Implement `workflow/conditions.py`: `ConditionEvaluator` protocol/ABC with `evaluate(spec, *, scene, world_state) -> bool`, and a registry `dict[str, ConditionEvaluator]` -- [ ] 4.2 Implement the `scene_contains_text` evaluator (checks the most recent `Scene`'s elements/text for a target substring from `spec.params`) -- [ ] 4.3 Implement the `world_variable_equals` evaluator (compares `world_state.variables.get(spec.params["name"])` to `spec.params["value"]`; returns `False` — never raises — when `world_state` is `None` or the key is absent) -- [ ] 4.4 Implement the `elapsed_seconds` evaluator (returns `True` once at least `spec.params["seconds"]` have elapsed since the wait step started polling) -- [ ] 4.5 Implement the `step_result_success` evaluator (looks up a prior step's recorded `WorkflowStepResult.success` by `spec.params["step_id"]` from the run's step-result log) -- [ ] 4.6 Implement `evaluate_condition(spec, *, scene, world_state) -> bool` as the registry lookup entry point, raising a distinguishable `UnknownConditionKindError` for an unregistered `kind` (caught by the runner and turned into a failed step, not an uncaught exception) -- [ ] 4.7 Write unit tests for each built-in evaluator (true case, false case) and for the unregistered-kind error path +- [x] 4.1 Implement `workflow/conditions.py`: `ConditionEvaluator` protocol/ABC with `evaluate(spec, *, scene, world_state) -> bool`, and a registry `dict[str, ConditionEvaluator]` +- [x] 4.2 Implement the `scene_contains_text` evaluator (checks the most recent `Scene`'s elements/text for a target substring from `spec.params`) +- [x] 4.3 Implement the `world_variable_equals` evaluator (compares `world_state.variables.get(spec.params["name"])` to `spec.params["value"]`; returns `False` — never raises — when `world_state` is `None` or the key is absent) +- [x] 4.4 Implement the `elapsed_seconds` evaluator (returns `True` once at least `spec.params["seconds"]` have elapsed since the wait step started polling) +- [x] 4.5 Implement the `step_result_success` evaluator (looks up a prior step's recorded `WorkflowStepResult.success` by `spec.params["step_id"]` from the run's step-result log) +- [x] 4.6 Implement `evaluate_condition(spec, *, scene, world_state) -> bool` as the registry lookup entry point, raising a distinguishable `UnknownConditionKindError` for an unregistered `kind` (caught by the runner and turned into a failed step, not an uncaught exception) +- [x] 4.7 Write unit tests for each built-in evaluator (true case, false case) and for the unregistered-kind error path ## 5. Skill-invocation execution path (capability: workflow-orchestration) -- [ ] 5.1 Implement `workflow/skill_exec.py`: `validate_skill_args(skill, args) -> None` raising a descriptive error listing missing required parameters or unrecognized argument keys, called before any tool dispatch -- [ ] 5.2 Implement `resolve_skill_steps(skill, args) -> list[dict]` substituting `{param}` placeholders in the skill's stored tool-call template with validated `args` values -- [ ] 5.3 Implement `run_flow_template_skill(skill, args) -> list[StepResult]` calling `validate_skill_args`, `resolve_skill_steps`, then dispatching each resolved tool call through the same `tools/*` functions `Executor.execute()` already uses, collecting one `StepResult` per resolved step -- [ ] 5.4 Implement the `Skill.kind != "flow_template"` guard: reject with a step-definition error before attempting resolution -- [ ] 5.5 Define (or import, if `skill-learning-runtime`/`skill-catalog-subscription` is already implemented) the minimal `Skill`/`FlowTemplateSkill` dataclass shape `skill_exec.py` depends on, with a code comment flagging reconciliation once one of those changes lands -- [ ] 5.6 Write unit tests for `skill_exec`: valid-args resolution produces the expected resolved tool-call sequence; missing required parameter fails before any tool call; non-flow-template skill kind is rejected +- [x] 5.1 Implement `workflow/skill_exec.py`: `validate_skill_args(skill, args) -> None` raising a descriptive error listing missing required parameters or unrecognized argument keys, called before any tool dispatch +- [x] 5.2 Implement `resolve_skill_steps(skill, args) -> list[dict]` substituting `{param}` placeholders in the skill's stored tool-call template with validated `args` values +- [x] 5.3 Implement `run_flow_template_skill(skill, args) -> list[StepResult]` calling `validate_skill_args`, `resolve_skill_steps`, then dispatching each resolved tool call through the same `tools/*` functions `Executor.execute()` already uses, collecting one `StepResult` per resolved step +- [x] 5.4 Implement the `Skill.kind != "flow_template"` guard: reject with a step-definition error before attempting resolution +- [x] 5.5 Define (or import, if `skill-learning-runtime`/`skill-catalog-subscription` is already implemented) the minimal `Skill`/`FlowTemplateSkill` dataclass shape `skill_exec.py` depends on, with a code comment flagging reconciliation once one of those changes lands +- [x] 5.6 Write unit tests for `skill_exec`: valid-args resolution produces the expected resolved tool-call sequence; missing required parameter fails before any tool call; non-flow-template skill kind is rejected ## 6. WorkflowRunner orchestration (capability: workflow-orchestration) -- [ ] 6.1 Implement `workflow/runner.py`: `WorkflowRunner(store, task_runner_factory, ...)` with `run(definition, device_id, initial_variables=None) -> WorkflowRun` that creates a run via `WorkflowStore.create_run()` and drives steps until a terminal status -- [ ] 6.2 Implement the planned-goal step handler: construct a `core.models.Task(goal=step.goal, device_id=...)`, call `TaskRunner(...).run(task)`, map `task.status`/`task.failure_reason` to a `WorkflowStepResult` -- [ ] 6.3 Implement the skill-invocation step handler: look up the referenced skill by `skill_id`, call `skill_exec.run_flow_template_skill`, map the returned `StepResult`s to one aggregate `WorkflowStepResult` -- [ ] 6.4 Implement the wait-for-condition step handler: poll `conditions.evaluate_condition()` at `poll_interval_seconds` until `True` or `timeout_seconds` elapses; record success or a timeout failure -- [ ] 6.5 Implement the branch step handler: evaluate the condition once and set the run's next step to `on_true`/`on_false` accordingly, bypassing default sequential advancement -- [ ] 6.6 Implement default sequential advancement (no explicit `next_step_id`/branch target) as "the next step in `WorkflowDefinition.steps` order" for non-branch step kinds -- [ ] 6.7 After each step handler returns, call `WorkflowStore.append_step_result()` and `WorkflowStore.update_run()` (new `current_step_id`, and `status` if terminal) before advancing — no step is considered "done" until this checkpoint write completes -- [ ] 6.8 Implement `WorkflowRunner.resume(run_id) -> WorkflowRun`: load the persisted run via `WorkflowStore.get_run()`, and continue driving from its persisted `current_step_id`, skipping any step already present in the loaded `step_results` log -- [ ] 6.9 Implement `resume()` on an already-`completed`/`failed`/`cancelled` run as a no-op that returns the existing run state unchanged -- [ ] 6.10 Write unit tests for `WorkflowRunner.run()` covering: a linear planned-goal-only workflow reaching `completed`; a workflow with a failing planned-goal step reaching `failed` with a recorded reason; a skill-invocation step executing resolved tool calls (mocked `tools/*`); a wait step succeeding before timeout and timing out after -- [ ] 6.11 Write unit tests for branch step selection (true path and false path) and for default sequential advancement between non-branch steps +- [x] 6.1 Implement `workflow/runner.py`: `WorkflowRunner(store, task_runner_factory, ...)` with `run(definition, device_id, initial_variables=None) -> WorkflowRun` that creates a run via `WorkflowStore.create_run()` and drives steps until a terminal status +- [x] 6.2 Implement the planned-goal step handler: construct a `core.models.Task(goal=step.goal, device_id=...)`, call `TaskRunner(...).run(task)`, map `task.status`/`task.failure_reason` to a `WorkflowStepResult` +- [x] 6.3 Implement the skill-invocation step handler: look up the referenced skill by `skill_id`, call `skill_exec.run_flow_template_skill`, map the returned `StepResult`s to one aggregate `WorkflowStepResult` +- [x] 6.4 Implement the wait-for-condition step handler: poll `conditions.evaluate_condition()` at `poll_interval_seconds` until `True` or `timeout_seconds` elapses; record success or a timeout failure +- [x] 6.5 Implement the branch step handler: evaluate the condition once and set the run's next step to `on_true`/`on_false` accordingly, bypassing default sequential advancement +- [x] 6.6 Implement default sequential advancement (no explicit `next_step_id`/branch target) as "the next step in `WorkflowDefinition.steps` order" for non-branch step kinds +- [x] 6.7 After each step handler returns, call `WorkflowStore.append_step_result()` and `WorkflowStore.update_run()` (new `current_step_id`, and `status` if terminal) before advancing — no step is considered "done" until this checkpoint write completes +- [x] 6.8 Implement `WorkflowRunner.resume(run_id) -> WorkflowRun`: load the persisted run via `WorkflowStore.get_run()`, and continue driving from its persisted `current_step_id`, skipping any step already present in the loaded `step_results` log +- [x] 6.9 Implement `resume()` on an already-`completed`/`failed`/`cancelled` run as a no-op that returns the existing run state unchanged +- [x] 6.10 Write unit tests for `WorkflowRunner.run()` covering: a linear planned-goal-only workflow reaching `completed`; a workflow with a failing planned-goal step reaching `failed` with a recorded reason; a skill-invocation step executing resolved tool calls (mocked `tools/*`); a wait step succeeding before timeout and timing out after +- [x] 6.11 Write unit tests for branch step selection (true path and false path) and for default sequential advancement between non-branch steps ## 7. Resumability end-to-end validation (capability: workflow-orchestration) -- [ ] 7.1 Write an end-to-end test that runs a multi-step workflow through `WorkflowRunner`, stops after two steps (simulating a crash by discarding the in-memory `WorkflowRunner`/`TaskRunner` instances), constructs a fresh `WorkflowRunner`/`WorkflowStore` pointed at the same db file, and asserts `resume(run_id)` continues from the third step without re-invoking the first two steps' tool calls or delegated tasks -- [ ] 7.2 Write an end-to-end test asserting a `WorkflowRun`'s persisted `variables` and `step_results` log are readable and correctly ordered after a full run via `WorkflowStore.get_run()` -- [ ] 7.3 Write an end-to-end test combining a branch step with a wait step and a skill-invocation step in one workflow, asserting the run reaches `completed` via the expected branch path +- [x] 7.1 Write an end-to-end test that runs a multi-step workflow through `WorkflowRunner`, stops after two steps (simulating a crash by discarding the in-memory `WorkflowRunner`/`TaskRunner` instances), constructs a fresh `WorkflowRunner`/`WorkflowStore` pointed at the same db file, and asserts `resume(run_id)` continues from the third step without re-invoking the first two steps' tool calls or delegated tasks +- [x] 7.2 Write an end-to-end test asserting a `WorkflowRun`'s persisted `variables` and `step_results` log are readable and correctly ordered after a full run via `WorkflowStore.get_run()` +- [x] 7.3 Write an end-to-end test combining a branch step with a wait step and a skill-invocation step in one workflow, asserting the run reaches `completed` via the expected branch path ## 8. Composition safety checks and full-suite validation -- [ ] 8.1 Confirm no existing file under `runtime/`, `storage/`, `tools/`, or `core/` is modified by this change (composition via import only, per design.md's D1/D8/D9) -- [ ] 8.2 Write a test that runs a real (non-mocked) `runtime.task.TaskRunner` instance inside a `WorkflowRunner`-driven planned-goal step, guarding against silent drift in `agent-runtime`'s public `run(task) -> Task` contract this change composes over -- [ ] 8.3 Run the full test suite (`pytest`) and confirm every existing test in `tests/` passes unmodified, with only new `tests/test_workflow_*.py`-style files added +- [x] 8.1 Confirm no existing file under `runtime/`, `storage/`, `tools/`, or `core/` is modified by this change (composition via import only, per design.md's D1/D8/D9) +- [x] 8.2 Write a test that runs a real (non-mocked) `runtime.task.TaskRunner` instance inside a `WorkflowRunner`-driven planned-goal step, guarding against silent drift in `agent-runtime`'s public `run(task) -> Task` contract this change composes over +- [x] 8.3 Run the full test suite (`pytest`) and confirm every existing test in `tests/` passes unmodified, with only new `tests/test_workflow_*.py`-style files added diff --git a/pyproject.toml b/pyproject.toml index dd9b6de..3ef2931 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ include = [ "storage*", "tools*", "world*", + "workflow*", ] [tool.pytest.ini_options] diff --git a/tests/test_smoke.py b/tests/test_smoke.py index d60e13e..8e1f850 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -16,5 +16,6 @@ def test_imports_new_packages() -> None: "storage", "tools", "world", + "workflow", ): importlib.import_module(package) diff --git a/tests/test_workflow_conditions.py b/tests/test_workflow_conditions.py new file mode 100644 index 0000000..db3f237 --- /dev/null +++ b/tests/test_workflow_conditions.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +from datetime import timedelta +from types import SimpleNamespace + +import pytest + +from core.models import Bounds, Scene, SceneElement, utc_now +from workflow.conditions import UnknownConditionKindError, evaluate_condition +from workflow.models import ConditionSpec, WorkflowStepResult + + +def _scene(text: str) -> Scene: + return Scene( + width=10, + height=20, + elements=[ + SceneElement( + id="label", + type="text", + text=text, + bounds=Bounds(1, 2, 3, 4), + ) + ], + ) + + +def test_scene_contains_text_evaluator_true_and_false() -> None: + spec = ConditionSpec("scene_contains_text", {"text": "hello"}) + + assert evaluate_condition(spec, scene=_scene("Hello world"), world_state=None) + assert not evaluate_condition(spec, scene=_scene("Goodbye"), world_state=None) + + +def test_world_variable_equals_evaluator_true_false_and_absent() -> None: + spec = ConditionSpec("world_variable_equals", {"name": "ready", "value": True}) + + assert evaluate_condition( + spec, + scene=None, + world_state=SimpleNamespace(variables={"ready": True}), + ) + assert not evaluate_condition( + spec, + scene=None, + world_state=SimpleNamespace(variables={"ready": False}), + ) + assert not evaluate_condition(spec, scene=None, world_state=None) + + +def test_elapsed_seconds_evaluator_true_and_false() -> None: + spec = ConditionSpec("elapsed_seconds", {"seconds": 1}) + + assert evaluate_condition( + spec, + scene=None, + world_state=None, + started_at=utc_now() - timedelta(seconds=2), + ) + assert not evaluate_condition( + spec, + scene=None, + world_state=None, + started_at=utc_now(), + ) + + +def test_step_result_success_evaluator_true_and_false() -> None: + spec = ConditionSpec("step_result_success", {"step_id": "first"}) + results = [WorkflowStepResult("first", "planned_goal", True)] + + assert evaluate_condition(spec, scene=None, world_state=None, step_results=results) + assert not evaluate_condition( + ConditionSpec("step_result_success", {"step_id": "missing"}), + scene=None, + world_state=None, + step_results=results, + ) + + +def test_unknown_condition_kind_raises_distinguishable_error() -> None: + with pytest.raises(UnknownConditionKindError): + evaluate_condition(ConditionSpec("unknown"), scene=None, world_state=None) diff --git a/tests/test_workflow_models.py b/tests/test_workflow_models.py new file mode 100644 index 0000000..28048a0 --- /dev/null +++ b/tests/test_workflow_models.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +import pytest + +from workflow.models import ( + BranchStep, + ConditionSpec, + PlannedGoalStep, + SkillInvocationStep, + WaitForConditionStep, + WorkflowDefinition, +) + + +def test_workflow_definition_accepts_heterogeneous_steps() -> None: + definition = WorkflowDefinition( + name="chat workflow", + entry_step_id="start", + steps=[ + PlannedGoalStep("start", "open chat", next_step_id="skill"), + SkillInvocationStep("skill", "skill-1", {"message": "hello"}, "wait"), + WaitForConditionStep( + "wait", + ConditionSpec("scene_contains_text", {"text": "Done"}), + timeout_seconds=1, + poll_interval_seconds=0, + next_step_id="branch", + ), + BranchStep( + "branch", + ConditionSpec("world_variable_equals", {"name": "ok", "value": True}), + on_true="start", + on_false="skill", + ), + ], + ) + + assert definition.step_by_id("skill").kind == "skill_invocation" + + +def test_workflow_definition_rejects_duplicate_step_id() -> None: + with pytest.raises(ValueError, match="duplicate"): + WorkflowDefinition( + name="bad", + entry_step_id="same", + steps=[ + PlannedGoalStep("same", "one"), + PlannedGoalStep("same", "two"), + ], + ) + + +@pytest.mark.parametrize( + "definition", + [ + WorkflowDefinition, + ], +) +def test_workflow_definition_rejects_dangling_references(definition) -> None: + with pytest.raises(ValueError, match="unknown|entry"): + definition( + name="bad next", + entry_step_id="start", + steps=[PlannedGoalStep("start", "one", next_step_id="missing")], + ) + with pytest.raises(ValueError, match="unknown|entry"): + definition( + name="bad branch", + entry_step_id="branch", + steps=[ + BranchStep( + "branch", + ConditionSpec("elapsed_seconds", {"seconds": 0}), + on_true="missing", + on_false="branch", + ) + ], + ) + with pytest.raises(ValueError, match="entry"): + definition( + name="bad entry", + entry_step_id="missing", + steps=[PlannedGoalStep("start", "one")], + ) diff --git a/tests/test_workflow_runner.py b/tests/test_workflow_runner.py new file mode 100644 index 0000000..b341dd9 --- /dev/null +++ b/tests/test_workflow_runner.py @@ -0,0 +1,387 @@ +from __future__ import annotations + +from core.models import Bounds, Scene, SceneElement, Task +from runtime.executor import Executor, ExecutorConfig +from runtime.planner import Planner +from runtime.task import TaskRunner, TaskRunnerConfig +from skills_learning.models import FlowStep, FlowTemplateSkill, SkillMetadata +from skills_learning.store import SkillStore +from tests.fakes import PNG_10X20 +from workflow.models import ( + BranchStep, + ConditionSpec, + PlannedGoalStep, + SkillInvocationStep, + WaitForConditionStep, + WorkflowDefinition, +) +from workflow.runner import WorkflowRunner +from workflow.store import WorkflowStore + + +class FakeTaskRunner: + def __init__( + self, + calls: list[str], + *, + status: str = "completed", + failure_reason: str | None = None, + ) -> None: + self.calls = calls + self.status = status + self.failure_reason = failure_reason + + def run(self, task: Task) -> Task: + self.calls.append(task.goal) + task.status = self.status # type: ignore[assignment] + task.failure_reason = self.failure_reason + return task + + +def _scene(text: str = "Ready") -> Scene: + return Scene( + width=10, + height=20, + elements=[ + SceneElement( + id="label", + type="text", + text=text, + bounds=Bounds(1, 2, 3, 4), + ) + ], + ) + + +def _skill_store() -> tuple[SkillStore, FlowTemplateSkill]: + store = SkillStore() + skill = store.create_version( + FlowTemplateSkill( + metadata=SkillMetadata( + name="send message", + description="Send message", + kind="flow_template", + ), + steps=[FlowStep("input_text", {"text": "{message}"})], + parameters={"message": {"type": "string"}}, + ) + ) + return store, skill + + +def _store(tmp_path) -> WorkflowStore: + return WorkflowStore(tmp_path / "workflows.sqlite3") + + +def test_workflow_runner_linear_planned_goal_completes(tmp_path) -> None: + calls: list[str] = [] + definition = WorkflowDefinition( + name="linear", + entry_step_id="first", + steps=[ + PlannedGoalStep("first", "open app", next_step_id="second"), + PlannedGoalStep("second", "send message"), + ], + ) + runner = WorkflowRunner( + _store(tmp_path), + task_runner_factory=lambda: FakeTaskRunner(calls), + ) + + run = runner.run(definition, "phone") + + assert run.status == "completed" + assert calls == ["open app", "send message"] + assert [result.step_id for result in run.step_results] == ["first", "second"] + + +def test_workflow_runner_failing_planned_goal_marks_run_failed(tmp_path) -> None: + definition = WorkflowDefinition( + name="fail", + entry_step_id="first", + steps=[PlannedGoalStep("first", "open app")], + ) + runner = WorkflowRunner( + _store(tmp_path), + task_runner_factory=lambda: FakeTaskRunner( + [], + status="failed", + failure_reason="device offline", + ), + ) + + run = runner.run(definition, "phone") + + assert run.status == "failed" + assert run.step_results[0].detail["failure_reason"] == "device offline" + + +def test_workflow_runner_skill_invocation_executes_resolved_tool_calls(tmp_path) -> None: + skill_store, skill = _skill_store() + calls: list[dict[str, object]] = [] + definition = WorkflowDefinition( + name="skill", + entry_step_id="skill", + steps=[SkillInvocationStep("skill", skill.id, {"message": "hello"})], + ) + runner = WorkflowRunner( + _store(tmp_path), + skill_store=skill_store, + tools={"input_text": lambda **kwargs: calls.append(kwargs) or {"ok": True}}, + ) + + run = runner.run(definition, "phone") + + assert run.status == "completed" + assert calls == [{"text": "hello"}] + + +def test_workflow_runner_wait_step_succeeds_and_times_out(tmp_path) -> None: + success_definition = WorkflowDefinition( + name="wait success", + entry_step_id="wait", + steps=[ + WaitForConditionStep( + "wait", + ConditionSpec("scene_contains_text", {"text": "ready"}), + timeout_seconds=0.1, + poll_interval_seconds=0, + ) + ], + ) + success_runner = WorkflowRunner( + _store(tmp_path / "success"), + scene_provider=lambda: _scene("Ready"), + sleep_func=lambda seconds: None, + ) + + success_run = success_runner.run(success_definition, "phone") + + assert success_run.status == "completed" + + timeout_definition = WorkflowDefinition( + name="wait timeout", + entry_step_id="wait", + steps=[ + WaitForConditionStep( + "wait", + ConditionSpec("scene_contains_text", {"text": "missing"}), + timeout_seconds=0, + poll_interval_seconds=0, + ) + ], + ) + timeout_runner = WorkflowRunner( + _store(tmp_path / "timeout"), + scene_provider=lambda: _scene("Ready"), + sleep_func=lambda seconds: None, + ) + + timeout_run = timeout_runner.run(timeout_definition, "phone") + + assert timeout_run.status == "failed" + assert "timed out" in timeout_run.step_results[0].detail["reason"] + + +def test_workflow_runner_branch_paths_and_sequential_advancement(tmp_path) -> None: + true_calls: list[str] = [] + definition = WorkflowDefinition( + name="branch", + entry_step_id="branch", + steps=[ + BranchStep( + "branch", + ConditionSpec("world_variable_equals", {"name": "ready", "value": True}), + on_true="true-step", + on_false="false-step", + ), + PlannedGoalStep("true-step", "true path", next_step_id="end"), + PlannedGoalStep("false-step", "false path", next_step_id="end"), + WaitForConditionStep( + "end", + ConditionSpec("elapsed_seconds", {"seconds": 0}), + timeout_seconds=0, + poll_interval_seconds=0, + ), + ], + ) + true_runner = WorkflowRunner( + _store(tmp_path / "true"), + task_runner_factory=lambda: FakeTaskRunner(true_calls), + ) + + true_run = true_runner.run(definition, "phone", {"ready": True}) + + assert true_run.status == "completed" + assert true_calls == ["true path"] + assert [result.step_id for result in true_run.step_results] == [ + "branch", + "true-step", + "end", + ] + + false_calls: list[str] = [] + false_runner = WorkflowRunner( + _store(tmp_path / "false"), + task_runner_factory=lambda: FakeTaskRunner(false_calls), + ) + + false_run = false_runner.run(definition, "phone", {"ready": False}) + + assert false_run.status == "completed" + assert false_calls == ["false path"] + + +def test_workflow_runner_resume_skips_completed_steps_after_restart(tmp_path) -> None: + calls: list[str] = [] + db_path = tmp_path / "workflows.sqlite3" + definition = WorkflowDefinition( + name="resume", + entry_step_id="one", + steps=[ + PlannedGoalStep("one", "one"), + PlannedGoalStep("two", "two"), + PlannedGoalStep("three", "three"), + ], + ) + first_runner = WorkflowRunner( + WorkflowStore(db_path), + task_runner_factory=lambda: FakeTaskRunner(calls), + step_limit=2, + ) + + partial = first_runner.run(definition, "phone") + + assert partial.status == "running" + assert partial.current_step_id == "three" + assert calls == ["one", "two"] + + resumed_runner = WorkflowRunner( + WorkflowStore(db_path), + task_runner_factory=lambda: FakeTaskRunner(calls), + ) + resumed = resumed_runner.resume(partial.id) + + assert resumed.status == "completed" + assert calls == ["one", "two", "three"] + + +def test_workflow_run_persists_variables_and_ordered_results(tmp_path) -> None: + store = _store(tmp_path) + definition = WorkflowDefinition( + name="persist", + entry_step_id="one", + steps=[PlannedGoalStep("one", "one"), PlannedGoalStep("two", "two")], + ) + run = WorkflowRunner( + store, + task_runner_factory=lambda: FakeTaskRunner([]), + ).run(definition, "phone", {"contact": "Zhang San"}) + + loaded = store.get_run(run.id) + + assert loaded is not None + assert loaded.variables == {"contact": "Zhang San"} + assert [result.step_id for result in loaded.step_results] == ["one", "two"] + + +def test_workflow_runner_branch_wait_and_skill_combination(tmp_path) -> None: + skill_store, skill = _skill_store() + calls: list[dict[str, object]] = [] + definition = WorkflowDefinition( + name="combo", + entry_step_id="branch", + steps=[ + BranchStep( + "branch", + ConditionSpec("world_variable_equals", {"name": "ready", "value": True}), + on_true="wait", + on_false="skip", + ), + WaitForConditionStep( + "wait", + ConditionSpec("elapsed_seconds", {"seconds": 0}), + timeout_seconds=1, + poll_interval_seconds=0, + next_step_id="skill", + ), + SkillInvocationStep( + "skill", + skill.id, + {"message": "hello"}, + next_step_id="end", + ), + PlannedGoalStep("skip", "skip", next_step_id="end"), + WaitForConditionStep( + "end", + ConditionSpec("elapsed_seconds", {"seconds": 0}), + timeout_seconds=0, + poll_interval_seconds=0, + ), + ], + ) + runner = WorkflowRunner( + _store(tmp_path), + skill_store=skill_store, + tools={"input_text": lambda **kwargs: calls.append(kwargs) or {"ok": True}}, + ) + + run = runner.run(definition, "phone", {"ready": True}) + + assert run.status == "completed" + assert [result.step_id for result in run.step_results] == [ + "branch", + "wait", + "skill", + "end", + ] + assert calls == [{"text": "hello"}] + + +def test_resume_completed_run_is_noop(tmp_path) -> None: + calls: list[str] = [] + runner = WorkflowRunner( + _store(tmp_path), + task_runner_factory=lambda: FakeTaskRunner(calls), + ) + definition = WorkflowDefinition( + name="done", + entry_step_id="one", + steps=[PlannedGoalStep("one", "one")], + ) + completed = runner.run(definition, "phone") + + resumed = runner.resume(completed.id) + + assert resumed == completed + assert calls == ["one"] + + +def test_workflow_runner_composes_real_task_runner_contract(tmp_path) -> None: + scene = _scene("Ready") + + def task_runner_factory() -> TaskRunner: + return TaskRunner( + planner=Planner(), + executor=Executor( + tools={"describe_screen": lambda **kwargs: scene}, + config=ExecutorConfig(max_retries=1, backoff_seconds=0), + ), + config=TaskRunnerConfig(max_steps=3), + observer=lambda device_id: scene, + screenshot_provider=lambda device_id: PNG_10X20, + ) + + definition = WorkflowDefinition( + name="real task runner", + entry_step_id="goal", + steps=[PlannedGoalStep("goal", "inspect")], + ) + + run = WorkflowRunner( + _store(tmp_path), + task_runner_factory=task_runner_factory, + ).run(definition, "phone") + + assert run.status == "completed" + assert run.step_results[0].task_id is not None diff --git a/tests/test_workflow_skill_exec.py b/tests/test_workflow_skill_exec.py new file mode 100644 index 0000000..5831507 --- /dev/null +++ b/tests/test_workflow_skill_exec.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import pytest + +from skills_learning.models import FlowStep, FlowTemplateSkill, SkillMetadata +from workflow.skill_exec import ( + SkillExecutionError, + resolve_skill_steps, + run_flow_template_skill, +) + + +def _skill() -> FlowTemplateSkill: + return FlowTemplateSkill( + metadata=SkillMetadata( + name="send message", + description="Send a message", + kind="flow_template", + ), + steps=[ + FlowStep("input_text", {"text": "{message}"}), + FlowStep("tap", {"x": 1, "y": 2}), + ], + parameters={"message": {"type": "string"}}, + ) + + +def test_resolve_skill_steps_substitutes_valid_args() -> None: + resolved = resolve_skill_steps(_skill(), {"message": "hello"}) + + assert resolved == [ + {"tool_name": "input_text", "args": {"text": "hello"}}, + {"tool_name": "tap", "args": {"x": 1, "y": 2}}, + ] + + +def test_missing_required_parameter_fails_before_tool_call() -> None: + calls: list[str] = [] + + with pytest.raises(SkillExecutionError, match="missing required"): + run_flow_template_skill( + _skill(), + {}, + tools={"input_text": lambda **kwargs: calls.append("input_text")}, + ) + + assert calls == [] + + +def test_non_flow_template_skill_is_rejected() -> None: + skill = FlowTemplateSkill( + metadata=SkillMetadata( + name="knowledge", + description="Not executable", + kind="knowledge", + ) + ) + + with pytest.raises(SkillExecutionError, match="flow_template"): + resolve_skill_steps(skill, {}) + + +def test_run_flow_template_skill_dispatches_resolved_tools() -> None: + calls: list[tuple[str, dict[str, object]]] = [] + + results = run_flow_template_skill( + _skill(), + {"message": "hello"}, + tools={ + "input_text": lambda **kwargs: calls.append(("input_text", kwargs)) or {"ok": True}, + "tap": lambda **kwargs: calls.append(("tap", kwargs)) or {"ok": True}, + }, + ) + + assert [result.success for result in results] == [True, True] + assert calls == [ + ("input_text", {"text": "hello"}), + ("tap", {"x": 1, "y": 2}), + ] diff --git a/tests/test_workflow_store.py b/tests/test_workflow_store.py new file mode 100644 index 0000000..b19a65d --- /dev/null +++ b/tests/test_workflow_store.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from workflow.models import PlannedGoalStep, WorkflowDefinition, WorkflowStepResult +from workflow.store import WorkflowStore + + +def _definition() -> WorkflowDefinition: + return WorkflowDefinition( + name="linear", + entry_step_id="first", + steps=[ + PlannedGoalStep("first", "one", next_step_id="second"), + PlannedGoalStep("second", "two"), + ], + ) + + +def test_workflow_store_round_trips_definition_and_run(tmp_path) -> None: + store = WorkflowStore(tmp_path / "workflows.sqlite3") + definition = _definition() + + store.save_definition(definition) + run = store.create_run( + definition.id, + {"contact": "Zhang San"}, + device_id="phone", + ) + loaded = store.get_run(run.id) + + assert store.get_definition(definition.id) == definition + assert loaded is not None + assert loaded.status == "running" + assert loaded.current_step_id == "first" + assert loaded.variables == {"contact": "Zhang San"} + assert loaded.device_id == "phone" + + +def test_workflow_store_appends_step_results_in_order(tmp_path) -> None: + store = WorkflowStore(tmp_path / "workflows.sqlite3") + definition = _definition() + store.save_definition(definition) + run = store.create_run(definition.id, {}) + + store.append_step_result(run.id, WorkflowStepResult("first", "planned_goal", True)) + store.append_step_result(run.id, WorkflowStepResult("second", "planned_goal", True)) + + loaded = store.get_run(run.id) + assert loaded is not None + assert [result.step_id for result in loaded.step_results] == ["first", "second"] + + +def test_workflow_store_survives_reopen(tmp_path) -> None: + db_path = tmp_path / "workflows.sqlite3" + first_store = WorkflowStore(db_path) + definition = _definition() + first_store.save_definition(definition) + run = first_store.create_run(definition.id, {"x": 1}) + first_store.append_step_result(run.id, WorkflowStepResult("first", "planned_goal", True)) + first_store.update_run(run.id, current_step_id="second", variables={"x": 2}) + + reopened = WorkflowStore(db_path) + loaded = reopened.get_run(run.id) + + assert loaded is not None + assert loaded.current_step_id == "second" + assert loaded.variables == {"x": 2} + assert [result.step_id for result in loaded.step_results] == ["first"] diff --git a/workflow/__init__.py b/workflow/__init__.py new file mode 100644 index 0000000..5f5be5a --- /dev/null +++ b/workflow/__init__.py @@ -0,0 +1,27 @@ +"""Persisted workflow orchestration over tasks, skills, and conditions.""" + +from workflow.models import ( + BranchStep, + ConditionSpec, + PlannedGoalStep, + SkillInvocationStep, + WaitForConditionStep, + WorkflowDefinition, + WorkflowRun, + WorkflowStepResult, +) +from workflow.runner import WorkflowRunner +from workflow.store import WorkflowStore + +__all__ = [ + "BranchStep", + "ConditionSpec", + "PlannedGoalStep", + "SkillInvocationStep", + "WaitForConditionStep", + "WorkflowDefinition", + "WorkflowRun", + "WorkflowRunner", + "WorkflowStepResult", + "WorkflowStore", +] diff --git a/workflow/conditions.py b/workflow/conditions.py new file mode 100644 index 0000000..8d15e2b --- /dev/null +++ b/workflow/conditions.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Protocol, Sequence + +from core.models import Scene +from workflow.models import ConditionSpec, WorkflowStepResult + + +class UnknownConditionKindError(Exception): + pass + + +class ConditionEvaluator(Protocol): + def evaluate( + self, + spec: ConditionSpec, + *, + scene: Scene | None, + world_state: object | None, + started_at: datetime | None = None, + step_results: Sequence[WorkflowStepResult] = (), + ) -> bool: + ... + + +class SceneContainsTextEvaluator: + def evaluate( + self, + spec: ConditionSpec, + *, + scene: Scene | None, + world_state: object | None, + started_at: datetime | None = None, + step_results: Sequence[WorkflowStepResult] = (), + ) -> bool: + if scene is None: + return False + target = str(spec.params.get("text") or spec.params.get("target") or "") + if not target: + return False + target_lower = target.lower() + return any( + target_lower in str(element.text or "").lower() + for element in scene.elements + ) + + +class WorldVariableEqualsEvaluator: + def evaluate( + self, + spec: ConditionSpec, + *, + scene: Scene | None, + world_state: object | None, + started_at: datetime | None = None, + step_results: Sequence[WorkflowStepResult] = (), + ) -> bool: + variables = getattr(world_state, "variables", None) + if not isinstance(variables, dict): + return False + name = spec.params.get("name") + if not isinstance(name, str): + return False + return variables.get(name) == spec.params.get("value") + + +class ElapsedSecondsEvaluator: + def evaluate( + self, + spec: ConditionSpec, + *, + scene: Scene | None, + world_state: object | None, + started_at: datetime | None = None, + step_results: Sequence[WorkflowStepResult] = (), + ) -> bool: + if started_at is None: + return False + seconds = float(spec.params.get("seconds") or 0) + return (datetime.now(started_at.tzinfo) - started_at).total_seconds() >= seconds + + +class StepResultSuccessEvaluator: + def evaluate( + self, + spec: ConditionSpec, + *, + scene: Scene | None, + world_state: object | None, + started_at: datetime | None = None, + step_results: Sequence[WorkflowStepResult] = (), + ) -> bool: + target_step_id = spec.params.get("step_id") + return any( + result.step_id == target_step_id and result.success + for result in step_results + ) + + +DEFAULT_CONDITION_REGISTRY: dict[str, ConditionEvaluator] = { + "scene_contains_text": SceneContainsTextEvaluator(), + "world_variable_equals": WorldVariableEqualsEvaluator(), + "elapsed_seconds": ElapsedSecondsEvaluator(), + "step_result_success": StepResultSuccessEvaluator(), +} + + +def evaluate_condition( + spec: ConditionSpec, + *, + scene: Scene | None, + world_state: object | None, + started_at: datetime | None = None, + step_results: Sequence[WorkflowStepResult] = (), + registry: dict[str, ConditionEvaluator] | None = None, +) -> bool: + evaluators = registry or DEFAULT_CONDITION_REGISTRY + evaluator = evaluators.get(spec.kind) + if evaluator is None: + raise UnknownConditionKindError(spec.kind) + return evaluator.evaluate( + spec, + scene=scene, + world_state=world_state, + started_at=started_at, + step_results=step_results, + ) diff --git a/workflow/config.py b/workflow/config.py new file mode 100644 index 0000000..65f1d20 --- /dev/null +++ b/workflow/config.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +from collections.abc import Mapping +from dataclasses import dataclass + +DEFAULT_POLL_INTERVAL_SECONDS = 0.25 +DEFAULT_WAIT_TIMEOUT_SECONDS = 10.0 +DEFAULT_WORKFLOW_DB_PATH = "workflows/workflows.sqlite3" + +POLL_INTERVAL_ENV = "WORKFLOW_POLL_INTERVAL_SECONDS" +WAIT_TIMEOUT_ENV = "WORKFLOW_WAIT_TIMEOUT_SECONDS" +DB_PATH_ENV = "WORKFLOW_DB_PATH" + + +@dataclass(frozen=True) +class WorkflowConfig: + poll_interval_seconds: float = DEFAULT_POLL_INTERVAL_SECONDS + wait_timeout_seconds: float = DEFAULT_WAIT_TIMEOUT_SECONDS + db_path: str = DEFAULT_WORKFLOW_DB_PATH + + +def load_config(env: Mapping[str, str] | None = None) -> WorkflowConfig: + values = env or os.environ + return WorkflowConfig( + poll_interval_seconds=_parse_float( + values.get(POLL_INTERVAL_ENV), + default=DEFAULT_POLL_INTERVAL_SECONDS, + ), + wait_timeout_seconds=_parse_float( + values.get(WAIT_TIMEOUT_ENV), + default=DEFAULT_WAIT_TIMEOUT_SECONDS, + ), + db_path=values.get(DB_PATH_ENV) or DEFAULT_WORKFLOW_DB_PATH, + ) + + +def _parse_float(value: str | None, *, default: float) -> float: + if value is None: + return default + try: + parsed = float(value) + except ValueError: + return default + return parsed if parsed >= 0 else default diff --git a/workflow/models.py b/workflow/models.py new file mode 100644 index 0000000..37e53f6 --- /dev/null +++ b/workflow/models.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, ClassVar, Literal +from uuid import uuid4 + +from core.models import utc_now + +WorkflowRunStatus = Literal["pending", "running", "waiting", "completed", "failed", "cancelled"] + + +@dataclass(frozen=True) +class ConditionSpec: + kind: str + params: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return {"kind": self.kind, "params": dict(self.params)} + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "ConditionSpec": + return cls(kind=str(data["kind"]), params=dict(data.get("params") or {})) + + +@dataclass(frozen=True) +class PlannedGoalStep: + step_id: str + goal: str + next_step_id: str | None = None + kind: ClassVar[str] = "planned_goal" + + def to_dict(self) -> dict[str, Any]: + return { + "kind": self.kind, + "step_id": self.step_id, + "goal": self.goal, + "next_step_id": self.next_step_id, + } + + +@dataclass(frozen=True) +class SkillInvocationStep: + step_id: str + skill_id: str + args: dict[str, Any] = field(default_factory=dict) + next_step_id: str | None = None + kind: ClassVar[str] = "skill_invocation" + + def to_dict(self) -> dict[str, Any]: + return { + "kind": self.kind, + "step_id": self.step_id, + "skill_id": self.skill_id, + "args": dict(self.args), + "next_step_id": self.next_step_id, + } + + +@dataclass(frozen=True) +class WaitForConditionStep: + step_id: str + condition: ConditionSpec + timeout_seconds: float + poll_interval_seconds: float + next_step_id: str | None = None + kind: ClassVar[str] = "wait_for_condition" + + def to_dict(self) -> dict[str, Any]: + return { + "kind": self.kind, + "step_id": self.step_id, + "condition": self.condition.to_dict(), + "timeout_seconds": self.timeout_seconds, + "poll_interval_seconds": self.poll_interval_seconds, + "next_step_id": self.next_step_id, + } + + +@dataclass(frozen=True) +class BranchStep: + step_id: str + condition: ConditionSpec + on_true: str + on_false: str + kind: ClassVar[str] = "branch" + + def to_dict(self) -> dict[str, Any]: + return { + "kind": self.kind, + "step_id": self.step_id, + "condition": self.condition.to_dict(), + "on_true": self.on_true, + "on_false": self.on_false, + } + + +WorkflowStep = PlannedGoalStep | SkillInvocationStep | WaitForConditionStep | BranchStep + + +@dataclass(frozen=True) +class WorkflowDefinition: + name: str + steps: list[WorkflowStep] + entry_step_id: str + id: str = field(default_factory=lambda: uuid4().hex) + + def __post_init__(self) -> None: + if not self.steps: + raise ValueError("workflow definition requires at least one step") + step_ids = [step.step_id for step in self.steps] + if len(step_ids) != len(set(step_ids)): + raise ValueError("workflow definition contains duplicate step_id values") + known = set(step_ids) + if self.entry_step_id not in known: + raise ValueError(f"entry_step_id {self.entry_step_id} is not a workflow step") + for step in self.steps: + for target in _step_targets(step): + if target is not None and target not in known: + raise ValueError( + f"step {step.step_id} references unknown step {target}" + ) + + def step_by_id(self, step_id: str) -> WorkflowStep: + for step in self.steps: + if step.step_id == step_id: + return step + raise KeyError(step_id) + + def next_step_id_after(self, step_id: str) -> str | None: + for index, step in enumerate(self.steps): + if step.step_id == step_id: + if index + 1 >= len(self.steps): + return None + return self.steps[index + 1].step_id + raise KeyError(step_id) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "name": self.name, + "entry_step_id": self.entry_step_id, + "steps": [step.to_dict() for step in self.steps], + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "WorkflowDefinition": + return cls( + id=str(data["id"]), + name=str(data["name"]), + entry_step_id=str(data["entry_step_id"]), + steps=[workflow_step_from_dict(step) for step in data.get("steps", [])], + ) + + +@dataclass(frozen=True) +class WorkflowStepResult: + step_id: str + kind: str + success: bool + detail: dict[str, Any] = field(default_factory=dict) + task_id: str | None = None + timestamp: datetime = field(default_factory=utc_now) + + def to_dict(self) -> dict[str, Any]: + return { + "step_id": self.step_id, + "kind": self.kind, + "success": self.success, + "detail": dict(self.detail), + "task_id": self.task_id, + "timestamp": self.timestamp.isoformat(), + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "WorkflowStepResult": + return cls( + step_id=str(data["step_id"]), + kind=str(data["kind"]), + success=bool(data["success"]), + detail=dict(data.get("detail") or {}), + task_id=data.get("task_id"), + timestamp=_parse_datetime(data.get("timestamp")), + ) + + +@dataclass(frozen=True) +class WorkflowRun: + definition_id: str + status: WorkflowRunStatus + current_step_id: str | None + variables: dict[str, Any] = field(default_factory=dict) + step_results: list[WorkflowStepResult] = field(default_factory=list) + id: str = field(default_factory=lambda: uuid4().hex) + device_id: str | None = None + created_at: datetime = field(default_factory=utc_now) + updated_at: datetime = field(default_factory=utc_now) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.id, + "definition_id": self.definition_id, + "status": self.status, + "current_step_id": self.current_step_id, + "device_id": self.device_id, + "variables": dict(self.variables), + "step_results": [result.to_dict() for result in self.step_results], + "created_at": self.created_at.isoformat(), + "updated_at": self.updated_at.isoformat(), + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "WorkflowRun": + return cls( + id=str(data["id"]), + definition_id=str(data["definition_id"]), + status=data["status"], + current_step_id=data.get("current_step_id"), + device_id=data.get("device_id"), + variables=dict(data.get("variables") or {}), + step_results=[ + WorkflowStepResult.from_dict(result) + for result in data.get("step_results", []) + ], + created_at=_parse_datetime(data.get("created_at")), + updated_at=_parse_datetime(data.get("updated_at")), + ) + + +def workflow_step_from_dict(data: dict[str, Any]) -> WorkflowStep: + kind = data.get("kind") + if kind == PlannedGoalStep.kind: + return PlannedGoalStep( + step_id=str(data["step_id"]), + goal=str(data["goal"]), + next_step_id=data.get("next_step_id"), + ) + if kind == SkillInvocationStep.kind: + return SkillInvocationStep( + step_id=str(data["step_id"]), + skill_id=str(data["skill_id"]), + args=dict(data.get("args") or {}), + next_step_id=data.get("next_step_id"), + ) + if kind == WaitForConditionStep.kind: + return WaitForConditionStep( + step_id=str(data["step_id"]), + condition=ConditionSpec.from_dict(data["condition"]), + timeout_seconds=float(data["timeout_seconds"]), + poll_interval_seconds=float(data["poll_interval_seconds"]), + next_step_id=data.get("next_step_id"), + ) + if kind == BranchStep.kind: + return BranchStep( + step_id=str(data["step_id"]), + condition=ConditionSpec.from_dict(data["condition"]), + on_true=str(data["on_true"]), + on_false=str(data["on_false"]), + ) + raise ValueError(f"unknown workflow step kind {kind}") + + +def _step_targets(step: WorkflowStep) -> list[str | None]: + if isinstance(step, BranchStep): + return [step.on_true, step.on_false] + return [step.next_step_id] + + +def _parse_datetime(value: Any) -> datetime: + if isinstance(value, datetime): + return value + if isinstance(value, str): + try: + return datetime.fromisoformat(value) + except ValueError: + pass + return utc_now() diff --git a/workflow/runner.py b/workflow/runner.py new file mode 100644 index 0000000..4eaf163 --- /dev/null +++ b/workflow/runner.py @@ -0,0 +1,321 @@ +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import replace +from datetime import datetime +from types import SimpleNamespace +from typing import Any +from time import sleep + +from core.models import Scene, Task +from runtime.task import TaskRunner +from skills_learning.store import SkillStore, get_default_store +from workflow.conditions import ( + ConditionEvaluator, + UnknownConditionKindError, + evaluate_condition, +) +from workflow.config import WorkflowConfig, load_config +from workflow.models import ( + BranchStep, + PlannedGoalStep, + SkillInvocationStep, + WaitForConditionStep, + WorkflowDefinition, + WorkflowRun, + WorkflowStep, + WorkflowStepResult, +) +from workflow.skill_exec import SkillExecutionError, run_flow_template_skill +from workflow.store import WorkflowStore + +TaskRunnerFactory = Callable[[], TaskRunner] +SceneProvider = Callable[[], Scene | None] +WorldStateProvider = Callable[[], object | None] +SleepFunc = Callable[[float], None] + +TERMINAL_STATUSES = {"completed", "failed", "cancelled"} + + +class WorkflowRunner: + def __init__( + self, + store: WorkflowStore | None = None, + *, + task_runner_factory: TaskRunnerFactory | None = None, + skill_store: SkillStore | None = None, + tools: dict[str, Callable[..., Any]] | None = None, + condition_registry: dict[str, ConditionEvaluator] | None = None, + scene_provider: SceneProvider | None = None, + world_state_provider: WorldStateProvider | None = None, + sleep_func: SleepFunc = sleep, + config: WorkflowConfig | None = None, + step_limit: int | None = None, + ) -> None: + self.config = config or load_config() + self.store = store or WorkflowStore(self.config.db_path) + self.task_runner_factory = task_runner_factory or (lambda: TaskRunner()) + self.skill_store = skill_store or get_default_store() + self.tools = tools + self.condition_registry = condition_registry + self.scene_provider = scene_provider or (lambda: None) + self.world_state_provider = world_state_provider + self.sleep_func = sleep_func + self.step_limit = step_limit + + def run( + self, + definition: WorkflowDefinition, + device_id: str, + initial_variables: dict[str, Any] | None = None, + ) -> WorkflowRun: + self.store.save_definition(definition) + run = self.store.create_run( + definition.id, + initial_variables or {}, + device_id=device_id, + ) + return self._drive(definition, run) + + def resume(self, run_id: str) -> WorkflowRun: + run = self.store.get_run(run_id) + if run is None: + raise KeyError(f"unknown workflow run {run_id}") + if run.status in TERMINAL_STATUSES: + return run + definition = self.store.get_definition(run.definition_id) + if definition is None: + raise KeyError(f"unknown workflow definition {run.definition_id}") + return self._drive(definition, run) + + def _drive( + self, + definition: WorkflowDefinition, + run: WorkflowRun, + ) -> WorkflowRun: + executed = 0 + while run.status not in TERMINAL_STATUSES and run.current_step_id: + if self.step_limit is not None and executed >= self.step_limit: + return run + step = definition.step_by_id(run.current_step_id) + if _already_recorded(run, step.step_id): + next_step_id = self._next_step_id(definition, step, None) + run = self._checkpoint(run, next_step_id, "running") + continue + + result, branch_next_step_id = self._execute_step(definition, run, step) + next_status = "running" + next_step_id = self._next_step_id(definition, step, branch_next_step_id) + if not result.success: + next_status = "failed" + next_step_id = None + elif next_step_id is None: + next_status = "completed" + + self.store.append_step_result(run.id, result) + run = self._checkpoint( + run, + next_step_id, + next_status, + ) + executed += 1 + return run + + def _checkpoint( + self, + run: WorkflowRun, + current_step_id: str | None, + status: str, + ) -> WorkflowRun: + self.store.update_run( + run.id, + status=status, # type: ignore[arg-type] + current_step_id=current_step_id, + variables=run.variables, + ) + updated = self.store.get_run(run.id) + if updated is None: + raise KeyError(f"unknown workflow run {run.id}") + return updated + + def _execute_step( + self, + definition: WorkflowDefinition, + run: WorkflowRun, + step: WorkflowStep, + ) -> tuple[WorkflowStepResult, str | None]: + if isinstance(step, PlannedGoalStep): + return self._execute_planned_goal_step(run, step), None + if isinstance(step, SkillInvocationStep): + return self._execute_skill_invocation_step(step), None + if isinstance(step, WaitForConditionStep): + return self._execute_wait_step(run, step), None + if isinstance(step, BranchStep): + return self._execute_branch_step(run, step) + return ( + WorkflowStepResult( + step_id=getattr(step, "step_id", "unknown"), + kind="unknown", + success=False, + detail={"reason": "unknown workflow step type"}, + ), + None, + ) + + def _execute_planned_goal_step( + self, + run: WorkflowRun, + step: PlannedGoalStep, + ) -> WorkflowStepResult: + task = Task(goal=step.goal, device_id=run.device_id or "") + task_runner = self.task_runner_factory() + result_task = task_runner.run(task) + success = result_task.status == "completed" + return WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=success, + detail={ + "task_status": result_task.status, + "failure_reason": result_task.failure_reason, + }, + task_id=result_task.id, + ) + + def _execute_skill_invocation_step( + self, + step: SkillInvocationStep, + ) -> WorkflowStepResult: + skill = self.skill_store.get_by_id(step.skill_id) + if skill is None: + return WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=False, + detail={"reason": f"unknown skill {step.skill_id}"}, + ) + try: + results = run_flow_template_skill(skill, step.args, tools=self.tools) + except SkillExecutionError as exc: + return WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=False, + detail={"reason": str(exc)}, + ) + success = all(result.success for result in results) + return WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=success, + detail={ + "step_results": [result.to_dict() for result in results], + }, + ) + + def _execute_wait_step( + self, + run: WorkflowRun, + step: WaitForConditionStep, + ) -> WorkflowStepResult: + started_at = datetime.now().astimezone() + timeout_seconds = step.timeout_seconds + poll_interval_seconds = step.poll_interval_seconds + while True: + try: + if self._condition_is_true(run, step.condition, started_at=started_at): + return WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=True, + detail={"condition": step.condition.to_dict()}, + ) + except UnknownConditionKindError as exc: + return WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=False, + detail={"reason": f"unknown condition kind: {exc}"}, + ) + + elapsed = ( + datetime.now(started_at.tzinfo) - started_at + ).total_seconds() + if elapsed >= timeout_seconds: + return WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=False, + detail={"reason": "condition timed out"}, + ) + self.sleep_func(poll_interval_seconds) + + def _execute_branch_step( + self, + run: WorkflowRun, + step: BranchStep, + ) -> tuple[WorkflowStepResult, str | None]: + try: + matched = self._condition_is_true(run, step.condition) + except UnknownConditionKindError as exc: + return ( + WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=False, + detail={"reason": f"unknown condition kind: {exc}"}, + ), + None, + ) + target = step.on_true if matched else step.on_false + return ( + WorkflowStepResult( + step_id=step.step_id, + kind=step.kind, + success=True, + detail={"condition_result": matched, "next_step_id": target}, + ), + target, + ) + + def _condition_is_true( + self, + run: WorkflowRun, + condition, + *, + started_at: datetime | None = None, + ) -> bool: + return evaluate_condition( + condition, + scene=self.scene_provider(), + world_state=self._world_state(run), + started_at=started_at, + step_results=run.step_results, + registry=self.condition_registry, + ) + + def _world_state(self, run: WorkflowRun) -> object: + if self.world_state_provider is not None: + world_state = self.world_state_provider() + if world_state is not None: + return world_state + return SimpleNamespace(variables=dict(run.variables)) + + def _next_step_id( + self, + definition: WorkflowDefinition, + step: WorkflowStep, + branch_next_step_id: str | None, + ) -> str | None: + if branch_next_step_id is not None: + return branch_next_step_id + explicit = getattr(step, "next_step_id", None) + if explicit is not None: + return explicit + if isinstance(step, BranchStep): + return None + return definition.next_step_id_after(step.step_id) + + +def _already_recorded(run: WorkflowRun, step_id: str) -> bool: + return any(result.step_id == step_id for result in run.step_results) diff --git a/workflow/skill_exec.py b/workflow/skill_exec.py new file mode 100644 index 0000000..15f0fe7 --- /dev/null +++ b/workflow/skill_exec.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from typing import Any + +from runtime.executor import Executor, ExecutorConfig, StepResult, ToolCallable +from runtime.planner import PlannedStep +from skills_learning.models import FlowTemplateSkill + + +class SkillExecutionError(Exception): + pass + + +def validate_skill_args(skill: FlowTemplateSkill, args: dict[str, Any]) -> None: + if skill.metadata.kind != "flow_template": + raise SkillExecutionError(f"skill {skill.id} is not a flow_template skill") + required = set(skill.parameters) + provided = set(args) + missing = sorted(required - provided) + extra = sorted(provided - required) + errors: list[str] = [] + if missing: + errors.append(f"missing required parameters: {', '.join(missing)}") + if extra: + errors.append(f"unrecognized parameters: {', '.join(extra)}") + if errors: + raise SkillExecutionError("; ".join(errors)) + + +def resolve_skill_steps( + skill: FlowTemplateSkill, + args: dict[str, Any], +) -> list[dict[str, Any]]: + validate_skill_args(skill, args) + return [ + { + "tool_name": step.tool_name, + "args": _resolve_value(step.args, args), + } + for step in skill.steps + ] + + +def run_flow_template_skill( + skill: FlowTemplateSkill, + args: dict[str, Any], + *, + tools: dict[str, ToolCallable] | None = None, +) -> list[StepResult]: + resolved_steps = resolve_skill_steps(skill, args) + executor = Executor( + tools=tools, + config=ExecutorConfig(max_retries=1, backoff_seconds=0), + ) + results: list[StepResult] = [] + for index, resolved in enumerate(resolved_steps, start=1): + step = PlannedStep( + action=resolved["tool_name"], + description=f"Run skill {skill.name} step {index}", + args=resolved["args"], + ) + result = executor.execute(step) + results.append(result) + if not result.success: + break + return results + + +def _resolve_value(value: Any, args: dict[str, Any]) -> Any: + if isinstance(value, str) and value.startswith("{") and value.endswith("}"): + name = value[1:-1] + return args[name] + if isinstance(value, dict): + return {key: _resolve_value(nested, args) for key, nested in value.items()} + if isinstance(value, list): + return [_resolve_value(item, args) for item in value] + return value diff --git a/workflow/store.py b/workflow/store.py new file mode 100644 index 0000000..68efa97 --- /dev/null +++ b/workflow/store.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Any + +from core.models import utc_now +from workflow.config import DEFAULT_WORKFLOW_DB_PATH +from workflow.models import ( + WorkflowDefinition, + WorkflowRun, + WorkflowRunStatus, + WorkflowStepResult, +) + + +class WorkflowStore: + def __init__(self, db_path: str | Path = DEFAULT_WORKFLOW_DB_PATH) -> None: + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._ensure_schema() + + def save_definition(self, definition: WorkflowDefinition) -> None: + with self._connect() as connection: + connection.execute( + """ + insert into workflow_definitions (id, name, definition_json) + values (?, ?, ?) + on conflict(id) do update set + name = excluded.name, + definition_json = excluded.definition_json + """, + ( + definition.id, + definition.name, + json.dumps(definition.to_dict(), ensure_ascii=False), + ), + ) + + def get_definition(self, definition_id: str) -> WorkflowDefinition | None: + with self._connect() as connection: + row = connection.execute( + "select definition_json from workflow_definitions where id = ?", + (definition_id,), + ).fetchone() + if row is None: + return None + return WorkflowDefinition.from_dict(json.loads(row["definition_json"])) + + def create_run( + self, + definition_id: str, + initial_variables: dict[str, Any] | None = None, + *, + device_id: str | None = None, + ) -> WorkflowRun: + definition = self.get_definition(definition_id) + if definition is None: + raise KeyError(f"unknown workflow definition {definition_id}") + run = WorkflowRun( + definition_id=definition_id, + status="running", + current_step_id=definition.entry_step_id, + variables=dict(initial_variables or {}), + device_id=device_id, + ) + with self._connect() as connection: + connection.execute( + """ + insert into workflow_runs ( + id, definition_id, status, current_step_id, device_id, + variables_json, created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + run.id, + run.definition_id, + run.status, + run.current_step_id, + run.device_id, + json.dumps(run.variables, ensure_ascii=False), + run.created_at.isoformat(), + run.updated_at.isoformat(), + ), + ) + return run + + def append_step_result( + self, + run_id: str, + step_result: WorkflowStepResult, + ) -> None: + with self._connect() as connection: + index = ( + connection.execute( + "select count(*) as count from workflow_step_results where run_id = ?", + (run_id,), + ).fetchone()["count"] + + 1 + ) + connection.execute( + """ + insert into workflow_step_results ( + run_id, step_index, step_id, kind, success, detail_json, + task_id, timestamp + ) values (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + run_id, + index, + step_result.step_id, + step_result.kind, + 1 if step_result.success else 0, + json.dumps(step_result.detail, ensure_ascii=False), + step_result.task_id, + step_result.timestamp.isoformat(), + ), + ) + + def update_run( + self, + run_id: str, + *, + status: WorkflowRunStatus | None = None, + current_step_id: str | None = None, + variables: dict[str, Any] | None = None, + ) -> None: + run = self.get_run(run_id) + if run is None: + raise KeyError(f"unknown workflow run {run_id}") + next_status = status or run.status + next_step_id = current_step_id + if current_step_id is None and next_status not in { + "completed", + "failed", + "cancelled", + }: + next_step_id = run.current_step_id + next_variables = dict(run.variables if variables is None else variables) + with self._connect() as connection: + connection.execute( + """ + update workflow_runs + set status = ?, + current_step_id = ?, + variables_json = ?, + updated_at = ? + where id = ? + """, + ( + next_status, + next_step_id, + json.dumps(next_variables, ensure_ascii=False), + utc_now().isoformat(), + run_id, + ), + ) + + def get_run(self, run_id: str) -> WorkflowRun | None: + with self._connect() as connection: + run_row = connection.execute( + "select * from workflow_runs where id = ?", + (run_id,), + ).fetchone() + if run_row is None: + return None + result_rows = connection.execute( + """ + select * from workflow_step_results + where run_id = ? + order by step_index + """, + (run_id,), + ).fetchall() + return WorkflowRun.from_dict( + { + "id": run_row["id"], + "definition_id": run_row["definition_id"], + "status": run_row["status"], + "current_step_id": run_row["current_step_id"], + "device_id": run_row["device_id"], + "variables": json.loads(run_row["variables_json"]), + "created_at": run_row["created_at"], + "updated_at": run_row["updated_at"], + "step_results": [ + { + "step_id": row["step_id"], + "kind": row["kind"], + "success": bool(row["success"]), + "detail": json.loads(row["detail_json"]), + "task_id": row["task_id"], + "timestamp": row["timestamp"], + } + for row in result_rows + ], + } + ) + + def _ensure_schema(self) -> None: + with self._connect() as connection: + connection.execute( + """ + create table if not exists workflow_definitions ( + id text primary key, + name text not null, + definition_json text not null + ) + """ + ) + connection.execute( + """ + create table if not exists workflow_runs ( + id text primary key, + definition_id text not null, + status text not null, + current_step_id text, + device_id text, + variables_json text not null, + created_at text not null, + updated_at text not null + ) + """ + ) + connection.execute( + """ + create table if not exists workflow_step_results ( + id integer primary key autoincrement, + run_id text not null, + step_index integer not null, + step_id text not null, + kind text not null, + success integer not null, + detail_json text not null, + task_id text, + timestamp text not null + ) + """ + ) + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self.db_path) + connection.row_factory = sqlite3.Row + return connection