Levandor

Frigate viewer alert — implementation plan

13 min read

Frigate Viewer Alert Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Send a Telegram message when someone opens the Frigate UI, identifying who by tailnet device name.

Architecture: A python3-stdlib daemon on telep-mainframe follows docker logs -f frigate, parses nginx access lines, and on a new viewing session (tailnet source IP, quiet > 5 min) resolves the IP to a tailscale device name and sends a Telegram message via the existing frigate-notify bot. Pure-logic modules are unit-tested; the daemon wires them and reconnects on Frigate restart.

Tech Stack: Python 3.13 stdlib only, docker logs, tailscale status, Telegram Bot API over urllib, systemd.

Design: 2026-07-23-frigate-viewer-alert-design

Global Constraints

  • Python 3 stdlib only. No pip installs.
  • No comments, docstrings, or inline annotations in any code (user’s global rule). Self-explanatory naming only.
  • Do not run git commit. Commit steps are omitted per the user’s global rule; each task ends in a verification step.
  • All code lives in /home/levander/frigate-viewer-alert/ on telep-mainframe. Runs as user levander (NOT root).
  • Reuse the existing Telegram bot: read token + chatid from ~/nvr/frigate-notify/config.yml at runtime. NEVER copy the token into this project’s files, tests, or logs.
  • The container name is frigate. Tailnet addresses match ^100\.\d{1,3}\.\d{1,3}\.\d{1,3}$.
  • COOLDOWN_SECONDS = 300.
  • The daemon must never crash the process and must never affect anything on the box (read-only: logs in, Telegram out).
  • Run all commands on the box: /usr/bin/ssh levander@telep-mainframe. On this Mac, bare ssh is broken — always use /usr/bin/ssh.

Test command (every task): cd /home/levander/frigate-viewer-alert && python3 -m unittest <module> -v

Task 1: Access-log parser

Files:

  • Create: /home/levander/frigate-viewer-alert/logparse.py
  • Test: /home/levander/frigate-viewer-alert/test_logparse.py

Interfaces:

  • Produces:

    • parse_access_line(line) -> dict | None — returns {"ip": str, "method": str, "path": str, "status": int} for an nginx access line whose X-Forwarded-For is an IPv4 address, else None
    • is_tailnet(ip) -> bool — True iff ip matches ^100\.\d{1,3}\.\d{1,3}\.\d{1,3}$
    • extract_camera(path) -> str | None — camera name from /live/<transport>/<camera>, else None

  • Step 1: Create the directory

/usr/bin/ssh levander@telep-mainframe 'mkdir -p /home/levander/frigate-viewer-alert'
  • Step 2: Write the failing test

Create /home/levander/frigate-viewer-alert/test_logparse.py:

import unittest
 
from logparse import extract_camera, is_tailnet, parse_access_line
 
LIVE_LINE = (
    '172.18.0.1 - - [20/Jul/2026:22:33:43 +0200] "GET /live/jsmpeg/telep_cam2 HTTP/1.1" 101 '
    '56992 "-" "Mozilla/5.0" "100.83.222.120" request_time="1.125" upstream_response_time="1.121"'
)
UI_LINE = (
    '172.18.0.1 - - [22/Jul/2026:13:06:55 +0200] "GET /api/config HTTP/1.1" 401 581 '
    '"https://telep-mainframe.taild4189d.ts.net/" "Mozilla/5.0" "100.83.222.120" '
    'request_time="0.003" upstream_response_time="-"'
)
INTERNAL_LINE = (
    '172.18.0.2 - - [23/Jul/2026:17:29:49 +0200] "GET /api/review HTTP/1.1" 200 1189 '
    '"-" "Frigate-Notify/v0.5.4" "-" request_time="0.014" upstream_response_time="0.008"'
)
GO2RTC_LINE = (
    "01:20:29.319432299  01:20:29.319 WRN github.com/AlexxIT/go2rtc/internal/streams/producer.go:170 "
    '> error="read tcp" url=rtsp://x'
)
 
 
class TestParse(unittest.TestCase):
    def test_live_line(self):
        result = parse_access_line(LIVE_LINE)
        self.assertEqual(result["ip"], "100.83.222.120")
        self.assertEqual(result["method"], "GET")
        self.assertEqual(result["path"], "/live/jsmpeg/telep_cam2")
        self.assertEqual(result["status"], 101)
 
    def test_ui_line(self):
        result = parse_access_line(UI_LINE)
        self.assertEqual(result["ip"], "100.83.222.120")
        self.assertEqual(result["path"], "/api/config")
        self.assertEqual(result["status"], 401)
 
    def test_internal_dash_xff_ignored(self):
        self.assertIsNone(parse_access_line(INTERNAL_LINE))
 
    def test_go2rtc_line_ignored(self):
        self.assertIsNone(parse_access_line(GO2RTC_LINE))
 
    def test_blank_line_ignored(self):
        self.assertIsNone(parse_access_line(""))
 
 
