feat(semaphore-sync): two-tier tag sync + scheduler mislog fix

srv/jobs/scheduler.js

@@ -170,7 +170,26 @@ async function runWithLock(jobName, durationMs, fn, opts = {}) {
   try {
     result = await fn(logId);
     const summary = formatJobSummary(jobName, result);
-    await logPipelineEnd(logId, 'SUCCESS', summary);
+    // A job may fail-shut by RETURNING { ok:false, error } rather than throwing
+    // (e.g. semaphore-tag-sync on a bad fetch). Treat that as a failed run so the
+    // PipelineLog STATUS and JobLastRun.lastSuccessAt/lastErrorAt reflect reality
+    // — otherwise a no-op/errored run is mislogged SUCCESS and looks healthy.
+    if (result && typeof result === 'object' && result.ok === false) {
+      outcome = 'error';
+      errorMessage = result.error ? String(result.error) : `${jobName} returned ok:false`;
+      LOG.error(`Job ${jobName} returned failure:`, errorMessage);
+      await logPipelineEnd(logId, 'FAILED', summary, errorMessage);
+      void alerting.raise({
+        eventType: 'ScheduledJobFailed',
+        severity: 'ERROR',
+        category: 'ALERT',
+        subject: `Scheduled job failed: ${jobName}`,
+        body: errorMessage,
+        resource: { resourceName: jobName, resourceType: 'job' }
+      }); // fail-open, non-blocking
+    } else {
+      await logPipelineEnd(logId, 'SUCCESS', summary);
+    }
   } catch (err) {
     outcome = 'error';
     errorMessage = err.message ?? String(err);

srv/jobs/semaphore-tag-sync-job.js

@@ -41,6 +41,7 @@ const CONFIG_KEYS = [
   'semaphore.sync.filter',
   'semaphore.sync.actualTagClasses',
   'semaphore.sync.interestItemClasses',
+  'semaphore.sync.intakeClasses',
   'semaphore.sync.dryRun',
 ];
 
@@ -51,6 +52,34 @@ function splitList(v) {
     .filter(Boolean);
 }
 
+// Shorten a class URI to its readable leaf ("…schema#Topic" → "Topic") for the
+// histogram; leave short names untouched.
+function shortClass(c) {
+  const s = String(c ?? '');
+  if (s.includes('#')) return s.slice(s.lastIndexOf('#') + 1);
+  if (s.includes('/')) return s.slice(s.lastIndexOf('/') + 1);
+  return s;
+}
+
+// Distinct SES classes across all mapped rows with a term count each, rendered
+// as a compact single-line string. formatJobSummary only renders scalar summary
+// fields (number/string/boolean) — an object/array would be silently dropped —
+// so the histogram MUST be a pre-formatted string to survive into the SUMMARY.
+// This is the whole point of the first dry run: it surfaces the real class
+// distribution so semaphore.sync.filter / intakeClasses can be chosen from data.
+function classHistogram(rows) {
+  const counts = new Map();
+  for (const r of rows) {
+    for (const c of r.classes ?? []) {
+      const k = shortClass(c);
+      counts.set(k, (counts.get(k) ?? 0) + 1);
+    }
+  }
+  const sorted = [...counts.entries()].sort((a, b) => b[1] - a[1]);
+  if (sorted.length === 0) return '(no classes on any term)';
+  return sorted.map(([k, n]) => `${k}=${n}`).join(', ');
+}
+
 // Read the semaphore.sync.* string config from ImsConfig in one SELECT.
 async function readConfig(db) {
   const { ImsConfig } = cds.entities(NS);
@@ -69,6 +98,10 @@ async function readConfig(db) {
     filter: map.get('semaphore.sync.filter') || undefined,
     actualTagClasses: splitList(map.get('semaphore.sync.actualTagClasses')),
     interestItemClasses: splitList(map.get('semaphore.sync.interestItemClasses')),
+    // Tier-2 intake allowlist: unmatched terms are INSERTed only if their SES
+    // class is listed here. Empty ⇒ intake OFF (adopt-only) — the safe default
+    // until the FILTER/classes are chosen from the dry-run histogram.
+    intakeClasses: splitList(map.get('semaphore.sync.intakeClasses')),
     // dryRun defaults to TRUE — first live runs report the plan without writing.
     dryRun: (map.get('semaphore.sync.dryRun') ?? 'true').toLowerCase() !== 'false',
   };
@@ -130,9 +163,9 @@ export async function runSemaphoreTagSync(_logId, opts = {}) {
     interestItemClasses: cfg.interestItemClasses,
   });
 
-  let applied = { inserted: 0, updated: 0, unchanged: 0, total: rows.length };
+  let applied = { inserted: 0, updated: 0, unchanged: 0, skippedIntake: 0, total: rows.length };
   try {
-    applied = await applyTerms(rows, { db, dryRun: cfg.dryRun });
+    applied = await applyTerms(rows, { db, dryRun: cfg.dryRun, intakeClasses: cfg.intakeClasses });
   } catch (e) {
     LOG.error(`semaphore-sync upsert failed: ${e.message}`);
     return { ok: false, error: e.message, mapped: rows.length, skipped: skipped.length };
@@ -146,6 +179,9 @@ export async function runSemaphoreTagSync(_logId, opts = {}) {
     fetched: Array.isArray(data.terms) ? data.terms.length : 0,
     mapped: rows.length,
     skipped: skipped.length,
+    intakeClasses: cfg.intakeClasses.join(', ') || '(none — adopt-only)',
+    // Pre-formatted string so formatJobSummary renders it into the SUMMARY.
+    classHistogram: classHistogram(rows),
     ...applied,
   };
   LOG.info(`semaphore-sync summary: ${JSON.stringify(summary)}`);

srv/lib/semaphore-sync/applier.js

@@ -19,42 +19,72 @@
 
 import cds from '@sap/cds';
 
-// Fields we consider when deciding whether an existing row needs an UPDATE.
-const TRACKED = ['label', 'name', 'titlePath', 'isActualTag', 'isInterestItem', 'semaphoreId'];
+// SES SAPCore is the ENTIRE SAP product/topic universe (~21k terms); the Tags
+// table is a deliberately curated subset. So the sync is two-tier (#2184):
+//
+//   Tier 1 — ADOPT (existing tags): match by semaphoreId, else by name. Backfill
+//     ONLY the taxonomy-owned fields (semaphoreId, titlePath, label). The
+//     isActualTag / isInterestItem flags are EDITORIAL — set per-term by hand,
+//     not derivable from any SES class — so the sync must never overwrite them
+//     (verified: 99 tags under the "Software Product" root, only 18/18 flagged).
+//
+//   Tier 2 — INTAKE (new terms): a term with no matching row is INSERTed ONLY
+//     when it passes the intake allowlist (opts.intakeClasses). New rows land
+//     INERT (isActualTag=false, isInterestItem=false) awaiting editor curation —
+//     never active, never flooding the table with all 21k. With no allowlist,
+//     intake is OFF and unmatched terms are counted `skippedIntake`, not inserted.
+//
+// Idempotent: a second run with the same payload reports every row `unchanged`.
+// dryRun:true computes the same plan without writing.
+
+// Taxonomy-owned fields the sync may write onto an EXISTING row. The editorial
+// flags are deliberately excluded so an adopt/update never clobbers curation.
+const ADOPT_FIELDS = ['label', 'name', 'titlePath', 'semaphoreId'];
 
-function differs(existing, row) {
-  return TRACKED.some((f) => (existing[f] ?? null) !== (row[f] ?? null));
+function adoptDiffers(existing, row) {
+  return ADOPT_FIELDS.some((f) => (existing[f] ?? null) !== (row[f] ?? null));
+}
+
+// Case-insensitive membership, tolerant of full class URIs vs short names —
+// mirrors mapper.classMatches so intake uses the same class-matching semantics.
+function classAllowed(termClasses, allow) {
+  if (!Array.isArray(allow) || allow.length === 0) return false;
+  if (!Array.isArray(termClasses) || termClasses.length === 0) return false;
+  const allowLc = allow.map((w) => String(w).toLowerCase());
+  return termClasses.some((c) => {
+    const cl = String(c ?? '').toLowerCase();
+    const short = cl.includes('#') ? cl.slice(cl.lastIndexOf('#') + 1)
+      : cl.includes('/') ? cl.slice(cl.lastIndexOf('/') + 1)
+      : cl;
+    return allowLc.some((w) => w === cl || w === short);
+  });
 }
 
 /**
- * Upsert mapper rows into Tags.
+ * Two-tier upsert of mapper rows into Tags.
  *
- * @param {Array}  rows  Output of mapAllTerms().rows
+ * @param {Array}  rows  Output of mapAllTerms().rows (each row also carries
+ *                       `classes` for the intake gate)
  * @param {object} [opts]
- * @param {boolean} [opts.dryRun=false]  compute the plan without writing
- * @param {object}  [opts.db]            cds db (defaults to cds.db / connect)
- * @returns {Promise<{inserted:number, updated:number, unchanged:number, total:number}>}
+ * @param {boolean}  [opts.dryRun=false]       compute the plan without writing
+ * @param {string[]} [opts.intakeClasses=[]]   class allowlist for Tier-2 INSERT;
+ *                                              empty ⇒ intake OFF (adopt-only)
+ * @param {object}   [opts.db]                 cds db (defaults to cds.db / connect)
+ * @returns {Promise<{inserted:number, updated:number, unchanged:number,
+ *                    skippedIntake:number, total:number}>}
  */
 export async function applyTerms(rows, opts = {}) {
-  const { dryRun = false } = opts;
+  const { dryRun = false, intakeClasses = [] } = opts;
   const db = opts.db ?? cds.db ?? (await cds.connect.to('db'));
   const { Tags } = cds.entities('com.sap.developers.ims');
 
   let inserted = 0;
   let updated = 0;
   let unchanged = 0;
+  let skippedIntake = 0;
   const total = Array.isArray(rows) ? rows.length : 0;
 
   for (const row of rows ?? []) {
-    const fields = {
-      semaphoreId: row.semaphoreId,
-      label: row.label,
-      name: row.name,
-      titlePath: row.titlePath,
-      isActualTag: !!row.isActualTag,
-      isInterestItem: !!row.isInterestItem,
-    };
-
     // 1. Existing by semaphoreId.
     let existing = await db.run(SELECT.one.from(Tags).where({ semaphoreId: row.semaphoreId }));
     // 2. Adopt a legacy/CSV row that matches by name but has no semaphoreId yet.
@@ -64,22 +94,42 @@ export async function applyTerms(rows, opts = {}) {
       );
     }
 
+    // Tier 1 — ADOPT: update ONLY taxonomy-owned fields; never the editorial flags.
     if (existing) {
-      if (differs(existing, fields)) {
-        if (!dryRun) await db.run(UPDATE(Tags, existing.ID).set(fields));
+      const adopt = {
+        semaphoreId: row.semaphoreId,
+        label: row.label,
+        name: row.name,
+        titlePath: row.titlePath,
+      };
+      if (adoptDiffers(existing, adopt)) {
+        if (!dryRun) await db.run(UPDATE(Tags, existing.ID).set(adopt));
         updated++;
       } else {
         unchanged++;
       }
       continue;
     }
 
-    // 3. Insert. Assign the UUID key explicitly for HANA parity.
+    // Tier 2 — INTAKE: insert a genuinely new term ONLY if class-allowlisted.
+    // New rows land inert; an editor promotes them by setting the flags later.
+    if (!classAllowed(row.classes, intakeClasses)) {
+      skippedIntake++;
+      continue;
+    }
     if (!dryRun) {
-      await db.run(INSERT.into(Tags).entries({ ID: cds.utils.uuid(), ...fields }));
+      await db.run(INSERT.into(Tags).entries({
+        ID: cds.utils.uuid(),
+        semaphoreId: row.semaphoreId,
+        label: row.label,
+        name: row.name,
+        titlePath: row.titlePath,
+        isActualTag: false,
+        isInterestItem: false,
+      }));
     }
     inserted++;
   }
 
-  return { inserted, updated, unchanged, total };
+  return { inserted, updated, unchanged, skippedIntake, total };
 }

srv/lib/semaphore-sync/mapper.js

@@ -161,6 +161,9 @@ export function mapAllTerms(data, opts = {}) {
       titlePath: deriveTitlePath(term),
       isActualTag,
       isInterestItem,
+      // Raw SES classes, carried through for the applier's Tier-2 intake gate
+      // and the dry-run class histogram (job). Not persisted to Tags.
+      classes: Array.isArray(term.classes) ? term.classes.map(String) : [],
     };
 
     // De-dupe on semaphoreId (last write wins) — SES should be unique but be safe.