diff --git a/chart/interloper/templates/NOTES.txt b/chart/interloper/templates/NOTES.txt
index e498dfbc..36c1e9f2 100644
--- a/chart/interloper/templates/NOTES.txt
+++ b/chart/interloper/templates/NOTES.txt
@@ -32,5 +32,12 @@ HTTPRoute hostnames: {{ join ", " .Values.httpRoute.hostnames }}
    Generate one and re-apply:   openssl rand -base64 32
 {{- end }}
 
+{{- if and (eq (default "" (dig "runner" "type" "" .Values.config)) "k8s") (not .Values.secrets.eventIngestToken) }}
+
+⚠  No secrets.eventIngestToken set — per-asset workers will persist events via
+   log-scraping (best-effort) instead of the durable ingest endpoint.
+   Generate one and re-apply:   openssl rand -base64 32
+{{- end }}
+
 Tail the scheduler logs:
   kubectl -n {{ .Release.Namespace }} logs -f deploy/{{ include "interloper.fullname" . }}-scheduler
diff --git a/chart/interloper/templates/_helpers.tpl b/chart/interloper/templates/_helpers.tpl
index 961adbed..320fa2ce 100644
--- a/chart/interloper/templates/_helpers.tpl
+++ b/chart/interloper/templates/_helpers.tpl
@@ -147,6 +147,18 @@ Postgres host: either the bundled subchart service or external config.
 {{- end -}}
 {{- end -}}
 
