Merge Phase 2: Meta sync, launch control, adopt and reconciliation

Campaigns now reach Meta. Sync builds the campaign/ad set/creative/ad
chain with every object PAUSED; launch is the only action that starts
spend and is gated on an identified operator and a funding source.
Campaigns authored in Ads Manager can be discovered and adopted, and a
background loop keeps local status in step with Meta.

Review of the phase found three defects, fixed here: launch activated
only the campaign so nothing delivered, POST payloads went in the query
string so image uploads could not work, and an ad set with no targeting
reached Meta and failed opaquely.

README.md

@@ -39,6 +39,27 @@ This repository never connects directly to the Maskan CRM PostgreSQL database.
 Open API documentation at `http://127.0.0.1:8088/docs` when documentation is
 enabled.
 
+### Running the tests
+
+```powershell
+npm run test
+```
+
+`npm run test` loads `.env.testing`, which points at a **separate** database so
+a test run cannot touch development data. Create it once:
+
+```sql
+CREATE DATABASE campaign_test;
+```
+
+```powershell
+npm run test:migrate
+```
+
+Without it, the repository integration tests skip with "PostgreSQL is not
+reachable" and the suite still passes — so check the skip count, not just the
+exit code.
+
 ### Upgrading from JSON storage
 
 Chats and cron jobs were originally stored as JSON files in the working
