Context aware steps

.claude/commands/map-efficient.md

@@ -390,6 +390,10 @@ pytest --tb=short 2>&1 || true
 When TDD mode is active, Actor receives `<TDD_Mode>code_only</TDD_Mode>` and must NOT modify test files. When TDD is off, standard behavior.
 
 ```python
+# Context assembly: use build_context_block() from map_step_runner.py
+# to generate <map_context> when called programmatically.
+# For manual invocation, construct the block from blueprint.json + step_state.json.
+
 Task(
   subagent_type="actor",
   description="Implement subtask [ID]",
@@ -399,16 +403,40 @@ Task(
 [AAG contract from decomposition: Actor -> Action -> Goal]
 </MAP_Contract>
 
-Subtask: [ID] [title]
+<map_context>
+# Goal
+[Goal from task_plan.md — one sentence]
+
+# Current Subtask: [ID] — [title]
+AAG Contract: [contract from blueprint]
 Affected files: [from blueprint]
-Validation criteria: [from blueprint]
+Validation criteria:
+- [criteria from blueprint]
+
+# Plan Overview ([N] subtasks):
+[For each subtask in blueprint, show one-liner with status:]
+- [x] ST-001: Title (complete)
+- [ ] ST-002: Title (pending)
+- [>>] ST-003: Title (IN PROGRESS) <- current
+
+# Upstream Results (dependencies of current subtask):
+[Only for subtasks that current depends on, from step_state.json subtask_results:]
+ST-001: files=[a.py, b.py], status=valid
+
+# Repo Delta (files changed since last subtask):
+[From compute_differential_insight(), if last_subtask_commit_sha available]
+[Omit this section entirely if no previous SHA (first subtask)]
+</map_context>
 
 Protocol:
-1. Parse MAP_Contract — this is your compilation target
-2. Read affected files to understand current state
-3. Implement: translate MAP_Contract into code
-4. Apply code with Edit/Write tools
-5. Output: approach + files_changed + trade-offs"""
+1. SCOPE: Implement ONLY the Current Subtask. Do NOT modify files belonging to other subtasks.
+2. Plan Overview is for orientation — do NOT implement other subtasks.
+3. Upstream Results show what dependencies produced — use as input context.
+4. Parse MAP_Contract — this is your compilation target.
+5. Read affected files to understand current state.
+6. Implement: translate MAP_Contract into code.
+7. Apply code with Edit/Write tools.
+8. Output: approach + files_changed + trade-offs"""
 )
 ```
 
@@ -452,6 +480,16 @@ if echo "$TEST_GATE" | python3 -c "import sys,json; d=json.load(sys.stdin); sys.
     # Tests passed — snapshot code state for artifact verification
     SNAPSHOT=$(python3 .map/scripts/map_step_runner.py snapshot_code_state)
     # Append git ref to review artifact header (if code-review file exists)
+
+    # Record subtask result for context-aware injection (Upstream Results + Repo Delta)
+    # Uses record_subtask_result CLI dispatch via stdin JSON (injection-safe, single source of truth).
+    FILES_JSON=$(echo "$SNAPSHOT" | python3 -c "import sys,json; print(json.dumps(json.load(sys.stdin).get('files_changed',[])))")
+    CURRENT_SHA=$(git rev-parse HEAD 2>/dev/null || echo "")
+    if [ -n "$CURRENT_SHA" ]; then
+        echo "{\"files\": ${FILES_JSON}, \"status\": \"valid\", \"summary\": \"Monitor passed + tests passed\", \"commit_sha\": \"${CURRENT_SHA}\"}" | python3 .map/scripts/map_step_runner.py record_subtask_result
+    else
+        echo "{\"files\": ${FILES_JSON}, \"status\": \"valid\", \"summary\": \"Monitor passed + tests passed\"}" | python3 .map/scripts/map_step_runner.py record_subtask_result
+    fi
 fi
 
 # After Monitor returns:

.claude/hooks/workflow-context-injector.py

@@ -18,6 +18,9 @@
 import os
 import re
 import sys
+
+# Keep in sync with map_step_runner.py GOAL_HEADING_RE
+GOAL_HEADING_RE = r"## (?:Goal|Overview)\n(.*?)(?=\n##|\Z)"
 from pathlib import Path
 
 # Bash commands that don't need workflow reminders
@@ -161,8 +164,61 @@ def required_action_for_step(step_id: str, step_phase: str, state: dict) -> str
     return None
 
 
