diff --git a/event/stream/topics.go b/event/stream/topics.go
new file mode 100644
index 0000000..35b1bf4
--- /dev/null
+++ b/event/stream/topics.go
@@ -0,0 +1,72 @@
+package stream
+
+import "strings"
+
+// Classify maps an ONVIF topic string (e.g. "tns1:VideoSource/MotionAlarm")
+// to the normalized EventKind that callers should switch on. Returns
+// KindUnknown when no rule matches.
+//
+// The classifier strips XML-namespace prefixes from each path segment so it
+// is robust to vendor-specific namespaces like tnsaxis:, tnsbosch:,
+// tnssamsung:. Matching is case-sensitive because ONVIF topic identifiers
+// are case-sensitive per the spec.
+func Classify(topic string) EventKind {
+	if topic == "" {
+		return KindUnknown
+	}
+	canonical := canonicalizeTopic(topic)
+	for _, rule := range topicRules {
+		if strings.Contains(canonical, rule.needle) {
+			return rule.kind
+		}
+	}
+	return KindUnknown
+}
+
+// canonicalizeTopic strips the XML-namespace prefix (e.g. "tns1:") from each
+// "/"-separated segment of the topic. This collapses vendor variants like
+// "tns1:Device/tnssamsung:DigitalInput" and the plain
+// "tns1:Device/DigitalInput" form to the same canonical path.
+func canonicalizeTopic(topic string) string {
+	segments := strings.Split(topic, "/")
+	for i, seg := range segments {
+		if idx := strings.Index(seg, ":"); idx >= 0 {
+			segments[i] = seg[idx+1:]
+		}
+	}
+	return strings.Join(segments, "/")
+}
+
+// topicRules is evaluated in order; first match wins. Keep the most specific
+// rules first when adding new entries — e.g. "MotionDetector/Motion" must
+// precede a hypothetical bare "/Motion" rule.
+var topicRules = []struct {
+	needle string
+	kind   EventKind
+}{
+	// Motion: covers AXIS VideoSource/MotionAlarm, ONVIF
+	// RuleEngine/CellMotionDetector and MotionRegionDetector, and
+	// vendor-namespaced MotionAlarm variants (Bosch).
+	{"MotionAlarm", KindMotion},
+	{"CellMotionDetector/Motion", KindMotion},
+	{"MotionRegionDetector/Motion", KindMotion},
+
+	// Tampering: ONVIF RuleEngine/TamperDetector.
+	{"TamperDetector", KindTampering},
+
+	// Digital I/O: ONVIF Device/Trigger/{DigitalInput,Relay}. The
+	// canonicalization step normalizes vendor-prefixed inner segments
+	// (tnssamsung:DigitalInput, tns1:Relay) to the bare names.
+	{"Trigger/DigitalInput", KindDigitalInput},
+	{"Trigger/Relay", KindDigitalOutput},
+
+	// Object analytics: AXIS ObjectAnalytics scenarios use dynamic suffixes
+	// (Device1ScenarioANY, Device1Scenario1, ...), so match the path prefix.
+	{"ObjectAnalytics/", KindObjectDetected},
+	{"ObjectsInside", KindObjectDetected},
+
+	// Audio: ONVIF AudioAnalytics/Audio/DetectedSound and AXIS
+	// AudioSource/TriggerLevel.
+	{"Audio/DetectedSound", KindAudioAlarm},
+	{"AudioSource/TriggerLevel", KindAudioAlarm},
+}
diff --git a/event/stream/topics_test.go b/event/stream/topics_test.go
new file mode 100644
index 0000000..1477d47
--- /dev/null
+++ b/event/stream/topics_test.go
@@ -0,0 +1,53 @@
+package stream
+
+import "testing"
+
+func TestClassifyTopic(t *testing.T) {
+	tests := []struct {
+		name  string
+		topic string
+		want  EventKind
+	}{
+		// AXIS: motion alarm on video source.
+		{"axis_video_source_motion", "tns1:VideoSource/MotionAlarm", KindMotion},
+		// AXIS / Hikvision / others: ONVIF cell-motion detector rule.
+		{"cell_motion_detector", "tns1:RuleEngine/CellMotionDetector/Motion", KindMotion},
+		// AXIS region motion.
+		{"motion_region_detector", "tns1:RuleEngine/MotionRegionDetector/Motion", KindMotion},
+		// Vendor-prefixed motion (e.g. Bosch).
+		{"bosch_motion", "tns1:VideoAnalytics/tnsbosch:MotionAlarm", KindMotion},
+		// Tampering / scene change.
+		{"tamper_detector", "tns1:RuleEngine/TamperDetector/Tamper", KindTampering},
+		{"axis_scene_tamper", "tns1:VideoSource/ImageTooDark/ImagingService", KindUnknown}, // not classified
+		// Digital input — vendor-namespaced variant from agent code.
+		{"digital_input", "tns1:Device/Trigger/DigitalInput", KindDigitalInput},
+		{"digital_input_samsung", "tns1:Device/tns1:Trigger/tnssamsung:DigitalInput", KindDigitalInput},
+		// Digital output / relay.
+		{"relay", "tns1:Device/Trigger/Relay", KindDigitalOutput},
+		{"relay_avigilon", "tns1:Device/tns1:Trigger/tns1:Relay", KindDigitalOutput},
+		// Analytics object detection (AXIS object analytics, generic motion analytics).
+		{"object_detected", "tns1:RuleEngine/MyRuleDetector/ObjectsInside", KindObjectDetected},
+		{"axis_object_analytics", "tnsaxis:CameraApplicationPlatform/ObjectAnalytics/Device1ScenarioANY", KindObjectDetected},
+		// Audio.
+		{"audio_alarm", "tns1:AudioAnalytics/Audio/DetectedSound", KindAudioAlarm},
+		{"axis_audio", "tnsaxis:AudioSource/TriggerLevel", KindAudioAlarm},
+		// Unknown — should not be force-classified.
+		{"empty", "", KindUnknown},
+		{"unknown_topic", "tns1:UserAlarm/IVA", KindUnknown},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := Classify(tc.topic); got != tc.want {
+				t.Errorf("Classify(%q) = %v, want %v", tc.topic, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestClassifyIsCaseSensitive(t *testing.T) {
+	// ONVIF topic names are case-sensitive per spec; we should not silently
+	// upper/lower-case. A lowercased topic must not match.
+	if got := Classify("tns1:videosource/motionalarm"); got != KindUnknown {
+		t.Errorf("Classify lowercase = %v, want KindUnknown (topics are case-sensitive)", got)
+	}
+}