feat: add MediaMTX process and reconciliation control (#25)

Sense/server/app/sense/adapters/mediamtx/client.go

@@ -0,0 +1,92 @@
+package mediamtx
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+)
+
+type Source struct {
+	Path     string
+	URI      string
+	Username string
+	Password string
+}
+type PathStatus struct {
+	Name          string `json:"name"`
+	Ready         bool   `json:"ready"`
+	Readers       int    `json:"readers"`
+	BytesReceived int64  `json:"bytes_received"`
+}
+type Controller interface {
+	Apply(context.Context, Source) error
+	Status(context.Context, string) (PathStatus, error)
+}
+type HTTPController struct {
+	base   string
+	client *http.Client
+}
+
+func NewHTTPController(base string) *HTTPController {
+	return &HTTPController{base: strings.TrimRight(base, "/"), client: &http.Client{Timeout: 5 * time.Second}}
+}
+func (c *HTTPController) Apply(ctx context.Context, source Source) error {
+	parsed, err := url.Parse(source.URI)
+	if err != nil || parsed.User != nil {
+		return fmt.Errorf("invalid credential-free media source")
+	}
+	if source.Username != "" {
+		parsed.User = url.UserPassword(source.Username, source.Password)
+	}
+	payload := map[string]any{"source": parsed.String(), "sourceOnDemand": true, "rtspTransport": "tcp"}
+	data, _ := json.Marshal(payload)
+	endpoint := c.base + "/v3/config/paths/replace/" + url.PathEscape(source.Path)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(data))
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Content-Type", "application/json")
+	res, err := c.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("mediamtx control unavailable: %w", err)
+	}
+	defer res.Body.Close()
+	io.Copy(io.Discard, io.LimitReader(res.Body, 1<<20))
+	if res.StatusCode < 200 || res.StatusCode >= 300 {
+		return fmt.Errorf("mediamtx apply returned %d", res.StatusCode)
+	}
+	return nil
+}
+func (c *HTTPController) Status(ctx context.Context, path string) (PathStatus, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+"/v3/paths/get/"+url.PathEscape(path), nil)
+	if err != nil {
+		return PathStatus{}, err
+	}
+	res, err := c.client.Do(req)
+	if err != nil {
+		return PathStatus{}, err
+	}
+	defer res.Body.Close()
+	if res.StatusCode == http.StatusNotFound {
+		return PathStatus{Name: path}, nil
+	}
+	if res.StatusCode < 200 || res.StatusCode >= 300 {
+		return PathStatus{}, fmt.Errorf("mediamtx status returned %d", res.StatusCode)
+	}
+	var raw struct {
+		Name          string `json:"name"`
+		Ready         bool   `json:"ready"`
+		BytesReceived int64  `json:"bytesReceived"`
+		Readers       []any  `json:"readers"`
+	}
+	if err := json.NewDecoder(io.LimitReader(res.Body, 1<<20)).Decode(&raw); err != nil {
+		return PathStatus{}, err
+	}
+	return PathStatus{Name: raw.Name, Ready: raw.Ready, Readers: len(raw.Readers), BytesReceived: raw.BytesReceived}, nil
+}

Sense/server/app/sense/adapters/mediamtx/client_test.go

@@ -0,0 +1,37 @@
+package mediamtx
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestApplyBuildsCredentialSourceOnlyInTransientBody(t *testing.T) {
+	var body map[string]any
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.Contains(r.URL.String(), "password") {
+			t.Fatal("credential leaked in control URL")
+		}
+		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+			t.Fatal(err)
+		}
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer server.Close()
+	client := NewHTTPController(server.URL)
+	if err := client.Apply(context.Background(), Source{Path: "sense_test", URI: "rtsp://camera.invalid/main", Username: "fixture-user", Password: "fixture-password"}); err != nil {
+		t.Fatal(err)
+	}
+	if body["source"] != "rtsp://fixture-user:fixture-password@camera.invalid/main" {
+		t.Fatalf("body=%#v", body)
+	}
+}
+func TestApplyRejectsCredentialURI(t *testing.T) {
+	client := NewHTTPController("http://127.0.0.1")
+	if err := client.Apply(context.Background(), Source{Path: "x", URI: "rtsp://user:pass@camera.invalid/main"}); err == nil {
+		t.Fatal("credential URI accepted")
+	}
+}

Sense/server/app/sense/adapters/mediamtx/supervisor.go

