automatyzacje-comit5

pytorrent/config.py

@@ -71,3 +71,4 @@ JOBS_RETENTION_DAYS = _env_int("PYTORRENT_JOBS_RETENTION_DAYS", 30, 1)
 SMART_QUEUE_HISTORY_RETENTION_DAYS = _env_int("PYTORRENT_SMART_QUEUE_HISTORY_RETENTION_DAYS", 30, 1)
 LOG_RETENTION_DAYS = _env_int("PYTORRENT_LOG_RETENTION_DAYS", 30, 1)
 SMART_QUEUE_LABEL = os.getenv("PYTORRENT_SMART_QUEUE_LABEL", "Smart Queue Paused")
+SMART_QUEUE_STALLED_LABEL = os.getenv("PYTORRENT_SMART_QUEUE_STALLED_LABEL", "Stalled")

pytorrent/db.py

@@ -139,6 +139,7 @@ CREATE TABLE IF NOT EXISTS smart_queue_settings (
   stalled_seconds INTEGER DEFAULT 300,
   min_speed_bytes INTEGER DEFAULT 1024,
   min_seeds INTEGER DEFAULT 1,
+  min_peers INTEGER DEFAULT 0,
   manage_stopped INTEGER DEFAULT 0,
   updated_at TEXT NOT NULL,
   PRIMARY KEY(user_id, profile_id)
@@ -280,6 +281,7 @@ MIGRATIONS = [
     "ALTER TABLE rtorrent_config_overrides ADD COLUMN baseline_value TEXT",
     "ALTER TABLE torrent_stats_cache ADD COLUMN updated_epoch REAL DEFAULT 0",
     "ALTER TABLE smart_queue_settings ADD COLUMN manage_stopped INTEGER DEFAULT 0",
+    "ALTER TABLE smart_queue_settings ADD COLUMN min_peers INTEGER DEFAULT 0",
 ]
 
 

pytorrent/routes/api.py

@@ -326,9 +326,11 @@ def cleanup_summary() -> dict:
         "jobs_total": _table_count("jobs"),
         "jobs_clearable": _table_count("jobs", "WHERE status NOT IN ('pending', 'running')"),
         "smart_queue_history_total": _table_count("smart_queue_history"),
+        "automation_history_total": _table_count("automation_history"),
         "retention_days": {
             "jobs": JOBS_RETENTION_DAYS,
             "smart_queue_history": SMART_QUEUE_HISTORY_RETENTION_DAYS,
+            "automation_history": SMART_QUEUE_HISTORY_RETENTION_DAYS,
         },
         "database": _db_size(),
     }
@@ -731,6 +733,19 @@ def cleanup_smart_queue():
     return ok({"deleted": deleted, "cleanup": cleanup_summary()})
 
 
+@bp.post("/cleanup/automations")
+def cleanup_automations():
+    with connect() as conn:
+        exists = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='automation_history'").fetchone()
+        if not exists:
+            deleted = 0
+        else:
+            # Note: Cleanup panel removes only automation logs, not saved automation rules.
+            cur = conn.execute("DELETE FROM automation_history")
+            deleted = int(cur.rowcount or 0)
+    return ok({"deleted": deleted, "cleanup": cleanup_summary()})
+
+
 @bp.post("/cleanup/all")
 def cleanup_all():
     deleted_jobs = clear_jobs()
@@ -741,7 +756,13 @@ def cleanup_all():
         else:
             cur = conn.execute("DELETE FROM smart_queue_history")
             deleted_smart = int(cur.rowcount or 0)
-    return ok({"deleted": {"jobs": deleted_jobs, "smart_queue_history": deleted_smart}, "cleanup": cleanup_summary()})
+        exists_auto = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='automation_history'").fetchone()
+        if not exists_auto:
+            deleted_auto = 0
+        else:
+            cur = conn.execute("DELETE FROM automation_history")
+            deleted_auto = int(cur.rowcount or 0)
+    return ok({"deleted": {"jobs": deleted_jobs, "smart_queue_history": deleted_smart, "automation_history": deleted_auto}, "cleanup": cleanup_summary()})
 
 
 @bp.post("/jobs/<job_id>/cancel")
@@ -1062,3 +1083,17 @@ def automations_check():
         return ok({'result': automation_rules.check(profile, force=True), 'history': automation_rules.list_history(profile['id'])})
     except Exception as exc:
         return jsonify({'ok': False, 'error': str(exc)}), 500
+
+
+@bp.delete('/automations/history')
+def automations_history_clear():
+    from ..services import automation_rules
+    profile = preferences.active_profile()
+    if not profile:
+        return jsonify({'ok': False, 'error': 'No profile'}), 400
+    try:
+        # Note: Clear only automation execution logs; rules and cooldown state stay unchanged.
+        deleted = automation_rules.clear_history(profile['id'])
+        return ok({'deleted': deleted, 'history': automation_rules.list_history(profile['id']), 'cleanup': cleanup_summary()})
+    except Exception as exc:
+        return jsonify({'ok': False, 'error': str(exc)}), 500

pytorrent/routes/main.py

@@ -106,7 +106,7 @@ def openapi():
         "/api/rss/feeds": {"post": {"summary": "Add RSS feed", "requestBody": {"content": {"application/json": {"schema": {"type": "object"}}}}, "responses": {"200": {"description": "RSS config"}}}},
         "/api/rss/rules": {"post": {"summary": "Add RSS rule", "requestBody": {"content": {"application/json": {"schema": {"type": "object"}}}}, "responses": {"200": {"description": "RSS config"}}}},
         "/api/rss/check": {"post": {"summary": "Manually check RSS feeds", "responses": {"200": {"description": "Queued matches"}}}},
-        "/api/smart-queue": {"get": {"summary": "Get Smart Queue settings, exceptions and history", "parameters": [{"name": "history_limit", "in": "query", "schema": {"type": "integer", "default": 10, "minimum": 1, "maximum": 100}, "description": "Number of Smart Queue history rows to return"}], "responses": {"200": {"description": "Smart Queue config with history and history_total"}}}, "post": {"summary": "Save Smart Queue settings", "requestBody": {"content": {"application/json": {"schema": {"type": "object", "properties": {"enabled": {"type": "boolean"}, "max_active_downloads": {"type": "integer"}, "stalled_seconds": {"type": "integer"}, "min_speed_bytes": {"type": "integer"}, "min_seeds": {"type": "integer"}}}}}}, "responses": {"200": {"description": "Saved"}}}},
+        "/api/smart-queue": {"get": {"summary": "Get Smart Queue settings, exceptions and history", "parameters": [{"name": "history_limit", "in": "query", "schema": {"type": "integer", "default": 10, "minimum": 1, "maximum": 100}, "description": "Number of Smart Queue history rows to return"}], "responses": {"200": {"description": "Smart Queue config with history and history_total"}}}, "post": {"summary": "Save Smart Queue settings", "requestBody": {"content": {"application/json": {"schema": {"type": "object", "properties": {"enabled": {"type": "boolean"}, "max_active_downloads": {"type": "integer"}, "stalled_seconds": {"type": "integer"}, "min_speed_bytes": {"type": "integer"}, "min_seeds": {"type": "integer"}, "min_peers": {"type": "integer"}}}}}}, "responses": {"200": {"description": "Saved"}}}},
         "/api/smart-queue/check": {"post": {"summary": "Run Smart Queue immediately", "responses": {"200": {"description": "Smart Queue action result"}}}},
         "/api/smart-queue/exclusion": {"post": {"summary": "Add or remove a torrent Smart Queue exception", "requestBody": {"content": {"application/json": {"schema": {"type": "object", "properties": {"hash": {"type": "string"}, "excluded": {"type": "boolean"}, "reason": {"type": "string"}}}}}}, "responses": {"200": {"description": "Exception list"}}}},
         "/api/traffic/history": {"get": {"summary": "Transfer history for charts", "parameters": [{"name": "range", "in": "query", "schema": {"type": "string", "enum": ["15m", "1h", "3h", "6h", "24h", "7d", "30d", "90d"]}}], "responses": {"200": {"description": "Aggregated traffic history"}}}}

