worker_plan: represent plan_raw.json as InitialPlanRawTask

worker_plan/worker_plan_internal/plan/nodes/initial_plan_raw.py

@@ -0,0 +1,16 @@
+"""InitialPlanRawTask - ExternalTask exposing the user-supplied plan_raw.json."""
+from pathlib import Path
+import luigi
+from worker_plan_api.filenames import FilenameEnum
+
+
+class InitialPlanRawTask(luigi.ExternalTask):
+    """The user-supplied plan_raw.json that must exist before the pipeline starts.
+
+    Tasks that want the bare prompt (rather than SetupTask's
+    formatted plan.txt) require this and read it via PlanFile.
+    """
+    run_id_dir = luigi.Parameter()
+
+    def output(self):
+        return luigi.LocalTarget(str(Path(self.run_id_dir) / FilenameEnum.INITIAL_PLAN_RAW.value))

worker_plan/worker_plan_internal/plan/nodes/prompt_adherence.py

@@ -4,7 +4,7 @@
 from worker_plan_internal.llm_util.llm_executor import LLMExecutor
 from worker_plan_api.filenames import FilenameEnum
 from worker_plan_api.plan_file import PlanFile
-from worker_plan_internal.plan.nodes.setup import SetupTask
+from worker_plan_internal.plan.nodes.initial_plan_raw import InitialPlanRawTask
 from worker_plan_internal.plan.nodes.project_plan import ProjectPlanTask
 from worker_plan_internal.plan.nodes.executive_summary import ExecutiveSummaryTask
 from worker_plan_internal.plan.nodes.consolidate_assumptions_markdown import ConsolidateAssumptionsMarkdownTask
@@ -21,7 +21,7 @@ def output(self):
 
     def requires(self):
         return {
-            'setup': self.clone(SetupTask),
+            'plan_raw': self.clone(InitialPlanRawTask),
             'project_plan': self.clone(ProjectPlanTask),
             'executive_summary': self.clone(ExecutiveSummaryTask),
             'consolidate_assumptions_markdown': self.clone(ConsolidateAssumptionsMarkdownTask),
@@ -30,8 +30,7 @@ def requires(self):
     def run_inner(self):
         llm_executor: LLMExecutor = self.create_llm_executor()
 
-        plan_raw_path = self.run_id_dir / FilenameEnum.INITIAL_PLAN_RAW.value
-        plan_file = PlanFile.load(str(plan_raw_path))
+        plan_file = PlanFile.load(self.input()['plan_raw'].path)
         plan_prompt = plan_file.plan_prompt
         with self.input()['project_plan']['markdown'].open("r") as f:
             project_plan_markdown = f.read()

worker_plan/worker_plan_internal/plan/nodes/screen_planning_prompt.py

@@ -3,13 +3,14 @@
 from worker_plan_internal.plan.run_plan_pipeline import PlanTask
 from worker_plan_internal.diagnostics.screen_planning_prompt import ScreenPlanningPrompt
 from worker_plan_api.filenames import FilenameEnum
-from worker_plan_internal.plan.nodes.setup import SetupTask
+from worker_plan_api.plan_file import PlanFile
+from worker_plan_internal.plan.nodes.initial_plan_raw import InitialPlanRawTask
 
 
 class ScreenPlanningPromptTask(PlanTask):
     """Flag prompts as UNUSABLE when there is high confidence the prompt is garbage."""
     def requires(self):
-        return self.clone(SetupTask)
+        return self.clone(InitialPlanRawTask)
 
     def output(self):
         return {
@@ -18,8 +19,8 @@ def output(self):
         }
 
     def run_with_llm(self, llm: LLM) -> None:
-        with self.input().open("r") as f:
-            plan_prompt = f.read()
+        plan_file = PlanFile.load(self.input().path)
+        plan_prompt = plan_file.plan_prompt
 
         result = ScreenPlanningPrompt.execute(llm, plan_prompt)
 

worker_plan/worker_plan_internal/plan/nodes/setup.py

@@ -2,21 +2,19 @@
 from worker_plan_internal.plan.run_plan_pipeline import PlanTask
 from worker_plan_api.filenames import FilenameEnum
 from worker_plan_api.plan_file import PlanFile
+from worker_plan_internal.plan.nodes.initial_plan_raw import InitialPlanRawTask
 
 
 class SetupTask(PlanTask):
     """Read plan_raw.json and produce plan.txt from the template."""
+    def requires(self):
+        return self.clone(InitialPlanRawTask)
+
     def output(self):
         return self.local_target(FilenameEnum.INITIAL_PLAN)
 
     def run(self):
-        raw_path = self.run_id_dir / FilenameEnum.INITIAL_PLAN_RAW.value
-        if not raw_path.exists():
-            raise FileNotFoundError(
-                f"Before starting the pipeline the '{FilenameEnum.INITIAL_PLAN_RAW.value}' file "
-                f"must be present in the run_id_dir: {self.run_id_dir!r}"
-            )
-        plan_file = PlanFile.load(str(raw_path))
+        plan_file = PlanFile.load(self.input().path)
         plan_text = plan_file.to_plan_text()
         with open(self.output().path, "w", encoding="utf-8") as f:
             f.write(plan_text)