fix: vendor safety, preserve domain notes, ask/step JSON hygiene

src/bedside/commands/ask_cmd.py

@@ -110,22 +110,24 @@ def run_ask(
                 "or use the agent host structured choice UI for this gate."
             )
             r.line(
-                f"Record: bedside.ask id={gate_id} choice= pending "
+                f"Record: bedside.ask id={gate_id} choice=pending "
                 f"recommended={recommended} matched=false"
             )
             return r
         pre = CommandResult(OK)
         _emit_prompt(pre, gate_id, prompt, ordered, recommended)
         # Print before blocking: CLI only flushes CommandResult after return.
-        _print_now(pre.messages)
+        # Keep stdout JSON-clean when --json.
+        _print_now(pre.messages, to_stderr=json_out)
         interactive_flushed = True
         try:
             if input_fn is not None:
                 raw = input_fn("Answer: ")
             else:
                 stream = stdin if stdin is not None else sys.stdin
-                sys.stdout.write("Answer: ")
-                sys.stdout.flush()
+                prompt_stream = sys.stderr if json_out else sys.stdout
+                prompt_stream.write("Answer: ")
+                prompt_stream.flush()
                 raw = stream.readline()
                 if not raw:
                     raw = ""
@@ -173,10 +175,14 @@ def run_ask(
     return r
 
 
-def _print_now(messages: list[str]) -> None:
-    """Flush operator-facing lines before a blocking stdin read."""
+def _print_now(messages: list[str], *, to_stderr: bool = False) -> None:
+    """Flush operator-facing lines before a blocking stdin read.
+
+    When json_out is true, human text goes to stderr so stdout stays JSON-only.
+    """
+    stream = sys.stderr if to_stderr else sys.stdout
     for line in messages:
-        print(line, flush=True)
+        print(line, file=stream, flush=True)
 
 
 def _emit_prompt(

src/bedside/commands/init_cmd.py

@@ -94,12 +94,14 @@ def run_init(
         dest = (root / vendor_dest).resolve()
         try:
             copied = vendor_copy(src, dest, include_src=include_src)
-        except (OSError, FileNotFoundError) as e:
+        except (OSError, FileNotFoundError, ValueError) as e:
+            # ValueError: self-vendor / dest nested under source safety guards.
             bad = CommandResult(SETUP_ERROR)
             bad.line(f"Vendor copy failed: {e}")
             bad.line(
                 "What to do next: pass a local tig/bedside checkout to "
-                "`--vendor-from` (must contain contract/)."
+                "`--vendor-from` (must contain contract/). Do not point "
+                "`--vendor-from` at the existing product vendor tree itself."
             )
             return bad
         detected = detect_pin(src)
@@ -139,7 +141,9 @@ def run_init(
 
     if not skip_domain_notes:
         notes = root / cfg.domain_notes
-        if notes.is_file() and not force:
+        # Never clobber product domain notes on --force re-vendor; only scaffold
+        # when missing. Agents fill BEDSIDE.md once; refresh updates third_party only.
+        if notes.is_file():
             r.line(f"Left existing {cfg.domain_notes}")
         else:
             notes.write_text(BEDSIDE_MD, encoding="utf-8")

src/bedside/commands/step_cmd.py

@@ -79,15 +79,17 @@ def run_step(
         pre = CommandResult(OK)
         _emit_step(pre, gate_id, prompt, expect)
         # Print before blocking: CLI only flushes CommandResult after return.
-        _print_now(pre.messages)
+        # Keep stdout JSON-clean when --json.
+        _print_now(pre.messages, to_stderr=json_out)
         interactive_flushed = True
         try:
             if input_fn is not None:
                 raw = input_fn("Confirm (yes/no): ")
             else:
                 stream = stdin if stdin is not None else sys.stdin
-                sys.stdout.write("Confirm (yes/no): ")
-                sys.stdout.flush()
+                prompt_stream = sys.stderr if json_out else sys.stdout
+                prompt_stream.write("Confirm (yes/no): ")
+                prompt_stream.flush()
                 raw = stream.readline()
                 if not raw:
                     raw = ""
@@ -132,10 +134,14 @@ def run_step(
     return r
 
 
-def _print_now(messages: list[str]) -> None:
-    """Flush operator-facing lines before a blocking stdin read."""
+def _print_now(messages: list[str], *, to_stderr: bool = False) -> None:
+    """Flush operator-facing lines before a blocking stdin read.
+
+    When json_out is true, human text goes to stderr so stdout stays JSON-only.
+    """
+    stream = sys.stderr if to_stderr else sys.stdout
     for line in messages:
-        print(line, flush=True)
+        print(line, file=stream, flush=True)
 
 
 def _emit_step(

src/bedside/vendor.py

@@ -52,6 +52,24 @@ def vendor_copy(
             "(point at a tig/bedside checkout, not a random folder)"
         )
 
+    # Never rmtree the source. Same path, or dest nested under source, would
+    # delete the only copy of contract/src before copytree runs.
+    if source == dest:
+        raise ValueError(
+            f"vendor source and dest are the same path: {source}. "
+            "Point --vendor-from at an upstream tig/bedside checkout, not the "
+            "existing product vendor tree."
+        )
+    try:
+        dest.relative_to(source)
+    except ValueError:
+        pass  # dest is not under source
+    else:
+        raise ValueError(
+            f"vendor dest {dest} is inside source {source}; refusing to delete "
+            "a nested destination that would wipe the source tree."
+        )
+
     if dest.exists():
         shutil.rmtree(dest)
     dest.mkdir(parents=True)

tests/test_ask_step.py

@@ -91,7 +91,11 @@ def test_ask_noninteractive_without_answer():
         stdin_isatty=False,
     )
     assert r.exit_code == HUMAN_NEEDED
-    assert "Human action needed" in "\n".join(r.messages)
+    joined = "\n".join(r.messages)
+    assert "Human action needed" in joined
+    # key=value form (not "choice= pending")
+    assert "choice=pending" in joined
+    assert "choice= pending" not in joined
 
 
 def test_ask_interactive_input_fn():

tests/test_cli_commands.py

@@ -108,3 +108,45 @@ def test_step_and_confirm_summary_uses_info_not_failed():
     assert "failed=" not in line
     if rep.info_failed:
         assert "info=" in line
+
+
+def test_init_force_preserves_existing_bedside_md(tmp_path: Path):
+    """Re-vendor with --force must not wipe product domain notes."""
+    notes = tmp_path / "BEDSIDE.md"
+    notes.write_text("# product metal notes\nkeep me\n", encoding="utf-8")
+    r = run_init(tmp_path, force=True)
+    assert r.exit_code == OK
+    text = notes.read_text(encoding="utf-8")
+    assert "keep me" in text
+    assert "Left existing BEDSIDE.md" in "\n".join(r.messages)
+
+
+def test_vendor_copy_rejects_self_path(tmp_path: Path):
+    from bedside.vendor import vendor_copy
+
+    src = tmp_path / "bedside"
+    (src / "contract").mkdir(parents=True)
+    (src / "contract" / "README.md").write_text("c\n", encoding="utf-8")
+    try:
+        vendor_copy(src, src)
+        raise AssertionError("expected ValueError for source == dest")
+    except ValueError as e:
+        assert "same path" in str(e)
+
+
+def test_init_vendor_self_path_is_setup_error(tmp_path: Path):
+    """Safety guard must exit 30 with recovery text, not traceback."""
+    vendor = tmp_path / "third_party" / "bedside"
+    (vendor / "contract").mkdir(parents=True)
+    (vendor / "contract" / "README.md").write_text("c\n", encoding="utf-8")
+    r = run_init(
+        tmp_path,
+        vendor_from=vendor,
+        vendor_dest="third_party/bedside",
+        force=True,
+    )
+    assert r.exit_code == SETUP_ERROR
+    joined = "\n".join(r.messages)
+    assert "Vendor copy failed" in joined
+    assert "same path" in joined or "inside source" in joined
+    assert "What to do next" in joined