fix(plugin): report when an installation failure happened, not when it was sampled

The connector stored only uuid/code/message, so the ops reporter stamped every entry with its own sampling time and a stale failure looked freshly broken on each report. Failures now carry failed_at and the payload publishes it.

src/langbot/pkg/plugin/connector.py

@@ -56,7 +56,9 @@ from langbot_plugin.api.entities.builtin.command import (
     errors as command_errors,
 )
 from langbot_plugin.runtime.plugin.mgr import PluginInstallSource
+from langbot_plugin.runtime.io.handler import SHARED_WORKER_FILE_STORAGE_DIR
 from langbot_plugin.runtime.security import (
+    PLUGIN_FILE_STORAGE_DIR_ENV,
     PLUGIN_RUNTIME_CONTROL_TOKEN_ENV,
     PLUGIN_RUNTIME_CONTROL_TOKEN_HEADER,
     validate_runtime_secret,
@@ -231,6 +233,7 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         self.runtime_profile: typing.Literal['oss_dev', 'shared'] = (
             'shared' if getattr(getattr(ap, 'deployment', None), 'mode', 'oss') == 'cloud' else 'oss_dev'
         )
+        self._align_plugin_file_transfer_root()
         self.runtime_identity: RuntimeIdentity | None = None
         self._runtime_id = self._build_runtime_id()
         self.worker_policy: PluginWorkerPolicy | None = None
@@ -256,6 +259,24 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             'missing_artifacts': 0,
         }
 
+    def _align_plugin_file_transfer_root(self) -> None:
+        """Use the Runtime's plugin file-transfer root when running shared.
+
+        The SDK binds its transfer root to one directory inode with mode 0700 and
+        picks the default per process: a shared worker uses
+        ``SHARED_WORKER_FILE_STORAGE_DIR``, every other process falls back to
+        ``data/temp/lbp``. A host talking to a shared Runtime must use the same root,
+        otherwise plugin icons and assets die with "Invalid file transfer
+        capability" (the icon route then answers 500/404 instead of the image).
+        An explicit operator value always wins.
+        """
+
+        if self.runtime_profile != 'shared':
+            return
+        if os.environ.get(PLUGIN_FILE_STORAGE_DIR_ENV):
+            return
+        os.environ[PLUGIN_FILE_STORAGE_DIR_ENV] = SHARED_WORKER_FILE_STORAGE_DIR
+
     @staticmethod
     def _build_runtime_id() -> str:
         """Return the durable identity of this instance's shared Runtime."""
@@ -634,6 +655,10 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             'installation_uuid': installation_uuid,
             'error_code': error_code,
             'message': message,
+            # When the failure actually happened. The ops reporter must publish this
+            # instead of its own sampling time, otherwise every sample re-stamps the
+            # same failure with the newest report time.
+            'failed_at': _utc_now_iso(),
         }
         self._installation_failures[installation_uuid] = failure
         self.ap.logger.error(
@@ -675,9 +700,15 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 'installation_uuid': installation_uuid,
                 'error_code': error_code,
                 'message': message,
+                # See _raise_apply_failure: never let the reporter substitute its own
+                # sampling time for the moment the failure was observed.
+                'failed_at': _utc_now_iso(),
             }
             failures[installation_uuid] = failure
-            if self._installation_failures.get(installation_uuid) != failure:
+            previous = self._installation_failures.get(installation_uuid) or {}
+            if {key: value for key, value in previous.items() if key != 'failed_at'} != {
+                key: value for key, value in failure.items() if key != 'failed_at'
+            }:
                 self.ap.logger.error(
                     'Plugin installation %s failed during reconcile [%s]: %s',
                     installation_uuid,

src/langbot/pkg/plugin/runtime_ops.py

@@ -356,7 +356,10 @@ class RuntimeOpsReporter:
                     'execution_mode': mode,
                     'error_code': str(record.get('error_code', '') or ''),
                     'message': _sanitize_failure_message(record.get('message')),
-                    'observed_at': observed_at,
+                    # The moment the failure was observed by the connector. Falling back to
+                    # the sampling time would make a stale failure look freshly broken on
+                    # every report.
+                    'observed_at': str(record.get('failed_at') or observed_at),
                 }
             )
         return failures

tests/unit_tests/plugin/test_connector_reconcile.py

@@ -735,11 +735,13 @@ async def test_apply_dependency_failure_raises_stable_observable_error():
     assert error.installation_uuid == setting.installation_uuid
     assert error.error_code == 'dependency_prepare_failed'
     assert '[dependency_prepare_failed]' in str(error)
-    assert connector._installation_failures[setting.installation_uuid] == {
+    recorded = connector._installation_failures[setting.installation_uuid]
+    assert {key: value for key, value in recorded.items() if key != 'failed_at'} == {
         'installation_uuid': setting.installation_uuid,
         'error_code': 'dependency_prepare_failed',
         'message': 'Plugin dependency installer exited with code 1',
     }
+    assert str(recorded['failed_at']).endswith('Z')
     connector.ap.logger.error.assert_called_once()
 
 
@@ -767,9 +769,12 @@ async def test_shared_reconcile_records_one_failure_without_blocking_other_state
         setting_a.installation_uuid,
         setting_b.installation_uuid,
     }
-    assert connector._installation_failures == {
-        setting_a.installation_uuid: failure,
-    }
+    recorded_failures = connector._installation_failures
+    assert {
+        uuid: {key: value for key, value in record.items() if key != 'failed_at'}
+        for uuid, record in recorded_failures.items()
+    } == {setting_a.installation_uuid: failure}
+    assert str(recorded_failures[setting_a.installation_uuid]['failed_at']).endswith('Z')
     assert set(connector._known_desired_states) == {
         setting_a.installation_uuid,
         setting_b.installation_uuid,

tests/unit_tests/plugin/test_runtime_ops_reporter.py

@@ -282,6 +282,39 @@ class TestUpload:
         await reporter.stop()
 
 
+class TestFailureTimestamps:
+    def test_publishes_when_the_failure_happened_not_when_it_was_sampled(self):
+        failures = {
+            'inst-1': {
+                'installation_uuid': 'inst-1',
+                'error_code': 'dependency_prepare_failed',
+                'message': 'prepare failed',
+                'failed_at': '2026-10-01T07:11:00Z',
+            }
+        }
+        reporter = RuntimeOpsReporter(make_app(connector=make_connector(failures=failures)))
+
+        payload = reporter.build_payload({'live': True})
+
+        entry = payload['failures'][0]
+        assert entry['observed_at'] == '2026-10-01T07:11:00Z'
+        assert entry['observed_at'] != payload['generated_at']
+
+    def test_legacy_records_without_a_timestamp_fall_back_to_the_sample_time(self):
+        failures = {
+            'inst-2': {
+                'installation_uuid': 'inst-2',
+                'error_code': 'worker_launch_failed',
+                'message': 'boom',
+            }
+        }
+        reporter = RuntimeOpsReporter(make_app(connector=make_connector(failures=failures)))
+
+        payload = reporter.build_payload({'live': True})
+
+        assert payload['failures'][0]['observed_at'] == payload['generated_at']
+
+
 def test_payload_never_contains_a_control_token(monkeypatch):
     monkeypatch.setenv(PLUGIN_RUNTIME_CONTROL_TOKEN_ENV, 'runtime-control-secret')
     monkeypatch.setenv(_CONTROL_PLANE_TOKEN_ENV, 'space-control-plane-secret')