Add input sources mechanism for pipeline artifact seeding

.gremlins/gh-terse.yaml

@@ -7,7 +7,13 @@ inputs:
   in:
     INSTRUCTIONS: instructions?
     PLAN: plan?
-    PR: pr?
+  sources:
+    plan:
+      type: [filepath, string]
+      optional: true
+    instructions:
+      type: string
+      optional: true
 
 land:
   in:

gremlins/executor/gremlin.py

@@ -2,14 +2,12 @@
 
 from __future__ import annotations
 
-import argparse
 import asyncio
 import dataclasses
 import json
 import logging
 import os
 import pathlib
-import re
 import shutil
 from collections.abc import Awaitable, Callable, Sequence
 from typing import TYPE_CHECKING, Any, cast
@@ -114,9 +112,6 @@ def __init__(
         worktree_dir: pathlib.Path | None = None,
         worktree_parent: pathlib.Path | None = None,
         resume_from: str | None = None,
-        instructions: str = "",
-        spec: str | None = None,
-        plan: str | None = None,
         repo: str = "",
         state_file: pathlib.Path | None = None,
         project_root: str = "",
@@ -147,9 +142,6 @@ def __init__(
         self.worktree_dir = worktree_dir
         self.worktree_parent = worktree_parent
         self.resume_from = resume_from
-        self.instructions = instructions
-        self.spec = spec
-        self.plan = plan
         self.repo = repo
         self.state_file = state_file
         self.project_root = project_root
@@ -254,7 +246,6 @@ async def fork(
             pipeline_data=effective_pipeline,
             repo=state.repo,
             cwd=child_cwd,
-            instructions=state.instructions,
             worktree=child_worktree,
             worktree_parent=state.worktree_parent,
             artifacts=child_registry,
@@ -288,12 +279,6 @@ def _set_gremlin_recursive(self, stage: StageProtocol) -> None:
     def _collect_stages(
         self, stages: Sequence[StageProtocol]
     ) -> list[tuple[str, Callable[[], Awaitable[Any]]]]:
-        args = argparse.Namespace(
-            plan=self.plan,
-            resume_from=self.resume_from,
-            spec=self.spec,
-            instructions=[self.instructions] if self.instructions else [],
-        )
         cwd = (
             str(self.worktree_dir)
             if self.worktree_dir is not None
@@ -307,11 +292,9 @@ def _collect_stages(
                 data=StateData(gremlin_id=self.gremlin_id, state_file=self.state_file),
                 client=stage_client,
                 artifact_dir=self.artifact_dir,
-                args=args,
                 pipeline_data=self.pipeline_data,
                 repo=self.repo,
                 cwd=cwd,
-                instructions=self.instructions,
                 worktree=self.worktree_dir,
                 worktree_parent=self.worktree_parent,
                 artifacts=self.registry,
@@ -374,7 +357,6 @@ def open(cls, gremlin_id: str) -> Gremlin:
         pipeline_args = cast(list[str], state_raw.get("pipeline_args") or [])
         pipeline_path = cast(str, state_raw.get("pipeline_path") or "")
         worktree_dir_str = cast(str, state_raw.get("workdir") or "")
-        instructions = cast(str, state_raw.get("instructions") or "")
 
         # Resolve pipeline (hermetic check first, then fallback)
         hermetic = state_dir / "pipeline.yaml"
@@ -426,7 +408,6 @@ def open(cls, gremlin_id: str) -> Gremlin:
             gremlin_id=gremlin_id,
             pipeline_data=pipeline,
             worktree_dir=worktree_dir,
-            instructions=instructions,
             project_root=project_root,
             pipeline_path=pipeline_path,
             pipeline_args=pipeline_args,
@@ -441,10 +422,7 @@ def initialize_with_runtime(
         project_dir: pathlib.Path,
         pipeline_ref: str,
         worktree_parent: pathlib.Path | None = None,
-        instructions: str = "",
         resume_from: str | None = None,
-        plan: str | None = None,
-        spec: str | None = None,
         project_root: str = "",
         base_ref_sha: str = "",
         base_ref: str = "",
@@ -486,9 +464,6 @@ def initialize_with_runtime(
             worktree_dir=worktree_dir,
             worktree_parent=worktree_parent,
             resume_from=resume_from,
-            instructions=instructions,
-            spec=spec,
-            plan=plan,
             project_root=project_root,
             base_ref_sha=base_ref_sha,
             base_ref=base_ref,
@@ -500,7 +475,6 @@ def initialize_with_runtime(
             self.state_dir,
             self.artifact_dir,
             self.gremlin_id,
-            instructions=self.instructions or "",
         )
 
         worktree_created: str | None = None
@@ -522,23 +496,6 @@ def initialize_with_runtime(
                     setup_kind="worktree-detached",
                 )
 
-            if self.spec:
-                spec_file = self.artifact_dir / "spec.md"
-                if not spec_file.exists():
-                    spec_src = pathlib.Path(self.spec)
-                    if not spec_src.is_file():
-                        raise ValueError(f"--spec: file not found: {self.spec}")
-                    if spec_src.stat().st_size == 0:
-                        raise ValueError(f"--spec: file is empty: {self.spec}")
-                    shutil.copyfile(spec_src, spec_file)
-
-            if self.plan and not self.pipeline_data.github_integration:
-                plan_file = self.artifact_dir / "plan.md"
-                if not plan_file.exists():
-                    src = pathlib.Path(self.plan)
-                    if src.is_file():
-                        shutil.copyfile(src, plan_file)
-
             if self.worktree_dir is not None:
                 os.chdir(self.worktree_dir)
 
@@ -555,35 +512,6 @@ def initialize_with_runtime(
                 sha = _git_mod.head_sha(cwd=self.worktree_dir)
                 if sha:
                     self.registry.bind("base_sha", Uri.parse(f"git://commit/{sha}"))
-            # When --plan is a GH issue ref on a github_integration pipeline,
-            # the opaque issue URI is what compose-pr's plan.uri? needs.
-            plan_issue_uri: str | None = None
-            if self.pipeline_data.github_integration and self.plan:
-                m = re.match(
-                    r"^(?:[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)?#([0-9]+)$", self.plan
-                )
-                if m:
-                    plan_issue_uri = f"gh://issue/{m.group(1)}"
-
-            if not self.registry.produced("plan"):
-                if (self.artifact_dir / "plan.md").exists():
-                    if not self.pipeline_data.github_integration:
-                        self.registry.bind("plan", Uri.parse("file://session/plan.md"))
-                    elif plan_issue_uri is not None:
-                        self.registry.bind("plan", Uri.parse(plan_issue_uri))
-                    elif self.registry.produced("plan-issue-number"):
-                        n = str(self.registry.read("plan-issue-number")).strip()
-                        self.registry.bind("plan", Uri.parse(f"gh://issue/{n}"))
-                    # else: github_integration with no issue ref/number yet —
-                    # publish-as-issue will bind plan (avoid DuplicateArtifact).
-            elif (
-                plan_issue_uri is not None
-                and self.registry.resolve("plan").scheme == "file"
-            ):
-                # resume: upgrade an existing file:// plan to the issue URI so
-                # compose-pr resolves it even when the plan stage was skipped.
-                self.registry.unbind("plan")
-                self.registry.bind("plan", Uri.parse(plan_issue_uri))
         except Exception:
             if worktree_created:
                 _git_mod.remove_worktree(self.project_root, worktree_created)

gremlins/executor/run.py

@@ -135,22 +135,9 @@ def _atexit_log() -> None:
 
 def _parse_args(argv: list[str]) -> argparse.Namespace:
     parser = argparse.ArgumentParser(add_help=False)
-    parser.add_argument("--plan", dest="plan", default=None)
-    parser.add_argument("--spec", default=None)
     parser.add_argument("--client", dest="client", default=None)
     parser.add_argument("--resume-from", dest="resume_from", default=None)
-    parser.add_argument("instructions", nargs="*")
-    args = parser.parse_args(argv)
-    if args.plan and args.instructions:
-        die("--plan and positional instructions are mutually exclusive")
-    if (
-        not args.plan
-        and not args.instructions
-        and not args.resume_from
-        and not os.environ.get("GREMLINS_RESUME_FROM")
-    ):
-        die("one of --plan or positional instructions is required")
-    return args
+    return parser.parse_args(argv)
 
 
 def _unique_clients(stages: Sequence[StageProtocol]) -> list[Client]:
@@ -210,9 +197,6 @@ async def run_pipeline(
     worktree_dir = pathlib.Path(_workdir) if _workdir else None
     project_root = str(state_json.get("project_root") or "")
     stage_inputs: dict[str, Any] = dict(state_json.get("stage_inputs") or {})
-    instructions: str = str(
-        stage_inputs.get("instructions") or " ".join(args.instructions or [])
-    )
 
     # base_ref_sha and base_ref are bound in registry.json at launch time
     _registry_path = state_dir / "registry.json"
@@ -257,10 +241,7 @@ async def run_pipeline(
             if project_root
             else paths.project_root(),
             pipeline_ref=str(pipeline_path),
-            instructions=instructions,
             resume_from=resume_from,
-            spec=args.spec,
-            plan=args.plan,
             worktree_dir=worktree_dir,
             project_root=project_root,
             base_ref_sha=base_ref_sha,
@@ -307,8 +288,6 @@ async def run_pipeline(
         if shutil.which("claude") is None:
             die("claude not found on PATH")
 
-    plan_file = artifact_dir / "plan.md"
-
     if not gh and resume_from:
         _expanded_stage_names = [s.name for s in gremlin.stages]
 
@@ -323,9 +302,6 @@ def _name_idx(stage_name: str) -> int:
             if resume_from in _expanded_stage_names
             else 0
         )
-        if start_idx >= _name_idx("implement"):
-            if not plan_file.exists() or plan_file.stat().st_size == 0:
-                die(f"--resume-from {resume_from} requires existing {plan_file}")
         if start_idx >= _name_idx("review-code"):
             if not has_dirty_worktree() and not has_commits():
                 die(

gremlins/executor/state.py

@@ -115,7 +115,6 @@ class StateData:
     worktree_base: str = ""
     status: str = ""
     started_at: str = ""
-    instructions: str = ""
     description: str = ""
     description_explicit: bool = False
     parent_id: str = ""
@@ -147,7 +146,6 @@ def load(cls, gremlin_id: str | None) -> StateData:
             worktree_base=sd.get("worktree_base") or "",
             status=sd.get("status") or "",
             started_at=sd.get("started_at") or "",
-            instructions=sd.get("instructions") or "",
             description=sd.get("description") or "",
             description_explicit=bool(sd.get("description_explicit")),
             parent_id=sd.get("parent_id") or "",
@@ -178,7 +176,6 @@ def persist(self, state_dir: pathlib.Path) -> None:
             "worktree_base": self.worktree_base,
             "status": self.status,
             "started_at": self.started_at,
-            "instructions": self.instructions,
             "description": self.description,
             "description_explicit": self.description_explicit,
             "parent_id": self.parent_id,
@@ -494,7 +491,6 @@ class State:
     cwd: str = ""
     args: argparse.Namespace = dataclasses.field(default_factory=argparse.Namespace)
     pipeline_data: Pipeline | None = None
-    instructions: str = ""
     current_scope: list[StageProtocol] = dataclasses.field(default_factory=_stage_list)
     child_key: str | None = None
     parent_stage: str = ""
@@ -507,7 +503,6 @@ class State:
             "name",
             "model",
             "artifact_dir",
-            "instructions",
             "repo",
             "cwd",
             "base_ref",
@@ -521,7 +516,6 @@ def framework_subs(self, stage: StageProtocol) -> dict[str, str]:
             "name": stage.name,
             "model": self.client.model,
             "artifact_dir": str(self.artifact_dir),
-            "instructions": self.instructions,
             "repo": self.repo,
             "cwd": self.cwd,
             "base_ref": self.base_ref,
@@ -533,12 +527,9 @@ def setup_dirs(
         state_dir: pathlib.Path,
         artifact_dir: pathlib.Path,
         gremlin_id: str | None,
-        *,
-        instructions: str = "",
     ) -> None:
         state_dir.mkdir(parents=True, exist_ok=True)
         artifact_dir.mkdir(parents=True, exist_ok=True)
-        (state_dir / "instructions.txt").write_text(instructions, encoding="utf-8")
         sf = state_dir / "state.json"
         if gremlin_id and not sf.exists():
             write_state(state_dir, {"id": gremlin_id})
@@ -624,7 +615,6 @@ def build_state(
     pipeline_data: Pipeline | None = None,
     repo: str = "",
     cwd: str = "",
-    instructions: str = "",
     worktree: pathlib.Path | None = None,
     worktree_parent: pathlib.Path | None = None,
     artifacts: ArtifactRegistry | None = None,
@@ -643,7 +633,6 @@ def build_state(
         or (str(worktree) if worktree is not None else str(_paths.project_root())),
         args=args if args is not None else argparse.Namespace(),
         pipeline_data=pipeline_data,
-        instructions=instructions,
         worktree=worktree,
         worktree_parent=worktree_parent,
         child_key=child_key,