pytorrent/services/automation_rules.py

@@ -5,6 +5,10 @@ import json
 from ..db import connect, default_user_id, utcnow
 from . import rtorrent
 from .preferences import active_profile
+from .workers import enqueue
+
+AUTOMATION_JOB_CHUNK_SIZE = 100
+
 
 
 def _loads(value: str | None, default: Any) -> Any:
@@ -94,6 +98,14 @@ def list_history(profile_id: int, user_id: int | None = None, limit: int = 30) -
         return conn.execute('SELECT * FROM automation_history WHERE user_id=? AND profile_id=? ORDER BY created_at DESC LIMIT ?', (user_id, profile_id, max(1, min(int(limit or 30), 100)))).fetchall()
 
 
+def clear_history(profile_id: int, user_id: int | None = None) -> int:
+    user_id = user_id or default_user_id()
+    with connect() as conn:
+        # Note: Manual automation log cleanup is scoped to the active profile and current user.
+        cur = conn.execute('DELETE FROM automation_history WHERE user_id=? AND profile_id=?', (user_id, profile_id))
+        return int(cur.rowcount or 0)
+
+
 def _condition_true(t: dict[str, Any], cond: dict[str, Any]) -> bool:
     typ = str(cond.get('type') or '')
     if typ == 'completed': return bool(int(t.get('complete') or 0))
