More iteration

nightshift/pipeline.py

@@ -529,10 +529,11 @@ class PipelineRunner:
                 format_semantic_index(index),
             )
             query = " ".join([task.title, task.description, *task.acceptance_criteria])
+            results = _task_semantic_results(index, query, task.id)
             context_path = self.artifacts.write_stage_output(
                 task.id,
                 stage.output or "semantic-context.md",
-                format_search_results(search_index(index, query, limit=8), query),
+                format_search_results(results, query),
             )
             self.logger.event(
                 "artifact.write",
@@ -1246,15 +1247,17 @@ class PipelineRunner:
     def _build_context_pack(self, task: Task) -> str:
         terms = _task_search_terms(task)
         lookup_paths = self.config.safety.scoped_paths or (".",)
-        files = self._list_context_files(lookup_paths)
+        files = self._list_context_files(lookup_paths, task.id)
         grep_sections: list[str] = []
         for term in terms[:5]:
             scoped_results = []
             for path in lookup_paths:
+                grep_output = self.repo_tools.grep(re.escape(term), path, max_matches=20).rstrip()
+                grep_output = _filter_future_task_test_lines(grep_output, task.id)
                 scoped_results.append(
                     f"#### Path: {path}\n\n"
                     "```text\n"
-                    f"{self.repo_tools.grep(re.escape(term), path, max_matches=20).rstrip()}\n"
+                    f"{grep_output}\n"
                     "```"
                 )
             grep_sections.extend(
@@ -1294,13 +1297,15 @@ class PipelineRunner:
             ]
         )
 
-    def _list_context_files(self, paths: tuple[str, ...]) -> str:
+    def _list_context_files(self, paths: tuple[str, ...], task_id: str) -> str:
         sections: list[str] = []
         for path in paths:
+            files = self.repo_tools.list_files(path, pattern="*", max_files=80).rstrip()
+            files = _filter_future_task_test_lines(files, task_id)
             sections.extend(
                 [
                     f"## Path: {path}",
-                    self.repo_tools.list_files(path, pattern="*", max_files=80).rstrip(),
+                    files,
                     "",
                 ]
             )
@@ -1480,6 +1485,39 @@ def _extract_exit_code(text: str) -> int | None:
         return None
 
 
+def _task_semantic_results(index, query: str, task_id: str):
+    current_test_path = _current_task_test_path(task_id)
+    current = tuple(item for item in index if item.path == current_test_path)
+    current_paths = {item.path for item in current}
+    searched = search_index(index, query, limit=8)
+    filtered = tuple(
+        item
+        for item in searched
+        if item.path not in current_paths and not _future_task_test_path(item.path, current_test_path)
+    )
+    return (*current, *filtered)[:8]
+
+
+def _future_task_test_path(path: str, current_test_path: str) -> bool:
+    return bool(re.fullmatch(r"tests/test_task\d+\.py", path)) and path != current_test_path
+
+
+def _current_task_test_path(task_id: str) -> str:
+    return f"tests/test_{task_id.lower().replace('-', '')}.py"
+
+
+def _filter_future_task_test_lines(text: str, task_id: str) -> str:
+    current_test_path = _current_task_test_path(task_id)
+    kept: list[str] = []
+    for line in text.splitlines():
+        normalized = line.replace("\\", "/")
+        matches = re.findall(r"tests/test_task\d+\.py", normalized)
+        if matches and all(path != current_test_path for path in matches):
+            continue
+        kept.append(line)
+    return "\n".join(kept)
+
+
 def _repeated_protected_path_violation(entries: tuple[RetryMemoryEntry, ...]) -> bool:
     recent = entries[-2:]
     if len(recent) < 2:

nightshift/project_templates/tutorial-deaddrop/.nightshift/agents/implementer.md

@@ -1,13 +1,14 @@
 You are the implementation agent for the NightShift DeadDrop tutorial.
 
-Implement the smallest application change that satisfies the current task and the generated tests.
+Implement the smallest application change that satisfies the current task and its fixed test file.
-Do not rewrite generated tests unless the retry context explicitly says they are inaccurate.
 Do not edit files under `tests/`. The tutorial tests are fixed; make the application satisfy them.
 Do not add behavior for future tasks unless needed to satisfy the current tests.
 Use Flask and `sqlite3` from the Python standard library. Do not use SQLAlchemy, Flask-SQLAlchemy, or undeclared dependencies.
 Keep the public package name `deaddrop_app`.
 Keep the public app entry point `create_app(database_path: str | None = None)`.
 Respect `database_path`; do not hard-code `snippets.db` when a database path is supplied.
+For `TASK-001`, satisfy only `tests/test_task001.py`: accept JSON `POST /snippets`, persist title/body, return an integer `id`, return exactly `id`, `title`, and `body` from `GET /snippets/<id>`, and return 404 for missing snippets.
+Do not add `language`, `tags`, `expires_at`, listing, forms, templates, or other future-task behavior while implementing `TASK-001`.
 Tests should interact through HTTP routes and `create_app`, not through ORM/session globals.
 Do not use `app.before_first_request`; recent Flask versions removed it. Initialize required database tables inside `create_app` or inside the route helper before use.
 When adding columns to an existing sqlite table, handle existing databases idempotently with `ALTER TABLE` checks or a simple migration helper. `CREATE TABLE IF NOT EXISTS` does not add columns to an existing table.

nightshift/project_templates/tutorial-deaddrop/.nightshift/agents/planner.md

@@ -1,15 +1,16 @@
 You are the planning agent for the NightShift DeadDrop tutorial.
 
