feat(host-agent): make execution history authoritative

tests/test_workflow_runner.py

@@ -6,6 +6,7 @@ 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 storage.task_metadata import TaskMetadataStore
 from tests.fakes import PNG_10X20
 from workflow.models import (
     BranchStep,
@@ -39,6 +40,14 @@ class FakeTaskRunner:
         return task
 
 
+class ImmediateCompletionPlanner(Planner):
+    def plan(self, *, goal, scene, context):
+        return []
+
+    def goal_reached(self, *, goal, scene, context):
+        return True
+
+
 def _scene(text: str = "Ready") -> Scene:
     return Scene(
         width=10,
@@ -96,6 +105,32 @@ def test_workflow_runner_linear_planned_goal_completes(tmp_path) -> None:
     assert [result.step_id for result in run.step_results] == ["first", "second"]
 
 
+def test_workflow_planned_goal_creates_task_metadata_without_precreation(
+    tmp_path,
+) -> None:
+    metadata_store = TaskMetadataStore(tmp_path / "tasks.sqlite3")
+    definition = WorkflowDefinition(
+        name="persisted goal",
+        entry_step_id="goal",
+        steps=[PlannedGoalStep("goal", "inspect screen")],
+    )
+    runner = WorkflowRunner(
+        _store(tmp_path),
+        task_runner_factory=lambda: TaskRunner(
+            planner=ImmediateCompletionPlanner(),
+            metadata_store=metadata_store,
+            observer=lambda device_id: _scene(),
+            config=TaskRunnerConfig(max_steps=1),
+        ),
+    )
+
+    run = runner.run(definition, "phone")
+
+    task_id = run.step_results[0].task_id
+    assert task_id is not None
+    assert metadata_store.get_task(task_id)["status"] == "completed"
+
+
 def test_workflow_runner_stops_before_the_next_step(tmp_path) -> None:
     stop_requested = False
     calls: list[str] = []