diff --git a/.env.example b/.env.example index 686d02d..4256383 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,6 @@ +# Runtime role: brain = full knowledge system; agent = lightweight source poller using the same binary/image. +BRAIN_MODE=brain + # HTTP BRAIN_LISTEN_ADDR=:8090 BRAIN_DATA_DIR=./data @@ -88,6 +91,17 @@ BRAIN_RELATION_THRESHOLD=0.72 BRAIN_TOP_K=8 BRAIN_MAX_CONTEXT_CHARS=16000 +# Distributed Source-Agent inbox (Brain mode). Agent documents are queued outside +# the graph, embedded in small batches, and only promoted to candidate status when +# they match existing knowledge. Adaptive article research checks this inbox before SearXNG. +BRAIN_SOURCE_INBOX_ENABLED=true +BRAIN_SOURCE_INBOX_INTERVAL=30s +BRAIN_SOURCE_INBOX_BATCH_SIZE=12 +BRAIN_SOURCE_INBOX_MIN_SIMILARITY=0.55 +BRAIN_SOURCE_INBOX_MIN_RESULTS=2 +# Freshness-sensitive article queries only reuse inbox documents newer than this. +BRAIN_SOURCE_INBOX_FRESH_MAX_AGE=168h + # Optional controlled web research through your own SearXNG instance. # Use the root URL or a URL ending in /search. Inside Docker, localhost points # to the Brain container; use the SearXNG service name or host.docker.internal. @@ -168,3 +182,20 @@ BRAIN_AUTONOMOUS_RESEARCH_LEASE=45m BRAIN_AUTONOMOUS_RESEARCH_MAX_ATTEMPTS=3 BRAIN_AUTONOMOUS_RESEARCH_QUERY_TRIGGERS=true BRAIN_AUTONOMOUS_RESEARCH_OPPORTUNITY_LIMIT=8 + +# ----------------------------------------------------------------------------- +# Agent mode only. These values are ignored in BRAIN_MODE=brain. +# Create/rotate the token under /source-agents.html in the Brain UI. +# A local JSON bootstrap file may be used instead of URL/ID/token envs. +# ----------------------------------------------------------------------------- +# BRAIN_MODE=agent +# BRAIN_AGENT_BRAIN_URL=https://brain.example.org +# BRAIN_AGENT_ID=security-news-01 +# BRAIN_AGENT_TOKEN=brain_agent_xxxxxxxxx +# BRAIN_AGENT_CONFIG_FILE= +# BRAIN_AGENT_CONFIG_REFRESH=5m +# BRAIN_AGENT_HTTP_TIMEOUT=30s +# BRAIN_AGENT_CONCURRENCY=3 +# BRAIN_AGENT_BATCH_SIZE=50 +# Allow polling RFC1918/private URLs only for intentionally trusted intranet sources. +# BRAIN_AGENT_ALLOW_PRIVATE=false diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8d802e8..897a47d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -70,3 +70,9 @@ Full Graph ``` Außenkanten werden nach sichtbaren Endpunkten, Relationstyp, Richtung, Herkunft und Status aggregiert. Aktivität öffnet nur den betroffenen Hierarchiepfad. + +## Distributed Source-Agent runtime + +`BRAIN_MODE` selects one of two startup graphs in the same executable. `brain` initializes the Graph, Ollama, knowledge scanner, article pipeline, Source Agent registry and Source Inbox. `agent` initializes only the source-poller runner, local dedupe/config cache and a minimal health/status HTTP server. + +Remote discovery follows `Agent -> authenticated batch ingest -> Source Inbox -> embedding/ANN relevance classification -> candidate archive`. Article evidence acquisition follows `internal KB -> Source Inbox -> SearXNG`, and permanent graph materialization happens only after claim-level grounding. diff --git a/CHANGELOG-SOURCE-AGENT-MODE.md b/CHANGELOG-SOURCE-AGENT-MODE.md new file mode 100644 index 0000000..62c10d6 --- /dev/null +++ b/CHANGELOG-SOURCE-AGENT-MODE.md @@ -0,0 +1,16 @@ +# Changelog: integrated Source-Agent mode + +- Added `BRAIN_MODE=brain|agent` to the single `neural-brain` binary/image. +- Agent mode skips Graph/Ollama/SearXNG/GLPI/THINK initialization. +- Added Brain-side Agent/task/token registry in `source-agents.db`. +- Added per-Agent SHA-256 hashed Bearer tokens with config/heartbeat/ingest-only permissions. +- Added RSS, Atom, sitemap (including bounded sitemap-index traversal) and generic Web polling. +- Source crawling applies SSRF-safe DNS/redirect/private-network checks independently from the Brain control-plane client. +- Added local Agent SQLite state for intervals and URL/content deduplication. +- Added cached centrally managed Agent task configuration. +- Added batch ingest and persistent Source Inbox queue. +- Added bounded Inbox embedding/cluster classification without a chat-model call. +- Adaptive article research and reviewer repair now query the Source Inbox before SearXNG. +- Freshness-sensitive queries apply a separate Inbox age limit. +- Inbox documents are not graph nodes until a reviewer actually grounds a supported article claim in them. +- Added `/source-agents.html` administration UI and `deployment/docker-compose.source-agent.yml`. diff --git a/README.md b/README.md index 602a0d4..743ef99 100644 --- a/README.md +++ b/README.md @@ -352,3 +352,22 @@ Die Artikelrecherche arbeitet ohne `site:`-Filter und ist im Cluster/Fast-Modus ## Cluster/Fast-Verarbeitungsmodus Für große Wissensgraphen kann neben dem bisherigen präzisen Vollscan der Modus `BRAIN_PROCESSING_MODE=clustered` verwendet werden. Er nutzt Semantic Hashing zur Kandidatenvorsortierung, berechnet exakte Cosine-Nähe nur für eine kleine Top-K-Shortlist, bündelt thematisch kompatible Relationen zu gemeinsamen Artikeljobs, nutzt adaptive statt obligatorischer Webrecherche und materialisiert Webquellen erst nach Reviewer-Grounding. Der finale Claim-Review bleibt erhalten. Details: `CLUSTER-FAST-MODE.md` und `ADAPTIVE-ARTICLE-WORKFLOW.md`. + +## Integrated distributed Source Agents + +The project can now run the exact same binary/image as either the full Brain or a lightweight source poller: + +```env +BRAIN_MODE=brain +``` + +or: + +```env +BRAIN_MODE=agent +BRAIN_AGENT_BRAIN_URL=https://brain.example.org +BRAIN_AGENT_ID=security-news-01 +BRAIN_AGENT_TOKEN=brain_agent_... +``` + +Create Agents and RSS/Atom/sitemap/Web polling tasks under `/source-agents.html`. Incoming documents first enter a persistent Source Inbox and are classified against the local KB; adaptive article research searches this Inbox before falling back to SearXNG. Discovered documents are not materialized as graph knowledge until a reviewer actually uses them to ground a supported claim. See `SOURCE-AGENT-MODE.md` for the API, security model and deployment example. diff --git a/SHA256SUMS.txt b/SHA256SUMS.txt index cc11ab1..57fff1f 100644 --- a/SHA256SUMS.txt +++ b/SHA256SUMS.txt @@ -1,10 +1,10 @@ -83729668d3f28a90401b1e39fe80abe71816d209581d1860a06e32b0c2de1ac3 .env.example +a68842eb81f6dd064912ca34e30f94b9473293b8e8b7a77cb96bb59997745093 .env.example 236713daf159ff0a8067e80a442ae3404fa28a5251ae6f24782f263bcfc17005 .gitea/workflows/registry.yml caf5847b0ca972e7701ec23222302ac72de05d20f620d1b0f508efa126f24bfd .gitignore 048f53e6ca01ac583b48784cd2f6f7d248e0534849955b144e75f017f73188a3 .vscode/settings.json 7cab0e9636d745e6a9565659bc062a604f5f0426f7b95cac5d75adcacf1fc65c ADAPTIVE-ARTICLE-WORKFLOW.md 8db46213d020b76bff1b6ea50b8540ed2b13da4bae506295c1dc4dd82e088006 ANALYSIS-DASHBOARD.md -972c1656febf1e22e93472345df9e079e79a602990a72add90096b0d233bf91c ARCHITECTURE.md +d881f2fc86c9ddac4339a5cbeefd5c909cea9552b41a4dba2530a46a4bb88a06 ARCHITECTURE.md 1451a75bb99c33dbfa970e01327d95137dbe27320b69c4a03c36780d1e2da5ef AUTONOMOUS-RESEARCH.md f66de3a0a5abba55f42475a1e79bad2129f033b95648f95fafeb9efe5b7716a8 CHANGELOG-ADAPTIVE-ARTICLE-WORKFLOW.md 7aed2194baab0fb66c561446e5f1cd7724c1166bfa3a08c9e59157d1878bf994 CHANGELOG-ANALYSIS-DASHBOARD.md @@ -27,6 +27,7 @@ fbf686a1acc2de6c4fbb56730a5f87dfdf28d93125fa56ae0c588c29ce492efe CHANGELOG-RESE 87f89a81e1124b18e092a4ea946cb037295cb9e884a46b392286272dc8134dd4 CHANGELOG-RUNTIME-HONEYCOMB.md 2d04e6d385f4b902080a0bcaab510a846a8ae6a76cb757423c50425c8433e7f9 CHANGELOG-SEARXNG-DIAGNOSTICS.md 9c760e167a9af2d3d8ca32c5aa4ef6bb4a3153047343a71c4b19ca9aef96ca32 CHANGELOG-SEARXNG-VISUALIZATION.md +6315493546a2d66022bcdff849e3895b300496cfcbab1ef891c966e55a54cb04 CHANGELOG-SOURCE-AGENT-MODE.md be9f133ae933bdc0e0a8aa5d176dd3e39488a191337043533379e23d179f3ad2 CHANGELOG-SOURCE-ONLY-FILTERS.md 5433a7c2e67ab35fb320bc872e9024fa5f3e765736184e9f878340b8b45407aa CHANGELOG-SQLITE-STARTUP-FIX.md 5b9deeab0cd59b3c649fd73f129361a1e773ed3955cded0048b8cb280bb32e88 CHANGELOG-SQLITE-STORAGE.md @@ -40,9 +41,10 @@ a1aed7c198bc1ffc7af4a8f69ccf137541e4887ce5d59a0d67f2be7216a37dcd FILTER-SCOPES- 696d2da2338cd8190b9614707e4059d78ce291e7334f273633aad815c3b6a6df Makefile e5e9a5268031fee9346462e4631e28ce8c8a8b46a4623e56327a1f310a09f646 OLLAMA-POOL.md 2c0062941ef3edbd40d46b823934a7d0a3a9da7581b83d0b9360e8aaa7694b1b PERSISTENCE.md -8ab574d12ab4f55c1e637ed7807f949faaa89283783280cd09f5b864c67bf3f0 README.md +770a9638e9dc76a93b68de3be809b2dfcd17f00542aaa03b130a631ab200de41 README.md 2838cd19ac2bfa35bebbef2541f631b99221b5997bbb6dbc27146a66a3a1ad34 RUNTIME-CONTROLS-HONEYCOMB.md c3da43b33e550901d55789f2ee526c2e50f61ee028a3f59e0a40e77e1057fde7 SEARXNG-VISUALIZATION.md +498d4174ba4e27e571d8d2ef09c4b977423c4bedea469f0d528840589c6452da SOURCE-AGENT-MODE.md c69419c0327425186cfb84f25746feff226ce813cf467b3f252e729517455047 SOURCE-ONLY-FILTERS.md ae7bc1f1959071f79b76ca8a4ba103064ec0d5c5752af346175731ef4f636d9d SQLITE-STORAGE.md 706b3912716d565082a44a0e707afd2ad07e4eb17cc23cae75772c1740205276 VALIDATION-ADAPTIVE-ARTICLE-WORKFLOW.md @@ -59,16 +61,18 @@ ed62866492c9f62732b6f54a60b0e38174586a873a63e56646c11c0e25b5b2ba VALIDATION-QUE e5d2e3da41bfb6e720f2a9a4b95d0002fb01835db5120674c137f57110312b33 VALIDATION-RESEARCH-ORCHESTRATION.md 044cf894b43d8ce1746adce9e58d75a6c7b0c1f3f3d2f0abcdeaab1db435014d VALIDATION-RESEARCH-PREFETCH-GATE.md e08b0eaab827fe03d5a72327e8fdfe9ce4025097e28208714246969e7d059561 VALIDATION-SEARXNG-DIAGNOSTICS.md +0afab72f31601a8e20eacc3d64a19a6e554ae7116e58e761073318fe5584b5b8 VALIDATION-SOURCE-AGENT-MODE.md f46938b5de7e139e1b21868b1bedce6e312d69cd61602b4f8f43512a327dd422 VALIDATION-SOURCE-ONLY-FILTERS.md 4cd120496664388717fe422a8c54380708df723fea26b35f81799665dfaf2c1c VALIDATION-SQLITE.md f6699e4cdbaadc720e4b8a22c557d02319325283775a2b8a87ff78ca202e3386 VISUALIZATION-PERFORMANCE.md -8970f2abf17bddcd0d84387e8a4975588df2776f03a7a54c5a283470769ca0c8 cmd/brain/main.go +d06d3c03883bde802d755ec36a68ee0394aad1bd49fa7667f155351aa25f511d cmd/brain/main.go e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 data/.gitkeep 7ad10f51cf26747b4f186c540ecb7c77662f001800f0f7803422d45e8d72a4db data/graph.db 587064aa2b583235d2235f4dcff2854b47347d47aef12ce266d56e9da6986843 data/runtime-settings.json 21b51d0e1b7ed07c20f7f3a5da76dedab8df44a94a51724b67b0c3411599fe15 deployment/README.md -033dff450d628d3b722b57815884ab02eb5b25bad9fd21d76dff62d5a455919f deployment/docker-compose.full.yml -c6dfdd509546608829960aee9a06ea3ab1e4f4a7224bb675a9c66fc1b2b493bf docker-compose.yml +0eaa10d68a1a6f9acef0e69c98476e53efac39689f6c1d7febf1994805d21780 deployment/docker-compose.full.yml +6d993bbb8ebebf099fb09b5d554905d1a1b5f1defb6402e8462ca20e31e07c08 deployment/docker-compose.source-agent.yml +3e51e03cd99c8a68ee5d208548f745dd2bb5e8ad5bc37a9798a0de30d641960e docker-compose.yml 1edabd3a7fc60aebaca37fae228a5f34ddbf9ed18478aa1b114204cb956a4027 go.mod 864c3376212497b070feca13d26cfc28e96876078ce7a0b5b0c0470e2dd4fbf8 go.sum ed7fa0e09e94aa9e89c93b00303d4ce6f0f20dac626ecb91181c3aabc76bc8ff integrations/agent/README.md @@ -76,22 +80,22 @@ ed7fa0e09e94aa9e89c93b00303d4ce6f0f20dac626ecb91181c3aabc76bc8ff integrations/a 3c0fc6913501976100521526e1ee8e7988d33fbce3f7b4bab26387d42b0966f5 integrations/knowledgebase/README.md 12f8424f863b13f19aab9c2e6c2828c0ad824a55a62fe336e062db534102bc3a integrations/knowledgebase/glpi-ai-knowledgebase-neural-brain.patch 93d8993e09473559a191c4e01252d6d1fdb214646d65e1271cde018b47d939ed internal/activity/broker.go -1d42342c14b58fdf6b6f650f3c30fc1e11ed6df4b7f2db5beb483e330f6bc90a internal/config/config.go -e08ec1f0465f49d053ac027d3dedea399ccb014171c33fcf1b20ca21cf493515 internal/config/config_test.go -b2cc8158185790189d1c508c37a5a8f146a22ff9510b31f1bcef8625230b4a63 internal/engine/article.go -6f86679fe337aa75aaa9422f9361208328a1988836ea24731cb81b72c7712ffa internal/engine/article_adaptive.go +57c30b450db2534accdd6d1bc07e921b36ee45c94eb7007cefd391460d96446a internal/config/config.go +6a5fdbc3b849363f80a4bef57a9028e4e1eefbbb96cec61f670a0ea1efe9dc28 internal/config/config_test.go +a03f945c1ce44480f21f855b25060c34f0a5132e537cb12787af34185f7b438c internal/engine/article.go +cfb3ac33a192391d8f85f77c23d5c4e277e03618cf32cce6269115b8c238b311 internal/engine/article_adaptive.go 9658461b16b32e664336018d1c7e068bc92dcd77b07f4ecc6a5f9abdc27c3dde internal/engine/article_adaptive_test.go 1825f890a6f033a64908233e8ca0e5d2b0b556c154cda2c74ff70b8e6fbbdcb1 internal/engine/article_batch.go b24b6b932eca7b73af7bcbdf552b139787ad4b7bf9e26ad1a3550e968dfd795d internal/engine/article_format_test.go -4c149a2c55515e5c413c66b88d67a85b027562d776fa6b5c9ea05be058ce1923 internal/engine/article_generate_then_review.go +77934cde83b3eeb052889cda703f5d0da0bfa12366d05fe9850afcaae7c13a66 internal/engine/article_generate_then_review.go 72d12cf78cbdcf7b476466462c624acbdb29b8a8903356dad81d5d3c9c20360e internal/engine/article_generate_then_review_test.go -69aa32d7c2c7494eaa2eeb9ffd100caa059384afa46d11f46ca5a6c95e1d7c29 internal/engine/article_research.go +b83bda20e2c37516c9f3159a04e463cd02583a97f8a1d364c66f949a86e8da27 internal/engine/article_research.go 69418f872101c9f055d518ee8a881d2af3cfebf438911e6c01cfc92c629e41ec internal/engine/article_research_cache.go 3ca7037231935329d49b6b80553fbfef206c079a6aed4c2379dadd46d39ebc0d internal/engine/article_research_cache_test.go 50e5bab3e2dfa4e64706035d7a858397112baf1db477022b50d1366d0effc932 internal/engine/article_research_test.go a4451f7712ca281e677e4a26ffda16d2a1f7dfd7ffcb658d77cfedcc563d5e9f internal/engine/autonomous_research.go 62a69f1af6fc3842e34f3847a5f868f81202feaaa6429e4e84db8115f5d14ad8 internal/engine/autonomous_research_test.go -72de9be17ff516caf5cecdbaba6691ed627a07f4a823843179873b7559f7b181 internal/engine/engine.go +443f1d6195946d73c04ed1a1086ec133967cb40d6830623b0863f9d9795b39c7 internal/engine/engine.go 0161090ea1ee9e22d9372a69b80131818190da6a3c0c850eae8356bc1763f032 internal/engine/engine_test.go 2038a3909b147a630e361faae3f5c1cc22218d0f1336c91d4c1b795c52665e9b internal/engine/research_diagnostics.go 0eb3f00e2ab73d2dc6a4cbc1a4038533a4bb20fbbbb6190abd39feff6b34b8e1 internal/engine/research_diagnostics_test.go @@ -100,6 +104,7 @@ a4451f7712ca281e677e4a26ffda16d2a1f7dfd7ffcb658d77cfedcc563d5e9f internal/engin 87a96d4a68e8e19bc15e787efa6d24fee3ae483aeb49910c6dd699a108d0a594 internal/engine/research_work_test.go 4788163f25060322da85597f6cf5d4881463497ca9a9ddfbefaf5a3a3d985adf internal/engine/runtime.go 02809a7dbbf10bdc086316314c57ddd4f20f7b7f1dffbb82fc9cc8927560a2b7 internal/engine/runtime_filter_test.go +13694ea5e73a614bc7b1ab7cd7d573eeae8c1c902a01bd3ae32b93af0179b975 internal/engine/source_inbox.go b82980a646a92751bdd27a866ba1ffc6d34a3ba81d537f7b6e5a78e1432ee6fa internal/glpi/client.go 525102be56bc51ce8a08655b1b2bb53b67f4a1828903585fd664ed6a5133f617 internal/glpi/client_test.go 5ec61f7f830721bdfb78d99617d5c5e8da0e15f8ab6e8ee18b7e7956b0c2c8bf internal/graph/analysis_dashboard.go @@ -124,18 +129,26 @@ a13910fb417484d56e78ae856b71fb66c63987d9abf3b513190bc52697321a18 internal/inges 40194b1b2e44a1c8796fd934346e4cd8ecf374ef2318ff1d80822e55e12afb47 internal/ollama/client_test.go 0bc8bb4c698c2c5dbef5c980d3e3fb88f10d35a8d7a230cbc81d218798bc1fe6 internal/persist/coordinator.go 46144aa719ff5c3ab787c214d8e32e4b3c6883bed068410e5403a26f6352cd6a internal/persist/coordinator_test.go -e7138877303bcc07c429323c4792fed112d8c16d35fd44222fe144ab0b69344f internal/research/fetch.go +09fc87e1e6147e0f3d221b51f6a5a39712a113ad16ce4048fad1b0fc3d5c82a9 internal/research/fetch.go 6a9f269783a7c41d5b63f9bd5022f415ed1b5572041ca5ccae1512d6dc2a0b80 internal/research/fetch_test.go edd033455bfd3925e5cdf5183a363ad527e7cf81beca183339fdc0b5410349ee internal/research/searxng.go 5a1155da5809f3a5dbd99a98094c5778424e3f4bf93fa14bd442e91cdcc08094 internal/research/searxng_test.go -984966a437944530008f4888944a91130604aad8891da3f67a89656bc991d9e4 internal/web/server.go +999720ab3db41e394f230ea9c29a37a9e29aec491a3f611a4381d288aa0ba86a internal/sourceagent/agent.go +05f918e48e58bc08fe365a95f79950e4d7ffb5d8dc891067618448673facebae internal/sourceagent/sourceagent_test.go +04e29ecf7370322256c5ec2f57bd7c2977e87434bbf1be8093bf513d53202217 internal/sourceagent/store.go +5761310d3810c57be0efd4d1f3a54b8711a7d23de75933a41e001f1d570f9b26 internal/sourceagent/types.go +769b4abedc8ae8762fe8441fc92dfa1cadf02e0480898853abe9ad383273f08e internal/web/server.go b5abd1c3591242a7e8835eb38866410558d0e7901c75f5b11b94039ee3747716 internal/web/server_test.go +b29a185248803336c8b46751b798cc03987d9d91c44d11b4e9fcb8a9f35173db internal/web/source_agents.go 3e27efdbeaf8aba34864f6d1dd47d101d04c87993d02e35ee08df34affaefe29 internal/web/static/analysis.css db27a3c62848dbb0f383886c1075d2c0c779363cea3e847793104ca708c1d6f0 internal/web/static/analysis.html 2ebcc27579c4fc477976d99a6c8116e7b4798b59d896dfcf6e4450cacfa7d4b7 internal/web/static/analysis.js 5887106080718a2cd3d97baddb37e09e9bc733768b567f553cce8084d1dec8bb internal/web/static/app.css f2afbfe0818847f6bab26ddc3279b60e8155f298db9e306aeefef86f0002bc06 internal/web/static/app.js -790abdd2ca39a0257c19fbb0ec4aa1360b98d8f86f0d423139d8ad8d2de31a99 internal/web/static/index.html +699a6b744cec5bfd4aafa7e736f28132700bc3af0bbbceefe1e1f650d431e739 internal/web/static/index.html +844021e5004b139377587973ff9027c463f454186fe6d331d8b5f7cda7f19a06 internal/web/static/source-agents.css +deb8d8b6a77bf4705d836f79fd80950182880c46fc0f726b22cdbe6766994487 internal/web/static/source-agents.html +6a1bf0eb066bfeec3ffcac2f092c71d657c98cc0963c981459852e3e31d1e260 internal/web/static/source-agents.js ee527efd31cc069b08ebbd53d3df7f6374cb245ae64e7e25278dc0b6381aefa4 internal/workqueue/limiter.go 12209426f68411da5bd793a2c499992e914fc5de9ab48fa3e0c4c89215d77b9e internal/workqueue/limiter_test.go 83aded814b6225395935e61fe957963c3c470f368fc9089f505b6de23e959115 preview.png diff --git a/SOURCE-AGENT-MODE.md b/SOURCE-AGENT-MODE.md new file mode 100644 index 0000000..eb14dd9 --- /dev/null +++ b/SOURCE-AGENT-MODE.md @@ -0,0 +1,123 @@ +# Integrated Source-Agent mode + +The same `neural-brain` binary and Docker image now has two runtime roles: + +```env +BRAIN_MODE=brain +``` + +starts the full graph/AI application, while + +```env +BRAIN_MODE=agent +``` + +starts only the distributed source poller. Agent mode does **not** initialize the graph, Ollama, SearXNG, GLPI sync, AI-THINK or article synthesis. + +## Brain-side flow + +The Brain owns Agent identities, source tasks and the Source Inbox in `source-agents.db`. + +1. Open `/source-agents.html`. +2. Create an Agent. The plaintext token is returned exactly when it is created or rotated; only its SHA-256 hash is stored by the Brain. +3. Add RSS, Atom, sitemap or generic Web tasks and a polling interval. +4. The remote Agent obtains its task list from `GET /api/v1/agent/config`. +5. New documents are sent in batches to `POST /api/v1/agent/ingest`. +6. Documents enter the Inbox as `received`, not as graph nodes. +7. The Brain embeds a bounded Inbox batch and checks it against the existing KB. Relevant documents become `candidate`; weak matches become `archived`. +8. Adaptive article/reviewer research queries the `candidate` Inbox before SearXNG. For freshness-sensitive requests, only sufficiently recent Inbox documents are reused. +9. A Web document becomes a graph node only when a generated article actually grounds a supported claim in it. Then the Inbox row becomes `used`. + +This deliberately separates `discovered` from `knowledge`. + +## Agent bootstrap + +Minimal remote Agent configuration: + +```env +BRAIN_MODE=agent +BRAIN_AGENT_BRAIN_URL=https://brain.example.org +BRAIN_AGENT_ID=security-news-01 +BRAIN_AGENT_TOKEN=brain_agent_... +``` + +Optional Agent settings: + +```env +BRAIN_AGENT_CONFIG_REFRESH=5m +BRAIN_AGENT_HTTP_TIMEOUT=30s +BRAIN_AGENT_CONCURRENCY=3 +BRAIN_AGENT_BATCH_SIZE=50 +BRAIN_AGENT_ALLOW_PRIVATE=false +``` + +The Agent caches the last successful remote task configuration in its data volume. Source polling uses SSRF-safe DNS/redirect handling and rejects private, loopback and link-local targets by default; set `BRAIN_AGENT_ALLOW_PRIVATE=true` only for intentionally trusted intranet sources. This restriction applies to source URLs, not to the Brain control-plane URL, so the Brain itself may live on a private LAN. + +A local bootstrap JSON can be used instead of the three connection ENV values: + +```json +{ + "brain_url": "https://brain.example.org", + "agent_id": "security-news-01", + "token": "brain_agent_..." +} +``` + +and: + +```env +BRAIN_MODE=agent +BRAIN_AGENT_CONFIG_FILE=/run/secrets/brain-agent.json +``` + +## Source task schema + +Example task returned by the Brain: + +```json +{ + "id": "microsoft-security", + "agent_id": "security-news-01", + "name": "Microsoft Security", + "type": "rss", + "url": "https://example.org/feed.xml", + "enabled": true, + "poll_interval": "2h", + "categories": ["Microsoft", "Security"], + "max_items": 20, + "config": { + "refetch_seen": "false" + } +} +``` + +Supported task types in v1 are `rss`, `atom`, `sitemap` and `web`. Set `config.refetch_seen=true` for sources whose existing URLs are expected to change (for example rolling advisories). Otherwise the Agent optimizes for discovery of new article URLs. + +## Agent API permissions + +Agent Bearer tokens are accepted only on: + +- `GET /api/v1/agent/config` +- `POST /api/v1/agent/heartbeat` +- `POST /api/v1/agent/ingest` + +They are separate from `BRAIN_API_KEY` and do not authorize graph, runtime, research, THINK or administration APIs. Agent-management and Source-Inbox management endpoints use the normal `BRAIN_API_KEY` whenever it is configured. + +## Source Inbox tuning + +Brain-side defaults: + +```env +BRAIN_SOURCE_INBOX_ENABLED=true +BRAIN_SOURCE_INBOX_INTERVAL=30s +BRAIN_SOURCE_INBOX_BATCH_SIZE=12 +BRAIN_SOURCE_INBOX_MIN_SIMILARITY=0.55 +BRAIN_SOURCE_INBOX_MIN_RESULTS=2 +BRAIN_SOURCE_INBOX_FRESH_MAX_AGE=168h +``` + +Classification is deliberately cheap: one batched embedding operation plus the configured precise/clustered nearest-neighbour retrieval. There is no Gemma/Qwen call just to classify every incoming news item. + +## Docker + +`deployment/docker-compose.source-agent.yml` is a minimal example that builds the exact same project with `BRAIN_MODE=agent`. The Agent only needs a persistent `/app/data` volume for local dedupe state and its cached remote configuration. diff --git a/VALIDATION-SOURCE-AGENT-MODE.md b/VALIDATION-SOURCE-AGENT-MODE.md new file mode 100644 index 0000000..342203a --- /dev/null +++ b/VALIDATION-SOURCE-AGENT-MODE.md @@ -0,0 +1,16 @@ +# Validation: integrated Source-Agent mode + +Validated in the available offline build environment: + +- Agent/Brain role configuration parsing and Agent-mode dependency bypass. +- Poll task validation for RSS/Atom/sitemap/Web, interval/URL rejection cases, and rejection of URL userinfo. +- Source-fetch HTTP uses the same DNS/redirect/private-network SSRF protections as Brain research fetches; Brain API connectivity remains a separate client so private Brain URLs still work. +- Document normalization, SHA-256 creation and semantic-hash sanity check. +- Compile checks for Source-Agent store/runner, Brain Source-Inbox worker, Agent HTTP API, UI integration and main role switch. +- Existing Engine/Web packages compile with the new Source-Inbox dependency. +- Source-agent management/inbox reads require the normal Brain API key whenever `BRAIN_API_KEY` is configured; Agent bearer tokens remain restricted to config/heartbeat/ingest. +- `source-agents.js` passes `node --check`. +- `go vet` is run in the temporary Go 1.23 compile copy. +- Patch application and ZIP integrity are checked for the release. + +Environment limitation: the project requires Go 1.26 and `modernc.org/sqlite v1.37.1`, while the local runner has Go 1.23 and no network access. As with the preceding releases, project-wide compile/vet validation uses an isolated copy with the compile-only SQLite stub. Consequently the new SQLite Agent/Inbox schema is compile-checked but cannot be exercised against the real modernc SQLite driver in this environment. The shipped `go.mod` remains unchanged at Go 1.26 with the real SQLite dependency. diff --git a/cmd/brain/main.go b/cmd/brain/main.go index 7563097..c595e34 100644 --- a/cmd/brain/main.go +++ b/cmd/brain/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "errors" "log/slog" "net/http" @@ -15,9 +16,12 @@ import ( "github.com/local/glpi-neural-brain/internal/engine" "github.com/local/glpi-neural-brain/internal/graph" "github.com/local/glpi-neural-brain/internal/ingest" + "github.com/local/glpi-neural-brain/internal/sourceagent" webui "github.com/local/glpi-neural-brain/internal/web" ) +const buildVersion = "source-agent-integrated-v1" + func main() { cfg, err := config.Load() if err != nil { @@ -25,23 +29,70 @@ func main() { os.Exit(1) } slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo}))) + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + if cfg.Mode == "agent" { + runAgent(ctx, cfg) + return + } + runBrain(ctx, cancel, cfg) +} + +func runAgent(ctx context.Context, cfg config.Config) { + runner, err := sourceagent.NewRunner(sourceagent.RunnerConfig{ + BrainURL: cfg.AgentBrainURL, AgentID: cfg.AgentID, Token: cfg.AgentToken, DataDir: cfg.DataDir, + ConfigFile: cfg.AgentConfigFile, ConfigRefresh: cfg.AgentConfigRefresh, HTTPTimeout: cfg.AgentHTTPTimeout, + Concurrency: cfg.AgentConcurrency, BatchSize: cfg.AgentBatchSize, AllowPrivate: cfg.AgentAllowPrivate, Version: buildVersion, + }) + if err != nil { + slog.Error("source agent initialization failed", "error", err) + os.Exit(1) + } + defer runner.Close() + runner.Start(ctx) + mux := http.NewServeMux() + mux.HandleFunc("GET /api/status", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(runner.Status()) + }) + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }) + srv := &http.Server{Addr: cfg.ListenAddr, Handler: mux, ReadHeaderTimeout: 5 * time.Second, IdleTimeout: 30 * time.Second} + go func() { + slog.Info("neural brain source-agent listening", "addr", cfg.ListenAddr, "agent_id", cfg.AgentID, "brain_url", cfg.AgentBrainURL) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("agent status server failed", "error", err) + } + }() + <-ctx.Done() + shutdown, c := context.WithTimeout(context.Background(), 10*time.Second) + defer c() + _ = srv.Shutdown(shutdown) +} + +func runBrain(ctx context.Context, cancel context.CancelFunc, cfg config.Config) { g, err := graph.Open(cfg.DataDir) if err != nil { slog.Error("graph open failed", "error", err) os.Exit(1) } broker := activity.New(120) + sourceStore, err := sourceagent.OpenStore(cfg.DataDir) + if err != nil { + slog.Error("source agent store open failed", "error", err) + _ = g.Close() + os.Exit(1) + } + defer sourceStore.Close() eng := engine.New(cfg, g, broker) - ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) - defer cancel() + eng.SetSourceInbox(sourceStore) eng.Start(ctx) watcher := ingest.NewAgentWatcher(cfg.AgentRunsFiles, g, broker) watcher.Start(ctx) - ui := &webui.Server{Engine: eng, Graph: g, Broker: broker, APIKey: cfg.APIKey} + ui := &webui.Server{Engine: eng, Graph: g, Broker: broker, APIKey: cfg.APIKey, SourceAgents: sourceStore} srv := &http.Server{Addr: cfg.ListenAddr, Handler: ui.Handler(), ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 30 * time.Second, WriteTimeout: 10 * time.Minute, IdleTimeout: 90 * time.Second} go func() { storage := g.StorageStatus() - slog.Info("neural brain listening", "addr", cfg.ListenAddr, "knowledge_dirs", cfg.KnowledgeDirs, "staging_dirs", cfg.StagingDirs, "graph_backend", storage.Backend, "graph_db", storage.Path) + slog.Info("neural brain listening", "mode", "brain", "addr", cfg.ListenAddr, "knowledge_dirs", cfg.KnowledgeDirs, "staging_dirs", cfg.StagingDirs, "graph_backend", storage.Backend, "graph_db", storage.Path) if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { slog.Error("server failed", "error", err) cancel() diff --git a/deployment/docker-compose.full.yml b/deployment/docker-compose.full.yml index 4978e81..f82d7d8 100644 --- a/deployment/docker-compose.full.yml +++ b/deployment/docker-compose.full.yml @@ -106,6 +106,7 @@ services: ports: - "${BRAIN_PORT:-8090}:8090" environment: + BRAIN_MODE: brain BRAIN_LISTEN_ADDR: :8090 BRAIN_DATA_DIR: /app/data BRAIN_KNOWLEDGE_DIRS: /sources/knowledge diff --git a/deployment/docker-compose.source-agent.yml b/deployment/docker-compose.source-agent.yml new file mode 100644 index 0000000..3520164 --- /dev/null +++ b/deployment/docker-compose.source-agent.yml @@ -0,0 +1,28 @@ +# Same image/binary as the Brain, but with the lightweight source-poller runtime. +# Create the Agent + token first in https:///source-agents.html. +services: + source-agent: + build: + context: .. + restart: unless-stopped + environment: + BRAIN_MODE: agent + BRAIN_LISTEN_ADDR: :8090 + BRAIN_DATA_DIR: /app/data + BRAIN_AGENT_BRAIN_URL: ${BRAIN_AGENT_BRAIN_URL:?set Brain URL} + BRAIN_AGENT_ID: ${BRAIN_AGENT_ID:?set Agent ID} + BRAIN_AGENT_TOKEN: ${BRAIN_AGENT_TOKEN:?set Agent token} + BRAIN_AGENT_CONFIG_REFRESH: ${BRAIN_AGENT_CONFIG_REFRESH:-5m} + BRAIN_AGENT_HTTP_TIMEOUT: ${BRAIN_AGENT_HTTP_TIMEOUT:-30s} + BRAIN_AGENT_CONCURRENCY: ${BRAIN_AGENT_CONCURRENCY:-3} + BRAIN_AGENT_BATCH_SIZE: ${BRAIN_AGENT_BATCH_SIZE:-50} + BRAIN_AGENT_ALLOW_PRIVATE: ${BRAIN_AGENT_ALLOW_PRIVATE:-false} + volumes: + - source-agent-state:/app/data + read_only: true + tmpfs: [/tmp:size=32m] + security_opt: [no-new-privileges:true] + cap_drop: [ALL] + +volumes: + source-agent-state: diff --git a/docker-compose.yml b/docker-compose.yml index 4651532..040ba1c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -6,6 +6,7 @@ services: ports: - "8090:8090" environment: + BRAIN_MODE: ${BRAIN_MODE:-brain} BRAIN_LISTEN_ADDR: ":8090" BRAIN_DATA_DIR: /app/data BRAIN_KNOWLEDGE_DIRS: /sources/knowledge @@ -85,6 +86,12 @@ services: BRAIN_TOP_K: ${BRAIN_TOP_K:-8} BRAIN_MAX_CONTEXT_CHARS: ${BRAIN_MAX_CONTEXT_CHARS:-16000} BRAIN_RESEARCH_ENABLED: ${BRAIN_RESEARCH_ENABLED:-false} + BRAIN_SOURCE_INBOX_ENABLED: ${BRAIN_SOURCE_INBOX_ENABLED:-true} + BRAIN_SOURCE_INBOX_INTERVAL: ${BRAIN_SOURCE_INBOX_INTERVAL:-30s} + BRAIN_SOURCE_INBOX_BATCH_SIZE: ${BRAIN_SOURCE_INBOX_BATCH_SIZE:-12} + BRAIN_SOURCE_INBOX_MIN_SIMILARITY: ${BRAIN_SOURCE_INBOX_MIN_SIMILARITY:-0.55} + BRAIN_SOURCE_INBOX_MIN_RESULTS: ${BRAIN_SOURCE_INBOX_MIN_RESULTS:-2} + BRAIN_SOURCE_INBOX_FRESH_MAX_AGE: ${BRAIN_SOURCE_INBOX_FRESH_MAX_AGE:-168h} BRAIN_AUTONOMOUS_RESEARCH_ENABLED: ${BRAIN_AUTONOMOUS_RESEARCH_ENABLED:-false} BRAIN_AUTONOMOUS_RESEARCH_IDLE_ONLY: ${BRAIN_AUTONOMOUS_RESEARCH_IDLE_ONLY:-true} BRAIN_AUTONOMOUS_RESEARCH_INTERVAL: ${BRAIN_AUTONOMOUS_RESEARCH_INTERVAL:-30m} diff --git a/internal/config/config.go b/internal/config/config.go index d07cb15..d4bc7c5 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -11,6 +11,7 @@ import ( ) type Config struct { + Mode string ListenAddr string DataDir string KnowledgeDirs []string @@ -108,6 +109,22 @@ type Config struct { LowPowerMode bool RuntimeDefaultsConfigured bool + SourceInboxEnabled bool + SourceInboxInterval time.Duration + SourceInboxBatchSize int + SourceInboxMinSimilarity float64 + SourceInboxMinResults int + SourceInboxFreshMaxAge time.Duration + AgentBrainURL string + AgentID string + AgentToken string + AgentConfigFile string + AgentConfigRefresh time.Duration + AgentHTTPTimeout time.Duration + AgentConcurrency int + AgentBatchSize int + AgentAllowPrivate bool + GLPIKBEnabled bool GLPIURL string GLPIAPIVersion string @@ -139,6 +156,7 @@ func Load() (Config, error) { ollamaURLs[i] = strings.TrimRight(ollamaURLs[i], "/") } cfg := Config{ + Mode: strings.ToLower(env("BRAIN_MODE", "brain")), ListenAddr: env("BRAIN_LISTEN_ADDR", ":8090"), DataDir: abs, KnowledgeDirs: paths("BRAIN_KNOWLEDGE_DIRS"), @@ -235,6 +253,21 @@ func Load() (Config, error) { MaxDisplayNodes: integer("BRAIN_MAX_DISPLAY_NODES", 0), LowPowerMode: boolean("BRAIN_LOW_POWER_MODE", false), RuntimeDefaultsConfigured: true, + SourceInboxEnabled: boolean("BRAIN_SOURCE_INBOX_ENABLED", true), + SourceInboxInterval: duration("BRAIN_SOURCE_INBOX_INTERVAL", 30*time.Second), + SourceInboxBatchSize: integer("BRAIN_SOURCE_INBOX_BATCH_SIZE", 12), + SourceInboxMinSimilarity: number("BRAIN_SOURCE_INBOX_MIN_SIMILARITY", 0.55), + SourceInboxMinResults: integer("BRAIN_SOURCE_INBOX_MIN_RESULTS", 2), + SourceInboxFreshMaxAge: duration("BRAIN_SOURCE_INBOX_FRESH_MAX_AGE", 168*time.Hour), + AgentBrainURL: strings.TrimRight(strings.TrimSpace(os.Getenv("BRAIN_AGENT_BRAIN_URL")), "/"), + AgentID: strings.TrimSpace(os.Getenv("BRAIN_AGENT_ID")), + AgentToken: strings.TrimSpace(os.Getenv("BRAIN_AGENT_TOKEN")), + AgentConfigFile: strings.TrimSpace(os.Getenv("BRAIN_AGENT_CONFIG_FILE")), + AgentConfigRefresh: duration("BRAIN_AGENT_CONFIG_REFRESH", 5*time.Minute), + AgentHTTPTimeout: duration("BRAIN_AGENT_HTTP_TIMEOUT", 30*time.Second), + AgentConcurrency: integer("BRAIN_AGENT_CONCURRENCY", 3), + AgentBatchSize: integer("BRAIN_AGENT_BATCH_SIZE", 50), + AgentAllowPrivate: boolean("BRAIN_AGENT_ALLOW_PRIVATE", false), GLPIKBEnabled: boolean("GLPI_KB_ENABLED", false), GLPIURL: strings.TrimRight(strings.TrimSpace(os.Getenv("GLPI_URL")), "/"), GLPIAPIVersion: env("GLPI_API_VERSION", "v2.3"), @@ -250,6 +283,45 @@ func Load() (Config, error) { GLPIKBSyncInterval: duration("GLPI_KB_SYNC_INTERVAL", 10*time.Minute), GLPIKBSource: env("GLPI_KB_SOURCE", "GLPI Knowledge Base"), } + if cfg.Mode != "brain" && cfg.Mode != "agent" { + return Config{}, fmt.Errorf("BRAIN_MODE must be brain or agent") + } + if cfg.Mode == "agent" { + if cfg.AgentConfigRefresh < time.Minute || cfg.AgentConfigRefresh > 24*time.Hour { + return Config{}, fmt.Errorf("BRAIN_AGENT_CONFIG_REFRESH must be between 1m and 24h") + } + if cfg.AgentHTTPTimeout < 5*time.Second || cfg.AgentHTTPTimeout > 5*time.Minute { + return Config{}, fmt.Errorf("BRAIN_AGENT_HTTP_TIMEOUT must be between 5s and 5m") + } + if cfg.AgentConcurrency < 1 || cfg.AgentConcurrency > 16 { + return Config{}, fmt.Errorf("BRAIN_AGENT_CONCURRENCY must be between 1 and 16") + } + if cfg.AgentBatchSize < 1 || cfg.AgentBatchSize > 500 { + return Config{}, fmt.Errorf("BRAIN_AGENT_BATCH_SIZE must be between 1 and 500") + } + if cfg.AgentConfigFile == "" && (cfg.AgentBrainURL == "" || cfg.AgentID == "" || cfg.AgentToken == "") { + return Config{}, fmt.Errorf("agent mode requires BRAIN_AGENT_BRAIN_URL, BRAIN_AGENT_ID and BRAIN_AGENT_TOKEN or BRAIN_AGENT_CONFIG_FILE") + } + if err := os.MkdirAll(cfg.DataDir, 0o750); err != nil { + return Config{}, err + } + return cfg, nil + } + if cfg.SourceInboxInterval < 5*time.Second || cfg.SourceInboxInterval > 24*time.Hour { + return Config{}, fmt.Errorf("BRAIN_SOURCE_INBOX_INTERVAL must be between 5s and 24h") + } + if cfg.SourceInboxBatchSize < 1 || cfg.SourceInboxBatchSize > 100 { + return Config{}, fmt.Errorf("BRAIN_SOURCE_INBOX_BATCH_SIZE must be between 1 and 100") + } + if cfg.SourceInboxMinSimilarity < 0 || cfg.SourceInboxMinSimilarity > 1 { + return Config{}, fmt.Errorf("BRAIN_SOURCE_INBOX_MIN_SIMILARITY must be between 0 and 1") + } + if cfg.SourceInboxMinResults < 1 || cfg.SourceInboxMinResults > 20 { + return Config{}, fmt.Errorf("BRAIN_SOURCE_INBOX_MIN_RESULTS must be between 1 and 20") + } + if cfg.SourceInboxFreshMaxAge < time.Hour || cfg.SourceInboxFreshMaxAge > 365*24*time.Hour { + return Config{}, fmt.Errorf("BRAIN_SOURCE_INBOX_FRESH_MAX_AGE must be between 1h and 8760h") + } if cfg.ScanInterval < 2*time.Second { return Config{}, fmt.Errorf("BRAIN_SCAN_INTERVAL must be at least 2s") } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 4342a6e..0937bcd 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -187,3 +187,32 @@ func TestLoadRejectsInvalidArticleResearchStrategy(t *testing.T) { t.Fatal("expected invalid article research strategy error") } } + +func TestLoadAgentModeDoesNotRequireBrainDependencies(t *testing.T) { + t.Setenv("BRAIN_DATA_DIR", t.TempDir()) + t.Setenv("BRAIN_MODE", "agent") + t.Setenv("BRAIN_AGENT_BRAIN_URL", "http://brain:8090") + t.Setenv("BRAIN_AGENT_ID", "poller-1") + t.Setenv("BRAIN_AGENT_TOKEN", "brain_agent_test") + t.Setenv("OLLAMA_URLS", "not-a-url") + t.Setenv("BRAIN_AUTONOMOUS_RESEARCH_ENABLED", "true") + cfg, err := Load() + if err != nil { + t.Fatal(err) + } + if cfg.Mode != "agent" || cfg.AgentID != "poller-1" || cfg.AgentBrainURL != "http://brain:8090" { + t.Fatalf("unexpected agent config: %+v", cfg) + } +} + +func TestLoadAgentModeRequiresBootstrapConnection(t *testing.T) { + t.Setenv("BRAIN_DATA_DIR", t.TempDir()) + t.Setenv("BRAIN_MODE", "agent") + t.Setenv("BRAIN_AGENT_BRAIN_URL", "") + t.Setenv("BRAIN_AGENT_ID", "") + t.Setenv("BRAIN_AGENT_TOKEN", "") + t.Setenv("BRAIN_AGENT_CONFIG_FILE", "") + if _, err := Load(); err == nil { + t.Fatal("expected missing agent bootstrap configuration error") + } +} diff --git a/internal/engine/article.go b/internal/engine/article.go index 49541ab..1bb0335 100644 --- a/internal/engine/article.go +++ b/internal/engine/article.go @@ -139,7 +139,7 @@ func (e *Engine) synthesizeKnowledgeArticle(ctx context.Context, trigger string, researchStrategy := e.effectiveArticleResearchStrategy() freshness := detectArticleFreshnessNeed(plan, relation, selected) initialWebResearch := false - if e.ResearchEnabledForRuntime() { + if e.evidenceAcquisitionEnabled() { switch researchStrategy { case "always": collected, report, researchErr := e.collectResearchMaterialForArticle(ctx, trigger, plan.SourceNodeIDs, selected, plan, brief, researchResults) @@ -160,7 +160,7 @@ func (e *Engine) synthesizeKnowledgeArticle(ctx context.Context, trigger string, } } } - e.Broker.Publish(model.Activity{Type: "article.research.strategy", Source: "brain", Phase: "knowledge-research-routing", NodeIDs: plan.SourceNodeIDs, Message: fmt.Sprintf("Artikelrecherche: %s · initiales Webmaterial: %t", researchStrategy, initialWebResearch), Strength: .44, Metadata: map[string]any{"trigger": trigger, "research_strategy": researchStrategy, "freshness_required": freshness.Required, "freshness_reason": freshness.Reason, "initial_web_research": initialWebResearch, "initial_research_queries": researchReport.Queries, "initial_research_fetched": researchReport.Fetched}}) + e.Broker.Publish(model.Activity{Type: "article.research.strategy", Source: "brain", Phase: "knowledge-research-routing", NodeIDs: plan.SourceNodeIDs, Message: fmt.Sprintf("Artikelrecherche: %s · initiales Webmaterial: %t", researchStrategy, initialWebResearch), Strength: .44, Metadata: map[string]any{"trigger": trigger, "research_strategy": researchStrategy, "freshness_required": freshness.Required, "freshness_reason": freshness.Reason, "initial_web_research": initialWebResearch, "initial_research_queries": researchReport.Queries, "source_inbox_results": researchReport.InboxResults, "initial_research_fetched": researchReport.Fetched}}) e.Broker.Publish(model.Activity{Type: "article.draft.started", Source: "brain", Phase: "knowledge-synthesis", NodeIDs: plan.SourceNodeIDs, Message: fmt.Sprintf("%s erstellt zuerst aus dem verfügbaren Evidenzsatz einen KB-Artikel; Webrecherche erfolgt nur bei Bedarf", e.Cfg.ArticleSynthesisModel), Strength: .95, Metadata: map[string]any{"trigger": trigger, "action": plan.Action, "article_type": plan.ArticleType, "target_article_id": plan.TargetArticleID, "source_count": len(selected), "research_material_count": len(researchResults), "research_rounds": researchReport.Rounds, "research_fetched": researchReport.Fetched, "research_strategy": researchStrategy, "freshness_required": freshness.Required, "synthesis_model": e.Cfg.ArticleSynthesisModel, "review_model": e.Cfg.ArticleReviewModel, "generation_depth": generationDepth}}) @@ -187,7 +187,7 @@ func (e *Engine) synthesizeKnowledgeArticle(ctx context.Context, trigger string, // In adaptive mode Gemma is allowed to say that the internal KB is not // sufficient. Only then do we pay for a focused Web round. This happens // before the reviewer, so an obvious evidence gap does not waste a Qwen call. - if researchStrategy == "adaptive" && !authorResearchAttempted && !freshness.Required && e.ResearchEnabledForRuntime() && (content.ResearchNeeded || content.FreshnessSensitive) { + if researchStrategy == "adaptive" && !authorResearchAttempted && !freshness.Required && e.evidenceAcquisitionEnabled() && (content.ResearchNeeded || content.FreshnessSensitive) { queries := append([]string{}, content.ResearchQueries...) if content.ResearchNeeded && len(queries) == 0 { queries = append(queries, plan.ResearchQuery) @@ -235,11 +235,11 @@ func (e *Engine) synthesizeKnowledgeArticle(ctx context.Context, trigger string, break } repairAttempts++ - if len(quality.MissingEvidenceQueries) > 0 && e.ResearchEnabledForRuntime() { + if len(quality.MissingEvidenceQueries) > 0 && e.evidenceAcquisitionEnabled() { e.Broker.Publish(model.Activity{Type: "article.review.research.started", Source: "brain", Phase: "quality-repair-research", NodeIDs: draft.SourceNodeIDs, Message: "Der Reviewer hat konkrete unbelegte Aussagen gefunden · nur diese Punkte werden nachrecherchiert", Strength: .84, Metadata: map[string]any{"trigger": trigger, "repair_round": repairAttempts, "queries": quality.MissingEvidenceQueries, "review_model": e.Cfg.ArticleReviewModel}}) additional, repairReport := e.collectReviewerRepairResearch(ctx, trigger, plan.SourceNodeIDs, quality.MissingEvidenceQueries, repairAttempts, attemptedRepairURLs) researchResults = uniqueResearchEvidence(append(researchResults, additional...)) - e.Broker.Publish(model.Activity{Type: "article.review.research.completed", Source: "brain", Phase: "quality-repair-research", NodeIDs: draft.SourceNodeIDs, Message: fmt.Sprintf("Gezielte Nachrecherche beendet · %d zusätzliche Volltextquellen", len(additional)), Strength: .82, Metadata: map[string]any{"trigger": trigger, "repair_round": repairAttempts, "new_material": len(additional), "queries": repairReport.Queries, "search_results": repairReport.SearchResults, "fetched": repairReport.Fetched}}) + e.Broker.Publish(model.Activity{Type: "article.review.research.completed", Source: "brain", Phase: "quality-repair-research", NodeIDs: draft.SourceNodeIDs, Message: fmt.Sprintf("Gezielte Nachrecherche beendet · %d zusätzliche Volltextquellen", len(additional)), Strength: .82, Metadata: map[string]any{"trigger": trigger, "repair_round": repairAttempts, "new_material": len(additional), "queries": repairReport.Queries, "source_inbox_results": repairReport.InboxResults, "search_results": repairReport.SearchResults, "fetched": repairReport.Fetched}}) } reviewCopy := quality reviewFeedback = &reviewCopy diff --git a/internal/engine/article_adaptive.go b/internal/engine/article_adaptive.go index 91f90fc..3e9111e 100644 --- a/internal/engine/article_adaptive.go +++ b/internal/engine/article_adaptive.go @@ -145,13 +145,27 @@ func (e *Engine) collectAdaptiveInitialResearch(ctx context.Context, trigger str out := make([]model.ResearchResult, 0) report := articleResearchReport{Rounds: 1} for i, query := range queries { + report.Queries++ + freshnessSensitive := containsFreshnessLanguage(query) + inbox := e.sourceInboxResearch(ctx, query, fetchCap, freshnessSensitive) + if len(inbox) > 0 { + report.InboxResults += len(inbox) + report.Accepted += len(inbox) + out = uniqueResearchEvidence(append(out, inbox...)) + } + if len(inbox) >= e.Cfg.SourceInboxMinResults || !e.ResearchEnabledForRuntime() { + continue + } + remaining := fetchCap - len(inbox) + if remaining < 1 { + remaining = 1 + } language := "en-US" if looksGermanResearchQuery(query) { language = "de-DE" } question := model.ResearchQuestion{GapID: fmt.Sprintf("ADAPTIVE-%d", i+1), Question: query, Critical: true, ExpectActionable: containsActionableLanguage(query)} - report.Queries++ - items, stats := e.executeArticleResearchQueryForSynthesis(ctx, trigger, nodeIDs, question, query, language, 1, attemptedURLs, fetchCap) + items, stats := e.executeArticleResearchQueryForSynthesis(ctx, trigger, nodeIDs, question, query, language, 1, attemptedURLs, remaining) report.SearchResults += stats.SearchResults report.Fetched += stats.Fetched report.Accepted += stats.Accepted @@ -307,7 +321,9 @@ func (e *Engine) materializeGroundedResearchEvidence(articleID string, sources [ } categories := categoriesFromArticleSources(sources) ids := make([]string, 0, len(results)) + usedURLs := make([]string, 0, len(results)) for _, result := range results { + usedURLs = append(usedURLs, result.URL) id := graph.ID("external", result.URL) path, contentHash, err := e.queueResearchEvidence(result) if err != nil { @@ -322,5 +338,8 @@ func (e *Engine) materializeGroundedResearchEvidence(articleID string, sources [ e.Graph.UpsertNode(node) ids = append(ids, id) } + if e.SourceInbox != nil && len(usedURLs) > 0 { + _ = e.SourceInbox.MarkUsed(context.Background(), usedURLs) + } return unique(ids) } diff --git a/internal/engine/article_generate_then_review.go b/internal/engine/article_generate_then_review.go index f84b2d1..760fcc4 100644 --- a/internal/engine/article_generate_then_review.go +++ b/internal/engine/article_generate_then_review.go @@ -25,7 +25,7 @@ func (e *Engine) collectResearchMaterialForArticle(ctx context.Context, trigger } } - if !e.ResearchEnabledForRuntime() { + if !e.evidenceAcquisitionEnabled() { return material, report, nil } maxRounds := e.Cfg.ArticleResearchRounds @@ -56,6 +56,23 @@ func (e *Engine) collectResearchMaterialForArticle(ctx context.Context, trigger if report.Queries >= maxQueries { break } + inboxMaterial := e.sourceInboxResearch(ctx, question.Question, e.Cfg.ArticleResearchFetchResults, containsFreshnessLanguage(question.Question)) + if len(inboxMaterial) > 0 { + report.InboxResults += len(inboxMaterial) + report.Accepted += len(inboxMaterial) + for _, item := range inboxMaterial { + key := canonicalResearchURL(item.URL) + if key == "" || seenURLs[key] { + continue + } + seenURLs[key] = true + material = append(material, item) + collectedThisRound++ + } + } + if len(inboxMaterial) >= e.Cfg.SourceInboxMinResults || !e.ResearchEnabledForRuntime() { + continue + } lease, reused, err := e.beginResearchIntent(ctx, "synthesis-material", question.Question) if err != nil { return material, report, err @@ -201,13 +218,26 @@ func (e *Engine) collectReviewerRepairResearch(ctx context.Context, trigger stri if query == "" { continue } + report.Queries++ + inbox := e.sourceInboxResearch(ctx, query, fetchCap, containsFreshnessLanguage(query)) + if len(inbox) > 0 { + report.InboxResults += len(inbox) + report.Accepted += len(inbox) + out = uniqueResearchEvidence(append(out, inbox...)) + } + if len(inbox) >= e.Cfg.SourceInboxMinResults || !e.ResearchEnabledForRuntime() { + continue + } + remaining := fetchCap - len(inbox) + if remaining < 1 { + remaining = 1 + } language := "en-US" if looksGermanResearchQuery(query) { language = "de-DE" } question := model.ResearchQuestion{GapID: fmt.Sprintf("REVIEW-%d", i+1), Question: query, Critical: true, ExpectActionable: containsActionableLanguage(query)} - report.Queries++ - items, stats := e.executeArticleResearchQueryForSynthesis(ctx, trigger, nodeIDs, question, query, language, round, attemptedURLs, fetchCap) + items, stats := e.executeArticleResearchQueryForSynthesis(ctx, trigger, nodeIDs, question, query, language, round, attemptedURLs, remaining) report.SearchResults += stats.SearchResults report.Fetched += stats.Fetched report.Accepted += stats.Accepted diff --git a/internal/engine/article_research.go b/internal/engine/article_research.go index 6963e2d..e1d42ca 100644 --- a/internal/engine/article_research.go +++ b/internal/engine/article_research.go @@ -26,6 +26,7 @@ type articleResearchReport struct { Rejected int FetchFailed int SearchFailed int + InboxResults int } type rankedResearchCandidate struct { @@ -492,6 +493,7 @@ type queryExecutionStats struct { Rejected int FetchFailed int SearchFailed int + InboxResults int } func (e *Engine) executeArticleResearchQuery(ctx context.Context, trigger string, nodeIDs []string, question model.ResearchQuestion, query, language string, round int, attemptedURLs map[string]bool, fetchCaps ...int) ([]model.ResearchResult, queryExecutionStats) { diff --git a/internal/engine/engine.go b/internal/engine/engine.go index 0d3f1a6..adcd4af 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -24,6 +24,7 @@ import ( "github.com/local/glpi-neural-brain/internal/ollama" "github.com/local/glpi-neural-brain/internal/persist" "github.com/local/glpi-neural-brain/internal/research" + "github.com/local/glpi-neural-brain/internal/sourceagent" "github.com/local/glpi-neural-brain/internal/workqueue" ) @@ -63,6 +64,7 @@ type Engine struct { Scanner *ingest.KnowledgeScanner GLPIKB *ingest.GLPIKBSyncer Persistence *persist.Coordinator + SourceInbox *sourceagent.Store mu sync.Mutex stateMu sync.RWMutex @@ -329,6 +331,9 @@ func (e *Engine) Start(ctx context.Context) { } go e.idle(ctx) e.startAutonomousResearch(ctx) + if e.SourceInbox != nil && e.Cfg.SourceInboxEnabled { + go e.sourceInboxLoop(ctx) + } } func (e *Engine) enrichmentScheduler(ctx context.Context) { @@ -1041,6 +1046,11 @@ func (e *Engine) Status() map[string]any { } e.stateMu.RUnlock() status["autonomous_research"] = e.AutonomousResearchStatus(context.Background()) + if e.SourceInbox != nil { + if inbox, err := e.SourceInbox.Stats(context.Background()); err == nil { + status["source_inbox"] = inbox + } + } return status } diff --git a/internal/engine/source_inbox.go b/internal/engine/source_inbox.go new file mode 100644 index 0000000..1de1125 --- /dev/null +++ b/internal/engine/source_inbox.go @@ -0,0 +1,120 @@ +package engine + +import ( + "context" + "fmt" + "log/slog" + "strings" + "time" + + "github.com/local/glpi-neural-brain/internal/model" + "github.com/local/glpi-neural-brain/internal/ollama" + "github.com/local/glpi-neural-brain/internal/sourceagent" +) + +func (e *Engine) SetSourceInbox(store *sourceagent.Store) { e.SourceInbox = store } + +func (e *Engine) evidenceAcquisitionEnabled() bool { + return (e.SourceInbox != nil && e.Cfg.SourceInboxEnabled) || e.ResearchEnabledForRuntime() +} + +func (e *Engine) sourceInboxLoop(ctx context.Context) { + ticker := time.NewTicker(e.Cfg.SourceInboxInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + e.processSourceInbox(ctx) + } + } +} + +func (e *Engine) processSourceInbox(ctx context.Context) { + if e.SourceInbox == nil || !e.Cfg.SourceInboxEnabled { + return + } + items, err := e.SourceInbox.ClaimInbox(ctx, e.Cfg.SourceInboxBatchSize) + if err != nil { + slog.Warn("source inbox claim failed", "error", err) + return + } + if len(items) == 0 { + return + } + texts := make([]string, 0, len(items)) + for _, item := range items { + text := item.Document.Title + "\n" + strings.Join(item.Document.Categories, " · ") + "\n" + item.Document.Text + if len([]rune(text)) > 12000 { + text = string([]rune(text)[:12000]) + } + texts = append(texts, text) + } + cctx, cancel := context.WithTimeout(ollama.WithLowPriority(ctx), 4*time.Minute) + vectors, embedErr := e.Ollama.Embed(cctx, texts) + cancel() + if embedErr != nil || len(vectors) != len(items) { + for _, item := range items { + _ = e.SourceInbox.ReleaseInbox(ctx, item.ID, fmt.Sprint(embedErr)) + } + return + } + candidateCount := 0 + for i, item := range items { + hits, stats := e.similarKnowledge(vectors[i], 1, e.effectiveLearningFilter(), 0) + if len(hits) == 0 { + _ = e.SourceInbox.ReleaseInbox(ctx, item.ID, "knowledge vectors are not ready") + continue + } + relevance := 0.0 + matched := "" + if len(hits) > 0 { + relevance = hits[0].Score + matched = hits[0].NodeID + } + status := "archived" + if relevance >= e.Cfg.SourceInboxMinSimilarity { + status = "candidate" + candidateCount++ + } + meta := make(map[string]any, len(item.Metadata)+4) + for key, value := range item.Metadata { + meta[key] = value + } + meta["processing_mode"] = e.RuntimeSettings().ProcessingMode + meta["exact_comparisons"] = stats.ExactComparisons + meta["coarse_comparisons"] = stats.CoarseComparisons + meta["candidate_pool"] = stats.CandidatePool + _ = e.SourceInbox.CompleteClassification(ctx, item.ID, status, relevance, matched, meta) + } + if e.Broker != nil { + e.Broker.Publish(model.Activity{Type: "source.inbox.classified", Source: "brain", Phase: "source-inbox", Message: fmt.Sprintf("Source-Inbox: %d Dokumente geprüft · %d als Wissenskandidaten vorgemerkt", len(items), candidateCount), Strength: .36, Metadata: map[string]any{"documents": len(items), "candidates": candidateCount, "minimum_similarity": e.Cfg.SourceInboxMinSimilarity}}) + } +} + +func (e *Engine) sourceInboxResearch(ctx context.Context, query string, limit int, freshnessSensitive bool) []model.ResearchResult { + if e.SourceInbox == nil || !e.Cfg.SourceInboxEnabled || limit < 1 { + return nil + } + maxAge := time.Duration(0) + if freshnessSensitive { + maxAge = e.Cfg.SourceInboxFreshMaxAge + } + items, err := e.SourceInbox.SearchCandidates(ctx, query, limit, maxAge) + if err != nil { + return nil + } + out := make([]model.ResearchResult, 0, len(items)) + for _, item := range items { + if item.QueryScore < .18 { + continue + } + d := item.Document + out = append(out, model.ResearchResult{Title: d.Title, URL: d.CanonicalURL, Snippet: clamp(d.Text, 1000), Content: d.Text, ContentType: d.ContentType, Query: query, Language: d.Language, Fetched: true, Relevant: true, Relevance: item.QueryScore, SourceQuality: "source_inbox", SourceQualityScore: .68, AssessmentReason: "Vorab durch Source-Agent gesammelt und vom Brain als thematisch passend zur Knowledgebase klassifiziert."}) + } + if len(out) > 0 && e.Broker != nil { + e.Broker.Publish(model.Activity{Type: "article.research.inbox", Source: "brain", Phase: "knowledge-research-routing", Message: fmt.Sprintf("Source-Inbox liefert %d bereits gecrawlte Kandidaten vor SearXNG", len(out)), Strength: .56, Metadata: map[string]any{"query": query, "results": len(out), "freshness_sensitive": freshnessSensitive}}) + } + return out +} diff --git a/internal/research/fetch.go b/internal/research/fetch.go index 3da47d4..e4ba459 100644 --- a/internal/research/fetch.go +++ b/internal/research/fetch.go @@ -46,6 +46,33 @@ type FetchedPage struct { ContentType string } +// NewSafeHTTPClient returns an HTTP client that applies the same SSRF protections +// as FetchPage. It is intended for source/feed fetchers that need the raw response +// body instead of extracted page text. Private, loopback and link-local targets are +// rejected unless allowPrivate is explicitly enabled. Redirects are validated too. +func NewSafeHTTPClient(allowPrivate bool, timeout time.Duration) *http.Client { + if timeout <= 0 { + timeout = 20 * time.Second + } + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.Proxy = nil + transport.DialContext = safeDialContext(allowPrivate) + transport.MaxIdleConns = 4 + transport.MaxIdleConnsPerHost = 2 + transport.IdleConnTimeout = 20 * time.Second + return &http.Client{ + Transport: transport, + Timeout: timeout, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 5 { + return errors.New("zu viele Weiterleitungen") + } + _, err := validateFetchURL(req.Context(), req.URL.String(), allowPrivate) + return err + }, + } +} + func (c *Client) FetchPage(ctx context.Context, rawURL string, options FetchOptions) (FetchedPage, FetchDiagnostic, error) { started := time.Now() diagnostic := FetchDiagnostic{URL: strings.TrimSpace(rawURL)} @@ -79,23 +106,7 @@ func (c *Client) FetchPage(ctx context.Context, rawURL string, options FetchOpti return finish(FetchedPage{}, err) } - transport := http.DefaultTransport.(*http.Transport).Clone() - transport.Proxy = nil - transport.DialContext = safeDialContext(options.AllowPrivate) - transport.MaxIdleConns = 4 - transport.MaxIdleConnsPerHost = 2 - transport.IdleConnTimeout = 20 * time.Second - client := &http.Client{ - Transport: transport, - Timeout: options.Timeout, - CheckRedirect: func(req *http.Request, via []*http.Request) error { - if len(via) >= 5 { - return errors.New("zu viele Weiterleitungen") - } - _, err := validateFetchURL(req.Context(), req.URL.String(), options.AllowPrivate) - return err - }, - } + client := NewSafeHTTPClient(options.AllowPrivate, options.Timeout) req, err := http.NewRequestWithContext(ctx, http.MethodGet, parsed.String(), nil) if err != nil { diff --git a/internal/sourceagent/agent.go b/internal/sourceagent/agent.go new file mode 100644 index 0000000..88ef3a2 --- /dev/null +++ b/internal/sourceagent/agent.go @@ -0,0 +1,658 @@ +package sourceagent + +import ( + "bytes" + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "encoding/xml" + "errors" + "fmt" + "html" + "io" + "log/slog" + "net/http" + "net/url" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "sync" + "time" + + "github.com/local/glpi-neural-brain/internal/research" + _ "modernc.org/sqlite" +) + +type RunnerConfig struct { + BrainURL string + AgentID string + Token string + DataDir string + ConfigFile string + ConfigRefresh time.Duration + HTTPTimeout time.Duration + Concurrency int + BatchSize int + AllowPrivate bool + Version string +} + +type Runner struct { + cfg RunnerConfig + http *http.Client + sourceHTTP *http.Client + state *localState + mu sync.RWMutex + remote RemoteConfig + wake chan struct{} +} + +type bootstrapConfig struct { + BrainURL string `json:"brain_url"` + AgentID string `json:"agent_id"` + Token string `json:"token"` +} + +func NewRunner(cfg RunnerConfig) (*Runner, error) { + if strings.TrimSpace(cfg.ConfigFile) != "" { + if data, err := os.ReadFile(cfg.ConfigFile); err == nil { + var b bootstrapConfig + if json.Unmarshal(data, &b) == nil { + if cfg.BrainURL == "" { + cfg.BrainURL = b.BrainURL + } + if cfg.AgentID == "" { + cfg.AgentID = b.AgentID + } + if cfg.Token == "" { + cfg.Token = b.Token + } + } + } + } + cfg.BrainURL = strings.TrimRight(strings.TrimSpace(cfg.BrainURL), "/") + cfg.AgentID = strings.TrimSpace(cfg.AgentID) + cfg.Token = strings.TrimSpace(cfg.Token) + if cfg.BrainURL == "" || cfg.AgentID == "" || cfg.Token == "" { + return nil, errors.New("agent mode requires BRAIN_AGENT_BRAIN_URL, BRAIN_AGENT_ID and BRAIN_AGENT_TOKEN (or BRAIN_AGENT_CONFIG_FILE)") + } + u, err := url.Parse(cfg.BrainURL) + if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") { + return nil, errors.New("BRAIN_AGENT_BRAIN_URL must be an absolute http(s) URL") + } + if u.User != nil { + return nil, errors.New("BRAIN_AGENT_BRAIN_URL must not contain userinfo") + } + if cfg.ConfigRefresh < time.Minute { + cfg.ConfigRefresh = 5 * time.Minute + } + if cfg.HTTPTimeout < 5*time.Second { + cfg.HTTPTimeout = 30 * time.Second + } + if cfg.Concurrency < 1 { + cfg.Concurrency = 3 + } + if cfg.Concurrency > 16 { + cfg.Concurrency = 16 + } + if cfg.BatchSize < 1 { + cfg.BatchSize = 50 + } + if cfg.BatchSize > 500 { + cfg.BatchSize = 500 + } + state, err := openLocalState(cfg.DataDir) + if err != nil { + return nil, err + } + return &Runner{ + cfg: cfg, + http: &http.Client{Timeout: cfg.HTTPTimeout}, + sourceHTTP: research.NewSafeHTTPClient(cfg.AllowPrivate, cfg.HTTPTimeout), + state: state, + wake: make(chan struct{}, 1), + }, nil +} +func (r *Runner) Close() error { + if r == nil || r.state == nil { + return nil + } + return r.state.Close() +} + +func (r *Runner) Start(ctx context.Context) { + r.loadCachedConfig() + go r.loop(ctx) +} + +func (r *Runner) Status() map[string]any { + r.mu.RLock() + remote := r.remote + r.mu.RUnlock() + return map[string]any{"ok": true, "mode": "agent", "agent_id": r.cfg.AgentID, "brain_url": r.cfg.BrainURL, "configured_tasks": len(remote.Tasks), "config_issued_at": remote.IssuedAt, "version": r.cfg.Version} +} + +func (r *Runner) loop(ctx context.Context) { + refresh := time.NewTicker(r.cfg.ConfigRefresh) + defer refresh.Stop() + run := time.NewTicker(30 * time.Second) + defer run.Stop() + _ = r.refreshConfig(ctx) + r.runDue(ctx) + for { + select { + case <-ctx.Done(): + return + case <-refresh.C: + _ = r.refreshConfig(ctx) + case <-run.C: + r.runDue(ctx) + case <-r.wake: + r.runDue(ctx) + } + } +} + +func (r *Runner) refreshConfig(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.cfg.BrainURL+"/api/v1/agent/config", nil) + if err != nil { + return err + } + r.auth(req) + resp, err := r.http.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return fmt.Errorf("brain config returned HTTP %d", resp.StatusCode) + } + var cfg RemoteConfig + if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&cfg); err != nil { + return err + } + if cfg.Agent.ID != "" && cfg.Agent.ID != r.cfg.AgentID { + return fmt.Errorf("brain returned config for unexpected agent %q", cfg.Agent.ID) + } + r.mu.Lock() + r.remote = cfg + r.mu.Unlock() + r.saveCachedConfig(cfg) + _ = r.sendHeartbeat(ctx, Heartbeat{AgentID: r.cfg.AgentID, Version: r.cfg.Version, Status: "online", Metadata: map[string]any{"configured_tasks": len(cfg.Tasks)}}) + return nil +} + +func (r *Runner) runDue(ctx context.Context) { + r.mu.RLock() + tasks := append([]Task(nil), r.remote.Tasks...) + r.mu.RUnlock() + if len(tasks) == 0 { + return + } + sem := make(chan struct{}, r.cfg.Concurrency) + var wg sync.WaitGroup + for _, task := range tasks { + task := task + if !task.Enabled || !r.state.Due(ctx, task) { + continue + } + wg.Add(1) + go func() { + defer wg.Done() + select { + case sem <- struct{}{}: + case <-ctx.Done(): + return + } + defer func() { <-sem }() + r.runTask(ctx, task) + }() + } + wg.Wait() +} + +func (r *Runner) runTask(ctx context.Context, task Task) { + started := time.Now().UTC() + docs, err := r.pollTask(ctx, task) + if err == nil && len(docs) > 0 { + for start := 0; start < len(docs); start += r.cfg.BatchSize { + end := start + r.cfg.BatchSize + if end > len(docs) { + end = len(docs) + } + if sendErr := r.sendBatch(ctx, task.ID, docs[start:end]); sendErr != nil { + err = sendErr + break + } + for _, d := range docs[start:end] { + _ = r.state.MarkSeen(ctx, task.ID, d.CanonicalURL, d.ContentSHA256) + } + } + } + _ = r.state.FinishTask(ctx, task.ID, started, err) + h := Heartbeat{AgentID: r.cfg.AgentID, Version: r.cfg.Version, Status: "ok", LastRunAt: started, TasksChecked: 1, Documents: len(docs)} + if err != nil { + h.Status = "error" + h.LastError = err.Error() + slog.Warn("source agent task failed", "task", task.ID, "error", err) + } else { + slog.Info("source agent task completed", "task", task.ID, "documents", len(docs)) + } + _ = r.sendHeartbeat(ctx, h) +} + +func (r *Runner) pollTask(ctx context.Context, task Task) ([]Document, error) { + switch task.Type { + case "rss", "atom": + return r.pollFeed(ctx, task) + case "sitemap": + return r.pollSitemap(ctx, task) + case "web": + return r.pollWeb(ctx, task) + default: + return nil, fmt.Errorf("unsupported task type %q", task.Type) + } +} + +type feedEnvelope struct { + Channel struct { + Items []struct { + Title string `xml:"title"` + Link string `xml:"link"` + Description string `xml:"description"` + PubDate string `xml:"pubDate"` + GUID string `xml:"guid"` + } `xml:"item"` + } `xml:"channel"` + Entries []struct { + Title string `xml:"title"` + ID string `xml:"id"` + Updated string `xml:"updated"` + Published string `xml:"published"` + Summary string `xml:"summary"` + Content string `xml:"content"` + Links []struct { + Href string `xml:"href,attr"` + Rel string `xml:"rel,attr"` + } `xml:"link"` + } `xml:"entry"` +} + +type feedItem struct { + Title, Link, Summary string + Published time.Time +} + +func (r *Runner) pollFeed(ctx context.Context, task Task) ([]Document, error) { + body, finalURL, _, _, err := r.fetchRaw(ctx, task.URL, 5<<20) + if err != nil { + return nil, err + } + var env feedEnvelope + if err := xml.Unmarshal(body, &env); err != nil { + return nil, fmt.Errorf("feed XML: %w", err) + } + var items []feedItem + for _, it := range env.Channel.Items { + link := strings.TrimSpace(it.Link) + if link == "" { + link = strings.TrimSpace(it.GUID) + } + items = append(items, feedItem{Title: strings.TrimSpace(it.Title), Link: resolveLink(finalURL, link), Summary: stripMarkup(it.Description), Published: parsePublished(it.PubDate)}) + } + for _, it := range env.Entries { + link := "" + for _, l := range it.Links { + if l.Rel == "" || l.Rel == "alternate" { + link = l.Href + break + } + } + pub := parsePublished(it.Published) + if pub.IsZero() { + pub = parsePublished(it.Updated) + } + summary := it.Summary + if summary == "" { + summary = it.Content + } + items = append(items, feedItem{Title: strings.TrimSpace(it.Title), Link: resolveLink(finalURL, link), Summary: stripMarkup(summary), Published: pub}) + } + sort.SliceStable(items, func(i, j int) bool { return items[i].Published.After(items[j].Published) }) + if len(items) > task.MaxItems { + items = items[:task.MaxItems] + } + return r.materializeItems(ctx, task, items) +} + +type sitemapEnvelope struct { + URLs []struct { + Loc string `xml:"loc"` + LastMod string `xml:"lastmod"` + } `xml:"url"` + Sitemaps []struct { + Loc string `xml:"loc"` + } `xml:"sitemap"` +} + +func (r *Runner) pollSitemap(ctx context.Context, task Task) ([]Document, error) { + items, err := r.collectSitemapItems(ctx, task.URL, task.MaxItems, 0) + if err != nil { + return nil, err + } + return r.materializeItems(ctx, task, items) +} + +func (r *Runner) collectSitemapItems(ctx context.Context, rawURL string, limit, depth int) ([]feedItem, error) { + if limit <= 0 || depth > 1 { + return nil, nil + } + body, finalURL, _, _, err := r.fetchRaw(ctx, rawURL, 8<<20) + if err != nil { + return nil, err + } + var sm sitemapEnvelope + if err := xml.Unmarshal(body, &sm); err != nil { + return nil, err + } + items := make([]feedItem, 0, limit) + for _, u := range sm.URLs { + link := resolveLink(finalURL, u.Loc) + if link == "" { + continue + } + items = append(items, feedItem{Link: link, Published: parsePublished(u.LastMod)}) + if len(items) >= limit { + return items, nil + } + } + // Sitemap indexes are common on larger publishers. Follow a bounded number of + // child maps once; source HTTP safety rules apply to every child URL. + for i, child := range sm.Sitemaps { + if len(items) >= limit || i >= 12 { + break + } + childURL := resolveLink(finalURL, child.Loc) + if childURL == "" { + continue + } + more, childErr := r.collectSitemapItems(ctx, childURL, limit-len(items), depth+1) + if childErr != nil { + continue + } + items = append(items, more...) + } + return items, nil +} + +var hrefPattern = regexp.MustCompile(`(?is)]*href\s*=\s*["']([^"'#]+)["'][^>]*>(.*?)`) + +func (r *Runner) pollWeb(ctx context.Context, task Task) ([]Document, error) { + body, finalURL, _, _, err := r.fetchRaw(ctx, task.URL, 5<<20) + if err != nil { + return nil, err + } + matches := hrefPattern.FindAllStringSubmatch(string(body), -1) + seen := map[string]bool{} + items := make([]feedItem, 0, task.MaxItems) + base, _ := url.Parse(finalURL) + for _, m := range matches { + link := resolveLink(finalURL, m[1]) + if link == "" || seen[link] { + continue + } + u, err := url.Parse(link) + if err != nil || u.Host != base.Host { + continue + } + if !looksArticleLink(u.Path, stripMarkup(m[2])) { + continue + } + seen[link] = true + items = append(items, feedItem{Title: stripMarkup(m[2]), Link: link}) + if len(items) >= task.MaxItems { + break + } + } + return r.materializeItems(ctx, task, items) +} + +func (r *Runner) materializeItems(ctx context.Context, task Task, items []feedItem) ([]Document, error) { + fetcher := research.New("") + out := make([]Document, 0, len(items)) + for _, item := range items { + if strings.TrimSpace(item.Link) == "" { + continue + } + if r.state.SeenURL(ctx, task.ID, item.Link) && !strings.EqualFold(strings.TrimSpace(task.Config["refetch_seen"]), "true") { + continue + } + page, diag, err := fetcher.FetchPage(ctx, item.Link, research.FetchOptions{MaxBytes: 2 << 20, MaxChars: 20000, Timeout: r.cfg.HTTPTimeout, AllowPrivate: r.cfg.AllowPrivate}) + text := strings.TrimSpace(item.Summary) + title := strings.TrimSpace(item.Title) + ctype := "text/html" + final := item.Link + if err == nil { + if page.Content != "" { + text = page.Content + } + if page.Title != "" { + title = page.Title + } + if page.URL != "" { + final = page.URL + } + ctype = page.ContentType + } else if len([]rune(text)) < 80 { + continue + } + if title == "" { + title = final + } + sum := sha256.Sum256([]byte(text)) + sha := hex.EncodeToString(sum[:]) + if r.state.SeenHash(ctx, task.ID, final, sha) { + continue + } + baseURL := "" + if u, e := url.Parse(task.URL); e == nil { + baseURL = u.Scheme + "://" + u.Host + } + out = append(out, Document{ExternalID: final, URL: final, CanonicalURL: final, Title: title, PublishedAt: item.Published, DiscoveredAt: time.Now().UTC(), ContentType: ctype, Text: text, ContentSHA256: sha, SourceName: task.Name, SourceBaseURL: baseURL, Categories: task.Categories, Metadata: map[string]any{"agent_task_type": task.Type, "fetch_error_kind": diag.ErrorKind}}) + if len(out) >= task.MaxItems { + break + } + } + return out, nil +} + +func (r *Runner) fetchRaw(ctx context.Context, raw string, max int64) ([]byte, string, string, string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, raw, nil) + if err != nil { + return nil, "", "", "", err + } + req.Header.Set("User-Agent", "glpi-neural-brain-source-agent/1.0") + req.Header.Set("Accept", "application/rss+xml, application/atom+xml, application/xml, text/xml, text/html;q=0.9, */*;q=0.5") + resp, err := r.sourceHTTP.Do(req) + if err != nil { + return nil, "", "", "", err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return nil, "", "", "", fmt.Errorf("source returned HTTP %d", resp.StatusCode) + } + body, err := io.ReadAll(io.LimitReader(resp.Body, max+1)) + if err != nil { + return nil, "", "", "", err + } + if int64(len(body)) > max { + return nil, "", "", "", errors.New("source response too large") + } + return body, resp.Request.URL.String(), resp.Header.Get("ETag"), resp.Header.Get("Last-Modified"), nil +} + +func (r *Runner) sendBatch(ctx context.Context, taskID string, docs []Document) error { + payload := IngestBatch{SchemaVersion: SchemaVersion, AgentID: r.cfg.AgentID, TaskID: taskID, Documents: docs} + data, _ := json.Marshal(payload) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.cfg.BrainURL+"/api/v1/agent/ingest", bytes.NewReader(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + r.auth(req) + resp, err := r.http.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return fmt.Errorf("brain ingest HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b))) + } + return nil +} +func (r *Runner) sendHeartbeat(ctx context.Context, h Heartbeat) error { + data, _ := json.Marshal(h) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.cfg.BrainURL+"/api/v1/agent/heartbeat", bytes.NewReader(data)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + r.auth(req) + resp, err := r.http.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode/100 != 2 { + return fmt.Errorf("heartbeat HTTP %d", resp.StatusCode) + } + return nil +} +func (r *Runner) auth(req *http.Request) { + req.Header.Set("Authorization", "Bearer "+r.cfg.Token) + req.Header.Set("X-Brain-Agent-ID", r.cfg.AgentID) +} + +func resolveLink(base, ref string) string { + ref = strings.TrimSpace(html.UnescapeString(ref)) + if ref == "" { + return "" + } + u, err := url.Parse(ref) + if err != nil { + return "" + } + b, err := url.Parse(base) + if err != nil { + return "" + } + return b.ResolveReference(u).String() +} +func stripMarkup(v string) string { + v = regexp.MustCompile(`(?is)<[^>]+>`).ReplaceAllString(v, " ") + return strings.Join(strings.Fields(html.UnescapeString(v)), " ") +} +func looksArticleLink(path, label string) bool { + p := strings.ToLower(path + " " + label) + if strings.Contains(p, "/tag/") || strings.Contains(p, "/category/") || strings.Contains(p, "/author/") || strings.Contains(p, "login") || strings.Contains(p, "privacy") || strings.Contains(p, "impress") || strings.Contains(p, "kontakt") { + return false + } + segments := strings.Split(strings.Trim(path, "/"), "/") + return len(segments) >= 2 || strings.Contains(p, "news") || strings.Contains(p, "blog") || strings.Contains(p, "advis") || strings.Contains(p, "release") || strings.Contains(p, "security") +} +func parsePublished(v string) time.Time { + v = strings.TrimSpace(v) + for _, layout := range []string{time.RFC3339, time.RFC1123Z, time.RFC1123, time.RFC822Z, time.RFC822, "2006-01-02"} { + if t, err := time.Parse(layout, v); err == nil { + return t.UTC() + } + } + return time.Time{} +} + +func (r *Runner) configCachePath() string { + return filepath.Join(r.cfg.DataDir, "source-agent-config-cache.json") +} +func (r *Runner) saveCachedConfig(cfg RemoteConfig) { + data, err := json.MarshalIndent(cfg, "", " ") + if err != nil { + return + } + _ = os.WriteFile(r.configCachePath(), append(data, '\n'), 0o600) +} +func (r *Runner) loadCachedConfig() { + data, err := os.ReadFile(r.configCachePath()) + if err != nil { + return + } + var cfg RemoteConfig + if json.Unmarshal(data, &cfg) != nil || cfg.Agent.ID != r.cfg.AgentID { + return + } + r.mu.Lock() + r.remote = cfg + r.mu.Unlock() +} + +type localState struct { + db *sql.DB + mu sync.Mutex +} + +func openLocalState(dataDir string) (*localState, error) { + if strings.TrimSpace(dataDir) == "" { + dataDir = "./data" + } + if err := os.MkdirAll(dataDir, 0o750); err != nil { + return nil, err + } + db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(filepath.Join(dataDir, "source-agent-local.db"))+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)") + if err != nil { + return nil, err + } + st := &localState{db: db} + for _, q := range []string{`CREATE TABLE IF NOT EXISTS task_state(task_id TEXT PRIMARY KEY,last_run_ns INTEGER NOT NULL DEFAULT 0,last_error TEXT NOT NULL DEFAULT '') WITHOUT ROWID`, `CREATE TABLE IF NOT EXISTS seen(task_id TEXT NOT NULL,url TEXT NOT NULL,content_sha256 TEXT NOT NULL,seen_at_ns INTEGER NOT NULL,PRIMARY KEY(task_id,url,content_sha256)) WITHOUT ROWID`, `CREATE INDEX IF NOT EXISTS idx_seen_task_url ON seen(task_id,url,seen_at_ns DESC)`} { + if _, err := db.Exec(q); err != nil { + db.Close() + return nil, err + } + } + return st, nil +} +func (s *localState) Close() error { return s.db.Close() } +func (s *localState) Due(ctx context.Context, t Task) bool { + d, err := time.ParseDuration(t.PollInterval) + if err != nil { + d = 4 * time.Hour + } + var last int64 + err = s.db.QueryRowContext(ctx, `SELECT last_run_ns FROM task_state WHERE task_id=?`, t.ID).Scan(&last) + return err == sql.ErrNoRows || err != nil || last == 0 || time.Since(time.Unix(0, last)) >= d +} +func (s *localState) FinishTask(ctx context.Context, id string, started time.Time, err error) error { + msg := "" + if err != nil { + msg = err.Error() + } + _, e := s.db.ExecContext(ctx, `INSERT INTO task_state(task_id,last_run_ns,last_error) VALUES(?,?,?) ON CONFLICT(task_id) DO UPDATE SET last_run_ns=excluded.last_run_ns,last_error=excluded.last_error`, id, started.UnixNano(), msg) + return e +} +func (s *localState) SeenURL(ctx context.Context, task, urlv string) bool { + var n int + _ = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM seen WHERE task_id=? AND url=?`, task, urlv).Scan(&n) + return n > 0 +} +func (s *localState) SeenHash(ctx context.Context, task, urlv, sha string) bool { + var n int + _ = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM seen WHERE task_id=? AND url=? AND content_sha256=?`, task, urlv, sha).Scan(&n) + return n > 0 +} +func (s *localState) MarkSeen(ctx context.Context, task, urlv, sha string) error { + _, err := s.db.ExecContext(ctx, `INSERT OR IGNORE INTO seen(task_id,url,content_sha256,seen_at_ns) VALUES(?,?,?,?)`, task, urlv, sha, time.Now().UTC().UnixNano()) + return err +} diff --git a/internal/sourceagent/sourceagent_test.go b/internal/sourceagent/sourceagent_test.go new file mode 100644 index 0000000..970ef5e --- /dev/null +++ b/internal/sourceagent/sourceagent_test.go @@ -0,0 +1,91 @@ +package sourceagent + +import ( + "context" + "math/bits" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/local/glpi-neural-brain/internal/research" +) + +func TestValidateTaskSupportsPollerTypes(t *testing.T) { + for _, typ := range []string{"rss", "atom", "sitemap", "web"} { + task, err := validateTask(Task{ID: "security", AgentID: "a", Name: "Security", Type: typ, URL: "https://example.org/feed", Enabled: true, PollInterval: "2h", MaxItems: 25}) + if err != nil { + t.Fatalf("%s: %v", typ, err) + } + if task.Type != typ || task.MaxItems != 25 { + t.Fatalf("unexpected task: %+v", task) + } + } +} + +func TestValidateTaskRejectsUnsafeShape(t *testing.T) { + if _, err := validateTask(Task{Type: "ftp", URL: "https://example.org", PollInterval: "1h"}); err == nil { + t.Fatal("expected unsupported type") + } + if _, err := validateTask(Task{Type: "rss", URL: "ftp://example.org/feed", PollInterval: "1h"}); err == nil { + t.Fatal("expected URL validation error") + } + if _, err := validateTask(Task{Type: "rss", URL: "https://example.org/feed", PollInterval: "1m"}); err == nil { + t.Fatal("expected interval validation error") + } + if _, err := validateTask(Task{Type: "rss", URL: "https://user:pass@example.org/feed", PollInterval: "1h"}); err == nil { + t.Fatal("expected userinfo URL validation error") + } +} + +func TestNormalizeDocumentBuildsContentHash(t *testing.T) { + d, err := normalizeDocument(Document{URL: "https://example.org/a", Title: "A", Text: "Dies ist ein ausreichend langer Dokumenttext mit sicherheitsrelevanten Informationen und mehreren Details für die Verarbeitung."}) + if err != nil { + t.Fatal(err) + } + if d.ContentSHA256 == "" || d.CanonicalURL != "https://example.org/a" || d.ExternalID != "https://example.org/a" { + t.Fatalf("unexpected normalized document: %+v", d) + } +} + +func TestSimhashKeepsRelatedTextCloser(t *testing.T) { + base := simhash64("Windows Backup Repository Ransomware Schutz immutable storage MFA") + related := simhash64("Ransomware Schutz für Windows Backup Repository mit MFA und immutable Storage") + unrelated := simhash64("Kaffee Bohnen Espresso Maschine Mahlgrad Temperatur") + if bits.OnesCount64(base^related) >= bits.OnesCount64(base^unrelated) { + t.Fatalf("related text should hash closer") + } +} + +func TestFetchRawBlocksPrivateSourceByDefault(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _, _ = w.Write([]byte("")) })) + defer srv.Close() + r := &Runner{cfg: RunnerConfig{HTTPTimeout: time.Second}, sourceHTTP: research.NewSafeHTTPClient(false, time.Second)} + if _, _, _, _, err := r.fetchRaw(context.Background(), srv.URL, 1024); err == nil { + t.Fatal("expected private/loopback source URL to be blocked") + } +} + +func TestSitemapIndexCollectsChildURLs(t *testing.T) { + mux := http.NewServeMux() + var base string + mux.HandleFunc("/index.xml", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write([]byte(`` + base + `/child.xml`)) + }) + mux.HandleFunc("/child.xml", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/xml") + _, _ = w.Write([]byte(`` + base + `/article-12026-08-07T12:00:00Z`)) + }) + srv := httptest.NewServer(mux) + defer srv.Close() + base = srv.URL + r := &Runner{cfg: RunnerConfig{HTTPTimeout: time.Second, AllowPrivate: true}, sourceHTTP: research.NewSafeHTTPClient(true, time.Second)} + items, err := r.collectSitemapItems(context.Background(), srv.URL+"/index.xml", 10, 0) + if err != nil { + t.Fatal(err) + } + if len(items) != 1 || items[0].Link != srv.URL+"/article-1" { + t.Fatalf("unexpected sitemap items: %+v", items) + } +} diff --git a/internal/sourceagent/store.go b/internal/sourceagent/store.go new file mode 100644 index 0000000..8684865 --- /dev/null +++ b/internal/sourceagent/store.go @@ -0,0 +1,698 @@ +package sourceagent + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "hash/fnv" + "math/bits" + "net/url" + "path/filepath" + "sort" + "strings" + "sync" + "time" + "unicode" + + _ "modernc.org/sqlite" +) + +type Store struct { + db *sql.DB + mu sync.Mutex +} + +func OpenStore(dataDir string) (*Store, error) { + path := filepath.Join(dataDir, "source-agents.db") + db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(path)+"?_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)&_pragma=journal_mode(WAL)") + if err != nil { + return nil, err + } + db.SetMaxOpenConns(4) + store := &Store{db: db} + if err := store.init(context.Background()); err != nil { + _ = db.Close() + return nil, err + } + return store, nil +} + +func (s *Store) Close() error { + if s == nil || s.db == nil { + return nil + } + return s.db.Close() +} + +func (s *Store) init(ctx context.Context) error { + statements := []string{ + `CREATE TABLE IF NOT EXISTS source_agents ( + id TEXT PRIMARY KEY, name TEXT NOT NULL, token_hash TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, + created_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL, last_seen_ns INTEGER NOT NULL DEFAULT 0, + last_error TEXT NOT NULL DEFAULT '', version TEXT NOT NULL DEFAULT '' + ) WITHOUT ROWID`, + `CREATE TABLE IF NOT EXISTS source_tasks ( + id TEXT PRIMARY KEY, agent_id TEXT NOT NULL REFERENCES source_agents(id) ON DELETE CASCADE, + name TEXT NOT NULL, type TEXT NOT NULL, url TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1, + poll_interval TEXT NOT NULL DEFAULT '4h', categories_json TEXT NOT NULL DEFAULT '[]', max_items INTEGER NOT NULL DEFAULT 20, + config_json TEXT NOT NULL DEFAULT '{}', created_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL + ) WITHOUT ROWID`, + `CREATE INDEX IF NOT EXISTS idx_source_tasks_agent ON source_tasks(agent_id, enabled, updated_at_ns)`, + `CREATE TABLE IF NOT EXISTS source_inbox ( + id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, task_id TEXT NOT NULL, external_id TEXT NOT NULL DEFAULT '', + url TEXT NOT NULL, canonical_url TEXT NOT NULL, title TEXT NOT NULL, published_at_ns INTEGER NOT NULL DEFAULT 0, + discovered_at_ns INTEGER NOT NULL DEFAULT 0, received_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL, + language TEXT NOT NULL DEFAULT '', content_type TEXT NOT NULL DEFAULT '', text_content TEXT NOT NULL, + content_sha256 TEXT NOT NULL, source_name TEXT NOT NULL DEFAULT '', source_base_url TEXT NOT NULL DEFAULT '', + categories_json TEXT NOT NULL DEFAULT '[]', metadata_json TEXT NOT NULL DEFAULT '{}', signature INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'received', relevance REAL NOT NULL DEFAULT 0, matched_node_id TEXT NOT NULL DEFAULT '', + UNIQUE(agent_id, task_id, canonical_url, content_sha256) + ) WITHOUT ROWID`, + `CREATE INDEX IF NOT EXISTS idx_source_inbox_status_received ON source_inbox(status, received_at_ns)`, + `CREATE INDEX IF NOT EXISTS idx_source_inbox_candidate_updated ON source_inbox(status, updated_at_ns DESC)`, + } + for _, statement := range statements { + if _, err := s.db.ExecContext(ctx, statement); err != nil { + return err + } + } + _, _ = s.db.ExecContext(ctx, `UPDATE source_inbox SET status='received' WHERE status='processing' AND updated_at_ns 30*24*time.Hour { + return t, fmt.Errorf("poll_interval must be between 5m and 720h") + } + if t.MaxItems <= 0 { + t.MaxItems = 20 + } + if t.MaxItems > 500 { + return t, fmt.Errorf("max_items must be <=500") + } + if t.Config == nil { + t.Config = map[string]string{} + } + return t, nil +} + +func (s *Store) UpsertTask(ctx context.Context, t Task) (Task, error) { + var err error + t, err = validateTask(t) + if err != nil { + return Task{}, err + } + if strings.TrimSpace(t.AgentID) == "" { + return Task{}, errors.New("agent_id required") + } + cats, _ := json.Marshal(uniqueStrings(t.Categories)) + cfg, _ := json.Marshal(t.Config) + now := time.Now().UTC() + _, err = s.db.ExecContext(ctx, `INSERT INTO source_tasks(id,agent_id,name,type,url,enabled,poll_interval,categories_json,max_items,config_json,created_at_ns,updated_at_ns) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET agent_id=excluded.agent_id,name=excluded.name,type=excluded.type,url=excluded.url,enabled=excluded.enabled,poll_interval=excluded.poll_interval,categories_json=excluded.categories_json,max_items=excluded.max_items,config_json=excluded.config_json,updated_at_ns=excluded.updated_at_ns`, t.ID, t.AgentID, t.Name, t.Type, t.URL, boolInt(t.Enabled), t.PollInterval, string(cats), t.MaxItems, string(cfg), now.UnixNano(), now.UnixNano()) + if err != nil { + return Task{}, err + } + t.UpdatedAt = now + if t.CreatedAt.IsZero() { + t.CreatedAt = now + } + return t, nil +} + +func (s *Store) DeleteTask(ctx context.Context, id string) error { + _, err := s.db.ExecContext(ctx, `DELETE FROM source_tasks WHERE id=?`, id) + return err +} +func (s *Store) ListTasks(ctx context.Context, agentID string) ([]Task, error) { + rows, err := s.db.QueryContext(ctx, `SELECT id,agent_id,name,type,url,enabled,poll_interval,categories_json,max_items,config_json,created_at_ns,updated_at_ns FROM source_tasks WHERE (?='' OR agent_id=?) ORDER BY name,id`, agentID, agentID) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Task + for rows.Next() { + var t Task + var en int + var cats, cfg string + var c, u int64 + if err := rows.Scan(&t.ID, &t.AgentID, &t.Name, &t.Type, &t.URL, &en, &t.PollInterval, &cats, &t.MaxItems, &cfg, &c, &u); err != nil { + return nil, err + } + t.Enabled = en != 0 + _ = json.Unmarshal([]byte(cats), &t.Categories) + _ = json.Unmarshal([]byte(cfg), &t.Config) + t.CreatedAt = nsTime(c) + t.UpdatedAt = nsTime(u) + out = append(out, t) + } + return out, rows.Err() +} + +func (s *Store) RemoteConfig(ctx context.Context, agent Agent) (RemoteConfig, error) { + tasks, err := s.ListTasks(ctx, agent.ID) + if err != nil { + return RemoteConfig{}, err + } + return RemoteConfig{SchemaVersion: SchemaVersion, Agent: agent, Tasks: tasks, IssuedAt: time.Now().UTC()}, nil +} + +func (s *Store) Heartbeat(ctx context.Context, agentID string, h Heartbeat) error { + _, err := s.db.ExecContext(ctx, `UPDATE source_agents SET last_seen_ns=?,last_error=?,version=?,updated_at_ns=? WHERE id=?`, time.Now().UTC().UnixNano(), strings.TrimSpace(h.LastError), strings.TrimSpace(h.Version), time.Now().UTC().UnixNano(), agentID) + return err +} + +func normalizeDocument(d Document) (Document, error) { + d.URL = strings.TrimSpace(d.URL) + if d.URL == "" { + return d, errors.New("document url required") + } + u, err := url.Parse(d.URL) + if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") { + return d, errors.New("invalid document url") + } + d.CanonicalURL = strings.TrimSpace(d.CanonicalURL) + if d.CanonicalURL == "" { + d.CanonicalURL = d.URL + } + d.Title = strings.TrimSpace(d.Title) + d.Text = strings.TrimSpace(d.Text) + if len([]rune(d.Text)) < 80 { + return d, errors.New("document text too short") + } + if d.DiscoveredAt.IsZero() { + d.DiscoveredAt = time.Now().UTC() + } + if d.ExternalID == "" { + d.ExternalID = d.CanonicalURL + } + d.Categories = uniqueStrings(d.Categories) + if d.Metadata == nil { + d.Metadata = map[string]any{} + } + if d.ContentSHA256 == "" { + sum := sha256.Sum256([]byte(d.Text)) + d.ContentSHA256 = hex.EncodeToString(sum[:]) + } + return d, nil +} + +func (s *Store) Ingest(ctx context.Context, agentID, taskID string, docs []Document) (IngestResult, error) { + var taskOwner string + if err := s.db.QueryRowContext(ctx, `SELECT agent_id FROM source_tasks WHERE id=? AND enabled=1`, taskID).Scan(&taskOwner); err != nil { + return IngestResult{}, fmt.Errorf("unknown or disabled source task: %w", err) + } + if taskOwner != agentID { + return IngestResult{}, errors.New("source task does not belong to authenticated agent") + } + result := IngestResult{BatchID: randomID("batch")} + now := time.Now().UTC() + s.mu.Lock() + defer s.mu.Unlock() + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return result, err + } + defer tx.Rollback() + for _, raw := range docs { + d, e := normalizeDocument(raw) + if e != nil { + result.Rejected++ + continue + } + cats, _ := json.Marshal(d.Categories) + meta, _ := json.Marshal(d.Metadata) + idSum := sha256.Sum256([]byte(agentID + "\x00" + taskID + "\x00" + d.CanonicalURL + "\x00" + d.ContentSHA256)) + id := hex.EncodeToString(idSum[:12]) + sig := int64(simhash64(d.Title + "\n" + d.Text)) + res, e := tx.ExecContext(ctx, `INSERT OR IGNORE INTO source_inbox(id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,signature,status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, agentID, taskID, d.ExternalID, d.URL, d.CanonicalURL, d.Title, timeNS(d.PublishedAt), timeNS(d.DiscoveredAt), now.UnixNano(), now.UnixNano(), d.Language, d.ContentType, d.Text, d.ContentSHA256, d.SourceName, d.SourceBaseURL, string(cats), string(meta), sig, "received") + if e != nil { + result.Rejected++ + continue + } + n, _ := res.RowsAffected() + if n == 0 { + result.Duplicates++ + } else { + result.Accepted++ + } + } + if err := tx.Commit(); err != nil { + return result, err + } + return result, nil +} + +func (s *Store) ClaimInbox(ctx context.Context, limit int) ([]InboxDocument, error) { + if limit < 1 { + limit = 1 + } + if limit > 100 { + limit = 100 + } + s.mu.Lock() + defer s.mu.Unlock() + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return nil, err + } + defer tx.Rollback() + rows, err := tx.QueryContext(ctx, `SELECT id FROM source_inbox WHERE status='received' ORDER BY received_at_ns LIMIT ?`, limit) + if err != nil { + return nil, err + } + var ids []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return nil, err + } + ids = append(ids, id) + } + rows.Close() + if len(ids) == 0 { + return nil, tx.Commit() + } + now := time.Now().UTC().UnixNano() + for _, id := range ids { + if _, err := tx.ExecContext(ctx, `UPDATE source_inbox SET status='processing',updated_at_ns=? WHERE id=? AND status='received'`, now, id); err != nil { + return nil, err + } + } + if err := tx.Commit(); err != nil { + return nil, err + } + return s.GetInboxByIDs(ctx, ids) +} + +func (s *Store) CompleteClassification(ctx context.Context, id, status string, relevance float64, matchedNodeID string, meta map[string]any) error { + if status != "candidate" && status != "archived" { + return errors.New("invalid inbox classification status") + } + data, _ := json.Marshal(meta) + _, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status=?,relevance=?,matched_node_id=?,metadata_json=?,updated_at_ns=? WHERE id=?`, status, relevance, matchedNodeID, string(data), time.Now().UTC().UnixNano(), id) + return err +} + +func (s *Store) MarkUsed(ctx context.Context, canonicalURLs []string) error { + now := time.Now().UTC().UnixNano() + for _, u := range uniqueStrings(canonicalURLs) { + if _, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='used',updated_at_ns=? WHERE canonical_url=?`, now, u); err != nil { + return err + } + } + return nil +} + +func (s *Store) GetInboxByIDs(ctx context.Context, ids []string) ([]InboxDocument, error) { + if len(ids) == 0 { + return nil, nil + } + out := make([]InboxDocument, 0, len(ids)) + for _, id := range ids { + doc, err := s.getInbox(ctx, id) + if err == nil { + out = append(out, doc) + } + } + return out, nil +} +func (s *Store) getInbox(ctx context.Context, id string) (InboxDocument, error) { + row := s.db.QueryRowContext(ctx, `SELECT id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,status,relevance,matched_node_id FROM source_inbox WHERE id=?`, id) + return scanInbox(row) +} + +type rowScanner interface{ Scan(...any) error } + +func scanInbox(row rowScanner) (InboxDocument, error) { + var d InboxDocument + var ext, urlv, canon, title, lang, ctype, text, sha, source, base, cats, meta, status, matched string + var pub, disc, recv, upd int64 + var rel float64 + err := row.Scan(&d.ID, &d.AgentID, &d.TaskID, &ext, &urlv, &canon, &title, &pub, &disc, &recv, &upd, &lang, &ctype, &text, &sha, &source, &base, &cats, &meta, &status, &rel, &matched) + if err != nil { + return d, err + } + d.Document = Document{ExternalID: ext, URL: urlv, CanonicalURL: canon, Title: title, PublishedAt: nsTime(pub), DiscoveredAt: nsTime(disc), Language: lang, ContentType: ctype, Text: text, ContentSHA256: sha, SourceName: source, SourceBaseURL: base} + _ = json.Unmarshal([]byte(cats), &d.Document.Categories) + _ = json.Unmarshal([]byte(meta), &d.Metadata) + d.Status = status + d.Relevance = rel + d.MatchedNodeID = matched + d.ReceivedAt = nsTime(recv) + d.UpdatedAt = nsTime(upd) + return d, nil +} + +func (s *Store) ListInbox(ctx context.Context, status string, limit int) ([]InboxDocument, error) { + if limit < 1 { + limit = 100 + } + if limit > 1000 { + limit = 1000 + } + rows, err := s.db.QueryContext(ctx, `SELECT id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,status,relevance,matched_node_id FROM source_inbox WHERE (?='' OR status=?) ORDER BY received_at_ns DESC LIMIT ?`, status, status, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []InboxDocument + for rows.Next() { + d, err := scanInbox(rows) + if err != nil { + return nil, err + } + out = append(out, d) + } + return out, rows.Err() +} + +func (s *Store) Stats(ctx context.Context) (InboxStats, error) { + rows, err := s.db.QueryContext(ctx, `SELECT status,COUNT(*) FROM source_inbox GROUP BY status`) + if err != nil { + return InboxStats{}, err + } + defer rows.Close() + var st InboxStats + for rows.Next() { + var status string + var n int + if err := rows.Scan(&status, &n); err != nil { + return st, err + } + st.Total += n + switch status { + case "received": + st.Received = n + case "processing": + st.Processing = n + case "candidate": + st.Candidate = n + case "archived": + st.Archived = n + case "used": + st.Used = n + } + } + return st, rows.Err() +} + +func (s *Store) SearchCandidates(ctx context.Context, query string, limit int, maxAge time.Duration) ([]ScoredDocument, error) { + if limit < 1 { + limit = 3 + } + if limit > 50 { + limit = 50 + } + since := int64(0) + if maxAge > 0 { + since = time.Now().UTC().Add(-maxAge).UnixNano() + } + rows, err := s.db.QueryContext(ctx, `SELECT id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,status,relevance,matched_node_id,signature FROM source_inbox WHERE status IN ('candidate','used') AND (?=0 OR COALESCE(NULLIF(published_at_ns,0),received_at_ns)>=?) ORDER BY updated_at_ns DESC LIMIT 5000`, since, since) + if err != nil { + return nil, err + } + defer rows.Close() + qsig := simhash64(query) + qterms := termSet(query) + var out []ScoredDocument + for rows.Next() { + var d InboxDocument + var ext, urlv, canon, title, lang, ctype, text, sha, source, base, cats, meta, status, matched string + var pub, disc, recv, upd, sig int64 + var rel float64 + if err := rows.Scan(&d.ID, &d.AgentID, &d.TaskID, &ext, &urlv, &canon, &title, &pub, &disc, &recv, &upd, &lang, &ctype, &text, &sha, &source, &base, &cats, &meta, &status, &rel, &matched, &sig); err != nil { + return nil, err + } + d.Document = Document{ExternalID: ext, URL: urlv, CanonicalURL: canon, Title: title, PublishedAt: nsTime(pub), DiscoveredAt: nsTime(disc), Language: lang, ContentType: ctype, Text: text, ContentSHA256: sha, SourceName: source, SourceBaseURL: base} + _ = json.Unmarshal([]byte(cats), &d.Document.Categories) + _ = json.Unmarshal([]byte(meta), &d.Metadata) + d.Status = status + d.Relevance = rel + d.MatchedNodeID = matched + d.ReceivedAt = nsTime(recv) + d.UpdatedAt = nsTime(upd) + hamming := bits.OnesCount64(qsig ^ uint64(sig)) + hashScore := 1 - float64(hamming)/64.0 + lex := jaccard(qterms, termSet(title+" "+text)) + score := 0.45*hashScore + 0.45*lex + 0.10*rel + if lex < 0.03 && hashScore < 0.58 { + continue + } + out = append(out, ScoredDocument{InboxDocument: d, QueryScore: score}) + } + sort.Slice(out, func(i, j int) bool { return out[i].QueryScore > out[j].QueryScore }) + if len(out) > limit { + out = out[:limit] + } + return out, nil +} + +func boolInt(v bool) int { + if v { + return 1 + } + return 0 +} +func timeNS(t time.Time) int64 { + if t.IsZero() { + return 0 + } + return t.UTC().UnixNano() +} +func nsTime(v int64) time.Time { + if v <= 0 { + return time.Time{} + } + return time.Unix(0, v).UTC() +} +func uniqueStrings(values []string) []string { + seen := map[string]bool{} + out := make([]string, 0, len(values)) + for _, v := range values { + v = strings.TrimSpace(v) + if v == "" || seen[v] { + continue + } + seen[v] = true + out = append(out, v) + } + return out +} +func termSet(value string) map[string]bool { + out := map[string]bool{} + for _, r := range strings.FieldsFunc(strings.ToLower(value), func(r rune) bool { return !unicode.IsLetter(r) && !unicode.IsDigit(r) }) { + if len([]rune(r)) >= 3 { + out[r] = true + } + } + return out +} +func jaccard(a, b map[string]bool) float64 { + if len(a) == 0 || len(b) == 0 { + return 0 + } + inter := 0 + union := len(a) + for k := range b { + if a[k] { + inter++ + } else { + union++ + } + } + if union == 0 { + return 0 + } + return float64(inter) / float64(union) +} +func simhash64(value string) uint64 { + weights := [64]int{} + for token := range termSet(value) { + h := fnv.New64a() + _, _ = h.Write([]byte(token)) + x := h.Sum64() + for i := 0; i < 64; i++ { + if x&(1<= 0 { + sig |= 1 << i + } + } + return sig +} + +func (s *Store) ReleaseInbox(ctx context.Context, id, message string) error { + meta, _ := json.Marshal(map[string]any{"classification_error": strings.TrimSpace(message)}) + _, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='received',metadata_json=?,updated_at_ns=? WHERE id=?`, string(meta), time.Now().UTC().UnixNano(), id) + return err +} diff --git a/internal/sourceagent/types.go b/internal/sourceagent/types.go new file mode 100644 index 0000000..7838ff8 --- /dev/null +++ b/internal/sourceagent/types.go @@ -0,0 +1,108 @@ +package sourceagent + +import "time" + +const SchemaVersion = 1 + +type Agent struct { + ID string `json:"id"` + Name string `json:"name"` + Enabled bool `json:"enabled"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + LastSeen time.Time `json:"last_seen,omitempty"` + LastError string `json:"last_error,omitempty"` + Version string `json:"version,omitempty"` + TaskCount int `json:"task_count,omitempty"` +} + +type Task struct { + ID string `json:"id"` + AgentID string `json:"agent_id"` + Name string `json:"name"` + Type string `json:"type"` + URL string `json:"url"` + Enabled bool `json:"enabled"` + PollInterval string `json:"poll_interval"` + Categories []string `json:"categories,omitempty"` + MaxItems int `json:"max_items,omitempty"` + Config map[string]string `json:"config,omitempty"` + CreatedAt time.Time `json:"created_at,omitempty"` + UpdatedAt time.Time `json:"updated_at,omitempty"` +} + +type RemoteConfig struct { + SchemaVersion int `json:"schema_version"` + Agent Agent `json:"agent"` + Tasks []Task `json:"tasks"` + IssuedAt time.Time `json:"issued_at"` +} + +type Document struct { + ExternalID string `json:"external_id,omitempty"` + URL string `json:"url"` + CanonicalURL string `json:"canonical_url,omitempty"` + Title string `json:"title"` + PublishedAt time.Time `json:"published_at,omitempty"` + DiscoveredAt time.Time `json:"discovered_at,omitempty"` + Language string `json:"language,omitempty"` + ContentType string `json:"content_type,omitempty"` + Text string `json:"text"` + ContentSHA256 string `json:"content_sha256,omitempty"` + SourceName string `json:"source_name,omitempty"` + SourceBaseURL string `json:"source_base_url,omitempty"` + Categories []string `json:"categories,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type IngestBatch struct { + SchemaVersion int `json:"schema_version"` + AgentID string `json:"agent_id,omitempty"` + TaskID string `json:"task_id"` + Documents []Document `json:"documents"` +} + +type IngestResult struct { + Accepted int `json:"accepted"` + Duplicates int `json:"duplicates"` + Rejected int `json:"rejected"` + BatchID string `json:"batch_id"` +} + +type Heartbeat struct { + AgentID string `json:"agent_id,omitempty"` + Version string `json:"version,omitempty"` + Status string `json:"status,omitempty"` + LastRunAt time.Time `json:"last_run_at,omitempty"` + LastError string `json:"last_error,omitempty"` + TasksChecked int `json:"tasks_checked,omitempty"` + Documents int `json:"documents,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type InboxDocument struct { + ID string `json:"id"` + AgentID string `json:"agent_id"` + TaskID string `json:"task_id"` + Document Document `json:"document"` + Status string `json:"status"` + Relevance float64 `json:"relevance,omitempty"` + MatchedNodeID string `json:"matched_node_id,omitempty"` + ReceivedAt time.Time `json:"received_at"` + UpdatedAt time.Time `json:"updated_at"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type InboxStats struct { + Total int `json:"total"` + Received int `json:"received"` + Processing int `json:"processing"` + Candidate int `json:"candidate"` + Archived int `json:"archived"` + Used int `json:"used"` +} + +type ScoredDocument struct { + InboxDocument + QueryScore float64 `json:"query_score"` +} diff --git a/internal/web/server.go b/internal/web/server.go index c679e0c..c8e69b9 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -18,16 +18,18 @@ import ( "github.com/local/glpi-neural-brain/internal/engine" "github.com/local/glpi-neural-brain/internal/graph" "github.com/local/glpi-neural-brain/internal/model" + "github.com/local/glpi-neural-brain/internal/sourceagent" ) //go:embed static/* var assets embed.FS type Server struct { - Engine *engine.Engine - Graph *graph.Store - Broker *activity.Broker - APIKey string + Engine *engine.Engine + Graph *graph.Store + Broker *activity.Broker + APIKey string + SourceAgents *sourceagent.Store } func (s *Server) Handler() http.Handler { @@ -55,6 +57,20 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/glpi-kb/sync", s.handleGLPIKBSync) mux.HandleFunc("POST /api/flush", s.handleFlush) mux.HandleFunc("GET /api/state/export", s.handleStateExport) + mux.HandleFunc("GET /api/source-agents", s.handleListSourceAgents) + mux.HandleFunc("POST /api/source-agents", s.handleCreateSourceAgent) + mux.HandleFunc("PATCH /api/source-agents/{id}", s.handlePatchSourceAgent) + mux.HandleFunc("DELETE /api/source-agents/{id}", s.handleDeleteSourceAgent) + mux.HandleFunc("POST /api/source-agents/{id}/rotate-token", s.handleRotateSourceAgentToken) + mux.HandleFunc("GET /api/source-agents/{id}/tasks", s.handleListSourceTasks) + mux.HandleFunc("POST /api/source-agents/{id}/tasks", s.handleCreateSourceTask) + mux.HandleFunc("PUT /api/source-tasks/{id}", s.handleUpdateSourceTask) + mux.HandleFunc("DELETE /api/source-tasks/{id}", s.handleDeleteSourceTask) + mux.HandleFunc("GET /api/source-inbox", s.handleSourceInbox) + mux.HandleFunc("GET /api/source-inbox/status", s.handleSourceInboxStatus) + mux.HandleFunc("GET /api/v1/agent/config", s.handleAgentConfig) + mux.HandleFunc("POST /api/v1/agent/heartbeat", s.handleAgentHeartbeat) + mux.HandleFunc("POST /api/v1/agent/ingest", s.handleAgentIngest) sub, _ := fs.Sub(assets, "static") mux.HandleFunc("GET /analysis", func(w http.ResponseWriter, r *http.Request) { http.Redirect(w, r, "/analysis.html", http.StatusTemporaryRedirect) diff --git a/internal/web/source_agents.go b/internal/web/source_agents.go new file mode 100644 index 0000000..0aa9385 --- /dev/null +++ b/internal/web/source_agents.go @@ -0,0 +1,308 @@ +package web + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "strconv" + "strings" + "time" + + "github.com/local/glpi-neural-brain/internal/sourceagent" +) + +func (s *Server) sourceStoreAvailable(w http.ResponseWriter) bool { + if s.SourceAgents != nil { + return true + } + writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "source agent service is not initialized"}) + return false +} +func (s *Server) adminAuthorized(w http.ResponseWriter, r *http.Request) bool { + if s.authorized(r) { + return true + } + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return false +} + +func (s *Server) handleListSourceAgents(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + agents, err := s.SourceAgents.ListAgents(ctx) + if err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + tasks, _ := s.SourceAgents.ListTasks(ctx, "") + stats, _ := s.SourceAgents.Stats(ctx) + writeJSON(w, 200, map[string]any{"agents": agents, "tasks": tasks, "inbox": stats}) +} +func (s *Server) handleCreateSourceAgent(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + var in struct { + ID string `json:"id"` + Name string `json:"name"` + } + if err := decode(r, &in); err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + a, token, err := s.SourceAgents.CreateAgent(ctx, in.ID, in.Name) + if err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusCreated, map[string]any{"agent": a, "token": token, "token_notice": "Der Token wird nur in dieser Antwort im Klartext ausgegeben."}) +} +func (s *Server) handlePatchSourceAgent(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + var in struct { + Enabled *bool `json:"enabled"` + } + if err := decode(r, &in); err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + if in.Enabled == nil { + writeJSON(w, 400, map[string]string{"error": "enabled is required"}) + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + if err := s.SourceAgents.SetAgentEnabled(ctx, r.PathValue("id"), *in.Enabled); err != nil { + writeJSON(w, 404, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, map[string]any{"ok": true}) +} +func (s *Server) handleDeleteSourceAgent(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + if err := s.SourceAgents.DeleteAgent(ctx, r.PathValue("id")); err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, map[string]any{"ok": true}) +} +func (s *Server) handleRotateSourceAgentToken(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + token, err := s.SourceAgents.RotateToken(ctx, r.PathValue("id")) + if err != nil { + writeJSON(w, 404, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, map[string]any{"token": token, "token_notice": "Der neue Token wird nur in dieser Antwort im Klartext ausgegeben."}) +} +func (s *Server) handleListSourceTasks(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + tasks, err := s.SourceAgents.ListTasks(ctx, r.PathValue("id")) + if err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, map[string]any{"tasks": tasks}) +} +func (s *Server) handleCreateSourceTask(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + var t sourceagent.Task + if err := decode(r, &t); err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + t.AgentID = r.PathValue("id") + if strings.TrimSpace(t.ID) == "" { + t.ID = "task-" + strconv.FormatInt(time.Now().UnixNano(), 36) + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + out, err := s.SourceAgents.UpsertTask(ctx, t) + if err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusCreated, out) +} +func (s *Server) handleUpdateSourceTask(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + var t sourceagent.Task + if err := decode(r, &t); err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + t.ID = r.PathValue("id") + if t.AgentID == "" { + writeJSON(w, 400, map[string]string{"error": "agent_id required"}) + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + out, err := s.SourceAgents.UpsertTask(ctx, t) + if err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, out) +} +func (s *Server) handleDeleteSourceTask(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + if err := s.SourceAgents.DeleteTask(ctx, r.PathValue("id")); err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, map[string]any{"ok": true}) +} +func (s *Server) handleSourceInbox(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + limit := 100 + if n, err := strconv.Atoi(r.URL.Query().Get("limit")); err == nil && n > 0 && n <= 1000 { + limit = n + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + docs, err := s.SourceAgents.ListInbox(ctx, strings.TrimSpace(r.URL.Query().Get("status")), limit) + if err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, map[string]any{"documents": docs}) +} +func (s *Server) handleSourceInboxStatus(w http.ResponseWriter, r *http.Request) { + if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) { + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + st, err := s.SourceAgents.Stats(ctx) + if err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, st) +} + +func bearerToken(r *http.Request) string { + v := strings.TrimSpace(r.Header.Get("Authorization")) + if strings.HasPrefix(v, "Bearer ") { + return strings.TrimSpace(strings.TrimPrefix(v, "Bearer ")) + } + return "" +} +func (s *Server) authenticateSourceAgent(r *http.Request) (sourceagent.Agent, error) { + if s.SourceAgents == nil { + return sourceagent.Agent{}, errors.New("source agent service unavailable") + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + a, err := s.SourceAgents.Authenticate(ctx, bearerToken(r)) + if err != nil { + return a, err + } + if want := strings.TrimSpace(r.Header.Get("X-Brain-Agent-ID")); want != "" && want != a.ID { + return sourceagent.Agent{}, errors.New("agent id does not match token") + } + return a, nil +} +func (s *Server) handleAgentConfig(w http.ResponseWriter, r *http.Request) { + a, err := s.authenticateSourceAgent(r) + if err != nil { + writeJSON(w, 401, map[string]string{"error": "unauthorized"}) + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + cfg, err := s.SourceAgents.RemoteConfig(ctx, a) + if err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, cfg) +} +func (s *Server) handleAgentHeartbeat(w http.ResponseWriter, r *http.Request) { + a, err := s.authenticateSourceAgent(r) + if err != nil { + writeJSON(w, 401, map[string]string{"error": "unauthorized"}) + return + } + var h sourceagent.Heartbeat + if err := decode(r, &h); err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + if h.AgentID != "" && h.AgentID != a.ID { + writeJSON(w, 403, map[string]string{"error": "agent_id mismatch"}) + return + } + ctx, cancel := contextTimeout(r, 10*time.Second) + defer cancel() + if err := s.SourceAgents.Heartbeat(ctx, a.ID, h); err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, 200, map[string]any{"ok": true, "server_time": time.Now().UTC()}) +} +func (s *Server) handleAgentIngest(w http.ResponseWriter, r *http.Request) { + a, err := s.authenticateSourceAgent(r) + if err != nil { + writeJSON(w, 401, map[string]string{"error": "unauthorized"}) + return + } + var batch sourceagent.IngestBatch + dec := json.NewDecoder(io.LimitReader(r.Body, 32<<20)) + if err := dec.Decode(&batch); err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + if batch.AgentID != "" && batch.AgentID != a.ID { + writeJSON(w, 403, map[string]string{"error": "agent_id mismatch"}) + return + } + if batch.SchemaVersion != 0 && batch.SchemaVersion != sourceagent.SchemaVersion { + writeJSON(w, 400, map[string]string{"error": "unsupported schema_version"}) + return + } + if strings.TrimSpace(batch.TaskID) == "" || len(batch.Documents) == 0 || len(batch.Documents) > 500 { + writeJSON(w, 400, map[string]string{"error": "task_id and 1..500 documents are required"}) + return + } + ctx, cancel := contextTimeout(r, 60*time.Second) + defer cancel() + result, err := s.SourceAgents.Ingest(ctx, a.ID, batch.TaskID, batch.Documents) + if err != nil { + writeJSON(w, 500, map[string]string{"error": err.Error()}) + return + } + _ = s.SourceAgents.Heartbeat(ctx, a.ID, sourceagent.Heartbeat{AgentID: a.ID, Status: "ingest", Documents: result.Accepted}) + writeJSON(w, http.StatusAccepted, result) +} diff --git a/internal/web/static/index.html b/internal/web/static/index.html index a81dfe8..a72c629 100644 --- a/internal/web/static/index.html +++ b/internal/web/static/index.html @@ -61,6 +61,7 @@ + AGENTS ANALYSE diff --git a/internal/web/static/source-agents.css b/internal/web/static/source-agents.css new file mode 100644 index 0000000..e22eea0 --- /dev/null +++ b/internal/web/static/source-agents.css @@ -0,0 +1,3 @@ +:root{color-scheme:dark;font-family:Inter,ui-sans-serif,system-ui,sans-serif;background:#040810;color:#dcecff}*{box-sizing:border-box}body{margin:0;background:radial-gradient(circle at 70% 0,#0b2637 0,#040810 42%);min-height:100vh}header{position:sticky;top:0;z-index:5;display:flex;justify-content:space-between;align-items:center;padding:18px 28px;background:#07111bd9;border-bottom:1px solid #173244;backdrop-filter:blur(18px)}header strong{display:block;letter-spacing:.16em;color:#66e3ff}header small{color:#7f9bad}nav{display:flex;gap:10px}a,button{color:#ccefff;background:#0a1a27;border:1px solid #285069;border-radius:9px;padding:9px 13px;text-decoration:none;cursor:pointer}button:hover,a:hover{border-color:#52dfff}main{width:min(1500px,96vw);margin:24px auto 60px;display:grid;gap:18px}.panel{background:#07121dcc;border:1px solid #17384b;border-radius:16px;padding:18px;box-shadow:0 20px 60px #0006}.panel h2{margin:0 0 8px;font-size:17px}.panel p{margin:0;color:#7896a9;font-size:13px}.auth{display:flex;gap:12px;align-items:center}.auth>div{margin-right:auto}.auth input{max-width:360px}.stats{display:grid;grid-template-columns:repeat(4,1fr);gap:14px}.stats article{padding:18px;border:1px solid #17384b;border-radius:14px;background:#07121dcc}.stats b{display:block;font-size:28px;color:#64e2ff}.stats span{color:#829dad}.grid{display:grid;grid-template-columns:1fr 1fr;gap:18px}label{display:grid;gap:6px;margin-top:12px;color:#8eaabb;font-size:12px}input,select{width:100%;padding:10px 11px;border:1px solid #25465b;border-radius:8px;background:#030a10;color:#e5f5ff}.two{display:grid;grid-template-columns:1fr 1fr;gap:10px}.panel>button{margin-top:14px}.cards{display:grid;grid-template-columns:repeat(auto-fit,minmax(310px,1fr));gap:12px;margin-top:14px}.card{border:1px solid #19384b;border-radius:12px;padding:14px;background:#050d15}.card-head{display:flex;justify-content:space-between;gap:12px}.card small,.muted{color:#7692a3}.status-online{color:#68f6b1}.status-offline{color:#ffb06b}.actions{display:flex;flex-wrap:wrap;gap:7px;margin-top:12px}.actions button{padding:6px 9px;font-size:11px}.task{margin-top:10px;padding:9px;border-left:2px solid #31a7cd;background:#081722}.task b{font-size:12px}.task span{display:block;color:#7994a5;font-size:11px;overflow-wrap:anywhere}.token-output{white-space:pre-wrap;word-break:break-all;background:#02070c;border:1px solid #2a5871;padding:12px;border-radius:10px;color:#9ff0ff}.hidden{display:none}.title-row{display:flex;justify-content:space-between;align-items:center;gap:12px}.title-row select{width:auto}.table-wrap{overflow:auto;margin-top:14px}table{width:100%;border-collapse:collapse;font-size:12px}th,td{padding:10px;border-bottom:1px solid #142b39;text-align:left;vertical-align:top}th{color:#79a6bb}td a{padding:0;border:0;background:none;color:#79dcff}.pill{display:inline-block;padding:3px 7px;border-radius:99px;background:#123246;color:#aeefff}.pill.candidate{background:#143c30;color:#8dffc5}.pill.archived{background:#2b2f36;color:#aab3bd}.pill.received,.pill.processing{background:#3a3215;color:#ffe59a}#toast{position:fixed;right:20px;bottom:20px;max-width:440px;padding:12px 16px;border-radius:10px;background:#102c3c;border:1px solid #3a718c;opacity:0;pointer-events:none;transition:.2s}#toast.show{opacity:1}@media(max-width:850px){.grid,.stats{grid-template-columns:1fr}.auth{align-items:stretch;flex-direction:column}.auth>div{margin:0}header{padding:14px}.title-row{align-items:flex-start;flex-direction:column}} + +label.check{display:flex;grid-template-columns:auto 1fr;align-items:center;gap:8px}.check input{width:auto} diff --git a/internal/web/static/source-agents.html b/internal/web/static/source-agents.html new file mode 100644 index 0000000..a20b7b2 --- /dev/null +++ b/internal/web/static/source-agents.html @@ -0,0 +1,24 @@ + + + + + + Source Agents · Neural Brain + + + +
SOURCE AGENTSVerteilte Quellenbeobachtung · Inbox vor SearXNG
+
+

