diff --git a/.env.example b/.env.example
index 4487faf..b82e51d 100644
--- a/.env.example
+++ b/.env.example
@@ -104,6 +104,8 @@ WORKER_SHELL_TIMEOUT=120
 WORKER_SHELL_TIMEOUT_MAX=600
 WORKER_SHELL_CAPTURE_MAX_BYTES=524288
 WORKER_HTTP_MAX_BYTES=1048576
+# 情报库有效期（秒）：last_seen 距今超过该值即视为过时，不再注入给后续 Worker。默认 7 天。
+INTEL_MAX_AGE_SECONDS=604800
 WORKER_OUTPUT_TRUNCATE=4096
 WORKER_LLM_TOOL_OUTPUT_TRUNCATE=4096
 WORKER_HISTORY_FULL_TOOL_ROUNDS=4
diff --git a/app/agents/intel.py b/app/agents/intel.py
index c4fb42e..fcfd063 100644
--- a/app/agents/intel.py
+++ b/app/agents/intel.py
@@ -15,7 +15,7 @@
 
 import hashlib
 import os
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -64,6 +64,9 @@
 _MAX_INJECT_CHARS = int(os.environ.get("INTEL_MAX_INJECT_CHARS", "1800"))
 # 内容字段写入前的硬截断（防异常超长）
 _MAX_FIELD = 300
+# 情报有效期（秒）：last_seen 距今超过该值即视为过时，不再注入给后续 Worker。
+# 过时凭证/端点会加大误用与 token 浪费，这里按时间窗口过滤而非全量注入。
+_MAX_AGE_SECONDS = int(os.environ.get("INTEL_MAX_AGE_SECONDS", str(7 * 24 * 3600)))
 
 
 def _now() -> datetime:
@@ -186,12 +189,16 @@ async def lookup_intel(
     try:
         root = (root or "").strip().lower()
         fps = [f for f in (fingerprints or []) if f]
+        # 只检索有效期内的情报；last_seen 在建行时有默认值，不会为 NULL。
+        # SQLite 在 naive DateTime 列上以 naive UTC 存储，这里用相同形态避免 aware/naive 比较出错。
+        fresh_cutoff = (datetime.now(timezone.utc) - timedelta(seconds=_MAX_AGE_SECONDS)).replace(tzinfo=None)
 
         # cred / profile 按 root 域命中
         if root:
             for kind in ("cred", "profile"):
                 rows = (await session.execute(
-                    select(Intel).where(Intel.kind == kind, Intel.match_key == root)
+                    select(Intel).where(Intel.kind == kind, Intel.match_key == root,
+                                        Intel.last_seen >= fresh_cutoff)
                     .order_by(Intel.confidence.desc(), Intel.hit_count.desc(), Intel.last_seen.desc())
                     .limit(_MAX_PER_KIND)
                 )).scalars().all()
@@ -201,7 +208,8 @@ async def lookup_intel(
         if fps:
             for kind in ("fingerprint", "endpoint"):
                 rows = (await session.execute(
-                    select(Intel).where(Intel.kind == kind, Intel.match_key.in_(fps))
+                    select(Intel).where(Intel.kind == kind, Intel.match_key.in_(fps),
+                                        Intel.last_seen >= fresh_cutoff)
                     .order_by(Intel.confidence.desc(), Intel.hit_count.desc(), Intel.last_seen.desc())
                     .limit(_MAX_PER_KIND)
                 )).scalars().all()
diff --git a/tests/test_intel_freshness.py b/tests/test_intel_freshness.py
new file mode 100644
index 0000000..d054328
--- /dev/null
+++ b/tests/test_intel_freshness.py
@@ -0,0 +1,65 @@
+"""覆盖情报库有效期过滤：lookup_intel 应忽略 last_seen 超过 INTEL_MAX_AGE_SECONDS 的过时情报。
+
+用内存 SQLite + aiosqlite 构造异步会话，插入一条新鲜、一条过时情报，断言只返回新鲜那条。
+"""
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.agents.intel import lookup_intel
+from app.db.models import Base, Intel
+
+# 与 intel.py 默认一致的 7 天有效期，便于测试可读。
+_DEFAULT_TTL_DAYS = 7
+
+
+class TestIntelFreshness(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+        async with self.engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+        self.Session = async_sessionmaker(self.engine, expire_on_commit=False)
+
+    async def asyncTearDown(self):
+        await self.engine.dispose()
+
+    async def _seed(self):
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
+        async with self.Session() as s:
+            s.add(Intel(
+                kind="cred", match_key="example.com", dedup_hash="fresh",
+                payload={"username": "u", "password": "p"}, summary="fresh cred",
+                confidence="verified", last_seen=now,
+            ))
+            s.add(Intel(
+                kind="cred", match_key="example.com", dedup_hash="stale",
+                payload={"username": "v", "password": "q"}, summary="stale cred",
+                confidence="verified", last_seen=now - timedelta(days=_DEFAULT_TTL_DAYS + 1),
+            ))
+            await s.commit()
+
+    async def test_stale_intel_excluded(self):
+        await self._seed()
+        async with self.Session() as s:
+            result = await lookup_intel(s, "example.com", [])
+        creds = result.get("cred") or []
+        self.assertEqual(len(creds), 1, creds)
+        self.assertEqual(creds[0].dedup_hash, "fresh")
+
+    async def test_fresh_intel_returned_when_within_ttl(self):
+        now = datetime.now(timezone.utc).replace(tzinfo=None)
+        async with self.Session() as s:
+            s.add(Intel(
+                kind="endpoint", match_key="framework_ruoyi", dedup_hash="ep",
+                payload={"path": "/actuator", "vuln_type": "unauthorized_access"},
+                summary="actuator exposed", confidence="likely", last_seen=now,
+            ))
+            await s.commit()
+            result = await lookup_intel(s, root="", fingerprints=["framework_ruoyi"])
+        endpoints = result.get("endpoint") or []
+        self.assertEqual([e.dedup_hash for e in endpoints], ["ep"])
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file