zincware
diff --git a/‎AGENTS.md‎
Lines changed: 89 additions & 4 deletions b/‎AGENTS.md‎
Lines changed: 89 additions & 4 deletions
diff --git a/‎laufband/db.py‎
Lines changed: 25 additions & 2 deletions b/‎laufband/db.py‎
Lines changed: 25 additions & 2 deletions
diff --git a/‎laufband/heartbeat.py‎
Lines changed: 12 additions & 35 deletions b/‎laufband/heartbeat.py‎
Lines changed: 12 additions & 35 deletions
diff --git a/‎laufband/time_provider.py‎
Lines changed: 51 additions & 0 deletions b/‎laufband/time_provider.py‎
Lines changed: 51 additions & 0 deletions
@@ -6,6 +6,14 @@ Laufband is a Python library that enables parallel iteration over datasets from
 
 ## Working Effectively
 
+This is a new application and you must not consider migrations or backwards compatibility.
+Design all new features with maintainability and performance in mind.
+Use KISS, DRY, SOLID and YAGNI principles.
+When refactoring, you can break backwards compatibility.
+Always consider a better design approach compared to the existing one.
+Consider multiple approaches, review them against the principles above, the existing methods and the overall architecture - and choose the best one.
+When in doubt, ask for a review of your design approach before implementing it.
+
 ### Build and Test Commands
 - **Run tests**: `uv run pytest --cov --tb=short`
 - **Code formatting and linting**: `uvx prek run --all-files` formats and lints all files
@@ -114,11 +122,88 @@ uv run laufband status --db test.sqlite --lock test.lock
 - **Error handling**: Tasks can fail gracefully, use `.close()` method for clean exits
 
 ### Testing Strategy
-- Tests use temporary directories (`tmp_path` fixture)
-- Database files are created with full paths in test temp directories
-- Mock scenarios test various failure modes and recovery patterns
-- Tests marked with `@pytest.mark.human_reviewed` should not be modified by automated tools
+
+The test suite is organized into **unit tests** and **integration tests** for optimal speed and maintainability:
+
+#### Test Organization
+- `tests/unit/` - Fast unit tests (<1s total, 32 tests)
+  - No multiprocessing, no time delays, no file I/O
+  - Test business logic, models, and algorithms in isolation
+  - Use `MockTimeProvider` for instant time control
+  - Use factories for consistent test data
+
+- `tests/integration/` - Integration tests (34 tests)
+  - Real multiprocessing, database I/O, and coordination
+  - Test end-to-end scenarios and worker interactions
+  - Use polling helpers to avoid fixed sleep delays
+  - Tests marked with `@pytest.mark.human_reviewed` should not be modified by automated tools
+
+#### Test Fixtures and Factories
+Available in `tests/conftest.py`:
+- `mock_time` - Controllable time provider (advance time instantly)
+- `db_engine` / `db_session` - In-memory database with auto-rollback
+- `workflow_factory` - Create test workflows
+- `worker_factory` - Create test workers (auto-incremented IDs)
+- `task_factory` - Create test tasks with various states
+- `wait_for_condition` - Polling utility for async operations
+- `db_wait_helpers` - Wait for database state changes
+
+#### Running Tests
+```bash
+# Fast unit tests only (development)
+uv run pytest -m unit  # ~0.3s
+
+# All tests with coverage (pre-commit)
+uv run pytest --cov --tb=short  # ~50s
+
+# Integration tests only
+uv run pytest tests/integration/  # ~50s
+
+# Specific test file
+uv run pytest tests/unit/test_heartbeat_logic.py -v
+```
+
+#### Writing New Tests
+
+**For Business Logic (Unit Tests)**:
+```python
+@pytest.mark.unit
+def test_heartbeat_expiration(worker_factory, mock_time):
+    """Test heartbeat logic without real delays."""
+    worker = worker_factory(heartbeat_timeout=5)
+
+    # Instantly advance time
+    mock_time.advance(6)
+
+    # Test logic
+    assert worker.is_heartbeat_expired(mock_time)
+```
+
+**For End-to-End Scenarios (Integration Tests)**:
+```python
+@pytest.mark.integration
+def test_worker_coordination(tmp_path, db_wait_helpers):
+    """Test real multiprocessing scenario."""
+    proc = multiprocessing.Process(...)
+    proc.start()
+
+    # Use polling instead of sleep
+    db_wait_helpers.wait_for_task_count(expected=5, timeout=3)
+
+    proc.join()
+```
+
+#### Testable Business Logic
+Core logic extracted to `laufband/worker_logic.py`:
+- `check_and_mark_expired_workers()` - Heartbeat monitoring
+- `should_retry_task()` - Retry policy decisions
+- `update_worker_heartbeat()` - Heartbeat updates
+- `create_worker_entry()` - Worker initialization
+
+These functions accept `TimeProvider` for controllable time in tests.
 
 ### Troubleshooting
 - If import errors occur, ensure running with `uv run` prefix
 - If tests timeout, ensure proper timeout settings (2+ minutes for test suite)
+- Unit test failures: Check factory usage and mock_time advancement
+- Integration test failures: Check for race conditions, use polling helpers
@@ -14,6 +14,8 @@
 )
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
+from laufband.time_provider import RealTimeProvider, TimeProvider
+
 # from sqlalchemy.orm import MappedAsDataclass
 
 
@@ -86,8 +88,18 @@ class WorkerEntry(Base):
     @property
     def heartbeat_expired(self) -> bool:
         """Check if the worker's heartbeat is expired."""
