fix(plugins): report the plugin that produced the event, not the one that sent it

harness is set when an event is recorded; source was set when its batch was
sent. Both came from module globals that stay at "generic" and "MEM0_PLUGIN"
until telemetry.init() runs, and two processes in the pipeline never run it:

- `python3 telemetry.py`, the detached sender spawn_flush() starts at session
  start, after every skill command, and when the MCP server exits. Everything it
  delivered was labelled source=MEM0_PLUGIN. Only batches flush_worker.py
  happened to drain got the real host.
- mcp_server.py, which records every manual search as harness=generic.

All six Python plugins ship the same files, so source could not tell any of them
apart and MCP searches from every plugin landed in one generic bucket. The
portable plugin is worse: it has no flush_worker at all, so its only sender is
the uninitialised one and 100% of its events were mislabelled.

Two changes. record() stamps source beside harness, so the sending process stops
mattering — flush() already spreads per-event properties last, so a per-event
source wins over any sender default. And the build generates core/_harness_id.py
per host, seeding both modules at import, so identity no longer depends on an
entrypoint remembering to call init(). The build already computed HARNESS_ID and
spent it only on skill templating, and bundle_drift already diffs core/
byte-for-byte, so --check catches drift for free.

Deliberately not adding MEM0_PLUGIN_HARNESS to the six manifests: they sit
outside the --sync and --check boundary, which is the property that caused this.

Also unifies two defaults that disagreed. configure_harness derived
`<host>_plugin` while telemetry.init derived `MEM0_<HOST>_PLUGIN`, so a third
value existed. It was unreachable only because hook_runner never calls flush();
moving source into record() would have made it live.

Events now carry a uuid so a resend can be collapsed.

The suite stayed green through all of this because the only tests live under one
host, behind a conftest that calls init() at import. New tests run in real
subprocesses with no init, and cover the portable plugin, which would pass a
native-only test vacuously.

Claude-Session: https://claude.ai/code/session_01C7tEmH86HAr7GoAAKCEHZb

integrations/agent-plugin-core/python/telemetry.py

@@ -34,8 +34,23 @@ from typing import Any
 
 import memory_core
 
-_harness: str = "generic"
-_source_tag: str = "MEM0_PLUGIN"
+# Seeded from the per-host module the build generates into core/. Two processes
+# in this pipeline never call init() — mcp_server.py, and the detached
+# `python3 telemetry.py` sender that spawn_flush() starts — so a module default
+# was what every one of their events got labelled with.
+try:  # pragma: no cover - absent only in the un-built shared source tree
+    from _harness_id import HARNESS_ID as _DEFAULT_HARNESS
+    from _harness_id import PLATFORM_APPLICATION as _PLATFORM_APPLICATION
+    from _harness_id import PLATFORM_SOURCE as _PLATFORM_SOURCE
+    from _harness_id import SOURCE_TAG as _DEFAULT_SOURCE_TAG
+except ImportError:
+    _DEFAULT_HARNESS = "generic"
+    _DEFAULT_SOURCE_TAG = "MEM0_PLUGIN"
+    _PLATFORM_SOURCE = "MEM0_PLUGIN"
+    _PLATFORM_APPLICATION = ""
+
+_harness: str = _DEFAULT_HARNESS
+_source_tag: str = _DEFAULT_SOURCE_TAG
 _PRIVATE_KEYS = {
     "apikey",
     "authorization",
@@ -61,10 +76,19 @@ _PRIVATE_KEYS = {
 }
 
 
-def init(harness: str = "generic", source_tag: str = "") -> None:
+def init(harness: str = "", source_tag: str = "") -> None:
+    """Override the generated identity. Optional — core/_harness_id.py is the default.
+
+    The fallback shape matches memory_core.configure_harness's (``<HOST>_PLUGIN``).
+    It used to be ``MEM0_<HOST>_PLUGIN`` here and ``<host>_plugin`` there, which
+    meant one plugin could emit three different source values depending on which
+    process happened to send the batch.
+    """
     global _harness, _source_tag
-    _harness = harness
-    _source_tag = source_tag or f"MEM0_{harness.upper().replace('-', '_')}_PLUGIN"
+    _harness = harness or _DEFAULT_HARNESS
+    _source_tag = source_tag or (
+        f"{_harness.upper().replace('-', '_')}_PLUGIN" if harness else _DEFAULT_SOURCE_TAG
+    )
 
 POSTHOG_API_KEY = "phc_hgJkUVJFYtmaJqrvf6CYN67TIQ8yhXAkWzUn9AMU4yX"
 POSTHOG_CAPTURE_URL = "https://us.i.posthog.com/i/v0/e/"
@@ -200,8 +224,13 @@ def record(
         except OSError:
             pass
         properties = _safe_value(properties)
+        # Stamped in the RECORDING process, beside harness. `source` used to be
+        # read in the sending process from a module global, so whichever process
+        # drained the spool named every event in it. flush() spreads per-event
+        # properties last, so this now wins over any sender's default.
         properties.update(
             harness=_harness,
+            source=_source_tag,
             plugin_version=memory_core.PLUGIN_VERSION,
             os=sys.platform,
             python_version=platform.python_version(),
@@ -213,6 +242,7 @@ def record(
         line = json.dumps(
             {
                 "event": f"{EVENT_PREFIX}.{event}",
+                "uuid": str(uuid.uuid4()),
                 "timestamp": memory_core.utc_now(),
                 "properties": {
                     key: value for key, value in properties.items() if value is not None
@@ -398,6 +428,8 @@ def flush() -> int:
                 "distinct_id": distinct_id,
                 "timestamp": event.get("timestamp"),
                 "properties": {
+                    # Fallback only: events recorded by a build before source
+                    # moved into record() have none of their own.
                     "source": _source_tag,
                     "language": "python",
                     "$process_person_profile": False,