Resolve geos via typeahead; drop dead targeting/payload helpers

pyproject.toml

@@ -31,6 +31,8 @@ meta = [
 linkedin = [
     "mcp>=1.2",
     "httpx>=0.27",
+    # ISO 3166 country data for resolving brief geo codes to LinkedIn locations.
+    "pycountry>=23",
 ]
 agent = [
     "langchain>=0.3",

src/yieldagent/agents/campaign_setup/nodes.py

@@ -111,7 +111,6 @@ def human_gate(state: AgentState) -> dict[str, Any]:
     reason = decision.get("reason", "")
     return {
         "approved": approved,
-        "rejection_reason": "" if approved else reason,
         "audit": _audit(
             state,
             AuditEntry(

src/yieldagent/agents/campaign_setup/state.py

@@ -24,6 +24,5 @@ class AgentState(TypedDict, total=False):
     brief: Brief
     campaign: Campaign
     approved: bool
-    rejection_reason: str
     publish_result: dict[str, Any]
     audit: list[AuditEntry]

src/yieldagent/integrations/linkedin/mapping.py

@@ -20,7 +20,9 @@
 from datetime import datetime, time, timezone
 from typing import Any
 
-from yieldagent.domain import Audience, Campaign, CreativeAsset, Flight, LineItem, Objective
+import pycountry
+
+from yieldagent.domain import Audience, Campaign, CreativeAsset, Flight, Objective
 
 # LinkedIn objectiveType values for Sponsored Content campaigns.
 OBJECTIVE_TO_LINKEDIN: dict[Objective, str] = {
@@ -32,27 +34,6 @@
     Objective.sales: "WEBSITE_CONVERSIONS",
 }
 
-# LinkedIn requires geo URNs (urn:li:geo:{id}) — ISO codes are not accepted.
-# Production use should resolve via the geo typeahead endpoint; the few entries
-# below cover the common-case countries so simple briefs work out of the box.
-ISO_TO_LINKEDIN_GEO_URN: dict[str, str] = {
-    "US": "urn:li:geo:103644278",
-    "GB": "urn:li:geo:101165590",
-    "CA": "urn:li:geo:101174742",
-    "DE": "urn:li:geo:101282230",
-    "FR": "urn:li:geo:105015875",
-    "AU": "urn:li:geo:101452733",
-    "IN": "urn:li:geo:102713980",
-    "IL": "urn:li:geo:101620260",
-    "BR": "urn:li:geo:106057199",
-    "JP": "urn:li:geo:101355337",
-    "NL": "urn:li:geo:102890719",
-    "ES": "urn:li:geo:105646813",
-    "IT": "urn:li:geo:103350119",
-    "SE": "urn:li:geo:105117694",
-    "SG": "urn:li:geo:102454443",
-}
-
 # LinkedIn campaign type for standard image / single-share Sponsored Content.
 DEFAULT_CAMPAIGN_TYPE = "SPONSORED_UPDATES"
 
@@ -86,68 +67,18 @@ def campaign_run_schedule(flights: list[Flight]) -> dict[str, int]:
     return flight_to_run_schedule(Flight(start_date=earliest, end_date=latest))
 
 
-def audience_to_targeting(audience: Audience) -> dict[str, Any]:
-    """Build a geo-only LinkedIn `targetingCriteria` payload.
-
-    This is the static, client-free baseline (locations only). Live B2B facet
-    resolution — industries, job functions, titles, seniorities, company sizes,
-    skills — needs API lookups and lives in `targeting.TargetingResolver`, which
-    the publish flow uses. This helper remains for inspection/test scaffolding.
-    """
-    includes: list[str] = []
-    for code in audience.geos:
-        urn = ISO_TO_LINKEDIN_GEO_URN.get(code.upper())
-        if urn:
-            includes.append(urn)
-    if not includes:
-        # LinkedIn requires at least one location; default to US.
-        includes.append(ISO_TO_LINKEDIN_GEO_URN["US"])
-
-    return {
-        "include": {
-            "and": [
-                {
-                    "or": {
-                        "urn:li:adTargetingFacet:locations": includes,
-                    }
-                }
-            ]
-        }
-    }
-
-
 def line_item_locale(audience: Audience) -> dict[str, str]:
     """LinkedIn campaigns require a `locale` (country + language).
 
-    Derived from the first audience geo if available; defaults to en/US.
+    Derived from the first audience geo if it is a valid ISO 3166-1 alpha-2
+    code; defaults to en/US otherwise.
     """
-    country = (audience.geos[0].upper() if audience.geos else "US")
-    if country not in ISO_TO_LINKEDIN_GEO_URN:
+    country = audience.geos[0].upper() if audience.geos else "US"
+    if pycountry.countries.get(alpha_2=country) is None:
         country = "US"
     return {"country": country, "language": "en"}
 
 
-def line_item_payload(
-    line_item: LineItem,
-    *,
-    campaign_group_urn: str,
-    objective_type: str,
-) -> dict[str, Any]:
-    """Strict-typed snapshot of the create_campaign call for testing/inspection."""
-    return {
-        "campaignGroup": campaign_group_urn,
-        "name": line_item.name,
-        "objectiveType": objective_type,
-        "type": DEFAULT_CAMPAIGN_TYPE,
-        "totalBudget": money_to_linkedin_amount(
-            line_item.budget.amount, line_item.budget.currency
-        ),
-        "runSchedule": flight_to_run_schedule(line_item.flight),
-        "targetingCriteria": audience_to_targeting(line_item.targeting.audience),
-        "locale": line_item_locale(line_item.targeting.audience),
-    }
-
-
 def post_article_content(creative: CreativeAsset) -> dict[str, Any]:
     """Build the `article` block for a Posts API dark post.
 
@@ -181,15 +112,12 @@ def creative_content_reference(post_urn: str) -> dict[str, Any]:
 
 __all__ = [
     "DEFAULT_CAMPAIGN_TYPE",
-    "ISO_TO_LINKEDIN_GEO_URN",
     "OBJECTIVE_TO_LINKEDIN",
-    "audience_to_targeting",
     "campaign_objective",
     "campaign_run_schedule",
     "creative_content_reference",
     "flight_to_run_schedule",
     "line_item_locale",
-    "line_item_payload",
     "money_to_linkedin_amount",
     "post_article_content",
     "post_commentary",

src/yieldagent/integrations/linkedin/targeting.py

@@ -11,9 +11,11 @@
     and match names against it. URNs are `urn:li:seniority:{id}` / `urn:li:function:{id}`.
   * Open taxonomies — industries, titles, skills — are resolved via the
     typeahead finder, which returns `{urn, name}` ranked by relevance.
-  * company_sizes — a fixed 9-bucket enum whose targeting values are plain
-    strings (`SIZE_11_TO_50`, ...), mapped statically.
-  * locations — geo URNs from the static ISO map in `mapping`.
+  * company_sizes — a fixed 9-bucket enum whose targeting values are range
+    tuple URNs (`urn:li:staffCountRange:(min,max)`), mapped statically.
+  * locations — the brief carries ISO 3166-1 alpha-2 codes; each is expanded to
+    its country name via `pycountry` and resolved through the same typeahead
+    finder, so any country works (not just a hardcoded shortlist).
 
 Names that resolve to no real LinkedIn URN are never guessed: they are collected
 in `ResolvedTargeting.unresolved` so the caller can surface them as a manual step
@@ -25,9 +27,9 @@
 from dataclasses import dataclass, field
 from typing import Any, Protocol
 
-from yieldagent.domain import Audience
+import pycountry
 
-from .mapping import ISO_TO_LINKEDIN_GEO_URN
+from yieldagent.domain import Audience
 
 FACET_LOCATIONS = "urn:li:adTargetingFacet:locations"
 FACET_SENIORITIES = "urn:li:adTargetingFacet:seniorities"
@@ -37,6 +39,11 @@
 FACET_SKILLS = "urn:li:adTargetingFacet:skills"
 FACET_STAFF_COUNT = "urn:li:adTargetingFacet:staffCountRanges"
 
+# LinkedIn requires at least one location. When a brief names no resolvable
+# country we fall back to this (United States) so the campaign is still valid;
+# the unresolved codes are still surfaced so the caller knows to fix them.
+DEFAULT_GEO_URN = "urn:li:geo:103644278"
+
 # Brief company-size buckets -> LinkedIn staffCountRange URNs. In targetingCriteria
 # the value is a range tuple URN (min,max), NOT the SIZE_* enum the facet listing
 # returns; the open-ended top bucket uses INT_MAX (2147483647) as its upper bound.
@@ -79,6 +86,19 @@ def _localized_name(entity: dict[str, Any]) -> str | None:
     return entity.get("name", {}).get("localized", {}).get("en_US")
 
 
+def _country_name(code: str) -> str | None:
+    """Map an ISO 3166-1 alpha-2 code to a country name LinkedIn will recognise.
+
+    Prefers the colloquial `common_name` (e.g. "South Korea" over "Korea,
+    Republic of") since that is what the locations typeahead indexes. Returns
+    None for codes `pycountry` does not know, so they surface as unresolved.
+    """
+    country = pycountry.countries.get(alpha_2=code.strip().upper())
+    if country is None:
+        return None
+    return getattr(country, "common_name", None) or country.name
+
+
 def _best_typeahead_match(query: str, hits: list[dict[str, Any]]) -> str | None:
     """Prefer an exact (normalized) name match; else LinkedIn's top-ranked hit.
 
@@ -163,21 +183,36 @@ def _resolve_company_sizes(sizes: list[str]) -> tuple[list[str], list[str]]:
                 unresolved.append(size)
         return values, unresolved
 
-    @staticmethod
-    def _geo_urns(audience: Audience) -> list[str]:
-        urns = [
-            ISO_TO_LINKEDIN_GEO_URN[code.upper()]
-            for code in audience.geos
-            if code.upper() in ISO_TO_LINKEDIN_GEO_URN
-        ]
-        # LinkedIn requires at least one location.
-        return urns or [ISO_TO_LINKEDIN_GEO_URN["US"]]
+    async def _resolve_geos(self, audience: Audience) -> tuple[list[str], list[str]]:
+        """Resolve ISO country codes to geo URNs via the locations typeahead.
+
+        Each code is expanded to a country name (`pycountry`) and looked up; a
+        code we cannot expand, or that the typeahead does not match, is returned
+        as unresolved. Falls back to the default location when nothing resolves,
+        since LinkedIn requires at least one.
+        """
+        urns: list[str] = []
+        unresolved: list[str] = []
+        for code in audience.geos:
+            name = _country_name(code)
+            urn = None
+            if name:
+                hits = await self._client.typeahead_targeting_entities(
+                    facet=FACET_LOCATIONS, query=name
+                )
+                urn = _best_typeahead_match(name, hits)
+            if urn:
+                urns.append(urn)
+            else:
+                unresolved.append(code)
+        return urns or [DEFAULT_GEO_URN], unresolved
 
     async def resolve(self, audience: Audience) -> ResolvedTargeting:
-        clauses: list[dict[str, Any]] = [
-            {"or": {FACET_LOCATIONS: self._geo_urns(audience)}}
-        ]
+        geo_urns, geo_unresolved = await self._resolve_geos(audience)
+        clauses: list[dict[str, Any]] = [{"or": {FACET_LOCATIONS: geo_urns}}]
         unresolved: dict[str, list[str]] = {}
+        if geo_unresolved:
+            unresolved["geos"] = geo_unresolved
 
         def _add(facet: str, urns: list[str], missing: list[str], key: str) -> None:
             if urns:
@@ -218,6 +253,7 @@ def _add(facet: str, urns: list[str], missing: list[str], key: str) -> None:
 
 __all__ = [
     "COMPANY_SIZE_TO_STAFF_RANGE",
+    "DEFAULT_GEO_URN",
     "FACET_INDUSTRIES",
     "FACET_JOB_FUNCTIONS",
     "FACET_LOCATIONS",

src/yieldagent/integrations/meta/mapping.py

@@ -10,7 +10,7 @@
 from decimal import Decimal
 from typing import Any
 
-from yieldagent.domain import Audience, Campaign, CreativeAsset, Flight, LineItem, Objective
+from yieldagent.domain import Audience, Campaign, CreativeAsset, Flight, Objective
 
 OBJECTIVE_TO_META: dict[Objective, str] = {
     Objective.awareness: "OUTCOME_AWARENESS",
@@ -67,20 +67,6 @@ def audience_to_targeting(audience: Audience) -> dict[str, Any]:
     return targeting
 
 
-def line_item_payload(line_item: LineItem, campaign_id: str) -> dict[str, Any]:
-    start, end = flight_to_meta_times(line_item.flight)
-    return {
-        "campaign_id": campaign_id,
-        "name": line_item.name,
-        "lifetime_budget_minor": to_minor_units(
-            line_item.budget.amount, line_item.budget.currency
-        ),
-        "start_time": start,
-        "end_time": end,
-        "targeting": audience_to_targeting(line_item.targeting.audience),
-    }
-
-
 def creative_payload(creative: CreativeAsset, page_id: str) -> dict[str, Any]:
     """Build a minimal link-ad creative.
 

tests/integrations/test_linkedin_publish.py

@@ -68,6 +68,13 @@ async def __aexit__(self, *_exc: object) -> None:
     def assert_account_allowed(self) -> None:
         pass
 
+    async def typeahead_targeting_entities(self, *, facet: str, query: str) -> list[dict]:
+        # The audience here is geo-only; resolve the locations typeahead and
+        # leave every other facet empty.
+        if facet.endswith("locations"):
+            return [{"urn": "urn:li:geo:103644278", "name": "United States"}]
+        return []
+
     async def get_ad_account(self) -> dict:
         return {"reference": _ORG_URN}
 

tests/integrations/test_linkedin_targeting.py

@@ -9,6 +9,7 @@
 
 from yieldagent.domain import Audience
 from yieldagent.integrations.linkedin.targeting import (
+    DEFAULT_GEO_URN,
     FACET_INDUSTRIES,
     FACET_JOB_FUNCTIONS,
     FACET_LOCATIONS,
@@ -20,6 +21,7 @@
 )
 
 _US_GEO = "urn:li:geo:103644278"
+_PT_GEO = "urn:li:geo:100364837"
 
 
 def _named(id_: int, name: str) -> dict:
@@ -35,6 +37,13 @@ def __init__(self) -> None:
         self.typeahead_calls: list[tuple[str, str]] = []
         # facet -> query(lowercased) -> hits
         self.typeahead: dict[str, dict[str, list[dict]]] = {
+            FACET_LOCATIONS: {
+                "united states": [{"urn": _US_GEO, "name": "United States"}],
+                "portugal": [
+                    {"urn": _PT_GEO, "name": "Portugal"},
+                    {"urn": "urn:li:geo:105374601", "name": "Porto, Portugal"},
+                ],
+            },
             FACET_INDUSTRIES: {
                 "advertising": [{"urn": "urn:li:industry:80", "name": "Advertising Services"}],
             },
@@ -76,10 +85,26 @@ def _clause_facets(criteria: dict) -> dict[str, list]:
 async def test_geo_only_defaults_to_us_when_empty() -> None:
     resolver = TargetingResolver(_FakeTargetingClient())
     resolved = await resolver.resolve(Audience(description="x", geos=[]))
-    assert _clause_facets(resolved.criteria) == {FACET_LOCATIONS: [_US_GEO]}
+    assert _clause_facets(resolved.criteria) == {FACET_LOCATIONS: [DEFAULT_GEO_URN]}
     assert resolved.unresolved == {}
 
 
+async def test_geos_resolve_via_typeahead_by_country_name() -> None:
+    resolver = TargetingResolver(_FakeTargetingClient())
+    resolved = await resolver.resolve(Audience(description="x", geos=["US", "PT"]))
+    # "PT" -> "Portugal" exact match wins over "Porto, Portugal".
+    assert _clause_facets(resolved.criteria) == {FACET_LOCATIONS: [_US_GEO, _PT_GEO]}
+    assert resolved.unresolved == {}
+
+
+async def test_unknown_geo_code_is_unresolved_not_guessed() -> None:
+    resolver = TargetingResolver(_FakeTargetingClient())
+    resolved = await resolver.resolve(Audience(description="x", geos=["ZZ"]))
+    # Invalid ISO code -> surfaced as unresolved; locations falls back to default.
+    assert _clause_facets(resolved.criteria) == {FACET_LOCATIONS: [DEFAULT_GEO_URN]}
+    assert resolved.unresolved == {"geos": ["ZZ"]}
+
+
 async def test_enum_facets_resolve_and_surface_misses() -> None:
     resolver = TargetingResolver(_FakeTargetingClient())
     resolved = await resolver.resolve(