+        return self.is_heartbeat_expired()
+
+    def is_heartbeat_expired(self, time_provider: TimeProvider | None = None) -> bool:
+        """Check if the worker's heartbeat is expired.
+
+        Args:
+            time_provider: Optional time provider for testing. Defaults to real time.
+        """
+        if time_provider is None:
+            time_provider = RealTimeProvider()
         return (
-            datetime.now() - self.last_heartbeat
+            time_provider.now() - self.last_heartbeat
         ).total_seconds() > self.heartbeat_timeout
 
     @property
@@ -99,9 +111,20 @@ def running_tasks(self) -> set["TaskEntry"]:
 
     @property
     def runtime(self) -> timedelta:
+        """Calculate worker runtime."""
+        return self.calculate_runtime()
+
+    def calculate_runtime(self, time_provider: TimeProvider | None = None) -> timedelta:
+        """Calculate worker runtime.
+
+        Args:
+            time_provider: Optional time provider for testing. Defaults to real time.
+        """
+        if time_provider is None:
+            time_provider = RealTimeProvider()
         if self.status in [WorkerStatus.OFFLINE, WorkerStatus.KILLED]:
             return self.last_heartbeat - self.started_at
-        return datetime.now() - self.started_at
+        return time_provider.now() - self.started_at
 
 
 # --- TaskStatusEntry ---
 
@@ -1,15 +1,13 @@
 import threading
-from datetime import datetime
 
 from flufl.lock import Lock, LockState
 from sqlalchemy import create_engine
-from sqlalchemy.orm import selectinload, sessionmaker
+from sqlalchemy.orm import sessionmaker
 
-from laufband.db import (
-    TaskStatusEntry,
-    TaskStatusEnum,
-    WorkerEntry,
-    WorkerStatus,
+from laufband.time_provider import RealTimeProvider
+from laufband.worker_logic import (
+    check_and_mark_expired_workers,
+    update_worker_heartbeat,
 )
 
 
@@ -22,15 +20,12 @@ def heartbeat(
 ):
     engine = create_engine(db, echo=False)
     Session = sessionmaker(bind=engine)  # noqa: N806
+    time_provider = RealTimeProvider()
 
     with db_lock:
         with Session() as session:
-            worker = session.get(WorkerEntry, identifier)
-            if worker is None:
-                raise ValueError(f"Worker with identifier {identifier} not found.")
-            worker.last_heartbeat = datetime.now()
+            worker = update_worker_heartbeat(session, identifier, time_provider)
             heartbeat_interval = worker.heartbeat_interval
-            session.add(worker)
             session.commit()
 
     while not stop_event.wait(heartbeat_interval):
@@ -41,34 +36,16 @@ def heartbeat(
             user_file_lock.refresh(int(heartbeat_interval * 1.5))
         with db_lock:
             with Session() as session:
-                worker = session.get(WorkerEntry, identifier)
-                if worker is None:
-                    raise ValueError(f"Worker with identifier {identifier} not found.")
-                worker.last_heartbeat = datetime.now()
-                session.add(worker)
-                # check expired heartbeats
+                worker = update_worker_heartbeat(session, identifier, time_provider)
                 workflow_id = worker.workflow_id
-                for w in (
-                    session.query(WorkerEntry)
-                    .options(selectinload(WorkerEntry.task_statuses))
-                    .filter(
-                        WorkerEntry.workflow_id == workflow_id,
-                        WorkerEntry.status.in_([WorkerStatus.BUSY, WorkerStatus.IDLE]),
-                    )
-                    .all()
-                ):
-                    if w.heartbeat_expired:
-                        w.status = WorkerStatus.KILLED
-                        for task in w.running_tasks:
-                            task_status = TaskStatusEntry(
-                                status=TaskStatusEnum.KILLED, worker=w, task=task
-                            )
-                            session.add(task_status)
-                        session.add(w)
+                # Check and mark expired workers
+                check_and_mark_expired_workers(session, workflow_id, time_provider)
                 session.commit()
 
     with db_lock:
         with Session() as session:
+            from laufband.db import WorkerEntry, WorkerStatus
+
             worker = session.get(WorkerEntry, identifier)
             if worker is not None:
                 worker.status = WorkerStatus.OFFLINE
 
@@ -0,0 +1,51 @@
+"""Time abstraction for testability."""
+
+from abc import ABC, abstractmethod
+from datetime import datetime, timedelta
+
+
+class TimeProvider(ABC):
+    """Abstract time provider for dependency injection."""
+
+    @abstractmethod
+    def now(self) -> datetime:
+        """Get current datetime."""
+        pass
+
+    @abstractmethod
+    def sleep(self, seconds: float) -> None:
+        """Sleep for given seconds."""
+        pass
+
+
+class RealTimeProvider(TimeProvider):
+    """Production time provider using real system time."""
+
+    def now(self) -> datetime:
+        return datetime.now()
+
+    def sleep(self, seconds: float) -> None:
+        import time
+
+        time.sleep(seconds)
+
+
+class MockTimeProvider(TimeProvider):
+    """Controllable time provider for testing.
+
+    Allows manual time advancement without actual waiting.
+    """
+
+    def __init__(self, start_time: datetime | None = None):
+        self._current_time = start_time or datetime(2024, 1, 1, 12, 0, 0)
+
+    def now(self) -> datetime:
+        return self._current_time
+
+    def advance(self, seconds: float) -> None:
+        """Manually advance time by given seconds."""
+        self._current_time += timedelta(seconds=seconds)
+
+    def sleep(self, seconds: float) -> None:
+        """Mock sleep that just advances time instantly."""
+        self.advance(seconds)