diff --git a/mvlog_worker.py b/mvlog_worker.py
index 38f0a02..b2b1416 100755
--- a/mvlog_worker.py
+++ b/mvlog_worker.py
@@ -26,6 +26,8 @@ PUSH_CONFIG = Path.home() / ".config" / "mvlog" / "push.json"
 STATE_NAME = ".movmaker-state.json"
 LOCK_NAME = ".movmaker-lock"
 ENABLED_NAME = ".movmaker-enabled"
+HIDDEN_NAME = ".mvlog-hidden"
+NOTIFIED_CACHE = "cache/push_notifications.json"
 
 
 def now() -> str:
@@ -85,17 +87,33 @@ print(hashlib.sha256(json.dumps(entries, ensure_ascii=False, separators=(',', ':
     return ssh(host, cmd).stdout.strip()
 
 
-def read_state(host: str, job_dir: str) -> dict:
-    path = f"{job_dir}/{STATE_NAME}"
+def read_remote_json(host: str, path: str) -> dict:
     res = ssh(host, f"cat {remote_q(path)} 2>/dev/null", check=False)
     if res.returncode != 0 or not res.stdout.strip():
         return {}
     try:
-        return json.loads(res.stdout)
+        data = json.loads(res.stdout)
+        return data if isinstance(data, dict) else {}
     except json.JSONDecodeError:
         return {}
 
 
+def write_remote_json(host: str, path: str, data: dict) -> None:
+    tmp_path = f"{path}.tmp.$$"
+    payload = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
+    cmd = (
+        f"mkdir -p {remote_q(str(Path(path).parent))} && "
+        f"cat > {remote_q(tmp_path)} && "
+        f"chmod 664 {remote_q(tmp_path)} && "
+        f"mv {remote_q(tmp_path)} {remote_q(path)}"
+    )
+    ssh(host, cmd, input_text=payload)
+
+
+def read_state(host: str, job_dir: str) -> dict:
+    return read_remote_json(host, f"{job_dir}/{STATE_NAME}")
+
+
 def write_state(host: str, job_dir: str, state: dict) -> None:
     path = f"{job_dir}/{STATE_NAME}"
     tmp_path = f"{path}.tmp.$$"
@@ -127,12 +145,12 @@ def copy_job_from_remote(host: str, job_dir: str, local_input: Path) -> None:
         raise
 
 
-def send_push_notification(host: str, remote_root: str, output: str) -> None:
+def send_push_notification(host: str, remote_root: str, output: str) -> bool:
     if not PUSH_SENDER.exists() or not PUSH_CONFIG.exists():
-        return
+        return False
     res = ssh(host, f"cat {remote_q(remote_root + '/cache/push_subscriptions.json')} 2>/dev/null", check=False)
     if res.returncode != 0 or not res.stdout.strip():
-        return
+        return False
     title = output.rsplit('.', 1)[0].replace('_', ' ').title()
     payload = json.dumps({
         "title": "New MVLog video",
@@ -145,6 +163,26 @@ def send_push_notification(host: str, remote_root: str, output: str) -> None:
         print(proc.stdout.strip())
     if proc.stderr.strip():
         print(proc.stderr.strip(), file=sys.stderr)
+    return proc.returncode == 0
+
+
+def maybe_send_first_show_notification(host: str, remote_root: str, job_dir: str, job: str, state: dict | None = None) -> None:
+    if remote_file_exists(host, f"{job_dir}/{HIDDEN_NAME}"):
+        return
+    state = state if state is not None else read_state(host, job_dir)
+    output = state.get("output") if isinstance(state.get("output"), str) else ""
+    if not output or not remote_file_exists(host, f"{remote_root}/out-dir/{output}"):
+        return
+    cache_path = f"{remote_root}/{NOTIFIED_CACHE}"
+    cache = read_remote_json(host, cache_path)
+    sent = cache.get("sent") if isinstance(cache.get("sent"), dict) else {}
+    if output in sent:
+        return
+    if send_push_notification(host, remote_root, output):
+        sent[output] = {"job": job, "sent_at": now()}
+        cache["sent"] = sent
+        write_remote_json(host, cache_path, cache)
+        print(f"notification sent: {job} -> {output}")
 
 
 def publish_output(host: str, remote_root: str, local_mp4: Path, old_output: str | None) -> str:
@@ -174,6 +212,8 @@ def render_job(local_input: Path, local_out: Path, extra_args: list[str]) -> Pat
 
 def process_job(host: str, remote_root: str, job: str, work_dir: Path, extra_args: list[str], force: bool) -> None:
     job_dir = f"{remote_root}/in-dir/{job}"
+    state = read_state(host, job_dir)
+    maybe_send_first_show_notification(host, remote_root, job_dir, job, state)
     if not remote_file_exists(host, f"{job_dir}/{ENABLED_NAME}"):
         print(f"disabled: {job}")
         return
@@ -219,14 +259,16 @@ def process_job(host: str, remote_root: str, job: str, work_dir: Path, extra_arg
             return
 
         output = publish_output(host, remote_root, local_mp4, old_output)
-        send_push_notification(host, remote_root, output)
-        write_state(host, job_dir, {
+        new_state = {
             "status": "done",
             "fingerprint": fingerprint_after,
             "output": output,
             "generated_at": now(),
-        })
+        }
+        maybe_send_first_show_notification(host, remote_root, job_dir, job, new_state)
+        write_state(host, job_dir, new_state)
         print(f"published: {job} -> {output}")
+        return
     except Exception as exc:
         try:
             write_state(host, job_dir, {