@@ -0,0 +1,113 @@
+package mediamtx
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/exec"
+	"sync"
+	"time"
+)
+
+type ProcessState struct {
+	Running  bool   `json:"running"`
+	PID      int    `json:"pid,omitempty"`
+	Restarts int    `json:"restarts"`
+	LastExit string `json:"last_exit,omitempty"`
+}
+type Process interface {
+	Start(context.Context) error
+	Stop(context.Context) error
+	State() ProcessState
+}
+type Supervisor struct {
+	binary      string
+	config      string
+	maxRestarts int
+	mu          sync.Mutex
+	command     *exec.Cmd
+	state       ProcessState
+}
+
+func NewSupervisor(binary, config string, maxRestarts int) *Supervisor {
+	if maxRestarts < 0 {
+		maxRestarts = 0
+	}
+	return &Supervisor{binary: binary, config: config, maxRestarts: maxRestarts}
+}
+func (s *Supervisor) Start(ctx context.Context) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.state.Running {
+		return nil
+	}
+	if s.binary == "" {
+		return fmt.Errorf("SENSE_MEDIAMTX_BINARY is not configured")
+	}
+	if s.state.LastExit != "" {
+		if s.state.Restarts >= s.maxRestarts {
+			return fmt.Errorf("MediaMTX restart limit reached")
+		}
+		s.state.Restarts++
+	}
+	arguments := []string{}
+	if s.config != "" {
+		arguments = append(arguments, s.config)
+	}
+	command := exec.CommandContext(context.Background(), s.binary, arguments...)
+	if err := command.Start(); err != nil {
+		return fmt.Errorf("start MediaMTX: %w", err)
+	}
+	s.command = command
+	s.state.Running = true
+	s.state.PID = command.Process.Pid
+	s.state.LastExit = ""
+	go s.wait(command)
+	return nil
+}
+func (s *Supervisor) wait(command *exec.Cmd) {
+	err := command.Wait()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.command != command {
+		return
+	}
+	s.state.Running = false
+	s.state.PID = 0
+	if err != nil {
+		s.state.LastExit = err.Error()
+	} else {
+		s.state.LastExit = "exited"
+	}
+}
+func (s *Supervisor) Stop(ctx context.Context) error {
+	s.mu.Lock()
+	command := s.command
+	if command == nil || !s.state.Running {
+		s.mu.Unlock()
+		return nil
+	}
+	process := command.Process
+	s.mu.Unlock()
+	if err := process.Signal(os.Interrupt); err != nil {
+		if killErr := process.Kill(); killErr != nil {
+			return fmt.Errorf("stop owned MediaMTX process: %w", killErr)
+		}
+	}
+	ticker := time.NewTicker(50 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			s.mu.Lock()
+			running := s.state.Running
+			s.mu.Unlock()
+			if !running {
+				return nil
+			}
+		}
+	}
+}
+func (s *Supervisor) State() ProcessState { s.mu.Lock(); defer s.mu.Unlock(); return s.state }

Sense/server/app/sense/admission/http.go

