diff --git a/tests/test_task_run_signals.py b/tests/test_task_run_signals.py
new file mode 100644
index 0000000..56973cf
--- /dev/null
+++ b/tests/test_task_run_signals.py
@@ -0,0 +1,247 @@
+"""FC-3i: Celery signal → task_run row tests.
+
+Uses task_always_eager so signals fire synchronously in-process during
+the test. Scoped autouse fixture keeps the eager mode confined to this
+file (other tests may break with eager mode enabled globally).
+
+Test assertions on signal-mutated rows use COLUMN SELECTS / db_sync.
+scalar (per reference_async_coredml_test_assertions). ORM attribute
+access on a row a signal handler just mutated risks MissingGreenlet
+across async/sync boundaries.
+"""
+
+import pytest
+from sqlalchemy import func, select
+
+from backend.app.celery_app import celery
+from backend.app.models import TaskRun
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture(autouse=True)
+def _eager_celery(monkeypatch):
+    """task_always_eager so signals fire in-process during the test."""
+    monkeypatch.setattr(celery.conf, "task_always_eager", True)
+    monkeypatch.setattr(celery.conf, "task_eager_propagates", False)
+
+
+@pytest.mark.asyncio
+async def test_prerun_inserts_running_row(db_sync):
+    """Fire a no-op task; assert task_run row inserted with running status."""
+    from backend.app.tasks.thumbnail import generate_thumbnail
+    generate_thumbnail.delay(999_999)  # nonexistent image_id → no-op
+
+    rows = db_sync.execute(
+        select(TaskRun.queue, TaskRun.target_id, TaskRun.task_name)
+        .where(TaskRun.target_id == 999_999)
+    ).all()
+    assert len(rows) == 1
+    queue, target_id, task_name = rows[0]
+    assert queue == "thumbnail"
+    assert target_id == 999_999
+    assert task_name.endswith(".generate_thumbnail")
+
+
+@pytest.mark.asyncio
+async def test_postrun_marks_ok_with_duration(db_sync):
+    from backend.app.tasks.thumbnail import generate_thumbnail
+    generate_thumbnail.delay(999_998)
+
+    row = db_sync.execute(
+        select(
+            TaskRun.status, TaskRun.finished_at, TaskRun.duration_ms,
+        ).where(TaskRun.target_id == 999_998)
+    ).one()
+    status, finished_at, duration_ms = row
+    assert status == "ok"
+    assert finished_at is not None
+    assert duration_ms is not None and duration_ms >= 0
+
+
+@pytest.mark.asyncio
+async def test_failure_records_error_type_and_message(db_sync, monkeypatch):
+    """Force a tracked task to raise; assert error fields populated."""
+    from backend.app.tasks.thumbnail import generate_thumbnail
+
+    def _explode(*a, **k):
+        raise ValueError("synthetic failure")
+    monkeypatch.setattr(
+        "backend.app.tasks.thumbnail.generate_thumbnail.run",
+        _explode,
+    )
+    with pytest.raises(ValueError):
+        generate_thumbnail.delay(999_997).get(propagate=True)
+
+    row = db_sync.execute(
+        select(
+            TaskRun.status, TaskRun.error_type, TaskRun.error_message,
+            TaskRun.finished_at,
+        ).where(TaskRun.target_id == 999_997)
+        .order_by(TaskRun.id.desc()).limit(1)
+    ).one()
+    status, error_type, error_message, finished_at = row
+    assert status == "error"
+    assert error_type == "ValueError"
+    assert "synthetic failure" in (error_message or "")
+    assert finished_at is not None
+
+
+@pytest.mark.asyncio
+async def test_soft_time_limit_records_timeout_status(db_sync, monkeypatch):
+    from celery.exceptions import SoftTimeLimitExceeded
+
+    from backend.app.tasks.thumbnail import generate_thumbnail
+
+    def _timeout(*a, **k):
+        raise SoftTimeLimitExceeded("simulated soft limit")
+    monkeypatch.setattr(
+        "backend.app.tasks.thumbnail.generate_thumbnail.run",
+        _timeout,
+    )
+    with pytest.raises(SoftTimeLimitExceeded):
+        generate_thumbnail.delay(999_996).get(propagate=True)
+
+    status = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.target_id == 999_996)
+        .order_by(TaskRun.id.desc()).limit(1)
+    ).scalar_one()
+    assert status == "timeout"
+
+
+@pytest.mark.asyncio
+async def test_untracked_celery_internal_tasks_skip(db_sync):
+    """celery.chord_unlock is in the untracked list; firing it should
+    NOT insert a task_run row."""
+    before = db_sync.execute(select(func.count(TaskRun.id))).scalar_one()
+    # Send a chord_unlock-style task via apply (no real task to dispatch).
+    # The signal handler checks task.name against _UNTRACKED_TASK_NAMES;
+    # if name is in the list, no row is inserted. We verify by firing
+    # a regular task and asserting only ONE row appears, not two.
+    from backend.app.tasks.thumbnail import generate_thumbnail
+    generate_thumbnail.delay(999_995)
+    after = db_sync.execute(select(func.count(TaskRun.id))).scalar_one()
+    assert after - before == 1
+
+
+@pytest.mark.asyncio
+async def test_args_summary_truncated_to_255(db_sync, monkeypatch):
+    """Task with a very large repr(args); assert args_summary <= 255."""
+    from backend.app.tasks.thumbnail import generate_thumbnail
+
+    # Force a giant arg by patching repr(args) target — actually just
+    # call with a large int that produces a long repr.
+    big_id = int("9" * 200)  # 200-char repr
+    # The signal handler reads repr(args), so this naturally produces
+    # a long args_summary that should get truncated to 255.
+    try:
+        generate_thumbnail.delay(big_id).get(timeout=5, propagate=False)
+    except Exception:
+        pass
+
+    args_summary = db_sync.execute(
+        select(TaskRun.args_summary)
+        .where(TaskRun.task_name.endswith(".generate_thumbnail"))
+        .order_by(TaskRun.id.desc()).limit(1)
+    ).scalar_one()
+    assert args_summary is None or len(args_summary) <= 255
+
+
+@pytest.mark.asyncio
+async def test_error_message_truncated_to_2000(db_sync, monkeypatch):
+    from backend.app.tasks.thumbnail import generate_thumbnail
+
+    huge_message = "X" * 5000
+
+    def _raise_huge(*a, **k):
+        raise RuntimeError(huge_message)
+    monkeypatch.setattr(
+        "backend.app.tasks.thumbnail.generate_thumbnail.run", _raise_huge,
+    )
+    with pytest.raises(RuntimeError):
+        generate_thumbnail.delay(999_994).get(propagate=True)
+
+    error_message = db_sync.execute(
+        select(TaskRun.error_message).where(TaskRun.target_id == 999_994)
+        .order_by(TaskRun.id.desc()).limit(1)
+    ).scalar_one()
+    assert error_message is not None
+    assert len(error_message) <= 2000
+
+
+@pytest.mark.asyncio
+async def test_signal_handler_failure_does_not_break_task(db_sync, monkeypatch):
+    """Monkeypatch sync_session_factory in the signals module to raise;
+    assert the task still returns its result and the worker doesn't
+    propagate the signal-handler exception."""
+    from backend.app.tasks.thumbnail import generate_thumbnail
+
+    def _broken_factory():
+        raise RuntimeError("db down")
+
+    monkeypatch.setattr(
+        "backend.app.celery_signals.sync_session_factory", _broken_factory,
+    )
+
+    # Task should still execute (no-op on missing image) without raising.
+    result = generate_thumbnail.delay(999_993).get(timeout=5)
+    assert result is not None  # task ran; signal handler errors swallowed
+
+
+@pytest.mark.parametrize("args,expected", [
+    ((42,), 42),
+    ((42, "ignored"), 42),
+    (("123",), 123),
+    (("not-int",), None),
+    ((), None),
+    (None, None),
+])
+def test_target_id_extracted_from_first_int_arg(args, expected):
+    """Unit test the helper directly; no DB needed."""
+    from backend.app.celery_signals import _target_id_from_args
+    assert _target_id_from_args(args) == expected
+
+
+@pytest.mark.asyncio
+async def test_retry_signal_records_retry_status(db_sync, monkeypatch):
+    """task_retry should mark the running row as 'retry' status."""
+    from backend.app.tasks.thumbnail import generate_thumbnail
+
+    call_count = [0]
+
+    def _retry_then_succeed(self_, *a, **k):
+        call_count[0] += 1
+        if call_count[0] == 1:
+            raise self_.retry(exc=RuntimeError("first attempt"), countdown=0)
+        return {"status": "ok"}
+
+    monkeypatch.setattr(
+        "backend.app.tasks.thumbnail.generate_thumbnail.run",
+        _retry_then_succeed,
+    )
+    try:
+        generate_thumbnail.delay(999_992).get(timeout=10, propagate=False)
+    except Exception:
+        pass
+
+    # Expect multiple rows for the same target_id: one with status='retry'
+    # for the first attempt, one with status='ok' for the second.
+    statuses = db_sync.execute(
+        select(TaskRun.status).where(TaskRun.target_id == 999_992)
+        .order_by(TaskRun.id.asc())
+    ).scalars().all()
+    assert "retry" in statuses
+
+
+@pytest.mark.asyncio
+async def test_worker_hostname_recorded(db_sync):
+    """Sender's hostname should land in worker_hostname."""
+    from backend.app.tasks.thumbnail import generate_thumbnail
+    generate_thumbnail.delay(999_991)
+
+    hostname = db_sync.execute(
+        select(TaskRun.worker_hostname).where(TaskRun.target_id == 999_991)
+    ).scalar_one()
+    # Under eager mode hostname may be None or "celery@local"; just
+    # assert the column is touched (either string or None — never throws).
+    assert hostname is None or isinstance(hostname, str)