class TestTailnet(unittest.TestCase):
    def test_tailnet_true(self):
        self.assertTrue(is_tailnet("100.83.222.120"))
 
    def test_docker_gateway_false(self):
        self.assertFalse(is_tailnet("172.18.0.1"))
 
    def test_public_hundred_prefix_false(self):
        self.assertFalse(is_tailnet("100.83.222.1200"))
 
 
class TestCamera(unittest.TestCase):
    def test_live_path(self):
        self.assertEqual(extract_camera("/live/jsmpeg/telep_cam2"), "telep_cam2")
 
    def test_webrtc_path(self):
        self.assertEqual(extract_camera("/live/webrtc/telep_cam1"), "telep_cam1")
 
    def test_non_live_path(self):
        self.assertIsNone(extract_camera("/api/config"))
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 3: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_logparse -v'

Expected: FAIL with ModuleNotFoundError: No module named 'logparse'

  • Step 4: Write the implementation

Create /home/levander/frigate-viewer-alert/logparse.py:

import re
 
_XFF = re.compile(r'"([^"]*)" request_time=')
_REQUEST = re.compile(r'"([A-Z]+) (\S+) HTTP/[0-9.]+" (\d+)')
_TAILNET = re.compile(r"^100\.\d{1,3}\.\d{1,3}\.\d{1,3}$")
_CAMERA = re.compile(r"^/live/[^/]+/([^/?]+)")
 
 
def is_tailnet(ip):
    return bool(_TAILNET.match(ip))
 
 
def extract_camera(path):
    match = _CAMERA.match(path)
    if match:
        return match.group(1)
    return None
 
 
def parse_access_line(line):
    xff = _XFF.search(line)
    if xff is None:
        return None
    ip = xff.group(1)
    if not re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$", ip):
        return None
    request = _REQUEST.search(line)
    if request is None:
        return None
    return {
        "ip": ip,
        "method": request.group(1),
        "path": request.group(2),
        "status": int(request.group(3)),
    }
  • Step 5: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_logparse -v'