@@ -10,6 +10,7 @@ type Module struct{ service *Service }
 
 func NewModule(service *Service) *Module { return &Module{service: service} }
 func (m *Module) Register(app *platform.App) {
+	setCurrentService(m.service)
 	app.Handle("GET /api/v1/admission/discover", identity.Require(identity.PermissionDeviceWrite, http.HandlerFunc(m.discover)))
 	app.Handle("POST /api/v1/admission/probe", identity.Require(identity.PermissionDeviceWrite, http.HandlerFunc(m.probe)))
 	app.Handle("GET /api/v1/admission/devices/{id}", identity.Require(identity.PermissionDeviceRead, http.HandlerFunc(m.get)))

Sense/server/app/sense/admission/profile_port.go

@@ -0,0 +1,24 @@
+package admission
+
+import "fmt"
+
+var currentService *Service
+
+func setCurrentService(service *Service) { currentService = service }
+
+// VerifiedProfile is a Sense-internal projection, not a cross-product contract.
+func VerifiedProfile(deviceID, token string) (Profile, error) {
+	if currentService == nil {
+		return Profile{}, fmt.Errorf("admission service is not ready")
+	}
+	result, ok := currentService.Get(deviceID)
+	if !ok {
+		return Profile{}, fmt.Errorf("device admission has not been verified")
+	}
+	for _, profile := range result.Profiles {
+		if profile.Token == token && profile.Verification.Status == "ready" {
+			return profile, nil
+		}
+	}
+	return Profile{}, fmt.Errorf("verified profile not found")
+}

Sense/server/app/sense/media/http.go

@@ -0,0 +1,64 @@
+package media
+
+import (
+	"net/http"
+	"yovision.local/sense/app/sense/identity"
+	"yovision.local/sense/internal/platform"
+)
+
+type Module struct{ service *Service }
+
+func NewModule(service *Service) *Module { return &Module{service: service} }
+func (m *Module) Register(app *platform.App) {
+	app.Handle("GET /api/v1/media/routes", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.list)))
+	app.Handle("POST /api/v1/media/routes", identity.Require(identity.PermissionMediaWrite, http.HandlerFunc(m.configure)))
+	app.Handle("POST /api/v1/media/routes/{id}/reconcile", identity.Require(identity.PermissionMediaWrite, http.HandlerFunc(m.reconcile)))
+	app.Handle("POST /api/v1/media/routes/{id}/stop", identity.Require(identity.PermissionMediaWrite, http.HandlerFunc(m.stop)))
+	app.Handle("GET /api/v1/media/process", identity.Require(identity.PermissionMediaRead, http.HandlerFunc(m.process)))
+}
+func (m *Module) list(w http.ResponseWriter, r *http.Request) {
+	items, err := m.service.List(r.Context())
+	if err != nil {
+		platform.WriteError(w, err)
+		return
+	}
+	platform.WriteJSON(w, http.StatusOK, map[string]any{"items": items, "total": len(items)})
+}
+func (m *Module) configure(w http.ResponseWriter, r *http.Request) {
+	var req struct {
+		DeviceID     string `json:"device_id"`
+		ProfileToken string `json:"profile_token"`
+	}
+	if err := platform.DecodeJSON(r, &req); err != nil {
+		platform.WriteError(w, err)
+		return
+	}
+	actor, _ := identity.PrincipalFromContext(r.Context())
+	item, err := m.service.Configure(r.Context(), actor, req.DeviceID, req.ProfileToken)
+	if err != nil {
+		platform.WriteError(w, &platform.APIError{Status: http.StatusBadRequest, Code: "media_invalid", Message: err.Error()})
+		return
+	}
+	platform.WriteJSON(w, http.StatusCreated, item)
+}
+func (m *Module) reconcile(w http.ResponseWriter, r *http.Request) {
+	actor, _ := identity.PrincipalFromContext(r.Context())
+	item, err := m.service.Reconcile(r.Context(), actor, r.PathValue("id"))
+	if err != nil {
+		platform.WriteError(w, &platform.APIError{Status: http.StatusBadRequest, Code: "media_reconcile_failed", Message: err.Error()})
+		return
+	}
+	platform.WriteJSON(w, http.StatusOK, item)
+}
+func (m *Module) stop(w http.ResponseWriter, r *http.Request) {
+	actor, _ := identity.PrincipalFromContext(r.Context())
+	item, err := m.service.Stop(r.Context(), actor, r.PathValue("id"))
+	if err != nil {
+		platform.WriteError(w, err)
+		return
+	}
+	platform.WriteJSON(w, http.StatusOK, item)
+}
+func (m *Module) process(w http.ResponseWriter, _ *http.Request) {
+	platform.WriteJSON(w, http.StatusOK, m.service.ProcessState())
+}

Sense/server/app/sense/media/migration.go

@@ -0,0 +1,3 @@
+package media
+
+const MigrationSQL = `CREATE TABLE IF NOT EXISTS sense_media_routes(id TEXT PRIMARY KEY,device_id TEXT NOT NULL,profile_token TEXT NOT NULL,path TEXT NOT NULL UNIQUE,desired TEXT NOT NULL,actual TEXT NOT NULL,readers INTEGER NOT NULL DEFAULT 0,detail TEXT NOT NULL DEFAULT '',version BIGINT NOT NULL,updated_at TIMESTAMPTZ NOT NULL);`

Sense/server/app/sense/media/service.go