+def load_goal_and_title(branch: str, subtask_id: str) -> tuple[str, str]:
+    """Load goal from task_plan and subtask title from blueprint.
+
+    Returns (truncated_goal, subtask_title) or ("", "") on any error.
+    Fast: single json.load + single regex — target <20ms.
+    """
+    project_dir = Path(os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd()))
+    goal = ""
+    title = ""
+
+    # Goal from task_plan.md — matches ## Goal or ## Overview headings
+    plan_file = project_dir / ".map" / branch / f"task_plan_{branch}.md"
+    try:
+        if plan_file.exists():
+            content = plan_file.read_text(encoding="utf-8")
+            match = re.search(GOAL_HEADING_RE, content, re.DOTALL)
+            if match:
+                goal = match.group(1).strip()
+                # Truncate to first sentence
+                if ". " in goal:
+                    goal = goal[: goal.index(". ") + 1]
+                if len(goal) > 80:
+                    goal = goal[:77] + "..."
+    except OSError:
+        pass
+
+    # Title from blueprint.json
+    blueprint_file = project_dir / ".map" / branch / "blueprint.json"
+    try:
+        if blueprint_file.exists():
+            bp = json.loads(blueprint_file.read_text(encoding="utf-8"))
+            for st in bp.get("subtasks", []):
+                if st.get("id") == subtask_id:
+                    title = st.get("title", "")
+                    break
+    except (json.JSONDecodeError, OSError):
+        pass
+
+    return (goal, title)
+
+
+def _truncate_at_word(text: str, limit: int) -> str:
+    """Truncate text at word boundary, appending '...' within limit."""
+    if len(text) <= limit:
+        return text
+    cut = text[: limit - 3]
+    # Find last space to avoid cutting mid-word
+    last_space = cut.rfind(" ")
+    if last_space > limit // 2:
+        cut = cut[:last_space]
+    return cut + "..."
+
+
 def format_reminder(state: dict, branch: str) -> str | None:
-    """Format terse workflow reminder (aim: ~150-200 chars)."""
+    """Format terse workflow reminder (aim: ≤500 chars)."""
     if not state:
         return None
 
@@ -216,9 +272,30 @@ def format_reminder(state: dict, branch: str) -> str | None:
     if not step_id and not step_phase:
         return None
 
-    base = f"[MAP] {step_id} {step_phase} | ST: {subtask_id} ({progress}) | plan:{plan_ok} mode:{mode}{wave_hint}{diag_hint}{files_hint}"
+    # Context-aware: add goal and subtask title
+    goal_hint = ""
+    title_hint = ""
+    if subtask_id != "-":
+        goal, title = load_goal_and_title(branch, subtask_id)
+        if goal:
+            goal_hint = f" | Goal: {goal}"
+        if title:
+            title_hint = f" {title}"
+
+    base = f"[MAP] {step_id} {step_phase}{goal_hint} | ST: {subtask_id}{title_hint} ({progress}) | plan:{plan_ok} mode:{mode}{wave_hint}{diag_hint}{files_hint}"
+
+    # Enforce 500-char limit: trim goal first, then word-boundary truncate
+    if len(base) > 500:
+        goal_hint = ""
+        base = f"[MAP] {step_id} {step_phase} | ST: {subtask_id}{title_hint} ({progress}) | plan:{plan_ok} mode:{mode}{wave_hint}{diag_hint}{files_hint}"
+    if len(base) > 500:
+        base = _truncate_at_word(base, 500)
+
     if required:
-        return f"{base} | REQUIRED: {required}"
+        result = f"{base} | REQUIRED: {required}"
+        if len(result) > 500:
+            result = _truncate_at_word(result, 500)
+        return result
     return base
 
 

.map/scripts/map_orchestrator.py

@@ -298,6 +298,25 @@ class StepState:
     subtask_files_changed: dict[str, list[str]] = field(default_factory=dict)
     guard_rework_counts: dict[str, int] = field(default_factory=dict)
     constraints: Optional[dict] = None
+    subtask_results: dict[str, dict] = field(default_factory=dict)
+    last_subtask_commit_sha: Optional[str] = None
+
+    def record_subtask_result(
+        self,
+        subtask_id: str,
+        files_changed: list[str],
+        status: str,
+        summary: str = "",
+        commit_sha: Optional[str] = None,
+    ) -> None:
+        """Record result of a completed subtask for context injection."""
+        self.subtask_results[subtask_id] = {
+            "files_changed": files_changed,
+            "status": status,
+            "summary": summary,
+        }
+        if commit_sha:
+            self.last_subtask_commit_sha = commit_sha
 
     def to_dict(self) -> dict:
         """Serialize to dictionary."""
@@ -325,6 +344,8 @@ def to_dict(self) -> dict:
             "subtask_files_changed": self.subtask_files_changed,
             "guard_rework_counts": self.guard_rework_counts,
             "constraints": self.constraints,
+            "subtask_results": self.subtask_results,
+            "last_subtask_commit_sha": self.last_subtask_commit_sha,
         }
 
     @classmethod
@@ -354,6 +375,8 @@ def from_dict(cls, data: dict) -> "StepState":
             subtask_files_changed=data.get("subtask_files_changed", {}),
             guard_rework_counts=data.get("guard_rework_counts", {}),
             constraints=data.get("constraints"),
+            subtask_results=data.get("subtask_results", {}),
+            last_subtask_commit_sha=data.get("last_subtask_commit_sha"),
         )
 
     @classmethod