verify(P4): migrate-ssm-except-narrowing — 4-layer verify PASS + ship

VERIFY: structural — narrowed excepts; behavioral — 48 tests + CI PASS; security — non-ParameterNotFound errors surface; quality — 2 new + 2 updated tests.

---ci---
project: acdl
phase: 4
milestone: v1.16
status: complete
phase_role: execution
requirements:
  covered: [REQ-168]
  partial: []
---/ci---

core/output_publisher.py

@@ -17,11 +17,15 @@ existing /acdl/... parameters to /nova/... and deletes the old ones.)
 import json
 import os
 import sys
+import urllib.error
+import urllib.request
 
 try:
     import boto3
+    from botocore.exceptions import ClientError
 except ImportError:
     boto3 = None
+    ClientError = Exception  # type: ignore[assignment,misc]
 
 # Repo root on sys.path so `from core import env` resolves to THIS package
 # when run as a script (avoids editable-installed third-party `core` shadow).
@@ -109,10 +113,12 @@ def publish_to_ssm(outputs, environment, contract_id):
                 Overwrite=True,
             )
             results[name] = param_name
-        except Exception as e:
+        except (ClientError, OSError) as e:
-            # Don't fail the pipeline if one output fails to publish, but log it
+            # P4 (REQ-168): narrow from bare `except Exception` to AWS +
+            # OS errors. Don't fail the pipeline if one output fails to
+            # publish, but log it with context.
             import sys
-            print(f"WARNING: SSM put_parameter failed for {name}: {e}", file=sys.stderr)
+            print(f"WARNING: SSM put_parameter failed for {name}: {type(e).__name__}: {e}", file=sys.stderr)
             results[name] = None
     return results
 
@@ -171,7 +177,6 @@ def post_github_comment(comment_text, token=None, repo=None, pr_number=None):
     if not token or not repo or not pr_number:
         return False  # not in a PR context or no token
     try:
-        import urllib.request
         url = f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments"
         data = json.dumps({"body": comment_text}).encode()
         req = urllib.request.Request(url, data=data, method="POST")
@@ -179,9 +184,12 @@ def post_github_comment(comment_text, token=None, repo=None, pr_number=None):
         req.add_header("Accept", "application/vnd.github+json")
         urllib.request.urlopen(req, timeout=10)
         return True
-    except Exception as e:
+    except (OSError, urllib.error.URLError, urllib.error.HTTPError) as e:
+        # P4 (REQ-168): narrow from bare `except Exception` to network +
+        # HTTP errors. Don't fail the pipeline if the PR comment can't be
+        # posted, but log it with context.
         import sys
-        print(f"WARNING: GitHub PR comment failed: {e}", file=sys.stderr)
+        print(f"WARNING: GitHub PR comment failed: {type(e).__name__}: {e}", file=sys.stderr)
         return False
 
 

scripts/migrate_ssm_paths.py

@@ -110,8 +110,18 @@ def copy_one_param(client, source_name: str, dest_name: str, force: bool = False
             return "skipped-equal"
         if not force:
             return "skipped-mismatch"
-    except Exception:  # ParameterNotFound → proceed to put
+    except client.exceptions.ParameterNotFound:
-        pass
+        pass  # target doesn't exist yet → proceed to put
+    except Exception as e:
+        # P4 (REQ-168): narrow the broad swallow — only ParameterNotFound
+        # is an expected "proceed to put" condition. Any other AWS error
+        # (auth, throttling, service) must surface, not be swallowed.
+        import sys
+        sys.stderr.write(
+            f"migrate_ssm_paths: get_parameter({dest_name}) failed: "
+            f"{type(e).__name__}: {e}\n"
+        )
+        raise
 
     put_kwargs = {
         "Name": dest_name,

tests/test_migrate_ssm_paths.py

@@ -74,4 +74,52 @@ class TestMapPath:
 
     def test_preserves_value_segment_exactly(self):
         # Hyphens, dots, underscores in output names are preserved
-        assert map_path("/acdl/dev/c-1/my.output-name_2") == "/nova/dev/c-1/my.output-name_2"
+        assert map_path("/acdl/dev/c-1/my.output-name_2") == "/nova/dev/c-1/my.output-name_2"
+
+class TestNarrowedException:
+    """P4 (REQ-168): the copy_one_param except is narrowed to
+    ParameterNotFound; non-ParameterNotFound errors surface (not swallowed)."""
+
+    def test_parameter_not_found_proceeds_to_put(self):
+        """A ParameterNotFound on the dest get_parameter (target absent) is
+        the expected 'proceed to put' path — not an error."""
+        from unittest import mock
+        import migrate_ssm_paths as m
+
+        class FakeExceptions:
+            ParameterNotFound = type("ParameterNotFound", (Exception,), {})
+
+        fake_client = mock.Mock()
+        fake_client.exceptions = FakeExceptions
+        # source get_parameter succeeds; dest get_parameter raises ParameterNotFound
+        fake_client.get_parameter.side_effect = [
+            {"Parameter": {"Value": "v", "Type": "String", "KeyId": None}},
+            FakeExceptions.ParameterNotFound(),
+        ]
+        fake_client.put_parameter.return_value = {"Version": 1}
+        result = m.copy_one_param(fake_client, "/acdl/dev/c/out", "/nova/dev/c/out")
+        assert result == "copied"
+        fake_client.put_parameter.assert_called_once()
+
+    def test_non_parameter_not_found_error_is_raised(self):
+        """A non-ParameterNotFound AWS error (e.g. ThrottlingException) on
+        the dest get_parameter is raised, not swallowed (P4, REQ-168)."""
+        from unittest import mock
+        import migrate_ssm_paths as m
+
+        class FakeExceptions:
+            ParameterNotFound = type("ParameterNotFound", (Exception,), {})
+
+        class ThrottlingException(Exception):
+            pass
+
+        fake_client = mock.Mock()
+        fake_client.exceptions = FakeExceptions
+        # source get_parameter succeeds; dest get_parameter raises Throttling
+        fake_client.get_parameter.side_effect = [
+            {"Parameter": {"Value": "v", "Type": "String", "KeyId": None}},
+            ThrottlingException("slow down"),
+        ]
+        with pytest.raises(ThrottlingException):
+            m.copy_one_param(fake_client, "/acdl/dev/c/out", "/nova/dev/c/out")
+        fake_client.put_parameter.assert_not_called()

tests/test_output_publisher.py

@@ -134,7 +134,11 @@ class TestPublishToSsm:
             def flaky_put(**kwargs):
                 call_count["n"] += 1
                 if "bad" in kwargs["Name"]:
-                    raise Exception("simulated failure")
+                    from botocore.exceptions import ClientError
+                    raise ClientError(
+                        {"Error": {"Code": "InternalError", "Message": "simulated"}},
+                        "PutParameter",
+                    )
                 return real_put(**kwargs)
 
             with mock.patch("core.output_publisher._ssm_client", return_value=ssm):
@@ -272,10 +276,11 @@ class TestPostGithubComment:
         assert "/issues/5/comments" in captured["url"]
 
     def test_returns_false_on_exception(self, monkeypatch):
+        import urllib.error
         monkeypatch.setenv("GITHUB_TOKEN", "tok")
         monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
         monkeypatch.setenv("GITHUB_REF", "refs/pull/1/merge")
-        with mock.patch("urllib.request.urlopen", side_effect=Exception("boom")):
+        with mock.patch("urllib.request.urlopen", side_effect=urllib.error.URLError("boom")):
             assert post_github_comment("body") is False
 
     def test_uses_gh_token_fallback(self, monkeypatch):