@@ -0,0 +1,140 @@
+package media
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"yovision.local/sense/app/sense/adapters/mediamtx"
+	"yovision.local/sense/app/sense/admission"
+	"yovision.local/sense/app/sense/device"
+	"yovision.local/sense/app/sense/identity"
+)
+
+type Service struct {
+	store      Store
+	process    mediamtx.Process
+	controller mediamtx.Controller
+	profile    func(string, string) (admission.Profile, error)
+	credential func(context.Context, string) (device.Credential, error)
+	now        func() time.Time
+}
+
+func NewService(store Store, process mediamtx.Process, controller mediamtx.Controller) *Service {
+	return &Service{store: store, process: process, controller: controller, profile: admission.VerifiedProfile, credential: device.ReadCredential, now: time.Now}
+}
+func (s *Service) Configure(ctx context.Context, actor identity.Principal, deviceID, profileToken string) (Route, error) {
+	if _, err := s.profile(deviceID, profileToken); err != nil {
+		return Route{}, err
+	}
+	id := deviceID + ":" + profileToken
+	path := "sense_" + safe(deviceID) + "_" + safe(profileToken)
+	now := s.now().UTC()
+	route := Route{ID: id, DeviceID: deviceID, ProfileToken: profileToken, Path: path, Desired: "running", Actual: "pending", Detail: "等待状态对账", Version: 1, UpdatedAt: now}
+	if previous, err := s.store.Get(ctx, id); err == nil {
+		route.Version = previous.Version + 1
+	}
+	if err := s.store.Save(ctx, route); err != nil {
+		return Route{}, err
+	}
+	identity.RecordAudit(ctx, actor.UserID, "media.configure", id, "success", map[string]any{"path": path})
+	return route, nil
+}
+func (s *Service) Reconcile(ctx context.Context, actor identity.Principal, id string) (Route, error) {
+	route, err := s.store.Get(ctx, id)
+	if err != nil {
+		return Route{}, err
+	}
+	if route.Desired == "stopped" {
+		route.Actual = "stopped"
+		route.Detail = "已停止"
+		route.Readers = 0
+		route.Version++
+		route.UpdatedAt = s.now().UTC()
+		return route, s.store.Save(ctx, route)
+	}
+	if err := s.process.Start(ctx); err != nil {
+		route.Actual = "process_failed"
+		route.Detail = err.Error()
+		route.Version++
+		route.UpdatedAt = s.now().UTC()
+		_ = s.store.Save(ctx, route)
+		return route, nil
+	}
+	profile, err := s.profile(route.DeviceID, route.ProfileToken)
+	if err != nil {
+		return Route{}, err
+	}
+	credential, err := s.credential(ctx, route.DeviceID)
+	if err != nil {
+		return Route{}, err
+	}
+	source := mediamtx.Source{Path: route.Path, URI: profile.StreamURI, Username: credential.Username, Password: credential.Password}
+	if err := s.controller.Apply(ctx, source); err != nil {
+		route.Actual = "apply_failed"
+		route.Detail = err.Error()
+		route.Version++
+		route.UpdatedAt = s.now().UTC()
+		_ = s.store.Save(ctx, route)
+		return route, nil
+	}
+	status, err := s.controller.Status(ctx, route.Path)
+	if err != nil {
+		route.Actual = "unconverged"
+		route.Detail = "尚未取得媒体状态"
+	} else if status.Ready {
+		route.Actual = "ready"
+		route.Detail = "上游拉流正常"
+		route.Readers = status.Readers
+	} else {
+		route.Actual = "waiting"
+		route.Detail = "等待上游拉流"
+		route.Readers = status.Readers
+	}
+	route.Version++
+	route.UpdatedAt = s.now().UTC()
+	if err := s.store.Save(ctx, route); err != nil {
+		return Route{}, err
+	}
+	identity.RecordAudit(ctx, actor.UserID, "media.reconcile", id, "success", map[string]any{"actual": route.Actual})
+	return route, nil
+}
+func (s *Service) Stop(ctx context.Context, actor identity.Principal, id string) (Route, error) {
+	route, err := s.store.Get(ctx, id)
+	if err != nil {
+		return Route{}, err
+	}
+	route.Desired = "stopped"
+	route.Actual = "stopped"
+	route.Readers = 0
+	route.Detail = "已停止"
+	route.Version++
+	route.UpdatedAt = s.now().UTC()
+	if err := s.store.Save(ctx, route); err != nil {
+		return Route{}, err
+	}
+	items, _ := s.store.List(ctx)
+	allStopped := true
+	for _, item := range items {
+		if item.ID != id && item.Desired == "running" {
+			allStopped = false
+		}
+	}
+	if allStopped {
+		stopCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+		defer cancel()
+		_ = s.process.Stop(stopCtx)
+	}
+	identity.RecordAudit(ctx, actor.UserID, "media.stop", id, "success", nil)
+	return route, nil
+}
+func (s *Service) List(ctx context.Context) ([]Route, error) { return s.store.List(ctx) }
+func (s *Service) ProcessState() mediamtx.ProcessState       { return s.process.State() }
+func safe(value string) string {
+	return strings.Map(func(r rune) rune {
+		if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '_' || r == '-' {
+			return r
+		}
+		return '_'
+	}, value)
+}

