diff --git a/cmd/mavend/tick.go b/cmd/mavend/tick.go index 1ce653a..3ebbeef 100644 --- a/cmd/mavend/tick.go +++ b/cmd/mavend/tick.go @@ -10,6 +10,7 @@ package main import ( "context" + "errors" "fmt" "log" "os" @@ -440,6 +441,14 @@ func (t *tickLoop) trace() *loop.TickTrace { type daemonAPI struct { ipc.CoreAPI getTrace func() *loop.TickTrace + chatFn func(ctx context.Context, text string) string +} + +func (d *daemonAPI) Chat(ctx context.Context, text string) (string, error) { + if d.chatFn == nil { + return "", errors.New("mavend: chat not available") + } + return d.chatFn(ctx, text), nil } func (d *daemonAPI) TickTrace(ctx context.Context) (ipc.TickTrace, error) { diff --git a/internal/ipc/api.go b/internal/ipc/api.go index 8f33df5..dc266f8 100644 --- a/internal/ipc/api.go +++ b/internal/ipc/api.go @@ -173,6 +173,14 @@ type Tool struct { Updated time.Time `json:"updated"` } +// chatReq / chatResp — text chat round-trip for the IPC Chat method. +type chatReq struct { + Text string `json:"text"` +} +type chatResp struct { + Reply string `json:"reply"` +} + type proposeToolReq struct { Name string `json:"name"` Scope string `json:"scope"` @@ -203,6 +211,25 @@ type listToolsResp struct { Tools []Tool `json:"tools"` } +// ProposedRoutine — a detected pattern awaiting human confirmation. +type ProposedRoutine struct { + ID int64 `json:"id"` + Action string `json:"action"` + Object string `json:"object"` + IntervalDays float64 `json:"interval_days"` + Status string `json:"status"` // proposed | accepted | dismissed + CreatedTs int64 `json:"created_ts"` + ReminderID *int64 `json:"reminder_id,omitempty"` +} + +type listProposedRoutinesResp struct { + Routines []ProposedRoutine `json:"routines"` +} + +type dismissProposedRoutineReq struct { + ID int64 `json:"id"` +} + // CoreAPI — what core exposes to modules. One Go interface, satisfied by: // - the in-process store adapter (server.go storeAPI) — used by the daemon // for modules that live in-process for now (router, delivery) and by tests, @@ -243,14 +270,25 @@ type CoreAPI interface { ProposeTool(ctx context.Context, name, utterance, scope string, ts time.Time) (bool, error) EnableTool(ctx context.Context, name string, cmd []string, destructive bool, scope string, ts time.Time) error DisableTool(ctx context.Context, name string) error + DeleteTool(ctx context.Context, name string) error LookupTool(ctx context.Context, name string) (Tool, error) ListTools(ctx context.Context, status string) ([]Tool, error) RevertFact(ctx context.Context, key string) (int64, error) + // ListProposedRoutines returns proposed routines, newest first. + ListProposedRoutines(ctx context.Context) ([]ProposedRoutine, error) + // DismissProposedRoutine flips a proposed routine to 'dismissed'. + DismissProposedRoutine(ctx context.Context, id int64) error + // TickTrace returns the most recent tick's rule trace. The daemon caches // this after every tick; the store adapter returns an error (trace is not // persisted — it's a daemon-level cache). TickTrace(ctx context.Context) (TickTrace, error) + + // Chat routes a text utterance through the reactive handler's core path + // (router → dialogue → action → replier) and returns the reply text. + // No audio or stt/tts — for text channels (mavweb, telegram). + Chat(ctx context.Context, text string) (string, error) } // --- Rule trace / explanation DTOs --- diff --git a/internal/ipc/client.go b/internal/ipc/client.go index 8080be4..1c61b30 100644 --- a/internal/ipc/client.go +++ b/internal/ipc/client.go @@ -369,6 +369,30 @@ func (c *Client) ListTools(ctx context.Context, status string) ([]Tool, error) { return r.Tools, nil } +func (c *Client) DeleteTool(ctx context.Context, name string) error { + return c.call(ctx, MethodDeleteTool, disableToolReq{Name: name}, nil) +} + +func (c *Client) ListProposedRoutines(ctx context.Context) ([]ProposedRoutine, error) { + var r listProposedRoutinesResp + if err := c.call(ctx, MethodListProposedRoutines, nil, &r); err != nil { + return nil, err + } + return r.Routines, nil +} + +func (c *Client) DismissProposedRoutine(ctx context.Context, id int64) error { + return c.call(ctx, MethodDismissProposedRoutine, dismissProposedRoutineReq{ID: id}, nil) +} + +func (c *Client) Chat(ctx context.Context, text string) (string, error) { + var r chatResp + if err := c.call(ctx, MethodChat, chatReq{Text: text}, &r); err != nil { + return "", err + } + return r.Reply, nil +} + func (c *Client) TickTrace(ctx context.Context) (TickTrace, error) { var t TickTrace if err := c.call(ctx, MethodTickTrace, nil, &t); err != nil { diff --git a/internal/ipc/ipc_test.go b/internal/ipc/ipc_test.go index db79005..f4faf85 100644 --- a/internal/ipc/ipc_test.go +++ b/internal/ipc/ipc_test.go @@ -375,6 +375,131 @@ func TestCaller_Peercred(t *testing.T) { } } +// TestChatViaClient — Chat round-trips over the wire. Uses a custom CoreAPI +// that implements Chat (storeAPI returns an error for it). +func TestChatViaClient(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "maven.sock") + chatAPI := &chatTestAPI{} + srv, err := Listen(sock, chatAPI) + if err != nil { + t.Fatalf("listen: %v", err) + } + done := make(chan struct{}) + go func() { + _ = srv.Serve() + close(done) + }() + t.Cleanup(func() { + _ = srv.Close() + <-done + }) + cli, err := Dial(srv.Path()) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { _ = cli.Close() }) + + reply, err := cli.Chat(context.Background(), "привет") + if err != nil { + t.Fatalf("Chat: %v", err) + } + if reply != "и тебе привет!" { + t.Fatalf("Chat = %q, want %q", reply, "и тебе привет!") + } +} + +// chatTestAPI — a minimal CoreAPI that only implements Chat for testing. +type chatTestAPI struct{} + +func (a *chatTestAPI) WriteFact(ctx context.Context, req WriteFactReq) (int64, error) { + return 0, ErrUnknownMethod +} +func (a *chatTestAPI) LatestFact(ctx context.Context, key string) (Fact, error) { + return Fact{}, ErrUnknownMethod +} +func (a *chatTestAPI) LatestFactBySource(ctx context.Context, key, source string) (Fact, error) { + return Fact{}, ErrUnknownMethod +} +func (a *chatTestAPI) Since(ctx context.Context, key string, now time.Time) (time.Duration, error) { + return 0, ErrUnknownMethod +} +func (a *chatTestAPI) Presence(ctx context.Context) (Presence, error) { + return Presence{}, ErrUnknownMethod +} +func (a *chatTestAPI) CreateReminder(ctx context.Context, fire time.Time, payload, cron string) (int64, error) { + return 0, ErrUnknownMethod +} +func (a *chatTestAPI) MarkReminder(ctx context.Context, id int64, status string) error { + return ErrUnknownMethod +} +func (a *chatTestAPI) ListReminders(ctx context.Context, n int) ([]Reminder, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) RecordNudge(ctx context.Context, rule, channel, message string, ts time.Time) (int64, error) { + return 0, ErrUnknownMethod +} +func (a *chatTestAPI) ResolveNudge(ctx context.Context, id int64, outcome string, ts time.Time) error { + return ErrUnknownMethod +} +func (a *chatTestAPI) RecentOutcomes(ctx context.Context, rule string, n int) ([]string, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) RecentFacts(ctx context.Context, n int) ([]Fact, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) CalendarEvents(ctx context.Context, from, to time.Time) ([]Fact, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) RecentNudges(ctx context.Context, n int) ([]Nudge, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error) { + return 0, ErrUnknownMethod +} +func (a *chatTestAPI) QueryNotes(ctx context.Context, embedding []float32, k int) ([]Note, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) RecentNotes(ctx context.Context, n int) ([]Note, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) ProposeTool(ctx context.Context, name, utterance, scope string, ts time.Time) (bool, error) { + return false, ErrUnknownMethod +} +func (a *chatTestAPI) EnableTool(ctx context.Context, name string, cmd []string, destructive bool, scope string, ts time.Time) error { + return ErrUnknownMethod +} +func (a *chatTestAPI) DisableTool(ctx context.Context, name string) error { + return ErrUnknownMethod +} +func (a *chatTestAPI) DeleteTool(ctx context.Context, name string) error { + return ErrUnknownMethod +} +func (a *chatTestAPI) ListProposedRoutines(ctx context.Context) ([]ProposedRoutine, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) DismissProposedRoutine(ctx context.Context, id int64) error { + return ErrUnknownMethod +} +func (a *chatTestAPI) LookupTool(ctx context.Context, name string) (Tool, error) { + return Tool{}, ErrUnknownMethod +} +func (a *chatTestAPI) ListTools(ctx context.Context, status string) ([]Tool, error) { + return nil, ErrUnknownMethod +} +func (a *chatTestAPI) RevertFact(ctx context.Context, key string) (int64, error) { + return 0, ErrUnknownMethod +} +func (a *chatTestAPI) TickTrace(ctx context.Context) (TickTrace, error) { + return TickTrace{}, ErrUnknownMethod +} +func (a *chatTestAPI) Chat(ctx context.Context, text string) (string, error) { + if text == "привет" { + return "и тебе привет!", nil + } + return "поговорили.", nil +} + // TestDispatch_UnknownMethod — an unknown method over the wire comes back as // ErrUnknownMethod, not a panic or a dropped conn. The server must stay up // for the next (legitimate) request on the same conn. diff --git a/internal/ipc/server.go b/internal/ipc/server.go index 845c28f..6eb8625 100644 --- a/internal/ipc/server.go +++ b/internal/ipc/server.go @@ -199,6 +199,10 @@ func (a *storeAPI) RevertFact(ctx context.Context, key string) (int64, error) { return newID, mapErr(err) } +func (a *storeAPI) Chat(ctx context.Context, text string) (string, error) { + return "", errors.New("store: chat not available via direct store API") +} + func (a *storeAPI) TickTrace(ctx context.Context) (TickTrace, error) { return TickTrace{}, errors.New("store: tick trace not available via direct store API") } @@ -215,6 +219,36 @@ func (a *storeAPI) ListTools(ctx context.Context, status string) ([]Tool, error) return out, nil } +func (a *storeAPI) DeleteTool(ctx context.Context, name string) error { + return mapErr(a.s.DeleteTool(ctx, name)) +} + +func (a *storeAPI) ListProposedRoutines(ctx context.Context) ([]ProposedRoutine, error) { + rs, err := a.s.ListProposedRoutines(ctx) + if err != nil { + return nil, mapErr(err) + } + out := make([]ProposedRoutine, len(rs)) + for i, r := range rs { + out[i] = ProposedRoutine{ + ID: r.ID, + Action: r.Action, + Object: r.Object, + IntervalDays: r.IntervalDays, + Status: r.Status, + CreatedTs: r.CreatedTs.UnixMilli(), + } + if r.ReminderID != nil { + out[i].ReminderID = r.ReminderID + } + } + return out, nil +} + +func (a *storeAPI) DismissProposedRoutine(ctx context.Context, id int64) error { + return mapErr(a.s.DismissProposedRoutine(ctx, id)) +} + func toTool(t store.Tool) Tool { return Tool{ Name: t.Name, Scope: t.Scope, Cmd: t.Cmd, Destructive: t.Destructive, @@ -712,6 +746,30 @@ func (s *Server) dispatch(ctx context.Context, req Request) (json.RawMessage, er } return marshalResult(listToolsResp{Tools: out}), nil + case MethodDeleteTool: + var p disableToolReq + if err := unmarshalParams(req.Params, &p); err != nil { + return nil, err + } + return marshalResult(nil), api.DeleteTool(ctx, p.Name) + + case MethodListProposedRoutines: + out, err := api.ListProposedRoutines(ctx) + if err != nil { + return nil, err + } + if out == nil { + out = []ProposedRoutine{} + } + return marshalResult(listProposedRoutinesResp{Routines: out}), nil + + case MethodDismissProposedRoutine: + var p dismissProposedRoutineReq + if err := unmarshalParams(req.Params, &p); err != nil { + return nil, err + } + return marshalResult(nil), api.DismissProposedRoutine(ctx, p.ID) + case MethodRevertFact: var p struct { Key string `json:"key"` @@ -725,6 +783,17 @@ func (s *Server) dispatch(ctx context.Context, req Request) (json.RawMessage, er } return marshalResult(map[string]int64{"new_id": newID}), nil + case MethodChat: + var p chatReq + if err := unmarshalParams(req.Params, &p); err != nil { + return nil, err + } + reply, err := api.Chat(ctx, p.Text) + if err != nil { + return nil, err + } + return marshalResult(chatResp{Reply: reply}), nil + case MethodTickTrace: t, err := api.TickTrace(ctx) if err != nil { diff --git a/internal/ipc/wire.go b/internal/ipc/wire.go index c03ded7..7682013 100644 --- a/internal/ipc/wire.go +++ b/internal/ipc/wire.go @@ -37,9 +37,13 @@ const ( MethodStoreEncryptionKey Method = "store_encryption_key" MethodUnlock Method = "unlock" MethodLookupTool Method = "lookup_tool" - MethodListTools Method = "list_tools" - MethodRevertFact Method = "revert_fact" + MethodListTools Method = "list_tools" + MethodDeleteTool Method = "delete_tool" + MethodListProposedRoutines Method = "list_proposed_routines" + MethodDismissProposedRoutine Method = "dismiss_proposed_routine" + MethodRevertFact Method = "revert_fact" MethodTickTrace Method = "tick_trace" + MethodChat Method = "chat" ) // Request — one frame from module to core. Params is the JSON-encoded argument diff --git a/internal/store/ack.go b/internal/store/ack.go new file mode 100644 index 0000000..58aa888 --- /dev/null +++ b/internal/store/ack.go @@ -0,0 +1,71 @@ +package store + +import ( + "context" + "fmt" + "time" +) + +// WasAcked returns true when the rule has no pending (un-acked) telegram +// nudges. A resolved nudge (acted/snoozed/ignored) means the user has seen +// and dealt with it — the alarm is considered acked. +func (s *Store) WasAcked(ctx context.Context, key string) (bool, error) { + var n int + err := s.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM nudges + WHERE rule = ? AND channel = 'telegram' AND outcome = 'pending'`, key).Scan(&n) + if err != nil { + return false, fmt.Errorf("was acked %s: %w", key, err) + } + return n == 0, nil +} + +// MarkSent records that a sev4 telegram nudge was sent (or re-sent) for the +// given rule at the given time. Used by the repeat-til-ack loop to clock the +// repeat interval. +func (s *Store) MarkSent(ctx context.Context, key string, ts time.Time) error { + _, err := s.db.ExecContext(ctx, + `INSERT INTO ack_sends (rule, sent_at) VALUES (?, ?)`, + key, ts.UnixMilli()) + if err != nil { + return fmt.Errorf("mark sent %s: %w", key, err) + } + return nil +} + +// LastSent returns the most recent send timestamp for the given rule's +// telegram nudge. Returns zero time if nothing has been sent yet (the initial +// send goes through RecordNudge, not MarkSent, so the first MarkSent comes on +// the repeat path — LastSent may legitimately be zero until then). +func (s *Store) LastSent(ctx context.Context, key string) (time.Time, error) { + var millis int64 + err := s.db.QueryRowContext(ctx, + `SELECT MAX(sent_at) FROM ack_sends WHERE rule = ?`, key).Scan(&millis) + if err != nil { + return time.Time{}, fmt.Errorf("last sent %s: %w", key, err) + } + if millis == 0 { + return time.Time{}, nil + } + return time.UnixMilli(millis).UTC(), nil +} + +// MarkAcked marks ALL pending telegram nudges for the rule as "acted" — +// stopping the repeat-til-ack loop. Called when the user acknowledges the +// alarm (voice acknowledgment, Telegram callback, etc.). +func (s *Store) MarkAcked(ctx context.Context, key string) error { + now := time.Now() + res, err := s.db.ExecContext(ctx, + `UPDATE nudges SET outcome = 'acted', outcome_ts = ? + WHERE rule = ? AND channel = 'telegram' AND outcome = 'pending'`, + now.UnixMilli(), key) + if err != nil { + return fmt.Errorf("mark acked %s: %w", key, err) + } + n, _ := res.RowsAffected() + if n == 0 { + // no pending nudges — already acked or never sent; not an error. + return nil + } + return nil +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 7bfd2af..bd8c8b2 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -26,6 +26,30 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2 meta TEXT NOT NULL DEFAULT '{}', created_ts INTEGER NOT NULL );`, // #3 — long-term vector memory (persistent backend for internal/memory) + `CREATE TABLE IF NOT EXISTS events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + fact_id INTEGER NOT NULL REFERENCES facts(id), + action TEXT NOT NULL, + object TEXT NOT NULL, + ts INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_events_action_object ON events (action, object, ts DESC); + CREATE TABLE IF NOT EXISTS proposed_routines ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + action TEXT NOT NULL, + object TEXT NOT NULL, + interval_days REAL NOT NULL, + status TEXT NOT NULL DEFAULT 'proposed' CHECK (status IN ('proposed','accepted','dismissed')), + created_ts INTEGER NOT NULL, + reminder_id INTEGER REFERENCES reminders(id), + UNIQUE(action, object) + );`, // #4 — event extraction + pattern inference + + `CREATE TABLE IF NOT EXISTS ack_sends ( + rule TEXT NOT NULL, + sent_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_ack_sends_rule ON ack_sends (rule, sent_at DESC);`, // #5 — sev4 telegram repeat-til-ack tracking } // migrate applies every migration with a number greater than the DB's current diff --git a/internal/store/tools.go b/internal/store/tools.go index 5d7d2a8..2c86af8 100644 --- a/internal/store/tools.go +++ b/internal/store/tools.go @@ -88,6 +88,15 @@ func (s *Store) EnableTool(ctx context.Context, name string, cmd []string, destr return nil } +// DeleteTool permanently removes a tool row. Used for "dismiss" on proposed +// tools — there's no dismissed status, the proposal is simply gone and maven +// can re-propose it later if the same gap is encountered. Idempotent: deleting +// a tool that doesn't exist is a no-op. +func (s *Store) DeleteTool(ctx context.Context, name string) error { + _, err := s.db.ExecContext(ctx, `DELETE FROM tools WHERE name = ?`, name) + return err +} + // DisableTool sets a tool's status from 'enabled' back to 'proposed'. This is // the "disable" act on the authed surface — the tool stays in the store (its // provenance preserved) but won't run until re-enabled. Idempotent: disabling