writeback.mjs

// Write-back — put the finding into the graph, then prove it landed.
//
// This is the half that makes the loop close: without it the agent knows
// something the catalog does not, so the next person to look at the table
// learns nothing. But writing is where an agent can do real damage, so three
// constraints are enforced here rather than trusted:
//
//  1. FIXED VOCABULARY. The agent may write only the terms declared in
//     scripts/setup_agent_metadata.py. DataHub already refuses unknown tags
//     (measured: "Failed to validate label ... Urn does not exist") — this is
//     the second lock, so a typo fails in our code with a clear message
//     rather than as a GraphQL error.
//
//  2. SELF-CORRECTING. State tags are mutually exclusive. A table that was
//     STALE and is now FRESH must not keep both. Every write removes the
//     state tags that no longer apply before adding the one that does,
//     so re-running never accumulates contradictions.
//
//  3. READBACK OR IT DIDN'T HAPPEN. A successful write response is not
//     evidence the graph changed. I filed datahub#18753 during this hackathon
//     after finding three storage layers giving three different answers about
//     the same write, 2+ hours after it landed. So every write is read back
//     and compared; a mismatch is reported as loudly as a failure, because it
//     is one.
//
// UNMEASURED is written, not skipped. That is the whole point: the catalog
// gets to carry "we checked this and could not tell, here is why" as a
// first-class fact. Tools that only write findings leave "nobody knows"
// looking exactly like "nothing wrong".

const STATE_TAGS = Object.freeze({
  FRESH: 'freshness_fresh',
  STALE: 'freshness_stale',
  UNMEASURED: 'freshness_unmeasured',
});
const ROLE_TAGS = Object.freeze({
  root: 'freshness_root_cause',
  echo: 'freshness_downstream_echo',
});
/** Everything the agent is allowed to write. Mirrors setup_agent_metadata.py. */
export const VOCABULARY = Object.freeze([
  ...Object.values(STATE_TAGS),
  ...Object.values(ROLE_TAGS),
]);
const PROP = Object.freeze({
  state: 'urn:li:structuredProperty:unmeasured.state',
  detail: 'urn:li:structuredProperty:unmeasured.detail',
  checkedAt: 'urn:li:structuredProperty:unmeasured.checked_at',
});

const tagUrn = (name) => `urn:li:tag:${name}`;

/** Which tags this finding should end up carrying. */
export function tagsFor({ state, role }) {
  const stateTag = STATE_TAGS[state];
  if (!stateTag) throw new Error(`writeback: no tag defined for state "${state}" — refusing to invent one`);
  const roleTag = role ? ROLE_TAGS[role] : null;
  if (role && !roleTag) throw new Error(`writeback: unknown role "${role}" — refusing to invent one`);
  return roleTag ? [stateTag, roleTag] : [stateTag];
}

/** Read back the agent-owned tags currently on an entity. */
async function readAgentTags(mcp, urn) {
  const res = await mcp.call('get_entities', { urns: [urn] });
  if (res.isError) return { ok: false, reason: `readback failed: ${res.text.slice(0, 160)}` };
  let parsed;
  try { parsed = JSON.parse(res.text); } catch { return { ok: false, reason: 'readback response is not JSON' }; }
  const entity = (Array.isArray(parsed) ? parsed : [parsed]).find((e) => e?.urn === urn);
  if (!entity) return { ok: false, reason: 'readback found no such entity' };
  if (typeof entity.error === 'string' && entity.error) {
    return { ok: false, reason: `readback error: ${entity.error}` };
  }
  const names = (entity.tags?.tags ?? [])
    .map((t) => t.tag?.properties?.name ?? t.tag?.urn?.split(':').pop())
    .filter(Boolean);
  return { ok: true, agentTags: names.filter((n) => VOCABULARY.includes(n)), allTags: names };
}

/**
 * Record one finding in the graph and verify it landed.
 *
 * @returns {{
 *   ok: boolean,          // wrote AND verified
 *   wrote: string[],      // tags intended
 *   verified: boolean,
 *   mismatch: string|null,// set when the write reported success but readback disagreed
 *   reason: string|null,  // set when the write itself failed
 *   preservedTags: string[], // non-agent tags still present after the write
 * }}
 */
export async function recordFinding(mcp, { urn, state, role = null, detail, checkedAt }) {
  const want = tagsFor({ state, role });

  const before = await readAgentTags(mcp, urn);
  if (!before.ok) {
    return { ok: false, wrote: [], verified: false, mismatch: null, reason: `cannot read entity before writing: ${before.reason}`, preservedTags: [] };
  }

  // Self-correcting: drop agent tags that no longer apply. Left alone, a table
  // that recovered would carry both freshness_stale and freshness_fresh.
  const stale = before.agentTags.filter((t) => !want.includes(t));
  if (stale.length > 0) {
    const rm = await mcp.call('remove_tags', { entity_urns: [urn], tag_urns: stale.map(tagUrn) });
    if (rm.isError) {
      return { ok: false, wrote: [], verified: false, mismatch: null, reason: `could not remove superseded tags ${stale.join(', ')}: ${rm.text.slice(0, 160)}`, preservedTags: [] };
    }
  }

  const add = await mcp.call('add_tags', { entity_urns: [urn], tag_urns: want.map(tagUrn) });
  if (add.isError) {
    return { ok: false, wrote: [], verified: false, mismatch: null, reason: `add_tags failed: ${add.text.slice(0, 200)}`, preservedTags: [] };
  }

  // Structured properties carry the evidence in typed form. A failure here is
  // reported but does not void the tag write — partial success is stated as
  // partial, never rounded up.
  let propReason = null;
  const props = await mcp.call('add_structured_properties', {
    entity_urns: [urn],
    property_values: {
      [PROP.state]: [state],
      [PROP.detail]: [detail],
      [PROP.checkedAt]: [checkedAt],
    },
  });
  if (props.isError) propReason = `structured properties not written: ${props.text.slice(0, 160)}`;

  // ---- readback: the write response is a claim, this is the evidence ----
  const after = await readAgentTags(mcp, urn);
  if (!after.ok) {
    return { ok: false, wrote: want, verified: false, mismatch: `write reported success but readback failed: ${after.reason}`, reason: propReason, preservedTags: [] };
  }

  const missing = want.filter((t) => !after.agentTags.includes(t));
  const extra = after.agentTags.filter((t) => !want.includes(t));
  const preserved = after.allTags.filter((t) => !VOCABULARY.includes(t));

  if (missing.length > 0 || extra.length > 0) {
    const bits = [];
    if (missing.length) bits.push(`missing ${missing.join(', ')}`);
    if (extra.length) bits.push(`unexpected ${extra.join(', ')}`);
    return {
      ok: false, wrote: want, verified: false,
      mismatch: `write reported success, readback disagrees: ${bits.join('; ')}`,
      reason: propReason, preservedTags: preserved,
    };
  }

  return { ok: propReason === null, wrote: want, verified: true, mismatch: null, reason: propReason, preservedTags: preserved };
}