fix: address augment validation findings

Author: bma-d

skills/bmad-story-automator/src/story_automator/commands/agent_config_cmd.py

@@ -102,3 +102,6 @@ def _load_presets_or_report(file_path: str) -> dict | None:
     except (OSError, UnicodeDecodeError) as exc:
         print_json({"ok": False, "error": "presets_file_error", "reason": str(exc)})
         return None
+    except ValueError:
+        print_json({"ok": False, "error": "invalid_presets_json"})
+        return None

skills/bmad-story-automator/src/story_automator/commands/orchestrator_state.py

@@ -76,8 +76,8 @@ def _replace_frontmatter_values(frontmatter: str, updates: list[tuple[str, str]]
 
 def _split_frontmatter(text: str) -> tuple[str, str]:
     if not text.startswith("---"):
-        return text, ""
+        return "", text
     parts = text.split("---", 2)
     if len(parts) < 3:
-        return text, ""
+        return "", text
     return f"{parts[0]}---{parts[1]}---", parts[2]

skills/bmad-story-automator/src/story_automator/commands/tmux.py

@@ -363,10 +363,10 @@ def cmd_monitor_session(args: list[str]) -> int:
     start = time.time()
     last_done = 0
     last_total = 0
+    session_state_issue = monitor_session_state_issue(session, project_root) if json_output else None
     for _poll in range(1, max_polls + 1):
         if time.time() - start >= timeout_minutes * 60:
             return emit_monitor_result(json_output, "timeout", last_done, last_total, "", f"exceeded_{timeout_minutes}m")
-        pre_status_issue = monitor_session_state_issue(session, project_root) if json_output else None
         status = session_status(session, full=False, codex=agent == "codex", project_root=project_root, mode=runtime_mode())
         if int(status["todos_done"]) or int(status["todos_total"]):
             last_done = int(status["todos_done"])
@@ -418,7 +418,7 @@ def cmd_monitor_session(args: list[str]) -> int:
             output = session_status(session, full=True, codex=agent == "codex", project_root=project_root, mode=runtime_mode())["active_task"]
             return emit_monitor_result(json_output, "stuck", 0, 0, str(output), "never_active")
         if state == "not_found":
-            issue = pre_status_issue or monitor_session_state_issue(session, project_root)
+            issue = session_state_issue or (monitor_session_state_issue(session, project_root) if json_output else None)
             return emit_monitor_result(json_output, "not_found", last_done, last_total, "", "session_gone", structured_issue=issue)
         time.sleep(min(180 if agent == "codex" else 120, max(5, int(status["wait_estimate"]))))
     output = session_status(session, full=True, codex=agent == "codex", project_root=project_root, mode=runtime_mode())["active_task"]

skills/bmad-story-automator/src/story_automator/core/agent_config.py

@@ -40,8 +40,12 @@ def load_presets_file(path: str | Path) -> dict[str, Any]:
     if not file_exists(preset_path):
         return {"version": "1.0.0", "presets": []}
     data = json.loads(read_text(preset_path))
+    if not isinstance(data, dict):
+        raise ValueError("presets file must be an object")
     data.setdefault("version", "1.0.0")
     data.setdefault("presets", [])
+    if not isinstance(data["presets"], list):
+        raise ValueError("presets file presets must be an array")
     return data
 
 
@@ -56,7 +60,7 @@ def parse_agent_config_json(raw: str) -> AgentConfigResolved:
         raise ValueError("agentConfig must be an object")
     config = AgentConfigResolved()
     if "agentConfig" in data and data.get("agentConfig") not in ("", None):
-        raise ValueError("agentConfig must be an object")
+        raise ValueError("unexpected nested agentConfig key; pass the inner config object directly")
     config.default_primary = data.get("defaultPrimary") or data.get("primary") or "auto"
     if "defaultFallback" in data:
         fallback_raw = data.get("defaultFallback")

skills/bmad-story-automator/src/story_automator/core/agent_plan.py

@@ -36,11 +36,9 @@ def validate_complexity_payload(payload: object) -> list[DiagnosticIssue]:
         story_id = story.get("storyId")
         if not isinstance(story_id, str) or not story_id.strip():
             issues.append(_issue("missing_field", f"{field}.storyId", "non-empty string", story_id, "Complexity storyId must be a non-empty string"))
-        complexity = story.get("complexity")
-        if complexity is None:
-            complexity = {}
-        elif not isinstance(complexity, dict):
-            issues.append(_issue("invalid_type", f"{field}.complexity", "object", complexity, "Complexity must be an object"))
+        complexity, issue = _story_complexity(story, field)
+        if issue:
+            issues.append(issue)
             continue
         level = str(complexity.get("level") or "medium").strip().lower()
         if level not in COMPLEXITY_LEVELS:
@@ -132,8 +130,8 @@ def build_agents_file(
         raise AgentPlanInputError("complexity-file", ValueError(message)) from None
 
     stories = []
-    for story in complexity_payload.get("stories", []):
-        level = _story_complexity_level(story)
+    for index, story in enumerate(complexity_payload.get("stories", [])):
+        level = _story_complexity_level(story, f"stories[{index}]")
         stories.append({"storyId": story.get("storyId"), "title": str(story.get("title") or ""), "complexity": level, "tasks": _tasks_for(config, level)})
     try:
         epic = find_frontmatter_value(state_file, "epic")
@@ -198,12 +196,19 @@ def _load_agents_plan_payload(path: str) -> tuple[dict[str, Any], list[Diagnosti
     return payload, []
 
 
-def _story_complexity_level(story: dict[str, Any]) -> str:
+def _story_complexity(story: dict[str, Any], field: str) -> tuple[dict[str, Any], DiagnosticIssue | None]:
     complexity = story.get("complexity")
     if complexity is None:
-        return "medium"
+        return {}, None
     if not isinstance(complexity, dict):
-        raise AgentPlanInputError("complexity-file", ValueError("Complexity must be an object"))
+        return {}, _issue("invalid_type", f"{field}.complexity", "object", complexity, "Complexity must be an object")
+    return complexity, None
+
+
+def _story_complexity_level(story: dict[str, Any], field: str) -> str:
+    complexity, issue = _story_complexity(story, field)
+    if issue:
+        raise AgentPlanInputError("complexity-file", ValueError(legacy_issue_message(issue)))
     return str(complexity.get("level") or "medium").strip().lower() or "medium"
 
 

skills/bmad-story-automator/src/story_automator/core/state_validation.py

@@ -145,7 +145,7 @@ def parse_state_update_argument(raw: str) -> tuple[str, str] | dict[str, Any]:
     key, value = raw.split("=", 1)
     if not key.strip():
         return state_update_argument_error_payload(raw)
-    return key.strip(), value
+    return key.strip(), value.strip()
 
 
 def state_validation_payload(issues: list[DiagnosticIssue]) -> dict[str, Any]:

tests/test_agent_config_model.py

@@ -60,6 +60,10 @@ def test_agent_cli_treats_empty_model_as_absent(self) -> None:
 
 
 class CoreAgentConfigModelTests(unittest.TestCase):
+    def test_parse_agent_config_json_rejects_nested_agent_config_with_clear_message(self) -> None:
+        with self.assertRaisesRegex(ValueError, "unexpected nested agentConfig key"):
+            parse_agent_config_json(json.dumps({"agentConfig": {"defaultPrimary": "codex"}}))
+
     def test_per_task_model_is_resolved(self) -> None:
         config = parse_agent_config_json(
             json.dumps(

tests/test_cli_contracts.py

@@ -170,6 +170,16 @@ def test_presets_decode_error_returns_stable_error(self) -> None:
         self.assertEqual(code, 1)
         self.assertEqual(payload["error"], "presets_file_error")
 
+    def test_presets_wrong_shape_returns_stable_error(self) -> None:
+        for payload_text in ("[]", '"bad"', '{"presets": {}}'):
+            with self.subTest(payload=payload_text):
+                self.presets.write_text(payload_text, encoding="utf-8")
+
+                code, payload = self._agent(["list", "--file", str(self.presets)])
+
+                self.assertEqual(code, 1)
+                self.assertEqual(payload["error"], "invalid_presets_json")
+
     def _agent(self, args: list[str]) -> tuple[int, dict[str, object]]:
         stdout = io.StringIO()
         with redirect_stdout(stdout):

tests/test_state_validation.py

@@ -265,6 +265,26 @@ def test_state_update_only_rewrites_frontmatter(self) -> None:
         self.assertIn("status: body-marker", body)
         self.assertIn("currentStep: body-step", body)
 
+    def test_state_update_rejects_file_without_frontmatter_without_rewriting_body(self) -> None:
+        state_file = self.project_root / "body-only.md"
+        state_file.write_text("body\nstatus: body-marker\n", encoding="utf-8")
+
+        code, payload = self._state_update(state_file, "status=READY")
+
+        self.assertEqual(code, 1)
+        self.assertEqual(payload, {"ok": False, "error": "keys_not_found", "updated": []})
+        self.assertEqual(state_file.read_text(encoding="utf-8"), "body\nstatus: body-marker\n")
+
+    def test_state_update_rejects_unterminated_frontmatter_without_rewriting_body(self) -> None:
+        state_file = self.project_root / "unterminated.md"
+        state_file.write_text("---\nstatus: body-marker\n", encoding="utf-8")
+
+        code, payload = self._state_update(state_file, "status=READY")
+
+        self.assertEqual(code, 1)
+        self.assertEqual(payload, {"ok": False, "error": "keys_not_found", "updated": []})
+        self.assertEqual(state_file.read_text(encoding="utf-8"), "---\nstatus: body-marker\n")
+
     def test_state_update_strips_set_key_whitespace(self) -> None:
         state_file = self._build_state_config(status="READY")
 
@@ -273,6 +293,15 @@ def test_state_update_strips_set_key_whitespace(self) -> None:
         self.assertEqual(code, 1)
         self.assertEqual(payload["error"], "invalid_status_transition")
 
+    def test_state_update_strips_set_value_whitespace(self) -> None:
+        state_file = self._build_state_config(status="READY")
+
+        code, payload = self._state_update(state_file, " status = IN_PROGRESS")
+
+        self.assertEqual(code, 0)
+        self.assertEqual(payload, {"ok": True, "updated": ["status"]})
+        self.assertIn("status: IN_PROGRESS", state_file.read_text(encoding="utf-8"))
+
     def _validate_state(self, state_file: Path) -> dict[str, object]:
         stdout = io.StringIO()
         with patch_env(self.project_root), redirect_stdout(stdout):

tests/test_success_verifiers.py

@@ -287,6 +287,23 @@ def test_monitor_session_json_reports_malformed_session_state_when_session_gone(
         self.assertEqual(payload["final_state"], "not_found")
         self.assertEqual(payload["structuredIssues"][0]["type"], "session_state.invalid_json")
 
+    def test_monitor_session_checks_session_state_issue_only_when_session_is_gone(self) -> None:
+        session = "sa-test-session"
+        statuses = [
+            {"active_task": "", "todos_done": 0, "todos_total": 0, "wait_estimate": 0, "session_state": "running"},
+            {"active_task": "", "todos_done": 0, "todos_total": 0, "wait_estimate": 0, "session_state": "running"},
+            {"active_task": "", "todos_done": 0, "todos_total": 0, "wait_estimate": 0, "session_state": "not_found"},
+        ]
+        stdout = io.StringIO()
+        with patch_env(self.project_root), patch("story_automator.commands.tmux.time.sleep"), patch(
+            "story_automator.commands.tmux.session_status",
+            side_effect=statuses,
+        ), patch("story_automator.commands.tmux.monitor_session_state_issue", return_value=None) as state_issue_mock, redirect_stdout(stdout):
+            code = cmd_monitor_session([session, "--json", "--max-polls", "3"])
+
+        self.assertEqual(code, 0)
+        self.assertEqual(state_issue_mock.call_count, 2)
+
     def test_monitor_session_csv_does_not_include_structured_issues(self) -> None:
         session = "sa-test-session"
         paths = session_paths(session, self.project_root)