Sense/server/app/sense/media/service_test.go

@@ -0,0 +1,110 @@
+package media
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"yovision.local/sense/app/sense/adapters/mediamtx"
+	"yovision.local/sense/app/sense/admission"
+	"yovision.local/sense/app/sense/device"
+	"yovision.local/sense/app/sense/identity"
+)
+
+type fakeProcess struct {
+	state    mediamtx.ProcessState
+	startErr error
+	stopped  bool
+}
+
+func (p *fakeProcess) Start(context.Context) error {
+	if p.startErr != nil {
+		return p.startErr
+	}
+	p.state.Running = true
+	p.state.PID = 123
+	return nil
+}
+func (p *fakeProcess) Stop(context.Context) error {
+	p.stopped = true
+	p.state.Running = false
+	p.state.PID = 0
+	return nil
+}
+func (p *fakeProcess) State() mediamtx.ProcessState { return p.state }
+
+type fakeController struct {
+	applyErr error
+	status   mediamtx.PathStatus
+}
+
+func (f fakeController) Apply(context.Context, mediamtx.Source) error { return f.applyErr }
+func (f fakeController) Status(context.Context, string) (mediamtx.PathStatus, error) {
+	if f.status.Name == "error" {
+		return mediamtx.PathStatus{}, errors.New("timeout")
+	}
+	if !f.status.Ready {
+		return mediamtx.PathStatus{Ready: true, Readers: 2}, nil
+	}
+	return f.status, nil
+}
+
+func TestStoreAndProcessBoundary(t *testing.T) {
+	store := NewMemoryStore()
+	process := &fakeProcess{}
+	service := NewService(store, process, fakeController{})
+	route := Route{ID: "dev:main", DeviceID: "dev", ProfileToken: "main", Path: "sense_dev_main", Desired: "running", Actual: "pending", Version: 1}
+	if err := store.Save(context.Background(), route); err != nil {
+		t.Fatal(err)
+	}
+	items, err := service.List(context.Background())
+	if err != nil || len(items) != 1 {
+		t.Fatalf("items=%#v err=%v", items, err)
+	}
+	if process.State().Running {
+		t.Fatal("process started without explicit reconcile")
+	}
+}
+
+func preparedService(t *testing.T, process *fakeProcess, controller fakeController) (*Service, Route) {
+	t.Helper()
+	service := NewService(NewMemoryStore(), process, controller)
+	service.profile = func(string, string) (admission.Profile, error) {
+		return admission.Profile{StreamURI: "rtsp://camera.invalid/main"}, nil
+	}
+	service.credential = func(context.Context, string) (device.Credential, error) {
+		return device.Credential{Username: "fixture", Password: "fixture"}, nil
+	}
+	route, err := service.Configure(context.Background(), identity.Principal{}, "device", "main")
+	if err != nil {
+		t.Fatal(err)
+	}
+	return service, route
+}
+
+func TestReconcileReadyAndStopOwnedProcess(t *testing.T) {
+	process := &fakeProcess{}
+	service, route := preparedService(t, process, fakeController{status: mediamtx.PathStatus{Ready: true, Readers: 2}})
+	result, err := service.Reconcile(context.Background(), identity.Principal{}, route.ID)
+	if err != nil || result.Actual != "ready" || result.Readers != 2 {
+		t.Fatalf("result=%#v err=%v", result, err)
+	}
+	if _, err := service.Stop(context.Background(), identity.Principal{}, route.ID); err != nil || !process.stopped {
+		t.Fatalf("stop err=%v stopped=%v", err, process.stopped)
+	}
+}
+
+func TestReconcileRecordsFailureStates(t *testing.T) {
+	process := &fakeProcess{startErr: errors.New("restart limit reached")}
+	service, route := preparedService(t, process, fakeController{})
+	result, err := service.Reconcile(context.Background(), identity.Principal{}, route.ID)
+	if err != nil || result.Actual != "process_failed" {
+		t.Fatalf("result=%#v err=%v", result, err)
+	}
+	process.startErr = nil
+	service.controller = fakeController{applyErr: errors.New("control timeout")}
+	result, err = service.Reconcile(context.Background(), identity.Principal{}, route.ID)
+	if err != nil || result.Actual != "apply_failed" {
+		t.Fatalf("result=%#v err=%v", result, err)
+	}
+}