Expected: OK — 11 tests pass.

  • Step 6: Verify against real Frigate log lines
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && docker logs frigate 2>&1 | grep "request_time=" | tail -30 > /tmp/real_lines.txt; python3 -c "
import logparse
hits = 0
for line in open(\"/tmp/real_lines.txt\"):
    r = logparse.parse_access_line(line)
    if r and logparse.is_tailnet(r[\"ip\"]):
        hits += 1
        print(r)
print(\"tailnet viewer lines:\", hits)
"'

Expected: parsed dicts for real tailnet requests; internal (- XFF) lines excluded.

Task 2: Session gate

Files:

  • Create: /home/levander/frigate-viewer-alert/sessions.py
  • Test: /home/levander/frigate-viewer-alert/test_sessions.py

Interfaces:

  • Produces: SessionGate(cooldown) with should_alert(ip, now) -> bool — returns True iff ip unseen or now - last_seen[ip] > cooldown, and records last_seen[ip] = now on every call.

  • Step 1: Write the failing test

Create /home/levander/frigate-viewer-alert/test_sessions.py:

import unittest
 
from sessions import SessionGate
 
 
class TestSessionGate(unittest.TestCase):
    def test_first_hit_alerts(self):
        gate = SessionGate(300)
        self.assertTrue(gate.should_alert("100.0.0.1", 1000.0))
 
    def test_second_hit_within_cooldown_silent(self):
        gate = SessionGate(300)
        gate.should_alert("100.0.0.1", 1000.0)
        self.assertFalse(gate.should_alert("100.0.0.1", 1200.0))
 
    def test_hit_after_cooldown_alerts(self):
        gate = SessionGate(300)
        gate.should_alert("100.0.0.1", 1000.0)
        self.assertTrue(gate.should_alert("100.0.0.1", 1301.0))
 
    def test_activity_refreshes_timer(self):
        gate = SessionGate(300)
        gate.should_alert("100.0.0.1", 1000.0)
        gate.should_alert("100.0.0.1", 1200.0)
        self.assertFalse(gate.should_alert("100.0.0.1", 1450.0))
 
    def test_independent_ips(self):
        gate = SessionGate(300)
        gate.should_alert("100.0.0.1", 1000.0)
        self.assertTrue(gate.should_alert("100.0.0.2", 1000.0))
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_sessions -v'

Expected: FAIL with ModuleNotFoundError: No module named 'sessions'

  • Step 3: Write the implementation

Create /home/levander/frigate-viewer-alert/sessions.py:

class SessionGate:
    def __init__(self, cooldown):
        self.cooldown = cooldown
        self.last_seen = {}
 
    def should_alert(self, ip, now):
        previous = self.last_seen.get(ip)
        self.last_seen[ip] = now
        if previous is None:
            return True
        return now - previous > self.cooldown
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_sessions -v'

Expected: OK — 5 tests pass.

Task 3: Name resolution

Files:

  • Create: /home/levander/frigate-viewer-alert/names.py
  • Test: /home/levander/frigate-viewer-alert/test_names.py

Interfaces:

  • Produces:

    • parse_status(text) -> dict[str, str] — maps 100.x IP → hostname from tailscale status output
    • resolve(ip, names) -> str — names[ip] if present, else ip

  • Step 1: Write the failing test

Create /home/levander/frigate-viewer-alert/test_names.py:

import unittest
 
from names import parse_status, resolve
 
STATUS = """100.115.209.87  telep-mainframe      myuser@      linux   -
100.83.222.120  personal-mac         myuser@      macOS   active; direct
100.72.53.4     iphone-14-pro        myuser@      iOS     idle
# a comment line
garbage line without ip
"""
 
 
class TestParseStatus(unittest.TestCase):
    def test_maps_ip_to_hostname(self):
        names = parse_status(STATUS)
        self.assertEqual(names["100.83.222.120"], "personal-mac")
        self.assertEqual(names["100.72.53.4"], "iphone-14-pro")
 
    def test_ignores_non_ip_lines(self):
        names = parse_status(STATUS)
        self.assertEqual(len(names), 3)
 
    def test_empty_input(self):
        self.assertEqual(parse_status(""), {})
 
 
class TestResolve(unittest.TestCase):
    def test_known_ip(self):
        self.assertEqual(resolve("100.83.222.120", {"100.83.222.120": "personal-mac"}), "personal-mac")
 
    def test_unknown_ip_returns_raw(self):
        self.assertEqual(resolve("100.9.9.9", {}), "100.9.9.9")
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_names -v'

Expected: FAIL with ModuleNotFoundError: No module named 'names'

  • Step 3: Write the implementation

Create /home/levander/frigate-viewer-alert/names.py:

import re
 
_ROW = re.compile(r"^(100\.\d{1,3}\.\d{1,3}\.\d{1,3})\s+(\S+)")
 
 
def parse_status(text):
    names = {}
    for line in text.splitlines():
        match = _ROW.match(line)
        if match:
            names[match.group(1)] = match.group(2)
    return names
 
 
def resolve(ip, names):
    return names.get(ip, ip)
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_names -v'

Expected: OK — 5 tests pass.

  • Step 5: Verify against real tailscale status
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -c "
import subprocess, names
out = subprocess.run([\"tailscale\",\"status\"], capture_output=True, text=True).stdout
m = names.parse_status(out)
print(len(m), \"devices\")
print(names.resolve(\"100.83.222.120\", m))
"'

Expected: a device count and personal-mac.

Task 4: Telegram creds + send

Files:

  • Create: /home/levander/frigate-viewer-alert/notify.py
  • Test: /home/levander/frigate-viewer-alert/test_notify.py

Interfaces:

  • Produces:

    • load_telegram_creds(config_path) -> tuple[str, str] — (token, chatid) from the telegram: block of the frigate-notify YAML; raises ValueError if either is missing
    • send(token, chatid, text) -> bool — POST to the Telegram API via urllib, 10s timeout, returns False on any error, never raises

  • Step 1: Write the failing test

Create /home/levander/frigate-viewer-alert/test_notify.py:

import os
import tempfile
import unittest
 
from notify import load_telegram_creds
 
CONFIG = """alerts:
  telegram:
    enabled: true
    chatid: -1004475187307
    token: 123456789:AA_fake_token_value
  zone_filter: false
other:
  key: value
"""
 
 
class TestLoadCreds(unittest.TestCase):
    def _write(self, text):
        handle = tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False)
        handle.write(text)
        handle.close()
        self.addCleanup(os.unlink, handle.name)
        return handle.name
 
    def test_extracts_token_and_chatid(self):
        token, chatid = load_telegram_creds(self._write(CONFIG))
        self.assertEqual(token, "123456789:AA_fake_token_value")
        self.assertEqual(chatid, "-1004475187307")
 
    def test_missing_token_raises(self):
        path = self._write("alerts:\n  telegram:\n    chatid: -100\n")
        with self.assertRaises(ValueError):
            load_telegram_creds(path)
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_notify -v'

Expected: FAIL with ModuleNotFoundError: No module named 'notify'

  • Step 3: Write the implementation

Create /home/levander/frigate-viewer-alert/notify.py:

import json
import re
import sys
import urllib.request
 
_TOKEN = re.compile(r"^\s*token:\s*(\S+)")
_CHATID = re.compile(r"^\s*chatid:\s*(\S+)")
 
 
def load_telegram_creds(config_path):
    token = None
    chatid = None
    with open(config_path, encoding="utf-8") as handle:
        for line in handle:
            if token is None:
                match = _TOKEN.match(line)
                if match:
                    token = match.group(1)
            if chatid is None:
                match = _CHATID.match(line)
                if match:
                    chatid = match.group(1)
    if not token or not chatid:
        raise ValueError("telegram token or chatid not found in config")
    return token, chatid
 
 
def send(token, chatid, text):
    url = "https://api.telegram.org/bot" + token + "/sendMessage"
    payload = json.dumps({"chat_id": chatid, "text": text}).encode("utf-8")
    request = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(request, timeout=10) as response:
            response.read()
        return True
    except Exception as error:
        print("telegram send failed:", error, file=sys.stderr, flush=True)
        return False
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_notify -v'

Expected: OK — 2 tests pass.

  • Step 5: Verify creds load from the real notify config (no secret printed)
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -c "
import notify
t, c = notify.load_telegram_creds(\"/home/levander/nvr/frigate-notify/config.yml\")
print(\"token len:\", len(t), \"chatid:\", c)
"'

Expected: a non-zero token length and the chat id. The token value itself must NOT be printed.

Task 5: Daemon + message formatting

Files:

  • Create: /home/levander/frigate-viewer-alert/watch.py
  • Test: /home/levander/frigate-viewer-alert/test_watch.py

Interfaces:

  • Consumes: everything from Tasks 1-4

  • Produces:

    • format_message(name, ip, camera, when) -> str — the Telegram text; when is a time.struct_time
    • handle_line(line, gate, names, now, emit) — parse + gate + resolve + call emit(text) on a session start
    • main() — the daemon loop
    • Constants CONTAINER = "frigate", NOTIFY_CONFIG, COOLDOWN_SECONDS = 300

  • Step 1: Write the failing test

Create /home/levander/frigate-viewer-alert/test_watch.py:

import time
import unittest
 
from sessions import SessionGate
from watch import format_message, handle_line
 
LIVE_LINE = (
    '172.18.0.1 - - [20/Jul/2026:22:33:43 +0200] "GET /live/jsmpeg/telep_cam2 HTTP/1.1" 101 '
    '56992 "-" "Mozilla/5.0" "100.83.222.120" request_time="1.125" upstream_response_time="1.121"'
)
UI_LINE = (
    '172.18.0.1 - - [22/Jul/2026:13:06:55 +0200] "GET /api/config HTTP/1.1" 401 581 '
    '"ref" "Mozilla/5.0" "100.83.222.120" request_time="0.003" upstream_response_time="-"'
)
INTERNAL_LINE = (
    '172.18.0.2 - - [23/Jul/2026:17:29:49 +0200] "GET /api/review HTTP/1.1" 200 1189 '
    '"-" "Frigate-Notify" "-" request_time="0.014" upstream_response_time="0.008"'
)
NAMES = {"100.83.222.120": "personal-mac"}
 
 
class TestFormatMessage(unittest.TestCase):
    def test_includes_name_ip_time(self):
        when = time.struct_time((2026, 7, 23, 17, 42, 0, 0, 0, 0))
        text = format_message("personal-mac", "100.83.222.120", None, when)
        self.assertIn("personal-mac", text)
        self.assertIn("100.83.222.120", text)
        self.assertIn("17:42", text)
 
    def test_includes_camera_when_present(self):
        when = time.struct_time((2026, 7, 23, 17, 42, 0, 0, 0, 0))
        text = format_message("personal-mac", "100.83.222.120", "telep_cam2", when)
        self.assertIn("telep_cam2", text)
 
 
class TestHandleLine(unittest.TestCase):
    def _emit_collector(self):
        sent = []
        return sent, lambda text: sent.append(text)
 
    def test_tailnet_ui_line_emits_once(self):
        sent, emit = self._emit_collector()
        gate = SessionGate(300)
        handle_line(UI_LINE, gate, NAMES, 1000.0, emit)
        self.assertEqual(len(sent), 1)
        self.assertIn("personal-mac", sent[0])
 
    def test_second_line_within_cooldown_silent(self):
        sent, emit = self._emit_collector()
        gate = SessionGate(300)
        handle_line(UI_LINE, gate, NAMES, 1000.0, emit)
        handle_line(LIVE_LINE, gate, NAMES, 1100.0, emit)
        self.assertEqual(len(sent), 1)
 
    def test_internal_line_ignored(self):
        sent, emit = self._emit_collector()
        gate = SessionGate(300)
        handle_line(INTERNAL_LINE, gate, NAMES, 1000.0, emit)
        self.assertEqual(len(sent), 0)
 
    def test_live_line_includes_camera(self):
        sent, emit = self._emit_collector()
        gate = SessionGate(300)
        handle_line(LIVE_LINE, gate, NAMES, 1000.0, emit)
        self.assertEqual(len(sent), 1)
        self.assertIn("telep_cam2", sent[0])
 
 
if __name__ == "__main__":
    unittest.main()
  • Step 2: Run the test to verify it fails
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_watch -v'

Expected: FAIL with ModuleNotFoundError: No module named 'watch'

  • Step 3: Write the implementation

Create /home/levander/frigate-viewer-alert/watch.py:

import subprocess
import sys
import time
 
from logparse import extract_camera, is_tailnet, parse_access_line
from names import parse_status, resolve
from notify import load_telegram_creds, send
from sessions import SessionGate
 
CONTAINER = "frigate"
NOTIFY_CONFIG = "/home/levander/nvr/frigate-notify/config.yml"
COOLDOWN_SECONDS = 300
NAME_REFRESH_SECONDS = 300
RECONNECT_MAX_BACKOFF = 30
 
 
def format_message(name, ip, camera, when):
    stamp = time.strftime("%H:%M", when)
    base = "\U0001f440 " + name + " (" + ip + ") opened Frigate — " + stamp
    if camera:
        return base + " · live: " + camera
    return base
 
 
def handle_line(line, gate, names, now, emit):
    parsed = parse_access_line(line)
    if parsed is None or not is_tailnet(parsed["ip"]):
        return
    if not gate.should_alert(parsed["ip"], now):
        return
    name = resolve(parsed["ip"], names)
    camera = extract_camera(parsed["path"])
    emit(format_message(name, parsed["ip"], camera, time.localtime()))
 
 
def load_names():
    try:
        output = subprocess.run(
            ["tailscale", "status"], capture_output=True, text=True, timeout=10
        ).stdout
    except (subprocess.SubprocessError, OSError):
        return {}
    return parse_status(output)
 
 
def follow_logs():
    return subprocess.Popen(
        ["docker", "logs", "-f", "--since", "0m", CONTAINER],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
        bufsize=1,
    )
 
 
def main():
    token, chatid = load_telegram_creds(NOTIFY_CONFIG)
    gate = SessionGate(COOLDOWN_SECONDS)
    names = load_names()
    names_refreshed = time.monotonic()
    backoff = 2
    emit = lambda text: send(token, chatid, text)
    while True:
        process = None
        try:
            process = follow_logs()
            backoff = 2
            for line in process.stdout:
                now = time.monotonic()
                if now - names_refreshed > NAME_REFRESH_SECONDS:
                    refreshed = load_names()
                    if refreshed:
                        names = refreshed
                    names_refreshed = now
                parsed = parse_access_line(line)
                if parsed is not None and is_tailnet(parsed["ip"]) and parsed["ip"] not in names:
                    refreshed = load_names()
                    if refreshed:
                        names = refreshed
                    names_refreshed = now
                handle_line(line, gate, names, now, emit)
        except Exception as error:
            print("watch loop error:", error, file=sys.stderr, flush=True)
        finally:
            if process is not None:
                try:
                    process.kill()
                except OSError:
                    pass
        print("frigate log stream ended; reconnecting", file=sys.stderr, flush=True)
        time.sleep(backoff)
        backoff = min(backoff * 2, RECONNECT_MAX_BACKOFF)
 
 
if __name__ == "__main__":
    main()
  • Step 4: Run the test to verify it passes
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest test_watch -v'

Expected: OK — 6 tests pass.

  • Step 5: Full suite
/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && python3 -m unittest discover -p "test_*.py" 2>&1 | tail -3'

Expected: OK — 29 tests pass.

  • Step 6: Foreground smoke test — send ONE real alert

Run the daemon in the foreground, then from a tailnet device (your Mac) open https://telep-mainframe.taild4189d.ts.net. Confirm a Telegram message arrives.

/usr/bin/ssh levander@telep-mainframe 'cd /home/levander/frigate-viewer-alert && timeout 45 python3 watch.py & sleep 2; echo "now open Frigate from a tailnet device within 40s"; wait'

Expected: within the window, one Telegram message like 👁 personal-mac (100.83.222.120) opened Frigate — HH:MM. If nothing arrives, check that the device actually hit the UI and that creds loaded. Report the real outcome.

Task 6: systemd service + end-to-end verification

Files:

  • Create: /etc/systemd/system/frigate-viewer-alert.service

  • Step 1: Install and enable the unit

/usr/bin/ssh levander@telep-mainframe 'sudo tee /etc/systemd/system/frigate-viewer-alert.service > /dev/null <<UNIT
[Unit]
Description=Telegram alert when someone opens the Frigate UI
After=docker.service network-online.target
Wants=network-online.target
 
[Service]
Type=simple
User=levander
WorkingDirectory=/home/levander/frigate-viewer-alert
ExecStart=/usr/bin/python3 /home/levander/frigate-viewer-alert/watch.py
Restart=always
RestartSec=5
 
[Install]
WantedBy=multi-user.target
UNIT
sudo systemctl daemon-reload && sudo systemctl enable --now frigate-viewer-alert.service && sleep 5 && systemctl is-active frigate-viewer-alert.service'

Expected: active.

  • Step 2: Confirm it is following the log (no errors)
/usr/bin/ssh levander@telep-mainframe 'journalctl -u frigate-viewer-alert.service -n 20 --no-pager'

Expected: no Python tracebacks; process running. (It is quiet until a viewer appears.)

  • Step 3: End-to-end — new session alerts, repeat within cooldown is silent

From a tailnet device, open Frigate → expect one Telegram message. Reload the page immediately → expect NO second message (within the 5-min cooldown). Wait > 5 min, open again → expect a new message.

Report the real observations (message text, whether the duplicate was correctly suppressed).

  • Step 4: Restart resilience
/usr/bin/ssh levander@telep-mainframe 'sudo systemctl restart docker 2>/dev/null || sudo docker restart frigate; sleep 20; journalctl -u frigate-viewer-alert.service -n 8 --no-pager | tail -8'

Expected: a reconnecting line then normal operation; the service stays active. Then open Frigate once more and confirm an alert still fires.

  • Step 5: Confirm enabled for reboot
/usr/bin/ssh levander@telep-mainframe 'systemctl is-enabled frigate-viewer-alert.service'

Expected: enabled.

Self-review notes

Spec coverage: log-follow detection (Task 5), tailnet-XFF match with internal exclusion (Task 1), session/cooldown dedup (Task 2, wired Task 5), tailscale name resolution (Task 3), reuse of frigate-notify creds without copying the token (Task 4), message format with optional camera (Task 5), reconnect on Frigate restart (Task 5, verified Task 6 Step 4), systemd service as levander (Task 6), end-to-end + dedup verification (Task 6 Step 3).

Deliberate deviation from skill defaults: git commit steps omitted per the user’s global no-commit rule.

Graph View

Backlinks