diff --git a/backend/packages/harness/deerflow/persistence/service_account/__init__.py b/backend/packages/harness/deerflow/persistence/service_account/__init__.py
index 3ff16eea..cad9febb 100644
--- a/backend/packages/harness/deerflow/persistence/service_account/__init__.py
+++ b/backend/packages/harness/deerflow/persistence/service_account/__init__.py
@@ -13,6 +13,6 @@ upgrade live in Stage 1 alongside the headless API surface.
 from __future__ import annotations
 
 from deerflow.persistence.service_account.model import ServiceAccountRow
-from deerflow.persistence.service_account.sql import ServiceAccountRepository
+from deerflow.persistence.service_account.sql import ServiceAccountRepository, ServiceAccountValidationError
 
-__all__ = ["ServiceAccountRepository", "ServiceAccountRow"]
+__all__ = ["ServiceAccountRepository", "ServiceAccountRow", "ServiceAccountValidationError"]
diff --git a/backend/packages/harness/deerflow/persistence/service_account/sql.py b/backend/packages/harness/deerflow/persistence/service_account/sql.py
index 3363bf1f..89dbf45b 100644
--- a/backend/packages/harness/deerflow/persistence/service_account/sql.py
+++ b/backend/packages/harness/deerflow/persistence/service_account/sql.py
@@ -20,6 +20,10 @@ from deerflow.persistence.service_account.model import ServiceAccountRow
 _VALID_STATUSES = frozenset({"active", "suspended", "deleted"})
 
 
+class ServiceAccountValidationError(ValueError):
+    """Raised when service account input fails application-layer validation."""
+
+
 class ServiceAccountRepository:
     def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
         self._sf = session_factory
@@ -48,6 +52,8 @@ class ServiceAccountRepository:
         identity_mode: str = "collapsed",
         status: str = "active",
     ) -> dict[str, Any]:
+        if status not in _VALID_STATUSES:
+            raise ServiceAccountValidationError(f"status {status!r} not in {_VALID_STATUSES!r}")
         now = datetime.now(UTC)
         row = ServiceAccountRow(
             id=str(uuid.uuid4()),
@@ -86,7 +92,7 @@ class ServiceAccountRepository:
 
     async def update_status(self, sa_id: str, status: str) -> None:
         if status not in _VALID_STATUSES:
-            raise ValueError(f"status {status!r} not in {_VALID_STATUSES!r}")
+            raise ServiceAccountValidationError(f"status {status!r} not in {_VALID_STATUSES!r}")
         async with self._sf() as session:
             await session.execute(update(ServiceAccountRow).where(ServiceAccountRow.id == sa_id).values(status=status, updated_at=datetime.now(UTC)))
             await session.commit()
diff --git a/backend/tests/test_service_account_repo.py b/backend/tests/test_service_account_repo.py
index 5ec8ce59..cb4d3174 100644
--- a/backend/tests/test_service_account_repo.py
+++ b/backend/tests/test_service_account_repo.py
@@ -95,3 +95,24 @@ async def test_list_by_workspace(tmp_path):
         assert {r["name"] for r in rows} == {"a", "b"}
     finally:
         await _cleanup()
+
+
+async def test_create_rejects_unknown_status(tmp_path):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_parents(repo)
+        with pytest.raises(ValueError):
+            await repo.create(workspace_id="w-1", name="x", created_by="u-alice", status="bogus")
+    finally:
+        await _cleanup()
+
+
+async def test_update_status_rejects_unknown_value(tmp_path):
+    repo = await _make_repo(tmp_path)
+    try:
+        await _seed_parents(repo)
+        sa = await repo.create(workspace_id="w-1", name="x", created_by="u-alice")
+        with pytest.raises(ValueError):
+            await repo.update_status(sa["id"], "not-a-status")
+    finally:
+        await _cleanup()