Sense/server/app/sense/media/store.go

@@ -0,0 +1,78 @@
+package media
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"sync"
+)
+
+var ErrNotFound = errors.New("media route not found")
+
+type Store interface {
+	Save(context.Context, Route) error
+	Get(context.Context, string) (Route, error)
+	List(context.Context) ([]Route, error)
+}
+type MemoryStore struct {
+	mu    sync.RWMutex
+	items map[string]Route
+}
+
+func NewMemoryStore() *MemoryStore { return &MemoryStore{items: map[string]Route{}} }
+func (s *MemoryStore) Save(_ context.Context, item Route) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.items[item.ID] = item
+	return nil
+}
+func (s *MemoryStore) Get(_ context.Context, id string) (Route, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	item, ok := s.items[id]
+	if !ok {
+		return Route{}, ErrNotFound
+	}
+	return item, nil
+}
+func (s *MemoryStore) List(_ context.Context) ([]Route, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	items := make([]Route, 0, len(s.items))
+	for _, item := range s.items {
+		items = append(items, item)
+	}
+	return items, nil
+}
+
+type PostgresStore struct{ database *sql.DB }
+
+func NewPostgresStore(db *sql.DB) *PostgresStore { return &PostgresStore{database: db} }
+func (s *PostgresStore) Save(ctx context.Context, item Route) error {
+	_, err := s.database.ExecContext(ctx, `INSERT INTO sense_media_routes(id,device_id,profile_token,path,desired,actual,readers,detail,version,updated_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT(id) DO UPDATE SET desired=EXCLUDED.desired,actual=EXCLUDED.actual,readers=EXCLUDED.readers,detail=EXCLUDED.detail,version=EXCLUDED.version,updated_at=EXCLUDED.updated_at`, item.ID, item.DeviceID, item.ProfileToken, item.Path, item.Desired, item.Actual, item.Readers, item.Detail, item.Version, item.UpdatedAt)
+	return err
+}
+func (s *PostgresStore) Get(ctx context.Context, id string) (Route, error) {
+	var item Route
+	err := s.database.QueryRowContext(ctx, `SELECT id,device_id,profile_token,path,desired,actual,readers,detail,version,updated_at FROM sense_media_routes WHERE id=$1`, id).Scan(&item.ID, &item.DeviceID, &item.ProfileToken, &item.Path, &item.Desired, &item.Actual, &item.Readers, &item.Detail, &item.Version, &item.UpdatedAt)
+	if errors.Is(err, sql.ErrNoRows) {
+		return Route{}, ErrNotFound
+	}
+	return item, err
+}
+func (s *PostgresStore) List(ctx context.Context) ([]Route, error) {
+	rows, err := s.database.QueryContext(ctx, `SELECT id,device_id,profile_token,path,desired,actual,readers,detail,version,updated_at FROM sense_media_routes ORDER BY updated_at DESC`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []Route
+	for rows.Next() {
+		var item Route
+		if err := rows.Scan(&item.ID, &item.DeviceID, &item.ProfileToken, &item.Path, &item.Desired, &item.Actual, &item.Readers, &item.Detail, &item.Version, &item.UpdatedAt); err != nil {
+			return nil, err
+		}
+		items = append(items, item)
+	}
+	return items, rows.Err()
+}

Sense/server/app/sense/media/types.go

@@ -0,0 +1,16 @@
+package media
+
+import "time"
+
+type Route struct {
+	ID           string    `json:"id"`
+	DeviceID     string    `json:"device_id"`
+	ProfileToken string    `json:"profile_token"`
+	Path         string    `json:"path"`
+	Desired      string    `json:"desired"`
+	Actual       string    `json:"actual"`
+	Readers      int       `json:"readers"`
+	Detail       string    `json:"detail"`
+	Version      int64     `json:"version"`
+	UpdatedAt    time.Time `json:"updated_at"`
+}

Sense/server/cmd/sense/modules_media.go

@@ -0,0 +1,30 @@
+package sense
+
+import (
+	"os"
+	"yovision.local/sense/app/sense/adapters/mediamtx"
+	"yovision.local/sense/app/sense/media"
+	"yovision.local/sense/internal/platform"
+)
+
+func init() {
+	registerModule(func(app *platform.App) error {
+		var store media.Store
+		if app.Config().DatabaseMode == platform.DatabaseModeMemory {
+			store = media.NewMemoryStore()
+		} else {
+			store = media.NewPostgresStore(app.Database())
+		}
+		process := mediamtx.NewSupervisor(os.Getenv("SENSE_MEDIAMTX_BINARY"), os.Getenv("SENSE_MEDIAMTX_CONFIG"), 3)
+		controller := mediamtx.NewHTTPController(valueOr("SENSE_MEDIAMTX_API", "http://127.0.0.1:9997"))
+		app.RegisterMigration(platform.Migration{Version: 2026081203, Name: "sense_media", SQL: media.MigrationSQL})
+		media.NewModule(media.NewService(store, process, controller)).Register(app)
+		return nil
+	})
+}
+func valueOr(key, fallback string) string {
+	if value := os.Getenv(key); value != "" {
+		return value
+	}
+	return fallback
+}

Sense/server/database/migrations/sense_media/2026081203_media.sql

@@ -0,0 +1,2 @@
+CREATE TABLE IF NOT EXISTS sense_media_routes(id TEXT PRIMARY KEY,device_id TEXT NOT NULL,profile_token TEXT NOT NULL,path TEXT NOT NULL UNIQUE,desired TEXT NOT NULL,actual TEXT NOT NULL,readers INTEGER NOT NULL DEFAULT 0,detail TEXT NOT NULL DEFAULT '',version BIGINT NOT NULL,updated_at TIMESTAMPTZ NOT NULL);
+

Sense/testdata/mediamtx/path-status.json

@@ -0,0 +1 @@
+{"name":"sense_fixture_main","ready":true,"bytesReceived":4096,"readers":[{"type":"webrtc"}]}

Sense/tests/media/README.md

@@ -0,0 +1,6 @@
+# Media acceptance
+
+Tests keep MediaMTX behind process and control interfaces. They verify that
+the credential-bearing MediaMTX control request is localhost-only and transient,
+never returned or persisted by Sense, and no process starts before an explicit reconcile. Real MediaMTX binary,
+camera pull, reader counts and crash recovery require the deployment lab.

Sense/ui/src/api/sense/media/index.js

@@ -0,0 +1,7 @@
+import request from '../../../bootstrap/request'
+export const listRoutes = () => request.get('/media/routes')
+export const configureRoute = (data) => request.post('/media/routes', data)
+export const reconcileRoute = (id) => request.post(`/media/routes/${encodeURIComponent(id)}/reconcile`)
+export const stopRoute = (id) => request.post(`/media/routes/${encodeURIComponent(id)}/stop`)
+export const processState = () => request.get('/media/process')
+

Sense/ui/src/router/modules/sense-media.js

@@ -0,0 +1,3 @@
+import Media from '../../views/sense/media/Media.vue'
+export default { path: 'media', name: 'sense-media', component: Media, meta: { title: '视频服务', icon: 'Monitor', order: 30, permission: 'media.read' } }
+

Sense/ui/src/views/sense/media/Media.vue

@@ -0,0 +1,38 @@
+<template>
+  <section class="page-container">
+    <div class="page-heading"><div><h1>视频服务</h1><p>管理 MediaMTX 独立进程和已验证视频的期望状态。</p></div><el-tag :type="process.running ? 'success' : 'info'">{{ process.running ? `进程运行中 · PID ${process.pid}` : '进程未运行' }}</el-tag></div>
+    <el-alert v-if="process.last_exit" :title="`最近退出：${process.last_exit}`" type="warning" show-icon :closable="false" class="process-alert" />
+    <div class="table-container">
+      <div class="table-actions"><strong>媒体路径</strong><el-button v-if="canWrite" type="primary" :icon="Plus" @click="dialog=true">添加视频路径</el-button></div>
+      <el-table v-loading="loading" :data="routes" border>
+        <el-table-column prop="device_id" label="设备" min-width="190" />
+        <el-table-column prop="profile_token" label="Profile" min-width="120" />
+        <el-table-column prop="path" label="媒体路径" min-width="210" />
+        <el-table-column label="期望" width="100"><template #default="scope"><el-tag>{{ scope.row.desired === 'running' ? '运行' : '停止' }}</el-tag></template></el-table-column>
+        <el-table-column label="实际状态" width="130"><template #default="scope"><el-tag :type="scope.row.actual === 'ready' ? 'success' : scope.row.actual.includes('failed') ? 'danger' : 'warning'">{{ stateLabel(scope.row.actual) }}</el-tag></template></el-table-column>
+        <el-table-column prop="readers" label="观看数" width="90" />
+        <el-table-column prop="detail" label="说明" min-width="180" />
+        <el-table-column v-if="canWrite" label="操作" width="150" fixed="right"><template #default="scope"><el-button link type="primary" @click="reconcile(scope.row)">对账</el-button><el-button v-if="scope.row.desired === 'running'" link type="danger" @click="stop(scope.row)">停止</el-button></template></el-table-column>
+      </el-table>
+    </div>
+    <el-dialog v-model="dialog" title="添加视频路径" width="520px">
+      <el-form ref="formRef" :model="form" :rules="rules" label-position="top">
+        <el-form-item label="已完成接入检查的设备" prop="device_id"><el-select v-model="form.device_id" filterable style="width:100%" @change="loadProfiles"><el-option v-for="item in devices" :key="item.id" :label="item.name" :value="item.id" /></el-select></el-form-item>
+        <el-form-item label="视频 Profile" prop="profile_token"><el-select v-model="form.profile_token" style="width:100%"><el-option v-for="item in profiles" :key="item.token" :label="`${item.kind === 'main' ? '主码流' : '子码流'} · ${item.width}×${item.height}`" :value="item.token" /></el-select></el-form-item>
+      </el-form>
+      <template #footer><el-button @click="dialog=false">取消</el-button><el-button type="primary" :loading="saving" @click="save">保存并等待对账</el-button></template>
+    </el-dialog>
+  </section>
+</template>
+<script setup>
+import{computed,onMounted,reactive,ref}from'vue';import{ElMessage}from'element-plus';import{Plus}from'@element-plus/icons-vue';import{useStore}from'vuex';import{listDevices}from'../../../api/sense/device';import{admissionResult}from'../../../api/sense/admission';import{configureRoute,listRoutes,processState,reconcileRoute,stopRoute}from'../../../api/sense/media'
+const store=useStore(),loading=ref(false),saving=ref(false),dialog=ref(false),formRef=ref(),routes=ref([]),devices=ref([]),profiles=ref([]),process=reactive({running:false});const form=reactive({device_id:'',profile_token:''});const rules={device_id:[{required:true,message:'请选择设备',trigger:'change'}],profile_token:[{required:true,message:'请选择 Profile',trigger:'change'}]};const canWrite=computed(()=>store.getters['sense-identity/hasPermission']?.('media.write'))
+function stateLabel(value){return({pending:'等待对账',ready:'拉流正常',waiting:'等待拉流',stopped:'已停止',process_failed:'进程启动失败',apply_failed:'配置失败',unconverged:'未收敛'})[value]||value}
+async function load(){loading.value=true;try{const[data,state,devicePage]=await Promise.all([listRoutes(),processState(),listDevices({page:1,page_size:100})]);routes.value=data.items||[];Object.assign(process,state);devices.value=devicePage.items.filter(item=>item.modality==='video'&&item.status!=='disabled')}finally{loading.value=false}}
+async function loadProfiles(){form.profile_token='';profiles.value=[];if(!form.device_id)return;try{profiles.value=(await admissionResult(form.device_id)).profiles.filter(item=>item.verification.status==='ready')}catch{ElMessage.warning('该设备尚未完成视频接入检查')}}
+async function save(){const valid=await formRef.value?.validate().catch(()=>false);if(!valid)return;saving.value=true;try{await configureRoute(form);ElMessage.success('媒体路径已保存，请执行对账');dialog.value=false;await load()}catch(error){ElMessage.error(error.message||'保存失败')}finally{saving.value=false}}
+async function reconcile(row){await reconcileRoute(row.id);ElMessage.success('已完成一次状态对账');await load()}
+async function stop(row){await stopRoute(row.id);ElMessage.success('媒体路径已停止');await load()}
+onMounted(load)
+</script>
+<style scoped>.process-alert{margin-bottom:16px}</style>