+{{/*
+Base API URL that worker pods POST events to. Defaults to the in-cluster API
+service; override via events.ingestUrl.
+*/}}
+{{- define "interloper.eventIngestUrl" -}}
+{{- if .Values.events.ingestUrl -}}
+{{ .Values.events.ingestUrl }}
+{{- else -}}
+http://{{ include "interloper.fullname" . }}-api.{{ .Release.Namespace }}.svc:{{ .Values.api.service.port }}
+{{- end -}}
+{{- end -}}
+
 {{/*
 Environment variables shared by all interloper pods (scheduler, api, frontend).
 Contains Postgres connection info + the encryption key secret.
@@ -177,4 +189,12 @@ Contains Postgres connection info + the encryption key secret.
       name: {{ include "interloper.secretName" . }}
       key: INTERLOPER_AUTH_GOOGLE_CLIENT_SECRET
       optional: true
+- name: INTERLOPER_EVENTS_INGEST_URL
+  value: {{ include "interloper.eventIngestUrl" . | quote }}
+- name: INTERLOPER_EVENTS_INGEST_TOKEN
+  valueFrom:
+    secretKeyRef:
+      name: {{ include "interloper.secretName" . }}
+      key: INTERLOPER_EVENTS_INGEST_TOKEN
+      optional: true
 {{- end -}}
diff --git a/chart/interloper/templates/secret.yaml b/chart/interloper/templates/secret.yaml
index 0eb02d26..ad01bc4a 100644
--- a/chart/interloper/templates/secret.yaml
+++ b/chart/interloper/templates/secret.yaml
@@ -17,4 +17,7 @@ stringData:
   {{- if .Values.secrets.googleClientSecret }}
   INTERLOPER_AUTH_GOOGLE_CLIENT_SECRET: {{ .Values.secrets.googleClientSecret | quote }}
   {{- end }}
+  {{- if .Values.secrets.eventIngestToken }}
+  INTERLOPER_EVENTS_INGEST_TOKEN: {{ .Values.secrets.eventIngestToken | quote }}
+  {{- end }}
 {{- end }}
diff --git a/chart/interloper/values.yaml b/chart/interloper/values.yaml
index f3acb998..fc2d7573 100644
--- a/chart/interloper/values.yaml
+++ b/chart/interloper/values.yaml
@@ -187,6 +187,20 @@ secrets:
   encryptionKey: ""  # recommended to generate: openssl rand -base64 32
   smtpPassword: ""
   googleClientSecret: ""  # OAuth client secret (paired with config.auth.google_client_id)
+  # Shared service token for the internal event-ingest endpoint. When set,
+  # per-asset worker pods POST their events to the API instead of relying on
+  # log-scraping. When empty, the ingest endpoint is disabled (workers fall
+  # back to the stderr path). Generate: openssl rand -base64 32
+  eventIngestToken: ""
+
+# =============================================================================
+# Events
+# =============================================================================
+
+events:
+  # Base API URL that worker pods POST events to. Defaults to the in-cluster
+  # API service; override only if workers must reach the API by another route.
+  ingestUrl: ""
 
 # =============================================================================
 # Postgres
diff --git a/packages/interloper-api/src/interloper_api/app.py b/packages/interloper-api/src/interloper_api/app.py
index ec349274..c779729a 100644
--- a/packages/interloper-api/src/interloper_api/app.py
+++ b/packages/interloper-api/src/interloper_api/app.py
@@ -9,7 +9,13 @@
 from interloper.catalog.base import Catalog
 from interloper_db import Store
 
-from interloper_api.dependencies import set_auth_config, set_catalog, set_smtp_config, set_store
+from interloper_api.dependencies import (
+    set_auth_config,
+    set_catalog,
+    set_ingest_token,
+    set_smtp_config,
+    set_store,
+)
 from interloper_api.routes import (
     admin,
     assets,
@@ -17,6 +23,7 @@
     backfills,
     destinations,
     external,
+    internal,
     jobs,
     oauth,
     organisations,
@@ -33,6 +40,7 @@ def create_app(
     catalog: Catalog | None = None,
     auth_config: Any | None = None,
     smtp_config: Any | None = None,
+    event_ingest_token: str | None = None,
     cors_origins: list[str] | None = None,
     **kwargs: Any,
 ) -> FastAPI:
@@ -69,6 +77,7 @@ def create_app(
         set_auth_config(auth_config)
     if smtp_config:
         set_smtp_config(smtp_config)
+    set_ingest_token(event_ingest_token)
 
     api = APIRouter(prefix="/api")
     api.include_router(auth.router, tags=["auth"])
@@ -80,6 +89,7 @@ def create_app(
     api.include_router(destinations.router, prefix="/destinations", tags=["destinations"])
     api.include_router(jobs.router, prefix="/jobs", tags=["jobs"])
     api.include_router(runs.router, prefix="/runs", tags=["runs"])
+    api.include_router(internal.router, prefix="/internal", tags=["internal"])
     api.include_router(backfills.router, prefix="/backfills", tags=["backfills"])
     api.include_router(assets.router, prefix="/assets", tags=["assets"])
     api.include_router(oauth.router, tags=["oauth"])
diff --git a/packages/interloper-api/src/interloper_api/dependencies.py b/packages/interloper-api/src/interloper_api/dependencies.py
index 0073294d..7c10188c 100644
--- a/packages/interloper-api/src/interloper_api/dependencies.py
+++ b/packages/interloper-api/src/interloper_api/dependencies.py
@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
+import hmac
 from typing import Any
 from uuid import UUID
 
-from fastapi import Cookie, Depends, HTTPException
+from fastapi import Cookie, Depends, Header, HTTPException
 from interloper.catalog.base import Catalog
 from interloper_db import Organisation, Profile, Store
 from interloper_db.models import Session as SessionModel
@@ -14,6 +15,7 @@
 _catalog: Catalog | None = None
 _auth_config: Any | None = None
 _smtp_config: Any | None = None
+_ingest_token: str | None = None
 
 # Role hierarchy: admin > editor > viewer
 _ROLE_RANK = {"viewer": 0, "editor": 1, "admin": 2}
@@ -110,6 +112,21 @@ def get_smtp_config() -> Any:
     return _smtp_config
 
 
+def set_ingest_token(token: str | None) -> None:
+    """Set the shared service token that authenticates internal event ingest.
+
+    Args:
+        token: The secret, or ``None``/empty to disable the ingest endpoint.
+    """
+    global _ingest_token  # noqa: PLW0603
+    _ingest_token = token or None
+
+
+def get_ingest_token() -> str | None:
+    """Return the configured event-ingest token, or ``None`` if disabled."""
+    return _ingest_token
+
+
 # -- Auth dependencies -------------------------------------------------------
 
 
@@ -310,3 +327,36 @@ def require_super_admin(
     if not user.is_super_admin:
         raise HTTPException(status_code=403, detail="Requires super-admin privileges")
     return user
+
+
+# -- Service-token auth (internal machine-to-machine) ------------------------
+
+
+def _bearer_token(authorization: str | None) -> str | None:
+    """Extract the token from an ``Authorization: Bearer <token>`` header."""
+    if not authorization:
+        return None
+    scheme, _, token = authorization.partition(" ")
+    if scheme.lower() != "bearer" or not token:
+        return None
+    return token
+
+
+def require_ingest_token(authorization: str | None = Header(default=None)) -> None:
+    """Authenticate an internal event-ingest request via a bearer service token.
+
+    This is machine-to-machine auth for trusted in-cluster callers (per-asset
+    worker processes), deliberately separate from the cookie-session user auth.
+    The token is a shared secret set via :func:`set_ingest_token`; when unset,
+    the ingest surface is disabled entirely.
+
+    Raises:
+        HTTPException: 503 if ingest is not configured, 401 if the token is
+            missing or wrong.
+    """
+    expected = get_ingest_token()
+    if not expected:
+        raise HTTPException(status_code=503, detail="Event ingest is not configured")
+    provided = _bearer_token(authorization)
+    if not provided or not hmac.compare_digest(provided, expected):
+        raise HTTPException(status_code=401, detail="Invalid or missing ingest token")
diff --git a/packages/interloper-api/src/interloper_api/routes/internal.py b/packages/interloper-api/src/interloper_api/routes/internal.py
new file mode 100644
index 00000000..905df44d
--- /dev/null
+++ b/packages/interloper-api/src/interloper_api/routes/internal.py
@@ -0,0 +1,77 @@
+"""Internal (machine-to-machine) API: reliable event ingest.
+
+Per-asset worker processes run in slim, DB-less pods, so instead of routing
+their events through stdout/log-scraping (lossy), they POST them here.  The
+endpoint is authenticated with a shared service token (not the user session)
+and writes each event with :meth:`Store.save_event`, which is idempotent — so
+the worker can retry a batch freely without creating duplicates.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from uuid import UUID
+
+import interloper as il
+from fastapi import APIRouter, Depends, HTTPException
+from interloper.errors import NotFoundError
+from interloper_db import Store
+from pydantic import BaseModel
+
+from interloper_api.dependencies import get_store, require_ingest_token
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+class EventIngestRequest(BaseModel):
+    """A batch of serialized events for a single run.
+
+    Each entry is the flat dict produced by ``Event.to_dict()`` (``event_id``,
+    ``type``, ``timestamp`` plus inlined metadata).
+    """
+
+    events: list[dict[str, Any]]
+
+
+class EventIngestResponse(BaseModel):
+    """Result of an ingest batch."""
+
+    accepted: int
+    rejected: int
+
+
+@router.post("/runs/{run_id}/events")
+def ingest_run_events(
+    run_id: UUID,
+    body: EventIngestRequest,
+    _: None = Depends(require_ingest_token),
+    store: Store = Depends(get_store),
+) -> EventIngestResponse:
+    """Persist a batch of events for *run_id*.
+
+    The ``org_id`` is resolved server-side from the run, so callers only need
+    the run id.  Malformed events are skipped (and counted in ``rejected``)
+    rather than failing the batch; a persistence error is allowed to surface as
+    a 5xx so the caller retries — idempotent writes make that safe.
+    """
+    try:
+        run = store.get_run(run_id)
+    except NotFoundError:
+        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
+
+    accepted = 0
+    rejected = 0
+    for raw in body.events:
+        try:
+            event = il.Event.from_dict(raw)
+        except Exception:  # noqa: BLE001 - malformed event: skip, don't retry
+            rejected += 1
+            logger.warning("Rejected malformed event for run %s", run_id)
+            continue
+        store.save_event(event, org_id=run.org_id, run_id=run_id)
+        accepted += 1
+
+    return EventIngestResponse(accepted=accepted, rejected=rejected)
diff --git a/packages/interloper-api/tests/test_ingest.py b/packages/interloper-api/tests/test_ingest.py
new file mode 100644
index 00000000..6ba40733
--- /dev/null
+++ b/packages/interloper-api/tests/test_ingest.py
@@ -0,0 +1,124 @@
+"""Tests for the internal event-ingest endpoint (``interloper_api.routes.internal``).
+
+The endpoint is machine-to-machine: authenticated by a shared service token
+(not a user session) and backed by ``Store.save_event`` (idempotent). A
+lightweight fake store keeps these as pure unit tests.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from uuid import UUID, uuid4
+
+import interloper as il
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from interloper.errors import NotFoundError
+
+from interloper_api.dependencies import get_store, set_ingest_token
+from interloper_api.routes import internal as internal_module
+
+TOKEN = "s3cret-token"
+
+
+class FakeStore:
+    """In-memory stand-in implementing only what the ingest route calls."""
+
+    def __init__(self) -> None:
+        self.org_id = uuid4()
+        self.run_id = uuid4()
+        self.saved: list[tuple[il.Event, UUID, UUID | None]] = []
+
+    def get_run(self, run_id: UUID):
+        if run_id != self.run_id:
+            raise NotFoundError(f"Run {run_id} not found")
+        return SimpleNamespace(id=run_id, org_id=self.org_id)
+
+    def save_event(self, event: il.Event, org_id: UUID, run_id: UUID | None = None):
+        self.saved.append((event, org_id, run_id))
+        return SimpleNamespace(id=event.id)
+
+
+def _client(store: FakeStore, *, token: str | None = TOKEN) -> TestClient:
+    set_ingest_token(token)
+    app = FastAPI()
+    app.include_router(internal_module.router)
+    app.dependency_overrides[get_store] = lambda: store
+    return TestClient(app)
+
+
+def _event(message: str = "hi") -> dict:
+    return il.Event(type=il.EventType.LOG, metadata={"message": message}).to_dict()
+
+
+@pytest.fixture
+def store() -> FakeStore:
+    return FakeStore()
+
+
+@pytest.fixture(autouse=True)
+def _reset_token():
+    yield
+    set_ingest_token(None)
+
+
+def test_ingest_persists_events(store: FakeStore) -> None:
+    client = _client(store)
+    resp = client.post(
+        f"/runs/{store.run_id}/events",
+        json={"events": [_event("a"), _event("b")]},
+        headers={"Authorization": f"Bearer {TOKEN}"},
+    )
+    assert resp.status_code == 200
+    assert resp.json() == {"accepted": 2, "rejected": 0}
+    assert len(store.saved) == 2
+    # org_id is resolved server-side from the run; run_id comes from the path.
+    assert all(org == store.org_id and run == store.run_id for _, org, run in store.saved)
+
+
+def test_ingest_requires_valid_token(store: FakeStore) -> None:
+    client = _client(store)
+    # Missing header.
+    assert client.post(f"/runs/{store.run_id}/events", json={"events": []}).status_code == 401
+    # Wrong token.
+    resp = client.post(
+        f"/runs/{store.run_id}/events",
+        json={"events": []},
+        headers={"Authorization": "Bearer wrong"},
+    )
+    assert resp.status_code == 401
+    assert store.saved == []
+
+
+def test_ingest_disabled_when_unconfigured(store: FakeStore) -> None:
+    client = _client(store, token=None)
+    resp = client.post(
+        f"/runs/{store.run_id}/events",
+        json={"events": [_event()]},
+        headers={"Authorization": f"Bearer {TOKEN}"},
+    )
+    assert resp.status_code == 503
+
+
+def test_ingest_unknown_run_is_404(store: FakeStore) -> None:
+    client = _client(store)
+    resp = client.post(
+        f"/runs/{uuid4()}/events",
+        json={"events": [_event()]},
+        headers={"Authorization": f"Bearer {TOKEN}"},
+    )
+    assert resp.status_code == 404
+    assert store.saved == []
+
+
+def test_ingest_skips_malformed_events(store: FakeStore) -> None:
+    client = _client(store)
+    resp = client.post(
+        f"/runs/{store.run_id}/events",
+        json={"events": [_event("ok"), {"type": "not_a_real_event", "timestamp": "nope"}]},
+        headers={"Authorization": f"Bearer {TOKEN}"},
+    )
+    assert resp.status_code == 200
+    assert resp.json() == {"accepted": 1, "rejected": 1}
+    assert len(store.saved) == 1
diff --git a/packages/interloper-core/src/interloper/cli/commands/run.py b/packages/interloper-core/src/interloper/cli/commands/run.py
index e6b1cc36..2e99da7c 100644
--- a/packages/interloper-core/src/interloper/cli/commands/run.py
+++ b/packages/interloper-core/src/interloper/cli/commands/run.py
@@ -90,7 +90,7 @@ def _cmd_run(args: argparse.Namespace) -> None:
     import sys
 
     from interloper.dag.spec import DAGSpec
-    from interloper.events import EventBus, StderrEventHandler
+    from interloper.events import EventBus, HttpEventSink, StderrEventHandler
     from interloper.runner import ExecutionStatus, build_runner
     from interloper.settings import AppSettings
 
@@ -103,9 +103,26 @@ def _cmd_run(args: argparse.Namespace) -> None:
     settings = AppSettings.get()
     is_container = os.environ.get("INTERLOPER_EVENTS_TO_STDERR") == "true"
 
-    # -- Subscribe stderr event handler (for container event forwarding) ------
+    # -- Event forwarding for child execution ---------------------------------
+    # http:   durable path — the child POSTs events to the ingest endpoint.
+    # stderr: fallback — the host scrapes container logs and re-emits; used for
+    #         local/dev runs and the docker runner when ingest isn't configured.
+    # The two are mutually exclusive: when ingest is configured the host stops
+    # re-emitting from logs, so writing @EVENT lines as well would just be noise.
+    events_cfg = settings.events
+    use_http = bool(args.run_id and events_cfg.ingest_url and events_cfg.ingest_token)
+
+    http_sink = None
+    if use_http:
+        http_sink = HttpEventSink(
+            base_url=events_cfg.ingest_url,
+            token=events_cfg.ingest_token,
+            run_id=args.run_id,
+        )
+        EventBus.subscribe(http_sink)
+
     stderr_handler = None
-    if is_container:
+    if is_container and not use_http:
         stderr_handler = StderrEventHandler()
         EventBus.subscribe(stderr_handler)
 
@@ -149,8 +166,12 @@ def _cmd_run(args: argparse.Namespace) -> None:
             raise SystemExit(1)
 
     finally:
-        if stderr_handler is not None:
+        if stderr_handler is not None or http_sink is not None:
             EventBus.flush(timeout=5.0)
+        if http_sink is not None:
+            http_sink.close()
+            EventBus.unsubscribe(http_sink)
+        if stderr_handler is not None:
             EventBus.unsubscribe(stderr_handler)
 
 
diff --git a/packages/interloper-core/src/interloper/cli/services.py b/packages/interloper-core/src/interloper/cli/services.py
index b62b7c6c..47603f0a 100644
--- a/packages/interloper-core/src/interloper/cli/services.py
+++ b/packages/interloper-core/src/interloper/cli/services.py
@@ -65,6 +65,7 @@ def run_services(
             catalog=catalog,
             auth_config=settings.auth,
             smtp_config=settings.smtp,
+            event_ingest_token=settings.events.ingest_token,
             cors_origins=cors_origins,
         )
 
diff --git a/packages/interloper-core/src/interloper/events/__init__.py b/packages/interloper-core/src/interloper/events/__init__.py
index 28e4ff19..e2e11483 100644
--- a/packages/interloper-core/src/interloper/events/__init__.py
+++ b/packages/interloper-core/src/interloper/events/__init__.py
@@ -2,6 +2,7 @@
 
 from interloper.events.bus import EventBus
 from interloper.events.event import Event
+from interloper.events.http import HttpEventSink
 from interloper.events.logger import EventLogger
 from interloper.events.stderr import StderrEventHandler
 from interloper.events.types import EventType
@@ -11,5 +12,6 @@
     "EventBus",
     "EventLogger",
     "EventType",
+    "HttpEventSink",
     "StderrEventHandler",
 ]
diff --git a/packages/interloper-core/src/interloper/events/bus.py b/packages/interloper-core/src/interloper/events/bus.py
index b31c0d4e..1c6d780a 100644
--- a/packages/interloper-core/src/interloper/events/bus.py
+++ b/packages/interloper-core/src/interloper/events/bus.py
@@ -82,6 +82,21 @@ def emit(event_type: EventType, *, metadata: dict[str, Any] | None = None) -> No
         """
         EventBus()._enqueue(Event(type=event_type, metadata=metadata or {}))
 
+    @staticmethod
+    def emit_event(event: Event) -> None:
+        """Emit an existing :class:`Event` as-is on the global bus.
+
+        Unlike :meth:`emit`, this preserves the event's ``id`` and
+        ``timestamp`` rather than creating a fresh event.  Use it to
+        re-emit an event received from another process (e.g. parsed from a
+        child container's stderr) so its identity survives end-to-end and
+        the event persists idempotently.
+
+        Args:
+            event: The event to enqueue unchanged.
+        """
+        EventBus()._enqueue(event)
+
     @staticmethod
     def subscribe(
         handler: Callable[[Event], None],
diff --git a/packages/interloper-core/src/interloper/events/event.py b/packages/interloper-core/src/interloper/events/event.py
index 5806ee5b..ceafd30f 100644
--- a/packages/interloper-core/src/interloper/events/event.py
+++ b/packages/interloper-core/src/interloper/events/event.py
@@ -4,6 +4,7 @@
 
 import datetime as dt
 import json
+import uuid
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -15,13 +16,19 @@
 class Event:
     """An event emitted during the framework lifecycle.
 
-    Carries a type, UTC timestamp, and an arbitrary metadata dict.
+    Carries a stable unique ``id``, a type, a UTC timestamp, and an
+    arbitrary metadata dict.  The ``id`` is assigned once by the producer
+    and preserved across serialization, so the same logical event can be
+    persisted idempotently even when it is delivered more than once (e.g.
+    re-emitted from a child process's log stream and also written directly).
+
     Supports JSON and dict serialization for forwarding and persistence.
     """
 
     type: EventType
     timestamp: dt.datetime = field(default_factory=lambda: dt.datetime.now(dt.timezone.utc))
     metadata: dict[str, Any] = field(default_factory=dict)
+    id: str = field(default_factory=lambda: str(uuid.uuid4()))
 
     def __str__(self) -> str:
         """Return a human-readable summary line for logging."""
@@ -36,13 +43,14 @@ def __str__(self) -> str:
     # ------------------------------------------------------------------
 
     def to_dict(self) -> dict[str, Any]:
-        """Serialize to a flat dict with ``type``, ``timestamp``, and metadata fields.
+        """Serialize to a flat dict with ``event_id``, ``type``, ``timestamp``, and metadata.
 
         Returns:
-            Dict with ``type`` and ``timestamp`` as top-level keys, plus all
-            metadata entries inlined.
+            Dict with ``event_id``, ``type`` and ``timestamp`` as top-level
+            keys, plus all metadata entries inlined.
         """
         return {
+            "event_id": self.id,
             "type": self.type.value,
             "timestamp": self.timestamp.isoformat(),
             **self.metadata,
@@ -87,8 +95,11 @@ def from_dict(cls, data: dict[str, Any]) -> Event:
         else:
             raise EventError(f"Invalid timestamp value for Event: {timestamp_val!r}")
 
-        metadata = {k: v for k, v in data.items() if k not in ("type", "timestamp")}
+        metadata = {k: v for k, v in data.items() if k not in ("event_id", "type", "timestamp")}
 
+        event_id = data.get("event_id")
+        if event_id is not None:
+            return cls(type=event_type, timestamp=timestamp, metadata=metadata, id=str(event_id))
         return cls(type=event_type, timestamp=timestamp, metadata=metadata)
 
     @classmethod
diff --git a/packages/interloper-core/src/interloper/events/http.py b/packages/interloper-core/src/interloper/events/http.py
new file mode 100644
index 00000000..85369fc4
--- /dev/null
+++ b/packages/interloper-core/src/interloper/events/http.py
@@ -0,0 +1,173 @@
+"""EventBus handler that POSTs events to the internal ingest endpoint.
+
+Per-asset worker processes run in slim, DB-less pods, so instead of relying on
+the host scraping their stdout (lossy), they ship events here.  Events are
+buffered and flushed in batches on a background thread; transient failures are
+retried, and :meth:`close` drains the buffer before the process exits.
+
+Delivery is at-least-once; the ingest endpoint deduplicates by event id, so
+retries never create duplicates.
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+import time
+from typing import TYPE_CHECKING
+
+import httpx
+
+from interloper.events.types import EventType
+
+if TYPE_CHECKING:
+    from interloper.events.event import Event
+
+logger = logging.getLogger(__name__)
+
+# Events the host orchestrator records itself: the run lifecycle, plus the bulk
+# asset-queued it emits at run start (which gives the UI every pending asset up
+# front). The per-asset child must not ship these or they'd be persisted twice.
+_HOST_OWNED_EVENTS = frozenset(
+    {
+        EventType.RUN_STARTED,
+        EventType.RUN_COMPLETED,
+        EventType.RUN_FAILED,
+        EventType.ASSET_QUEUED,
+    }
+)
+
+
+class HttpEventSink:
+    """Buffer events and POST them in batches to the event-ingest endpoint.
+
+    Subscribe an instance to the :class:`~interloper.events.EventBus` and call
+    :meth:`close` before the process exits to drain any buffered events.
+    """
+
+    def __init__(
+        self,
+        *,
+        base_url: str,
+        token: str,
+        run_id: str,
+        batch_size: int = 100,
+        flush_interval: float = 0.5,
+        shutdown_timeout: float = 30.0,
+        exclude_types: frozenset[EventType] = _HOST_OWNED_EVENTS,
+        client: httpx.Client | None = None,
+    ) -> None:
+        """Initialize the sink and start its background flush thread.
+
+        Args:
+            base_url: Base API URL (e.g. the in-cluster service URL).
+            token: Shared ingest service token (sent as a bearer token).
+            run_id: The run these events belong to.
+            batch_size: Max events per POST; also the buffer threshold that
+                triggers an immediate flush.
+            flush_interval: Seconds between idle flushes.
+            shutdown_timeout: Max seconds :meth:`close` spends draining.
+            exclude_types: Event types never shipped (run-level by default).
+            client: Optional pre-built HTTP client (for tests); otherwise a
+                default ``httpx.Client`` is created and owned by the sink.
+        """
+        self._url = f"{base_url.rstrip('/')}/api/internal/runs/{run_id}/events"
+        self._headers = {"Authorization": f"Bearer {token}"}
+        self._batch_size = batch_size
+        self._flush_interval = flush_interval
+        self._shutdown_timeout = shutdown_timeout
+        self._exclude_types = exclude_types
+        self._client = client or httpx.Client(timeout=10.0)
+        self._owns_client = client is None
+
+        self._buffer: list[dict] = []
+        self._lock = threading.Lock()
+        self._wake = threading.Event()
+        self._closing = threading.Event()
+        self._thread = threading.Thread(target=self._loop, name="http-event-sink", daemon=True)
+        self._thread.start()
+
+    def __call__(self, event: Event) -> None:
+        """Buffer *event* for delivery (non-blocking)."""
+        if event.type in self._exclude_types:
+            return
+        with self._lock:
+            self._buffer.append(event.to_dict())
+            pending = len(self._buffer)
+        if pending >= self._batch_size:
+            self._wake.set()
+
+    def close(self) -> None:
+        """Stop the background thread after a final, time-bounded drain."""
+        self._closing.set()
+        self._wake.set()
+        self._thread.join(timeout=self._shutdown_timeout + 5.0)
+        if self._owns_client:
+            self._client.close()
+
+    # -- internals ------------------------------------------------------------
+
+    def _loop(self) -> None:
+        """Background loop: flush on an interval until closed, then drain."""
+        while not self._closing.is_set():
+            self._wake.wait(self._flush_interval)
+            self._wake.clear()
+            self._drain(deadline=None)
+        # Final drain on shutdown, bounded by shutdown_timeout.
+        self._drain(deadline=time.monotonic() + self._shutdown_timeout)
+
+    def _drain(self, *, deadline: float | None) -> None:
+        """Flush buffered events.
+
+        With ``deadline=None`` (periodic flush) a transient failure simply
+        returns and is retried next tick.  With a deadline (shutdown) the same
+        batch is retried with backoff until it succeeds or the deadline passes.
+        """
+        backoff = 0.5
+        while True:
+            with self._lock:
+                if not self._buffer:
+                    return
+                batch = list(self._buffer[: self._batch_size])
+            outcome = self._post(batch)
+            if outcome == "ok":
+                with self._lock:
+                    del self._buffer[: len(batch)]
+                backoff = 0.5
+                continue
+            if outcome == "drop":
+                logger.error("Dropping %d event(s): ingest rejected the batch", len(batch))
+                with self._lock:
+                    del self._buffer[: len(batch)]
+                continue
+            # transient failure
+            if deadline is None:
+                return  # retry on the next tick
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                with self._lock:
+                    lost = len(self._buffer)
+                logger.warning("Event ingest unreachable at shutdown; dropping %d unsent event(s)", lost)
+                return
+            time.sleep(min(backoff, remaining))
+            backoff = min(backoff * 2, 5.0)
+
+    def _post(self, batch: list[dict]) -> str:
+        """POST a batch of events.
+
+        Returns:
+            ``ok`` if persisted, ``retry`` for a transient failure, or
+            ``drop`` for a permanent one.
+        """
+        try:
+            resp = self._client.post(self._url, headers=self._headers, json={"events": batch})
+        except httpx.HTTPError as e:
+            logger.debug("Event ingest POST failed: %s", e)
+            return "retry"
+        if resp.status_code < 300:
+            return "ok"
+        if resp.status_code >= 500:
+            logger.debug("Event ingest %s (will retry)", resp.status_code)
+            return "retry"
+        logger.error("Event ingest rejected batch: %s %s", resp.status_code, resp.text[:200])
+        return "drop"
diff --git a/packages/interloper-core/src/interloper/settings.py b/packages/interloper-core/src/interloper/settings.py
index 0fe088ba..5d975d33 100644
--- a/packages/interloper-core/src/interloper/settings.py
+++ b/packages/interloper-core/src/interloper/settings.py
@@ -101,6 +101,21 @@ def enabled(self) -> bool:
         return bool(self.host and self.user and self.password)
 
 
+class EventsSettings(BaseSettings):
+    """Event-ingest settings.
+
+    Used by the API to authenticate the internal event-ingest endpoint, and
+    by per-asset worker processes to reach it.  ``ingest_token`` is a shared
+    service secret; when empty the ingest endpoint is disabled.  ``ingest_url``
+    is the base API URL workers POST events to (e.g. the in-cluster service).
+    """
+
+    model_config = SettingsConfigDict(env_prefix=f"{PREFIX}EVENTS_")
+
+    ingest_token: str = ""
+    ingest_url: str = ""
+
+
 class LauncherSettings(BaseSettings):
     """Launcher settings (type + launcher-specific config)."""
 
@@ -158,6 +173,7 @@ class AppSettings(BaseSettings):
     smtp: SmtpSettings = Field(default_factory=SmtpSettings)
     worker: WorkerSettings = Field(default_factory=WorkerSettings)
     reaper: ReaperSettings = Field(default_factory=ReaperSettings)
+    events: EventsSettings = Field(default_factory=EventsSettings)
     launcher: LauncherSettings = Field(default_factory=LauncherSettings)
     runner: RunnerSettings = Field(default_factory=RunnerSettings)
     catalog: list[str] = Field(default_factory=list)
diff --git a/packages/interloper-core/tests/events/__init__.py b/packages/interloper-core/tests/events/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/packages/interloper-core/tests/events/test_event.py b/packages/interloper-core/tests/events/test_event.py
new file mode 100644
index 00000000..6705fb30
--- /dev/null
+++ b/packages/interloper-core/tests/events/test_event.py
@@ -0,0 +1,64 @@
+"""Tests for :class:`Event` identity and serialization round-trips."""
+
+from __future__ import annotations
+
+from interloper.events import Event, EventBus, EventType
+
+
+def test_events_get_unique_ids_by_default() -> None:
+    """Each freshly constructed event gets its own id."""
+    e1 = Event(type=EventType.LOG)
+    e2 = Event(type=EventType.LOG)
+    assert e1.id
+    assert e2.id
+    assert e1.id != e2.id
+
+
+def test_to_dict_includes_event_id() -> None:
+    """``event_id`` is serialized as a top-level key alongside type/timestamp."""
+    event = Event(type=EventType.ASSET_STARTED, metadata={"asset_key": "foo"})
+    data = event.to_dict()
+    assert data["event_id"] == event.id
+    assert data["type"] == "asset_started"
+    assert data["asset_key"] == "foo"
+
+
+def test_json_round_trip_preserves_id_and_timestamp() -> None:
+    """Serializing and parsing an event keeps its id, type, timestamp and metadata."""
+    event = Event(type=EventType.ASSET_FAILED, metadata={"asset_key": "foo", "error": "boom"})
+
+    restored = Event.from_json(event.to_json())
+
+    assert restored.id == event.id
+    assert restored.type == event.type
+    assert restored.timestamp == event.timestamp
+    assert restored.metadata["asset_key"] == "foo"
+    assert restored.metadata["error"] == "boom"
+    # event_id is a top-level field, not metadata.
+    assert "event_id" not in restored.metadata
+
+
+def test_from_dict_without_event_id_generates_one() -> None:
+    """A legacy payload lacking ``event_id`` still yields an event with an id."""
+    event = Event.from_dict({"type": "log", "timestamp": "2026-06-04T12:00:00+00:00"})
+    assert event.id
+
+
+def test_emit_event_preserves_identity() -> None:
+    """``emit_event`` delivers the event unchanged, keeping its id and timestamp."""
+    captured: list[Event] = []
+
+    def handler(event: Event) -> None:
+        captured.append(event)
+
+    EventBus.subscribe(handler)
+    try:
+        original = Event(type=EventType.LOG, metadata={"message": "hi"})
+        EventBus.emit_event(original)
+        EventBus.flush(timeout=5.0)
+    finally:
+        EventBus.unsubscribe(handler)
+
+    match = [e for e in captured if e.id == original.id]
+    assert match, "emit_event should deliver an event preserving its id"
+    assert match[0].timestamp == original.timestamp
diff --git a/packages/interloper-core/tests/events/test_http.py b/packages/interloper-core/tests/events/test_http.py
new file mode 100644
index 00000000..ba3b49ba
--- /dev/null
+++ b/packages/interloper-core/tests/events/test_http.py
@@ -0,0 +1,97 @@
+"""Tests for :class:`HttpEventSink`."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from uuid import uuid4
+
+import httpx
+
+from interloper.events import Event, EventType, HttpEventSink
+
+
+def _sink(handler: Callable[[httpx.Request], httpx.Response]) -> HttpEventSink:
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    return HttpEventSink(
+        base_url="http://api",
+        token="tok",
+        run_id=str(uuid4()),
+        client=client,
+        flush_interval=0.02,
+    )
+
+
+def test_posts_buffered_events_and_excludes_host_owned() -> None:
+    """Asset lifecycle/log events ship; host-owned events (run-level, queued) don't."""
+    posted: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        assert request.headers["authorization"] == "Bearer tok"
+        posted.append(json.loads(request.content))
+        return httpx.Response(200, json={"accepted": 1, "rejected": 0})
+
+    sink = _sink(handler)
+    sink(Event(type=EventType.LOG, metadata={"message": "a"}))
+    sink(Event(type=EventType.RUN_STARTED))
+    sink(Event(type=EventType.ASSET_QUEUED, metadata={"asset_key": "x"}))
+    sink(Event(type=EventType.ASSET_STARTED, metadata={"asset_key": "x"}))
+    sink.close()
+
+    types = [e["type"] for batch in posted for e in batch["events"]]
+    assert "log" in types
+    assert "asset_started" in types
+    assert "run_started" not in types
+    assert "asset_queued" not in types
+
+
+def test_targets_run_scoped_ingest_url() -> None:
+    """Events POST to /api/internal/runs/<run_id>/events."""
+    seen: dict[str, str] = {}
+    run_id = str(uuid4())
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        return httpx.Response(200, json={})
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    sink = HttpEventSink(base_url="http://api/", token="t", run_id=run_id, client=client, flush_interval=0.02)
+    sink(Event(type=EventType.LOG))
+    sink.close()
+
+    assert seen["url"] == f"http://api/api/internal/runs/{run_id}/events"
+
+
+def test_retries_transient_failure_then_succeeds() -> None:
+    """A 5xx is retried until it succeeds."""
+    attempts = {"n": 0}
+    delivered: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        attempts["n"] += 1
+        if attempts["n"] == 1:
+            return httpx.Response(503, json={})
+        delivered.append(json.loads(request.content))
+        return httpx.Response(200, json={})
+
+    sink = _sink(handler)
+    sink(Event(type=EventType.LOG, metadata={"message": "boom"}))
+    sink.close()
+
+    assert attempts["n"] >= 2
+    assert delivered
+
+
+def test_drops_on_client_error_without_spinning() -> None:
+    """A 4xx (e.g. bad token) drops the batch instead of retrying forever."""
+    attempts = {"n": 0}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        attempts["n"] += 1
+        return httpx.Response(401, json={})
+
+    sink = _sink(handler)
+    sink(Event(type=EventType.LOG))
+    sink.close()
+
+    assert attempts["n"] == 1
diff --git a/packages/interloper-db/src/interloper_db/store/runs.py b/packages/interloper-db/src/interloper_db/store/runs.py
index 989ad01b..41c4760a 100644
--- a/packages/interloper-db/src/interloper_db/store/runs.py
+++ b/packages/interloper-db/src/interloper_db/store/runs.py
@@ -5,11 +5,12 @@
 import datetime as dt
 import logging
 from datetime import datetime, timedelta, timezone
-from uuid import UUID
+from uuid import UUID, uuid4
 
 import interloper as il
 from interloper.errors import NotFoundError, RunNotFoundError
 from sqlalchemy import func, text
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlmodel import Session, col, select
 
 from interloper_db.engine import get_engine
@@ -17,12 +18,42 @@
 
 logger = logging.getLogger(__name__)
 
+_MAX_EVENT_TEXT = 60_000
+"""Defensive cap for free-text event fields (well under Postgres limits)."""
+
+
+def _sanitize_text(value: str | None, *, max_len: int = _MAX_EVENT_TEXT) -> str | None:
+    """Make a free-text event field safe to persist.
+
+    Postgres ``text`` columns cannot store NUL bytes (``0x00``) — a single
+    one makes the whole INSERT raise, which (because event persistence is
+    best-effort) would silently drop the event.  Strip NULs and cap the
+    length so an oversized traceback can't fail the write either.
+
+    Returns:
+        The cleaned string, or ``None`` if *value* is ``None``.
+    """
+    if value is None:
+        return None
+    cleaned = value.replace("\x00", "")
+    if len(cleaned) > max_len:
+        cleaned = cleaned[:max_len] + "…[truncated]"
+    return cleaned
+
 
 class RunMixin:
     """Store methods for runs, events, and backfills."""
 
     def save_event(self, event: il.Event, org_id: UUID, run_id: UUID | None = None) -> Event:
-        """Persist a framework event to the database.
+        """Persist a framework event to the database, idempotently.
+
+        The event's producer-assigned ``id`` becomes the row primary key
+        and the insert is an upsert (``ON CONFLICT DO NOTHING``), so the
+        same event delivered more than once — e.g. re-emitted from a child
+        container's log stream and also written directly — yields a single
+        row rather than a duplicate or an error.  Free-text fields are
+        sanitized so a stray NUL byte or oversized traceback can't fail the
+        write and silently drop the event.
 
         Args:
             event: The framework Event.
@@ -32,25 +63,35 @@ def save_event(self, event: il.Event, org_id: UUID, run_id: UUID | None = None)
         Returns:
             The saved Event row.
         """
+        meta = event.metadata
+        try:
+            event_id = UUID(event.id)
+        except (ValueError, TypeError):
+            event_id = uuid4()
+
+        values: dict[str, object | None] = {
+            "id": event_id,
+            "org_id": org_id,
+            "run_id": run_id,
+            "backfill_id": UUID(meta["backfill_id"]) if meta.get("backfill_id") else None,
+            "event_type": event.type.value,
+            "asset_id": UUID(meta["asset_id"]) if meta.get("asset_id") else None,
+            "asset_key": _sanitize_text(meta.get("asset_key")),
+            "partition_or_window": _sanitize_text(meta.get("partition_or_window")),
+            "error": _sanitize_text(meta.get("error")),
+            "traceback": _sanitize_text(meta.get("traceback")),
+            "message": _sanitize_text(meta.get("message")),
+            "timestamp": event.timestamp,
+        }
 
         with Session(get_engine()) as session:
-            db_event = Event(
-                org_id=org_id,
-                run_id=run_id,
-                backfill_id=UUID(event.metadata["backfill_id"]) if event.metadata.get("backfill_id") else None,
-                event_type=event.type.value,
-                asset_id=UUID(event.metadata["asset_id"]) if event.metadata.get("asset_id") else None,
-                asset_key=event.metadata.get("asset_key"),
-                partition_or_window=event.metadata.get("partition_or_window"),
-                error=event.metadata.get("error"),
-                traceback=event.metadata.get("traceback"),
-                message=event.metadata.get("message"),
-                timestamp=event.timestamp,
-            )
-            session.add(db_event)
+            stmt = pg_insert(Event).values(**values).on_conflict_do_nothing(index_elements=["id"])
+            session.execute(stmt)
             session.commit()
-            session.refresh(db_event)
-            return db_event
+            saved = session.get(Event, event_id)
+            if saved is None:  # pragma: no cover - only if the row was concurrently deleted
+                raise RuntimeError(f"Event {event_id} missing immediately after upsert")
+            return saved
 
     def list_events(
         self,
diff --git a/packages/interloper-db/tests/test_events.py b/packages/interloper-db/tests/test_events.py
new file mode 100644
index 00000000..b573953e
--- /dev/null
+++ b/packages/interloper-db/tests/test_events.py
@@ -0,0 +1,29 @@
+"""Tests for event persistence helpers."""
+
+from __future__ import annotations
+
+from interloper_db.store.runs import _sanitize_text
+
+
+def test_sanitize_strips_nul_bytes() -> None:
+    """NUL bytes (which Postgres text rejects) are removed."""
+    assert _sanitize_text("a\x00b\x00c") == "abc"
+
+
+def test_sanitize_passes_through_none() -> None:
+    """``None`` stays ``None``."""
+    assert _sanitize_text(None) is None
+
+
+def test_sanitize_keeps_normal_text() -> None:
+    """Ordinary text is returned unchanged."""
+    assert _sanitize_text("hello world") == "hello world"
+
+
+def test_sanitize_truncates_oversized() -> None:
+    """Oversized values are capped and marked as truncated."""
+    out = _sanitize_text("x" * 100, max_len=10)
+    assert out is not None
+    assert out.startswith("x" * 10)
+    assert out.endswith("[truncated]")
+    assert len(out) < 100
diff --git a/packages/interloper-docker/src/interloper_docker/runner.py b/packages/interloper-docker/src/interloper_docker/runner.py
index bb0903e3..62d9bea9 100644
--- a/packages/interloper-docker/src/interloper_docker/runner.py
+++ b/packages/interloper-docker/src/interloper_docker/runner.py
@@ -318,7 +318,7 @@ def stream_logs() -> None:
                                 event_asset_id = event.metadata.get("asset_id")
                                 if event_asset_id and event_asset_id != target_asset_id:
                                     continue
-                                EventBus.emit(event.type, metadata=event.metadata)
+                                EventBus.emit_event(event)
                                 continue
                         except Exception:  # noqa: BLE001, S110
                             pass
diff --git a/packages/interloper-k8s/src/interloper_k8s/launcher.py b/packages/interloper-k8s/src/interloper_k8s/launcher.py
index 64730737..940a7472 100644
--- a/packages/interloper-k8s/src/interloper_k8s/launcher.py
+++ b/packages/interloper-k8s/src/interloper_k8s/launcher.py
@@ -237,6 +237,13 @@ def _build_env(self) -> list[client.V1EnvVar]:
         if encryption_key:
             env_map["INTERLOPER_ENCRYPTION_KEY"] = encryption_key
 
+        # Forward event-ingest config to the run pod, which in turn forwards it
+        # to per-asset child pods so they can persist events directly.
+        for var in ("INTERLOPER_EVENTS_INGEST_URL", "INTERLOPER_EVENTS_INGEST_TOKEN"):
+            value = os.environ.get(var)
+            if value:
+                env_map[var] = value
+
         return [client.V1EnvVar(name=k, value=v) for k, v in env_map.items()]
 
     def _build_resources(self) -> client.V1ResourceRequirements | None:
diff --git a/packages/interloper-k8s/src/interloper_k8s/runner.py b/packages/interloper-k8s/src/interloper_k8s/runner.py
index 4322a4b2..c91e9dd5 100644
--- a/packages/interloper-k8s/src/interloper_k8s/runner.py
+++ b/packages/interloper-k8s/src/interloper_k8s/runner.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import json
+import os
 import threading
 import time
 from concurrent.futures import Future, ThreadPoolExecutor
@@ -25,12 +26,15 @@
 from kubernetes.client import V1Job
 from pydantic import Field, PrivateAttr
 
-# Events emitted by the container's inner run — not forwarded to the host.
-_RUN_EVENTS = frozenset(
+# Events the host orchestrator records itself — the run lifecycle plus the bulk
+# asset-queued emitted at run start. They are dropped from the child's log
+# stream so they aren't persisted twice.
+_HOST_OWNED_EVENTS = frozenset(
     {
         EventType.RUN_STARTED,
         EventType.RUN_COMPLETED,
         EventType.RUN_FAILED,
+        EventType.ASSET_QUEUED,
     }
 )
 
@@ -308,10 +312,15 @@ def _build_command(
         return cmd
 
     def _build_env(self) -> list[client.V1EnvVar]:
-        """Build the environment variables for the container."""
-        env = [client.V1EnvVar(name=k, value=v) for k, v in self.env_vars.items()]
-        env.append(client.V1EnvVar(name="INTERLOPER_EVENTS_TO_STDERR", value="true"))
-        return env
+        """Build the environment variables for the per-asset container."""
+        env_map = dict(self.env_vars)
+        env_map["INTERLOPER_EVENTS_TO_STDERR"] = "true"
+        # Forward event-ingest config so the child can persist events directly.
+        for var in ("INTERLOPER_EVENTS_INGEST_URL", "INTERLOPER_EVENTS_INGEST_TOKEN"):
+            value = os.environ.get(var)
+            if value:
+                env_map[var] = value
+        return [client.V1EnvVar(name=k, value=v) for k, v in env_map.items()]
 
     def _build_resources(self) -> client.V1ResourceRequirements | None:
         """Build the resource requirements for the container."""
@@ -339,12 +348,27 @@ def _build_tolerations(self) -> list[client.V1Toleration]:
             for t in self.tolerations
         ]
 
+    @property
+    def _child_persists_directly(self) -> bool:
+        """Whether per-asset child pods persist their own events via the ingest endpoint.
+
+        When true, the host must not re-emit events parsed from container logs:
+        the child already wrote them (idempotently) to the API, so re-emitting
+        would be redundant work and a second, lossier persistence path.
+        """
+        return bool(
+            os.environ.get("INTERLOPER_EVENTS_INGEST_URL")
+            and os.environ.get("INTERLOPER_EVENTS_INGEST_TOKEN")
+        )
+
     def _start_log_streaming(self, job_name: str, *, target_asset_id: str) -> None:
         """Stream events from a K8s pod's logs to the host EventBus.
 
-        Only events for the **target asset** are forwarded.
-        Container-internal ``RUN_*`` events and events for
-        non-materializable parent assets are dropped.
+        Only events for the **target asset** are forwarded, and only when the
+        child is not persisting its own events directly (see
+        :attr:`_child_persists_directly`). Host-owned events (run lifecycle and
+        the bulk asset-queued) and events for non-materializable parent assets
+        are always dropped.
 
         Args:
             job_name: The K8s Job name to stream logs from.
@@ -356,6 +380,9 @@ def _start_log_streaming(self, job_name: str, *, target_asset_id: str) -> None:
         core_v1 = self._core_v1
 
         def stream_logs() -> None:
+            # When the child persists its own events directly, the host doesn't
+            # re-emit them — it just drains the log stream.
+            forward_events = not self._child_persists_directly
             pod_name: str | None = None
             while not self._stop_log_streaming.is_set():
                 try:
@@ -396,22 +423,24 @@ def stream_logs() -> None:
                         try:
                             event = parse_event_from_log_line(line)
                             if event is not None:
-                                if event.type in _RUN_EVENTS:
+                                if event.type in _HOST_OWNED_EVENTS:
                                     continue
                                 event_asset_id = event.metadata.get("asset_id")
                                 if event_asset_id and event_asset_id != target_asset_id:
                                     continue
-                                EventBus.emit(event.type, metadata=event.metadata)
+                                if forward_events:
+                                    EventBus.emit_event(event)
                         except Exception:  # noqa: BLE001, S110
                             pass
                 buf = buf.rstrip()
                 if buf:
                     try:
                         event = parse_event_from_log_line(buf)
-                        if event is not None and event.type not in _RUN_EVENTS:
+                        if event is not None and event.type not in _HOST_OWNED_EVENTS:
                             event_asset_id = event.metadata.get("asset_id")
                             if not event_asset_id or event_asset_id == target_asset_id:
-                                EventBus.emit(event.type, metadata=event.metadata)
+                                if forward_events:
+                                    EventBus.emit_event(event)
                     except Exception:  # noqa: BLE001, S110
                         pass
             except Exception:  # noqa: BLE001, S110