e7cc57a929
New tool batches session discoveries into the knowledge graph: - Creates a session-audit knowledge entry with summary - Links it to all touched entities via 'documents' relationships - Creates individual discovery knowledge entries - Stamps each entity with last_agent_session attribute - Updates AGENTS.md with tool listing
485 lines
24 KiB
Go
485 lines
24 KiB
Go
package mcp
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/dtoro/oikos/internal/db"
|
|
"github.com/google/uuid"
|
|
"github.com/modelcontextprotocol/go-sdk/mcp"
|
|
)
|
|
|
|
func KnowledgeTools(pool *db.Pool, agentID uuid.UUID, sec secretBackend) []toolReg {
|
|
return []toolReg{
|
|
{tool: &mcp.Tool{Name: "search_knowledge", Description: "Full-text search across documentation (PostgreSQL FTS with ts_rank ranking). Returns a short snippet per hit, not the full note — call get_knowledge_content with the returned slug to read the whole thing.",
|
|
InputSchema: objSchema(prop{"query", "string", "Search terms"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
q := nStr(args["query"])
|
|
return annotateJSONResult(queryRows(ctx, pool, `
|
|
SELECT ke.title, e.slug,
|
|
ts_rank(ke.search, plainto_tsquery('english', $1)) AS rank,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', $1),
|
|
'MaxWords=40, MinWords=15, ShortWord=3, MaxFragments=3,
|
|
FragmentDelimiter=" ... "') AS snippet,
|
|
ke.source, ke.tags
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE ke.search @@ plainto_tsquery('english', $1)
|
|
ORDER BY rank DESC
|
|
LIMIT 20`, q), "knowledge_results"), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_entity_knowledge", Description: "All documents, investigations, and runbooks linked to an entity. Returns a headline per note, not the full text — call get_knowledge_content with the returned slug to read the whole thing.",
|
|
InputSchema: objSchema(prop{"entity_slug", "string", "Entity slug (e.g. lxc:jellyfin, service:caddy)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["entity_slug"].(string)
|
|
return annotateJSONResult(queryRows(ctx, pool, `
|
|
SELECT ke.title, ke.source, e.type AS kind, e.slug,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
JOIN relationships r ON r.source_id = ke.entity_id
|
|
JOIN entities target ON target.id = r.target_id
|
|
WHERE target.slug = $1
|
|
AND r.valid_to IS NULL
|
|
AND r.type IN ('documents', 'about')
|
|
UNION
|
|
SELECT ke.title, ke.source, e.type AS kind, e.slug,
|
|
ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
JOIN relationships r ON r.source_id = ke.entity_id
|
|
JOIN entity_types target_type ON target_type.name = (SELECT type FROM entities WHERE slug = $1)
|
|
JOIN entities ent ON ent.type = target_type.name AND ent.slug = $1
|
|
WHERE r.valid_to IS NULL
|
|
AND r.type = 'procedure-for'
|
|
ORDER BY 1`, slug), "knowledge_results"), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_knowledge_content", Description: "Full markdown body of one document/investigation/runbook, by its own entity slug. search_knowledge and get_entity_knowledge only return short snippets/headlines — once you know which note you need (from either of those, or because you already know its slug), call this to read the whole thing before acting on it.",
|
|
InputSchema: objSchema(prop{"slug", "string", "The knowledge entity's own slug (e.g. document:containers/101-jellyfin, runbook:client-enrollment) — not the slug of an entity it's about."}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["slug"].(string)
|
|
return queryRows(ctx, pool, `
|
|
SELECT ke.title, e.slug, e.type AS kind, ke.content, ke.source, ke.tags, ke.updated_at::text
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE e.slug = $1`, slug), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "upsert_knowledge", Description: "Write back what you learned so future sessions (and future you) benefit — this is how the system gets smarter over time. Use it AFTER solving a non-obvious problem, deploying a service, or discovering a gotcha: record the finding, the fix, and any caveats. Re-calling with the same title updates the existing note instead of duplicating. This is the ONLY way to persist knowledge; a chat message alone is forgotten. search_knowledge/get_entity_knowledge find it, get_knowledge_content reads the full body back.",
|
|
InputSchema: objSchema(
|
|
prop{"title", "string", "Short, specific, searchable title (e.g. 'Dragonfly memlock rlimit in unprivileged LXCs', not 'notes')."},
|
|
prop{"content", "string", "The knowledge itself, in markdown. Be concrete: symptom, root cause, the exact fix/commands, and any caveats. Written for someone hitting this fresh."},
|
|
prop{"about", "string", "Optional entity slug(s) this knowledge concerns. Pass a single slug (e.g. 'lxc:nfs-export') or a JSON array of slugs (e.g. '[\"lxc:nfs-export\", \"lxc:gitea\"]') to link to multiple entities. get_entity_knowledge surfaces it for each."},
|
|
prop{"tags", "string", "Optional comma-separated tags (e.g. 'docker,networking,gotcha')."},
|
|
prop{"kind", "string", "One of: investigation (a finding/incident analysis — default), document (reference), runbook (a repeatable procedure)."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
return upsertKnowledge(ctx, pool, args)
|
|
}},
|
|
{tool: &mcp.Tool{Name: "delete_knowledge", Description: "Soft-delete a knowledge entry (move to trash, restorable with restore_knowledge). The content and revision history survive.",
|
|
InputSchema: objSchema(prop{"knowledge_slug", "string", "Knowledge entity slug or UUID"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["knowledge_slug"].(string)
|
|
var entityID uuid.UUID
|
|
if u, err := uuid.Parse(slug); err == nil {
|
|
entityID = u
|
|
} else {
|
|
pool.QueryRow(ctx, `SELECT id FROM entities WHERE slug = $1`, slug).Scan(&entityID)
|
|
}
|
|
if entityID == uuid.Nil {
|
|
return textResult(fmt.Sprintf("knowledge entry not found: %s", slug)), nil
|
|
}
|
|
// Snapshot before tombstoning.
|
|
pool.Exec(ctx, `
|
|
INSERT INTO knowledge_revisions (entity_id, title, content, source, tags, edited_by, version_at)
|
|
SELECT entity_id, title, content, source, tags, COALESCE(edited_by,''), updated_at
|
|
FROM knowledge_entities WHERE entity_id = $1`, entityID)
|
|
tag, err := pool.Exec(ctx,
|
|
`UPDATE knowledge_entities SET deleted_at = now(), edited_by = 'nomos'
|
|
WHERE entity_id = $1 AND deleted_at IS NULL`, entityID)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err)), nil
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return textResult("knowledge entry already deleted"), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Knowledge %s soft-deleted. Restore with restore_knowledge.", slug)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "restore_knowledge", Description: "Restore a soft-deleted knowledge entry from trash. Undoes delete_knowledge.",
|
|
InputSchema: objSchema(prop{"knowledge_slug", "string", "Knowledge entity slug or UUID"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["knowledge_slug"].(string)
|
|
var entityID uuid.UUID
|
|
if u, err := uuid.Parse(slug); err == nil {
|
|
entityID = u
|
|
} else {
|
|
pool.QueryRow(ctx, `SELECT id FROM entities WHERE slug = $1`, slug).Scan(&entityID)
|
|
}
|
|
if entityID == uuid.Nil {
|
|
return textResult(fmt.Sprintf("knowledge entry not found: %s", slug)), nil
|
|
}
|
|
tag, err := pool.Exec(ctx,
|
|
`UPDATE knowledge_entities SET deleted_at = NULL, edited_by = 'nomos'
|
|
WHERE entity_id = $1 AND deleted_at IS NOT NULL`, entityID)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err)), nil
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return textResult("knowledge entry is not deleted"), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Knowledge %s restored from trash.", slug)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "merge_knowledge", Description: "Fold one or more knowledge entries into a target. Source content is appended under a provenance heading, and the union of all tags is kept. Sources are soft-deleted afterwards.",
|
|
InputSchema: objSchema(
|
|
prop{"target_slug", "string", "Knowledge entry to merge INTO (slug or UUID)"},
|
|
prop{"source_slugs", "string", "Comma-separated slugs of entries to fold into the target"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
targetSlug, _ := args["target_slug"].(string)
|
|
sourceStr, _ := args["source_slugs"].(string)
|
|
|
|
var targetID uuid.UUID
|
|
if u, err := uuid.Parse(targetSlug); err == nil {
|
|
targetID = u
|
|
} else {
|
|
pool.QueryRow(ctx, `
|
|
SELECT ke.entity_id FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE (e.slug = $1 OR e.id::text = $1) AND ke.deleted_at IS NULL`,
|
|
targetSlug).Scan(&targetID)
|
|
}
|
|
if targetID == uuid.Nil {
|
|
return textResult(fmt.Sprintf("target knowledge entry not found: %s", targetSlug)), nil
|
|
}
|
|
|
|
sources := []string{}
|
|
for _, s := range strings.Split(sourceStr, ",") {
|
|
if s = strings.TrimSpace(s); s != "" && s != targetSlug {
|
|
sources = append(sources, s)
|
|
}
|
|
}
|
|
if len(sources) == 0 {
|
|
return textResult("no valid source entries to merge"), nil
|
|
}
|
|
|
|
var appended strings.Builder
|
|
merged := []string{}
|
|
for _, srcSlug := range sources {
|
|
var title, content, updated string
|
|
var tags []string
|
|
err := pool.QueryRow(ctx, `
|
|
SELECT ke.title, ke.content, COALESCE(ke.tags,'{}'), ke.updated_at::text
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE (e.slug = $1 OR e.id::text = $1) AND ke.deleted_at IS NULL`,
|
|
srcSlug).Scan(&title, &content, &tags, &updated)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
appended.WriteString("\n\n---\n\n## Merged: ")
|
|
appended.WriteString(title)
|
|
appended.WriteString("\n\n*Originally ")
|
|
appended.WriteString(srcSlug)
|
|
appended.WriteString(", last updated ")
|
|
appended.WriteString(updated)
|
|
appended.WriteString("*\n\n")
|
|
appended.WriteString(content)
|
|
for _, t := range tags {
|
|
fmt.Fprintf(&appended, "\ntag: %s", strings.ToLower(strings.TrimSpace(t)))
|
|
}
|
|
merged = append(merged, srcSlug)
|
|
}
|
|
|
|
if len(merged) == 0 {
|
|
return textResult("no source entries could be read"), nil
|
|
}
|
|
|
|
_, err := pool.Exec(ctx, `
|
|
UPDATE knowledge_entities SET content = content || $2, edited_by = 'nomos', updated_at = now()
|
|
WHERE entity_id = $1`, targetID, appended.String())
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error appending content: %v", err)), nil
|
|
}
|
|
|
|
for _, srcSlug := range merged {
|
|
pool.Exec(ctx, `
|
|
UPDATE knowledge_entities ke SET deleted_at = now(), edited_by = 'nomos'
|
|
FROM entities e
|
|
WHERE e.id = ke.entity_id AND (e.slug = $1 OR e.id::text = $1)`,
|
|
srcSlug)
|
|
}
|
|
|
|
return textResult(fmt.Sprintf("Merged %d entries into %s: %s", len(merged), targetSlug, strings.Join(merged, ", "))), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "rename_knowledge_tag", Description: "Bulk-rename one or more tags across all knowledge entries. Case-insensitive matching — 'oom' and 'OOM' are treated as the same tag. Deduplicates after rename.",
|
|
InputSchema: objSchema(
|
|
prop{"from", "string", "Comma-separated tag names to rename FROM"},
|
|
prop{"to", "string", "New tag name"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
fromStr, _ := args["from"].(string)
|
|
to, _ := args["to"].(string)
|
|
to = strings.ToLower(strings.TrimSpace(to))
|
|
|
|
from := []string{}
|
|
for _, f := range strings.Split(fromStr, ",") {
|
|
if f = strings.TrimSpace(f); f != "" {
|
|
from = append(from, strings.ToLower(f))
|
|
}
|
|
}
|
|
if to == "" || len(from) == 0 {
|
|
return textResult("from and to are required"), nil
|
|
}
|
|
|
|
tag, err := pool.Exec(ctx, `
|
|
UPDATE knowledge_entities ke
|
|
SET tags = sub.new_tags, updated_at = now()
|
|
FROM (
|
|
SELECT k.entity_id,
|
|
ARRAY(SELECT DISTINCT CASE WHEN lower(t) = ANY($1) THEN $2 ELSE t END
|
|
FROM unnest(k.tags) AS t) AS new_tags
|
|
FROM knowledge_entities k
|
|
WHERE k.deleted_at IS NULL
|
|
AND EXISTS (SELECT 1 FROM unnest(k.tags) AS t WHERE lower(t) = ANY($1))
|
|
) AS sub
|
|
WHERE ke.entity_id = sub.entity_id`, from, to)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error: %v", err)), nil
|
|
}
|
|
return textResult(fmt.Sprintf("Tag %s → %s: %d entries updated.", strings.Join(from, ", "), to, tag.RowsAffected())), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_knowledge_revisions", Description: "Version history for a knowledge entry. Returns title, content, editor, tags, and timestamps for each revision.",
|
|
InputSchema: objSchema(
|
|
prop{"knowledge_slug", "string", "Knowledge entity slug (e.g. document:nomos/something)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
slug, _ := args["knowledge_slug"].(string)
|
|
return queryRows(ctx, pool, `
|
|
SELECT kr.id, kr.title, kr.content, COALESCE(kr.edited_by, '') AS edited_by,
|
|
COALESCE(kr.tags::text, '{}') AS tags,
|
|
kr.version_at::text, kr.revised_at::text
|
|
FROM knowledge_revisions kr
|
|
JOIN entities e ON e.id = kr.entity_id
|
|
WHERE e.slug = $1
|
|
ORDER BY kr.version_at DESC LIMIT 50`, slug), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_knowledge_duplicates", Description: "Near-duplicate knowledge entries detected via trigram similarity. Returns clusters of similar documents with similarity scores. Use before creating new knowledge to avoid pileup.",
|
|
InputSchema: objSchema(
|
|
prop{"threshold", "number", "Similarity threshold 0-1 (default 0.6, lower = more matches)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
threshold := getFloat(args, "threshold", 0.6)
|
|
return queryRows(ctx, pool, `
|
|
SELECT a.slug AS doc_a, b.slug AS doc_b, similarity(ka.title, kb.title) AS sim
|
|
FROM knowledge_entities ka
|
|
JOIN knowledge_entities kb ON ka.entity_id < kb.entity_id
|
|
JOIN entities a ON a.id = ka.entity_id
|
|
JOIN entities b ON b.id = kb.entity_id
|
|
WHERE ka.deleted_at IS NULL AND kb.deleted_at IS NULL
|
|
AND similarity(ka.title, kb.title) > $1
|
|
ORDER BY sim DESC LIMIT 100`, threshold), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "get_knowledge_orphans", Description: "Knowledge entries with no entity links (unlinked), no tags (untagged), or stale (not updated in N days). Helps identify abandoned or disconnected knowledge to clean up.",
|
|
InputSchema: objSchema(
|
|
prop{"stale_days", "integer", "Days without update to consider stale (default 90)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
staleDays := int(getFloat(args, "stale_days", 90))
|
|
return queryRows(ctx, pool, fmt.Sprintf(`
|
|
SELECT e.slug, ke.title, e.type AS kind, COALESCE(ke.edited_by, '') AS edited_by,
|
|
ke.updated_at::text,
|
|
(ke.tags IS NULL OR cardinality(ke.tags) = 0) AS untagged,
|
|
NOT EXISTS (
|
|
SELECT 1 FROM relationships r
|
|
WHERE r.source_id = ke.entity_id AND r.valid_to IS NULL
|
|
AND r.type IN ('documents', 'about')
|
|
) AS unlinked,
|
|
(ke.updated_at < now() - interval '%d days') AS stale
|
|
FROM knowledge_entities ke
|
|
JOIN entities e ON e.id = ke.entity_id
|
|
WHERE ke.deleted_at IS NULL
|
|
ORDER BY ke.updated_at ASC`, staleDays)), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "list_knowledge_tags", Description: "All tags used across the knowledge base with usage counts. Returns normalized tag, count, and any casing variants (e.g. 'oom' and 'OOM' surface as variants so you can spot drift).",
|
|
InputSchema: objSchema(),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
return queryRows(ctx, pool, `
|
|
SELECT lower(tag) AS tag, count(*) AS uses,
|
|
array_agg(DISTINCT tag ORDER BY tag) AS variants
|
|
FROM knowledge_entities ke, unnest(ke.tags) AS tag
|
|
WHERE ke.deleted_at IS NULL
|
|
GROUP BY lower(tag) ORDER BY uses DESC, lower(tag)`), nil
|
|
}},
|
|
{tool: &mcp.Tool{Name: "list_my_secrets", Description: "List secrets accessible to this client by public key",
|
|
InputSchema: objSchema(prop{"caller_pubkey", "string", "Age public key of the caller (optional)"}),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
pubkey, _ := args["caller_pubkey"].(string)
|
|
// Match entities where age_pubkey attribute contains the caller's key.
|
|
query := `
|
|
SELECT e.slug, e.type, e.name,
|
|
e.attributes->>'age_pubkey' AS age_pubkey
|
|
FROM entities e
|
|
WHERE e.attributes->>'age_pubkey' IS NOT NULL`
|
|
var dbArgs []any
|
|
if pubkey != "" {
|
|
query += ` AND e.attributes->>'age_pubkey' = $1`
|
|
dbArgs = append(dbArgs, pubkey)
|
|
}
|
|
query += ` ORDER BY e.slug LIMIT 100`
|
|
return queryRows(ctx, pool, query, dbArgs...), nil
|
|
}},
|
|
// upsert_session_summary persists what was learned in a session
|
|
// back into the knowledge graph. Call this at the end of a
|
|
// session to close the loop: knowledge entries are created,
|
|
// entity attributes are updated, and the graph stays current.
|
|
{tool: &mcp.Tool{Name: "upsert_session_summary", Description: "Batch-write session discoveries into the knowledge graph. Call at session end to persist everything learned: creates knowledge entries for each discovery, links them to the entities they affect, and records a session-audit knowledge entry tagged with the session_id. This is how the system gets smarter over time — without it, everything discovered in a session is lost when the conversation ends.",
|
|
InputSchema: objSchema(
|
|
prop{"session_id", "string", "Unique session identifier (e.g. chat session UUID, cron job ID). Used to tag and group knowledge entries from this session."},
|
|
prop{"summary", "string", "Free-text summary of what was accomplished in this session. Written as a readable paragraph for future reference."},
|
|
prop{"entities_touched", "string", "JSON array of entity slugs that were modified or inspected (e.g. [\"lxc:arriman\", \"service:sabnzbd\"]). Each gets linked to the session-audit knowledge entry."},
|
|
prop{"discoveries", "string", "Optional JSON array of discovery objects. Each object: {title, content, kind (document|investigation|runbook), tags (comma-separated)}. Each creates a separate searchable knowledge entry linked to the relevant entities."},
|
|
),
|
|
}, handler: func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
|
|
args := argsMap(req)
|
|
sessionID, _ := args["session_id"].(string)
|
|
summary, _ := args["summary"].(string)
|
|
entitiesRaw, _ := args["entities_touched"].(string)
|
|
discoveriesRaw, _ := args["discoveries"].(string)
|
|
|
|
if sessionID == "" || summary == "" {
|
|
return textResult("error: session_id and summary are required"), nil
|
|
}
|
|
|
|
var entitySlugs []string
|
|
if entitiesRaw != "" {
|
|
if err := json.Unmarshal([]byte(entitiesRaw), &entitySlugs); err != nil {
|
|
return textResult(fmt.Sprintf("error: entities_touched is not valid JSON array: %v", err)), nil
|
|
}
|
|
}
|
|
|
|
var discoveries []struct {
|
|
Title string `json:"title"`
|
|
Content string `json:"content"`
|
|
Kind string `json:"kind"`
|
|
Tags string `json:"tags"`
|
|
}
|
|
if discoveriesRaw != "" {
|
|
if err := json.Unmarshal([]byte(discoveriesRaw), &discoveries); err != nil {
|
|
return textResult(fmt.Sprintf("error: discoveries is not valid JSON: %v", err)), nil
|
|
}
|
|
}
|
|
|
|
created := 0
|
|
linked := 0
|
|
updated := 0
|
|
|
|
// Create the session-audit knowledge entry
|
|
auditTitle := fmt.Sprintf("Session audit: %s", sessionID)
|
|
auditSlug := fmt.Sprintf("session-audit/%s", sessionID)
|
|
tags := []string{"audit", "session"}
|
|
auditDocID, _ := uuid.NewV7()
|
|
err := pool.QueryRow(ctx, `
|
|
INSERT INTO entities (id, slug, type, name, attributes)
|
|
VALUES ($1, $2, 'investigation', $3, '{}')
|
|
ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name, updated_at = now()
|
|
RETURNING id`, auditDocID, auditSlug, auditTitle).Scan(&auditDocID)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error creating audit entity: %v", err)), nil
|
|
}
|
|
_, err = pool.Exec(ctx, `
|
|
INSERT INTO knowledge_entities (entity_id, title, content, source, tags, updated_at)
|
|
VALUES ($1, $2, $3, 'nomos-agent', $4, now())
|
|
ON CONFLICT (entity_id) DO UPDATE
|
|
SET title = EXCLUDED.title, content = EXCLUDED.content,
|
|
tags = EXCLUDED.tags, updated_at = now()`,
|
|
auditDocID, auditTitle, summary, tags)
|
|
if err != nil {
|
|
return textResult(fmt.Sprintf("error writing audit knowledge: %v", err)), nil
|
|
}
|
|
created++
|
|
|
|
// Link the audit entry to each touched entity
|
|
for _, slug := range entitySlugs {
|
|
var targetID uuid.UUID
|
|
if qerr := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", slug).Scan(&targetID); qerr == nil {
|
|
pool.Exec(ctx, `
|
|
INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
|
|
SELECT $1, $2, 'documents', '{"by":"nomos"}'::jsonb, now()
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM relationships
|
|
WHERE source_id = $1 AND target_id = $2 AND type = 'documents' AND valid_to IS NULL)`,
|
|
auditDocID, targetID)
|
|
linked++
|
|
}
|
|
}
|
|
|
|
// Create discovery knowledge entries
|
|
for _, d := range discoveries {
|
|
kind := d.Kind
|
|
switch kind {
|
|
case "document", "investigation", "runbook":
|
|
case "":
|
|
kind = "investigation"
|
|
default:
|
|
continue
|
|
}
|
|
discTitle := strings.TrimSpace(d.Title)
|
|
discContent := strings.TrimSpace(d.Content)
|
|
if discTitle == "" || discContent == "" {
|
|
continue
|
|
}
|
|
slug := fmt.Sprintf("discovery/%s/%s", sessionID, strings.ReplaceAll(strings.ToLower(discTitle), " ", "-"))
|
|
discID, _ := uuid.NewV7()
|
|
err := pool.QueryRow(ctx, `
|
|
INSERT INTO entities (id, slug, type, name, attributes)
|
|
VALUES ($1, $2, $3, $4, '{}')
|
|
ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name, updated_at = now()
|
|
RETURNING id`, discID, slug, kind, discTitle).Scan(&discID)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
var discTags []string
|
|
for _, t := range strings.Split(d.Tags, ",") {
|
|
if t = strings.TrimSpace(t); t != "" {
|
|
discTags = append(discTags, t)
|
|
}
|
|
}
|
|
discTags = append(discTags, "discovery", "session")
|
|
_, err = pool.Exec(ctx, `
|
|
INSERT INTO knowledge_entities (entity_id, title, content, source, tags, updated_at)
|
|
VALUES ($1, $2, $3, 'nomos-agent', $4, now())
|
|
ON CONFLICT (entity_id) DO UPDATE
|
|
SET title = EXCLUDED.title, content = EXCLUDED.content,
|
|
tags = EXCLUDED.tags, updated_at = now()`,
|
|
discID, discTitle, discContent, discTags)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
created++
|
|
}
|
|
|
|
// Update attributes on each touched entity
|
|
for _, slug := range entitySlugs {
|
|
pool.Exec(ctx, `
|
|
UPDATE entities
|
|
SET attributes = jsonb_set(COALESCE(attributes, '{}'), '{last_agent_session}', to_jsonb($2::text), true),
|
|
updated_at = now()
|
|
WHERE slug = $1`, slug, sessionID)
|
|
updated++
|
|
}
|
|
|
|
return textResult(fmt.Sprintf(
|
|
"Session summary saved. Created %d knowledge entries, linked to %d entities, updated %d entity attributes.",
|
|
created, linked, updated)), nil
|
|
}},
|
|
}
|
|
}
|