up

reme4/components/job/background_job.py

@@ -1,7 +1,9 @@
 """Long-running background job with optional supervisor."""
 
 import asyncio
+import contextlib
 import random
+import time
 
 from .base_job import BaseJob
 from ..component_registry import R
@@ -18,6 +20,12 @@ class BackgroundJob(BaseJob):
     exponential backoff (backoff_base * 2**attempt, capped at backoff_cap)
     plus ±50% jitter. __call__ must NOT swallow exceptions, otherwise the
     supervisor cannot trigger a restart.
+
+    On close, the stop_event is set and the task is given up to
+    ``close_timeout`` seconds to exit gracefully; after that it is cancelled.
+    If a single run survives at least ``attempt_reset_after`` seconds before
+    crashing, the backoff attempt counter resets — so a long-stable job that
+    eventually crashes restarts quickly rather than at the capped delay.
     """
 
     def __init__(
@@ -25,13 +33,17 @@ class BackgroundJob(BaseJob):
         supervisor: bool = True,
         backoff_base: float = 1.0,
         backoff_cap: float = 60.0,
+        close_timeout: float = 5.0,
+        attempt_reset_after: float = 60.0,
         **kwargs,
     ):
         super().__init__(**kwargs)
         self.supervisor: bool = supervisor
         self.backoff_base: float = backoff_base
         self.backoff_cap: float = backoff_cap
-        self._stop_event: asyncio.Event = asyncio.Event()
+        self.close_timeout: float = close_timeout
+        self.attempt_reset_after: float = attempt_reset_after
+        self._stop_event: asyncio.Event | None = None
         self._task: asyncio.Task | None = None
 
     async def _start(self) -> None:
@@ -40,25 +52,35 @@ class BackgroundJob(BaseJob):
         self._task = asyncio.create_task(self._run_with_supervisor())
 
     async def _close(self) -> None:
-        self._stop_event.set()
+        if self._stop_event is not None:
+            self._stop_event.set()
         if self._task is not None:
             try:
-                await self._task
+                await asyncio.wait_for(asyncio.shield(self._task), timeout=self.close_timeout)
+            except asyncio.TimeoutError:
+                self._task.cancel()
+                with contextlib.suppress(BaseException):
+                    await self._task
             except Exception:
                 self.logger.exception(f"Background task '{self.name}' raised during close")
             self._task = None
         await super()._close()
 
     async def _run_with_supervisor(self) -> None:
+        assert self._stop_event is not None
         attempt = 0
         while not self._stop_event.is_set():
+            started_at = time.monotonic()
             try:
                 await self()
                 return
             except Exception as e:
                 if not self.supervisor:
                     raise
-                delay = min(self.backoff_base * (2**attempt), self.backoff_cap) * (0.5 + random.random())
+                if time.monotonic() - started_at >= self.attempt_reset_after:
+                    attempt = 0
+                capped = min(self.backoff_base * (2**attempt), self.backoff_cap)
+                delay = min(capped * (0.5 + random.random()), self.backoff_cap)
                 self.logger.exception(f"job body crashed, restart in {delay:.2f}s error={e}")
                 attempt += 1
                 try:
@@ -67,8 +89,9 @@ class BackgroundJob(BaseJob):
                     pass
 
     async def __call__(self, **kwargs) -> Response:
-        """Default body: run step_components in order; errors propagate to supervisor."""
-        context = RuntimeContext(stop_event=self._stop_event, **self.kwargs)
-        for step in self.step_components:
+        """Default body: run steps in order; errors propagate to supervisor."""
+        merged = {**self.kwargs, **kwargs}
+        context = RuntimeContext(stop_event=self._stop_event, **merged)
+        for step in self._build_steps():
             await step(context)
         return context.response

reme4/components/job/base_job.py

@@ -1,11 +1,16 @@
 """Base job component for sequential step execution."""
 
+from typing import TYPE_CHECKING
+
 from ..base_component import BaseComponent
 from ..component_registry import R
 from ..runtime_context import RuntimeContext
 from ...enumeration import ComponentEnum
 from ...schema import ComponentConfig, Response
 
+if TYPE_CHECKING:
+    from ...steps import BaseStep
+
 
 @R.register("base")
 class BaseJob(BaseComponent):
@@ -24,14 +29,12 @@ class BaseJob(BaseComponent):
         self.description = description
         self.parameters = parameters or {}
         self.step_configs = steps or []
-
-        from ...steps import BaseStep
-
-        self.step_components: list[BaseStep] = []
+        self.step_specs: list[tuple[type["BaseStep"], dict]] = []
 
     async def _start(self) -> None:
-        """Resolve step configs into instantiated step components."""
-        assert self.app_context is not None, "app_context must be provided"
+        """Resolve step configs into (cls, params) pairs; defer instantiation to __call__."""
+        if self.app_context is None:
+            raise RuntimeError(f"app_context must be provided for job '{self.name}'")
         for raw in self.step_configs:
             config = raw if isinstance(raw, ComponentConfig) else ComponentConfig(**raw)
             if not config.backend:
@@ -41,17 +44,21 @@ class BaseJob(BaseComponent):
                 raise ValueError(f"Unregistered backend '{config.backend}' of type '{ComponentEnum.STEP}'")
             params = config.model_dump()
             params["app_context"] = self.app_context
-            self.step_components.append(step_cls(**params))
+            self.step_specs.append((step_cls, params))
 
     async def _close(self) -> None:
-        """Release all step components."""
-        self.step_components.clear()
+        """Release all step specs."""
+        self.step_specs.clear()
+
+    def _build_steps(self) -> list["BaseStep"]:
+        """Instantiate fresh step instances from stored specs."""
+        return [step_cls(**dict(params)) for step_cls, params in self.step_specs]
 
     async def __call__(self, **kwargs) -> Response:
         """Execute all steps in order and return the final response."""
         context = RuntimeContext(**kwargs)
         try:
-            for step in self.step_components:
+            for step in self._build_steps():
                 await step(context)
         except Exception as e:
             self.logger.exception(f"Failed to execute job: {e}")

reme4/components/job/stream_job.py

@@ -14,7 +14,7 @@ class StreamJob(BaseJob):
         """Execute steps and stream output; errors are sent as ERROR chunks."""
         context = RuntimeContext(**kwargs)
         try:
-            for step in self.step_components:
+            for step in self._build_steps():
                 await step(context)
         except Exception as e:
             await context.add_stream_string(str(e), ChunkEnum.ERROR)