diff --git a/crates/js/lib/build-all.mjs b/crates/js/lib/build-all.mjs
index cc5690e0..87ff479f 100644
--- a/crates/js/lib/build-all.mjs
+++ b/crates/js/lib/build-all.mjs
@@ -13,6 +13,10 @@
  *     names to include in the bundle (e.g. "rubicon,appnexus,openx").
  *     Each name must have a corresponding {name}BidAdapter.js module in
  *     the prebid.js package. Default: "rubicon".
+ *   TSJS_PREBID_USER_ID_MODULES — Comma-separated list of Prebid.js User ID
+ *     modules to include in the bundle (e.g. "sharedId,pubProvided,uid2").
+ *     Names may be module bases ("sharedId") or exact module names
+ *     ("sharedIdSystem"). Default: empty.
  */
 
 import fs from 'node:fs';
@@ -30,11 +34,10 @@ const integrationsDir = path.join(srcDir, 'integrations');
 // ---------------------------------------------------------------------------
 
 const DEFAULT_PREBID_ADAPTERS = 'rubicon';
-const ADAPTERS_FILE = path.join(
-  integrationsDir,
-  'prebid',
-  '_adapters.generated.ts',
-);
+const DEFAULT_PREBID_USER_ID_MODULES = '';
+const prebidIntegrationDir = path.join(integrationsDir, 'prebid');
+const ADAPTERS_FILE = path.join(prebidIntegrationDir, '_adapters.generated.ts');
+const USER_ID_MODULES_FILE = path.join(prebidIntegrationDir, '_user_id_modules.generated.ts');
 
 /**
  * Generate `_adapters.generated.ts` with import statements for each adapter
@@ -43,36 +46,34 @@ const ADAPTERS_FILE = path.join(
  * Invalid adapter names (those without a matching module in prebid.js) are
  * logged and skipped.
  */