-Create a concise TDD implementation plan for the current task.
+Create a concise implementation plan for the current task.
 
 Plan in this order:
-1. Which acceptance tests should be generated for only this task.
+1. Which fixed current-task test file defines the contract.
 2. Which application files likely need to change.
-3. The smallest implementation slice that should make those tests pass.
+3. The smallest implementation slice that should make the current-task tests pass.
 
 If repository context is needed, request it with lookup_requests.
 Prefer small edits and deterministic tests.
 Use the actual package and files from repository context. For this tutorial the public app entry point is `deaddrop_app.app:create_app`.
+For `TASK-001`, the current contract is `tests/test_task001.py`; ignore future task tests.
 Do not assume top-level modules such as `app`, `models`, `routes`, or `main` exist.
 Do not propose SQLAlchemy, Flask-SQLAlchemy, or ORM globals. Use Flask plus `sqlite3` from the Python standard library.
 Do not propose tests that import `session`, `Snippet`, `engine`, or other implementation internals.

nightshift/semantic_index.py

@@ -126,6 +126,8 @@ def _skip(path: Path, root: Path) -> bool:
     parts = set(Path(relative).parts)
     if parts & {".git", ".nightshift", "__pycache__", ".venv", "venv", "integ_runs"}:
         return True
+    if any(part.endswith(".egg-info") for part in parts):
+        return True
     return path.suffix.lower() not in {".py", ".md", ".txt", ".yaml", ".yml", ".toml", ".html", ".css", ".js"}
 
 

tests/test_telemetry_index.py

@@ -71,10 +71,15 @@ class TelemetryAndIndexTests(unittest.TestCase):
             root = Path(directory)
             (root / "src").mkdir()
             (root / "tests").mkdir()
+            (root / "src" / "demo.egg-info").mkdir()
             (root / "src" / "service.py").write_text(
                 "import sqlite3\n\nclass SnippetStore:\n    pass\n\ndef create_snippet():\n    return True\n",
                 encoding="utf-8",
             )
+            (root / "src" / "demo.egg-info" / "PKG-INFO").write_text(
+                "Name: generated-metadata\n",
+                encoding="utf-8",
+            )
             (root / "tests" / "test_service.py").write_text(
                 "def test_create_snippet():\n    assert True\n",
                 encoding="utf-8",
@@ -91,6 +96,7 @@ class TelemetryAndIndexTests(unittest.TestCase):
 
             self.assertTrue(any("create_snippet" in item.symbols for item in index))
             self.assertTrue(any(item.path == "src/service.py" for item in results))
+            self.assertFalse(any(".egg-info" in item.path for item in index))
 
     def test_semantic_context_stage_writes_artifacts(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
@@ -108,6 +114,66 @@ class TelemetryAndIndexTests(unittest.TestCase):
             self.assertTrue((task_dir / "semantic-index.md").exists())
             self.assertTrue((task_dir / "semantic-context.md").exists())
 
+    def test_semantic_context_prefers_current_task_test_and_excludes_future_task_tests(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            _write_common_files(root)
+            (root / "tests").mkdir(exist_ok=True)
+            (root / "tests" / "test_task001.py").write_text(
+                "def test_create_snippet_returns_id():\n    assert True\n",
+                encoding="utf-8",
+            )
+            (root / "tests" / "test_task002.py").write_text(
+                "def test_create_snippet_accepts_language_and_tags():\n    assert True\n",
+                encoding="utf-8",
+            )
+            stages = (StageConfig(id="semantic", type="semantic_context", output="semantic-context.md"),)
+            config = make_config(root, stages)
+            runner = PipelineRunner(config, ArtifactStore(root, ".nightshift", run_id="test-run"))
+
+            result = runner.run_task(parse_tasks(TASK_MD)[0])
+
+            task_dir = root / ".nightshift" / "runs" / "test-run" / "tasks" / "TASK-001"
+            context = (task_dir / "semantic-context.md").read_text(encoding="utf-8")
+            self.assertEqual(result.status, "complete")
+            self.assertIn("## `tests/test_task001.py`", context)
+            self.assertNotIn("## `tests/test_task002.py`", context)
+
+    def test_repo_context_excludes_future_task_test_hits(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            _write_common_files(root)
+            (root / "tests").mkdir(exist_ok=True)
+            (root / "tests" / "test_task001.py").write_text(
+                "def test_create_snippet_returns_id():\n    assert True\n",
+                encoding="utf-8",
+            )
+            (root / "tests" / "test_task002.py").write_text(
+                "def test_create_snippet_accepts_language_and_tags():\n    assert True\n",
+                encoding="utf-8",
+            )
+            task_md = """# Tasks
+
+- [ ] TASK-001: Snippet creation
+
+Description:
+Create snippets.
+
+Acceptance Criteria:
+- POST /snippets creates a snippet
+"""
+            stages = (StageConfig(id="context", type="repo_context", output="context-pack.md"),)
+            config = make_config(root, stages)
+            runner = PipelineRunner(config, ArtifactStore(root, ".nightshift", run_id="test-run"))
+
+            result = runner.run_task(parse_tasks(task_md)[0])
+
+            task_dir = root / ".nightshift" / "runs" / "test-run" / "tasks" / "TASK-001"
+            context = (task_dir / "context-pack.md").read_text(encoding="utf-8")
+            self.assertEqual(result.status, "complete")
+            self.assertIn("tests/test_task001.py", context)
+            self.assertNotIn("tests/test_task002.py", context)
+
 
 if __name__ == "__main__":
     unittest.main()