@@ -65,25 +86,127 @@ Campaign records live in PostgreSQL. Apply the schema before first use:
 npm run local:migrate
 ```
 
-Set `META_ADS_ACCESS_TOKEN` in Settings > Environments, and enable the
-`meta_ads` MCP client if you also want campaign tools in chat.
+Set `META_ADS_ACCESS_TOKEN` and `META_PAGE_ID` in Settings > Environments, and
+enable the `meta_ads` MCP client if you also want campaign tools in chat.
+`META_PAGE_ID` is the Facebook Page ads are published from; sync refuses to
+start without it, rather than failing part-way through building the object
+chain. A campaign can override it with `advanced.page_id`.
 
-### What Phase 1 does and does not do
+### The campaign lifecycle
 
-Phase 1 owns campaign records, the approval workflow, budget guardrails and ad
-preview. **It creates nothing in Meta.** The only Meta calls are two reads:
+```
+draft -> pending_approval -> approved -> synced -> live
+                                            |       |
+                                            +-> paused <-+
+                                                  |
+                                                stopped
+```
 
-- `GET /act_<id>` for balance, spend and `min_daily_budget`
-- `GET /act_<id>/generatepreviews` to render ad previews
+`approved` is where a human signs off. `synced` means the objects exist in
+Meta but are **paused**. Only `live` spends money.
 
-There is deliberately no launch or sync endpoint. A campaign moves
-`draft -> pending_approval -> approved` and stops there; pushing objects to
-Meta arrives in Phase 2.
+### Ad previews create nothing
 
-Ad previews are rendered from a creative spec and create no Meta objects, so
-the wizard can show exactly how an ad will look across desktop feed, mobile
-feed, Instagram feed, Instagram story, Facebook story and right column before
-anything exists.
+Previews are rendered from a creative spec via `generatepreviews`, so the
+wizard shows exactly how an ad will look across desktop feed, mobile feed,
+Instagram feed, Instagram story, Facebook story and right column before any
+Meta object exists.
+
+### Sync creates paused objects only
+
+`POST /campaigns/{id}/sync` builds the Meta chain — campaign, ad set,
+creative, ad — and **every object is created `PAUSED`**. Nothing sync does can
+start spending. The guard is enforced in three places: the typed `create_*`
+methods, `MetaClient._post` itself (which refuses a non-`PAUSED` status on any
+create path), and the launch endpoint being the only caller allowed to send
+`ACTIVE`.
+
+Each id is persisted the moment it is obtained, so a sync that fails part-way
+is resumable: re-running it reuses the ids already stored and creates only
+what is missing. That is what stops a retry from leaving a second set of
+objects in a real ad account.
+
+### Launching requires an operator and a payment method
+
+`POST /campaigns/{id}/launch` sets the whole object chain active — ad, then
+ad set, then campaign. Meta only delivers when all three are active, so
+activating just the campaign would launch nothing. The campaign goes **last**
+on purpose: nothing under a paused campaign delivers, so a failure part-way
+leaves it unable to spend. That is also why pause and stop only have to flip
+the campaign.
+
+It is the only action that starts spend, and it is refused unless:
+
+- the campaign is `synced` or `paused`,
+- `approved_by` names a resolved operator — an approval nobody can be named
+  for does not authorise spend, and
+- the ad account has a `funding_source`.
+
+Set `MASKANX_OPERATOR_TOKENS` to a comma-separated list of `name:token`
+pairs. Approve, sync, launch, pause, stop and adopt read the token from the
+`X-MaskanX-Operator` header and record the resolved name as the actor; any
+actor in the request body is ignored. With no tokens configured the operator
+resolves to `unauthenticated`, which is enough to draft and approve but not
+to launch. The Campaigns page has a field for storing the token in the
+browser.
+
+### Importing campaigns made in Ads Manager
+
+`GET /campaigns/discover?ad_account_id=act_...` lists Meta campaigns MaskanX
+does not know about; `POST /campaigns/adopt` imports one. Adopting twice is
+safe — the second call returns the existing record.
+
+Imported campaigns are marked `origin=imported` and treated as
+Ads-Manager-owned: they are never synced, launching one touches only the
+campaign (its ad sets and ads keep the statuses set in Ads Manager), and they
+cannot be deleted from MaskanX.
+
+### Deleting
+
+`DELETE /campaigns/{id}` deletes the campaign on Meta first and only then
+removes the local row. If the Meta delete fails the local row is kept and the
+call answers `502`: a campaign forgotten here but still live on Meta would
+keep spending with nothing left to show it exists. Deleting a campaign
+cascades to its ad sets and ads; the creative is left behind, being an
+account-level asset other ads may reference.
+
+Deleting an **imported** campaign is refused with `409`. Deleting it on Meta
+would destroy work MaskanX did not author, and deleting only the local row
+would achieve nothing — the reconciler would import it again on the next
+cycle. Delete it in Ads Manager instead; the reconciler then archives the
+local record, keeping its spend history reportable.
+
+### Drift reconciliation
+
+A background loop reconciles each ad account with Meta: campaigns created in
+Ads Manager are imported, statuses are copied back so a campaign paused
+directly in Meta stops showing as `live` here, and campaigns that have
+disappeared are archived (never hard-deleted, so their spend history stays
+reportable).
+
+Meta is authoritative for delivery status; MaskanX is authoritative for its
+own metadata — guardrails, approvals and audit history are never overwritten.
+Campaigns that have not reached Meta yet (`draft`, `pending_approval`,
+`approved`) are left alone entirely.
+
+The interval defaults to 120 seconds and is set with
+`MASKANX_CAMPAIGN_RECONCILE_SECONDS`; `0` disables the loop.
+
+### Live smoke test
+
+The unit suite runs entirely against a fake transport: it proves the code
+sends what we think it sends, not that Meta accepts it. One opt-in test
+closes that gap by syncing a real campaign, reading back from Graph that all
+three objects are `PAUSED`, then deleting them:
+
+```powershell
+$env:MASKANX_LIVE_TESTS = "1"
+$env:MASKANX_LIVE_AD_ACCOUNT_ID = "act_<your account id>"
+npm run test -- tests/test_campaign_sync_live.py
+```
+
+It creates and deletes real objects. They are paused for their whole life, so
+it cannot spend money, and cleanup runs in a `finally` block.
 
 ### Budget minimums
 

src/adclaw/app/_app.py

@@ -259,6 +259,29 @@ async def lifespan(app: FastAPI):  # pylint: disable=too-many-statements
     watchdog_task = asyncio.create_task(watchdog.start())
     app.state.watchdog = watchdog
 
+    # --- Campaign reconciliation (Meta has no webhooks for ad objects) ---
+    reconcile_task = None
+    reconcile_account = (os.environ.get("META_ADS_ACCOUNT_ID") or "").strip()
+    if reconcile_account:
+        from ..campaigns.reconcile import reconcile_interval_seconds, reconcile_loop
+
+        if reconcile_interval_seconds() > 0:
+            from ..campaigns.repo import CampaignRepository
+            from ..meta.client import MetaClient, access_token_from_env
+
+            def _meta_client():
+                return MetaClient(access_token=access_token_from_env())
+
+            reconcile_task = asyncio.create_task(
+                reconcile_loop(CampaignRepository, _meta_client, reconcile_account),
+                name="campaign_reconcile",
+            )
+            app.state.reconcile_task = reconcile_task
+    else:
+        logger.debug(
+            "META_ADS_ACCOUNT_ID not set; campaign reconciliation not started",
+        )
+
     try:
         if mcp_initial_config is not None:
             mcp_init_task = _schedule_mcp_initialization(
@@ -270,6 +293,8 @@ async def lifespan(app: FastAPI):  # pylint: disable=too-many-statements
     finally:
         if hasattr(app.state, "watchdog"):
             app.state.watchdog.stop()
+        if reconcile_task is not None:
+            reconcile_task.cancel()
         # stop order: watchers -> cron -> channels -> mcp -> runner
         try:
             await config_watcher.stop()

src/adclaw/app/routers/_operator.py

@@ -0,0 +1,76 @@
+# -*- coding: utf-8 -*-
+"""Operator identity for state changes that authorise spend.
+
+MaskanX has no login system. Approving a campaign and launching it are the
+two actions that lead to money being spent, so they must be attributable to
+a named operator rather than to a string the client chose.
+
+Configure `MASKANX_OPERATOR_TOKENS` as comma-separated `name:token` pairs.
+When it is unset the dependency returns "unauthenticated" so local
+development is not blocked, and logs a warning.
+"""
+from __future__ import annotations
+
+import logging
+import os
+
+from fastapi import Header, HTTPException, status
+
+logger = logging.getLogger(__name__)
+
+OPERATOR_TOKENS_ENV = "MASKANX_OPERATOR_TOKENS"
+UNAUTHENTICATED = "unauthenticated"
+
+_warned = False
+
+
+class OperatorAuthError(Exception):
+    """Raised when an operator token is missing or unknown."""
+
+
+def _token_map() -> dict[str, str]:
+    raw = (os.environ.get(OPERATOR_TOKENS_ENV) or "").strip()
+    if not raw:
+        return {}
+    mapping: dict[str, str] = {}
+    for entry in raw.split(","):
+        name, _, token = entry.partition(":")
+        name = name.strip()
+        token = token.strip()
+        if name and token:
+            mapping[token] = name
+    return mapping
+
+
+def resolve_operator(token: str | None) -> str:
+    """Return the operator name for a token, or raise OperatorAuthError."""
+    global _warned
+    tokens = _token_map()
+    if not tokens:
+        if not _warned:
+            logger.warning(
+                "%s is not set. Campaign approval and launch are "
+                "unauthenticated; set it before running live campaigns.",
+                OPERATOR_TOKENS_ENV,
+            )
+            _warned = True
+        return UNAUTHENTICATED
+    if not token or token not in tokens:
+        raise OperatorAuthError("Unknown or missing operator token.")
+    return tokens[token]
+
+
+def require_operator(
+    x_maskanx_operator: str | None = Header(default=None),
+) -> str:
+    """FastAPI dependency resolving the calling operator."""
+    try:
+        return resolve_operator(x_maskanx_operator)
+    except OperatorAuthError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail=(
+                "A valid X-MaskanX-Operator header is required to approve or "
+                "launch a campaign."
+            ),
+        ) from exc

src/adclaw/app/routers/campaigns.py

@@ -10,7 +10,7 @@ import logging
 import uuid
 from typing import Any
 
-from fastapi import APIRouter, HTTPException, status as http_status
+from fastapi import APIRouter, Depends, HTTPException, status as http_status
 from pydantic import BaseModel, Field
 
 from ...campaigns.models import CampaignSpec
@@ -20,8 +20,16 @@ from ...campaigns.state import (
     TransitionError,
     next_status,
 )
+from ...campaigns.adopt import adopt_campaign, discover_campaigns
+from ...campaigns.sync import (
+    SyncConfigurationError,
+    activation_order,
+    sync_campaign,
+    unsync_campaign,
+)
 from ...campaigns.validation import validate_campaign
 from ...meta.client import MetaClient, MetaError, access_token_from_env
+from ._operator import UNAUTHENTICATED, require_operator
 
 logger = logging.getLogger(__name__)
 
@@ -56,6 +64,12 @@ class ActorPayload(BaseModel):
     reason: str | None = None
 
 
+class AdoptRequest(BaseModel):
+    ad_account_id: str
+    meta_campaign_id: str
+    company_id: str | None = None
+
+
 class PreviewRequest(BaseModel):
     ad_account_id: str
     creative: dict[str, Any]
@@ -131,18 +145,17 @@ async def _transition(
     if action == "approve":
         campaign.approved_by = payload.actor
     try:
-        saved = await repo.update_campaign(campaign)
+        saved = await repo.update_campaign_with_event(
+            campaign,
+            event_type=f"campaign.{action}",
+            actor=payload.actor,
+            reason=payload.reason,
+        )
     except LookupError as exc:
         raise HTTPException(
             status_code=http_status.HTTP_404_NOT_FOUND,
             detail="Campaign not found.",
         ) from exc
-    await repo.add_event(
-        campaign_id,
-        event_type=f"campaign.{action}",
-        actor=payload.actor,
-        reason=payload.reason,
-    )
     return saved
 
 
@@ -186,6 +199,48 @@ async def preview_campaign(payload: PreviewRequest) -> PreviewResponse:
     return PreviewResponse(previews=previews)
 
 
+@router.get("/discover")
+async def discover_meta_campaigns(ad_account_id: str) -> list[dict[str, Any]]:
+    """List campaigns in the ad account that MaskanX does not know about."""
+    try:
+        return await discover_campaigns(
+            get_repository(), get_meta_client(), ad_account_id,
+        )
+    except MetaError as exc:
+        raise _meta_http_error(exc) from exc
+
+
+@router.post(
+    "/adopt",
+    response_model=CampaignSpec,
+    status_code=http_status.HTTP_201_CREATED,
+)
+async def adopt_meta_campaign(
+    payload: AdoptRequest,
+    operator: str = Depends(require_operator),
+) -> CampaignSpec:
+    """Import an existing Meta campaign as a MaskanX record.
+
+    Idempotent: adopting the same campaign twice returns the existing record.
+    """
+    try:
+        return await adopt_campaign(
+            get_repository(),
+            get_meta_client(),
+            ad_account_id=payload.ad_account_id,
+            meta_campaign_id=payload.meta_campaign_id,
+            actor=operator,
+            company_id=payload.company_id,
+        )
+    except LookupError as exc:
+        raise HTTPException(
+            status_code=http_status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except MetaError as exc:
+        raise _meta_http_error(exc) from exc
+
+
 @router.get("/account/{ad_account_id}")
 async def get_ad_account(ad_account_id: str) -> dict[str, Any]:
     """Return billing and limit fields for an ad account."""
@@ -237,7 +292,40 @@ async def update_campaign(
 
 @router.delete("/{campaign_id}", status_code=http_status.HTTP_204_NO_CONTENT)
 async def delete_campaign(campaign_id: str) -> None:
-    await _load_or_404(campaign_id)
+    """Delete the campaign here and, if MaskanX created it, on Meta too.
+
+    Meta is deleted first. If that fails the local row is kept and the
+    caller gets a 502: a campaign forgotten here but left live on Meta
+    would keep spending with nothing in MaskanX recording that it exists.
+
+    An imported campaign cannot be deleted here at all. MaskanX did not
+    author it, so deleting it on Meta would destroy work done in Ads
+    Manager; and deleting only the local row would achieve nothing, because
+    the reconciler would import it again on the next cycle. Deleting it in
+    Ads Manager is the one action that sticks — the reconciler then archives
+    the local record, keeping its spend history reportable.
+    """
+    campaign = await _load_or_404(campaign_id)
+    if campaign.origin == "imported":
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail=(
+                "This campaign was created in Ads Manager, so it has to be "
+                "deleted there. MaskanX will archive its record automatically "
+                "once it is gone, keeping the spend history."
+            ),
+        )
+    if campaign.meta_campaign_id:
+        try:
+            await unsync_campaign(get_meta_client(), campaign)
+        except MetaError as exc:
+            raise HTTPException(
+                status_code=http_status.HTTP_502_BAD_GATEWAY,
+                detail=(
+                    f"Could not delete this campaign on Meta, so it was kept "
+                    f"here as well: {exc}. Delete it in Ads Manager, or retry."
+                ),
+            ) from exc
     await get_repository().delete_campaign(campaign_id)
 
 
@@ -253,8 +341,14 @@ async def submit_campaign(
 async def approve_campaign(
     campaign_id: str,
     payload: ActorPayload | None = None,
+    operator: str = Depends(require_operator),
 ) -> CampaignSpec:
-    return await _transition(campaign_id, "approve", payload or ActorPayload())
+    body = payload or ActorPayload()
+    return await _transition(
+        campaign_id,
+        "approve",
+        ActorPayload(actor=operator, reason=body.reason),
+    )
 
 
 @router.post("/{campaign_id}/reject", response_model=CampaignSpec)
@@ -263,3 +357,158 @@ async def reject_campaign(
     payload: ActorPayload | None = None,
 ) -> CampaignSpec:
     return await _transition(campaign_id, "reject", payload or ActorPayload())
+
+
+async def _assert_launchable(campaign: CampaignSpec) -> None:
+    """Refuse to launch unless every precondition for spending is met.
+
+    These are the last checks before a campaign can consume real budget, so
+    each failure names what to do rather than letting Meta reject it later.
+    """
+    if not campaign.meta_campaign_id:
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail="This campaign has not been synced to Meta yet.",
+        )
+
+    # An approval that cannot be attributed to a person is not an approval.
+    if not campaign.approved_by or campaign.approved_by == UNAUTHENTICATED:
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail=(
+                "This campaign was approved without an identified operator, "
+                "so it cannot be launched. Set MASKANX_OPERATOR_TOKENS, then "
+                "have an operator approve it again."
+            ),
+        )
+
+    try:
+        account = await get_meta_client().get_ad_account(campaign.ad_account_id)
+    except MetaError as exc:
+        raise _meta_http_error(exc) from exc
+
+    if not account.get("funding_source"):
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail=(
+                "This ad account has no payment method, so Meta cannot "
+                "deliver the campaign. Add one in Meta Business Manager; it "
+                "cannot be set from MaskanX."
+            ),
+        )
+
+
+@router.post("/{campaign_id}/launch", response_model=CampaignSpec)
+async def launch_campaign(
+    campaign_id: str,
+    operator: str = Depends(require_operator),
+) -> CampaignSpec:
+    """Set the campaign live on Meta. This starts spending."""
+    campaign = await _load_or_404(campaign_id)
+    if campaign.status not in ("synced", "paused"):
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail=(
+                f"A campaign in status '{campaign.status}' cannot be "
+                "launched. Sync it to Meta first."
+            ),
+        )
+    await _assert_launchable(campaign)
+
+    # The whole chain has to be active for Meta to deliver, and the campaign
+    # goes last so a partial failure leaves it paused and unable to spend.
+    # See `activation_order`.
+    meta = get_meta_client()
+    try:
+        for object_id in activation_order(campaign):
+            await meta.update_object_status(object_id, "ACTIVE")
+    except MetaError as exc:
+        raise _meta_http_error(exc) from exc
+
+    return await _transition(campaign_id, "launch", ActorPayload(actor=operator))
+
+
+@router.post("/{campaign_id}/pause", response_model=CampaignSpec)
+async def pause_campaign(
+    campaign_id: str,
+    operator: str = Depends(require_operator),
+) -> CampaignSpec:
+    """Pause a live campaign on Meta, halting spend."""
+    campaign = await _load_or_404(campaign_id)
+    if campaign.status != "live":
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail=f"Only a live campaign can be paused; this one is "
+            f"'{campaign.status}'.",
+        )
+    if campaign.meta_campaign_id:
+        try:
+            await get_meta_client().update_object_status(
+                campaign.meta_campaign_id, "PAUSED",
+            )
+        except MetaError as exc:
+            raise _meta_http_error(exc) from exc
+
+    return await _transition(campaign_id, "pause", ActorPayload(actor=operator))
+
+
+@router.post("/{campaign_id}/stop", response_model=CampaignSpec)
+async def stop_campaign(
+    campaign_id: str,
+    operator: str = Depends(require_operator),
+) -> CampaignSpec:
+    """Stop a campaign permanently, pausing it on Meta first."""
+    campaign = await _load_or_404(campaign_id)
+    if campaign.status not in ("approved", "synced", "live", "paused"):
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail=f"A campaign in status '{campaign.status}' cannot be "
+            f"stopped.",
+        )
+    # Pause on Meta before recording the stop, so a Meta failure cannot leave
+    # a campaign that is stopped locally but still delivering.
+    if campaign.meta_campaign_id:
+        try:
+            await get_meta_client().update_object_status(
+                campaign.meta_campaign_id, "PAUSED",
+            )
+        except MetaError as exc:
+            raise _meta_http_error(exc) from exc
+
+    return await _transition(campaign_id, "stop", ActorPayload(actor=operator))
+
+
+@router.post("/{campaign_id}/sync", response_model=CampaignSpec)
+async def sync_campaign_to_meta(
+    campaign_id: str,
+    operator: str = Depends(require_operator),
+) -> CampaignSpec:
+    """Create the Meta object chain for an approved campaign.
+
+    Every object is created PAUSED. This does not start delivery.
+    """
+    campaign = await _load_or_404(campaign_id)
+    if campaign.status != "approved":
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT,
+            detail=(
+                f"Only an approved campaign can be synced; this one is "
+                f"'{campaign.status}'."
+            ),
+        )
+    try:
+        return await sync_campaign(
+            get_repository(), get_meta_client(), campaign, actor=operator,
+        )
+    except SyncConfigurationError as exc:
+        raise HTTPException(
+            status_code=http_status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=str(exc),
+        ) from exc
+    except MetaError as exc:
+        raise _meta_http_error(exc) from exc
+    except LookupError as exc:
+        raise HTTPException(
+            status_code=http_status.HTTP_404_NOT_FOUND,
+            detail="Campaign not found.",
+        ) from exc

src/adclaw/app/routers/envs.py

@@ -228,6 +228,7 @@ _KEY_REGISTRY: List[Dict[str, str]] = [
     {"key": "META_ADS_ACCESS_TOKEN", "plugin": "Meta Ads", "description": "Facebook & Instagram ads access token"},
     {"key": "META_APP_ID", "plugin": "Meta Ads", "description": "Meta app ID, enables automatic token refresh"},
     {"key": "META_APP_SECRET", "plugin": "Meta Ads", "description": "Meta app secret, enables automatic token refresh"},
+    {"key": "MASKANX_OPERATOR_TOKENS", "plugin": "Campaigns", "description": "Comma-separated name:token pairs authorised to approve and launch campaigns"},
     # Analytics
     # (Google Analytics uses OAuth, no static key needed)
     # Social Media

src/adclaw/campaigns/adopt.py

@@ -0,0 +1,112 @@
+# -*- coding: utf-8 -*-
+"""Discover campaigns created directly in Meta Ads Manager, and adopt them.
+
+Campaigns can be created on either side. Anything MaskanX did not create is
+`origin="imported"`: MaskanX reports on it and can pause or stop it, but it
+was authored elsewhere.
+
+Adoption is keyed on `meta_campaign_id`, which carries a unique index, so
+adopting the same campaign twice returns the existing record rather than
+creating a duplicate.
+"""
+from __future__ import annotations
+
+import logging
+import uuid
+from typing import Any
+
+from .models import CampaignSpec
+
+logger = logging.getLogger(__name__)
+
+# Meta delivery status -> MaskanX status. An imported campaign has already
+# been through whatever approval its author used, so it lands in a delivery
+# state rather than back in `draft`.
+_STATUS_MAP = {
+    "ACTIVE": "live",
+    "PAUSED": "paused",
+    "DELETED": "archived",
+    "ARCHIVED": "archived",
+}
+
+
+def local_status_for(meta_status: str | None) -> str:
+    """Map a Meta campaign status onto a MaskanX status."""
+    return _STATUS_MAP.get((meta_status or "").upper(), "paused")
+
+
+def spec_from_meta(
+    meta_campaign: dict[str, Any],
+    ad_account_id: str,
+    company_id: str | None = None,
+) -> CampaignSpec:
+    """Build a local record describing a campaign that already exists in Meta."""
+    return CampaignSpec(
+        id=f"camp_{uuid.uuid4().hex[:16]}",
+        company_id=company_id,
+        name=meta_campaign.get("name") or "Imported campaign",
+        status=local_status_for(meta_campaign.get("status")),
+        origin="imported",
+        objective=meta_campaign.get("objective"),
+        ad_account_id=ad_account_id,
+        meta_campaign_id=meta_campaign["id"],
+        sync_status="synced",
+        advanced={
+            "imported_from_meta": {
+                "effective_status": meta_campaign.get("effective_status"),
+                "created_time": meta_campaign.get("created_time"),
+            },
+        },
+    )
+
+
+async def _known_meta_ids(repo) -> dict[str, CampaignSpec]:
+    campaigns = await repo.list_campaigns()
+    return {c.meta_campaign_id: c for c in campaigns if c.meta_campaign_id}
+
+
+async def discover_campaigns(repo, meta, ad_account_id: str) -> list[dict[str, Any]]:
+    """Return Meta campaigns in this account that MaskanX does not know about."""
+    known = await _known_meta_ids(repo)
+    remote = await meta.list_campaigns(ad_account_id)
+    return [c for c in remote if c.get("id") and c["id"] not in known]
+
+
+async def adopt_campaign(
+    repo,
+    meta,
+    ad_account_id: str,
+    meta_campaign_id: str,
+    actor: str,
+    company_id: str | None = None,
+) -> CampaignSpec:
+    """Create a local record for an existing Meta campaign.
+
+    Idempotent: adopting an already-adopted campaign returns the existing
+    record untouched rather than creating a second one.
+    """
+    known = await _known_meta_ids(repo)
+    existing = known.get(meta_campaign_id)
+    if existing is not None:
+        logger.info(
+            "Campaign %s is already adopted as %s", meta_campaign_id, existing.id,
+        )
+        return existing
+
+    remote = await meta.list_campaigns(ad_account_id)
+    match = next((c for c in remote if c.get("id") == meta_campaign_id), None)
+    if match is None:
+        raise LookupError(
+            f"Campaign {meta_campaign_id} was not found in {ad_account_id}.",
+        )
+
+    spec = spec_from_meta(match, ad_account_id, company_id=company_id)
+    created = await repo.create_campaign(spec)
+    await repo.add_event(
+        created.id,
+        event_type="campaign.adopted",
+        actor=actor,
+        reason=f"Imported from Meta campaign {meta_campaign_id}",
+        payload={"meta_campaign_id": meta_campaign_id},
+    )
+    return created

src/adclaw/campaigns/reconcile.py

@@ -0,0 +1,159 @@
+# -*- coding: utf-8 -*-
+"""Keep local campaign records in step with Meta.
+
+Meta publishes no webhooks for campaign create or delete, so this polls.
+
+Conflict rule, from the design spec:
+  Meta is authoritative for delivery state (status).
+  MaskanX is authoritative for its own metadata (guardrails, approvals,
+  audit history), which this loop never touches.
+
+A campaign that disappears from Meta is archived, never hard-deleted: its
+spend history has to stay reportable.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+from .adopt import local_status_for, spec_from_meta
+
+logger = logging.getLogger(__name__)
+
+INTERVAL_ENV = "MASKANX_CAMPAIGN_RECONCILE_SECONDS"
+DEFAULT_INTERVAL_SECONDS = 120.0
+
+# Statuses MaskanX owns outright. A local campaign that has not reached Meta
+# yet must not be touched by reconciliation.
+_LOCAL_ONLY_STATUSES = frozenset(
+    {"draft", "pending_approval", "approved", "archived"},
+)
+
+
+@dataclass
+class ReconcileResult:
+    imported: list[str] = field(default_factory=list)
+    status_changed: list[str] = field(default_factory=list)
+    archived: list[str] = field(default_factory=list)
+
+    @property
+    def changed(self) -> bool:
+        return bool(self.imported or self.status_changed or self.archived)
+
+
+def reconcile_interval_seconds() -> float:
+    """Return the poll interval; 0 or less disables the loop."""
+    raw = os.environ.get(INTERVAL_ENV)
+    if raw is None:
+        return DEFAULT_INTERVAL_SECONDS
+    try:
+        return max(0.0, float(raw))
+    except ValueError:
+        logger.warning(
+            "%s is not a number (%r); using the default of %.0fs",
+            INTERVAL_ENV,
+            raw,
+            DEFAULT_INTERVAL_SECONDS,
+        )
+        return DEFAULT_INTERVAL_SECONDS
+
+
+async def reconcile_once(repo, meta, ad_account_id: str) -> ReconcileResult:
+    """Bring local records in step with one ad account. Returns what changed."""
+    result = ReconcileResult()
+
+    remote = await meta.list_campaigns(ad_account_id)
+    remote_by_id = {c["id"]: c for c in remote if c.get("id")}
+
+    local = await repo.list_campaigns()
+    local_by_meta_id = {c.meta_campaign_id: c for c in local if c.meta_campaign_id}
+
+    # Campaigns that exist in Meta but not here.
+    for meta_id, remote_campaign in remote_by_id.items():
+        if meta_id in local_by_meta_id:
+            continue
+        spec = spec_from_meta(remote_campaign, ad_account_id)
+        created = await repo.create_campaign(spec)
+        await repo.add_event(
+            created.id,
+            event_type="campaign.reconciled_import",
+            reason=f"Discovered in Meta as {meta_id}",
+        )
+        result.imported.append(created.id)
+
+    for meta_id, campaign in local_by_meta_id.items():
+        if campaign.ad_account_id != ad_account_id:
+            continue
+
+        remote_campaign = remote_by_id.get(meta_id)
+
+        if remote_campaign is None:
+            # Gone from Meta. Archive rather than delete: the insight history
+            # attached to this campaign must stay reportable.
+            if campaign.status == "archived":
+                continue
+            campaign.status = "archived"
+            campaign.advanced["deleted_in_meta_at"] = datetime.now(
+                timezone.utc,
+            ).isoformat()
+            await repo.update_campaign_with_event(
+                campaign,
+                event_type="campaign.deleted_in_meta",
+                reason="No longer present in the Meta ad account",
+            )
+            result.archived.append(campaign.id)
+            continue
+
+        # Meta owns delivery state. Local-only statuses are left alone: a
+        # draft has never been near Meta, and an archived campaign is done.
+        if campaign.status in _LOCAL_ONLY_STATUSES:
+            continue
+
+        remote_status = local_status_for(remote_campaign.get("status"))
+        if remote_status != campaign.status:
+            previous = campaign.status
+            campaign.status = remote_status
+            await repo.update_campaign_with_event(
+                campaign,
+                event_type="campaign.status_reconciled",
+                reason=f"Meta reports '{remote_campaign.get('status')}' "
+                f"(was '{previous}')",
+            )
+            result.status_changed.append(campaign.id)
+
+    return result
+
+
+async def reconcile_loop(repo_factory, meta_factory, ad_account_id: str) -> None:
+    """Poll Meta forever. Never raises: a bad cycle is logged and retried."""
+    interval = reconcile_interval_seconds()
+    if interval <= 0:
+        logger.info("Campaign reconciliation disabled (%s=0)", INTERVAL_ENV)
+        return
+
+    logger.info(
+        "Campaign reconciliation started for %s every %.0fs",
+        ad_account_id,
+        interval,
+    )
+    while True:
+        try:
+            await asyncio.sleep(interval)
+            result = await reconcile_once(repo_factory(), meta_factory(), ad_account_id)
+            if result.changed:
+                logger.info(
+                    "Reconciled %s: %d imported, %d status changes, %d archived",
+                    ad_account_id,
+                    len(result.imported),
+                    len(result.status_changed),
+                    len(result.archived),
+                )
+        except asyncio.CancelledError:
+            logger.info("Campaign reconciliation stopped")
+            raise
+        except Exception:
+            # A failed cycle must not kill the loop; the next one retries.
+            logger.exception("Campaign reconciliation cycle failed")

src/adclaw/campaigns/repo.py

@@ -183,6 +183,78 @@ WHERE id = %s
             await conn.close()
         return campaign_from_row(row)
 
+    async def update_campaign_with_event(
+        self,
+        spec: CampaignSpec,
+        event_type: str,
+        actor: str | None = None,
+        reason: str | None = None,
+        payload: dict[str, Any] | None = None,
+    ) -> CampaignSpec:
+        """Update a campaign and record its audit event atomically.
+
+        A status change without its audit row is a compliance hole once
+        launches authorise spend, so both statements share one transaction.
+        """
+        conn = await connect_database()
+        try:
+            async with conn.cursor() as cur:
+                await cur.execute(
+                    f"""
+UPDATE maskanx_campaigns SET
+    company_id = %s, name = %s, status = %s, objective = %s,
+    ad_account_id = %s, budget = %s, guardrails = %s, targeting = %s,
+    advanced = %s, channels = %s, schedule = %s, meta_campaign_id = %s,
+    sync_status = %s, sync_error = %s, approved_by = %s, updated_at = NOW()
+WHERE id = %s
+{_RETURNING_CLAUSE}
+""",
+                    (
+                        spec.company_id,
+                        spec.name,
+                        spec.status,
+                        spec.objective,
+                        spec.ad_account_id,
+                        jsonb(spec.budget),
+                        jsonb(spec.guardrails),
+                        jsonb(spec.targeting),
+                        jsonb(spec.advanced),
+                        jsonb(spec.channels),
+                        jsonb(spec.schedule),
+                        spec.meta_campaign_id,
+                        spec.sync_status,
+                        spec.sync_error,
+                        spec.approved_by,
+                        spec.id,
+                    ),
+                )
+                row = await cur.fetchone()
+                if row is None:
+                    raise LookupError(f"Campaign {spec.id} does not exist")
+
+                await cur.execute(
+                    """
+INSERT INTO maskanx_campaign_events (
+    id, campaign_id, event_type, actor, reason, payload
+) VALUES (%s, %s, %s, %s, %s, %s)
+""",
+                    (
+                        str(uuid.uuid4()),
+                        spec.id,
+                        event_type,
+                        actor,
+                        reason,
+                        jsonb(payload or {}),
+                    ),
+                )
+            await conn.commit()
+        except Exception:
+            await conn.rollback()
+            raise
+        finally:
+            await conn.close()
+        return campaign_from_row(row)
+
     async def delete_campaign(self, campaign_id: str) -> bool:
         conn = await connect_database()
         try:

src/adclaw/campaigns/sync.py

@@ -0,0 +1,189 @@
+# -*- coding: utf-8 -*-
+"""Push an approved campaign into Meta as campaign -> ad set -> creative -> ad.
+
+Every Meta object is created PAUSED; nothing here can start spending. Launch
+is a separate, explicit action.
+
+Each id is persisted as soon as it is obtained, so a failure part-way through
+the chain is resumable: re-running sync reuses the ids already stored and
+creates only what is still missing. That is what stops a retry from creating
+duplicate ad objects in a real advertising account.
+"""
+from __future__ import annotations
+
+import logging
+import os
+from datetime import datetime, timezone
+from typing import Any
+
+from .models import CampaignSpec
+from .state import next_status
+from ..meta.client import MetaError
+from ..meta.objects import (
+    build_ad_set_payload,
+    build_campaign_payload,
+    build_creative_payload,
+)
+
+logger = logging.getLogger(__name__)
+
+PAGE_ID_ENV = "META_PAGE_ID"
+SYNC_KEY = "meta_sync"
+
+
+class SyncConfigurationError(Exception):
+    """Raised before any Meta call when required configuration is missing."""
+
+
+def _sync_state(campaign: CampaignSpec) -> dict[str, Any]:
+    state = campaign.advanced.get(SYNC_KEY)
+    return dict(state) if isinstance(state, dict) else {}
+
+
+def _resolve_page_id(campaign: CampaignSpec) -> str:
+    page_id = (campaign.advanced.get("page_id") or "").strip()
+    if not page_id:
+        page_id = (os.environ.get(PAGE_ID_ENV) or "").strip()
+    if not page_id:
+        raise SyncConfigurationError(
+            "No Facebook Page is configured for this campaign. Set "
+            "`page_id` on the campaign or the META_PAGE_ID environment "
+            "variable before syncing.",
+        )
+    return page_id
+
+
+def _meta_error_text(exc: MetaError) -> str:
+    parts = [exc.message]
+    if exc.code is not None:
+        parts.append(f"code={exc.code}")
+    if exc.subcode is not None:
+        parts.append(f"subcode={exc.subcode}")
+    return " ".join(parts)
+
+
+async def sync_campaign(repo, meta, campaign: CampaignSpec, actor: str) -> CampaignSpec:
+    """Create the Meta object chain for an approved campaign.
+
+    Returns the persisted campaign. Raises MetaError with the failure already
+    recorded, or SyncConfigurationError before any Meta call.
+    """
+    if campaign.status != "approved":
+        raise ValueError(
+            f"Only an approved campaign can be synced; this one is "
+            f"'{campaign.status}'.",
+        )
+    if not campaign.ad_account_id:
+        raise SyncConfigurationError(
+            "This campaign has no ad account, so there is nothing to sync to.",
+        )
+
+    # Resolved before any network call so a missing Page cannot leave a
+    # half-built chain in the ad account.
+    page_id = _resolve_page_id(campaign)
+    account = campaign.ad_account_id
+    state = _sync_state(campaign)
+
+    async def _persist(event_type: str, reason: str | None = None) -> CampaignSpec:
+        campaign.advanced[SYNC_KEY] = state
+        return await repo.update_campaign_with_event(
+            campaign,
+            event_type=event_type,
+            actor=actor,
+            reason=reason,
+        )
+
+    campaign.sync_status = "syncing"
+    campaign.sync_error = None
+    await _persist("campaign.sync_started")
+
+    try:
+        if not campaign.meta_campaign_id:
+            campaign.meta_campaign_id = await meta.create_campaign(
+                account, **build_campaign_payload(campaign),
+            )
+            await _persist("campaign.sync_progress", reason="campaign created")
+
+        if not state.get("adset_id"):
+            state["adset_id"] = await meta.create_ad_set(
+                account,
+                **build_ad_set_payload(campaign, campaign.meta_campaign_id),
+            )
+            await _persist("campaign.sync_progress", reason="ad set created")
+
+        if not state.get("creative_id"):
+            image_hash = state.get("image_hash")
+            image_path = campaign.advanced.get("image_path")
+            if not image_hash and image_path:
+                image_hash = await meta.upload_ad_image(account, image_path)
+                state["image_hash"] = image_hash
+                await _persist("campaign.sync_progress", reason="image uploaded")
+            state["creative_id"] = await meta.create_ad_creative(
+                account,
+                **build_creative_payload(campaign, page_id, image_hash or ""),
+            )
+            await _persist("campaign.sync_progress", reason="creative created")
+
+        if not state.get("ad_id"):
+            state["ad_id"] = await meta.create_ad(
+                account,
+                name=f"{campaign.name} - Ad",
+                adset_id=state["adset_id"],
+                creative_id=state["creative_id"],
+            )
+            await _persist("campaign.sync_progress", reason="ad created")
+    except MetaError as exc:
+        # Keep every id obtained so far: the retry reuses them instead of
+        # creating a second set of objects in the account.
+        campaign.sync_status = "failed"
+        campaign.sync_error = _meta_error_text(exc)
+        await _persist("campaign.sync_failed", reason=campaign.sync_error)
+        logger.warning(
+            "Campaign %s failed to sync to Meta: %s", campaign.id, campaign.sync_error,
+        )
+        raise
+
+    campaign.status = next_status(campaign.status, "sync")
+    campaign.sync_status = "synced"
+    campaign.sync_error = None
+    campaign.advanced["last_synced_at"] = datetime.now(timezone.utc).isoformat()
+    return await _persist("campaign.sync")
+
+
+def activation_order(campaign: CampaignSpec) -> list[str]:
+    """Meta ids to set ACTIVE when launching, children before the campaign.
+
+    Sync creates every object PAUSED, and Meta only delivers an ad when the
+    ad, its ad set AND its campaign are all active. Setting just the
+    campaign active therefore launches nothing — the campaign would read
+    `live` here while delivering nothing there.
+
+    The campaign is deliberately last. Nothing under a paused campaign
+    delivers, so if activation fails part-way the campaign is still paused
+    and no money can be spent. It is the master switch, which is also why
+    pause and stop only have to flip the campaign.
+
+    An imported campaign has no stored ad set or ad ids, so only its
+    campaign is returned: its children keep whatever status the operator
+    gave them in Ads Manager.
+    """
+    state = _sync_state(campaign)
+    ids = [state.get("ad_id"), state.get("adset_id"), campaign.meta_campaign_id]
+    return [object_id for object_id in ids if object_id]
+
+
+async def unsync_campaign(meta, campaign: CampaignSpec) -> bool:
+    """Delete this campaign's objects on Meta. Returns whether it deleted.
+
+    Deleting a campaign cascades to its ad sets and ads, so the campaign id
+    is all that is needed. The creative is deliberately left behind: it is
+    an account-level asset that other ads may reference.
+
+    An imported campaign is never deleted on Meta. MaskanX did not author
+    it, and forgetting the local record must not destroy work the operator
+    did in Ads Manager.
+    """
+    if campaign.origin == "imported" or not campaign.meta_campaign_id:
+        return False
+    await meta.delete_object(campaign.meta_campaign_id)
+    return True

src/adclaw/campaigns/validation.py

@@ -78,11 +78,26 @@ def validate_budget(
                 ),
             )
 
-    if lifetime is not None and lifetime <= 0:
+    if lifetime is not None:
+        if lifetime <= 0:
+            issues.append(
+                ValidationIssue(
+                    field="budget.lifetime_budget",
+                    message="Lifetime budget must be greater than zero.",
+                ),
+            )
+        # Reached only when daily is None (the daily-and-lifetime-together
+        # case already returned above), i.e. this is a lifetime-only
+        # budget. create_ad_set only accepts daily_budget, so without this
+        # a lifetime-only campaign would pass validation and approval and
+        # only fail once Task 5's sync tries to create the Meta ad set.
         issues.append(
             ValidationIssue(
                 field="budget.lifetime_budget",
-                message="Lifetime budget must be greater than zero.",
+                message=(
+                    "Lifetime budgets are not supported yet. Set a daily "
+                    "budget instead."
+                ),
             ),
         )
 

src/adclaw/meta/client.py

@@ -1,11 +1,18 @@
 # -*- coding: utf-8 -*-
-"""Read-only Meta Graph API client.
+"""Meta Graph API client.
 
-Phase 1 only reads: ad account fields and ad previews. Previews are generated
-from a creative spec and create no objects in Meta.
+Phase 1 added reads: ad account fields and ad previews. Phase 2 adds writes:
+creating campaigns, ad sets, creatives and ads, and updating object status.
+
+Money-safety contract for the write half: every `create_*` helper is hard
+-coded to create its object with status="PAUSED" and raises `ValueError`
+(with no network call at all) if the caller asks for anything else.
+`update_object_status` is the only method allowed to send "ACTIVE"; it is
+what the Launch endpoint (Task 6) calls once a human has approved spend.
 """
 from __future__ import annotations
 
+import base64
 import json
 import os
 from typing import Any, Awaitable, Callable
@@ -39,6 +46,17 @@ DEFAULT_AD_FORMATS = (
     "RIGHT_COLUMN_STANDARD",
 )
 
+# The only status any create_* helper is allowed to send. Launching (setting
+# ACTIVE) happens exclusively through update_object_status.
+CREATE_STATUS = "PAUSED"
+
+LIST_FIELDS = "id,name,status,effective_status,created_time"
+
+# Path suffixes that create a campaign/ad set/ad. _post refuses any of these
+# with a non-PAUSED status as defence in depth, in case a future caller
+# bypasses the typed create_* guards and calls _post directly.
+_PAUSED_ONLY_SUFFIXES = ("/campaigns", "/adsets", "/ads")
+
 Transport = Callable[[str, str, dict[str, Any]], Awaitable[dict[str, Any]]]
 
 
@@ -77,10 +95,24 @@ async def _httpx_transport(
     url: str,
     params: dict[str, Any],
 ) -> dict[str, Any]:
+    """Send one Graph request.
+
+    POST payloads go in the request **body**, not the query string. Graph
+    accepts either, but a query string has a practical length limit of a few
+    kilobytes, and `upload_ad_image` base64-encodes an entire image file —
+    that request cannot fit in a URL for any real image.
+
+    Reads and deletes keep their parameters in the query string, which is
+    where Graph expects them.
+    """
     import httpx
 
-    async with httpx.AsyncClient(timeout=30.0) as client:
-        response = await client.request(method, url, params=params)
+    timeout = 30.0 if method != "POST" else 120.0
+    async with httpx.AsyncClient(timeout=timeout) as client:
+        if method == "POST":
+            response = await client.post(url, data=params)
+        else:
+            response = await client.request(method, url, params=params)
     try:
         return response.json()
     except ValueError as exc:
@@ -90,7 +122,7 @@ async def _httpx_transport(
 
 
 class MetaClient:
-    """Minimal read-only Graph API client."""
+    """Graph API client: ad account/preview reads plus paused-object writes."""
 
     def __init__(
         self,
@@ -113,6 +145,57 @@ class MetaClient:
             )
         return result
 
+    async def _post(self, path: str, data: dict[str, Any]) -> dict[str, Any]:
+        """POST to Graph, mirroring `_get`'s body-based error detection.
+
+        Graph returns errors with HTTP 200, so both `_get` and `_post` must
+        inspect the response body for an `"error"` key rather than trust the
+        HTTP status code.
+
+        Defence in depth: a POST to a campaign/ad set/ad creation path with
+        a status other than PAUSED is refused here too, before any
+        transport call, even if a caller bypasses the typed create_*
+        guards. `update_object_status` posts to `/{object_id}`, which never
+        matches these suffixes, so Launch (the only path allowed to send
+        ACTIVE) is unaffected.
+        """
+        if (
+            path.endswith(_PAUSED_ONLY_SUFFIXES)
+            and data.get("status") != CREATE_STATUS
+        ):
+            raise ValueError(
+                "Refusing to create a campaign/ad set/ad with a non-PAUSED "
+                "status. Use update_object_status to launch.",
+            )
+        payload = dict(data)
+        payload["access_token"] = self._token
+        result = await self._transport("POST", f"{GRAPH_BASE_URL}{path}", payload)
+        error = result.get("error") if isinstance(result, dict) else None
+        if error:
+            raise MetaError(
+                error.get("message") or "Meta request failed.",
+                code=error.get("code"),
+                subcode=error.get("error_subcode"),
+            )
+        return result
+
+    async def _delete(self, path: str) -> dict[str, Any]:
+        """DELETE from Graph, with the same body-based error detection.
+
+        Graph reports failures with HTTP 200 and an `"error"` key, so this
+        mirrors `_get`/`_post` rather than trusting the status code.
+        """
+        payload = {"access_token": self._token}
+        result = await self._transport("DELETE", f"{GRAPH_BASE_URL}{path}", payload)
+        error = result.get("error") if isinstance(result, dict) else None
+        if error:
+            raise MetaError(
+                error.get("message") or "Meta request failed.",
+                code=error.get("code"),
+                subcode=error.get("error_subcode"),
+            )
+        return result
+
     async def get_ad_account(self, ad_account_id: str) -> dict[str, Any]:
         """Return billing and configuration fields for an ad account."""
         return await self._get(
@@ -144,3 +227,206 @@ class MetaClient:
             if entries and entries[0].get("body"):
                 previews[ad_format] = entries[0]["body"]
         return previews
+
+    async def create_campaign(
+        self,
+        ad_account_id: str,
+        name: str,
+        objective: str,
+        status: str = CREATE_STATUS,
+        special_ad_categories: list[str] | None = None,
+    ) -> str:
+        """Create a paused campaign and return its id.
+
+        Raises `ValueError` (making no network call) if `status` is
+        anything other than "PAUSED" — campaigns are always created paused;
+        use `update_object_status` to launch.
+        """
+        if status != CREATE_STATUS:
+            raise ValueError(
+                "Campaigns are always created PAUSED. Use "
+                "update_object_status to launch.",
+            )
+        result = await self._post(
+            f"/{ad_account_id}/campaigns",
+            {
+                "name": name,
+                "objective": objective,
+                "status": CREATE_STATUS,
+                "special_ad_categories": json.dumps(special_ad_categories or []),
+            },
+        )
+        return result["id"]
+
+    async def create_ad_set(
+        self,
+        ad_account_id: str,
+        campaign_id: str,
+        name: str,
+        daily_budget: int,
+        targeting: dict[str, Any],
+        optimization_goal: str,
+        billing_event: str,
+        status: str = CREATE_STATUS,
+    ) -> str:
+        """Create a paused ad set and return its id.
+
+        `targeting` is JSON-encoded before being sent, since Graph expects
+        it as a JSON string rather than a nested object in form/query
+        params. Raises `ValueError` (making no network call) if `status`
+        is anything other than "PAUSED".
+        """
+        if status != CREATE_STATUS:
+            raise ValueError(
+                "Ad sets are always created PAUSED. Use "
+                "update_object_status to launch.",
+            )
+        result = await self._post(
+            f"/{ad_account_id}/adsets",
+            {
+                "name": name,
+                "campaign_id": campaign_id,
+                "daily_budget": daily_budget,
+                "targeting": json.dumps(targeting),
+                "optimization_goal": optimization_goal,
+                "billing_event": billing_event,
+                "status": CREATE_STATUS,
+            },
+        )
+        return result["id"]
+
+    async def upload_ad_image(self, ad_account_id: str, image_path: str) -> str:
+        """Upload an image file and return the hash Meta assigns it.
+
+        The file's bytes are base64-encoded and posted under `bytes`, per
+        Graph's `/adimages` contract. The response shape is
+        `{"images": {"<key>": {"hash": "...", ...}}}`; this returns the
+        first hash found and raises `MetaError` if none is present.
+        """
+        with open(image_path, "rb") as image_file:
+            encoded = base64.b64encode(image_file.read()).decode("ascii")
+        result = await self._post(
+            f"/{ad_account_id}/adimages",
+            {"bytes": encoded},
+        )
+        images = result.get("images") or {}
+        for entry in images.values():
+            image_hash = entry.get("hash") if isinstance(entry, dict) else None
+            if image_hash:
+                return image_hash
+        raise MetaError("Meta did not return an image hash for the upload.")
+
+    async def create_ad_creative(
+        self,
+        ad_account_id: str,
+        name: str,
+        page_id: str,
+        message: str,
+        headline: str,
+        description: str,
+        link: str,
+        image_hash: str,
+    ) -> str:
+        """Create an ad creative and return its id.
+
+        Ad creatives have no status of their own in Graph (only the ad
+        that references one does), so there is no PAUSED guard here.
+        `object_story_spec` is JSON-encoded before being sent.
+        """
+        object_story_spec = {
+            "page_id": page_id,
+            "link_data": {
+                "message": message,
+                "name": headline,
+                "description": description,
+                "link": link,
+                "image_hash": image_hash,
+            },
+        }
+        result = await self._post(
+            f"/{ad_account_id}/adcreatives",
+            {
+                "name": name,
+                "object_story_spec": json.dumps(object_story_spec),
+            },
+        )
+        return result["id"]
+
+    async def create_ad(
+        self,
+        ad_account_id: str,
+        name: str,
+        adset_id: str,
+        creative_id: str,
+        status: str = CREATE_STATUS,
+    ) -> str:
+        """Create a paused ad linking an ad set to a creative, return its id.
+
+        Raises `ValueError` (making no network call) if `status` is
+        anything other than "PAUSED".
+        """
+        if status != CREATE_STATUS:
+            raise ValueError(
+                "Ads are always created PAUSED. Use "
+                "update_object_status to launch.",
+            )
+        result = await self._post(
+            f"/{ad_account_id}/ads",
+            {
+                "name": name,
+                "adset_id": adset_id,
+                "creative": json.dumps({"creative_id": creative_id}),
+                "status": CREATE_STATUS,
+            },
+        )
+        return result["id"]
+
+    async def update_object_status(self, object_id: str, status: str) -> None:
+        """Update any campaign/ad set/ad's status.
+
+        This is the ONLY method permitted to send "ACTIVE" — it is what
+        the Launch endpoint (Task 6) calls once a human has approved spend.
+        """
+        await self._post(f"/{object_id}", {"status": status})
+
+    async def delete_object(self, object_id: str) -> None:
+        """Delete a campaign, ad set or ad on Meta.
+
+        Deleting a campaign cascades to its ad sets and ads, so callers
+        holding the whole chain only need to delete the campaign.
+
+        Meta answers a delete of an already-absent object with code 100
+        ("Unsupported get request" / object does not exist). That is the
+        state the caller wanted, so it is swallowed and the delete is
+        idempotent — retrying a partially-failed cleanup is safe.
+        """
+        try:
+            await self._delete(f"/{object_id}")
+        except MetaError as exc:
+            if exc.code == 100:
+                return
+            raise
+
+    async def list_campaigns(self, ad_account_id: str) -> list[dict[str, Any]]:
+        """Return this ad account's campaigns."""
+        result = await self._get(
+            f"/{ad_account_id}/campaigns",
+            {"fields": LIST_FIELDS},
+        )
+        return result.get("data") or []
+
+    async def list_ad_sets(self, ad_account_id: str) -> list[dict[str, Any]]:
+        """Return this ad account's ad sets."""
+        result = await self._get(
+            f"/{ad_account_id}/adsets",
+            {"fields": LIST_FIELDS},
+        )
+        return result.get("data") or []
+
+    async def list_ads(self, ad_account_id: str) -> list[dict[str, Any]]:
+        """Return this ad account's ads."""
+        result = await self._get(
+            f"/{ad_account_id}/ads",
+            {"fields": LIST_FIELDS},
+        )
+        return result.get("data") or []

src/adclaw/meta/objects.py

@@ -0,0 +1,168 @@
+# -*- coding: utf-8 -*-
+"""Pure mappers from a CampaignSpec's stored JSON blobs to Graph API params.
+
+These functions do no I/O and make no Meta calls; they only reshape data so
+`MetaClient`'s write methods stay transport-only and this mapping logic is
+unit-testable without a fake transport at all. Callers spread the returned
+dict as keyword arguments into the matching `MetaClient` method, e.g.:
+
+    await client.create_campaign(ad_account_id, **build_campaign_payload(spec))
+
+Field mapping decisions
+------------------------
+Phase 2 does not yet have separate AdSet/Ad spec models — `CampaignSpec` is
+the only spec in play — so ad-set- and ad-level fields that don't fit
+`CampaignSpec`'s campaign-level columns are read out of `spec.advanced`.
+This is a deliberate, documented choice rather than an oversight:
+
+- `spec.budget["daily_budget"]` -> ad set `daily_budget`, passed through
+  unchanged (already minor currency units / cents, matching
+  `adclaw.campaigns.validation` and Graph's own expectation). Lifetime
+  budgets are out of scope for this mapping: `daily_budget` is required and
+  `build_ad_set_payload` raises `ValueError` if it is missing.
+- `spec.targeting["age_min"]` / `["age_max"]` -> Graph
+  `targeting.age_min` / `targeting.age_max`.
+- `spec.targeting["countries"]` -> Graph
+  `targeting.geo_locations.countries`.
+- `spec.targeting["genders"]` -> Graph `targeting.genders`, if present.
+- `spec.advanced["special_ad_categories"]` -> campaign
+  `special_ad_categories` (defaults to an empty list). A bare string is
+  normalised to a single-element list rather than exploded into
+  characters, since `advanced` is unvalidated free-form JSON and this is
+  the compliance field for regulated advertising (housing, credit,
+  employment, ...).
+- `spec.advanced["optimization_goal"]` / `["billing_event"]` -> ad set
+  fields of the same name, defaulting to `LEAD_GENERATION` /
+  `IMPRESSIONS` (MaskanX's default lead-gen objective). Both are checked
+  against an allowlist (`ALLOWED_OPTIMIZATION_GOALS` /
+  `ALLOWED_BILLING_EVENTS`) before being forwarded, since `advanced` is
+  client-controlled and `billing_event` in particular determines how the
+  ad account is charged; an unrecognised value raises `ValueError` naming
+  the offending value and the allowed set rather than reaching Meta.
+- `spec.advanced["message"]` / `["headline"]` / `["description"]` /
+  `["link"]` -> ad creative `object_story_spec.link_data` fields of the
+  same purpose (headline defaults to the campaign name if unset).
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from ..campaigns.models import CampaignSpec
+
+DEFAULT_OPTIMIZATION_GOAL = "LEAD_GENERATION"
+DEFAULT_BILLING_EVENT = "IMPRESSIONS"
+
+# advanced.optimization_goal / advanced.billing_event are client-controlled
+# and billing_event determines how the ad account is charged, so both are
+# checked against an allowlist before being forwarded to a spending API.
+ALLOWED_OPTIMIZATION_GOALS = frozenset({
+    "LEAD_GENERATION", "LINK_CLICKS", "IMPRESSIONS", "REACH",
+    "OFFSITE_CONVERSIONS", "LANDING_PAGE_VIEWS", "THRUPLAY",
+    "POST_ENGAGEMENT", "QUALITY_CALL",
+})
+ALLOWED_BILLING_EVENTS = frozenset({
+    "IMPRESSIONS", "LINK_CLICKS", "THRUPLAY", "POST_ENGAGEMENT",
+})
+
+
+def build_campaign_payload(spec: CampaignSpec) -> dict[str, Any]:
+    """Map a CampaignSpec onto `MetaClient.create_campaign` kwargs.
+
+    Raises `ValueError` if `spec.objective` is unset — Graph requires an
+    objective on every campaign and there is no sensible default to guess.
+    """
+    if not spec.objective:
+        raise ValueError(
+            "CampaignSpec.objective is required to create a Meta campaign.",
+        )
+    # A bare string ("HOUSING") must become a single-element list, not be
+    # exploded into characters by list(); advanced is unvalidated JSON from
+    # the API and special_ad_categories is the compliance field for
+    # regulated advertising, so this cannot be left to Meta's opaque error.
+    raw_categories = spec.advanced.get("special_ad_categories") or []
+    categories = (
+        [raw_categories] if isinstance(raw_categories, str) else list(raw_categories)
+    )
+    return {
+        "name": spec.name,
+        "objective": spec.objective,
+        "special_ad_categories": categories,
+    }
+
+
+def build_ad_set_payload(spec: CampaignSpec, campaign_id: str) -> dict[str, Any]:
+    """Map a CampaignSpec onto `MetaClient.create_ad_set` kwargs.
+
+    Raises `ValueError` if `spec.budget["daily_budget"]` is unset — this
+    mapping does not yet support lifetime-budget campaigns.
+    """
+    daily_budget = spec.budget.get("daily_budget")
+    if daily_budget is None:
+        raise ValueError(
+            "CampaignSpec.budget.daily_budget is required to create an ad "
+            "set (lifetime budgets are not yet supported).",
+        )
+
+    targeting: dict[str, Any] = {}
+    age_min = spec.targeting.get("age_min")
+    age_max = spec.targeting.get("age_max")
+    if age_min is not None:
+        targeting["age_min"] = age_min
+    if age_max is not None:
+        targeting["age_max"] = age_max
+    countries = spec.targeting.get("countries")
+    if countries:
+        targeting["geo_locations"] = {"countries": list(countries)}
+    genders = spec.targeting.get("genders")
+    if genders:
+        targeting["genders"] = list(genders)
+
+    # Graph requires targeting on every ad set. Catching it here names the
+    # missing field; letting it through produces an opaque Meta rejection
+    # part-way into building the object chain.
+    if not targeting:
+        raise ValueError(
+            "CampaignSpec.targeting is empty, so the ad set has no audience. "
+            "Set at least one of countries, age_min, age_max or genders.",
+        )
+
+    optimization_goal = spec.advanced.get(
+        "optimization_goal", DEFAULT_OPTIMIZATION_GOAL,
+    )
+    if optimization_goal not in ALLOWED_OPTIMIZATION_GOALS:
+        raise ValueError(
+            f"Unsupported optimization_goal {optimization_goal!r}. Allowed: "
+            f"{sorted(ALLOWED_OPTIMIZATION_GOALS)}",
+        )
+    billing_event = spec.advanced.get("billing_event", DEFAULT_BILLING_EVENT)
+    if billing_event not in ALLOWED_BILLING_EVENTS:
+        raise ValueError(
+            f"Unsupported billing_event {billing_event!r}. Allowed: "
+            f"{sorted(ALLOWED_BILLING_EVENTS)}",
+        )
+
+    return {
+        "campaign_id": campaign_id,
+        "name": f"{spec.name} - Ad Set",
+        "daily_budget": daily_budget,
+        "targeting": targeting,
+        "optimization_goal": optimization_goal,
+        "billing_event": billing_event,
+    }
+
+
+def build_creative_payload(
+    spec: CampaignSpec,
+    page_id: str,
+    image_hash: str,
+) -> dict[str, Any]:
+    """Map a CampaignSpec onto `MetaClient.create_ad_creative` kwargs."""
+    return {
+        "name": f"{spec.name} - Creative",
+        "page_id": page_id,
+        "message": spec.advanced.get("message", ""),
+        "headline": spec.advanced.get("headline") or spec.name,
+        "description": spec.advanced.get("description", ""),
+        "link": spec.advanced.get("link", ""),
+        "image_hash": image_hash,
+    }

tests/test_campaign_adopt.py

@@ -0,0 +1,202 @@
+# -*- coding: utf-8 -*-
+"""Discovering and adopting campaigns created directly in Meta Ads Manager."""
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from adclaw.app.routers import campaigns as campaigns_router
+from adclaw.campaigns.adopt import (
+    adopt_campaign,
+    discover_campaigns,
+    local_status_for,
+)
+from adclaw.campaigns.models import CampaignSpec
+
+
+class FakeRepo:
+    def __init__(self, campaigns=None):
+        self.items: dict[str, CampaignSpec] = {c.id: c for c in (campaigns or [])}
+        self.events: list[dict] = []
+
+    async def list_campaigns(self, company_id=None, status=None):
+        return list(self.items.values())
+
+    async def get_campaign(self, campaign_id):
+        return self.items.get(campaign_id)
+
+    async def create_campaign(self, spec):
+        self.items[spec.id] = spec
+        return spec
+
+    async def add_event(
+        self, campaign_id, event_type, actor=None, reason=None, payload=None,
+    ):
+        self.events.append({"campaign_id": campaign_id, "event_type": event_type})
+
+
+class FakeMeta:
+    def __init__(self, campaigns=None):
+        self.campaigns = campaigns if campaigns is not None else [
+            {"id": "meta_1", "name": "Remote one", "status": "ACTIVE"},
+            {"id": "meta_2", "name": "Remote two", "status": "PAUSED"},
+        ]
+        self.list_calls = 0
+
+    async def list_campaigns(self, ad_account_id):
+        self.list_calls += 1
+        return self.campaigns
+
+
+def _local(meta_id: str | None, **overrides) -> CampaignSpec:
+    data = {
+        "id": f"camp_{meta_id or 'local'}",
+        "name": "Local campaign",
+        "status": "draft",
+        "ad_account_id": "act_1",
+        "meta_campaign_id": meta_id,
+    }
+    data.update(overrides)
+    return CampaignSpec(**data)
+
+
+@pytest.mark.parametrize(
+    ("meta_status", "expected"),
+    [
+        ("ACTIVE", "live"),
+        ("PAUSED", "paused"),
+        ("DELETED", "archived"),
+        ("ARCHIVED", "archived"),
+        ("active", "live"),
+        (None, "paused"),
+        ("SOMETHING_NEW", "paused"),
+    ],
+)
+def test_meta_status_maps_to_local_status(meta_status, expected):
+    assert local_status_for(meta_status) == expected
+
+
+@pytest.mark.asyncio
+async def test_discover_lists_only_unknown_campaigns():
+    repo = FakeRepo([_local("meta_1")])
+    meta = FakeMeta()
+
+    found = await discover_campaigns(repo, meta, "act_1")
+
+    assert [c["id"] for c in found] == ["meta_2"]
+
+
+@pytest.mark.asyncio
+async def test_discover_ignores_local_campaigns_with_no_meta_id():
+    repo = FakeRepo([_local(None)])
+    meta = FakeMeta()
+
+    found = await discover_campaigns(repo, meta, "act_1")
+
+    assert [c["id"] for c in found] == ["meta_1", "meta_2"]
+
+
+@pytest.mark.asyncio
+async def test_adopt_creates_an_imported_record():
+    repo, meta = FakeRepo(), FakeMeta()
+
+    created = await adopt_campaign(
+        repo, meta, ad_account_id="act_1", meta_campaign_id="meta_2", actor="owner",
+    )
+
+    assert created.origin == "imported"
+    assert created.meta_campaign_id == "meta_2"
+    assert created.sync_status == "synced"
+    assert created.status == "paused"
+    assert created.name == "Remote two"
+    assert repo.events[-1]["event_type"] == "campaign.adopted"
+
+
+@pytest.mark.asyncio
+async def test_adopting_twice_returns_the_existing_record():
+    """The unique index on meta_campaign_id must never be violated."""
+    repo, meta = FakeRepo(), FakeMeta()
+
+    first = await adopt_campaign(
+        repo, meta, ad_account_id="act_1", meta_campaign_id="meta_1", actor="owner",
+    )
+    second = await adopt_campaign(
+        repo, meta, ad_account_id="act_1", meta_campaign_id="meta_1", actor="owner",
+    )
+
+    assert second.id == first.id
+    assert len(repo.items) == 1
+    # Only the first adoption writes an event.
+    assert len(repo.events) == 1
+
+
+@pytest.mark.asyncio
+async def test_adopting_an_unknown_campaign_raises_lookup_error():
+    repo, meta = FakeRepo(), FakeMeta()
+
+    with pytest.raises(LookupError):
+        await adopt_campaign(
+            repo, meta, ad_account_id="act_1", meta_campaign_id="nope", actor="owner",
+        )
+
+
+@pytest.mark.asyncio
+async def test_adopted_record_keeps_meta_provenance():
+    repo = FakeRepo()
+    meta = FakeMeta([
+        {
+            "id": "meta_9",
+            "name": "With detail",
+            "status": "ACTIVE",
+            "effective_status": "ACTIVE",
+            "created_time": "2026-07-01T10:00:00+0000",
+        },
+    ])
+
+    created = await adopt_campaign(
+        repo, meta, ad_account_id="act_1", meta_campaign_id="meta_9", actor="owner",
+    )
+
+    provenance = created.advanced["imported_from_meta"]
+    assert provenance["effective_status"] == "ACTIVE"
+    assert provenance["created_time"] == "2026-07-01T10:00:00+0000"
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    repo, meta = FakeRepo(), FakeMeta()
+    monkeypatch.setattr(campaigns_router, "get_repository", lambda: repo)
+    monkeypatch.setattr(campaigns_router, "get_meta_client", lambda: meta)
+    app = FastAPI()
+    app.include_router(campaigns_router.router, prefix="/api")
+    test_client = TestClient(app)
+    test_client.repo = repo
+    return test_client
+
+
+def test_discover_endpoint_is_not_shadowed_by_the_id_route(client):
+    """`/discover` is a literal path and must not be read as a campaign id."""
+    response = client.get("/api/campaigns/discover?ad_account_id=act_1")
+
+    assert response.status_code == 200
+    assert [c["id"] for c in response.json()] == ["meta_1", "meta_2"]
+
+
+def test_adopt_endpoint_creates_and_is_idempotent(client):
+    body = {"ad_account_id": "act_1", "meta_campaign_id": "meta_1"}
+
+    first = client.post("/api/campaigns/adopt", json=body)
+    second = client.post("/api/campaigns/adopt", json=body)
+
+    assert first.status_code == 201
+    assert first.json()["origin"] == "imported"
+    assert second.json()["id"] == first.json()["id"]
+    assert len(client.repo.items) == 1
+
+
+def test_adopt_unknown_campaign_returns_404(client):
+    response = client.post(
+        "/api/campaigns/adopt",
+        json={"ad_account_id": "act_1", "meta_campaign_id": "missing"},
+    )
+
+    assert response.status_code == 404

tests/test_campaign_api.py

@@ -28,6 +28,14 @@ class FakeRepo:
         self.items[spec.id] = spec
         return spec
 
+    async def update_campaign_with_event(self, spec, event_type, actor=None,
+                                          reason=None, payload=None):
+        if spec.id not in self.items:
+            raise LookupError(f"Campaign {spec.id} does not exist")
+        self.items[spec.id] = spec
+        self.events.append({"campaign_id": spec.id, "event_type": event_type})
+        return spec
+
     async def delete_campaign(self, campaign_id):
         return self.items.pop(campaign_id, None) is not None
 
@@ -97,9 +105,14 @@ def test_submit_then_approve_moves_through_states(client):
     submitted = client.post(f"/api/campaigns/{cid}/submit")
     assert submitted.json()["status"] == "pending_approval"
 
+    # `actor` in the body is a forged-identity attempt: approval must be
+    # attributable to the resolved operator, never to a client-supplied
+    # string. With MASKANX_OPERATOR_TOKENS unset (as in this test), the
+    # operator dependency resolves to "unauthenticated" rather than trusting
+    # the body.
     approved = client.post(f"/api/campaigns/{cid}/approve", json={"actor": "owner"})
     assert approved.json()["status"] == "approved"
-    assert approved.json()["approved_by"] == "owner"
+    assert approved.json()["approved_by"] == "unauthenticated"
 
 
 def test_approve_from_draft_is_rejected(client):
@@ -118,10 +131,17 @@ def test_edit_after_approval_is_rejected(client):
     assert response.status_code == 409
 
 
-def test_launch_endpoint_is_absent_in_phase_1(client):
+def test_launch_is_refused_on_a_freshly_created_campaign(client):
+    """Phase 2 added /launch, so it must be unreachable from `draft`.
+
+    This replaces the Phase 1 test that asserted the endpoint did not exist.
+    The property being guarded is the same one: a campaign cannot start
+    spending without passing through approval and sync first.
+    """
     created = client.post("/api/campaigns", json=_payload()).json()
     response = client.post(f"/api/campaigns/{created['id']}/launch")
-    assert response.status_code == 404
+    assert response.status_code == 409
+    assert "cannot be launched" in response.json()["detail"]
 
 
 def test_preview_returns_iframe_per_format(client):
@@ -212,10 +232,10 @@ def test_submit_missing_campaign_returns_404(client, monkeypatch):
     created = client.post("/api/campaigns", json=_payload()).json()
     cid = created["id"]
 
-    async def raise_lookup(spec):
+    async def raise_lookup(spec, event_type, actor=None, reason=None, payload=None):
         raise LookupError("gone")
 
-    monkeypatch.setattr(client.repo, "update_campaign", raise_lookup)
+    monkeypatch.setattr(client.repo, "update_campaign_with_event", raise_lookup)
     response = client.post(f"/api/campaigns/{cid}/submit")
     assert response.status_code == 404
 

tests/test_campaign_delete.py

@@ -0,0 +1,166 @@
+# -*- coding: utf-8 -*-
+"""Deleting a campaign must not leave objects spending on Meta.
+
+The dangerous case is a campaign forgotten in MaskanX but still live on
+Meta: it keeps drawing budget with nothing left here to show it exists.
+So Meta is deleted first, and a failure there keeps the local row.
+
+The opposite mistake matters too: deleting an imported campaign here must
+not destroy work the operator did in Ads Manager.
+"""
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from adclaw.app.routers import campaigns as campaigns_router
+from adclaw.campaigns.models import CampaignSpec
+from adclaw.campaigns.sync import unsync_campaign
+from adclaw.meta.client import MetaClient, MetaError
+
+
+class FakeRepo:
+    def __init__(self):
+        self.items: dict[str, CampaignSpec] = {}
+        self.deleted: list[str] = []
+
+    async def get_campaign(self, campaign_id):
+        return self.items.get(campaign_id)
+
+    async def delete_campaign(self, campaign_id):
+        self.deleted.append(campaign_id)
+        self.items.pop(campaign_id, None)
+
+
+class FakeMeta:
+    def __init__(self, error: MetaError | None = None):
+        self.error = error
+        self.deleted: list[str] = []
+
+    async def delete_object(self, object_id):
+        if self.error:
+            raise self.error
+        self.deleted.append(object_id)
+
+
+def _campaign(**overrides) -> CampaignSpec:
+    data = {
+        "id": "camp_1",
+        "name": "Q3 lead gen",
+        "status": "live",
+        "origin": "maskanx",
+        "ad_account_id": "act_1",
+        "meta_campaign_id": "meta_camp_1",
+        "budget": {"daily_budget": 10000},
+    }
+    data.update(overrides)
+    return CampaignSpec(**data)
+
+
+@pytest.fixture()
+def env(monkeypatch):
+    repo, meta = FakeRepo(), FakeMeta()
+    monkeypatch.setattr(campaigns_router, "get_repository", lambda: repo)
+    monkeypatch.setattr(campaigns_router, "get_meta_client", lambda: meta)
+    app = FastAPI()
+    app.include_router(campaigns_router.router, prefix="/api")
+    return TestClient(app), repo, meta
+
+
+def test_delete_removes_the_campaign_on_meta_as_well(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign()
+
+    response = client.delete("/api/campaigns/camp_1")
+
+    assert response.status_code == 204
+    assert meta.deleted == ["meta_camp_1"]
+    assert repo.deleted == ["camp_1"]
+
+
+def test_delete_keeps_the_local_row_when_meta_delete_fails(env):
+    """Forgetting it here while it still spends there is the worst outcome."""
+    client, repo, meta = env
+    meta.error = MetaError("Permission denied", code=200)
+    repo.items["camp_1"] = _campaign()
+
+    response = client.delete("/api/campaigns/camp_1")
+
+    assert response.status_code == 502
+    assert "Ads Manager" in response.json()["detail"]
+    assert repo.deleted == []
+    assert "camp_1" in repo.items
+
+
+def test_delete_of_an_imported_campaign_is_refused(env):
+    """Deleting it on Meta would destroy Ads Manager work; deleting only the
+    local row would achieve nothing, since the reconciler re-imports it."""
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(origin="imported")
+
+    response = client.delete("/api/campaigns/camp_1")
+
+    assert response.status_code == 409
+    assert "Ads Manager" in response.json()["detail"]
+    assert meta.deleted == []
+    assert repo.deleted == []
+
+
+def test_delete_of_an_unsynced_campaign_does_not_call_meta(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(status="draft", meta_campaign_id=None)
+
+    response = client.delete("/api/campaigns/camp_1")
+
+    assert response.status_code == 204
+    assert meta.deleted == []
+    assert repo.deleted == ["camp_1"]
+
+
+def test_delete_of_a_missing_campaign_is_404(env):
+    client, repo, meta = env
+
+    response = client.delete("/api/campaigns/nope")
+
+    assert response.status_code == 404
+    assert meta.deleted == []
+
+
+async def test_unsync_reports_whether_it_deleted():
+    meta = FakeMeta()
+
+    assert await unsync_campaign(meta, _campaign()) is True
+    assert await unsync_campaign(meta, _campaign(origin="imported")) is False
+    assert await unsync_campaign(meta, _campaign(meta_campaign_id=None)) is False
+    assert meta.deleted == ["meta_camp_1"]
+
+
+# --- client transport ---
+
+
+async def test_delete_object_issues_a_graph_delete():
+    calls = []
+
+    async def transport(method, url, params):
+        calls.append((method, url))
+        return {"success": True}
+
+    await MetaClient("tok", transport=transport).delete_object("meta_camp_1")
+
+    assert calls == [("DELETE", "https://graph.facebook.com/v23.0/meta_camp_1")]
+
+
+async def test_delete_object_treats_an_absent_object_as_deleted():
+    """Retrying a partially-failed cleanup must not fail on the done part."""
+
+    async def transport(method, url, params):
+        return {"error": {"message": "Unsupported get request.", "code": 100}}
+
+    await MetaClient("tok", transport=transport).delete_object("gone")
+
+
+async def test_delete_object_raises_on_any_other_meta_error():
+    async def transport(method, url, params):
+        return {"error": {"message": "Permission denied", "code": 200}}
+
+    with pytest.raises(MetaError):
+        await MetaClient("tok", transport=transport).delete_object("meta_camp_1")

tests/test_campaign_launch.py

@@ -0,0 +1,258 @@
+# -*- coding: utf-8 -*-
+"""Launch, pause and stop.
+
+Launch is the only action that starts real spend, so most of these tests are
+about the guards that must refuse it.
+"""
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from adclaw.app.routers import campaigns as campaigns_router
+from adclaw.campaigns.models import CampaignSpec
+from adclaw.meta.client import MetaError
+
+
+class FakeRepo:
+    def __init__(self):
+        self.items: dict[str, CampaignSpec] = {}
+        self.events: list[dict] = []
+
+    async def get_campaign(self, campaign_id):
+        return self.items.get(campaign_id)
+
+    async def update_campaign_with_event(
+        self, spec, event_type, actor=None, reason=None, payload=None,
+    ):
+        self.items[spec.id] = spec
+        self.events.append({"event_type": event_type, "actor": actor})
+        return spec
+
+
+class FakeMeta:
+    def __init__(self, funding="card_1"):
+        self.funding = funding
+        self.status_calls: list[tuple[str, str]] = []
+        # Object id whose status update should fail, for partial-failure tests.
+        self.fail_on: str | None = None
+
+    async def get_ad_account(self, ad_account_id):
+        account = {"id": ad_account_id, "min_daily_budget": 9709}
+        if self.funding is not None:
+            account["funding_source"] = self.funding
+        return account
+
+    async def update_object_status(self, object_id, status):
+        if object_id == self.fail_on:
+            raise MetaError("Meta refused the status change.", code=100)
+        self.status_calls.append((object_id, status))
+
+
+def _campaign(**overrides) -> CampaignSpec:
+    data = {
+        "id": "camp_1",
+        "name": "Q3 lead gen",
+        "status": "synced",
+        "ad_account_id": "act_1",
+        "meta_campaign_id": "meta_camp_1",
+        "approved_by": "owner",
+        "budget": {"daily_budget": 10000},
+    }
+    data.update(overrides)
+    return CampaignSpec(**data)
+
+
+@pytest.fixture()
+def env(monkeypatch):
+    repo, meta = FakeRepo(), FakeMeta()
+    monkeypatch.setattr(campaigns_router, "get_repository", lambda: repo)
+    monkeypatch.setattr(campaigns_router, "get_meta_client", lambda: meta)
+    app = FastAPI()
+    app.include_router(campaigns_router.router, prefix="/api")
+    return TestClient(app), repo, meta
+
+
+def test_launch_activates_the_whole_chain_campaign_last(env):
+    """Meta only delivers when ad, ad set and campaign are all active.
+
+    The campaign goes last so a failure part-way leaves it paused, unable
+    to spend.
+    """
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(
+        advanced={"meta_sync": {"adset_id": "meta_set_1", "ad_id": "meta_ad_1"}},
+    )
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 200
+    assert response.json()["status"] == "live"
+    assert meta.status_calls == [
+        ("meta_ad_1", "ACTIVE"),
+        ("meta_set_1", "ACTIVE"),
+        ("meta_camp_1", "ACTIVE"),
+    ]
+    assert repo.events[-1]["event_type"] == "campaign.launch"
+
+
+def test_launch_leaves_the_campaign_paused_when_a_child_fails(env):
+    """A half-activated chain must not be able to spend."""
+    client, repo, meta = env
+    meta.fail_on = "meta_set_1"
+    repo.items["camp_1"] = _campaign(
+        advanced={"meta_sync": {"adset_id": "meta_set_1", "ad_id": "meta_ad_1"}},
+    )
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code >= 400
+    assert ("meta_camp_1", "ACTIVE") not in meta.status_calls
+    assert repo.items["camp_1"].status == "synced"
+
+
+def test_launch_of_an_imported_campaign_only_touches_the_campaign(env):
+    """Its ad sets and ads keep the status the operator set in Ads Manager."""
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(origin="imported")
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 200
+    assert meta.status_calls == [("meta_camp_1", "ACTIVE")]
+
+
+def test_launch_is_refused_without_a_payment_method(env):
+    """The account cannot deliver, so refuse rather than let Meta fail."""
+    client, repo, meta = env
+    meta.funding = None
+    repo.items["camp_1"] = _campaign()
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 409
+    assert "Meta Business Manager" in response.json()["detail"]
+    assert meta.status_calls == []
+
+
+def test_launch_is_refused_when_approval_was_not_attributable(env):
+    """An approval nobody can be named for must not authorise spend."""
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(approved_by="unauthenticated")
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 409
+    assert "identified operator" in response.json()["detail"]
+    assert meta.status_calls == []
+
+
+def test_launch_is_refused_when_never_approved(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(approved_by=None)
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 409
+    assert meta.status_calls == []
+
+
+@pytest.mark.parametrize("status", ["draft", "pending_approval", "approved", "stopped"])
+def test_launch_is_refused_from_a_non_launchable_status(env, status):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(status=status)
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 409
+    assert meta.status_calls == []
+
+
+def test_launch_is_refused_when_not_synced(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(meta_campaign_id=None)
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 409
+    assert "not been synced" in response.json()["detail"]
+    assert meta.status_calls == []
+
+
+def test_launch_requires_a_valid_operator_token(env, monkeypatch):
+    monkeypatch.setenv("MASKANX_OPERATOR_TOKENS", "owner:s3cr3t")
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign()
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 401
+    assert meta.status_calls == []
+
+
+def test_launch_records_the_operator_not_a_client_string(env, monkeypatch):
+    monkeypatch.setenv("MASKANX_OPERATOR_TOKENS", "owner:s3cr3t")
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign()
+
+    response = client.post(
+        "/api/campaigns/camp_1/launch",
+        headers={"X-MaskanX-Operator": "s3cr3t"},
+    )
+
+    assert response.status_code == 200
+    assert repo.events[-1]["actor"] == "owner"
+
+
+def test_paused_campaign_can_be_relaunched(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(status="paused")
+
+    response = client.post("/api/campaigns/camp_1/launch")
+
+    assert response.status_code == 200
+    assert response.json()["status"] == "live"
+
+
+def test_pause_halts_spend_on_meta(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(status="live")
+
+    response = client.post("/api/campaigns/camp_1/pause")
+
+    assert response.status_code == 200
+    assert response.json()["status"] == "paused"
+    assert meta.status_calls == [("meta_camp_1", "PAUSED")]
+
+
+def test_pause_is_refused_when_not_live(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(status="synced")
+
+    response = client.post("/api/campaigns/camp_1/pause")
+
+    assert response.status_code == 409
+    assert meta.status_calls == []
+
+
+def test_stop_pauses_on_meta_before_recording(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(status="live")
+
+    response = client.post("/api/campaigns/camp_1/stop")
+
+    assert response.status_code == 200
+    assert response.json()["status"] == "stopped"
+    # Meta must be paused, otherwise a campaign stopped locally would keep
+    # delivering.
+    assert meta.status_calls == [("meta_camp_1", "PAUSED")]
+
+
+def test_stop_works_on_an_unsynced_campaign(env):
+    client, repo, meta = env
+    repo.items["camp_1"] = _campaign(status="approved", meta_campaign_id=None)
+
+    response = client.post("/api/campaigns/camp_1/stop")
+
+    assert response.status_code == 200
+    assert response.json()["status"] == "stopped"
+    assert meta.status_calls == []

tests/test_campaign_operator_auth.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+"""Approve and launch must be attributable to a known operator."""
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from adclaw.app.routers import _operator
+
+
+def _app():
+    from adclaw.app.routers import campaigns as campaigns_router
+
+    app = FastAPI()
+    app.include_router(campaigns_router.router, prefix="/api")
+    return app
+
+
+def test_unset_env_allows_and_reports_unauthenticated(monkeypatch):
+    monkeypatch.delenv("MASKANX_OPERATOR_TOKENS", raising=False)
+    assert _operator.resolve_operator(None) == "unauthenticated"
+
+
+def test_known_token_resolves_to_its_name(monkeypatch):
+    monkeypatch.setenv("MASKANX_OPERATOR_TOKENS", "owner:s3cr3t,ops:t0ken")
+    assert _operator.resolve_operator("s3cr3t") == "owner"
+    assert _operator.resolve_operator("t0ken") == "ops"
+
+
+def test_unknown_token_is_rejected(monkeypatch):
+    monkeypatch.setenv("MASKANX_OPERATOR_TOKENS", "owner:s3cr3t")
+    with pytest.raises(_operator.OperatorAuthError):
+        _operator.resolve_operator("wrong")
+
+
+def test_missing_header_is_rejected_when_configured(monkeypatch):
+    monkeypatch.setenv("MASKANX_OPERATOR_TOKENS", "owner:s3cr3t")
+    with pytest.raises(_operator.OperatorAuthError):
+        _operator.resolve_operator(None)
+
+
+def test_malformed_config_entries_are_ignored(monkeypatch):
+    monkeypatch.setenv("MASKANX_OPERATOR_TOKENS", "bad,owner:s3cr3t,:empty,x:")
+    assert _operator.resolve_operator("s3cr3t") == "owner"
+    with pytest.raises(_operator.OperatorAuthError):
+        _operator.resolve_operator("empty")
+
+
+def test_approve_uses_operator_identity_not_client_actor(monkeypatch):
+    """The body's `actor` must never override the authenticated operator."""
+    monkeypatch.setenv("MASKANX_OPERATOR_TOKENS", "owner:s3cr3t")
+    client = TestClient(_app())
+    # Full flow is covered in test_campaign_api.py; here we assert the
+    # dependency is wired so a forged actor cannot win.
+    from adclaw.app.routers import campaigns as campaigns_router
+
+    import inspect
+
+    source = inspect.getsource(campaigns_router.approve_campaign)
+    assert "require_operator" in source
+    assert "payload.actor" not in source

tests/test_campaign_reconcile.py

@@ -0,0 +1,173 @@
+# -*- coding: utf-8 -*-
+"""Two-way reconciliation between local records and Meta."""
+import pytest
+
+from adclaw.campaigns.models import CampaignSpec
+from adclaw.campaigns.reconcile import (
+    DEFAULT_INTERVAL_SECONDS,
+    reconcile_interval_seconds,
+    reconcile_once,
+)
+
+
+class FakeRepo:
+    def __init__(self, campaigns=None):
+        self.items: dict[str, CampaignSpec] = {c.id: c for c in (campaigns or [])}
+        self.events: list[dict] = []
+
+    async def list_campaigns(self, company_id=None, status=None):
+        return list(self.items.values())
+
+    async def create_campaign(self, spec):
+        self.items[spec.id] = spec
+        return spec
+
+    async def update_campaign_with_event(
+        self, spec, event_type, actor=None, reason=None, payload=None,
+    ):
+        self.items[spec.id] = spec
+        self.events.append({"id": spec.id, "event_type": event_type})
+        return spec
+
+    async def add_event(
+        self, campaign_id, event_type, actor=None, reason=None, payload=None,
+    ):
+        self.events.append({"id": campaign_id, "event_type": event_type})
+
+
+class FakeMeta:
+    def __init__(self, campaigns):
+        self.campaigns = campaigns
+
+    async def list_campaigns(self, ad_account_id):
+        return self.campaigns
+
+
+def _local(**overrides) -> CampaignSpec:
+    data = {
+        "id": "camp_1",
+        "name": "Local",
+        "status": "live",
+        "origin": "maskanx",
+        "ad_account_id": "act_1",
+        "meta_campaign_id": "meta_1",
+        "guardrails": {"max_cost_per_lead": 150},
+        "approved_by": "owner",
+    }
+    data.update(overrides)
+    return CampaignSpec(**data)
+
+
+def test_interval_defaults_when_unset(monkeypatch):
+    monkeypatch.delenv("MASKANX_CAMPAIGN_RECONCILE_SECONDS", raising=False)
+    assert reconcile_interval_seconds() == DEFAULT_INTERVAL_SECONDS
+
+
+def test_interval_zero_disables_the_loop(monkeypatch):
+    monkeypatch.setenv("MASKANX_CAMPAIGN_RECONCILE_SECONDS", "0")
+    assert reconcile_interval_seconds() == 0
+
+
+def test_interval_falls_back_on_a_bad_value(monkeypatch):
+    monkeypatch.setenv("MASKANX_CAMPAIGN_RECONCILE_SECONDS", "not-a-number")
+    assert reconcile_interval_seconds() == DEFAULT_INTERVAL_SECONDS
+
+
+@pytest.mark.asyncio
+async def test_unknown_meta_campaign_is_imported():
+    repo = FakeRepo()
+    meta = FakeMeta([{"id": "meta_9", "name": "Remote", "status": "ACTIVE"}])
+
+    result = await reconcile_once(repo, meta, "act_1")
+
+    assert len(result.imported) == 1
+    imported = repo.items[result.imported[0]]
+    assert imported.origin == "imported"
+    assert imported.meta_campaign_id == "meta_9"
+    assert imported.status == "live"
+
+
+@pytest.mark.asyncio
+async def test_status_change_in_meta_updates_the_local_record():
+    repo = FakeRepo([_local(status="live")])
+    meta = FakeMeta([{"id": "meta_1", "name": "Local", "status": "PAUSED"}])
+
+    result = await reconcile_once(repo, meta, "act_1")
+
+    assert result.status_changed == ["camp_1"]
+    assert repo.items["camp_1"].status == "paused"
+    assert repo.events[-1]["event_type"] == "campaign.status_reconciled"
+
+
+@pytest.mark.asyncio
+async def test_campaign_deleted_in_meta_is_archived_not_removed():
+    """Spend history must stay reportable, so the record survives."""
+    repo = FakeRepo([_local(status="live")])
+    meta = FakeMeta([])
+
+    result = await reconcile_once(repo, meta, "act_1")
+
+    assert result.archived == ["camp_1"]
+    assert "camp_1" in repo.items
+    assert repo.items["camp_1"].status == "archived"
+    assert repo.items["camp_1"].advanced["deleted_in_meta_at"]
+    assert repo.events[-1]["event_type"] == "campaign.deleted_in_meta"
+
+
+@pytest.mark.asyncio
+async def test_already_archived_campaign_is_not_rearchived():
+    repo = FakeRepo([_local(status="archived")])
+    meta = FakeMeta([])
+
+    result = await reconcile_once(repo, meta, "act_1")
+
+    assert result.archived == []
+    assert repo.events == []
+
+
+@pytest.mark.asyncio
+async def test_reconciliation_never_touches_maskanx_owned_metadata():
+    """Meta owns delivery state; guardrails and approvals stay ours."""
+    repo = FakeRepo([_local(status="live")])
+    meta = FakeMeta([{"id": "meta_1", "name": "Renamed in Meta", "status": "PAUSED"}])
+
+    await reconcile_once(repo, meta, "act_1")
+
+    campaign = repo.items["camp_1"]
+    assert campaign.guardrails == {"max_cost_per_lead": 150}
+    assert campaign.approved_by == "owner"
+
+
+@pytest.mark.parametrize("status", ["draft", "pending_approval", "approved"])
+@pytest.mark.asyncio
+async def test_local_only_statuses_are_left_alone(status):
+    """A campaign that never reached Meta must not be reconciled."""
+    repo = FakeRepo([_local(status=status)])
+    meta = FakeMeta([{"id": "meta_1", "name": "Local", "status": "ACTIVE"}])
+
+    result = await reconcile_once(repo, meta, "act_1")
+
+    assert result.status_changed == []
+    assert repo.items["camp_1"].status == status
+
+
+@pytest.mark.asyncio
+async def test_campaigns_from_other_ad_accounts_are_ignored():
+    repo = FakeRepo([_local(ad_account_id="act_other")])
+    meta = FakeMeta([])
+
+    result = await reconcile_once(repo, meta, "act_1")
+
+    assert result.archived == []
+    assert repo.items["camp_1"].status == "live"
+
+
+@pytest.mark.asyncio
+async def test_matching_status_produces_no_write():
+    repo = FakeRepo([_local(status="live")])
+    meta = FakeMeta([{"id": "meta_1", "name": "Local", "status": "ACTIVE"}])
+
+    result = await reconcile_once(repo, meta, "act_1")
+
+    assert result.changed is False
+    assert repo.events == []

tests/test_campaign_repo_integration.py

@@ -0,0 +1,142 @@
+# -*- coding: utf-8 -*-
+"""Repository SQL exercised against a real PostgreSQL database.
+
+Skipped when PostgreSQL is unreachable so the suite stays runnable offline.
+"""
+import sys
+import uuid
+
+import pytest
+
+from adclaw.campaigns.models import CampaignSpec
+from adclaw.campaigns.repo import CampaignRepository
+
+# psycopg's async mode cannot run on Windows' default ProactorEventLoop
+# (see https://www.psycopg.org/psycopg3/docs/advanced/async.html#async-and-windows).
+# pytest-asyncio creates its per-test event loop lazily using whatever
+# policy is active when each test runs, so switching the policy here at
+# collection time (module import, before any test executes) is sufficient
+# to make every event loop created afterwards -- in this file or any test
+# module collected after it -- selector-based instead. This mirrors what
+# adclaw.cli.main already does for the real app entrypoint; it is set here
+# only for the test process and does not touch any production file.
+if sys.platform == "win32":
+    import asyncio
+
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+
+async def _database_available() -> bool:
+    try:
+        from adclaw.db.connection import connect_database
+
+        conn = await connect_database()
+        await conn.close()
+        return True
+    except Exception:
+        return False
+
+
+@pytest.fixture()
+async def repo():
+    if not await _database_available():
+        pytest.skip("PostgreSQL is not reachable")
+    return CampaignRepository()
+
+
+def _spec(**overrides) -> CampaignSpec:
+    data = {
+        "id": f"test_{uuid.uuid4().hex[:12]}",
+        "name": "Integration test campaign",
+        "status": "draft",
+        "origin": "maskanx",
+        "objective": "OUTCOME_LEADS",
+        "ad_account_id": "act_test",
+        "budget": {"daily_budget": 10000},
+        "guardrails": {"max_cost_per_lead": 150},
+        "targeting": {"age_min": 25},
+        "channels": ["facebook"],
+    }
+    data.update(overrides)
+    return CampaignSpec(**data)
+
+
+@pytest.mark.asyncio
+async def test_create_get_update_delete_round_trip(repo):
+    spec = _spec()
+    created = await repo.create_campaign(spec)
+    try:
+        assert created.id == spec.id
+        assert created.budget["daily_budget"] == 10000
+
+        fetched = await repo.get_campaign(spec.id)
+        assert fetched is not None
+        assert fetched.name == "Integration test campaign"
+
+        fetched.name = "Renamed"
+        fetched.company_id = "co_1"
+        updated = await repo.update_campaign(fetched)
+        assert updated.name == "Renamed"
+        assert updated.company_id == "co_1"
+    finally:
+        assert await repo.delete_campaign(spec.id) is True
+    assert await repo.get_campaign(spec.id) is None
+
+
+@pytest.mark.asyncio
+async def test_update_unknown_id_raises_lookup_error(repo):
+    with pytest.raises(LookupError):
+        await repo.update_campaign(_spec(id="test_does_not_exist"))
+
+
+@pytest.mark.asyncio
+async def test_delete_unknown_id_returns_false(repo):
+    assert await repo.delete_campaign("test_does_not_exist") is False
+
+
+@pytest.mark.asyncio
+async def test_events_are_recorded_and_listed(repo):
+    spec = _spec()
+    await repo.create_campaign(spec)
+    try:
+        await repo.add_event(spec.id, "campaign.created", actor="tester")
+        events = await repo.list_events(spec.id)
+        assert [e["event_type"] for e in events] == ["campaign.created"]
+        assert events[0]["actor"] == "tester"
+    finally:
+        await repo.delete_campaign(spec.id)
+
+
+@pytest.mark.asyncio
+async def test_transactional_update_writes_both_rows(repo):
+    spec = await repo.create_campaign(_spec())
+    try:
+        spec.status = "pending_approval"
+        saved = await repo.update_campaign_with_event(
+            spec, event_type="campaign.submit", actor="tester",
+        )
+        assert saved.status == "pending_approval"
+        events = await repo.list_events(spec.id)
+        assert "campaign.submit" in [e["event_type"] for e in events]
+    finally:
+        await repo.delete_campaign(spec.id)
+
+
+@pytest.mark.asyncio
+async def test_transactional_update_rolls_back_on_unknown_id(repo):
+    ghost = _spec(id="test_ghost")
+    with pytest.raises(LookupError):
+        await repo.update_campaign_with_event(ghost, event_type="campaign.submit")
+    assert await repo.list_events("test_ghost") == []
+
+
+@pytest.mark.asyncio
+async def test_list_filters_by_status(repo):
+    spec = await repo.create_campaign(_spec(status="draft"))
+    try:
+        drafts = await repo.list_campaigns(status="draft")
+        assert spec.id in [c.id for c in drafts]
+        approved = await repo.list_campaigns(status="approved")
+        assert spec.id not in [c.id for c in approved]
+    finally:
+        await repo.delete_campaign(spec.id)

tests/test_campaign_repo_txn.py

@@ -0,0 +1,41 @@
+# -*- coding: utf-8 -*-
+"""The transactional write must issue both statements on one connection."""
+import inspect
+
+from adclaw.campaigns import repo as repo_module
+
+
+def test_transactional_method_exists():
+    assert hasattr(repo_module.CampaignRepository, "update_campaign_with_event")
+
+
+def test_transactional_method_opens_exactly_one_connection():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert source.count("connect_database()") == 1
+
+
+def test_transactional_method_commits_once_after_both_statements():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert source.count("await conn.commit()") == 1
+    assert "maskanx_campaign_events" in source
+    assert "UPDATE maskanx_campaigns" in source
+    assert source.index("UPDATE maskanx_campaigns") < source.index("await conn.commit()")
+    assert source.index("maskanx_campaign_events") < source.index("await conn.commit()")
+
+
+def test_transactional_method_rolls_back_on_error():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert "await conn.rollback()" in source
+
+
+def test_transactional_method_raises_lookup_error_for_unknown_id():
+    source = inspect.getsource(
+        repo_module.CampaignRepository.update_campaign_with_event,
+    )
+    assert "LookupError" in source

tests/test_campaign_sync.py

@@ -0,0 +1,237 @@
+# -*- coding: utf-8 -*-
+"""Syncing an approved campaign into Meta.
+
+No test here touches the network: the Meta client is always a fake that
+records the calls it was asked to make.
+"""
+import pytest
+
+from adclaw.campaigns.models import CampaignSpec
+from adclaw.campaigns.sync import SyncConfigurationError, sync_campaign
+from adclaw.meta.client import MetaError
+
+
+class FakeRepo:
+    """Records persisted state and the audit events written."""
+
+    def __init__(self):
+        self.events: list[str] = []
+        self.saved: CampaignSpec | None = None
+
+    async def update_campaign_with_event(
+        self, spec, event_type, actor=None, reason=None, payload=None,
+    ):
+        self.events.append(event_type)
+        self.saved = spec
+        return spec
+
+
+class FakeMeta:
+    """Records every create call, and can fail on a chosen step."""
+
+    def __init__(self, fail_on: str | None = None):
+        self.calls: list[dict] = []
+        self.fail_on = fail_on
+
+    def _maybe_fail(self, step: str):
+        if self.fail_on == step:
+            raise MetaError("Meta rejected this", code=100, subcode=1487079)
+
+    async def create_campaign(self, ad_account_id, **kwargs):
+        self._maybe_fail("campaign")
+        self.calls.append({"step": "campaign", "account": ad_account_id, **kwargs})
+        return "meta_camp_1"
+
+    async def create_ad_set(self, ad_account_id, **kwargs):
+        self._maybe_fail("adset")
+        self.calls.append({"step": "adset", "account": ad_account_id, **kwargs})
+        return "meta_adset_1"
+
+    async def upload_ad_image(self, ad_account_id, image_path):
+        self._maybe_fail("image")
+        self.calls.append({"step": "image", "path": image_path})
+        return "img_hash_1"
+
+    async def create_ad_creative(self, ad_account_id, **kwargs):
+        self._maybe_fail("creative")
+        self.calls.append({"step": "creative", **kwargs})
+        return "meta_creative_1"
+
+    async def create_ad(self, ad_account_id, **kwargs):
+        self._maybe_fail("ad")
+        self.calls.append({"step": "ad", **kwargs})
+        return "meta_ad_1"
+
+    def steps(self) -> list[str]:
+        return [c["step"] for c in self.calls]
+
+
+def _campaign(**overrides) -> CampaignSpec:
+    data = {
+        "id": "camp_1",
+        "name": "Q3 lead gen",
+        "status": "approved",
+        "objective": "OUTCOME_LEADS",
+        "ad_account_id": "act_1",
+        "budget": {"daily_budget": 10000},
+        # Graph requires an audience on every ad set, so every campaign that
+        # can be synced has one.
+        "targeting": {"countries": ["IN"]},
+        "advanced": {"page_id": "page_1", "link": "https://example.com"},
+    }
+    data.update(overrides)
+    return CampaignSpec(**data)
+
+
+@pytest.mark.asyncio
+async def test_happy_path_creates_full_chain_and_marks_synced():
+    repo, meta = FakeRepo(), FakeMeta()
+
+    result = await sync_campaign(repo, meta, _campaign(), actor="owner")
+
+    assert meta.steps() == ["campaign", "adset", "creative", "ad"]
+    assert result.status == "synced"
+    assert result.sync_status == "synced"
+    assert result.sync_error is None
+    assert result.meta_campaign_id == "meta_camp_1"
+    assert result.advanced["meta_sync"] == {
+        "adset_id": "meta_adset_1",
+        "creative_id": "meta_creative_1",
+        "ad_id": "meta_ad_1",
+    }
+    assert "campaign.sync" in repo.events
+
+
+@pytest.mark.asyncio
+async def test_sync_never_requests_a_non_paused_object():
+    """The money-safety property: nothing in the chain may be created live."""
+    repo, meta = FakeRepo(), FakeMeta()
+
+    await sync_campaign(repo, meta, _campaign(), actor="owner")
+
+    for call in meta.calls:
+        assert call.get("status", "PAUSED") == "PAUSED", call
+
+
+@pytest.mark.asyncio
+async def test_only_approved_campaigns_may_sync():
+    repo, meta = FakeRepo(), FakeMeta()
+
+    with pytest.raises(ValueError):
+        await sync_campaign(repo, meta, _campaign(status="draft"), actor="owner")
+
+    assert meta.calls == []
+
+
+@pytest.mark.asyncio
+async def test_missing_page_id_fails_before_any_meta_call(monkeypatch):
+    monkeypatch.delenv("META_PAGE_ID", raising=False)
+    repo, meta = FakeRepo(), FakeMeta()
+    campaign = _campaign(advanced={"link": "https://example.com"})
+
+    with pytest.raises(SyncConfigurationError):
+        await sync_campaign(repo, meta, campaign, actor="owner")
+
+    assert meta.calls == []
+
+
+@pytest.mark.asyncio
+async def test_page_id_falls_back_to_environment(monkeypatch):
+    monkeypatch.setenv("META_PAGE_ID", "page_from_env")
+    repo, meta = FakeRepo(), FakeMeta()
+    campaign = _campaign(advanced={"link": "https://example.com"})
+
+    await sync_campaign(repo, meta, campaign, actor="owner")
+
+    creative = next(c for c in meta.calls if c["step"] == "creative")
+    assert creative["page_id"] == "page_from_env"
+
+
+@pytest.mark.asyncio
+async def test_missing_ad_account_fails_before_any_meta_call():
+    repo, meta = FakeRepo(), FakeMeta()
+
+    with pytest.raises(SyncConfigurationError):
+        await sync_campaign(repo, meta, _campaign(ad_account_id=None), actor="owner")
+
+    assert meta.calls == []
+
+
+@pytest.mark.asyncio
+async def test_failure_mid_chain_records_error_and_keeps_earlier_ids():
+    repo, meta = FakeRepo(), FakeMeta(fail_on="adset")
+    campaign = _campaign()
+
+    with pytest.raises(MetaError):
+        await sync_campaign(repo, meta, campaign, actor="owner")
+
+    assert campaign.sync_status == "failed"
+    assert "Meta rejected this" in campaign.sync_error
+    assert "code=100" in campaign.sync_error
+    assert "subcode=1487079" in campaign.sync_error
+    # Status must NOT advance: the chain is incomplete.
+    assert campaign.status == "approved"
+    # The campaign id obtained before the failure is kept, so the retry
+    # reuses it instead of creating a second campaign in the account.
+    assert campaign.meta_campaign_id == "meta_camp_1"
+    assert "campaign.sync_failed" in repo.events
+
+
+@pytest.mark.asyncio
+async def test_retry_after_partial_failure_does_not_duplicate_objects():
+    repo, meta = FakeRepo(), FakeMeta(fail_on="adset")
+    campaign = _campaign()
+
+    with pytest.raises(MetaError):
+        await sync_campaign(repo, meta, campaign, actor="owner")
+
+    # Second attempt with a healthy Meta.
+    repo2, meta2 = FakeRepo(), FakeMeta()
+    result = await sync_campaign(repo2, meta2, campaign, actor="owner")
+
+    # create_campaign must NOT be called again — the id already exists.
+    assert "campaign" not in meta2.steps()
+    assert meta2.steps() == ["adset", "creative", "ad"]
+    assert result.status == "synced"
+    assert result.meta_campaign_id == "meta_camp_1"
+
+
+@pytest.mark.asyncio
+async def test_image_is_uploaded_only_when_a_path_is_configured():
+    repo, meta = FakeRepo(), FakeMeta()
+    campaign = _campaign(
+        advanced={"page_id": "page_1", "image_path": "C:/tmp/creative.jpg"},
+    )
+
+    await sync_campaign(repo, meta, campaign, actor="owner")
+
+    assert "image" in meta.steps()
+    creative = next(c for c in meta.calls if c["step"] == "creative")
+    assert creative["image_hash"] == "img_hash_1"
+    assert campaign.advanced["meta_sync"]["image_hash"] == "img_hash_1"
+
+
+@pytest.mark.asyncio
+async def test_no_image_path_skips_upload():
+    repo, meta = FakeRepo(), FakeMeta()
+
+    await sync_campaign(repo, meta, _campaign(), actor="owner")
+
+    assert "image" not in meta.steps()
+
+
+@pytest.mark.asyncio
+async def test_uploaded_image_hash_is_reused_on_retry():
+    repo, meta = FakeRepo(), FakeMeta(fail_on="creative")
+    campaign = _campaign(
+        advanced={"page_id": "page_1", "image_path": "C:/tmp/creative.jpg"},
+    )
+
+    with pytest.raises(MetaError):
+        await sync_campaign(repo, meta, campaign, actor="owner")
+
+    repo2, meta2 = FakeRepo(), FakeMeta()
+    await sync_campaign(repo2, meta2, campaign, actor="owner")
+
+    # The image must not be uploaded a second time.
+    assert "image" not in meta2.steps()

tests/test_campaign_sync_live.py

@@ -0,0 +1,129 @@
+# -*- coding: utf-8 -*-
+"""Sync one campaign into the real ad account, then delete it.
+
+Opt-in: set MASKANX_LIVE_TESTS=1. Every object is created PAUSED, so the
+test cannot spend money, and the objects are deleted in a finally block
+so the account is left as it was found.
+
+This exists because the unit tests all run against a fake transport. They
+prove the code sends what we think it sends; only this proves Meta accepts
+it and that what lands in the account is paused.
+"""
+from __future__ import annotations
+
+import os
+import sys
+import uuid
+
+import pytest
+
+from adclaw.campaigns.models import CampaignSpec
+from adclaw.campaigns.sync import sync_campaign, unsync_campaign
+from adclaw.envs.store import load_envs_into_environ
+from adclaw.meta.client import CREATE_STATUS, MetaClient, access_token_from_env
+
+if sys.platform == "win32":
+    import asyncio
+
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+LIVE = os.environ.get("MASKANX_LIVE_TESTS") == "1"
+AD_ACCOUNT_ENV = "MASKANX_LIVE_AD_ACCOUNT_ID"
+
+pytestmark = pytest.mark.skipif(
+    not LIVE,
+    reason=(
+        "Live Meta tests are opt-in. Set MASKANX_LIVE_TESTS=1 and "
+        f"{AD_ACCOUNT_ENV}=act_... to run them. They create and delete "
+        "real (paused) objects in that ad account."
+    ),
+)
+
+
+class _MemoryRepo:
+    """Stands in for the database so the live test needs only Meta."""
+
+    def __init__(self):
+        self.events: list[str] = []
+
+    async def update_campaign_with_event(
+        self, spec, event_type, actor=None, reason=None, payload=None,
+    ):
+        self.events.append(event_type)
+        return spec
+
+
+def _live_campaign(ad_account_id: str) -> CampaignSpec:
+    return CampaignSpec(
+        id=f"live_{uuid.uuid4().hex[:12]}",
+        name=f"MaskanX live smoke {uuid.uuid4().hex[:6]}",
+        status="approved",
+        origin="maskanx",
+        objective="OUTCOME_TRAFFIC",
+        ad_account_id=ad_account_id,
+        approved_by="live-smoke-test",
+        # Comfortably above the account minimum; never spent, since the
+        # objects stay paused for their whole (short) life.
+        budget={"daily_budget": 50000},
+        targeting={"countries": ["IN"], "age_min": 25, "age_max": 55},
+        advanced={
+            "message": "MaskanX live smoke test. Paused; delete on sight.",
+            "headline": "MaskanX smoke test",
+            "link": "https://maskan.technology",
+        },
+        channels=["facebook"],
+    )
+
+
+@pytest.fixture()
+def live_account() -> str:
+    load_envs_into_environ()
+    account = (os.environ.get(AD_ACCOUNT_ENV) or "").strip()
+    if not account:
+        pytest.skip(f"{AD_ACCOUNT_ENV} is not set.")
+    return account
+
+
+async def test_sync_creates_paused_objects_and_delete_removes_them(live_account):
+    meta = MetaClient(access_token=access_token_from_env())
+    campaign = _live_campaign(live_account)
+
+    synced = None
+    try:
+        synced = await sync_campaign(
+            _MemoryRepo(), meta, campaign, actor="live-smoke-test",
+        )
+
+        assert synced.sync_status == "synced"
+        assert synced.status == "synced"
+        assert synced.meta_campaign_id
+
+        state = synced.advanced["meta_sync"]
+        assert state["adset_id"] and state["creative_id"] and state["ad_id"]
+
+        # What actually landed in the account is what matters, so read it
+        # back from Graph rather than trusting the create responses.
+        for object_id in (
+            synced.meta_campaign_id,
+            state["adset_id"],
+            state["ad_id"],
+        ):
+            live = await meta._get(f"/{object_id}", {"fields": "id,status"})
+            assert live["status"] == CREATE_STATUS, (
+                f"{object_id} is {live['status']}, not {CREATE_STATUS} — "
+                f"this object could be spending money."
+            )
+    finally:
+        if synced and synced.meta_campaign_id:
+            await unsync_campaign(meta, synced)
+
+    # Deleting the campaign cascades, so nothing from this test should be
+    # left anywhere in the account.
+    remaining = {c["id"] for c in await meta.list_campaigns(live_account)}
+    assert synced.meta_campaign_id not in remaining
+
+    ad_sets = {a["id"] for a in await meta.list_ad_sets(live_account)}
+    assert synced.advanced["meta_sync"]["adset_id"] not in ad_sets
+
+    ads = {a["id"] for a in await meta.list_ads(live_account)}
+    assert synced.advanced["meta_sync"]["ad_id"] not in ads

tests/test_campaign_validation.py

@@ -30,6 +30,25 @@ def test_daily_and_lifetime_budget_together_is_rejected():
     assert [i.field for i in issues] == ["budget"]
 
 
+def test_lifetime_only_budget_is_rejected_as_unsupported():
+    # create_ad_set only accepts daily_budget (objects.build_ad_set_payload
+    # raises ValueError for a lifetime-only spec), so a lifetime-only
+    # budget must be caught here rather than passing validation/approval
+    # and only failing once sync tries to create the Meta ad set.
+    issues = validate_budget({"lifetime_budget": 50000}, min_daily_budget=9709)
+    assert [i.field for i in issues] == ["budget.lifetime_budget"]
+    assert "not supported yet" in issues[0].message
+
+
+def test_lifetime_only_negative_budget_reports_both_issues():
+    issues = validate_budget({"lifetime_budget": -5}, min_daily_budget=9709)
+    fields = [i.field for i in issues]
+    assert fields == ["budget.lifetime_budget", "budget.lifetime_budget"]
+    messages = " ".join(i.message for i in issues)
+    assert "greater than zero" in messages
+    assert "not supported yet" in messages
+
+
 def test_auto_pause_above_daily_budget_is_rejected():
     issues = validate_guardrails(
         {"daily_budget": 10000},

tests/test_meta_transport.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+"""The default HTTP transport. Still no live calls: httpx itself is faked.
+
+Every other Meta test injects a transport, so nothing exercised the real
+one. The property that matters here is that POST payloads go in the body:
+`upload_ad_image` base64-encodes a whole image file, which cannot fit in a
+query string for any real image.
+"""
+import httpx
+import pytest
+
+from adclaw.meta import client as meta_client
+
+
+class _Recorder:
+    """Stands in for httpx.AsyncClient, capturing where params landed."""
+
+    def __init__(self, calls, **kwargs):
+        self.calls = calls
+        self.init_kwargs = kwargs
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    def _response(self):
+        return httpx.Response(200, json={"id": "obj_1"})
+
+    async def post(self, url, data=None):
+        self.calls.append({"method": "POST", "url": url, "data": data})
+        return self._response()
+
+    async def request(self, method, url, params=None):
+        self.calls.append({"method": method, "url": url, "params": params})
+        return self._response()
+
+
+@pytest.fixture()
+def calls(monkeypatch):
+    recorded: list[dict] = []
+    monkeypatch.setattr(
+        httpx,
+        "AsyncClient",
+        lambda **kwargs: _Recorder(recorded, **kwargs),
+    )
+    return recorded
+
+
+async def test_post_sends_its_payload_in_the_body(calls):
+    """A base64 image cannot fit in a URL, so POST must not use the query."""
+    await meta_client._httpx_transport(
+        "POST", "https://graph.facebook.com/v23.0/act_1/adimages", {"bytes": "x" * 5000},
+    )
+
+    assert calls[0]["method"] == "POST"
+    assert calls[0]["data"] == {"bytes": "x" * 5000}
+    assert "?" not in calls[0]["url"]
+
+
+async def test_get_keeps_its_params_in_the_query_string(calls):
+    await meta_client._httpx_transport(
+        "GET", "https://graph.facebook.com/v23.0/act_1", {"fields": "id"},
+    )
+
+    assert calls[0] == {
+        "method": "GET",
+        "url": "https://graph.facebook.com/v23.0/act_1",
+        "params": {"fields": "id"},
+    }
+
+
+async def test_delete_keeps_its_params_in_the_query_string(calls):
+    await meta_client._httpx_transport(
+        "DELETE", "https://graph.facebook.com/v23.0/obj_1", {"access_token": "t"},
+    )
+
+    assert calls[0]["method"] == "DELETE"
+    assert calls[0]["params"] == {"access_token": "t"}
+
+
+async def test_a_non_json_response_becomes_a_readable_error(monkeypatch):
+    class _Html(_Recorder):
+        def _response(self):
+            return httpx.Response(502, text="<html>bad gateway</html>")
+
+    monkeypatch.setattr(httpx, "AsyncClient", lambda **kwargs: _Html([], **kwargs))
+
+    with pytest.raises(meta_client.MetaError, match="non-JSON"):
+        await meta_client._httpx_transport("GET", "https://graph.facebook.com/x", {})

tests/test_meta_writes.py

@@ -0,0 +1,561 @@
+# -*- coding: utf-8 -*-
+"""Meta write helpers. No live calls: the transport is injected.
+
+Covers the money-safety properties Task 4 exists to enforce:
+  1. Every create_* helper hard-codes status="PAUSED" and raises ValueError
+     (with zero network calls) if asked for anything else.
+  2. update_object_status is the only method allowed to send ACTIVE.
+  3. Dict params Graph expects as JSON strings (targeting,
+     object_story_spec, special_ad_categories) are json.dumps-encoded.
+  4. Meta's code/error_subcode survive on MetaError.
+  5. No test makes a live network call or creates a real Meta object -
+     every client here is constructed with a FakeTransport.
+"""
+import base64
+import json
+
+import pytest
+
+from adclaw.meta.client import MetaClient, MetaError
+from adclaw.meta.objects import (
+    build_ad_set_payload,
+    build_campaign_payload,
+    build_creative_payload,
+)
+from adclaw.campaigns.models import CampaignSpec
+
+
+class FakeTransport:
+    """Records calls and returns queued responses. Never touches a network."""
+
+    def __init__(self, responses):
+        self.responses = list(responses)
+        self.calls = []
+
+    async def __call__(self, method, url, params):
+        self.calls.append({"method": method, "url": url, "params": params})
+        return self.responses.pop(0)
+
+
+# ---------------------------------------------------------------------------
+# create_campaign
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_create_campaign_is_always_paused():
+    transport = FakeTransport([{"id": "23851234567890"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    campaign_id = await client.create_campaign(
+        "act_1", name="Q3", objective="OUTCOME_LEADS",
+    )
+
+    assert campaign_id == "23851234567890"
+    call = transport.calls[0]
+    assert call["method"] == "POST"
+    assert call["url"].endswith("/act_1/campaigns")
+    assert call["params"]["status"] == "PAUSED"
+    assert call["params"]["objective"] == "OUTCOME_LEADS"
+
+
+@pytest.mark.asyncio
+async def test_create_campaign_refuses_active_status():
+    transport = FakeTransport([{"id": "1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    with pytest.raises(ValueError):
+        await client.create_campaign(
+            "act_1", name="Q3", objective="OUTCOME_LEADS", status="ACTIVE",
+        )
+    assert transport.calls == []
+
+
+@pytest.mark.asyncio
+async def test_create_campaign_serialises_special_ad_categories():
+    transport = FakeTransport([{"id": "c1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    await client.create_campaign(
+        "act_1",
+        name="Q3",
+        objective="OUTCOME_LEADS",
+        special_ad_categories=["HOUSING"],
+    )
+
+    sent = transport.calls[0]["params"]["special_ad_categories"]
+    assert isinstance(sent, str)
+    assert json.loads(sent) == ["HOUSING"]
+
+
+# ---------------------------------------------------------------------------
+# create_ad_set
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_create_ad_set_sends_paused_and_serialises_targeting():
+    transport = FakeTransport([{"id": "adset_1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    await client.create_ad_set(
+        "act_1",
+        campaign_id="c1",
+        name="Ad set",
+        daily_budget=10000,
+        targeting={"geo_locations": {"countries": ["IN"]}},
+        optimization_goal="LEAD_GENERATION",
+        billing_event="IMPRESSIONS",
+    )
+
+    params = transport.calls[0]["params"]
+    assert params["status"] == "PAUSED"
+    assert params["daily_budget"] == 10000
+    assert isinstance(params["targeting"], str)
+    assert json.loads(params["targeting"]) == {
+        "geo_locations": {"countries": ["IN"]},
+    }
+
+
+@pytest.mark.asyncio
+async def test_create_ad_set_refuses_active_status():
+    transport = FakeTransport([{"id": "adset_1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    with pytest.raises(ValueError):
+        await client.create_ad_set(
+            "act_1",
+            campaign_id="c1",
+            name="Ad set",
+            daily_budget=10000,
+            targeting={},
+            optimization_goal="LEAD_GENERATION",
+            billing_event="IMPRESSIONS",
+            status="ACTIVE",
+        )
+    assert transport.calls == []
+
+
+# ---------------------------------------------------------------------------
+# create_ad
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_create_ad_is_always_paused():
+    transport = FakeTransport([{"id": "ad_1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    ad_id = await client.create_ad(
+        "act_1", name="Ad", adset_id="adset_1", creative_id="creative_1",
+    )
+
+    assert ad_id == "ad_1"
+    params = transport.calls[0]["params"]
+    assert params["status"] == "PAUSED"
+    assert params["adset_id"] == "adset_1"
+    assert json.loads(params["creative"]) == {"creative_id": "creative_1"}
+
+
+@pytest.mark.asyncio
+async def test_create_ad_refuses_active_status():
+    transport = FakeTransport([{"id": "ad_1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    with pytest.raises(ValueError):
+        await client.create_ad(
+            "act_1",
+            name="Ad",
+            adset_id="adset_1",
+            creative_id="creative_1",
+            status="ACTIVE",
+        )
+    assert transport.calls == []
+
+
+# ---------------------------------------------------------------------------
+# create_ad_creative
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_create_ad_creative_serialises_object_story_spec():
+    transport = FakeTransport([{"id": "creative_1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    creative_id = await client.create_ad_creative(
+        "act_1",
+        name="Creative",
+        page_id="page_1",
+        message="Hello",
+        headline="Headline",
+        description="Description",
+        link="https://example.com",
+        image_hash="hash123",
+    )
+
+    assert creative_id == "creative_1"
+    sent = transport.calls[0]["params"]["object_story_spec"]
+    assert isinstance(sent, str)
+    decoded = json.loads(sent)
+    assert decoded["page_id"] == "page_1"
+    assert decoded["link_data"]["image_hash"] == "hash123"
+    assert decoded["link_data"]["message"] == "Hello"
+
+
+# ---------------------------------------------------------------------------
+# upload_ad_image
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_upload_ad_image_encodes_bytes_and_returns_hash(tmp_path):
+    image_path = tmp_path / "creative.png"
+    raw_bytes = b"\x89PNG\r\n\x1a\nfake-image-bytes"
+    image_path.write_bytes(raw_bytes)
+
+    transport = FakeTransport(
+        [{"images": {"creative.png": {"hash": "abc123", "url": "https://x"}}}],
+    )
+    client = MetaClient(access_token="tok", transport=transport)
+
+    image_hash = await client.upload_ad_image("act_1", str(image_path))
+
+    assert image_hash == "abc123"
+    call = transport.calls[0]
+    assert call["method"] == "POST"
+    assert call["url"].endswith("/act_1/adimages")
+    assert base64.b64decode(call["params"]["bytes"]) == raw_bytes
+
+
+@pytest.mark.asyncio
+async def test_upload_ad_image_raises_meta_error_when_no_hash(tmp_path):
+    image_path = tmp_path / "creative.png"
+    image_path.write_bytes(b"data")
+
+    transport = FakeTransport([{"images": {}}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    with pytest.raises(MetaError):
+        await client.upload_ad_image("act_1", str(image_path))
+
+
+# ---------------------------------------------------------------------------
+# _post error handling
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_post_error_preserves_meta_code_and_subcode():
+    transport = FakeTransport([
+        {"error": {"message": "Invalid budget", "code": 100, "error_subcode": 1487079}},
+    ])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    with pytest.raises(MetaError) as excinfo:
+        await client.create_campaign("act_1", name="Q3", objective="OUTCOME_LEADS")
+
+    assert excinfo.value.code == 100
+    assert excinfo.value.subcode == 1487079
+    assert "Invalid budget" in str(excinfo.value)
+
+
+# ---------------------------------------------------------------------------
+# update_object_status - the only path allowed to send ACTIVE
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_update_object_status_accepts_active():
+    """Launch is the one path allowed to send ACTIVE."""
+    transport = FakeTransport([{"success": True}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    await client.update_object_status("c1", "ACTIVE")
+
+    call = transport.calls[0]
+    assert call["method"] == "POST"
+    assert call["url"].endswith("/c1")
+    assert call["params"]["status"] == "ACTIVE"
+
+
+@pytest.mark.asyncio
+async def test_update_object_status_returns_none():
+    transport = FakeTransport([{"success": True}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    result = await client.update_object_status("c1", "PAUSED")
+
+    assert result is None
+
+
+# ---------------------------------------------------------------------------
+# _post defence in depth - reachable directly, still refuses non-PAUSED
+# creates on campaign/adset/ad paths, before any transport call
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_post_refuses_non_paused_status_on_campaign_create_path():
+    transport = FakeTransport([{"id": "c1"}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    with pytest.raises(ValueError):
+        await client._post(
+            "/act_1/campaigns",
+            {"name": "Q3", "objective": "OUTCOME_LEADS", "status": "ACTIVE"},
+        )
+    assert transport.calls == []
+
+
+@pytest.mark.asyncio
+async def test_post_still_allows_update_object_status_to_send_active():
+    """update_object_status posts to /{object_id}, which never matches the
+    campaign/adset/ad creation suffixes, so the _post guard above must not
+    block Launch."""
+    transport = FakeTransport([{"success": True}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    await client.update_object_status("c1", "ACTIVE")
+
+    assert transport.calls[0]["params"]["status"] == "ACTIVE"
+
+
+# ---------------------------------------------------------------------------
+# list_campaigns / list_ad_sets / list_ads
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_list_campaigns_returns_data_array():
+    transport = FakeTransport([{"data": [{"id": "c1", "name": "X", "status": "PAUSED"}]}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    result = await client.list_campaigns("act_1")
+
+    assert result[0]["id"] == "c1"
+    assert transport.calls[0]["method"] == "GET"
+    assert transport.calls[0]["url"].endswith("/act_1/campaigns")
+
+
+@pytest.mark.asyncio
+async def test_list_campaigns_returns_empty_list_when_no_data():
+    transport = FakeTransport([{}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    assert await client.list_campaigns("act_1") == []
+
+
+@pytest.mark.asyncio
+async def test_list_ad_sets_returns_data_array():
+    transport = FakeTransport([{"data": [{"id": "as1"}]}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    result = await client.list_ad_sets("act_1")
+
+    assert result == [{"id": "as1"}]
+    assert transport.calls[0]["method"] == "GET"
+    assert transport.calls[0]["url"].endswith("/act_1/adsets")
+
+
+@pytest.mark.asyncio
+async def test_list_ads_returns_data_array():
+    transport = FakeTransport([{"data": [{"id": "ad1"}]}])
+    client = MetaClient(access_token="tok", transport=transport)
+
+    result = await client.list_ads("act_1")
+
+    assert result == [{"id": "ad1"}]
+    assert transport.calls[0]["method"] == "GET"
+    assert transport.calls[0]["url"].endswith("/act_1/ads")
+
+
+# ---------------------------------------------------------------------------
+# objects.py - pure payload builders, no transport at all
+# ---------------------------------------------------------------------------
+
+
+def _spec(**overrides) -> CampaignSpec:
+    defaults = dict(
+        id="camp_1",
+        name="Test Campaign",
+        objective="OUTCOME_LEADS",
+        budget={"daily_budget": 5000},
+        targeting={"age_min": 25, "age_max": 45, "countries": ["US", "CA"]},
+        advanced={},
+    )
+    defaults.update(overrides)
+    return CampaignSpec(**defaults)
+
+
+def test_build_campaign_payload_maps_fields():
+    spec = _spec(advanced={"special_ad_categories": ["HOUSING"]})
+
+    payload = build_campaign_payload(spec)
+
+    assert payload == {
+        "name": "Test Campaign",
+        "objective": "OUTCOME_LEADS",
+        "special_ad_categories": ["HOUSING"],
+    }
+
+
+def test_build_campaign_payload_defaults_special_ad_categories_to_empty():
+    spec = _spec()
+
+    payload = build_campaign_payload(spec)
+
+    assert payload["special_ad_categories"] == []
+
+
+def test_build_campaign_payload_requires_objective():
+    spec = _spec(objective=None)
+
+    with pytest.raises(ValueError):
+        build_campaign_payload(spec)
+
+
+def test_build_campaign_payload_normalises_bare_string_special_ad_category():
+    # advanced is unvalidated free-form JSON, so a bare string is a
+    # plausible client input for this compliance field. It must become a
+    # single-element list, not be exploded into characters by list().
+    spec = _spec(advanced={"special_ad_categories": "HOUSING"})
+
+    payload = build_campaign_payload(spec)
+
+    assert payload["special_ad_categories"] == ["HOUSING"]
+
+
+def test_build_campaign_payload_passes_through_a_list_unchanged():
+    spec = _spec(advanced={"special_ad_categories": ["HOUSING", "EMPLOYMENT"]})
+
+    payload = build_campaign_payload(spec)
+
+    assert payload["special_ad_categories"] == ["HOUSING", "EMPLOYMENT"]
+
+
+def test_build_ad_set_payload_maps_targeting_and_budget():
+    spec = _spec()
+
+    payload = build_ad_set_payload(spec, campaign_id="c1")
+
+    assert payload["campaign_id"] == "c1"
+    assert payload["daily_budget"] == 5000
+    assert payload["targeting"] == {
+        "age_min": 25,
+        "age_max": 45,
+        "geo_locations": {"countries": ["US", "CA"]},
+    }
+    assert payload["optimization_goal"] == "LEAD_GENERATION"
+    assert payload["billing_event"] == "IMPRESSIONS"
+
+
+def test_build_ad_set_payload_honours_advanced_overrides():
+    spec = _spec(
+        advanced={
+            "optimization_goal": "OFFSITE_CONVERSIONS",
+            "billing_event": "LINK_CLICKS",
+        },
+    )
+
+    payload = build_ad_set_payload(spec, campaign_id="c1")
+
+    assert payload["optimization_goal"] == "OFFSITE_CONVERSIONS"
+    assert payload["billing_event"] == "LINK_CLICKS"
+
+
+def test_build_ad_set_payload_requires_daily_budget():
+    spec = _spec(budget={})
+
+    with pytest.raises(ValueError):
+        build_ad_set_payload(spec, campaign_id="c1")
+
+
+def test_build_ad_set_payload_requires_an_audience():
+    """Graph rejects an ad set with no targeting; say so here instead."""
+    spec = _spec(targeting={})
+
+    with pytest.raises(ValueError, match="no audience"):
+        build_ad_set_payload(spec, campaign_id="c1")
+
+
+def test_build_ad_set_payload_accepts_any_single_targeting_field():
+    payload = build_ad_set_payload(_spec(targeting={"countries": ["IN"]}), "c1")
+
+    assert payload["targeting"] == {"geo_locations": {"countries": ["IN"]}}
+
+
+def test_build_ad_set_payload_accepts_allowlisted_optimization_goal():
+    spec = _spec(advanced={"optimization_goal": "LINK_CLICKS"})
+
+    payload = build_ad_set_payload(spec, campaign_id="c1")
+
+    assert payload["optimization_goal"] == "LINK_CLICKS"
+
+
+def test_build_ad_set_payload_rejects_unknown_optimization_goal():
+    spec = _spec(advanced={"optimization_goal": "SOMETHING_MADE_UP"})
+
+    with pytest.raises(ValueError):
+        build_ad_set_payload(spec, campaign_id="c1")
+
+
+def test_build_ad_set_payload_accepts_allowlisted_billing_event():
+    spec = _spec(advanced={"billing_event": "LINK_CLICKS"})
+
+    payload = build_ad_set_payload(spec, campaign_id="c1")
+
+    assert payload["billing_event"] == "LINK_CLICKS"
+
+
+def test_build_ad_set_payload_rejects_unknown_billing_event():
+    # billing_event determines how the ad account is charged, so an
+    # unrecognised value from client-controlled `advanced` must never reach
+    # Meta - a spending-relevant field with no allowlist is the finding
+    # this guards against.
+    spec = _spec(advanced={"billing_event": "SOMETHING_MADE_UP"})
+
+    with pytest.raises(ValueError):
+        build_ad_set_payload(spec, campaign_id="c1")
+
+
+def test_build_ad_set_payload_targeting_can_produce_json_dumpable_dict():
+    # This is what create_ad_set will json.dumps() before sending - make
+    # sure the mapping never emits anything that would break that step.
+    spec = _spec()
+
+    payload = build_ad_set_payload(spec, campaign_id="c1")
+
+    encoded = json.dumps(payload["targeting"])
+    assert json.loads(encoded) == payload["targeting"]
+
+
+def test_build_creative_payload_maps_fields():
+    spec = _spec(
+        advanced={
+            "message": "Come see our homes",
+            "headline": "New Listings",
+            "description": "Fresh inventory weekly",
+            "link": "https://example.com/listings",
+        },
+    )
+
+    payload = build_creative_payload(spec, page_id="page_1", image_hash="hash123")
+
+    assert payload == {
+        "name": "Test Campaign - Creative",
+        "page_id": "page_1",
+        "message": "Come see our homes",
+        "headline": "New Listings",
+        "description": "Fresh inventory weekly",
+        "link": "https://example.com/listings",
+        "image_hash": "hash123",
+    }
+
+
+def test_build_creative_payload_falls_back_to_campaign_name_for_headline():
+    spec = _spec(advanced={})
+
+    payload = build_creative_payload(spec, page_id="page_1", image_hash="hash123")
+
+    assert payload["headline"] == "Test Campaign"