@@ -111,8 +123,11 @@ def _conditions_match(conn, rule: dict[str, Any], profile_id: int, t: dict[str,
     if not h: return False
     immediate_ok = True; delayed_ok = True; now = utcnow(); now_ts = _now_ts()
     for cond in rule.get('conditions') or []:
-        ok = _condition_true(t, cond)
-        if cond.get('type') == 'no_seeds' and int(cond.get('minutes') or 0) > 0:
+        raw_ok = _condition_true(t, cond)
+        negated = bool(cond.get('negate'))
+        # Note: Negation is applied in the backend, so UI and API only store the condition flag.
+        ok = (not raw_ok) if negated else raw_ok
+        if cond.get('type') == 'no_seeds' and int(cond.get('minutes') or 0) > 0 and not negated:
             row = conn.execute('SELECT condition_since_at FROM automation_rule_state WHERE rule_id=? AND profile_id=? AND torrent_hash=?', (rule['id'], profile_id, h)).fetchone()
             if ok:
                 since = row['condition_since_at'] if row and row.get('condition_since_at') else now
@@ -125,33 +140,143 @@ def _conditions_match(conn, rule: dict[str, Any], profile_id: int, t: dict[str,
     return immediate_ok and delayed_ok
 
 
-def _cooldown_ok(conn, rule: dict[str, Any], profile_id: int, torrent_hash: str) -> bool:
+def _cooldown_ok(conn, rule: dict[str, Any], profile_id: int, torrent_hash: str = '__rule__') -> bool:
     cooldown = int(rule.get('cooldown_minutes') or 0)
+    if cooldown <= 0: return True
     row = conn.execute('SELECT last_applied_at FROM automation_rule_state WHERE rule_id=? AND profile_id=? AND torrent_hash=?', (rule['id'], profile_id, torrent_hash)).fetchone()
     if not row or not row.get('last_applied_at'): return True
     return _now_ts() - _ts(row['last_applied_at']) >= cooldown * 60
 
 
-def _apply_effects(c: Any, profile: dict[str, Any], torrent: dict[str, Any], effects: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    h = str(torrent.get('hash') or ''); labels = _label_names(torrent.get('label')); applied = []
+def _mark_rule_cooldown(conn, rule: dict[str, Any], profile_id: int, now: str) -> None:
+    # Note: Cooldown is rule-level, so one batch execution blocks the whole automation until the cooldown expires.
+    conn.execute('INSERT INTO automation_rule_state(rule_id,profile_id,torrent_hash,last_applied_at,updated_at) VALUES(?,?,?,?,?) ON CONFLICT(rule_id,profile_id,torrent_hash) DO UPDATE SET last_applied_at=excluded.last_applied_at, updated_at=excluded.updated_at', (rule['id'], profile_id, '__rule__', now, now))
+
+
+def _chunk_hashes(hashes: list[str], size: int = AUTOMATION_JOB_CHUNK_SIZE) -> list[list[str]]:
+    # Note: Automation jobs use the same small-batch idea as manual bulk jobs, so long move/remove/actions remain visible and recoverable.
+    safe_size = max(1, int(size or AUTOMATION_JOB_CHUNK_SIZE))
+    return [hashes[index:index + safe_size] for index in range(0, len(hashes), safe_size)]
+
+
+def _job_context(rule: dict[str, Any], eff_type: str, hashes: list[str], torrents_by_hash: dict[str, dict[str, Any]], extra: dict[str, Any] | None = None) -> dict[str, Any]:
+    # Note: Job context marks jobs created by automations, making the Jobs log explain what rule queued the work.
+    ctx = {
+        'source': 'automation',
+        'rule_id': rule.get('id'),
+        'rule_name': str(rule.get('name') or ''),
+        'effect': eff_type,
+        'bulk': len(hashes) > 1,
+        'hash_count': len(hashes),
+        'requested_at': utcnow(),
+        'items': [
+            {
+                'hash': h,
+                'name': str((torrents_by_hash.get(h) or {}).get('name') or ''),
+                'path': str((torrents_by_hash.get(h) or {}).get('path') or ''),
+            }
+            for h in hashes
+        ],
+    }
+    if extra:
+        ctx.update(extra)
+    return ctx
+
+
+def _enqueue_automation_job(profile: dict[str, Any], rule: dict[str, Any], action_name: str, hashes: list[str], payload: dict[str, Any], torrents_by_hash: dict[str, dict[str, Any]], user_id: int | None = None, context_extra: dict[str, Any] | None = None) -> list[str]:
+    # Note: Every automation side effect is queued as a normal job instead of running inline, so it appears in Jobs and uses worker retries/ordering.
+    job_ids: list[str] = []
+    chunks = _chunk_hashes(hashes)
+    for index, chunk in enumerate(chunks, start=1):
+        part_payload = dict(payload or {})
+        part_payload['hashes'] = chunk
+        part_payload['automation_ordered'] = True
+        extra = dict(context_extra or {})
+        if len(chunks) > 1:
+            extra.update({'bulk_label': f'automation-{index}', 'bulk_part': index, 'bulk_parts': len(chunks), 'parent_hash_count': len(hashes)})
+        if action_name == 'move':
+            extra.update({'target_path': str(part_payload.get('path') or ''), 'move_data': bool(part_payload.get('move_data'))})
+        if action_name == 'remove':
+            extra.update({'remove_data': bool(part_payload.get('remove_data'))})
+        part_payload['job_context'] = _job_context(rule, str(context_extra.get('effect_type') if context_extra else action_name), chunk, torrents_by_hash, extra)
+        job_ids.append(enqueue(action_name, int(profile['id']), part_payload, user_id=user_id))
+    return job_ids
+
+
+def _apply_effects_bulk(c: Any, profile: dict[str, Any], torrents: list[dict[str, Any]], effects: list[dict[str, Any]], rule: dict[str, Any], user_id: int | None = None) -> list[dict[str, Any]]:
+    hashes = [str(t.get('hash') or '') for t in torrents if str(t.get('hash') or '')]
+    torrents_by_hash = {str(t.get('hash') or ''): t for t in torrents if str(t.get('hash') or '')}
+    labels_by_hash = {str(t.get('hash') or ''): _label_names(t.get('label')) for t in torrents}
+    applied: list[dict[str, Any]] = []
+    if not hashes: return applied
     for eff in effects:
         typ = str(eff.get('type') or '')
         if typ == 'move':
-            # Note: Automation move-to-path now uses the same move implementation as the main app action.
             path = str(eff.get('path') or '').strip() or rtorrent.default_download_path(profile)
-            move_payload = {'path': path, 'move_data': bool(eff.get('move_data')), 'recheck': bool(eff.get('recheck', eff.get('move_data'))), 'keep_seeding': bool(eff.get('keep_seeding'))}
-            result = rtorrent.move_torrents(profile, [h], move_payload) if path else None
-            if path: applied.append({'type': 'move', 'path': path, 'move_data': bool(eff.get('move_data')), 'recheck': bool(move_payload['recheck']), 'keep_seeding': bool(eff.get('keep_seeding')), 'result': result})
+            payload = {
+                'path': path,
+                'move_data': bool(eff.get('move_data')),
+                'recheck': bool(eff.get('recheck', eff.get('move_data'))),
+                'keep_seeding': bool(eff.get('keep_seeding')),
+            }
+            job_ids = _enqueue_automation_job(profile, rule, 'move', hashes, payload, torrents_by_hash, user_id, {'effect_type': 'move'})
+            applied.append({'type': 'move', 'path': path, 'count': len(hashes), 'target_hashes': hashes, 'move_data': payload['move_data'], 'recheck': payload['recheck'], 'keep_seeding': payload['keep_seeding'], 'job_ids': job_ids})
         elif typ == 'add_label':
             label = str(eff.get('label') or '').strip()
-            if label and label not in labels: labels.append(label); c.call('d.custom1.set', h, _label_value(labels))
-            if label: applied.append({'type': 'add_label', 'label': label})
+            if label:
+                # Note: Add-label automations are idempotent and queue only torrents that need a changed label value.
+                grouped: dict[str, list[str]] = {}
+                for h in hashes:
+                    labels = labels_by_hash.get(h, [])
+                    if label in labels:
+                        continue
+                    new_labels = list(labels) + [label]
+                    value = _label_value(new_labels)
+                    labels_by_hash[h] = _label_names(value)
+                    grouped.setdefault(value, []).append(h)
+                target_hashes = [h for group in grouped.values() for h in group]
+                job_ids: list[str] = []
+                for value, group_hashes in grouped.items():
+                    job_ids.extend(_enqueue_automation_job(profile, rule, 'set_label', group_hashes, {'label': value}, torrents_by_hash, user_id, {'effect_type': 'add_label', 'label': label}))
+                if target_hashes:
+                    applied.append({'type': 'add_label', 'label': label, 'count': len(target_hashes), 'target_hashes': target_hashes, 'job_ids': job_ids})
         elif typ == 'remove_label':
-            label = str(eff.get('label') or '').strip(); labels = [x for x in labels if x != label]; c.call('d.custom1.set', h, _label_value(labels)); applied.append({'type': 'remove_label', 'label': label})
+            label = str(eff.get('label') or '').strip()
+            if label:
+                # Note: Remove-label automations are queued only for torrents where the requested label exists.
+                grouped: dict[str, list[str]] = {}
+                for h in hashes:
+                    labels = labels_by_hash.get(h, [])
+                    if label not in labels:
+                        continue
+                    value = _label_value([x for x in labels if x != label])
+                    labels_by_hash[h] = _label_names(value)
+                    grouped.setdefault(value, []).append(h)
+                target_hashes = [h for group in grouped.values() for h in group]
+                job_ids: list[str] = []
+                for value, group_hashes in grouped.items():
+                    job_ids.extend(_enqueue_automation_job(profile, rule, 'set_label', group_hashes, {'label': value}, torrents_by_hash, user_id, {'effect_type': 'remove_label', 'label': label}))
+                if target_hashes:
+                    applied.append({'type': 'remove_label', 'label': label, 'count': len(target_hashes), 'target_hashes': target_hashes, 'job_ids': job_ids})
         elif typ == 'set_labels':
-            value = _label_value(_label_names(eff.get('labels'))); c.call('d.custom1.set', h, value); labels = _label_names(value); applied.append({'type': 'set_labels', 'labels': value})
-        elif typ in {'pause', 'stop', 'start', 'resume', 'recheck'}:
-            method = {'pause':'d.pause','stop':'d.stop','start':'d.start','resume':'d.resume','recheck':'d.check_hash'}[typ]; c.call(method, h); applied.append({'type': typ})
+            value = _label_value(_label_names(eff.get('labels')))
+            target_labels = _label_names(value)
+            # Note: Set-labels queues a job only if the current labels differ from the requested exact list.
+            target_hashes = [h for h in hashes if labels_by_hash.get(h, []) != target_labels]
+            for h in target_hashes:
+                labels_by_hash[h] = list(target_labels)
+            if target_hashes:
+                job_ids = _enqueue_automation_job(profile, rule, 'set_label', target_hashes, {'label': value}, torrents_by_hash, user_id, {'effect_type': 'set_labels', 'labels': value})
+                applied.append({'type': 'set_labels', 'labels': value, 'count': len(target_hashes), 'target_hashes': target_hashes, 'job_ids': job_ids})
+        elif typ in {'pause', 'stop', 'start', 'resume', 'recheck', 'reannounce'}:
+            # Note: Runtime actions are queued as jobs too, so automation activity is visible in the Jobs panel.
+            job_ids = _enqueue_automation_job(profile, rule, typ, hashes, {}, torrents_by_hash, user_id, {'effect_type': typ})
+            applied.append({'type': typ, 'count': len(hashes), 'target_hashes': hashes, 'job_ids': job_ids})
+        elif typ == 'remove':
+            # Note: Remove is supported for automation payloads and still goes through ordered worker jobs.
+            payload = {'remove_data': bool(eff.get('remove_data'))}
+            job_ids = _enqueue_automation_job(profile, rule, 'remove', hashes, payload, torrents_by_hash, user_id, {'effect_type': 'remove'})
+            applied.append({'type': 'remove', 'count': len(hashes), 'target_hashes': hashes, 'remove_data': payload['remove_data'], 'job_ids': job_ids})
     return applied
 
 
@@ -160,17 +285,44 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
     if not profile: return {'ok': False, 'error': 'No active rTorrent profile'}
     user_id = user_id or default_user_id(); profile_id = int(profile['id'])
     rules = [r for r in list_rules(profile_id, user_id) if force or int(r.get('enabled') or 0)]
-    if not rules: return {'ok': True, 'checked': 0, 'applied': [], 'rules': 0}
-    torrents = rtorrent.list_torrents(profile); c = rtorrent.client_for(profile); applied = []; now = utcnow()
+    if not rules: return {'ok': True, 'checked': 0, 'applied': [], 'batches': [], 'rules': 0}
+    torrents = rtorrent.list_torrents(profile); applied = []; batches = []; now = utcnow()
+    planned: list[dict[str, Any]] = []
     with connect() as conn:
         for rule in rules:
-            for t in torrents:
-                h = str(t.get('hash') or '')
-                if not _conditions_match(conn, rule, profile_id, t): continue
-                if not force and not _cooldown_ok(conn, rule, profile_id, h): continue
-                try: actions = _apply_effects(c, profile, t, rule.get('effects') or [])
-                except Exception as exc: actions = [{'error': str(exc)}]
+            # Note: This pass only matches rules and updates condition timers; job creation is intentionally delayed until after this DB transaction commits.
+            if not force and not _cooldown_ok(conn, rule, profile_id):
+                continue
+            matched = [t for t in torrents if _conditions_match(conn, rule, profile_id, t)]
+            if not matched:
+                continue
+            hashes = [str(t.get('hash') or '') for t in matched if str(t.get('hash') or '')]
+            if hashes:
+                planned.append({'rule': rule, 'matched': matched, 'hashes': hashes})
+    for item in planned:
+        rule = item['rule']
+        matched = item['matched']
+        hashes = item['hashes']
+        # Note: Automation jobs are enqueued outside the rule-state transaction, preventing SQLite self-locks when enqueue() writes to jobs.
+        try:
+            actions = _apply_effects_bulk(None, profile, matched, rule.get('effects') or [], rule, user_id)
+        except Exception as exc:
+            actions = [{'error': str(exc), 'count': len(hashes), 'target_hashes': hashes}]
+        changed_hashes = sorted({h for a in actions for h in (a.get('target_hashes') or [])})
+        if not actions or not changed_hashes:
+            # Note: Matching torrents with no real action are not logged and do not restart the cooldown.
+            continue
+        history_actions = [{k: v for k, v in a.items() if k != 'target_hashes'} for a in actions]
+        matched_by_hash = {str(t.get('hash') or ''): t for t in matched}
+        with connect() as conn:
+            # Note: State/history writes happen after enqueue succeeds, so failed job creation does not create misleading automation history.
+            for h in changed_hashes:
+                t = matched_by_hash.get(h, {})
                 conn.execute('INSERT INTO automation_rule_state(rule_id,profile_id,torrent_hash,last_matched_at,last_applied_at,updated_at) VALUES(?,?,?,?,?,?) ON CONFLICT(rule_id,profile_id,torrent_hash) DO UPDATE SET last_matched_at=excluded.last_matched_at, last_applied_at=excluded.last_applied_at, updated_at=excluded.updated_at', (rule['id'], profile_id, h, now, now, now))
-                conn.execute('INSERT INTO automation_history(user_id,profile_id,rule_id,torrent_hash,torrent_name,rule_name,actions_json,created_at) VALUES(?,?,?,?,?,?,?,?)', (user_id, profile_id, rule['id'], h, str(t.get('name') or ''), str(rule.get('name') or ''), json.dumps(actions), now))
-                applied.append({'rule_id': rule['id'], 'rule_name': rule.get('name'), 'hash': h, 'name': t.get('name'), 'actions': actions})
-    return {'ok': True, 'checked': len(torrents), 'rules': len(rules), 'applied': applied}
+                applied.append({'rule_id': rule['id'], 'rule_name': rule.get('name'), 'hash': h, 'name': t.get('name'), 'actions': [{'type': a.get('type', 'error'), 'count': a.get('count', len(changed_hashes))} for a in actions]})
+            _mark_rule_cooldown(conn, rule, profile_id, now)
+            torrent_name = str(matched_by_hash.get(changed_hashes[0], {}).get('name') or '') if len(changed_hashes) == 1 else f'{len(changed_hashes)} torrents'
+            torrent_hash = changed_hashes[0] if len(changed_hashes) == 1 else f'batch:{rule["id"]}:{now}'
+            conn.execute('INSERT INTO automation_history(user_id,profile_id,rule_id,torrent_hash,torrent_name,rule_name,actions_json,created_at) VALUES(?,?,?,?,?,?,?,?)', (user_id, profile_id, rule['id'], torrent_hash, torrent_name, str(rule.get('name') or ''), json.dumps(history_actions), now))
+        batches.append({'rule_id': rule['id'], 'rule_name': rule.get('name'), 'count': len(changed_hashes), 'actions': history_actions})
+    return {'ok': True, 'checked': len(torrents), 'rules': len(rules), 'applied': applied, 'batches': batches}

pytorrent/services/retention.py

@@ -30,6 +30,8 @@ def cleanup(force: bool = False) -> dict[str, int]:
         targets = {
             "traffic_history": ("created_at", TRAFFIC_HISTORY_RETENTION_DAYS),
             "smart_queue_history": ("created_at", SMART_QUEUE_HISTORY_RETENTION_DAYS),
+            # Note: Automation history follows Smart Queue retention; rules and rule state are never deleted here.
+            "automation_history": ("created_at", SMART_QUEUE_HISTORY_RETENTION_DAYS),
             "jobs": ("updated_at", JOBS_RETENTION_DAYS),
             "logs": ("created_at", LOG_RETENTION_DAYS),
         }

pytorrent/services/rtorrent.py

@@ -1274,16 +1274,31 @@ def start_or_resume_hash(c: ScgiRtorrentClient, torrent_hash: str) -> dict:
     result['ok'] = result.get('ok', True)
     return result
 
-
-def move_torrents(profile: dict, torrent_hashes: list[str], payload: dict | None = None) -> dict:
-    # Note: Shared move implementation keeps API move and automation move-to-path identical.
+def action(profile: dict, torrent_hashes: list[str], name: str, payload: dict | None = None) -> dict:
     payload = payload or {}
     c = client_for(profile)
+    methods = {
+        "stop": "d.stop",
+        "recheck": "d.check_hash",
+        "reannounce": "d.tracker_announce",
+        "remove": "d.erase",
+    }
+    if name == "set_label":
+        label = str(payload.get("label") or "").strip()
+        for h in torrent_hashes:
+            c.call("d.custom1.set", h, label)
+        return {"ok": True, "count": len(torrent_hashes), "label": label}
+    if name == "set_ratio_group":
+        group = str(payload.get("ratio_group") or "").strip()
+        for h in torrent_hashes:
+            c.call("d.custom.set", h, "py_ratio_group", group)
+        return {"ok": True, "count": len(torrent_hashes), "ratio_group": group}
+    if name == "move":
         path = _remote_clean_path(payload.get("path") or "")
         move_data = bool(payload.get("move_data"))
         recheck = bool(payload.get("recheck", move_data))
         keep_seeding = bool(payload.get("keep_seeding"))
-    # Note: keep_seeding lets automation move completed data to another path and force the torrent back into seeding.
+        # Note: Automations can force seeding after a physical move even if the torrent was not active before.
         if not path:
             raise ValueError("Missing path")
         results = []
@@ -1329,40 +1344,11 @@ def move_torrents(profile: dict, torrent_hashes: list[str], payload: dict | None
                         c.call("d.start", h)
                         item["started_after_move"] = True
                     except Exception as exc:
-                    item["start_error"] = str(exc)
+                        item["start_after_move_error"] = str(exc)
             else:
                 c.call("d.directory.set", h, path)
-            if keep_seeding:
-                try:
-                    c.call("d.start", h)
-                    item["started_after_path_change"] = True
-                except Exception as exc:
-                    item["start_error"] = str(exc)
             results.append(item)
         return {"ok": True, "count": len(torrent_hashes), "move_data": move_data, "keep_seeding": keep_seeding, "results": results}
-
-def action(profile: dict, torrent_hashes: list[str], name: str, payload: dict | None = None) -> dict:
-    payload = payload or {}
-    c = client_for(profile)
-    methods = {
-        "stop": "d.stop",
-        "recheck": "d.check_hash",
-        "reannounce": "d.tracker_announce",
-        "remove": "d.erase",
-    }
-    if name == "set_label":
-        label = str(payload.get("label") or "").strip()
-        for h in torrent_hashes:
-            c.call("d.custom1.set", h, label)
-        return {"ok": True, "count": len(torrent_hashes), "label": label}
-    if name == "set_ratio_group":
-        group = str(payload.get("ratio_group") or "").strip()
-        for h in torrent_hashes:
-            c.call("d.custom.set", h, "py_ratio_group", group)
-        return {"ok": True, "count": len(torrent_hashes), "ratio_group": group}
-    if name == "move":
-        # Note: Main move delegates to the shared helper used by automations.
-        return move_torrents(profile, torrent_hashes, payload)
     if name == "pause":
         # Note: The app pause action is now a pure d.pause so later resume works without stop/start.
         results = [pause_hash(c, h) for h in torrent_hashes]

pytorrent/services/smart_queue.py

@@ -5,7 +5,7 @@ from typing import Any
 import json
 import time
 
-from ..config import SMART_QUEUE_LABEL
+from ..config import SMART_QUEUE_LABEL, SMART_QUEUE_STALLED_LABEL
 from ..db import connect, default_user_id, utcnow
 from . import rtorrent
 from .preferences import active_profile, get_profile
@@ -20,6 +20,14 @@ def _ts(value: str | None) -> float:
         return 0.0
 
 
+def _int_setting(data: dict[str, Any], current: dict[str, Any], key: str, default: int, minimum: int = 0) -> int:
+    raw = data.get(key) if key in data else current.get(key)
+    try:
+        return max(minimum, int(raw if raw is not None and raw != '' else default))
+    except (TypeError, ValueError):
+        return max(minimum, int(default))
+
+
 def _default_settings(user_id: int, profile_id: int) -> dict[str, Any]:
     return {
         'user_id': user_id,
@@ -29,6 +37,7 @@ def _default_settings(user_id: int, profile_id: int) -> dict[str, Any]:
         'stalled_seconds': 300,
         'min_speed_bytes': 1024,
         'min_seeds': 1,
+        'min_peers': 0,
         'manage_stopped': 0,
         'updated_at': utcnow(),
     }
@@ -49,27 +58,30 @@ def save_settings(profile_id: int, data: dict[str, Any], user_id: int | None = N
     current = get_settings(profile_id, user_id)
     settings = {
         'enabled': 1 if data.get('enabled', current.get('enabled')) else 0,
-        'max_active_downloads': max(1, int(data.get('max_active_downloads') or current.get('max_active_downloads') or 5)),
-        'stalled_seconds': max(30, int(data.get('stalled_seconds') or current.get('stalled_seconds') or 300)),
-        'min_speed_bytes': max(0, int(data.get('min_speed_bytes') or current.get('min_speed_bytes') or 0)),
-        'min_seeds': max(0, int(data.get('min_seeds') or current.get('min_seeds') or 0)),
+        'max_active_downloads': _int_setting(data, current, 'max_active_downloads', 5, 1),
+        'stalled_seconds': _int_setting(data, current, 'stalled_seconds', 300, 30),
+        'min_speed_bytes': _int_setting(data, current, 'min_speed_bytes', 0, 0),
+        'min_seeds': _int_setting(data, current, 'min_seeds', 0, 0),
+        # Note: Min peers is optional; when set, stalled detection requires low speed, low seeds and low peers.
+        'min_peers': _int_setting(data, current, 'min_peers', 0, 0),
         # Note: This switch protects fully stopped torrents from automatic starts; by default Smart Queue manages only paused items.
         'manage_stopped': 1 if data.get('manage_stopped', current.get('manage_stopped')) else 0,
     }
     now = utcnow()
     with connect() as conn:
         conn.execute(
-            '''INSERT INTO smart_queue_settings(user_id,profile_id,enabled,max_active_downloads,stalled_seconds,min_speed_bytes,min_seeds,manage_stopped,updated_at)
-               VALUES(?,?,?,?,?,?,?,?,?)
+            '''INSERT INTO smart_queue_settings(user_id,profile_id,enabled,max_active_downloads,stalled_seconds,min_speed_bytes,min_seeds,min_peers,manage_stopped,updated_at)
+               VALUES(?,?,?,?,?,?,?,?,?,?)
                ON CONFLICT(user_id, profile_id) DO UPDATE SET
                enabled=excluded.enabled,
                max_active_downloads=excluded.max_active_downloads,
                stalled_seconds=excluded.stalled_seconds,
                min_speed_bytes=excluded.min_speed_bytes,
                min_seeds=excluded.min_seeds,
+               min_peers=excluded.min_peers,
                manage_stopped=excluded.manage_stopped,
                updated_at=excluded.updated_at''',
-            (user_id, profile_id, settings['enabled'], settings['max_active_downloads'], settings['stalled_seconds'], settings['min_speed_bytes'], settings['min_seeds'], settings['manage_stopped'], now),
+            (user_id, profile_id, settings['enabled'], settings['max_active_downloads'], settings['stalled_seconds'], settings['min_speed_bytes'], settings['min_seeds'], settings['min_peers'], settings['manage_stopped'], now),
         )
     return get_settings(profile_id, user_id)
 
@@ -132,6 +144,60 @@ def _excluded_hashes(profile_id: int, user_id: int) -> set[str]:
     return {r['torrent_hash'] for r in list_exclusions(profile_id, user_id)}
 
 
+
+def _label_names(value: str | None) -> list[str]:
+    names: list[str] = []
+    for part in str(value or '').replace(';', ',').replace('|', ',').split(','):
+        label = part.strip()
+        if label and label not in names:
+            names.append(label)
+    return names
+
+
+def _label_value(labels: list[str]) -> str:
+    output: list[str] = []
+    for label in labels:
+        item = str(label or '').strip()
+        if item and item not in output:
+            output.append(item)
+    return ', '.join(output)
+
+
+def _has_smart_queue_label(value: str | None) -> bool:
+    return SMART_QUEUE_LABEL in _label_names(value)
+
+
+def _without_smart_queue_label(value: str | None) -> str:
+    return _label_value([label for label in _label_names(value) if label != SMART_QUEUE_LABEL])
+
+
+def _has_stalled_label(value: str | None) -> bool:
+    # Note: Stalled is treated case-insensitively so manually edited labels still block Smart Queue.
+    target = SMART_QUEUE_STALLED_LABEL.casefold()
+    return any(label.casefold() == target for label in _label_names(value))
+
+
+def _without_queue_technical_labels(value: str | None) -> str:
+    return _label_value([label for label in _label_names(value) if label != SMART_QUEUE_LABEL])
+
+
+def _ensure_stalled_label(client: Any, torrent_hash: str, current_label: str = '') -> bool:
+    labels = [label for label in _label_names(current_label) if label != SMART_QUEUE_LABEL]
+    changed = False
+    if not any(label.casefold() == SMART_QUEUE_STALLED_LABEL.casefold() for label in labels):
+        labels.append(SMART_QUEUE_STALLED_LABEL)
+        changed = True
+    if SMART_QUEUE_LABEL in _label_names(current_label):
+        changed = True
+    if not changed:
+        return True
+    try:
+        # Note: Stalled marking is idempotent; it adds Stalled and removes only the Smart Queue technical marker.
+        client.call('d.custom1.set', torrent_hash, _label_value(labels))
+        return True
+    except Exception:
+        return False
+
 def _remember_auto_label(profile_id: int, torrent_hash: str, previous_label: str) -> None:
     now = utcnow()
     with connect() as conn:
@@ -166,19 +232,18 @@ def _restore_auto_label(client: Any, profile_id: int, torrent_hash: str, current
         ).fetchone()
         live_label = _read_label(client, torrent_hash, current_label or '')
         if not row:
-            if live_label != SMART_QUEUE_LABEL:
+            if not _has_smart_queue_label(live_label):
                 return False
             try:
-                # Note: Clear the Smart Queue label even when the torrent was marked earlier but no previous-label entry remains.
-                client.call('d.custom1.set', torrent_hash, '')
+                # Note: Remove only the Smart Queue technical label and keep every user label untouched.
+                client.call('d.custom1.set', torrent_hash, _without_smart_queue_label(live_label))
                 return True
             except Exception:
                 return False
-        previous = row.get('previous_label') or ''
         try:
-            # Note: On resume, Smart Queue restores the previous label only while it still sees its own technical label.
-            if live_label == SMART_QUEUE_LABEL or current_label is None:
-                client.call('d.custom1.set', torrent_hash, previous)
+            # Note: Starting a torrent removes only Smart Queue's technical marker, so labels added while paused stay untouched.
+            if _has_smart_queue_label(live_label):
+                client.call('d.custom1.set', torrent_hash, _without_smart_queue_label(live_label))
             conn.execute('DELETE FROM smart_queue_auto_labels WHERE profile_id=? AND torrent_hash=?', (profile_id, torrent_hash))
             return True
         except Exception:
@@ -282,10 +347,16 @@ def _read_live_start_state(client: Any, torrent_hash: str) -> dict[str, Any]:
     result['started'] = bool(int(result.get('active') or 0))
     return result
 
-def _set_smart_queue_label(client: Any, torrent_hash: str, attempts: int = 3) -> bool:
+def _set_smart_queue_label(client: Any, torrent_hash: str, current_label: str = '', attempts: int = 3) -> bool:
+    labels = _label_names(current_label)
+    if SMART_QUEUE_LABEL in labels:
+        return True
+    labels.append(SMART_QUEUE_LABEL)
+    value = _label_value(labels)
     for attempt in range(max(1, attempts)):
         try:
-            client.call('d.custom1.set', torrent_hash, SMART_QUEUE_LABEL)
+            # Note: Smart Queue appends its technical label instead of overwriting existing torrent labels.
+            client.call('d.custom1.set', torrent_hash, value)
             return True
         except Exception:
             if attempt < attempts - 1:
@@ -298,15 +369,17 @@ def _mark_auto_paused(client: Any, profile_id: int, torrent: dict[str, Any]) ->
     if not torrent_hash:
         return False
     previous = str(torrent.get('label') or '')
-    if previous != SMART_QUEUE_LABEL:
+    if not _has_smart_queue_label(previous):
         _remember_auto_label(profile_id, torrent_hash, previous)
-    return _set_smart_queue_label(client, torrent_hash)
+    return _set_smart_queue_label(client, torrent_hash, previous)
 
 
 def _is_smart_queue_hold(torrent: dict[str, Any] | None, manage_stopped: bool = True) -> bool:
     if not torrent or int(torrent.get('complete') or 0):
         return False
-    if str(torrent.get('label') or '') == SMART_QUEUE_LABEL:
+    if _has_stalled_label(str(torrent.get('label') or '')):
+        return False
+    if _has_smart_queue_label(str(torrent.get('label') or '')):
         return True
     # Note: Paused in rTorrent usually has state=1 and active=0, so state=0 must not be required.
     # This lets Smart Queue treat paused torrents as pending and fill the queue target later.
@@ -319,11 +392,11 @@ def _is_smart_queue_hold(torrent: dict[str, Any] | None, manage_stopped: bool =
 
 
 def _clear_untracked_smart_queue_label(client: Any, torrent_hash: str, current_label: str) -> bool:
-    if current_label != SMART_QUEUE_LABEL:
+    if not _has_smart_queue_label(current_label):
         return False
     try:
-        # Note: Clear an orphaned Smart Queue label when no previous-label entry exists in the database.
-        client.call('d.custom1.set', torrent_hash, '')
+        # Note: Clear only the orphaned Smart Queue marker and keep unrelated labels intact.
+        client.call('d.custom1.set', torrent_hash, _without_smart_queue_label(current_label))
         return True
     except Exception:
         return False
@@ -346,8 +419,8 @@ def _cleanup_auto_labels(client: Any, profile_id: int, torrents: list[dict[str,
             if _restore_auto_label(client, profile_id, h, None if t is None else current_label):
                 restored.append(h)
             continue
-        if current_label != SMART_QUEUE_LABEL:
-            _set_smart_queue_label(client, h)
+        if not _has_smart_queue_label(current_label):
+            _set_smart_queue_label(client, h, current_label)
 
     for h, t in by_hash.items():
         if not h or h in keep_hashes or h in tracked_hashes or _is_smart_queue_hold(t, manage_stopped):
@@ -363,7 +436,7 @@ def _is_running_download_slot(t: dict[str, Any]) -> bool:
     # Paused can have state=1/open=1, so a slot is counted only after d.is_active=1.
     if int(t.get('complete') or 0):
         return False
-    if str(t.get('label') or '') == SMART_QUEUE_LABEL:
+    if _has_smart_queue_label(str(t.get('label') or '')) or _has_stalled_label(str(t.get('label') or '')):
         return False
     status = str(t.get('status') or '').lower()
     if status == 'checking' or status == 'paused' or bool(t.get('paused')):
@@ -375,7 +448,9 @@ def _is_waiting_download_candidate(t: dict[str, Any], manage_stopped: bool) -> b
     """Return True for paused/held torrents Smart Queue may resume later."""
     if int(t.get('complete') or 0):
         return False
-    if str(t.get('label') or '') == SMART_QUEUE_LABEL:
+    if _has_stalled_label(str(t.get('label') or '')):
+        return False
+    if _has_smart_queue_label(str(t.get('label') or '')):
         return True
     # Note: Paused items are the primary source for filling the queue, regardless of manage_stopped.
     if bool(t.get('paused')) or str(t.get('status') or '').lower() == 'paused':
@@ -403,10 +478,12 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
         return {'ok': True, 'enabled': False, 'paused': [], 'resumed': [], 'labels_restored': restored, 'message': 'Smart Queue disabled'}
 
     torrents = rtorrent.list_torrents(profile)
-    excluded = _excluded_hashes(profile_id, user_id)
+    # Note: Torrents marked as Stalled are treated as queue-blocked even when there are no other pending downloads.
+    stalled_label_hashes = {str(t.get('hash') or '') for t in torrents if _has_stalled_label(str(t.get('label') or '')) and t.get('hash')}
+    excluded = _excluded_hashes(profile_id, user_id) | stalled_label_hashes
     manage_stopped = bool(settings.get('manage_stopped'))
     def is_managed_hold(t: dict[str, Any]) -> bool:
-        return str(t.get('label') or '') == SMART_QUEUE_LABEL
+        return _has_smart_queue_label(str(t.get('label') or ''))
 
     # Note: Count Smart Queue slots by d.is_active because Paused can have state=1/open=1 and must not occupy the limit.
     downloading = [
@@ -425,6 +502,7 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
     ]
     min_speed = int(settings.get('min_speed_bytes') or 0)
     min_seeds = int(settings.get('min_seeds') or 0)
+    min_peers = int(settings.get('min_peers') or 0)
     stalled_seconds = int(settings.get('stalled_seconds') or 300)
     now = utcnow()
     now_ts = datetime.now(timezone.utc).timestamp()
@@ -432,7 +510,8 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
 
     with connect() as conn:
         for t in downloading:
-            is_stalled = int(t.get('down_rate') or 0) <= min_speed and int(t.get('seeds') or 0) <= min_seeds
+            # Note: Stalled detection requires low speed plus low seeds and, when configured, low peers.
+            is_stalled = int(t.get('down_rate') or 0) <= min_speed and int(t.get('seeds') or 0) <= min_seeds and (min_peers <= 0 or int(t.get('peers') or 0) <= min_peers)
             h = t.get('hash')
             if not h:
                 continue
@@ -472,13 +551,12 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
     to_pause: list[dict[str, Any]] = pause_rank[:max(0, len(downloading) - max_active)]
     pause_hashes = {str(t.get('hash') or '') for t in to_pause}
 
-    # Note: Stalled rotation runs only when the queue is full. When slots are missing, Smart Queue should
-    # first add missing items instead of pausing existing or incorrectly detected stalled items.
-    if candidates and len(downloading) >= max_active:
-        replaceable_stalled = [t for t in stalled if str(t.get('hash') or '') not in pause_hashes]
-        for t in replaceable_stalled[:max(0, len(candidates) - len(to_pause))]:
+    # Note: Confirmed stalled downloads are removed from the active queue immediately, then new candidates can fill those slots.
+    for t in stalled:
+        h = str(t.get('hash') or '')
+        if h and h not in pause_hashes:
             to_pause.append(t)
-            pause_hashes.add(str(t.get('hash') or ''))
+            pause_hashes.add(h)
 
     active_after_pause = max(0, len(downloading) - len(to_pause))
     available_slots = max(0, max_active - active_after_pause)
@@ -491,6 +569,7 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
     paused: list[str] = []
     resumed: list[str] = []
     label_failed: list[str] = []
+    stalled_labeled: list[str] = []
     start_failed: list[dict[str, str]] = []
     start_no_effect: list[dict[str, Any]] = []
     resume_requested: list[str] = []
@@ -498,12 +577,18 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
 
     for t in to_pause:
         try:
-            pause_result = rtorrent.pause_hash(c, t['hash'])
+            h = str(t.get('hash') or '')
+            pause_result = rtorrent.pause_hash(c, h)
             if not pause_result.get('ok'):
                 raise RuntimeError(pause_result.get('error') or 'pause failed')
-            if not _mark_auto_paused(c, profile_id, t):
-                label_failed.append(t['hash'])
-            paused.append(t['hash'])
+            if h in stalled_hashes:
+                if _ensure_stalled_label(c, h, _read_label(c, h, str(t.get('label') or ''))):
+                    stalled_labeled.append(h)
+                else:
+                    label_failed.append(h)
+            elif not _mark_auto_paused(c, profile_id, t):
+                label_failed.append(h)
+            paused.append(h)
         except Exception:
             pass
 
@@ -538,9 +623,9 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
     keep_labels = (
         set(paused)
         | {str(t.get('hash') or '') for t in to_label_waiting}
-        | {str(t.get('hash') or '') for t in stopped if str(t.get('label') or '') == SMART_QUEUE_LABEL and str(t.get('hash') or '') not in set(resumed)}
+        | {str(t.get('hash') or '') for t in stopped if _has_smart_queue_label(str(t.get('label') or '')) and str(t.get('hash') or '') not in set(resumed)}
     )
     restored = _cleanup_auto_labels(c, profile_id, torrents, keep_labels, manage_stopped)
-    details = {'excluded': len(excluded), 'enabled': bool(settings.get('enabled')), 'auto_label': SMART_QUEUE_LABEL, 'labels_restored': restored, 'labels_failed': label_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'start_results': start_results, 'resume_requested': resume_requested, 'active_verified': active_verified, 'waiting_labeled': len(to_label_waiting), 'manage_stopped': manage_stopped, 'max_active_downloads': max_active, 'active_before': len(downloading), 'active_after_expected': active_after_pause + len(resumed), 'paused_planned': len(to_pause), 'resumed_planned': len(to_resume), 'rtorrent_cap': rtorrent_cap}
+    details = {'excluded': len(excluded), 'excluded_stalled': len(stalled_label_hashes), 'enabled': bool(settings.get('enabled')), 'auto_label': SMART_QUEUE_LABEL, 'stalled_label': SMART_QUEUE_STALLED_LABEL, 'stalled_labeled': stalled_labeled, 'labels_restored': restored, 'labels_failed': label_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'start_results': start_results, 'resume_requested': resume_requested, 'active_verified': active_verified, 'waiting_labeled': len(to_label_waiting), 'manage_stopped': manage_stopped, 'max_active_downloads': max_active, 'active_before': len(downloading), 'active_after_expected': active_after_pause + len(resumed), 'paused_planned': len(to_pause), 'resumed_planned': len(to_resume), 'rtorrent_cap': rtorrent_cap}
     add_history(profile_id, 'force_check' if force else 'auto_check', paused, resumed, len(torrents), details, user_id)
-    return {'ok': True, 'enabled': bool(settings.get('enabled')), 'paused': paused, 'resumed': resumed, 'resume_requested': resume_requested, 'waiting_labeled': len(to_label_waiting), 'labels_restored': restored, 'labels_failed': label_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'active_verified': active_verified, 'rtorrent_cap': rtorrent_cap, 'checked': len(torrents), 'excluded': len(excluded), 'settings': settings}
+    return {'ok': True, 'enabled': bool(settings.get('enabled')), 'paused': paused, 'resumed': resumed, 'resume_requested': resume_requested, 'waiting_labeled': len(to_label_waiting), 'stalled_labeled': stalled_labeled, 'excluded_stalled': len(stalled_label_hashes), 'labels_restored': restored, 'labels_failed': label_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'active_verified': active_verified, 'rtorrent_cap': rtorrent_cap, 'checked': len(torrents), 'excluded': len(excluded), 'settings': settings}

pytorrent/services/workers.py

@@ -54,25 +54,33 @@ def _job_row(job_id: str):
         return conn.execute("SELECT rowid AS _rowid, * FROM jobs WHERE id=?", (job_id,)).fetchone()
 
 
-def _is_ordered_action(action_name: str) -> bool:
-    return action_name in {"move", "remove"}
+def _job_payload(row) -> dict:
+    try:
+        return json.loads((row or {}).get("payload_json") or "{}")
+    except Exception:
+        return {}
+
+
+def _is_ordered_job(row) -> bool:
+    payload = _job_payload(row)
+    # Note: Move/remove remain ordered, and automation-created jobs can opt in so effect order is visible and predictable.
+    return str((row or {}).get("action") or "") in {"move", "remove"} or bool(payload.get("automation_ordered"))
 
 
 def _has_prior_ordered_jobs(profile_id: int, rowid: int) -> bool:
     with connect() as conn:
-        row = conn.execute(
+        rows = conn.execute(
             """
-            SELECT 1
+            SELECT rowid AS _rowid, action, payload_json
             FROM jobs
             WHERE profile_id=?
               AND rowid<?
-              AND action IN ('move', 'remove')
               AND status IN ('pending', 'running')
-            LIMIT 1
+            ORDER BY rowid
             """,
             (profile_id, rowid),
-        ).fetchone()
-    return bool(row)
+        ).fetchall()
+    return any(_is_ordered_job(row) for row in rows)
 
 
 def _wait_for_prior_ordered_jobs(job_id: str, profile_id: int, rowid: int) -> bool:
@@ -140,7 +148,7 @@ def _run(job_id: str):
         return
     profile_id = int(profile["id"])
     ordered_lock = None
-    if _is_ordered_action(str(job["action"])):
+    if _is_ordered_job(job):
         if not _wait_for_prior_ordered_jobs(job_id, profile_id, int(job["_rowid"])):
             return
         ordered_lock = _get_exclusive_lock(profile_id)

pytorrent/static/styles.css

@@ -1238,36 +1238,60 @@ body.mobile-mode .mobile-card {
   color: var(--bs-primary-text-emphasis);
 }
 
-.automation-form-grid {
+.automation-shell {
+  display: grid;
+  gap: 0.75rem;
+}
+.automation-main-card {
+  padding: 0.75rem;
+  border: 1px solid var(--bs-border-color);
+  border-radius: 0.75rem;
+  background: var(--bs-body-bg);
+}
+.automation-card-title {
+  margin-bottom: 0.5rem;
+  font-weight: 700;
+}
+.automation-rule-grid,
+.automation-builder-grid {
   display: grid;
   grid-template-columns: repeat(4, minmax(160px, 1fr));
   gap: 0.5rem;
   align-items: center;
 }
-
-.auto-move-option {
-  gap: 0.45rem;
+.automation-enabled,
+.automation-negate {
   margin: 0;
-}
-
-
-.automation-builder-list {
-  display: grid;
-  grid-column: 1 / -1;
-  gap: 0.4rem;
-}
-
-.automation-chip {
-  display: flex;
-  align-items: center;
-  justify-content: space-between;
-  gap: 0.5rem;
-  padding: 0.4rem 0.55rem;
+  padding: 0.45rem 0.6rem 0.45rem 2.5rem;
   border: 1px solid var(--bs-border-color);
-  border-radius: 0.55rem;
-  background: var(--bs-secondary-bg);
+  border-radius: 0.5rem;
+}
+.automation-path-input {
+  grid-column: span 2;
+}
+.automation-chip-list {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 0.45rem;
+}
+.automation-chip {
+  display: inline-flex;
+  align-items: center;
+  gap: 0.35rem;
+  max-width: 100%;
+  padding: 0.25rem 0.5rem;
+  border: 1px solid var(--bs-border-color);
+  border-radius: 999px;
+  background: var(--bs-tertiary-bg);
+  font-size: 0.82rem;
+}
+.automation-actions,
+.automation-row-actions {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 0.4rem;
+  align-items: center;
 }
-
 .automation-row {
   display: flex;
   justify-content: space-between;
@@ -1279,10 +1303,109 @@ body.mobile-mode .mobile-card {
   margin-bottom: 0.45rem;
   background: var(--bs-body-bg);
 }
+.automation-row-main {
+  min-width: 0;
+}
+.automation-rule-summary {
+  overflow-wrap: anywhere;
+}
+.automation-action-pill {
+  display: inline-flex;
+  max-width: 100%;
+  margin: 0.1rem;
+  padding: 0.15rem 0.4rem;
+  border-radius: 999px;
+  background: var(--bs-secondary-bg);
+  font-size: 0.78rem;
+  overflow-wrap: anywhere;
+  white-space: normal;
+  word-break: break-word;
+}
+.automation-history-toolbar {
+  display: flex;
+  justify-content: flex-end;
+  margin-bottom: 0.5rem;
+}
+/* Note: Automation history has fixed compact metadata columns and a flexible Actions column, so long JSON cannot overlap Time/Rule. */
+.automation-history-table {
+  width: 100%;
+  table-layout: fixed;
+  white-space: normal;
+}
+.automation-history-table th,
+.automation-history-table td {
+  min-width: 0;
+  vertical-align: top;
+}
+.automation-history-table th:nth-child(1),
+.automation-history-table td:nth-child(1) {
+  width: 9rem;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+.automation-history-table th:nth-child(2),
+.automation-history-table td:nth-child(2) {
+  width: 11rem;
+  overflow: hidden;
+  overflow-wrap: anywhere;
+  word-break: break-word;
+}
+.automation-history-table th:nth-child(3),
+.automation-history-table td:nth-child(3) {
+  width: 12rem;
+  overflow: hidden;
+  overflow-wrap: anywhere;
+  word-break: break-word;
+}
+.automation-history-table th:nth-child(4),
+.automation-history-table td:nth-child(4) {
+  width: auto;
+  overflow: hidden;
+  overflow-wrap: anywhere;
+  word-break: break-word;
+}
+.automation-history-details {
+  display: block;
+  min-width: 0;
+  max-width: 100%;
+}
+.automation-history-details summary {
+  display: block;
+  max-width: 100%;
+  cursor: pointer;
+  list-style-position: inside;
+  overflow-wrap: anywhere;
+  white-space: normal;
+  word-break: break-word;
+}
+.automation-history-details pre,
+.automation-history-raw {
+  max-width: 100%;
+  max-height: 220px;
+  margin: 0.35rem 0 0;
+  padding: 0.5rem;
+  overflow: auto;
+  border: 1px solid var(--bs-border-color);
+  border-radius: 0.5rem;
+  background: var(--bs-tertiary-bg);
+  overflow-wrap: anywhere;
+  white-space: pre-wrap;
+  word-break: break-word;
+}
 @media (max-width: 900px) {
-  .automation-form-grid {
+  .automation-rule-grid,
+  .automation-builder-grid {
     grid-template-columns: 1fr;
   }
+  .automation-path-input,
+  .automation-history-details {
+    grid-column: auto;
+    max-width: 100%;
+  }
+  .automation-history-toolbar {
+    justify-content: flex-start;
+  }
 }
 .disk-status {
   display: inline-flex;