Skip to content

Commit 789a59f

Browse files
PoppaBuzzPoppaBuzz
committed
Terminal push-based output + vchiq system info package
Terminal: push-based output via WebSocket (ws.Send, session mutex, StartOutputLoop goroutine). Android: remove fake prompt, simplify input handler, add session_id to resize. Device: add 8 new BPP handlers powered by vchiq package adapted from RaspConnect (CPU, throttling, GPU, voltage, processes, USB, OS info, device model). Platform-safe with Linux build tags. Fix test files for NewService signature, fix auth_test vet warning.
1 parent ba30e18 commit 789a59f

30 files changed

Lines changed: 1042 additions & 84 deletions

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ agent/all_cover
2020
.omo/
2121
*.swp
2222
*.swo
23+
problems/
2324

2425
# OS
2526
.DS_Store

‎agent/cmd/runtime/e2e_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,7 @@ func TestE2E_FullStack(t *testing.T) {
106106
// Device service
107107
cfg, _ := config.Load("")
108108
cfg.Runtime.DeviceName = "e2e-test-device"
109-
devSvc, err := device.NewService(cfg, devID.DeviceID, logger)
109+
devSvc, err := device.NewService(cfg, "", devID.DeviceID, logger)
110110
if err != nil {
111111
t.Fatalf("new device service: %v", err)
112112
}

‎agent/cmd/runtime/main.go‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -116,6 +116,14 @@ func main() {
116116
eng.RegisterMethod("device.info", devSvc.HandleInfo)
117117
eng.RegisterMethod("device.stats", devSvc.HandleStats)
118118
eng.RegisterMethod("device.rename", devSvc.HandleRename)
119+
eng.RegisterMethod("device.cpu", devSvc.HandleCPU)
120+
eng.RegisterMethod("device.throttling", devSvc.HandleThrottling)
121+
eng.RegisterMethod("device.gpu", devSvc.HandleGPU)
122+
eng.RegisterMethod("device.voltage", devSvc.HandleVoltage)
123+
eng.RegisterMethod("device.processes", devSvc.HandleProcesses)
124+
eng.RegisterMethod("device.usb", devSvc.HandleUSB)
125+
eng.RegisterMethod("device.os", devSvc.HandleOS)
126+
eng.RegisterMethod("device.model", devSvc.HandleModel)
119127

120128
pairH := pairing.NewHandler(store, devID.DeviceID, logger)
121129
eng.RegisterMethod("pair.initiate", pairH.HandleInitiate)
@@ -281,6 +289,14 @@ func (h *bppWebSocketHandler) HandleMessage(ctx context.Context, data []byte) ([
281289
ctx = context.WithValue(ctx, engine.CtxSessionToken, session)
282290
ctx = context.WithValue(ctx, engine.CtxClientID, clientID)
283291

292+
conn, _ := ctx.Value(ws.ContextKeyConn).(*ws.Connection)
293+
if conn != nil {
294+
sender := func(data []byte) error {
295+
return conn.Send(data)
296+
}
297+
ctx = context.WithValue(ctx, terminal.SenderKey, sender)
298+
}
299+
284300
resp, err := h.engine.Handle(ctx, env)
285301
if err != nil {
286302
h.log.Error("engine handler error", "error", err)

‎agent/internal/device/device.go‎

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515
"time"
1616

1717
"github.com/buzzpi/agent/internal/config"
18+
"github.com/buzzpi/agent/internal/vchiq"
1819
"github.com/buzzpi/agent/internal/version"
1920
)
2021

@@ -209,3 +210,177 @@ func (s *Service) Health() interface{} {
209210
"uptime_ms": time.Since(s.startTime).Milliseconds(),
210211
}
211212
}
213+
214+
// CPUInfoResponse is the response for device.cpu.
215+
type CPUInfoResponse struct {
216+
Model string `json:"model"`
217+
Cores string `json:"cores"`
218+
MHz string `json:"mhz"`
219+
Temperature string `json:"temperature"`
220+
Frequency float64 `json:"frequency_mhz"`
221+
UsagePercent float64 `json:"usage_percent"`
222+
Serial string `json:"serial,omitempty"`
223+
Revision string `json:"revision,omitempty"`
224+
Info []vchiq.LscpuField `json:"info,omitempty"`
225+
}
226+
227+
func (s *Service) HandleCPU(ctx context.Context, params json.RawMessage) (interface{}, error) {
228+
cpuUsage, _, _ := s.readCPUStats()
229+
230+
model, _ := vchiq.GetCPUModel()
231+
cores, _ := vchiq.GetCPUCores()
232+
mhz, _ := vchiq.GetCPUMHz()
233+
temp, _ := vchiq.GetCPUTemperature()
234+
freq, _ := vchiq.GetCPUFrequency()
235+
serial, _ := vchiq.GetCPUSerial()
236+
revision, _ := vchiq.GetCPURevision()
237+
info, _ := vchiq.GetCPUInfo()
238+
239+
return &CPUInfoResponse{
240+
Model: model,
241+
Cores: cores,
242+
MHz: mhz,
243+
Temperature: temp,
244+
Frequency: freq,
245+
UsagePercent: cpuUsage,
246+
Serial: serial,
247+
Revision: revision,
248+
Info: info,
249+
}, nil
250+
}
251+
252+
// ThrottlingResponse is the response for device.throttling.
253+
type ThrottlingResponse struct {
254+
Throttled int64 `json:"throttled"`
255+
ThrottledInfo string `json:"throttled_info"`
256+
}
257+
258+
func (s *Service) HandleThrottling(ctx context.Context, params json.RawMessage) (interface{}, error) {
259+
throttled, err := vchiq.GetThrottled()
260+
if err != nil {
261+
return nil, fmt.Errorf("get throttled: %w", err)
262+
}
263+
info, err := vchiq.GetThrottledInfo()
264+
if err != nil {
265+
return nil, fmt.Errorf("get throttled info: %w", err)
266+
}
267+
return &ThrottlingResponse{
268+
Throttled: throttled,
269+
ThrottledInfo: info,
270+
}, nil
271+
}
272+
273+
// GPUDetailsResponse is the response for device.gpu.
274+
type GPUDetailsResponse struct {
275+
GPUTemperature string `json:"gpu_temperature"`
276+
ARM string `json:"arm_memory"`
277+
GPU string `json:"gpu_memory"`
278+
}
279+
280+
func (s *Service) HandleGPU(ctx context.Context, params json.RawMessage) (interface{}, error) {
281+
temp, _ := vchiq.GetGPUTemperature()
282+
arm, gpu, _ := vchiq.GetVCGencmdMemory()
283+
return &GPUDetailsResponse{
284+
GPUTemperature: temp,
285+
ARM: arm,
286+
GPU: gpu,
287+
}, nil
288+
}
289+
290+
// VoltageResponse is the response for device.voltage.
291+
type VoltageResponse struct {
292+
CoreVoltage string `json:"core_voltage"`
293+
}
294+
295+
func (s *Service) HandleVoltage(ctx context.Context, params json.RawMessage) (interface{}, error) {
296+
volt, err := vchiq.GetCoreVoltage()
297+
if err != nil {
298+
return nil, fmt.Errorf("get core voltage: %w", err)
299+
}
300+
return &VoltageResponse{CoreVoltage: volt}, nil
301+
}
302+
303+
// ProcessResponse is the response for device.processes.
304+
type ProcessResponse struct {
305+
Processes []vchiq.ProcessInfo `json:"processes"`
306+
}
307+
308+
func (s *Service) HandleProcesses(ctx context.Context, params json.RawMessage) (interface{}, error) {
309+
procs, err := vchiq.ListProcesses()
310+
if err != nil {
311+
return nil, fmt.Errorf("list processes: %w", err)
312+
}
313+
return &ProcessResponse{Processes: procs}, nil
314+
}
315+
316+
// USBResponse is the response for device.usb.
317+
type USBResponse struct {
318+
Devices []vchiq.USBDevice `json:"devices"`
319+
}
320+
321+
func (s *Service) HandleUSB(ctx context.Context, params json.RawMessage) (interface{}, error) {
322+
devices, err := vchiq.GetUSBList()
323+
if err != nil {
324+
return nil, fmt.Errorf("list USB devices: %w", err)
325+
}
326+
return &USBResponse{Devices: devices}, nil
327+
}
328+
329+
// OSInfoResponse is the response for device.os.
330+
type OSInfoResponse struct {
331+
Hostname string `json:"hostname"`
332+
OSName string `json:"os_name"`
333+
KernelVersion string `json:"kernel_version"`
334+
Uptime string `json:"uptime"`
335+
LoadAverage string `json:"load_average"`
336+
FQDN string `json:"fqdn"`
337+
IPs []string `json:"ips"`
338+
}
339+
340+
func (s *Service) HandleOS(ctx context.Context, params json.RawMessage) (interface{}, error) {
341+
hostname, _ := vchiq.GetHostname()
342+
osName, _ := vchiq.GetOSName()
343+
kernel, _ := vchiq.GetKernelVersion()
344+
uptime, _ := vchiq.GetUptime()
345+
loadAvg, _ := vchiq.GetLoadAverage()
346+
fqdn, _ := vchiq.GetFQDN()
347+
348+
var ips []string
349+
netIPs, _ := vchiq.GetIPs()
350+
for _, ip := range netIPs {
351+
ips = append(ips, ip.String())
352+
}
353+
354+
return &OSInfoResponse{
355+
Hostname: hostname,
356+
OSName: osName,
357+
KernelVersion: kernel,
358+
Uptime: uptime,
359+
LoadAverage: loadAvg,
360+
FQDN: fqdn,
361+
IPs: ips,
362+
}, nil
363+
}
364+
365+
// ModelResponse is the response for device.model.
366+
type ModelResponse struct {
367+
DeviceName string `json:"device_name"`
368+
Revision string `json:"revision"`
369+
MinimalPowerSupply float64 `json:"minimal_power_supply_amps"`
370+
VcgencmdInstalled bool `json:"vcgencmd_installed"`
371+
}
372+
373+
func (s *Service) HandleModel(ctx context.Context, params json.RawMessage) (interface{}, error) {
374+
deviceName, err := vchiq.GetDeviceName()
375+
if err != nil {
376+
return nil, fmt.Errorf("get device name: %w", err)
377+
}
378+
revision, _ := vchiq.GetCPURevision()
379+
power, _ := vchiq.GetMinimalPowerSupply(deviceName)
380+
return &ModelResponse{
381+
DeviceName: deviceName,
382+
Revision: revision,
383+
MinimalPowerSupply: power,
384+
VcgencmdInstalled: vchiq.IsVcgencmdInstalled(),
385+
}, nil
386+
}

‎agent/internal/device/device_test.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ const testDeviceID = "test_device_abc123"
1616
func newTestService(t *testing.T) *Service {
1717
t.Helper()
1818
cfg := config.DefaultConfig()
19-
s, err := NewService(cfg, testDeviceID, slog.Default())
19+
s, err := NewService(cfg, "", testDeviceID, slog.Default())
2020
if err != nil {
2121
t.Fatalf("NewService() failed: %v", err)
2222
}
@@ -36,7 +36,7 @@ func TestNewService(t *testing.T) {
3636

3737
t.Run("with nil logger", func(t *testing.T) {
3838
cfg := config.DefaultConfig()
39-
s, err := NewService(cfg, testDeviceID, nil)
39+
s, err := NewService(cfg, "", testDeviceID, nil)
4040
if err != nil {
4141
t.Fatalf("NewService(nil logger) failed: %v", err)
4242
}
@@ -89,7 +89,7 @@ func TestHandleInfoFriendlyName(t *testing.T) {
8989
t.Run("uses device_name from config", func(t *testing.T) {
9090
cfg := config.DefaultConfig()
9191
cfg.Runtime.DeviceName = "MyBuzzPi"
92-
s, err := NewService(cfg, testDeviceID, slog.Default())
92+
s, err := NewService(cfg, "", testDeviceID, slog.Default())
9393
if err != nil {
9494
t.Fatalf("NewService() failed: %v", err)
9595
}
@@ -107,7 +107,7 @@ func TestHandleInfoFriendlyName(t *testing.T) {
107107
t.Run("falls back to hostname when empty", func(t *testing.T) {
108108
cfg := config.DefaultConfig()
109109
cfg.Runtime.DeviceName = ""
110-
s, err := NewService(cfg, testDeviceID, slog.Default())
110+
s, err := NewService(cfg, "", testDeviceID, slog.Default())
111111
if err != nil {
112112
t.Fatalf("NewService() failed: %v", err)
113113
}
@@ -336,7 +336,7 @@ var os_hostname = os.Hostname
336336
func TestServiceFriendlyNameFallback(t *testing.T) {
337337
cfg := config.DefaultConfig()
338338
cfg.Runtime.DeviceName = ""
339-
s, err := NewService(cfg, "dev_test", slog.Default())
339+
s, err := NewService(cfg, "", "dev_test", slog.Default())
340340
if err != nil {
341341
t.Fatalf("NewService() failed: %v", err)
342342
}
@@ -349,7 +349,7 @@ func TestServiceFriendlyNameFallback(t *testing.T) {
349349
func TestServiceFriendlyNameFromConfig(t *testing.T) {
350350
cfg := config.DefaultConfig()
351351
cfg.Runtime.DeviceName = "CustomName"
352-
s, err := NewService(cfg, "dev_test", slog.Default())
352+
s, err := NewService(cfg, "", "dev_test", slog.Default())
353353
if err != nil {
354354
t.Fatalf("NewService() failed: %v", err)
355355
}

‎agent/internal/engine/auth_test.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -242,7 +242,8 @@ func TestHandleErrorResponseHasCorrectRID(t *testing.T) {
242242
t.Parallel()
243243
m := newTestManager(t)
244244
m.RegisterMethod("test", func(ctx context.Context, params json.RawMessage) (interface{}, error) {
245-
return nil, json.Unmarshal([]byte("bad"), nil)
245+
var v interface{}
246+
return nil, json.Unmarshal([]byte("bad"), &v)
246247
})
247248

248249
env, _ := bpp.NewRequest("test", json.RawMessage(`{}`))

‎agent/internal/terminal/manager.go‎

Lines changed: 25 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"fmt"
77
"log/slog"
88
"sync"
9-
"time"
109

1110
"github.com/google/uuid"
1211
)
@@ -17,6 +16,10 @@ type Manager struct {
1716
sessions sync.Map
1817
}
1918

19+
type senderKeyType struct{}
20+
21+
var SenderKey = senderKeyType{}
22+
2023
// NewManager creates a new terminal session manager.
2124
func NewManager(logger *slog.Logger) *Manager {
2225
if logger == nil {
@@ -122,6 +125,8 @@ func (m *Manager) Health() interface{} {
122125
func (m *Manager) HandleOpen(ctx context.Context, params json.RawMessage) (interface{}, error) {
123126
var req struct {
124127
Shell string `json:"shell,omitempty"`
128+
Cols int `json:"cols,omitempty"`
129+
Rows int `json:"rows,omitempty"`
125130
}
126131
if err := json.Unmarshal(params, &req); err != nil {
127132
return nil, fmt.Errorf("invalid params: %w", err)
@@ -132,14 +137,23 @@ func (m *Manager) HandleOpen(ctx context.Context, params json.RawMessage) (inter
132137
return nil, fmt.Errorf("open terminal: %w", err)
133138
}
134139

140+
if req.Cols > 0 && req.Rows > 0 {
141+
if err := s.Resize(uint16(req.Rows), uint16(req.Cols)); err != nil {
142+
m.logger.Warn("failed to apply initial terminal size", "error", err)
143+
}
144+
}
145+
146+
if sender, ok := ctx.Value(SenderKey).(func([]byte) error); ok && sender != nil {
147+
s.StartOutputLoop(sender)
148+
}
149+
135150
return map[string]interface{}{
136151
"session_id": s.ID,
137152
"created_at": s.CreatedAt,
138153
}, nil
139154
}
140155

141156
// HandleInput is the BPP handler for terminal.input.
142-
// Writes input data to the PTY and returns any available output.
143157
func (m *Manager) HandleInput(ctx context.Context, params json.RawMessage) (interface{}, error) {
144158
var req struct {
145159
SessionID string `json:"session_id"`
@@ -158,17 +172,8 @@ func (m *Manager) HandleInput(ctx context.Context, params json.RawMessage) (inte
158172
return nil, fmt.Errorf("write: %w", err)
159173
}
160174

161-
output, err := s.ReadOutput(200 * time.Millisecond)
162-
if err != nil {
163-
m.logger.Warn("read output after input", "error", err, "session", req.SessionID)
164-
}
165-
166-
if output == nil {
167-
output = []byte{}
168-
}
169-
170175
return map[string]interface{}{
171-
"output": string(output),
176+
"ok": true,
172177
}, nil
173178
}
174179

@@ -183,6 +188,10 @@ func (m *Manager) HandleResize(ctx context.Context, params json.RawMessage) (int
183188
return nil, fmt.Errorf("invalid params: %w", err)
184189
}
185190

191+
if req.SessionID == "" {
192+
return nil, fmt.Errorf("session_id is required")
193+
}
194+
186195
s, ok := m.Get(req.SessionID)
187196
if !ok {
188197
return nil, fmt.Errorf("session not found: %s", req.SessionID)
@@ -207,6 +216,10 @@ func (m *Manager) HandleClose(ctx context.Context, params json.RawMessage) (inte
207216
return nil, fmt.Errorf("invalid params: %w", err)
208217
}
209218

219+
if req.SessionID == "" {
220+
return nil, fmt.Errorf("session_id is required")
221+
}
222+
210223
if err := m.Close(req.SessionID); err != nil {
211224
return nil, fmt.Errorf("close: %w", err)
212225
}

0 commit comments

Comments
 (0)