Administration

Falls BRAIN_API_KEY gesetzt ist, wird er nur in dieser Browser-Session gehalten.

+
0Agents
0Tasks
0Inbox-Kandidaten
0Queue
+
+

Agent anlegen

+

Polling-Aufgabe

+
+

Agents

Tokenrechte: config · heartbeat · ingest. Kein Graph-/Adminzugriff.

+

Source Inbox

Nur als candidate klassifizierte Dokumente werden vor SearXNG durchsucht. Graph-Nodes entstehen erst nach tatsächlichem Claim-Grounding.

StatusTitel / QuelleRelevanzAgent / TaskEingang
+
+
+ + + diff --git a/internal/web/static/source-agents.js b/internal/web/static/source-agents.js new file mode 100644 index 0000000..2602e96 --- /dev/null +++ b/internal/web/static/source-agents.js @@ -0,0 +1,18 @@ +(() => { + const $=id=>document.getElementById(id); let snapshot={agents:[],tasks:[],inbox:{}}; + $('apiKey').value=sessionStorage.getItem('brain_api_key')||''; + $('saveKey').onclick=()=>{sessionStorage.setItem('brain_api_key',$('apiKey').value.trim());toast('API-Key für diese Session gespeichert.');loadAll()}; + function headers(){const h={'Content-Type':'application/json'},key=sessionStorage.getItem('brain_api_key')||'';if(key)h.Authorization=`Bearer ${key}`;return h} + async function api(path,opt={}){const res=await fetch(path,{...opt,headers:{...headers(),...(opt.headers||{})}});const data=await res.json().catch(()=>({}));if(!res.ok)throw new Error(data.error||`HTTP ${res.status}`);return data} + function toast(msg){const t=$('toast');t.textContent=msg;t.classList.add('show');setTimeout(()=>t.classList.remove('show'),3500)} + function esc(v){return String(v??'').replace(/[&<>"']/g,c=>({'&':'&','<':'<','>':'>','"':'"',"'":'''}[c]))} + function when(v){if(!v)return 'nie';const d=new Date(v);if(Number.isNaN(d.getTime()))return 'nie';return d.toLocaleString()} + async function loadAll(){try{snapshot=await api('/api/source-agents');render();await loadInbox()}catch(e){toast(e.message)}} + function render(){const {agents=[],tasks=[],inbox={}}=snapshot;$('agentCount').textContent=agents.length;$('taskCount').textContent=tasks.length;$('candidateCount').textContent=inbox.candidate||0;$('receivedCount').textContent=(inbox.received||0)+(inbox.processing||0);$('taskAgent').innerHTML=agents.map(a=>``).join('');$('agents').innerHTML=agents.length?agents.map(a=>{const own=tasks.filter(t=>t.agent_id===a.id),online=a.last_seen&&Date.now()-new Date(a.last_seen).getTime()<15*60*1000;return `
${esc(a.name)}${esc(a.id)}
● ${online?'online':'offline'}
Version ${esc(a.version||'—')} · letzter Kontakt ${esc(when(a.last_seen))}${a.last_error?`

${esc(a.last_error)}

`:''}
${own.map(t=>`
${esc(t.name)} · ${esc(t.type)} · ${esc(t.poll_interval)}${esc(t.url)}
`).join('')}
`}).join(''):'

Noch keine Agenten konfiguriert.

';wireActions()} + function wireActions(){document.querySelectorAll('[data-rotate]').forEach(b=>b.onclick=async()=>{try{const d=await api(`/api/source-agents/${b.dataset.rotate}/rotate-token`,{method:'POST'});showToken(b.dataset.rotate,d.token);await loadAll()}catch(e){toast(e.message)}});document.querySelectorAll('[data-toggle]').forEach(b=>b.onclick=async()=>{try{await api(`/api/source-agents/${b.dataset.toggle}`,{method:'PATCH',body:JSON.stringify({enabled:b.dataset.enabled!=='true'})});await loadAll()}catch(e){toast(e.message)}});document.querySelectorAll('[data-delete-agent]').forEach(b=>b.onclick=async()=>{if(!confirm('Agent inklusive Tasks löschen?'))return;try{await api(`/api/source-agents/${b.dataset.deleteAgent}`,{method:'DELETE'});await loadAll()}catch(e){toast(e.message)}});document.querySelectorAll('[data-delete-task]').forEach(b=>b.onclick=async()=>{try{await api(`/api/source-tasks/${b.dataset.deleteTask}`,{method:'DELETE'});await loadAll()}catch(e){toast(e.message)}})} + function showToken(id,token){const brain=location.origin;$('tokenOutput').textContent=`Token nur jetzt kopieren:\n${token}\n\nAgent-ENV:\nBRAIN_MODE=agent\nBRAIN_AGENT_BRAIN_URL=${brain}\nBRAIN_AGENT_ID=${id}\nBRAIN_AGENT_TOKEN=${token}`;$('tokenOutput').classList.remove('hidden')} + $('createAgent').onclick=async()=>{try{const d=await api('/api/source-agents',{method:'POST',body:JSON.stringify({id:$('agentID').value.trim(),name:$('agentName').value.trim()})});showToken(d.agent.id,d.token);$('agentName').value='';$('agentID').value='';await loadAll()}catch(e){toast(e.message)}}; + $('createTask').onclick=async()=>{const agent=$('taskAgent').value;if(!agent)return toast('Zuerst einen Agent anlegen.');const task={name:$('taskName').value.trim(),type:$('taskType').value,url:$('taskURL').value.trim(),enabled:true,poll_interval:$('taskInterval').value.trim(),categories:$('taskCategories').value.split(',').map(x=>x.trim()).filter(Boolean),max_items:Number($('taskMaxItems').value)||20,config:{refetch_seen:$('taskRefetchSeen').checked?'true':'false'}};try{await api(`/api/source-agents/${encodeURIComponent(agent)}/tasks`,{method:'POST',body:JSON.stringify(task)});$('taskName').value='';$('taskURL').value='';await loadAll();toast('Polling-Aufgabe gespeichert.')}catch(e){toast(e.message)}}; + async function loadInbox(){try{const status=$('inboxFilter').value;const d=await api(`/api/source-inbox?limit=100${status?`&status=${encodeURIComponent(status)}`:''}`);$('inbox').innerHTML=(d.documents||[]).map(x=>`${esc(x.status)}${esc(x.document.title)}
${esc(x.document.source_name||x.document.source_base_url||'')}${(Number(x.relevance||0)*100).toFixed(0)}%${esc(x.agent_id)}
${esc(x.task_id)}${esc(when(x.received_at))}`).join('')}catch(e){toast(e.message)}} + $('inboxFilter').onchange=loadInbox;$('reload').onclick=loadAll;loadAll();setInterval(loadAll,30000); +})();