-function generatePrebidAdapters() {
-  const raw = process.env.TSJS_PREBID_ADAPTERS || DEFAULT_PREBID_ADAPTERS;
-  const names = raw
+const prebidModulesDir = path.join(__dirname, 'node_modules', 'prebid.js', 'modules');
+
+function parseCommaSeparatedEnv(name, fallback) {
+  return (process.env[name] || fallback)
     .split(',')
     .map((s) => s.trim())
     .filter(Boolean);
+}
+
+function generatePrebidAdapters() {
+  const names = parseCommaSeparatedEnv('TSJS_PREBID_ADAPTERS', DEFAULT_PREBID_ADAPTERS);
 
   if (names.length === 0) {
     console.warn(
       '[build-all] TSJS_PREBID_ADAPTERS is empty, falling back to default:',
-      DEFAULT_PREBID_ADAPTERS,
+      DEFAULT_PREBID_ADAPTERS
     );
     names.push(DEFAULT_PREBID_ADAPTERS);
   }
 
-  const modulesDir = path.join(
-    __dirname,
-    'node_modules',
-    'prebid.js',
-    'modules',
-  );
-
   // Validate each adapter and build import lines
   const imports = [];
   for (const name of names) {
     const moduleFile = `${name}BidAdapter.js`;
-    const modulePath = path.join(modulesDir, moduleFile);
+    const modulePath = path.join(prebidModulesDir, moduleFile);
     if (!fs.existsSync(modulePath)) {
       console.error(
-        `[build-all] WARNING: Prebid adapter "${name}" not found (expected ${moduleFile}), skipping`,
+        `[build-all] WARNING: Prebid adapter "${name}" not found (expected ${moduleFile}), skipping`
       );
       continue;
     }
@@ -81,7 +82,7 @@ function generatePrebidAdapters() {
 
   if (imports.length === 0) {
     console.error(
-      '[build-all] WARNING: No valid Prebid adapters found, bundle will have no client-side adapters',
+      '[build-all] WARNING: No valid Prebid adapters found, bundle will have no client-side adapters'
     );
   }
 
@@ -94,18 +95,77 @@ function generatePrebidAdapters() {
     `// Default: "${DEFAULT_PREBID_ADAPTERS}"`,
     '',
     ...imports,
-    '',
   ].join('\n');
 
-  fs.writeFileSync(ADAPTERS_FILE, content);
+  fs.writeFileSync(ADAPTERS_FILE, `${content}\n`);
 
   const adapterNames = names.filter((name) =>
-    fs.existsSync(path.join(modulesDir, `${name}BidAdapter.js`)),
+    fs.existsSync(path.join(prebidModulesDir, `${name}BidAdapter.js`))
   );
   console.log('[build-all] Prebid adapters:', adapterNames);
 }
 
+function prebidModuleExists(moduleFile) {
+  const sourceFile = path.join(prebidModulesDir, moduleFile);
+  const sourceTsFile = path.join(prebidModulesDir, moduleFile.replace(/\.js$/, '.ts'));
+  const publicDistFile = path.join(
+    __dirname,
+    'node_modules',
+    'prebid.js',
+    'dist',
+    'src',
+    'public',
+    moduleFile
+  );
+
+  return fs.existsSync(sourceFile) || fs.existsSync(sourceTsFile) || fs.existsSync(publicDistFile);
+}
+
+function resolvePrebidUserIdModule(name) {
+  const candidates = [
+    `${name}.js`,
+    `${name}System.js`,
+    `${name}IdSystem.js`,
+    `${name}IdSubmodule.js`,
+  ];
+
+  return candidates.find(prebidModuleExists);
+}
+
+function generatePrebidUserIdModules() {
+  const names = parseCommaSeparatedEnv(
+    'TSJS_PREBID_USER_ID_MODULES',
+    DEFAULT_PREBID_USER_ID_MODULES
+  );
+
+  const imports = [];
+  const moduleNames = [];
+  for (const name of names) {
+    const moduleFile = resolvePrebidUserIdModule(name);
+    if (!moduleFile) {
+      console.error(`[build-all] WARNING: Prebid User ID module "${name}" not found, skipping`);
+      continue;
+    }
+    imports.push(`import 'prebid.js/modules/${moduleFile}';`);
+    moduleNames.push(moduleFile.replace(/\.js$/, ''));
+  }
+
+  const content = [
+    '// Auto-generated by build-all.mjs — manual edits will be overwritten at build time.',
+    '//',
+    '// Controls which Prebid.js User ID modules are included in the bundle.',
+    '// Set the TSJS_PREBID_USER_ID_MODULES environment variable to a comma-separated list',
+    '// of module names (e.g. "sharedId,pubProvided,uid2") before building.',
+    '// Default: empty',
+    ...(imports.length > 0 ? ['', ...imports] : []),
+  ].join('\n');
+
+  fs.writeFileSync(USER_ID_MODULES_FILE, `${content}\n`);
+  console.log('[build-all] Prebid User ID modules:', moduleNames);
+}
+
 generatePrebidAdapters();
+generatePrebidUserIdModules();
 
 // ---------------------------------------------------------------------------
 
@@ -120,8 +180,7 @@ const integrationModules = fs.existsSync(integrationsDir)
       .filter((name) => {
         const fullPath = path.join(integrationsDir, name);
         return (
-          fs.statSync(fullPath).isDirectory() &&
-          fs.existsSync(path.join(fullPath, 'index.ts'))
+          fs.statSync(fullPath).isDirectory() && fs.existsSync(path.join(fullPath, 'index.ts'))
         );
       })
       .sort()
@@ -143,7 +202,15 @@ async function buildModule(name, entryPath) {
         // "exports" map, but we need it for client-side bidder validation.
         'prebid.js/src/adapterManager.js': path.resolve(
           __dirname,
-          'node_modules/prebid.js/dist/src/src/adapterManager.js',
+          'node_modules/prebid.js/dist/src/src/adapterManager.js'
+        ),
+        // The published liveIntentIdSystem module contains a build-time
+        // require() switch that is not replaced by our Vite build. Import the
+        // standard implementation directly so browser bundles do not contain
+        // CommonJS require calls.
+        'prebid.js/modules/liveIntentIdSystem.js': path.resolve(
+          __dirname,
+          'node_modules/prebid.js/dist/src/libraries/liveIntentId/idSystem.js'
         ),
       },
     },
@@ -176,9 +243,7 @@ async function buildModule(name, entryPath) {
 await buildModule('core', path.join(srcDir, 'core', 'index.ts'));
 
 await Promise.all(
-  integrationModules.map((name) =>
-    buildModule(name, path.join(integrationsDir, name, 'index.ts')),
-  ),
+  integrationModules.map((name) => buildModule(name, path.join(integrationsDir, name, 'index.ts')))
 );
 
 // List all built files
diff --git a/crates/js/lib/src/integrations/prebid/_user_id_modules.generated.ts b/crates/js/lib/src/integrations/prebid/_user_id_modules.generated.ts
new file mode 100644
index 00000000..9aaf1625
--- /dev/null
+++ b/crates/js/lib/src/integrations/prebid/_user_id_modules.generated.ts
@@ -0,0 +1,6 @@
+// Auto-generated by build-all.mjs — manual edits will be overwritten at build time.
+//
+// Controls which Prebid.js User ID modules are included in the bundle.
+// Set the TSJS_PREBID_USER_ID_MODULES environment variable to a comma-separated list
+// of module names (e.g. "sharedId,pubProvided,uid2") before building.
+// Default: empty
diff --git a/crates/js/lib/src/integrations/prebid/index.ts b/crates/js/lib/src/integrations/prebid/index.ts
index b8d6d73b..39e98c7c 100644
--- a/crates/js/lib/src/integrations/prebid/index.ts
+++ b/crates/js/lib/src/integrations/prebid/index.ts
@@ -18,6 +18,11 @@ import 'prebid.js/modules/consentManagementGpp.js';
 import 'prebid.js/modules/consentManagementUsp.js';
 import 'prebid.js/modules/userId.js';
 
+// User ID submodules — self-register with prebid.js on import.
+// The set of submodules is controlled by the TSJS_PREBID_USER_ID_MODULES env var
+// at build time. See _user_id_modules.generated.ts (written by build-all.mjs).
+import './_user_id_modules.generated';
+
 // Client-side bid adapters — self-register with prebid.js on import.
 // The set of adapters is controlled by the TSJS_PREBID_ADAPTERS env var at
 // build time. See _adapters.generated.ts (written by build-all.mjs).
@@ -42,6 +47,8 @@ export interface PrebidNpmConfig {
   timeout?: number;
   /** Enable Prebid.js debug logging. Defaults to false. */
   debug?: boolean;
+  /** Prebid.js userSync configuration for User ID modules. */
+  userSync?: unknown;
 }
 
 /**
@@ -53,6 +60,8 @@ interface InjectedPrebidConfig {
   timeout?: number;
   debug?: boolean;
   bidders?: string[];
+  /** Prebid.js userSync configuration for User ID modules. */
+  userSync?: unknown;
   /** Bidders that run client-side via native Prebid.js adapters. */
   clientSideBidders?: string[];
 }
@@ -177,6 +186,10 @@ function isDefined<T>(value: T | undefined): value is T {
   return value !== undefined;
 }
 
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return !!value && typeof value === 'object' && !Array.isArray(value);
+}
+
 function collectAuctionEids(): AuctionEid[] | undefined {
   if (typeof pbjs.getUserIdsAsEids !== 'function') {
     return undefined;
@@ -218,6 +231,7 @@ export function installPrebidNpm(config?: Partial<PrebidNpmConfig>): typeof pbjs
     endpoint: config?.endpoint,
     timeout: config?.timeout ?? injected?.timeout,
     debug: config?.debug ?? injected?.debug,
+    userSync: config?.userSync ?? injected?.userSync,
   };
 
   auctionEndpoint = merged.endpoint ?? '/auction';
@@ -366,12 +380,15 @@ export function installPrebidNpm(config?: Partial<PrebidNpmConfig>): typeof pbjs
   };
 
   // Apply initial configuration
-  const pbjsConfig: PbjsConfig & { bidderTimeout?: number } = {
+  const pbjsConfig: PbjsConfig & { bidderTimeout?: number; userSync?: unknown } = {
     debug: merged.debug ?? false,
   };
   if (typeof merged.timeout === 'number') {
     pbjsConfig.bidderTimeout = merged.timeout;
   }
+  if (isRecord(merged.userSync)) {
+    pbjsConfig.userSync = merged.userSync;
+  }
   pbjs.setConfig(pbjsConfig as PbjsConfig);
 
   // processQueue() must be called after all modules are loaded when using
diff --git a/crates/js/lib/src/integrations/sourcepoint/index.ts b/crates/js/lib/src/integrations/sourcepoint/index.ts
index 0e31b086..f732b34c 100644
--- a/crates/js/lib/src/integrations/sourcepoint/index.ts
+++ b/crates/js/lib/src/integrations/sourcepoint/index.ts
@@ -5,6 +5,7 @@ const GPP_COOKIE_NAME = '__gpp';
 const GPP_SID_COOKIE_NAME = '__gpp_sid';
 const GPP_SOURCE_COOKIE_NAME = '_ts_gpp_src';
 const GPP_SOURCE_SOURCEPOINT = 'sp';
+const TCF_COOKIE_NAME = 'euconsent-v2';
 const INITIAL_RETRY_DELAY_MS = 500;
 
 interface SourcepointGppData {
@@ -12,15 +13,148 @@ interface SourcepointGppData {
   applicableSections: number[];
 }
 
+interface SourcepointConsentStringEntry {
+  sectionId?: number;
+  consentString?: string;
+}
+
+interface SourcepointSectionPayload {
+  consentString?: string;
+  applicableSections?: number[];
+  consentStrings?: SourcepointConsentStringEntry[];
+  tcString?: string;
+  euconsent?: string;
+  euconsentV2?: string;
+}
+
 interface SourcepointConsentPayload {
   gppData?: SourcepointGppData;
+  gdpr?: SourcepointSectionPayload;
+  usnat?: SourcepointSectionPayload;
+  [key: string]: unknown;
+}
+
+interface MirroredConsent {
+  gppString?: string;
+  gppSections?: number[];
+  tcString?: string;
 }
 
 let initialized = false;
 
-function findSourcepointConsent(): SourcepointConsentPayload | null {
-  // Sourcepoint stores one consent payload per property under `_sp_user_consent_*`.
-  // We intentionally take the first valid match and mirror that origin-scoped payload.
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === 'object' && value !== null;
+}
+
+function isNumberArray(value: unknown): value is number[] {
+  return Array.isArray(value) && value.every((item) => typeof item === 'number');
+}
+
+function isConsentStringEntryArray(value: unknown): value is SourcepointConsentStringEntry[] {
+  return (
+    Array.isArray(value) &&
+    value.every(
+      (item) =>
+        isRecord(item) &&
+        (typeof item.sectionId === 'number' || typeof item.sectionId === 'undefined') &&
+        (typeof item.consentString === 'string' || typeof item.consentString === 'undefined')
+    )
+  );
+}
+
+function normalizeSectionPayload(value: unknown): SourcepointSectionPayload | null {
+  if (!isRecord(value)) return null;
+
+  return {
+    consentString: typeof value.consentString === 'string' ? value.consentString : undefined,
+    applicableSections: isNumberArray(value.applicableSections)
+      ? value.applicableSections
+      : undefined,
+    consentStrings: isConsentStringEntryArray(value.consentStrings)
+      ? value.consentStrings
+      : undefined,
+    tcString: typeof value.tcString === 'string' ? value.tcString : undefined,
+    euconsent: typeof value.euconsent === 'string' ? value.euconsent : undefined,
+    euconsentV2: typeof value.euconsentV2 === 'string' ? value.euconsentV2 : undefined,
+  };
+}
+
+function sectionIdsFromConsentStrings(
+  consentStrings: SourcepointConsentStringEntry[] | undefined
+): number[] | undefined {
+  const ids = consentStrings
+    ?.map((entry) => entry.sectionId)
+    .filter((sectionId): sectionId is number => typeof sectionId === 'number');
+
+  return ids && ids.length > 0 ? ids : undefined;
+}
+
+function looksLikeGpp(consentString: string): boolean {
+  return consentString.includes('~');
+}
+
+function extractConsentFromSection(
+  sectionName: string,
+  section: SourcepointSectionPayload
+): MirroredConsent | null {
+  const gppSections =
+    section.applicableSections ?? sectionIdsFromConsentStrings(section.consentStrings);
+
+  if (section.consentString && (gppSections || looksLikeGpp(section.consentString))) {
+    return {
+      gppString: section.consentString,
+      gppSections,
+    };
+  }
+
+  if (sectionName === 'gdpr') {
+    const tcString =
+      section.tcString ?? section.euconsentV2 ?? section.euconsent ?? section.consentString;
+    if (tcString) {
+      return { tcString };
+    }
+  }
+
+  return null;
+}
+
+function mergeConsent(
+  primary: MirroredConsent | null,
+  secondary: MirroredConsent | null
+): MirroredConsent | null {
+  if (!primary) return secondary;
+  if (!secondary) return primary;
+
+  return {
+    gppString: primary.gppString ?? secondary.gppString,
+    gppSections: primary.gppSections ?? secondary.gppSections,
+    tcString: primary.tcString ?? secondary.tcString,
+  };
+}
+
+function extractMirroredConsent(payload: SourcepointConsentPayload): MirroredConsent | null {
+  let consent: MirroredConsent | null = null;
+
+  if (payload.gppData?.gppString) {
+    consent = mergeConsent(consent, {
+      gppString: payload.gppData.gppString,
+      gppSections: payload.gppData.applicableSections,
+    });
+  }
+
+  for (const [sectionName, rawSection] of Object.entries(payload)) {
+    if (sectionName === 'gppData') continue;
+
+    const section = normalizeSectionPayload(rawSection);
+    if (!section) continue;
+
+    consent = mergeConsent(consent, extractConsentFromSection(sectionName, section));
+  }
+
+  return consent;
+}
+
+function findSourcepointConsent(): MirroredConsent | null {
   for (let i = 0; i < localStorage.length; i++) {
     const key = localStorage.key(i);
     if (!key?.startsWith(SP_CONSENT_PREFIX)) continue;
@@ -30,8 +164,9 @@ function findSourcepointConsent(): SourcepointConsentPayload | null {
 
     try {
       const payload = JSON.parse(raw) as SourcepointConsentPayload;
-      if (payload.gppData?.gppString) {
-        return payload;
+      const consent = extractMirroredConsent(payload);
+      if (consent?.gppString || consent?.tcString) {
+        return consent;
       }
     } catch {
       log.debug('sourcepoint: failed to parse localStorage value', { key });
@@ -87,8 +222,14 @@ function scheduleInitialRetry(): void {
 }
 
 /**
- * Reads Sourcepoint consent from localStorage and mirrors it into
- * `__gpp` and `__gpp_sid` cookies for Trusted Server to read.
+ * Reads Sourcepoint consent from localStorage and mirrors it into cookies for
+ * Trusted Server to read on the next request.
+ *
+ * Sourcepoint localStorage differs by campaign/module. US National data is
+ * commonly stored under `usnat.consentString`/`usnat.applicableSections`, while
+ * some setups expose `gppData.gppString`. GDPR/UK setups may expose a TCF string
+ * under `gdpr` fields. Trusted Server cannot read localStorage server-side, so
+ * this bridge writes the standard IAB cookies.
  *
  * Returns `true` if cookies were written, `false` otherwise.
  */
@@ -97,35 +238,44 @@ export function mirrorSourcepointConsent(): boolean {
     return false;
   }
 
-  const payload = findSourcepointConsent();
-  if (!payload?.gppData) {
+  const consent = findSourcepointConsent();
+  if (!consent) {
     clearSourcepointCookies();
-    log.debug('sourcepoint: no GPP data found in localStorage');
+    log.debug('sourcepoint: no consent data found in localStorage');
     return false;
   }
 
-  const { gppString, applicableSections } = payload.gppData;
-  if (!gppString) {
+  let wroteCookie = false;
+
+  if (consent.gppString) {
+    writeCookie(GPP_SOURCE_COOKIE_NAME, GPP_SOURCE_SOURCEPOINT);
+    writeCookie(GPP_COOKIE_NAME, consent.gppString);
+
+    if (Array.isArray(consent.gppSections) && consent.gppSections.length > 0) {
+      writeCookie(GPP_SID_COOKIE_NAME, consent.gppSections.join(','));
+    } else {
+      clearCookie(GPP_SID_COOKIE_NAME);
+    }
+
+    wroteCookie = true;
+  } else {
     clearSourcepointCookies();
-    log.debug('sourcepoint: gppString is empty');
-    return false;
   }
 
-  writeCookie(GPP_SOURCE_COOKIE_NAME, GPP_SOURCE_SOURCEPOINT);
-  writeCookie(GPP_COOKIE_NAME, gppString);
-
-  if (Array.isArray(applicableSections) && applicableSections.length > 0) {
-    writeCookie(GPP_SID_COOKIE_NAME, applicableSections.join(','));
-  } else if (hasSourcepointMarker()) {
-    clearCookie(GPP_SID_COOKIE_NAME);
+  if (consent.tcString) {
+    writeCookie(TCF_COOKIE_NAME, consent.tcString);
+    wroteCookie = true;
   }
 
-  log.info('sourcepoint: mirrored GPP consent to cookies', {
-    gppLength: gppString.length,
-    sections: applicableSections,
-  });
+  if (wroteCookie) {
+    log.info('sourcepoint: mirrored consent to cookies', {
+      gppLength: consent.gppString?.length ?? 0,
+      sections: consent.gppSections,
+      tcLength: consent.tcString?.length ?? 0,
+    });
+  }
 
-  return true;
+  return wroteCookie;
 }
 
 /**
@@ -142,8 +292,6 @@ export function initializeSourcepointConsentMirror(): void {
     scheduleInitialRetry();
   }
 
-  // Sourcepoint persists consent changes to localStorage. Re-mirror when a
-  // user returns to the page so session cookies do not remain stale.
   document.addEventListener('visibilitychange', mirrorOnVisible);
   window.addEventListener('focus', mirrorSourcepointConsent);
 }
diff --git a/crates/js/lib/test/integrations/prebid/index.test.ts b/crates/js/lib/test/integrations/prebid/index.test.ts
index 47034861..5e8124ec 100644
--- a/crates/js/lib/test/integrations/prebid/index.test.ts
+++ b/crates/js/lib/test/integrations/prebid/index.test.ts
@@ -53,7 +53,8 @@ vi.mock('prebid.js/modules/consentManagementGpp.js', () => ({}));
 vi.mock('prebid.js/modules/consentManagementUsp.js', () => ({}));
 vi.mock('prebid.js/modules/userId.js', () => ({}));
 
-// Mock the build-generated adapter imports (no-op in tests)
+// Mock the build-generated module imports (no-op in tests)
+vi.mock('../../../src/integrations/prebid/_user_id_modules.generated', () => ({}));
 vi.mock('../../../src/integrations/prebid/_adapters.generated', () => ({}));
 
 import {
@@ -233,14 +234,33 @@ describe('prebid/installPrebidNpm', () => {
   });
 
   it('respects custom config values', () => {
+    const userSync = {
+      userIds: [
+        {
+          name: 'sharedId',
+          storage: { name: '_sharedID', type: 'cookie', expires: 365 },
+        },
+      ],
+      auctionDelay: 100,
+    };
+
     installPrebidNpm({
       endpoint: '/custom/auction',
       timeout: 2000,
       debug: true,
+      userSync,
     });
 
     expect(mockSetConfig).toHaveBeenCalledWith(
-      expect.objectContaining({ debug: true, bidderTimeout: 2000 })
+      expect.objectContaining({ debug: true, bidderTimeout: 2000, userSync })
+    );
+  });
+
+  it('ignores non-object userSync config values', () => {
+    installPrebidNpm({ userSync: 'sharedId' });
+
+    expect(mockSetConfig).toHaveBeenCalledWith(
+      expect.not.objectContaining({ userSync: expect.anything() })
     );
   });
 
@@ -756,6 +776,18 @@ describe('prebid/installPrebidNpm with server-injected config', () => {
     );
   });
 
+  it('reads userSync from window.__tsjs_prebid', () => {
+    const userSync = {
+      userIds: [{ name: 'sharedId', storage: { name: '_sharedID', type: 'cookie' } }],
+      auctionDelay: 50,
+    };
+    (window as any).__tsjs_prebid = { userSync };
+
+    installPrebidNpm();
+
+    expect(mockSetConfig).toHaveBeenCalledWith(expect.objectContaining({ userSync }));
+  });
+
   it('works with no config argument and no injected config', () => {
     installPrebidNpm();
 
diff --git a/crates/js/lib/test/integrations/sourcepoint/index.test.ts b/crates/js/lib/test/integrations/sourcepoint/index.test.ts
index 3914c2ef..57ef5812 100644
--- a/crates/js/lib/test/integrations/sourcepoint/index.test.ts
+++ b/crates/js/lib/test/integrations/sourcepoint/index.test.ts
@@ -49,6 +49,39 @@ describe('integrations/sourcepoint', () => {
     expect(getCookie(SOURCEPOINT_MARKER_COOKIE)).toBe('sp');
   });
 
+  it('mirrors __gpp and __gpp_sid from current Sourcepoint usnat localStorage shape', () => {
+    const payload = {
+      usnat: {
+        applicableSections: [7],
+        consentString: 'DBABLA~BVQqAAAAAgA.QA',
+        consentStatus: {
+          consentedToAll: true,
+        },
+      },
+    };
+    localStorage.setItem('_sp_user_consent_36922', JSON.stringify(payload));
+
+    const result = mirrorSourcepointConsent();
+
+    expect(result).toBe(true);
+    expect(getCookie('__gpp')).toBe('DBABLA~BVQqAAAAAgA.QA');
+    expect(getCookie('__gpp_sid')).toBe('7');
+  });
+
+  it('mirrors euconsent-v2 from Sourcepoint gdpr localStorage shape', () => {
+    const payload = {
+      gdpr: {
+        consentString: 'CPXxGfAPXxGfAAHABBENBCCsAP_AAH_AAAAAHftf',
+      },
+    };
+    localStorage.setItem('_sp_user_consent_36922', JSON.stringify(payload));
+
+    const result = mirrorSourcepointConsent();
+
+    expect(result).toBe(true);
+    expect(getCookie('euconsent-v2')).toBe('CPXxGfAPXxGfAAHABBENBCCsAP_AAH_AAAAAHftf');
+  });
+
   it('handles multiple applicable sections', () => {
     localStorage.setItem(
       '_sp_user_consent_99999',
diff --git a/crates/trusted-server-adapter-fastly/src/logging.rs b/crates/trusted-server-adapter-fastly/src/logging.rs
index f110c63b..09d43f62 100644
--- a/crates/trusted-server-adapter-fastly/src/logging.rs
+++ b/crates/trusted-server-adapter-fastly/src/logging.rs
@@ -3,6 +3,8 @@
 use chrono::{SecondsFormat, Utc};
 use log_fastly::Logger;
 
+const ENV_TS_LOG_LEVEL: &str = "TS_LOG_LEVEL";
+
 /// Extracts the final `::` segment from a Rust module path for use as a log label.
 ///
 /// When the input has no `::` separator, returns the full target. When the
@@ -27,14 +29,16 @@ fn target_label(target: &str) -> &str {
 /// Panics if the Fastly logger cannot be built or if the global logger has already
 /// been set.
 pub(crate) fn init_logger() {
+    let max_level = configured_log_level();
     let logger = Logger::builder()
         .default_endpoint("tslog")
         .echo_stdout(true)
-        .max_level(log::LevelFilter::Info)
+        .max_level(max_level)
         .build()
         .expect("should build Logger");
 
     fern::Dispatch::new()
+        .level(max_level)
         .format(|out, message, record| {
             out.finish(format_args!(
                 "{} {} [{}] {}",
@@ -49,6 +53,25 @@ pub(crate) fn init_logger() {
         .expect("should initialize logger");
 }
 
+fn configured_log_level() -> log::LevelFilter {
+    std::env::var(ENV_TS_LOG_LEVEL)
+        .ok()
+        .and_then(|value| parse_log_level(&value))
+        .unwrap_or(log::LevelFilter::Info)
+}
+
+fn parse_log_level(value: &str) -> Option<log::LevelFilter> {
+    match value.trim().to_ascii_lowercase().as_str() {
+        "off" => Some(log::LevelFilter::Off),
+        "error" => Some(log::LevelFilter::Error),
+        "warn" | "warning" => Some(log::LevelFilter::Warn),
+        "info" => Some(log::LevelFilter::Info),
+        "debug" => Some(log::LevelFilter::Debug),
+        "trace" => Some(log::LevelFilter::Trace),
+        _ => None,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/crates/trusted-server-adapter-fastly/src/main.rs b/crates/trusted-server-adapter-fastly/src/main.rs
index 30c6e391..e3242707 100644
--- a/crates/trusted-server-adapter-fastly/src/main.rs
+++ b/crates/trusted-server-adapter-fastly/src/main.rs
@@ -9,16 +9,19 @@ use trusted_server_core::constants::{
     COOKIE_SHAREDID, COOKIE_TS_EIDS, ENV_FASTLY_IS_STAGING, ENV_FASTLY_SERVICE_VERSION,
     HEADER_X_GEO_INFO_AVAILABLE, HEADER_X_TS_ENV, HEADER_X_TS_VERSION,
 };
+use trusted_server_core::ec::admin::handle_register_partner;
 use trusted_server_core::ec::batch_sync::handle_batch_sync;
 use trusted_server_core::ec::device::DeviceSignals;
 use trusted_server_core::ec::finalize::ec_finalize_response;
 use trusted_server_core::ec::identify::{cors_preflight_identify, handle_identify};
 use trusted_server_core::ec::kv::KvIdentityGraph;
+use trusted_server_core::ec::partner::PartnerStore;
 use trusted_server_core::ec::pull_sync::{
     build_pull_sync_context, dispatch_pull_sync, PullSyncContext,
 };
 use trusted_server_core::ec::rate_limiter::{FastlyRateLimiter, RATE_COUNTER_NAME};
 use trusted_server_core::ec::registry::PartnerRegistry;
+use trusted_server_core::ec::sync_pixel::handle_sync;
 use trusted_server_core::ec::EcContext;
 use trusted_server_core::error::TrustedServerError;
 use trusted_server_core::geo::GeoInfo;
@@ -290,6 +293,10 @@ async fn route_request(
             handle_deactivate_key(settings, runtime_services, req),
             false,
         ),
+        (Method::POST, "/_ts/admin/partners/register") => (
+            require_partner_store(settings).and_then(|store| handle_register_partner(&store, req)),
+            false,
+        ),
         (Method::GET, "/_ts/api/v1/identify") => (
             // Bot gate is intentionally write-only in this PR. `/identify` reads
             // remain gated by bearer auth + consent, even when request-classification
@@ -302,6 +309,15 @@ async fn route_request(
             (cors_preflight_identify(settings, &req), false)
         }
 
+        (Method::GET, "/sync") => (
+            require_identity_graph(settings).and_then(|kv| {
+                require_partner_store(settings).and_then(|partner_store| {
+                    handle_sync(settings, &kv, &partner_store, &req, &mut ec_context)
+                })
+            }),
+            false,
+        ),
+
         // Unified auction endpoint (returns creative HTML inline)
         (Method::POST, "/auction") => {
             let registry_ref = if partner_registry.is_empty() {
@@ -339,7 +355,15 @@ async fn route_request(
         }
         (m, path) if integration_registry.has_route(&m, path) => {
             let result = integration_registry
-                .handle_proxy(&m, path, settings, kv_graph.as_ref(), &mut ec_context, req)
+                .handle_proxy(
+                    &m,
+                    path,
+                    settings,
+                    runtime_services,
+                    kv_graph.as_ref(),
+                    &mut ec_context,
+                    req,
+                )
                 .await
                 .unwrap_or_else(|| {
                     Err(Report::new(TrustedServerError::BadRequest {
@@ -523,6 +547,18 @@ fn require_identity_graph(
     Ok(KvIdentityGraph::new(store_name))
 }
 
+/// Constructs a `PartnerStore` from settings, or returns 503 if the
+/// `partner_store` config is not set.
+fn require_partner_store(settings: &Settings) -> Result<PartnerStore, Report<TrustedServerError>> {
+    let store_name = settings.ec.partner_store.as_deref().ok_or_else(|| {
+        Report::new(TrustedServerError::KvStore {
+            store_name: "ec.partner_store".to_owned(),
+            message: "ec.partner_store is not configured".to_owned(),
+        })
+    })?;
+    Ok(PartnerStore::new(store_name))
+}
+
 /// Extracts a named cookie value from the request's `Cookie` header.
 fn extract_cookie_value(req: &Request, name: &str) -> Option<String> {
     let cookie_header = req.get_header_str("cookie")?;
diff --git a/crates/trusted-server-core/src/consent/mod.rs b/crates/trusted-server-core/src/consent/mod.rs
index 6dff4121..1159cf07 100644
--- a/crates/trusted-server-core/src/consent/mod.rs
+++ b/crates/trusted-server-core/src/consent/mod.rs
@@ -103,7 +103,7 @@ pub fn build_consent_context(input: &ConsentPipelineInput<'_>) -> ConsentContext
         let gdpr_applies =
             has_eu_tcf_signal(signals.raw_tc_string.is_some(), gpp_section_ids.as_deref());
         log::debug!("Consent proxy mode: jurisdiction={jur}, skipping decode");
-        return ConsentContext {
+        let ctx = ConsentContext {
             raw_tc_string: signals.raw_tc_string,
             raw_gpp_string: signals.raw_gpp_string,
             gpp_section_ids,
@@ -118,6 +118,8 @@ pub fn build_consent_context(input: &ConsentPipelineInput<'_>) -> ConsentContext
             jurisdiction: jur,
             source: ConsentSource::Cookie,
         };
+        log_consent_context(&ctx);
+        return ctx;
     }
 
     let mut ctx = build_context_from_signals(&signals);
@@ -456,6 +458,120 @@ pub fn gate_eids_by_consent<T>(
 // EC consent gating
 // ---------------------------------------------------------------------------
 
+/// Reason for an Edge Cookie creation consent decision.
+#[derive(Debug, Copy, Clone, Eq, PartialEq, derive_more::Display)]
+pub enum EcConsentReason {
+    /// GDPR applies and TCF Purpose 1 storage consent is granted.
+    GdprStorageConsentGranted,
+    /// GDPR applies but no decoded TCF signal is available.
+    GdprMissingTcf,
+    /// GDPR applies and TCF Purpose 1 storage consent is denied.
+    GdprStorageConsentDenied,
+    /// A US state privacy regime applies and GPC opts the user out.
+    UsGpcOptOut,
+    /// A US state privacy regime applies and TCF Purpose 1 storage consent is granted.
+    UsTcfStorageConsentGranted,
+    /// A US state privacy regime applies and TCF Purpose 1 storage consent is denied.
+    UsTcfStorageConsentDenied,
+    /// A US state privacy regime applies and GPP does not opt the user out of sale.
+    UsGppNoSaleOptOut,
+    /// A US state privacy regime applies and GPP opts the user out of sale.
+    UsGppSaleOptOut,
+    /// A US state privacy regime applies and US Privacy does not opt the user out of sale.
+    UsPrivacyNoSaleOptOut,
+    /// A US state privacy regime applies and US Privacy opts the user out of sale.
+    UsPrivacySaleOptOut,
+    /// A US state privacy regime applies but no usable consent signal is present.
+    UsMissingSignals,
+    /// The request is outside regulated jurisdictions.
+    NonRegulated,
+    /// Jurisdiction cannot be determined, so EC creation fails closed.
+    UnknownJurisdiction,
+}
+
+/// Result of evaluating whether Edge Cookie creation is permitted.
+#[derive(Debug, Copy, Clone, Eq, PartialEq)]
+pub struct EcConsentDecision {
+    /// Whether EC creation is permitted.
+    pub allowed: bool,
+    /// The branch responsible for the decision.
+    pub reason: EcConsentReason,
+}
+
+/// Determines whether Edge Cookie (EC) creation is permitted and why.
+#[must_use]
+pub fn ec_creation_decision(ctx: &ConsentContext) -> EcConsentDecision {
+    match &ctx.jurisdiction {
+        jurisdiction::Jurisdiction::Gdpr => match effective_tcf(ctx) {
+            Some(tcf) if tcf.has_storage_consent() => EcConsentDecision {
+                allowed: true,
+                reason: EcConsentReason::GdprStorageConsentGranted,
+            },
+            Some(_) => EcConsentDecision {
+                allowed: false,
+                reason: EcConsentReason::GdprStorageConsentDenied,
+            },
+            None => EcConsentDecision {
+                allowed: false,
+                reason: EcConsentReason::GdprMissingTcf,
+            },
+        },
+        jurisdiction::Jurisdiction::UsState(_) => {
+            if ctx.gpc {
+                return EcConsentDecision {
+                    allowed: false,
+                    reason: EcConsentReason::UsGpcOptOut,
+                };
+            }
+            if let Some(tcf) = effective_tcf(ctx) {
+                return EcConsentDecision {
+                    allowed: tcf.has_storage_consent(),
+                    reason: if tcf.has_storage_consent() {
+                        EcConsentReason::UsTcfStorageConsentGranted
+                    } else {
+                        EcConsentReason::UsTcfStorageConsentDenied
+                    },
+                };
+            }
+            if let Some(gpp) = &ctx.gpp {
+                if let Some(opted_out) = gpp.us_sale_opt_out {
+                    return EcConsentDecision {
+                        allowed: !opted_out,
+                        reason: if opted_out {
+                            EcConsentReason::UsGppSaleOptOut
+                        } else {
+                            EcConsentReason::UsGppNoSaleOptOut
+                        },
+                    };
+                }
+            }
+            if let Some(usp) = &ctx.us_privacy {
+                let opted_out = usp.opt_out_sale == PrivacyFlag::Yes;
+                return EcConsentDecision {
+                    allowed: !opted_out,
+                    reason: if opted_out {
+                        EcConsentReason::UsPrivacySaleOptOut
+                    } else {
+                        EcConsentReason::UsPrivacyNoSaleOptOut
+                    },
+                };
+            }
+            EcConsentDecision {
+                allowed: false,
+                reason: EcConsentReason::UsMissingSignals,
+            }
+        }
+        jurisdiction::Jurisdiction::NonRegulated => EcConsentDecision {
+            allowed: true,
+            reason: EcConsentReason::NonRegulated,
+        },
+        jurisdiction::Jurisdiction::Unknown => EcConsentDecision {
+            allowed: false,
+            reason: EcConsentReason::UnknownJurisdiction,
+        },
+    }
+}
+
 /// Determines whether Edge Cookie (EC) creation is permitted based on the
 /// user's consent and detected jurisdiction.
 ///
@@ -473,46 +589,7 @@ pub fn gate_eids_by_consent<T>(
 ///   blocked as a precaution.
 #[must_use]
 pub fn allows_ec_creation(ctx: &ConsentContext) -> bool {
-    match &ctx.jurisdiction {
-        jurisdiction::Jurisdiction::Gdpr => {
-            // EU/UK: explicit opt-in required (TCF Purpose 1 = store/access device).
-            match effective_tcf(ctx) {
-                Some(tcf) => tcf.has_storage_consent(),
-                None => false,
-            }
-        }
-        jurisdiction::Jurisdiction::UsState(_) => {
-            // GPC is an independent opt-out signal — it always blocks EC
-            // creation regardless of what the US Privacy string says.
-            if ctx.gpc {
-                return false;
-            }
-            // When a CMP uses TCF in the US (e.g. Didomi), respect the
-            // TCF Purpose 1 decision — this is an explicit opt-in signal.
-            // The Sourcepoint GPP design documents this precedence decision.
-            if let Some(tcf) = effective_tcf(ctx) {
-                return tcf.has_storage_consent();
-            }
-            // Check GPP US section for sale opt-out.
-            if let Some(gpp) = &ctx.gpp {
-                if let Some(opted_out) = gpp.us_sale_opt_out {
-                    return !opted_out;
-                }
-            }
-            // Check US Privacy string for explicit opt-out.
-            if let Some(usp) = &ctx.us_privacy {
-                return usp.opt_out_sale != PrivacyFlag::Yes;
-            }
-            // Spec §6.1.1: "In regulated jurisdictions (GDPR, US state),
-            // consent cookies/headers must be present for
-            // allows_ec_creation() to return true." No signals = block.
-            false
-        }
-        jurisdiction::Jurisdiction::NonRegulated => true,
-        // No geolocation data — cannot determine jurisdiction.
-        // Fail-closed: block EC creation as a precaution.
-        jurisdiction::Jurisdiction::Unknown => false,
-    }
+    ec_creation_decision(ctx).allowed
 }
 
 /// Returns `true` only when the request contains an explicit EC opt-out signal.
@@ -575,10 +652,6 @@ fn signal_status(decoded: bool, raw: bool) -> &'static str {
 
 /// Logs a structured summary of the fully-processed consent context.
 fn log_consent_context(ctx: &ConsentContext) {
-    if ctx.is_empty() {
-        return;
-    }
-
     let tcf_status = match (&ctx.tcf, ctx.expired) {
         (Some(_), _) => "present",
         (None, true) => "expired",
@@ -589,13 +662,17 @@ fn log_consent_context(ctx: &ConsentContext) {
     let gpp_status = signal_status(ctx.gpp.is_some(), ctx.raw_gpp_string.is_some());
     let usp_status = signal_status(ctx.us_privacy.is_some(), ctx.raw_us_privacy.is_some());
 
+    let ec_decision = ec_creation_decision(ctx);
     log::info!(
         "Consent context: jurisdiction={}, tcf={tcf_status}, gpp={gpp_status}, \
-         us_privacy={usp_status}, gpc={}, gdpr_applies={}, source={:?}",
+         us_privacy={usp_status}, gpc={}, gdpr_applies={}, source={:?}, \
+         ec_allowed={}, ec_reason={}",
         ctx.jurisdiction,
         ctx.gpc,
         ctx.gdpr_applies,
         ctx.source,
+        ec_decision.allowed,
+        ec_decision.reason,
     );
 }
 
diff --git a/crates/trusted-server-core/src/ec/admin.rs b/crates/trusted-server-core/src/ec/admin.rs
new file mode 100644
index 00000000..8aaf34ee
--- /dev/null
+++ b/crates/trusted-server-core/src/ec/admin.rs
@@ -0,0 +1,380 @@
+//! Admin endpoints for partner management.
+//!
+//! Provides `POST /_ts/admin/partners/register` for registering and updating
+//! partner configurations. Authentication is handled by the `[[handlers]]`
+//! basic-auth layer before this code runs.
+
+use error_stack::{Report, ResultExt};
+use fastly::{Request, Response};
+use serde::{Deserialize, Serialize};
+use std::collections::HashSet;
+use url::Host;
+
+use crate::error::TrustedServerError;
+
+use super::partner::{
+    hash_api_key, validate_partner_id, validate_pull_sync_config, PartnerRecord, PartnerStore,
+};
+
+/// Request body for `POST /_ts/admin/partners/register`.
+///
+/// Accepts `api_key` as plaintext — it is hashed before storage and
+/// never persisted in cleartext.
+#[derive(Debug, Deserialize)]
+pub struct RegisterPartnerRequest {
+    pub id: String,
+    pub name: String,
+    pub allowed_return_domains: Vec<String>,
+    /// Raw API key — will be SHA-256 hashed before storage.
+    pub api_key: String,
+    #[serde(default)]
+    pub bidstream_enabled: bool,
+    pub source_domain: String,
+    #[serde(default = "default_openrtb_atype")]
+    pub openrtb_atype: u8,
+    #[serde(default = "default_sync_rate_limit")]
+    pub sync_rate_limit: u32,
+    #[serde(default = "default_batch_rate_limit")]
+    pub batch_rate_limit: u32,
+    #[serde(default)]
+    pub pull_sync_enabled: bool,
+    #[serde(default)]
+    pub pull_sync_url: Option<String>,
+    #[serde(default)]
+    pub pull_sync_allowed_domains: Vec<String>,
+    #[serde(default = "default_pull_sync_ttl_sec")]
+    pub pull_sync_ttl_sec: u64,
+    #[serde(default = "default_pull_sync_rate_limit")]
+    pub pull_sync_rate_limit: u32,
+    #[serde(default)]
+    pub ts_pull_token: Option<String>,
+}
+
+fn default_openrtb_atype() -> u8 {
+    3
+}
+fn default_sync_rate_limit() -> u32 {
+    100
+}
+fn default_batch_rate_limit() -> u32 {
+    60
+}
+fn default_pull_sync_ttl_sec() -> u64 {
+    86400
+}
+fn default_pull_sync_rate_limit() -> u32 {
+    10
+}
+
+fn bad_request(message: impl Into<String>) -> Report<TrustedServerError> {
+    Report::new(TrustedServerError::BadRequest {
+        message: message.into(),
+    })
+}
+
+fn normalize_required_text(
+    value: &str,
+    field_name: &str,
+) -> Result<String, Report<TrustedServerError>> {
+    let trimmed = value.trim();
+    if trimmed.is_empty() {
+        return Err(bad_request(format!("{field_name} is required")));
+    }
+    Ok(trimmed.to_owned())
+}
+
+fn normalize_hostname(value: &str, field_name: &str) -> Result<String, Report<TrustedServerError>> {
+    let trimmed = value.trim().trim_end_matches('.');
+    if trimmed.is_empty() {
+        return Err(bad_request(format!("{field_name} is required")));
+    }
+
+    let normalized = trimmed.to_ascii_lowercase();
+    Host::parse(&normalized)
+        .map_err(|_| bad_request(format!("{field_name} must be a valid hostname")))?;
+
+    Ok(normalized)
+}
+
+fn normalize_hostname_list(
+    values: Vec<String>,
+    field_name: &str,
+) -> Result<Vec<String>, Report<TrustedServerError>> {
+    let mut normalized_values = Vec::with_capacity(values.len());
+    let mut seen = HashSet::with_capacity(values.len());
+
+    for value in values {
+        let trimmed = value.trim().trim_end_matches('.');
+        if trimmed.is_empty() {
+            return Err(bad_request(format!(
+                "{field_name} entries must not be empty"
+            )));
+        }
+
+        let normalized = trimmed.to_ascii_lowercase();
+        Host::parse(&normalized).map_err(|_| {
+            bad_request(format!("{field_name} contains invalid hostname '{value}'"))
+        })?;
+
+        if seen.insert(normalized.clone()) {
+            normalized_values.push(normalized);
+        }
+    }
+
+    Ok(normalized_values)
+}
+
+/// Response body for `POST /_ts/admin/partners/register`.
+///
+/// Echoes key fields without exposing sensitive data (`api_key_hash`,
+/// `ts_pull_token`).
+#[derive(Debug, Serialize)]
+pub struct RegisterPartnerResponse {
+    pub id: String,
+    pub name: String,
+    pub pull_sync_enabled: bool,
+    pub bidstream_enabled: bool,
+    pub created: bool,
+}
+
+/// Handles `POST /_ts/admin/partners/register`.
+///
+/// Registers a new partner or updates an existing one. Authentication is
+/// handled upstream by the `[[handlers]]` basic-auth layer.
+///
+/// # Errors
+///
+/// Returns `Report<TrustedServerError>` on validation failure (400),
+/// KV store failure (503), or JSON parse failure (400).
+pub fn handle_register_partner(
+    partner_store: &PartnerStore,
+    mut req: Request,
+) -> Result<Response, Report<TrustedServerError>> {
+    // Parse request body.
+    let body_bytes = req.take_body_bytes();
+    let request: RegisterPartnerRequest =
+        serde_json::from_slice(&body_bytes).change_context(TrustedServerError::BadRequest {
+            message: "Invalid JSON in request body".to_owned(),
+        })?;
+
+    let RegisterPartnerRequest {
+        id,
+        name,
+        allowed_return_domains,
+        api_key,
+        bidstream_enabled,
+        source_domain,
+        openrtb_atype,
+        sync_rate_limit,
+        batch_rate_limit,
+        pull_sync_enabled,
+        pull_sync_url,
+        pull_sync_allowed_domains,
+        pull_sync_ttl_sec,
+        pull_sync_rate_limit,
+        ts_pull_token,
+    } = request;
+
+    // Validate partner ID.
+    validate_partner_id(&id).map_err(bad_request)?;
+
+    // Validate and normalize required fields.
+    let name = normalize_required_text(&name, "name")?;
+    if api_key.trim().is_empty() {
+        return Err(bad_request("api_key is required"));
+    }
+    let source_domain = normalize_hostname(&source_domain, "source_domain")?;
+
+    if allowed_return_domains.is_empty() {
+        return Err(bad_request(
+            "allowed_return_domains must have at least one entry",
+        ));
+    }
+    let allowed_return_domains =
+        normalize_hostname_list(allowed_return_domains, "allowed_return_domains")?;
+    let pull_sync_allowed_domains =
+        normalize_hostname_list(pull_sync_allowed_domains, "pull_sync_allowed_domains")?;
+
+    // Build the PartnerRecord with hashed API key.
+    let record = PartnerRecord {
+        id,
+        name,
+        allowed_return_domains,
+        api_key_hash: hash_api_key(&api_key),
+        bidstream_enabled,
+        source_domain,
+        openrtb_atype,
+        sync_rate_limit,
+        batch_rate_limit,
+        pull_sync_enabled,
+        pull_sync_url,
+        pull_sync_allowed_domains,
+        pull_sync_ttl_sec,
+        pull_sync_rate_limit,
+        ts_pull_token,
+    };
+
+    // Validate pull sync configuration.
+    validate_pull_sync_config(&record).map_err(bad_request)?;
+
+    // Persist to KV store.
+    let created = partner_store.upsert(&record)?;
+
+    let status = if created {
+        log::info!("Registered new partner '{}'", record.id);
+        fastly::http::StatusCode::CREATED
+    } else {
+        log::info!("Updated existing partner '{}'", record.id);
+        fastly::http::StatusCode::OK
+    };
+
+    let response_body = RegisterPartnerResponse {
+        id: record.id,
+        name: record.name,
+        pull_sync_enabled: record.pull_sync_enabled,
+        bidstream_enabled: record.bidstream_enabled,
+        created,
+    };
+
+    let body = serde_json::to_string(&response_body).change_context(
+        TrustedServerError::Configuration {
+            message: "Failed to serialize registration response".to_owned(),
+        },
+    )?;
+
+    Ok(Response::from_status(status)
+        .with_content_type(fastly::mime::APPLICATION_JSON)
+        .with_body(body))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn request_deserializes_with_defaults() {
+        let json = r#"{
+            "id": "ssp_x",
+            "name": "SSP Example",
+            "allowed_return_domains": ["sync.example-ssp.com"],
+            "api_key": "raw-secret-key",
+            "source_domain": "example-ssp.com"
+        }"#;
+
+        let req: RegisterPartnerRequest =
+            serde_json::from_str(json).expect("should deserialize with defaults");
+
+        assert_eq!(req.id, "ssp_x");
+        assert_eq!(req.openrtb_atype, 3, "should default to 3");
+        assert_eq!(req.sync_rate_limit, 100, "should default to 100");
+        assert_eq!(req.batch_rate_limit, 60, "should default to 60");
+        assert_eq!(req.pull_sync_ttl_sec, 86400, "should default to 86400");
+        assert_eq!(req.pull_sync_rate_limit, 10, "should default to 10");
+        assert!(!req.bidstream_enabled, "should default to false");
+        assert!(!req.pull_sync_enabled, "should default to false");
+        assert!(req.pull_sync_url.is_none());
+        assert!(req.ts_pull_token.is_none());
+    }
+
+    #[test]
+    fn response_does_not_contain_sensitive_fields() {
+        let response = RegisterPartnerResponse {
+            id: "ssp_x".to_owned(),
+            name: "SSP Example".to_owned(),
+            pull_sync_enabled: false,
+            bidstream_enabled: true,
+            created: true,
+        };
+
+        let json = serde_json::to_string(&response).expect("should serialize");
+        assert!(!json.contains("api_key"), "should not contain api_key");
+        assert!(
+            !json.contains("api_key_hash"),
+            "should not contain api_key_hash"
+        );
+        assert!(
+            !json.contains("ts_pull_token"),
+            "should not contain ts_pull_token"
+        );
+    }
+
+    #[test]
+    fn request_deserializes_full_payload() {
+        let json = r#"{
+            "id": "ssp_x",
+            "name": "SSP Example",
+            "allowed_return_domains": ["sync.example-ssp.com"],
+            "api_key": "raw-secret-key",
+            "bidstream_enabled": true,
+            "source_domain": "example-ssp.com",
+            "openrtb_atype": 3,
+            "sync_rate_limit": 200,
+            "batch_rate_limit": 120,
+            "pull_sync_enabled": true,
+            "pull_sync_url": "https://sync.example-ssp.com/pull",
+            "pull_sync_allowed_domains": ["sync.example-ssp.com"],
+            "pull_sync_ttl_sec": 43200,
+            "pull_sync_rate_limit": 5,
+            "ts_pull_token": "bearer-token-123"
+        }"#;
+
+        let req: RegisterPartnerRequest =
+            serde_json::from_str(json).expect("should deserialize full payload");
+
+        assert_eq!(req.sync_rate_limit, 200);
+        assert_eq!(req.batch_rate_limit, 120);
+        assert!(req.pull_sync_enabled);
+        assert_eq!(
+            req.pull_sync_url.as_deref(),
+            Some("https://sync.example-ssp.com/pull")
+        );
+    }
+
+    #[test]
+    fn normalize_required_text_rejects_whitespace_only() {
+        let err = normalize_required_text("   ", "name")
+            .expect_err("should reject whitespace-only required field");
+        assert!(
+            err.to_string().contains("name is required"),
+            "should mention required field"
+        );
+    }
+
+    #[test]
+    fn normalize_hostname_normalizes_case_and_trailing_dot() {
+        let normalized = normalize_hostname("  Sync.Example.COM.  ", "source_domain")
+            .expect("should parse host");
+        assert_eq!(normalized, "sync.example.com");
+    }
+
+    #[test]
+    fn normalize_hostname_list_rejects_empty_entry() {
+        let err = normalize_hostname_list(
+            vec!["sync.example.com".to_owned(), "   ".to_owned()],
+            "allowed_return_domains",
+        )
+        .expect_err("should reject empty domain entries");
+        assert!(
+            err.to_string()
+                .contains("allowed_return_domains entries must not be empty"),
+            "should surface empty-entry error"
+        );
+    }
+
+    #[test]
+    fn normalize_hostname_list_deduplicates_normalized_values() {
+        let normalized = normalize_hostname_list(
+            vec![
+                "Sync.Example.com".to_owned(),
+                "sync.example.com.".to_owned(),
+                "cdn.example.com".to_owned(),
+            ],
+            "allowed_return_domains",
+        )
+        .expect("should normalize hostnames");
+        assert_eq!(
+            normalized,
+            vec!["sync.example.com".to_owned(), "cdn.example.com".to_owned()]
+        );
+    }
+}
diff --git a/crates/trusted-server-core/src/ec/consent.rs b/crates/trusted-server-core/src/ec/consent.rs
index ad9f5dd2..a59db914 100644
--- a/crates/trusted-server-core/src/ec/consent.rs
+++ b/crates/trusted-server-core/src/ec/consent.rs
@@ -6,7 +6,7 @@
 //! eventual migration path (renaming, adding EC-specific conditions) is
 //! contained here.
 
-use crate::consent::ConsentContext;
+use crate::consent::{ConsentContext, EcConsentDecision};
 
 /// Determines whether Edge Cookie creation is permitted based on the
 /// user's consent and detected jurisdiction.
@@ -18,7 +18,13 @@ use crate::consent::ConsentContext;
 /// See [`crate::consent::allows_ec_creation`] for the full decision matrix.
 #[must_use]
 pub fn ec_consent_granted(consent_context: &ConsentContext) -> bool {
-    crate::consent::allows_ec_creation(consent_context)
+    ec_consent_decision(consent_context).allowed
+}
+
+/// Determines whether Edge Cookie creation is permitted and why.
+#[must_use]
+pub fn ec_consent_decision(consent_context: &ConsentContext) -> EcConsentDecision {
+    crate::consent::ec_creation_decision(consent_context)
 }
 
 /// Returns `true` when the request carries an explicit EC withdrawal signal.
diff --git a/crates/trusted-server-core/src/ec/finalize.rs b/crates/trusted-server-core/src/ec/finalize.rs
index f08120e9..f17003a8 100644
--- a/crates/trusted-server-core/src/ec/finalize.rs
+++ b/crates/trusted-server-core/src/ec/finalize.rs
@@ -7,7 +7,7 @@ use std::collections::HashSet;
 
 use fastly::Response;
 
-use super::consent::{ec_consent_granted, ec_consent_withdrawn};
+use super::consent::{ec_consent_decision, ec_consent_withdrawn};
 use crate::constants::HEADER_X_TS_EC;
 use crate::settings::Settings;
 
@@ -47,7 +47,8 @@ pub fn ec_finalize_response(
     sharedid_cookie: Option<&str>,
     response: &mut Response,
 ) {
-    let consent_allows_ec = ec_consent_granted(ec_context.consent());
+    let consent_decision = ec_consent_decision(ec_context.consent());
+    let consent_allows_ec = consent_decision.allowed;
     let consent_withdrawn = ec_consent_withdrawn(ec_context.consent());
 
     if !consent_allows_ec {
@@ -56,6 +57,17 @@ pub fn ec_finalize_response(
         // revocation and fail-closed cases such as missing geo or undecodable
         // consent input.
         clear_ec_headers_on_response(response, Some(registry));
+        log::info!(
+            "EC finalize: action=clear_headers, consent_allowed=false, \
+             consent_reason={}, consent_withdrawn={}, cookie_present={}, \
+             ec_present={}, ec_generated={}, kv_available={}",
+            consent_decision.reason,
+            consent_withdrawn,
+            ec_context.cookie_was_present(),
+            ec_context.ec_value().is_some(),
+            ec_context.ec_generated(),
+            kv.is_some(),
+        );
 
         // Only expire the browser cookie and tombstone the identity-graph row
         // when the request carries an explicit withdrawal signal.
@@ -64,6 +76,13 @@ pub fn ec_finalize_response(
 
             // Compute once for the authoritative identity-graph tombstones.
             let ids_to_withdraw = withdrawal_ec_ids(ec_context);
+            log::info!(
+                "EC finalize: action=expire_cookie_and_tombstone, tombstone_count={}, \
+                 consent_reason={}, kv_available={}",
+                ids_to_withdraw.len(),
+                consent_decision.reason,
+                kv.is_some(),
+            );
 
             // The identity-graph tombstone is the authoritative withdrawal marker
             // for subsequent EC behavior.
@@ -74,6 +93,11 @@ pub fn ec_finalize_response(
                             "Failed to write withdrawal tombstone for EC ID '{}': {err:?}",
                             log_id(ec_id),
                         );
+                    } else {
+                        log::info!(
+                            "EC finalize: action=tombstone_written, ec_id={}",
+                            log_id(ec_id),
+                        );
                     }
                 });
             }
@@ -97,6 +121,14 @@ pub fn ec_finalize_response(
         // Returning users keep the active EC visible for this response, but
         // ordinary page views no longer refresh the browser cookie or KV TTL.
         set_ec_header_on_response(ec_context, response);
+        log::info!(
+            "EC finalize: action=set_returning_header, consent_allowed=true, \
+             consent_reason={}, cookie_present={}, ec_present=true, ec_generated=false, \
+             kv_available={}",
+            consent_decision.reason,
+            ec_context.cookie_was_present(),
+            kv.is_some(),
+        );
 
         return;
     }
@@ -106,7 +138,14 @@ pub fn ec_finalize_response(
     // identity-graph row, producing a phantom ID on later requests.
     if ec_context.ec_generated() {
         let (Some(graph), Some(ec_id)) = (kv, ec_context.ec_value()) else {
-            log::debug!("Skipping generated EC response write because KV graph is unavailable");
+            log::info!(
+                "EC finalize: action=skip_generated_no_kv, consent_allowed=true, \
+                 consent_reason={}, cookie_present={}, ec_present={}, ec_generated=true, \
+                 kv_available=false",
+                consent_decision.reason,
+                ec_context.cookie_was_present(),
+                ec_context.ec_value().is_some(),
+            );
             return;
         };
 
@@ -117,7 +156,25 @@ pub fn ec_finalize_response(
             ingest_sharedid_cookie(cookie, ec_id, graph, registry);
         }
         set_ec_cookie_and_header_on_response(settings, ec_context, response);
+        log::info!(
+            "EC finalize: action=set_generated_cookie_and_header, consent_allowed=true, \
+             consent_reason={}, cookie_present={}, ec_present=true, ec_generated=true, \
+             kv_available=true, ec_id={}",
+            consent_decision.reason,
+            ec_context.cookie_was_present(),
+            log_id(ec_id),
+        );
+        return;
     }
+
+    log::info!(
+        "EC finalize: action=no_ec_output, consent_allowed=true, consent_reason={}, \
+         cookie_present={}, ec_present={}, ec_generated=false, kv_available={}",
+        consent_decision.reason,
+        ec_context.cookie_was_present(),
+        ec_context.ec_value().is_some(),
+        kv.is_some(),
+    );
 }
 
 /// Sets the EC response header when an EC ID is available.
diff --git a/crates/trusted-server-core/src/ec/mod.rs b/crates/trusted-server-core/src/ec/mod.rs
index 5e945617..acccf8e5 100644
--- a/crates/trusted-server-core/src/ec/mod.rs
+++ b/crates/trusted-server-core/src/ec/mod.rs
@@ -20,16 +20,19 @@
 //! - [`kv`] — KV Store identity graph operations (CAS, tombstones, debounce)
 //! - [`kv_types`] — Schema types for KV identity graph entries
 //! - [`device`] — Device signal derivation (UA, JA4, H2 fingerprinting)
-//! - [`partner`] — Partner validation helpers (ID format, pull sync config)
+//! - [`partner`] — Partner validation helpers plus KV-backed admin store types
 //! - [`registry`] — In-memory partner registry built from config
 //! - [`rate_limiter`] — Rate limiting abstraction (Fastly Edge Rate Limiting)
 //! - [`identify`] — Identity read endpoint (`GET /_ts/api/v1/identify`)
 //! - [`eids`] — Shared EID resolution and formatting helpers
 //! - [`batch_sync`] — S2S batch sync endpoint (`POST /_ts/api/v1/batch-sync`)
 //! - [`pull_sync`] — Background pull-sync dispatcher for organic routes
+//! - [`admin`] — Admin endpoints for partner management
+//! - [`sync_pixel`] — Pixel sync write endpoint (`GET /sync`)
 
 mod auth;
 
+pub mod admin;
 pub mod batch_sync;
 pub mod consent;
 pub mod cookies;
@@ -45,6 +48,7 @@ pub mod prebid_eids;
 pub mod pull_sync;
 pub mod rate_limiter;
 pub mod registry;
+pub mod sync_pixel;
 
 /// Truncates an EC ID for safe inclusion in log messages.
 ///
@@ -89,6 +93,21 @@ struct RequestEc {
     jar: Option<CookieJar>,
 }
 
+#[derive(Debug, Copy, Clone, Eq, PartialEq, derive_more::Display)]
+enum EcInputStatus {
+    Absent,
+    Valid,
+    Invalid,
+}
+
+fn ec_input_status(value: Option<&str>) -> EcInputStatus {
+    match value {
+        Some(v) if is_valid_ec_id(v) => EcInputStatus::Valid,
+        Some(_) => EcInputStatus::Invalid,
+        None => EcInputStatus::Absent,
+    }
+}
+
 /// Parses EC identity from request headers and cookies in a single pass.
 ///
 /// # Errors
@@ -208,12 +227,30 @@ impl EcContext {
         // Malformed values are discarded per §4.2: "If the header is
         // present but malformed, it is discarded and the cookie value
         // is used instead."
+        let header_status = ec_input_status(parsed.header_ec.as_deref());
+        let cookie_status = ec_input_status(parsed.cookie_ec.as_deref());
+        let active_source = if header_status == EcInputStatus::Valid {
+            "header"
+        } else if cookie_status == EcInputStatus::Valid {
+            "cookie"
+        } else {
+            "none"
+        };
+        let header_cookie_differ = matches!(
+            (parsed.header_ec.as_deref(), parsed.cookie_ec.as_deref()),
+            (Some(header), Some(cookie)) if header != cookie
+        );
         let ec_value = parsed
             .header_ec
             .filter(|v| is_valid_ec_id(v))
             .or_else(|| parsed.cookie_ec.clone().filter(|v| is_valid_ec_id(v)));
         let ec_was_present = ec_value.is_some();
 
+        log::info!(
+            "EC request state: header_ec={header_status}, cookie_ec={cookie_status}, \
+             active_source={active_source}, ec_present={ec_was_present}, \
+             header_cookie_differ={header_cookie_differ}"
+        );
         if let Some(ref id) = ec_value {
             log::trace!("Existing EC ID found: {}", log_id(id));
         }
@@ -259,14 +296,30 @@ impl EcContext {
         settings: &Settings,
         kv: Option<&KvIdentityGraph>,
     ) -> Result<(), Report<TrustedServerError>> {
-        if self.ec_value.is_some() {
+        if let Some(ec_id) = self.ec_value.as_deref() {
+            let decision = consent::ec_consent_decision(&self.consent);
+            log::info!(
+                "EC generation decision: action=existing_ec, ec_present=true, \
+                 ec_generated=false, consent_allowed={}, consent_reason={}, \
+                 jurisdiction={}, kv_available={}, ec_id={}",
+                decision.allowed,
+                decision.reason,
+                self.consent.jurisdiction,
+                kv.is_some(),
+                log_id(ec_id),
+            );
             return Ok(());
         }
 
-        if !consent::ec_consent_granted(&self.consent) {
-            log::debug!(
-                "EC generation skipped: consent not granted (jurisdiction={})",
+        let decision = consent::ec_consent_decision(&self.consent);
+        if !decision.allowed {
+            log::info!(
+                "EC generation decision: action=blocked_by_consent, ec_present=false, \
+                 ec_generated=false, consent_allowed=false, consent_reason={}, \
+                 jurisdiction={}, kv_available={}",
+                decision.reason,
                 self.consent.jurisdiction,
+                kv.is_some(),
             );
             return Ok(());
         }
@@ -302,9 +355,29 @@ impl EcContext {
                 );
                 self.ec_value = None;
                 self.ec_generated = false;
+                log::info!(
+                    "EC generation decision: action=generated_but_kv_write_failed, \
+                     ec_present=false, ec_generated=false, consent_allowed=true, \
+                     consent_reason={}, jurisdiction={}, kv_available=true",
+                    decision.reason,
+                    self.consent.jurisdiction,
+                );
+                return Ok(());
             }
         }
 
+        if let Some(ec_id) = self.ec_value.as_deref() {
+            log::info!(
+                "EC generation decision: action=generated, ec_present=false, \
+                 ec_generated=true, consent_allowed=true, consent_reason={}, \
+                 jurisdiction={}, kv_available={}, ec_id={}",
+                decision.reason,
+                self.consent.jurisdiction,
+                kv.is_some(),
+                log_id(ec_id),
+            );
+        }
+
         Ok(())
     }
 
diff --git a/crates/trusted-server-core/src/ec/partner.rs b/crates/trusted-server-core/src/ec/partner.rs
index f836724e..a783d934 100644
--- a/crates/trusted-server-core/src/ec/partner.rs
+++ b/crates/trusted-server-core/src/ec/partner.rs
@@ -1,12 +1,19 @@
-//! Partner validation helpers and ID hashing.
+//! Partner registry — `PartnerRecord` schema and `PartnerStore` operations.
 //!
-//! Provides partner ID format validation, reserved name checks, and
-//! API key hashing. The actual partner registry is in [`super::registry`].
+//! Each partner (SSP, DSP, identity vendor) is stored as a JSON record in
+//! the Fastly KV Store keyed by `partner_id`. A secondary index
+//! `apikey:{sha256_hex}` provides O(1) API key lookups for batch sync auth.
 
-use std::sync::OnceLock;
+use std::{collections::HashSet, sync::OnceLock};
 
+use error_stack::{Report, ResultExt};
+use fastly::kv_store::KVStore;
 use regex::Regex;
+use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
+use subtle::ConstantTimeEq;
+
+use crate::error::TrustedServerError;
 
 /// Regex pattern for valid partner IDs.
 /// Lowercase alphanumeric, hyphens, underscores; 1-32 characters.
@@ -24,6 +31,9 @@ const RESERVED_PARTNER_IDS: &[&str] = &[
     "env",
 ];
 
+/// Prefix for the API key hash secondary index keys.
+const APIKEY_INDEX_PREFIX: &str = "apikey:";
+
 /// Cached compiled regex for partner ID validation.
 static PARTNER_ID_REGEX: OnceLock<Result<Regex, String>> = OnceLock::new();
 
@@ -37,6 +47,49 @@ fn partner_id_regex() -> Result<&'static Regex, String> {
         .map_err(Clone::clone)
 }
 
+/// A registered partner configuration stored in the partner KV store.
+///
+/// Created via `POST /_ts/admin/partners/register`. Used by pixel sync, batch
+/// sync, pull sync, and auction bidstream decoration.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct PartnerRecord {
+    /// Unique partner identifier. Must match [`PARTNER_ID_PATTERN`] and
+    /// not be in [`RESERVED_PARTNER_IDS`]. Used to build `X-ts-<id>`
+    /// response headers.
+    pub id: String,
+    /// Human-readable partner name.
+    pub name: String,
+    /// Exact hostnames allowed as `return` URL domains in pixel sync.
+    pub allowed_return_domains: Vec<String>,
+    /// SHA-256 hex of the partner's API key. Plaintext is never stored.
+    pub api_key_hash: String,
+    /// Whether this partner's UIDs appear in auction `user.eids`.
+    pub bidstream_enabled: bool,
+    /// `OpenRTB` `source.domain` for EID entries (e.g. `"liveramp.com"`).
+    pub source_domain: String,
+    /// `OpenRTB` `atype` value (typically 3).
+    pub openrtb_atype: u8,
+    /// Max pixel sync writes per EC hash per partner per hour.
+    pub sync_rate_limit: u32,
+    /// Max batch sync API requests per partner per minute.
+    pub batch_rate_limit: u32,
+    /// Whether server-to-server pull sync is enabled for this partner.
+    pub pull_sync_enabled: bool,
+    /// URL to call for pull sync. Required when `pull_sync_enabled`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub pull_sync_url: Option<String>,
+    /// Allowlist of domains TS may call for this partner's pull sync.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub pull_sync_allowed_domains: Vec<String>,
+    /// Seconds between pull sync refreshes (default 86400).
+    pub pull_sync_ttl_sec: u64,
+    /// Max pull sync calls per EC hash per partner per hour.
+    pub pull_sync_rate_limit: u32,
+    /// Outbound bearer token for pull sync requests.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ts_pull_token: Option<String>,
+}
+
 /// Validates a partner ID format and checks against reserved names.
 ///
 /// # Errors
@@ -55,6 +108,66 @@ pub fn validate_partner_id(id: &str) -> Result<(), String> {
     Ok(())
 }
 
+/// Validates pull sync configuration consistency.
+///
+/// When `pull_sync_enabled` is true, both `pull_sync_url` and
+/// `ts_pull_token` must be present, and the URL's hostname must
+/// appear in `pull_sync_allowed_domains`.
+///
+/// # Errors
+///
+/// Returns a descriptive error string on validation failure.
+pub fn validate_pull_sync_config(record: &PartnerRecord) -> Result<(), String> {
+    if !record.pull_sync_enabled {
+        return Ok(());
+    }
+
+    let url_str = record.pull_sync_url.as_deref().unwrap_or("");
+    if url_str.is_empty() {
+        return Err(
+            "pull_sync_url and ts_pull_token are required when pull_sync_enabled is true"
+                .to_owned(),
+        );
+    }
+
+    if record.ts_pull_token.as_deref().unwrap_or("").is_empty() {
+        return Err(
+            "pull_sync_url and ts_pull_token are required when pull_sync_enabled is true"
+                .to_owned(),
+        );
+    }
+
+    // Validate that the pull sync URL uses HTTPS (bearer tokens must not
+    // travel over plaintext).
+    let parsed =
+        url::Url::parse(url_str).map_err(|e| format!("pull_sync_url is not a valid URL: {e}"))?;
+    if parsed.scheme() != "https" {
+        return Err(format!(
+            "pull_sync_url must use HTTPS, got scheme '{}'",
+            parsed.scheme()
+        ));
+    }
+
+    // Validate that the pull sync URL hostname is in the allowed domains.
+    let host = parsed
+        .host_str()
+        .ok_or("pull_sync_url has no hostname")?
+        .trim_end_matches('.')
+        .to_ascii_lowercase();
+
+    let allowed: HashSet<String> = record
+        .pull_sync_allowed_domains
+        .iter()
+        .map(|domain| domain.trim().trim_end_matches('.').to_ascii_lowercase())
+        .collect();
+
+    if !allowed.contains(&host) {
+        return Err("pull_sync_url domain must be in pull_sync_allowed_domains".to_owned());
+    }
+
+    Ok(())
+}
+
 /// Computes the SHA-256 hex digest of an API key.
 #[must_use]
 pub fn hash_api_key(api_key: &str) -> String {
@@ -63,87 +176,644 @@ pub fn hash_api_key(api_key: &str) -> String {
     hex::encode(hasher.finalize())
 }
 
+/// Wraps a Fastly KV Store for partner registry operations.
+///
+/// Partner records are keyed by `partner_id`. A secondary index
+/// `apikey:{sha256_hex}` maps API key hashes to partner IDs for
+/// O(1) auth lookups during batch sync.
+pub struct PartnerStore {
+    store_name: String,
+}
+
+impl PartnerStore {
+    /// Creates a new partner store backed by the named KV store.
+    #[must_use]
+    pub fn new(store_name: impl Into<String>) -> Self {
+        Self {
+            store_name: store_name.into(),
+        }
+    }
+
+    /// Returns the configured store name.
+    #[must_use]
+    pub fn store_name(&self) -> &str {
+        &self.store_name
+    }
+
+    /// Opens the underlying Fastly KV store.
+    fn open_store(&self) -> Result<KVStore, Report<TrustedServerError>> {
+        KVStore::open(&self.store_name)
+            .change_context(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: "Failed to open partner store".to_owned(),
+            })?
+            .ok_or_else(|| {
+                Report::new(TrustedServerError::KvStore {
+                    store_name: self.store_name.clone(),
+                    message: "Partner store not found".to_owned(),
+                })
+            })
+    }
+
+    /// Reads a partner record by ID.
+    ///
+    /// Returns `Ok(None)` when the partner is not registered.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`TrustedServerError::KvStore`] on store or deserialization failure.
+    pub fn get(
+        &self,
+        partner_id: &str,
+    ) -> Result<Option<PartnerRecord>, Report<TrustedServerError>> {
+        let store = self.open_store()?;
+        let mut response = match store.lookup(partner_id) {
+            Ok(resp) => resp,
+            Err(fastly::kv_store::KVStoreError::ItemNotFound) => return Ok(None),
+            Err(err) => {
+                return Err(
+                    Report::new(err).change_context(TrustedServerError::KvStore {
+                        store_name: self.store_name.clone(),
+                        message: format!("Failed to read partner '{partner_id}'"),
+                    }),
+                );
+            }
+        };
+
+        let body_bytes = response.take_body_bytes();
+        let record: PartnerRecord =
+            serde_json::from_slice(&body_bytes).change_context(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: format!("Failed to deserialize partner '{partner_id}'"),
+            })?;
+
+        Ok(Some(record))
+    }
+
+    /// Lists all registered partner records.
+    ///
+    /// Scans the partner KV store and returns records for non-index keys.
+    /// Secondary index entries (e.g. `apikey:*`) are skipped.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`TrustedServerError::KvStore`] on list, lookup, or
+    /// deserialization failure.
+    pub fn list_registered(&self) -> Result<Vec<PartnerRecord>, Report<TrustedServerError>> {
+        let store = self.open_store()?;
+        let mut records = Vec::new();
+
+        for page in store.build_list().limit(1000).iter() {
+            let page = page.change_context(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: "Failed to list partner keys".to_owned(),
+            })?;
+
+            for key in page.keys() {
+                if key.starts_with(APIKEY_INDEX_PREFIX) {
+                    continue;
+                }
+
+                let mut response = match store.lookup(key) {
+                    Ok(resp) => resp,
+                    Err(fastly::kv_store::KVStoreError::ItemNotFound) => continue,
+                    Err(err) => {
+                        return Err(
+                            Report::new(err).change_context(TrustedServerError::KvStore {
+                                store_name: self.store_name.clone(),
+                                message: format!("Failed to read partner '{key}' while listing"),
+                            }),
+                        );
+                    }
+                };
+
+                let body_bytes = response.take_body_bytes();
+                let record = serde_json::from_slice::<PartnerRecord>(&body_bytes).change_context(
+                    TrustedServerError::KvStore {
+                        store_name: self.store_name.clone(),
+                        message: format!("Failed to deserialize partner '{key}' while listing"),
+                    },
+                )?;
+
+                records.push(record);
+            }
+        }
+
+        Ok(records)
+    }
+
+    /// Writes or updates a partner record and maintains the API key index.
+    ///
+    /// Returns `true` if this was a new partner (create), `false` if an
+    /// existing partner was updated.
+    ///
+    /// Index maintenance order:
+    /// 1. Read existing `apikey:` index value for rollback
+    /// 2. Write new `apikey:` index
+    /// 3. Write primary record
+    /// 4. Delete old `apikey:` index (if key rotated)
+    ///
+    /// Writes are still **not fully atomic**, but this order ensures
+    /// registration does not return success after a failed index write and
+    /// performs best-effort rollback when primary write fails.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`TrustedServerError::KvStore`] on store failure.
+    pub fn upsert(&self, record: &PartnerRecord) -> Result<bool, Report<TrustedServerError>> {
+        let store = self.open_store()?;
+
+        // Read existing record to detect API key rotation.
+        let existing = match store.lookup(&record.id) {
+            Ok(mut resp) => {
+                let bytes = resp.take_body_bytes();
+                serde_json::from_slice::<PartnerRecord>(&bytes).ok()
+            }
+            Err(_) => None,
+        };
+
+        let is_create = existing.is_none();
+        let old_api_key_hash = existing.as_ref().map(|r| r.api_key_hash.clone());
+
+        let index_key = format!("{APIKEY_INDEX_PREFIX}{}", record.api_key_hash);
+        let previous_index_partner_id = self.read_index_partner_id(&store, &index_key)?;
+
+        // 1. Write new API key index.
+        store
+            .build_insert()
+            .execute(&index_key, record.id.as_str())
+            .change_context(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: format!(
+                    "Failed to write API key index for partner '{}' (hash '{}')",
+                    record.id, record.api_key_hash
+                ),
+            })?;
+
+        // 2. Write primary record.
+        let body = serde_json::to_string(record).change_context(TrustedServerError::KvStore {
+            store_name: self.store_name.clone(),
+            message: format!("Failed to serialize partner '{}'", record.id),
+        })?;
+
+        if let Err(err) = store.build_insert().execute(&record.id, body) {
+            self.restore_previous_index_mapping(
+                &store,
+                &index_key,
+                previous_index_partner_id.as_deref(),
+                &record.id,
+            );
+            return Err(
+                Report::new(err).change_context(TrustedServerError::KvStore {
+                    store_name: self.store_name.clone(),
+                    message: format!("Failed to write partner '{}'", record.id),
+                }),
+            );
+        }
+
+        // 3. Delete old API key index if key rotated.
+        if let Some(ref old_hash) = old_api_key_hash {
+            if *old_hash != record.api_key_hash {
+                let old_key = format!("{APIKEY_INDEX_PREFIX}{old_hash}");
+                if let Err(err) = store.delete(&old_key) {
+                    log::warn!(
+                        "Failed to delete old API key index for partner '{}': {err:?}",
+                        record.id,
+                    );
+                }
+            }
+        }
+
+        Ok(is_create)
+    }
+
+    fn read_index_partner_id(
+        &self,
+        store: &KVStore,
+        index_key: &str,
+    ) -> Result<Option<String>, Report<TrustedServerError>> {
+        let mut response = match store.lookup(index_key) {
+            Ok(resp) => resp,
+            Err(fastly::kv_store::KVStoreError::ItemNotFound) => return Ok(None),
+            Err(err) => {
+                return Err(
+                    Report::new(err).change_context(TrustedServerError::KvStore {
+                        store_name: self.store_name.clone(),
+                        message: format!(
+                            "Failed to read existing API key index before upsert ('{index_key}')"
+                        ),
+                    }),
+                );
+            }
+        };
+
+        let partner_id = String::from_utf8(response.take_body_bytes()).map_err(|_| {
+            Report::new(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: format!(
+                    "Existing API key index value is not valid UTF-8 before upsert ('{index_key}')"
+                ),
+            })
+        })?;
+
+        Ok(Some(partner_id))
+    }
+
+    fn restore_previous_index_mapping(
+        &self,
+        store: &KVStore,
+        index_key: &str,
+        previous_partner_id: Option<&str>,
+        partner_id: &str,
+    ) {
+        if let Some(previous_partner_id) = previous_partner_id {
+            if previous_partner_id == partner_id {
+                return;
+            }
+
+            if let Err(err) = store.build_insert().execute(index_key, previous_partner_id) {
+                log::warn!(
+                    "Failed to restore previous API key index mapping after write failure for partner '{}': {err:?}",
+                    partner_id,
+                );
+            }
+            return;
+        }
+
+        match store.delete(index_key) {
+            Ok(()) | Err(fastly::kv_store::KVStoreError::ItemNotFound) => {}
+            Err(err) => {
+                log::warn!(
+                    "Failed to roll back API key index after write failure for partner '{}': {err:?}",
+                    partner_id,
+                );
+            }
+        }
+    }
+
+    /// Looks up a partner by API key hash using the `apikey:` secondary index.
+    ///
+    /// After resolving the index to a partner ID, re-verifies that the
+    /// stored `api_key_hash` matches the lookup hash (guards against stale
+    /// index entries from key rotation).
+    ///
+    /// # Errors
+    ///
+    /// Returns [`TrustedServerError::KvStore`] on store failure.
+    pub fn find_by_api_key_hash(
+        &self,
+        hash: &str,
+    ) -> Result<Option<PartnerRecord>, Report<TrustedServerError>> {
+        let store = self.open_store()?;
+
+        // Look up the secondary index.
+        let index_key = format!("{APIKEY_INDEX_PREFIX}{hash}");
+        let mut index_response = match store.lookup(&index_key) {
+            Ok(resp) => resp,
+            Err(fastly::kv_store::KVStoreError::ItemNotFound) => return Ok(None),
+            Err(err) => {
+                return Err(
+                    Report::new(err).change_context(TrustedServerError::KvStore {
+                        store_name: self.store_name.clone(),
+                        message: format!("Failed to read API key index for hash '{hash}'"),
+                    }),
+                );
+            }
+        };
+
+        let partner_id = String::from_utf8(index_response.take_body_bytes()).map_err(|_| {
+            Report::new(TrustedServerError::KvStore {
+                store_name: self.store_name.clone(),
+                message: format!("API key index value for hash '{hash}' is not valid UTF-8"),
+            })
+        })?;
+
+        // Fetch the actual partner record.
+        let record = match self.get(&partner_id)? {
+            Some(r) => r,
+            None => {
+                // Stale index — partner was deleted.
+                log::warn!(
+                    "API key index points to non-existent partner '{partner_id}' (stale index)"
+                );
+                return Ok(None);
+            }
+        };
+
+        // Verify the stored hash matches — guards against stale index from
+        // key rotation.
+        if record.api_key_hash != hash {
+            log::warn!(
+                "API key hash mismatch for partner '{}' (stale index after key rotation)",
+                record.id,
+            );
+            return Ok(None);
+        }
+
+        Ok(Some(record))
+    }
+
+    /// Verifies an API key against the stored hash for a given partner.
+    ///
+    /// Uses SHA-256 hashing and constant-time comparison to prevent
+    /// timing attacks.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`TrustedServerError::KvStore`] if the partner lookup fails.
+    pub fn verify_api_key(
+        &self,
+        partner_id: &str,
+        api_key: &str,
+    ) -> Result<bool, Report<TrustedServerError>> {
+        let record = match self.get(partner_id)? {
+            Some(r) => r,
+            None => return Ok(false),
+        };
+
+        let incoming_hash = hash_api_key(api_key);
+        let stored_bytes = record.api_key_hash.as_bytes();
+        let incoming_bytes = incoming_hash.as_bytes();
+
+        Ok(stored_bytes.ct_eq(incoming_bytes).into())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
     #[test]
-    fn validate_partner_id_accepts_valid_ids() {
-        assert!(
-            validate_partner_id("ssp_x").is_ok(),
-            "should accept underscored ID"
-        );
-        assert!(
-            validate_partner_id("dsp-y").is_ok(),
-            "should accept hyphenated ID"
-        );
-        assert!(
-            validate_partner_id("liveramp").is_ok(),
-            "should accept lowercase alpha"
-        );
-        assert!(
-            validate_partner_id("id5").is_ok(),
-            "should accept alphanumeric"
-        );
+    fn partner_record_serialization_roundtrip() {
+        let record = PartnerRecord {
+            id: "ssp_x".to_owned(),
+            name: "SSP Example".to_owned(),
+            allowed_return_domains: vec!["sync.example-ssp.com".to_owned()],
+            api_key_hash: hash_api_key("test-api-key"),
+            bidstream_enabled: true,
+            source_domain: "example-ssp.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: false,
+            pull_sync_url: None,
+            pull_sync_allowed_domains: vec![],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: None,
+        };
+
+        let json = serde_json::to_string(&record).expect("should serialize");
+        let deserialized: PartnerRecord = serde_json::from_str(&json).expect("should deserialize");
+
+        assert_eq!(deserialized, record);
     }
 
     #[test]
-    fn validate_partner_id_rejects_invalid_ids() {
-        assert!(validate_partner_id("").is_err(), "should reject empty ID");
-        assert!(
-            validate_partner_id("SSP").is_err(),
-            "should reject uppercase"
-        );
+    fn hash_api_key_is_deterministic() {
+        let h1 = hash_api_key("my-secret-key");
+        let h2 = hash_api_key("my-secret-key");
+        assert_eq!(h1, h2);
+        assert_eq!(h1.len(), 64, "should be 64-char hex SHA-256");
+    }
+
+    #[test]
+    fn hash_api_key_differs_for_different_keys() {
+        let h1 = hash_api_key("key-a");
+        let h2 = hash_api_key("key-b");
+        assert_ne!(h1, h2);
+    }
+
+    #[test]
+    fn validate_partner_id_accepts_valid() {
+        assert!(validate_partner_id("ssp_x").is_ok());
+        assert!(validate_partner_id("liveramp").is_ok());
+        assert!(validate_partner_id("a-b_c-1").is_ok());
+        assert!(validate_partner_id("a").is_ok());
+    }
+
+    #[test]
+    fn validate_partner_id_rejects_uppercase() {
+        let err = validate_partner_id("SSP_X").unwrap_err();
         assert!(
-            validate_partner_id("a".repeat(33).as_str()).is_err(),
-            "should reject >32 chars"
+            err.contains("must match"),
+            "should reject uppercase, got: {err}"
         );
+    }
+
+    #[test]
+    fn validate_partner_id_rejects_too_long() {
+        let long = "a".repeat(33);
+        let err = validate_partner_id(&long).unwrap_err();
         assert!(
-            validate_partner_id("has space").is_err(),
-            "should reject spaces"
+            err.contains("must match"),
+            "should reject >32 chars, got: {err}"
         );
     }
 
     #[test]
-    fn validate_partner_id_rejects_reserved_ids() {
+    fn validate_partner_id_rejects_empty() {
+        let err = validate_partner_id("").unwrap_err();
         assert!(
-            validate_partner_id("ec").is_err(),
-            "should reject reserved 'ec'"
+            err.contains("must match"),
+            "should reject empty, got: {err}"
         );
+    }
+
+    #[test]
+    fn validate_partner_id_rejects_reserved() {
+        for reserved in RESERVED_PARTNER_IDS {
+            let err = validate_partner_id(reserved).unwrap_err();
+            assert!(
+                err.contains("reserved"),
+                "should reject '{reserved}', got: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn validate_partner_id_rejects_special_chars() {
+        assert!(validate_partner_id("ssp.x").is_err(), "should reject dots");
         assert!(
-            validate_partner_id("ts").is_err(),
-            "should reject reserved 'ts'"
+            validate_partner_id("ssp x").is_err(),
+            "should reject spaces"
         );
         assert!(
-            validate_partner_id("eids").is_err(),
-            "should reject reserved 'eids'"
+            validate_partner_id("ssp/x").is_err(),
+            "should reject slashes"
         );
     }
 
     #[test]
-    fn hash_api_key_produces_hex_digest() {
-        let hash = hash_api_key("test-key");
-        assert_eq!(hash.len(), 64, "should produce 64-char hex digest");
+    fn validate_pull_sync_ok_when_disabled() {
+        let record = PartnerRecord {
+            id: "test".to_owned(),
+            name: "Test".to_owned(),
+            allowed_return_domains: vec![],
+            api_key_hash: String::new(),
+            bidstream_enabled: false,
+            source_domain: "test.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: false,
+            pull_sync_url: None,
+            pull_sync_allowed_domains: vec![],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: None,
+        };
+        assert!(validate_pull_sync_config(&record).is_ok());
+    }
+
+    #[test]
+    fn validate_pull_sync_rejects_missing_url() {
+        let record = PartnerRecord {
+            id: "test".to_owned(),
+            name: "Test".to_owned(),
+            allowed_return_domains: vec![],
+            api_key_hash: String::new(),
+            bidstream_enabled: false,
+            source_domain: "test.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: true,
+            pull_sync_url: None,
+            pull_sync_allowed_domains: vec![],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: Some("token".to_owned()),
+        };
+        let err = validate_pull_sync_config(&record).unwrap_err();
+        assert!(err.contains("pull_sync_url"), "got: {err}");
+    }
+
+    #[test]
+    fn validate_pull_sync_rejects_missing_token() {
+        let record = PartnerRecord {
+            id: "test".to_owned(),
+            name: "Test".to_owned(),
+            allowed_return_domains: vec![],
+            api_key_hash: String::new(),
+            bidstream_enabled: false,
+            source_domain: "test.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: true,
+            pull_sync_url: Some("https://sync.test.com/pull".to_owned()),
+            pull_sync_allowed_domains: vec!["sync.test.com".to_owned()],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: None,
+        };
+        let err = validate_pull_sync_config(&record).unwrap_err();
+        assert!(err.contains("ts_pull_token"), "got: {err}");
+    }
+
+    #[test]
+    fn validate_pull_sync_rejects_http_scheme() {
+        let record = PartnerRecord {
+            id: "test".to_owned(),
+            name: "Test".to_owned(),
+            allowed_return_domains: vec![],
+            api_key_hash: String::new(),
+            bidstream_enabled: false,
+            source_domain: "test.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: true,
+            pull_sync_url: Some("http://sync.test.com/pull".to_owned()),
+            pull_sync_allowed_domains: vec!["sync.test.com".to_owned()],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: Some("token".to_owned()),
+        };
+        let err = validate_pull_sync_config(&record).unwrap_err();
         assert!(
-            hash.chars().all(|c| c.is_ascii_hexdigit()),
-            "should only contain hex characters"
+            err.contains("HTTPS"),
+            "should reject HTTP scheme, got: {err}"
         );
     }
 
     #[test]
-    fn hash_api_key_is_deterministic() {
-        let hash1 = hash_api_key("same-key");
-        let hash2 = hash_api_key("same-key");
-        assert_eq!(hash1, hash2, "should produce same hash for same input");
+    fn validate_pull_sync_rejects_url_not_in_allowed_domains() {
+        let record = PartnerRecord {
+            id: "test".to_owned(),
+            name: "Test".to_owned(),
+            allowed_return_domains: vec![],
+            api_key_hash: String::new(),
+            bidstream_enabled: false,
+            source_domain: "test.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: true,
+            pull_sync_url: Some("https://evil.com/pull".to_owned()),
+            pull_sync_allowed_domains: vec!["sync.test.com".to_owned()],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: Some("token".to_owned()),
+        };
+        let err = validate_pull_sync_config(&record).unwrap_err();
+        assert!(err.contains("pull_sync_allowed_domains"), "got: {err}");
     }
 
     #[test]
-    fn hash_api_key_differs_for_different_keys() {
-        let hash1 = hash_api_key("key-a");
-        let hash2 = hash_api_key("key-b");
-        assert_ne!(
-            hash1, hash2,
-            "should produce different hashes for different inputs"
+    fn validate_pull_sync_accepts_valid_config() {
+        let record = PartnerRecord {
+            id: "test".to_owned(),
+            name: "Test".to_owned(),
+            allowed_return_domains: vec![],
+            api_key_hash: String::new(),
+            bidstream_enabled: false,
+            source_domain: "test.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: true,
+            pull_sync_url: Some("https://sync.test.com/pull".to_owned()),
+            pull_sync_allowed_domains: vec!["sync.test.com".to_owned()],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: Some("token".to_owned()),
+        };
+        assert!(validate_pull_sync_config(&record).is_ok());
+    }
+
+    #[test]
+    fn optional_fields_omitted_from_json() {
+        let record = PartnerRecord {
+            id: "test".to_owned(),
+            name: "Test".to_owned(),
+            allowed_return_domains: vec![],
+            api_key_hash: String::new(),
+            bidstream_enabled: false,
+            source_domain: "test.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: false,
+            pull_sync_url: None,
+            pull_sync_allowed_domains: vec![],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: None,
+        };
+        let json = serde_json::to_string(&record).expect("should serialize");
+        assert!(
+            !json.contains("pull_sync_url"),
+            "None pull_sync_url should be omitted"
+        );
+        assert!(
+            !json.contains("ts_pull_token"),
+            "None ts_pull_token should be omitted"
+        );
+        assert!(
+            !json.contains("pull_sync_allowed_domains"),
+            "empty pull_sync_allowed_domains should be omitted"
         );
     }
 }
diff --git a/crates/trusted-server-core/src/ec/sync_pixel.rs b/crates/trusted-server-core/src/ec/sync_pixel.rs
new file mode 100644
index 00000000..60c48e72
--- /dev/null
+++ b/crates/trusted-server-core/src/ec/sync_pixel.rs
@@ -0,0 +1,424 @@
+//! Pixel sync endpoint (`GET /sync`).
+
+use error_stack::{Report, ResultExt};
+use fastly::erl::{CounterDuration, RateCounter};
+use fastly::http::StatusCode;
+use fastly::{Request, Response};
+use url::Url;
+
+use crate::consent::{allows_ec_creation, gpp, tcf, ConsentContext};
+use crate::error::TrustedServerError;
+use crate::settings::Settings;
+
+use super::generation::{ec_hash, is_valid_ec_id};
+use super::kv::KvIdentityGraph;
+use super::partner::{PartnerRecord, PartnerStore};
+use super::EcContext;
+
+const RATE_COUNTER_NAME: &str = "counter_store";
+
+/// Handles `GET /sync` pixel sync requests.
+///
+/// # Errors
+///
+/// Returns [`TrustedServerError`] when request validation fails (`400`) or
+/// required stores are unavailable (`503`).
+pub fn handle_sync(
+    _settings: &Settings,
+    kv: &KvIdentityGraph,
+    partner_store: &PartnerStore,
+    req: &Request,
+    ec_context: &mut EcContext,
+) -> Result<Response, Report<TrustedServerError>> {
+    let query = SyncQuery::parse(req)?;
+
+    let partner = partner_store.get(&query.partner)?.ok_or_else(|| {
+        Report::new(TrustedServerError::BadRequest {
+            message: format!("unknown partner '{}'", query.partner),
+        })
+    })?;
+
+    let return_url = validate_return_url(&query.return_url, &partner)?;
+
+    let Some(cookie_ec_id) = ec_context
+        .existing_cookie_ec_id()
+        .filter(|v| is_valid_ec_id(v))
+        .map(str::to_owned)
+    else {
+        return Ok(redirect_with_status(&return_url, "0", Some("no_ec")));
+    };
+
+    if ec_context.consent().is_empty() {
+        if let Some(consent_query) = query.consent.as_deref() {
+            if let Some(fallback) =
+                decode_query_fallback_consent(ec_context.consent(), consent_query)
+            {
+                *ec_context.consent_mut() = fallback;
+            }
+        }
+    }
+
+    if !allows_ec_creation(ec_context.consent()) {
+        return Ok(redirect_with_status(&return_url, "0", Some("no_consent")));
+    }
+
+    let hash = ec_hash(&cookie_ec_id);
+    let limiter = FastlyRateLimiter::new(RATE_COUNTER_NAME);
+    if limiter.exceeded(&format!("{}:{hash}", partner.id), partner.sync_rate_limit)? {
+        return Ok(Response::from_status(StatusCode::TOO_MANY_REQUESTS)
+            .with_body_text_plain("rate_limit_exceeded"));
+    }
+
+    if let Err(err) = kv.upsert_partner_id(hash, &partner.id, &query.uid) {
+        log::warn!(
+            "Pixel sync write failed for partner '{}' and hash '{}': {err:?}",
+            partner.id,
+            hash,
+        );
+        return Ok(redirect_with_status(&return_url, "0", Some("write_failed")));
+    }
+
+    Ok(redirect_with_status(&return_url, "1", None))
+}
+
+#[derive(Debug)]
+struct SyncQuery {
+    partner: String,
+    uid: String,
+    return_url: String,
+    consent: Option<String>,
+}
+
+impl SyncQuery {
+    fn parse(req: &Request) -> Result<Self, Report<TrustedServerError>> {
+        let mut partner = None;
+        let mut uid = None;
+        let mut return_url = None;
+        let mut consent = None;
+
+        let raw_query = req.get_query_str().unwrap_or("");
+        for (key, value) in url::form_urlencoded::parse(raw_query.as_bytes()) {
+            match key.as_ref() {
+                "partner" => partner = Some(value.into_owned()),
+                "uid" => uid = Some(value.into_owned()),
+                "return" => return_url = Some(value.into_owned()),
+                "consent" => consent = Some(value.into_owned()),
+                _ => {}
+            }
+        }
+
+        Ok(Self {
+            partner: required_query_param(partner, "partner")?,
+            uid: required_query_param(uid, "uid")?,
+            return_url: required_query_param(return_url, "return")?,
+            consent,
+        })
+    }
+}
+
+fn required_query_param(
+    value: Option<String>,
+    key: &str,
+) -> Result<String, Report<TrustedServerError>> {
+    let Some(value) = value else {
+        return Err(Report::new(TrustedServerError::BadRequest {
+            message: format!("missing required query parameter '{key}'"),
+        }));
+    };
+
+    if value.trim().is_empty() {
+        return Err(Report::new(TrustedServerError::BadRequest {
+            message: format!("query parameter '{key}' must not be empty"),
+        }));
+    }
+
+    Ok(value)
+}
+
+fn validate_return_url(
+    return_url: &str,
+    partner: &PartnerRecord,
+) -> Result<Url, Report<TrustedServerError>> {
+    let parsed = Url::parse(return_url).change_context(TrustedServerError::BadRequest {
+        message: "return URL must be a valid absolute URL".to_owned(),
+    })?;
+
+    let host = parsed
+        .host_str()
+        .ok_or_else(|| {
+            Report::new(TrustedServerError::BadRequest {
+                message: "return URL must include a hostname".to_owned(),
+            })
+        })?
+        .trim_end_matches('.')
+        .to_ascii_lowercase();
+
+    let allowed = partner
+        .allowed_return_domains
+        .iter()
+        .map(|domain| domain.trim().trim_end_matches('.').to_ascii_lowercase())
+        .any(|domain| domain == host);
+
+    if !allowed {
+        return Err(Report::new(TrustedServerError::BadRequest {
+            message: format!(
+                "return URL host '{host}' is not allowed for partner '{}'",
+                partner.id
+            ),
+        }));
+    }
+
+    Ok(parsed)
+}
+
+fn redirect_with_status(return_url: &Url, synced: &str, reason: Option<&str>) -> Response {
+    let mut url = return_url.clone();
+    {
+        let mut query = url.query_pairs_mut();
+        query.append_pair("ts_synced", synced);
+        if let Some(reason) = reason {
+            query.append_pair("ts_reason", reason);
+        }
+    }
+
+    Response::from_status(StatusCode::FOUND).with_header("location", url.as_str())
+}
+
+fn decode_query_fallback_consent(
+    base: &ConsentContext,
+    raw_consent: &str,
+) -> Option<ConsentContext> {
+    if raw_consent.trim().is_empty() {
+        return None;
+    }
+
+    let mut consent = ConsentContext {
+        jurisdiction: base.jurisdiction.clone(),
+        gpc: base.gpc,
+        ..ConsentContext::default()
+    };
+
+    if raw_consent.contains('~') || raw_consent.starts_with("DB") {
+        match gpp::decode_gpp_string(raw_consent) {
+            Ok(decoded) => {
+                consent.raw_gpp_string = Some(raw_consent.to_owned());
+                consent.gpp_section_ids = Some(decoded.section_ids.clone());
+                consent.tcf = decoded.eu_tcf.clone();
+                consent.gpp = Some(decoded);
+                consent.gdpr_applies = consent
+                    .gpp_section_ids
+                    .as_ref()
+                    .is_some_and(|sids| sids.contains(&2));
+                return Some(consent);
+            }
+            Err(err) => {
+                log::warn!("Failed to decode GPP consent query fallback: {err:?}");
+                return None;
+            }
+        }
+    }
+
+    match tcf::decode_tc_string(raw_consent) {
+        Ok(decoded) => {
+            consent.raw_tc_string = Some(raw_consent.to_owned());
+            consent.tcf = Some(decoded);
+            consent.gdpr_applies = true;
+            Some(consent)
+        }
+        Err(err) => {
+            log::warn!("Failed to decode TCF consent query fallback: {err:?}");
+            None
+        }
+    }
+}
+
+trait RateLimiter {
+    fn exceeded(&self, key: &str, hourly_limit: u32) -> Result<bool, Report<TrustedServerError>>;
+}
+
+struct FastlyRateLimiter {
+    counter: RateCounter,
+}
+
+impl FastlyRateLimiter {
+    fn new(counter_name: &str) -> Self {
+        Self {
+            counter: RateCounter::open(counter_name),
+        }
+    }
+}
+
+impl RateLimiter for FastlyRateLimiter {
+    fn exceeded(&self, key: &str, hourly_limit: u32) -> Result<bool, Report<TrustedServerError>> {
+        // Fastly's public rate-counter API currently exposes windows up to 60s.
+        // Approximate the story's 1h limit by converting to a per-minute budget.
+        //
+        // Follow-up: move to exact 1-hour enforcement once platform counters
+        // expose longer windows or we add a dedicated KV-backed hour bucket.
+        let per_minute_limit = hourly_limit.saturating_add(59) / 60;
+        let per_minute_limit = per_minute_limit.max(1);
+
+        let current = self
+            .counter
+            .lookup_count(key, CounterDuration::SixtySecs)
+            .map_err(|e| {
+                Report::new(TrustedServerError::KvStore {
+                    store_name: RATE_COUNTER_NAME.to_owned(),
+                    message: format!("Failed to read sync rate counter: {e}"),
+                })
+            })?;
+
+        if current >= per_minute_limit {
+            return Ok(true);
+        }
+
+        self.counter.increment(key, 1).map_err(|e| {
+            Report::new(TrustedServerError::KvStore {
+                store_name: RATE_COUNTER_NAME.to_owned(),
+                message: format!("Failed to increment sync rate counter: {e}"),
+            })
+        })?;
+
+        Ok(false)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn sample_partner() -> PartnerRecord {
+        PartnerRecord {
+            id: "ssp_x".to_owned(),
+            name: "SSP X".to_owned(),
+            allowed_return_domains: vec!["sync.example.com".to_owned()],
+            api_key_hash: "deadbeef".to_owned(),
+            bidstream_enabled: false,
+            source_domain: "ssp.example.com".to_owned(),
+            openrtb_atype: 3,
+            sync_rate_limit: 100,
+            batch_rate_limit: 60,
+            pull_sync_enabled: false,
+            pull_sync_url: None,
+            pull_sync_allowed_domains: vec![],
+            pull_sync_ttl_sec: 86400,
+            pull_sync_rate_limit: 10,
+            ts_pull_token: None,
+        }
+    }
+
+    #[test]
+    fn redirect_appends_query_when_url_has_none() {
+        let url = Url::parse("https://sync.example.com/return").expect("should parse URL");
+        let response = redirect_with_status(&url, "1", None);
+        let location = response
+            .get_header("location")
+            .expect("should set location header")
+            .to_str()
+            .expect("should convert location to UTF-8");
+
+        assert_eq!(
+            location, "https://sync.example.com/return?ts_synced=1",
+            "should append query with ? when missing"
+        );
+    }
+
+    #[test]
+    fn redirect_appends_query_when_url_already_has_query() {
+        let url = Url::parse("https://sync.example.com/return?foo=bar").expect("should parse URL");
+        let response = redirect_with_status(&url, "0", Some("no_ec"));
+        let location = response
+            .get_header("location")
+            .expect("should set location header")
+            .to_str()
+            .expect("should convert location to UTF-8");
+
+        assert_eq!(
+            location, "https://sync.example.com/return?foo=bar&ts_synced=0&ts_reason=no_ec",
+            "should append sync status after existing query"
+        );
+    }
+
+    #[test]
+    fn fallback_decodes_tcf() {
+        let base = ConsentContext::default();
+        let decoded =
+            decode_query_fallback_consent(&base, "CPXxGfAPXxGfAAfKABENB-CgAAAAAAAAAAYgAAAAAAAA")
+                .expect("should decode TCF fallback");
+
+        assert!(
+            decoded.raw_tc_string.is_some(),
+            "should store raw TC string"
+        );
+    }
+
+    #[test]
+    fn query_parse_rejects_missing_required_param() {
+        let req = Request::new("GET", "https://edge.example.com/sync?partner=ssp&uid=u1");
+        let err = SyncQuery::parse(&req).expect_err("should fail when return param is missing");
+        assert!(
+            err.to_string()
+                .contains("missing required query parameter 'return'"),
+            "should mention missing required return parameter"
+        );
+    }
+
+    #[test]
+    fn query_parse_rejects_empty_required_param() {
+        let req = Request::new(
+            "GET",
+            "https://edge.example.com/sync?partner=ssp&uid=u1&return=   ",
+        );
+        let err = SyncQuery::parse(&req).expect_err("should fail when return param is empty");
+        assert!(
+            err.to_string()
+                .contains("query parameter 'return' must not be empty"),
+            "should reject empty required return parameter"
+        );
+    }
+
+    #[test]
+    fn return_url_validation_rejects_subdomain_spoofing() {
+        let partner = sample_partner();
+        let err = validate_return_url("https://a.sync.example.com/callback", &partner)
+            .expect_err("should reject return host not exactly allowlisted");
+
+        assert!(
+            err.to_string().contains("is not allowed"),
+            "should reject non-exact allowlist host"
+        );
+    }
+
+    #[test]
+    fn return_url_validation_rejects_relative_url() {
+        let partner = sample_partner();
+        let err = validate_return_url("/callback", &partner)
+            .expect_err("should reject non-absolute return URL");
+        assert!(
+            err.to_string().contains("valid absolute URL"),
+            "should require absolute return URLs"
+        );
+    }
+
+    #[test]
+    fn fallback_decodes_gpp() {
+        let base = ConsentContext::default();
+        let decoded = decode_query_fallback_consent(&base, "DBABTA~1YNN")
+            .expect("should decode valid GPP fallback");
+
+        assert!(
+            decoded.raw_gpp_string.is_some(),
+            "should store raw GPP string"
+        );
+    }
+
+    #[test]
+    fn fallback_returns_none_for_invalid_consent_string() {
+        let base = ConsentContext::default();
+        let decoded = decode_query_fallback_consent(&base, "not-a-valid-consent");
+        assert!(
+            decoded.is_none(),
+            "should ignore undecodable consent fallback"
+        );
+    }
+}
diff --git a/crates/trusted-server-core/src/integrations/datadome.rs b/crates/trusted-server-core/src/integrations/datadome.rs
index da3a64a2..0ce83b2f 100644
--- a/crates/trusted-server-core/src/integrations/datadome.rs
+++ b/crates/trusted-server-core/src/integrations/datadome.rs
@@ -71,6 +71,7 @@ use crate::integrations::{
     AttributeRewriteAction, IntegrationAttributeContext, IntegrationAttributeRewriter,
     IntegrationEndpoint, IntegrationProxy, IntegrationRegistration,
 };
+use crate::platform::RuntimeServices;
 use crate::settings::{IntegrationConfig, Settings};
 
 const DATADOME_INTEGRATION_ID: &str = "datadome";
@@ -404,6 +405,7 @@ impl IntegrationProxy for DataDomeIntegration {
     async fn handle(
         &self,
         _settings: &Settings,
+        _services: &RuntimeServices,
         req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let path = req.get_path();
diff --git a/crates/trusted-server-core/src/integrations/didomi.rs b/crates/trusted-server-core/src/integrations/didomi.rs
index 7042af70..2cada38f 100644
--- a/crates/trusted-server-core/src/integrations/didomi.rs
+++ b/crates/trusted-server-core/src/integrations/didomi.rs
@@ -11,6 +11,7 @@ use validator::Validate;
 use crate::backend::BackendConfig;
 use crate::error::TrustedServerError;
 use crate::integrations::{IntegrationEndpoint, IntegrationProxy, IntegrationRegistration};
+use crate::platform::RuntimeServices;
 use crate::settings::{IntegrationConfig, Settings};
 
 const DIDOMI_INTEGRATION_ID: &str = "didomi";
@@ -198,6 +199,7 @@ impl IntegrationProxy for DidomiIntegration {
     async fn handle(
         &self,
         _settings: &Settings,
+        _services: &RuntimeServices,
         req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let path = req.get_path();
diff --git a/crates/trusted-server-core/src/integrations/google_tag_manager.rs b/crates/trusted-server-core/src/integrations/google_tag_manager.rs
index 64f27415..1ae8ec50 100644
--- a/crates/trusted-server-core/src/integrations/google_tag_manager.rs
+++ b/crates/trusted-server-core/src/integrations/google_tag_manager.rs
@@ -28,6 +28,7 @@ use crate::integrations::{
     IntegrationEndpoint, IntegrationProxy, IntegrationRegistration, IntegrationScriptContext,
     IntegrationScriptRewriter, ScriptRewriteAction,
 };
+use crate::platform::RuntimeServices;
 use crate::proxy::{proxy_request, ProxyRequestConfig};
 use crate::settings::{IntegrationConfig, Settings};
 
@@ -427,6 +428,7 @@ impl IntegrationProxy for GoogleTagManagerIntegration {
     async fn handle(
         &self,
         settings: &Settings,
+        _services: &RuntimeServices,
         mut req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let path = req.get_path().to_string();
diff --git a/crates/trusted-server-core/src/integrations/gpt.rs b/crates/trusted-server-core/src/integrations/gpt.rs
index 0affbe95..d849f1dd 100644
--- a/crates/trusted-server-core/src/integrations/gpt.rs
+++ b/crates/trusted-server-core/src/integrations/gpt.rs
@@ -49,6 +49,7 @@ use crate::integrations::{
     IntegrationEndpoint, IntegrationHeadInjector, IntegrationHtmlContext, IntegrationProxy,
     IntegrationRegistration,
 };
+use crate::platform::RuntimeServices;
 use crate::proxy::{proxy_request, ProxyRequestConfig};
 use crate::settings::{IntegrationConfig, Settings};
 
@@ -376,6 +377,7 @@ impl IntegrationProxy for GptIntegration {
     async fn handle(
         &self,
         settings: &Settings,
+        _services: &RuntimeServices,
         req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let path = req.get_path();
diff --git a/crates/trusted-server-core/src/integrations/lockr.rs b/crates/trusted-server-core/src/integrations/lockr.rs
index 5469c294..ad009130 100644
--- a/crates/trusted-server-core/src/integrations/lockr.rs
+++ b/crates/trusted-server-core/src/integrations/lockr.rs
@@ -24,6 +24,7 @@ use crate::integrations::{
     AttributeRewriteAction, IntegrationAttributeContext, IntegrationAttributeRewriter,
     IntegrationEndpoint, IntegrationProxy, IntegrationRegistration,
 };
+use crate::platform::RuntimeServices;
 use crate::settings::{IntegrationConfig, Settings};
 
 const LOCKR_INTEGRATION_ID: &str = "lockr";
@@ -304,6 +305,7 @@ impl IntegrationProxy for LockrIntegration {
     async fn handle(
         &self,
         settings: &Settings,
+        _services: &RuntimeServices,
         req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let path = req.get_path();
diff --git a/crates/trusted-server-core/src/integrations/permutive.rs b/crates/trusted-server-core/src/integrations/permutive.rs
index 179d7766..0946a54a 100644
--- a/crates/trusted-server-core/src/integrations/permutive.rs
+++ b/crates/trusted-server-core/src/integrations/permutive.rs
@@ -19,6 +19,7 @@ use crate::integrations::{
     AttributeRewriteAction, IntegrationAttributeContext, IntegrationAttributeRewriter,
     IntegrationEndpoint, IntegrationProxy, IntegrationRegistration,
 };
+use crate::platform::RuntimeServices;
 use crate::settings::{IntegrationConfig, Settings};
 
 const PERMUTIVE_INTEGRATION_ID: &str = "permutive";
@@ -560,6 +561,7 @@ impl IntegrationProxy for PermutiveIntegration {
     async fn handle(
         &self,
         settings: &Settings,
+        _services: &RuntimeServices,
         req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let path = req.get_path();
diff --git a/crates/trusted-server-core/src/integrations/prebid.rs b/crates/trusted-server-core/src/integrations/prebid.rs
index d38c43a4..75379bd8 100644
--- a/crates/trusted-server-core/src/integrations/prebid.rs
+++ b/crates/trusted-server-core/src/integrations/prebid.rs
@@ -30,6 +30,7 @@ use crate::openrtb::{
     OpenRtbRequest, PrebidExt, PrebidImpExt, Publisher, Regs, RegsExt, RequestExt, Site, ToExt,
     TrustedServerExt, User, UserExt,
 };
+use crate::platform::RuntimeServices;
 use crate::request_signing::{RequestSigner, SigningParams, SIGNING_VERSION};
 use crate::settings::{IntegrationConfig, Settings};
 
@@ -98,6 +99,14 @@ pub struct PrebidIntegrationConfig {
     /// manages both lists explicitly.
     #[serde(default, deserialize_with = "crate::settings::vec_from_seq_or_map")]
     pub client_side_bidders: Vec<String>,
+    /// Prebid.js `userSync` client configuration for User ID modules.
+    ///
+    /// The configured object is injected into `window.__tsjs_prebid.userSync`
+    /// and passed through to `pbjs.setConfig({ userSync })`. The corresponding
+    /// Prebid.js User ID modules must be included in the JS bundle at build
+    /// time with `TSJS_PREBID_USER_ID_MODULES`.
+    #[serde(default)]
+    pub user_sync: Option<Json>,
     /// Per-bidder, per-zone param overrides. The outer key is a bidder name, the
     /// inner key is a zone name (sent by the JS adapter from `mediaTypes.banner.name`
     /// — a non-standard Prebid.js field used as a temporary workaround),
@@ -315,6 +324,7 @@ impl IntegrationProxy for PrebidIntegration {
     async fn handle(
         &self,
         _settings: &Settings,
+        _services: &RuntimeServices,
         req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let path = req.get_path().to_string();
@@ -364,6 +374,8 @@ impl IntegrationHeadInjector for PrebidIntegration {
             timeout: u32,
             debug: bool,
             bidders: &'a [String],
+            #[serde(skip_serializing_if = "Option::is_none")]
+            user_sync: Option<&'a Json>,
             #[serde(skip_serializing_if = "<[String]>::is_empty")]
             client_side_bidders: &'a [String],
         }
@@ -373,6 +385,7 @@ impl IntegrationHeadInjector for PrebidIntegration {
             timeout: self.config.timeout_ms,
             debug: self.config.debug,
             bidders: &self.config.bidders,
+            user_sync: self.config.user_sync.as_ref(),
             client_side_bidders: &self.config.client_side_bidders,
         };
 
@@ -1348,6 +1361,7 @@ mod tests {
             debug_query_params: None,
             script_patterns: default_script_patterns(),
             client_side_bidders: Vec::new(),
+            user_sync: None,
             bid_param_zone_overrides: HashMap::new(),
             consent_forwarding: ConsentForwardingMode::Both,
         }
@@ -1843,6 +1857,36 @@ server_url = "https://prebid.example"
         );
     }
 
+    #[test]
+    fn head_injector_includes_user_sync_when_configured() {
+        let mut config = base_config();
+        config.user_sync = Some(json!({
+            "userIds": [
+                {
+                    "name": "sharedId",
+                    "storage": {"name": "_sharedID", "type": "cookie", "expires": 365}
+                }
+            ],
+            "auctionDelay": 100
+        }));
+        let integration = PrebidIntegration::new(config);
+        let document_state = IntegrationDocumentState::default();
+        let ctx = IntegrationHtmlContext {
+            request_host: "pub.example",
+            request_scheme: "https",
+            origin_host: "origin.example",
+            document_state: &document_state,
+        };
+
+        let inserts = integration.head_inserts(&ctx);
+        let script = &inserts[0];
+        assert!(
+            script.contains(r#""userSync":{"auctionDelay":100,"userIds":[{"name":"sharedId""#),
+            "should include userSync object: {}",
+            script
+        );
+    }
+
     #[test]
     fn to_openrtb_includes_debug_flags_when_enabled() {
         let mut config = base_config();
@@ -2711,7 +2755,7 @@ server_url = "https://prebid.example"
 
         let settings = make_settings();
         let request = Request::get("https://pub.example/auction");
-        let context = create_test_auction_context(&settings, &request);
+        let context = create_test_auction_context(&settings, &request, &EMPTY_CLIENT_INFO);
 
         let openrtb = provider.to_openrtb(&auction_request, &context, None);
 
@@ -2736,7 +2780,7 @@ server_url = "https://prebid.example"
 
         let settings = make_settings();
         let request = Request::get("https://pub.example/auction");
-        let context = create_test_auction_context(&settings, &request);
+        let context = create_test_auction_context(&settings, &request, &EMPTY_CLIENT_INFO);
 
         let openrtb = provider.to_openrtb(&auction_request, &context, None);
 
diff --git a/crates/trusted-server-core/src/integrations/registry.rs b/crates/trusted-server-core/src/integrations/registry.rs
index 968696a9..2c199318 100644
--- a/crates/trusted-server-core/src/integrations/registry.rs
+++ b/crates/trusted-server-core/src/integrations/registry.rs
@@ -13,6 +13,7 @@ use crate::ec::kv::KvIdentityGraph;
 use crate::ec::EcContext;
 use crate::error::TrustedServerError;
 use crate::http_util::is_navigation_request;
+use crate::platform::RuntimeServices;
 use crate::settings::Settings;
 
 /// Action returned by attribute rewriters to describe how the runtime should mutate the element.
@@ -259,6 +260,7 @@ pub trait IntegrationProxy: Send + Sync {
     async fn handle(
         &self,
         settings: &Settings,
+        services: &RuntimeServices,
         req: Request,
     ) -> Result<Response, Report<TrustedServerError>>;
 
@@ -649,11 +651,13 @@ impl IntegrationRegistry {
     /// Response-side cookie/header mutation is centralized in EC finalize.
     #[allow(clippy::too_many_arguments)]
     #[must_use]
+    #[allow(clippy::too_many_arguments)]
     pub async fn handle_proxy(
         &self,
         method: &Method,
         path: &str,
         settings: &Settings,
+        services: &RuntimeServices,
         kv: Option<&KvIdentityGraph>,
         ec_context: &mut EcContext,
         mut req: Request,
@@ -667,9 +671,13 @@ impl IntegrationRegistry {
                     log::warn!("EC generation failed for integration proxy: {err:?}");
                 }
             } else {
-                log::debug!(
-                    "EC generation skipped for integration proxy: non-document request (path={})",
+                log::info!(
+                    "EC generation decision: action=skipped_non_navigation, route=integration_proxy, \
+                     path={}, ec_present={}, ec_generated={}, consent_allowed={}",
                     path,
+                    ec_context.ec_value().is_some(),
+                    ec_context.ec_generated(),
+                    ec_context.ec_allowed(),
                 );
             }
 
@@ -678,7 +686,7 @@ impl IntegrationRegistry {
                 req.set_header(HEADER_X_TS_EC, ec_id);
             }
 
-            Some(proxy.handle(settings, req).await)
+            Some(proxy.handle(settings, services, req).await)
         } else {
             None
         }
@@ -961,6 +969,7 @@ mod tests {
         async fn handle(
             &self,
             _settings: &Settings,
+            _services: &RuntimeServices,
             _req: Request,
         ) -> Result<Response, Report<TrustedServerError>> {
             Ok(Response::new())
@@ -1243,6 +1252,7 @@ mod tests {
         async fn handle(
             &self,
             _settings: &Settings,
+            _services: &RuntimeServices,
             req: Request,
         ) -> Result<Response, Report<TrustedServerError>> {
             let mut response =
@@ -1277,10 +1287,12 @@ mod tests {
             EcContext::read_from_request(&settings, &req).expect("should read EC context");
 
         // Call handle_proxy (uses futures executor in test environment)
+        let services = crate::platform::test_support::noop_services();
         let result = futures::executor::block_on(registry.handle_proxy(
             &Method::GET,
             "/integrations/test/ec",
             &settings,
+            &services,
             None,
             &mut ec_context,
             req,
@@ -1319,10 +1331,12 @@ mod tests {
         let mut ec_context =
             EcContext::read_from_request(&settings, &req).expect("should read EC context");
 
+        let services = crate::platform::test_support::noop_services();
         let result = futures::executor::block_on(registry.handle_proxy(
             &Method::GET,
             "/integrations/test/ec",
             &settings,
+            &services,
             None,
             &mut ec_context,
             req,
@@ -1357,10 +1371,12 @@ mod tests {
         let mut ec_context =
             EcContext::read_from_request(&settings, &req).expect("should read EC context");
 
+        let services = crate::platform::test_support::noop_services();
         let result = futures::executor::block_on(registry.handle_proxy(
             &Method::POST,
             "/integrations/test/ec",
             &settings,
+            &services,
             None,
             &mut ec_context,
             req,
diff --git a/crates/trusted-server-core/src/integrations/testlight.rs b/crates/trusted-server-core/src/integrations/testlight.rs
index e63baea3..47665c5a 100644
--- a/crates/trusted-server-core/src/integrations/testlight.rs
+++ b/crates/trusted-server-core/src/integrations/testlight.rs
@@ -14,6 +14,7 @@ use crate::integrations::{
     AttributeRewriteAction, IntegrationAttributeContext, IntegrationAttributeRewriter,
     IntegrationEndpoint, IntegrationProxy, IntegrationRegistration,
 };
+use crate::platform::RuntimeServices;
 use crate::proxy::{proxy_request, ProxyRequestConfig};
 use crate::settings::{IntegrationConfig, Settings};
 use crate::tsjs;
@@ -140,6 +141,7 @@ impl IntegrationProxy for TestlightIntegration {
     async fn handle(
         &self,
         settings: &Settings,
+        _services: &RuntimeServices,
         mut req: Request,
     ) -> Result<Response, Report<TrustedServerError>> {
         let mut payload = serde_json::from_slice::<TestlightRequestBody>(&req.take_body_bytes())
diff --git a/crates/trusted-server-core/src/platform/test_support.rs b/crates/trusted-server-core/src/platform/test_support.rs
index 818819f5..a108699d 100644
--- a/crates/trusted-server-core/src/platform/test_support.rs
+++ b/crates/trusted-server-core/src/platform/test_support.rs
@@ -239,6 +239,16 @@ impl StubHttpClient {
     pub fn recorded_backend_names(&self) -> Vec<String> {
         self.calls.lock().expect("should lock calls").clone()
     }
+
+    /// Return the request headers captured per `send` call, in order.
+    ///
+    /// Each entry is the set of `(name, value)` pairs from one call.
+    pub fn recorded_request_headers(&self) -> Vec<Vec<(String, String)>> {
+        self.request_headers
+            .lock()
+            .expect("should lock request_headers")
+            .clone()
+    }
 }
 
 // ?Send matches PlatformHttpClient. See http.rs for the full rationale.
@@ -399,6 +409,27 @@ pub(crate) fn build_request_signing_services() -> RuntimeServices {
     )
 }
 
+/// Build a [`RuntimeServices`] with a [`StubBackend`] and the given HTTP client.
+///
+/// Useful for tests that need to verify `services.http_client()` call sites.
+pub(crate) fn build_services_with_http_client(
+    http_client: Arc<dyn PlatformHttpClient>,
+) -> RuntimeServices {
+    RuntimeServices::builder()
+        .config_store(Arc::new(NoopConfigStore))
+        .secret_store(Arc::new(NoopSecretStore))
+        .kv_store(Arc::new(edgezero_core::key_value_store::NoopKvStore))
+        .backend(Arc::new(StubBackend))
+        .http_client(http_client)
+        .geo(Arc::new(NoopGeo))
+        .client_info(ClientInfo {
+            client_ip: None,
+            tls_protocol: None,
+            tls_cipher: None,
+        })
+        .build()
+}
+
 pub(crate) fn build_services_with_config(
     config_store: impl PlatformConfigStore + 'static,
 ) -> RuntimeServices {
diff --git a/crates/trusted-server-core/src/publisher.rs b/crates/trusted-server-core/src/publisher.rs
index aaba0bcf..8c007f18 100644
--- a/crates/trusted-server-core/src/publisher.rs
+++ b/crates/trusted-server-core/src/publisher.rs
@@ -446,16 +446,21 @@ pub fn handle_publisher_request(
             log::warn!("EC generation failed: {err:?}");
         }
     } else {
-        log::debug!(
-            "EC generation skipped: non-document request (path={})",
+        log::info!(
+            "EC generation decision: action=skipped_non_navigation, path={}, \
+             ec_present={}, ec_generated={}, consent_allowed={}",
             req.get_path(),
+            ec_context.ec_value().is_some(),
+            ec_context.ec_generated(),
+            ec_context.ec_allowed(),
         );
     }
 
     let ec_allowed = ec_context.ec_allowed();
-    log::debug!(
-        "Proxy EC ID: {:?}, ec_allowed: {ec_allowed}",
-        ec_context.ec_value(),
+    log::info!(
+        "Publisher EC state: ec_present={}, ec_generated={}, ec_allowed={ec_allowed}",
+        ec_context.ec_value().is_some(),
+        ec_context.ec_generated(),
     );
 
     let backend_name = BackendConfig::from_url(
diff --git a/crates/trusted-server-core/src/settings.rs b/crates/trusted-server-core/src/settings.rs
index a75eb2c8..b65eebaf 100644
--- a/crates/trusted-server-core/src/settings.rs
+++ b/crates/trusted-server-core/src/settings.rs
@@ -349,6 +349,12 @@ pub struct Ec {
     #[serde(default)]
     pub ec_store: Option<String>,
 
+    /// Fastly KV store name for the partner registry.
+    ///
+    /// Required for the partner registration admin endpoint.
+    #[serde(default)]
+    pub partner_store: Option<String>,
+
     /// Maximum number of concurrent pull-sync requests.
     #[serde(default = "Ec::default_pull_sync_concurrency")]
     pub pull_sync_concurrency: usize,
@@ -759,8 +765,11 @@ impl Settings {
     /// endpoints are always protected by authentication.
     /// Update [`ADMIN_ENDPOINTS`](Self::ADMIN_ENDPOINTS) when adding new
     /// admin routes to `crates/trusted-server-adapter-fastly/src/main.rs`.
-    pub(crate) const ADMIN_ENDPOINTS: &[&str] =
-        &["/_ts/admin/keys/rotate", "/_ts/admin/keys/deactivate"];
+    pub(crate) const ADMIN_ENDPOINTS: &[&str] = &[
+        "/_ts/admin/keys/rotate",
+        "/_ts/admin/keys/deactivate",
+        "/_ts/admin/partners/register",
+    ];
 
     /// Returns admin endpoint paths that no configured handler covers.
     ///
@@ -2340,7 +2349,11 @@ mod tests {
             .expect("should check admin coverage");
         assert_eq!(
             uncovered,
-            vec!["/_ts/admin/keys/rotate", "/_ts/admin/keys/deactivate",],
+            vec![
+                "/_ts/admin/keys/rotate",
+                "/_ts/admin/keys/deactivate",
+                "/_ts/admin/partners/register",
+            ],
             "should report all admin endpoints as uncovered"
         );
     }
@@ -2374,7 +2387,7 @@ mod tests {
             .expect("should check admin coverage");
         assert_eq!(
             uncovered,
-            vec!["/_ts/admin/keys/deactivate"],
+            vec!["/_ts/admin/keys/deactivate", "/_ts/admin/partners/register",],
             "should detect endpoints not covered by the rotate-only handler"
         );
     }
diff --git a/trusted-server.toml b/trusted-server.toml
index 1b51f872..4c458682 100644
--- a/trusted-server.toml
+++ b/trusted-server.toml
@@ -84,6 +84,15 @@ debug = false
 # be statically imported in the JS bundle.
 client_side_bidders = ["rubicon"]
 
+# Optional Prebid.js User ID configuration. The matching User ID modules must
+# be included at JS build time with TSJS_PREBID_USER_ID_MODULES, for example:
+# TSJS_PREBID_USER_ID_MODULES=sharedId,pubProvided,uid2 node build-all.mjs
+# [integrations.prebid.user_sync]
+# auctionDelay = 100
+# [[integrations.prebid.user_sync.userIds]]
+# name = "sharedId"
+# storage = {name = "_sharedID", type = "cookie", expires = 365}
+
 # Zone-specific bid param overrides for Kargo s2s placement IDs.
 # The JS adapter reads the zone from mediaTypes.banner.name on each ad unit
 # and includes it in the request. The server maps zone → s2s placementId here.