# FULL MERGED MANAGER HYBRID — Unified Extended Spec
**Version:** 1.3 · **Date:** 2026-08-01
**Status:** Production-oriented merge (best-of, de-duplicated, working notes + code)
**Role:** Single extended operator manual for browser-first social automation, growth, lead gen, research, and ops.
**Scan:** All 40 folder sources tip-scanned + hybrid permanent memory cross-check. See §18.0.
---
## START HERE — Merged Omni + Hybrid (Index)
1. **Runtime CLI:** `omni-hybrid-merged/README.md` → `python3 omni_cli.py boot`
2. **AI system prompt:** `omni-hybrid-merged/SYSTEM_PROMPT.md` (or `python3 omni_cli.py prompt`)
3. **Full constitution:** this document (below)
4. **Scores / why merge:** `COMPARISON-OMNI-AGENTS-vs-UNIFIED.md`
5. **Executors:** `../social media omni credentials/media omni agents/`
```bash
cd omni-hybrid-merged && python3 check_merge.py && python3 omni_cli.py boot
```
---
## Competitive Pro-Tips Pack (from Hybrid 4.7) — Agent Edge
> Full battle cards: **§18.7**. Condensed power rules below — load these first.
| # | Power rule (must) |
|---|-------------------|
| P1 | Search opt-in only (`--factual` / `needs_search`) |
| P2 | Boot warnings block publish (unless `--force`) |
| P3 | SameSite Titlecase before cookie inject |
| P4 | IG/TT: local_storage + refresh ~45m |
| P5 | Bezier `smart_click` — never centroid `page.click` |
| P6 | PhasedGrowth <21d; playbooks = ceilings |
| P7 | Shadowban canary empty → L3 48h, no ban-loop |
| P8 | `first_60_minutes` is part of publish |
| P9 | Absolute media paths + file header check |
| P10 | One browser / one account process (sequential) |
| P11 | No sync SQLite on Playwright loop |
| P12 | Incidents 301→L1 · 308→L2 · 310→L3 → `incidents.jsonl` |
| P13 | Debug from `agent_monitor.db` last 5 errors |
| P14 | Phoenix morph **once** then halt |
| P15 | `check_ip_risk()` at boot; prefer native IP |
| P16 | Reject out-of-niche before upload footprint |
| P17 | Jitter schedules — never rigid cron |
| P18 | NSRE skip `parasitic_risk` |
| P19 | Rescue Window 5m **manual** only |
| P20 | `self_check` + runtime `health_check` dual gate |
**Adopted:** 42 full PRO TIPs from `/Users/khaledahmedmohamed/designs-content/vip/new-hybrid-4.7.md` → §18.7 (untruncated + agent actions).
---
## Table of Contents
| # | Section |
|---|---------|
| — | [Competitive Pro-Tips Pack (Hybrid 4.7)](#competitive-pro-tips-pack-from-hybrid-47--agent-edge) |
| 0 | [Merge Map & Source Attribution](#0-merge-map--source-attribution) |
| 1 | [Operating Contract](#1-operating-contract-non-negotiable) |
| 2 | [Architecture — Three Paths](#2-architecture--three-compatible-paths) |
| 3 | [Stack Decision Tree](#3-stack-decision-tree-ultimate-hybrid-stack) |
| 4 | [Accounts, Cookies, Servers](#4-accounts-cookies-servers-mode-c) |
| 5 | [Boot, Secrets, Self-Check](#5-boot-secrets-self-check) |
| 6 | [Stealth, Humanization, Resilience](#6-stealth-humanization-resilience) |
| 7 | [Pre-Publish Gates](#7-pre-publish-gates--content-quality) |
| 7b | [Extended Hybrid Modules (code library)](#7b-extended-hybrid-modules-extracted-best-of-code) |
| 8 | [Growth Playbook](#8-growth-playbook-merged-overgrowth) |
| 9 | [Lead Generation](#9-lead-generation-playbook-ethical) |
| 10 | [Research / CLEAR Lite Search](#10-research--search-extension-clear--lite) |
| 11 | [Mastery Takeaways 2026](#11-mastery-takeaways-2026) |
| 12 | [Platform Rate & Media Specs](#12-platform-rate--media-specs-working-defaults) |
| 13 | [CLI Surface](#13-cli-surface-unified-mental-model) |
| 14 | [VPS Ops (Mode C)](#14-vps-ops-notes-mode-c) |
| 15 | [Execution Checklists](#15-execution-checklists) |
| 16 | [Working Notes / Pro Tips](#16-working-notes--pro-tips-that-survive-reality) |
| 17 | [Self-Check Snippet](#17-self-check-snippet-doc-integrity) |
| 18 | [AI-Agent Acting Logic, Golden Rules, Prompts](#18-ai-agent-acting-logic--golden-rules-pro-tips--error-avoidance-prompts) |
| 19 | [OMNI + HYBRID Runtime Merge](#19-omni--hybrid-runtime-merge-implemented) |
| A–D | [Appendices](#appendix-a--full-hybrid-module-index-pointer) |
**Paths at a glance:** C = single-account `social-autopilot` · B = browser-use + Patchright · A = Hybrid 4.7 multi-account Mode C.
---
## 0. Merge Map & Source Attribution
| Priority | Source | What we kept |
|---------|--------|--------------|
| **Spine** | `new-hybrid-4.7.md` (OMNI-SOCIAL) | Operating contract, boot, auth, stealth, publish pipeline, governors, shadowban, phased growth, CLI, VPS, ethics |
| **Stack** | `Ultimate hybrid stack-media automation.md` | browser-use + Patchright/CloakBrowser hybrid, anti-bot checklist, fallback tree |
| **Lite path** | `social-autopilot/*` | One-account honest publish/reply/analytics; encrypted session; dual deterministic/agent publish |
| **Growth** | `social media overgrowth.md` + `Advanced media overgrowth.md` | Retention blueprint, hub-spoke, platform tactics, ethical algorithm playbooks |
| **Leads** | `Lead generation and media automation.md` | 11 free lead methods, Meta ads structure, tool stack (ethical only) |
| **Research** | `pro-mistreal-searcher-agent-extension.md` + slim-search | Triangulation, confidence scoring, rate-safe fetch |
| **Mastery** | `Social Media Mastery Guide.md` | 2026 takeaways, cookie failure modes, defense layers |
| **Slim skills** | `unified-slim-package/*` | Deploy discipline, hard rules, when to upgrade |
### Merge principles
1. **One spine** — Hybrid 4.7 contract is law; other docs *extend*, they do not override ethics or three-state verification.
2. **Best practice wins** — Prefer working patterns (encrypt cookies, Bezier click, boot self-check) over longer speculative code.
3. **Compatible layers** — Stack = engine; Hybrid 4.7 = product system; social-autopilot = minimal one-account path; slim skills = deploy filters.
4. **No raw dump** — Large modules truncated with pointers to full source when long.
5. **Ethics hard gate** — `passive_only`, no captcha bypass, no platform APIs in default path.
6. **Runnable-shaped code** — Prefer complete imports + functions; mark shortcuts with `ponytail:`.
### Conflict resolution
| Conflict | Winner | Why |
|----------|--------|-----|
| `page.click` vs smart_click | **smart_click / Bezier** | Hybrid + stack anti-bot consensus |
| Shared IP multi-account vs 1:1 | **1 account → 1 VPS IP** | Mode C / ServerRouter |
| API posting vs browser-only | **Browser-only** | Omni contract |
| Auto-send replies vs draft | **Draft default** | social-autopilot safety |
| Always-on search vs lite | **Lite OFF unless `--factual`** | Mode router |
| INCONCLUSIVE verification | **Blocks done** | Never promote to PASS |
---
## 1. Operating Contract (Non-Negotiable)
```
1. Self-check boot() before any browser context
2. Never raw page.click() — smart_click() cubic Bezier only
3. No platform APIs (instagrapi / tweepy / Graph API)
4. CircuitBreaker on publish/engage — abort after 3 failures
5. Lite search OFF by default; --factual / needs_search only
6. Encrypt cookies at rest; SameSite Titlecase; chmod 600
7. Strip GPT-isms: delve, tapestry, unlock, landscape, elevate, moreover, testament
8. Three-state verdict: PASS | FAIL | INCONCLUSIVE (INCONCLUSIVE ≠ PASS)
9. Factual posts: confidence ≥ 65 across ≥ 2 domains when search runs
10. Sanitize scraped text before LLM prompts
11. Quarantine on shadowban | checkpoint | session_expired | account_locked
12. NEVER parse Google HTML with regex — DDG / APIs / Playwright fallback
13. NEVER f-string URL-encode — urllib.parse.quote_plus
14. NEVER block event loop with subprocess.run(ollama) — aiohttp /api/generate
15. ethics.yaml passive_only: true blocks gray-hat modules at boot
16. One account → one VPS IP forever (Mode C)
17. PhasedGrowth caps for accounts < 21 days
18. Virality score ≥ 60 before live publish (when scorer enabled)
```
### Pipeline order (every live publish)
```
boot() → secrets_scan → ServerRouter → HybridDispatcher
→ EngagementHook.warm_up()
→ PrePublishScorer + AlgorithmMapper (+ TopicSanity)
→ maybe_search() IF --factual / needs_search
→ CaptionTransformer → publish
→ first_60_minutes() → log_post() → analytics
```
### Mode router
```python
# publish.py — mode dispatch
MODES = {
"automate": "V38AutomatorSystem", # default — publish, grow, engage
"verify": "VerificationEngine", # media + pipeline checks
"search": "ResearchPipelineV43", # optional 10% — only with --factual or needs_search
}
def resolve_mode(args) -> str:
if getattr(args, "factual", False) or getattr(args, "needs_search", False):
return "search"
if getattr(args, "verify", False):
return "verify"
return "automate"
```
> **Working note:** `publish/grow/engage` must never auto-invoke search. Pass `--factual` or set `needs_search: true` in the content brief.
---
## 2. Architecture — Three Compatible Paths
```
┌─────────────────────────────────────────────────────────────────┐
│ PATH A — OMNI / Hybrid 4.7 (full product) │
│ multi-account · ServerRouter · NSRE · shadowban · phased growth │
│ VPS Mode C · encrypted .enc · V38AutomatorSystem │
├─────────────────────────────────────────────────────────────────┤
│ PATH B — Ultimate Hybrid Stack (engine pattern) │
│ browser-use (brain) + Patchright/CloakBrowser (stealth) │
│ + residential proxy · AI adaptive when DOM breaks │
├─────────────────────────────────────────────────────────────────┤
│ PATH C — social-autopilot (minimal / one account) │
│ login once · queue.json · deterministic steps OR agent fallback │
│ draft replies · own-post analytics · no multi-account routing │
└─────────────────────────────────────────────────────────────────┘
▲ slim skill gates: automate 9.3 · media 9.1 · search 9.2
```
| Need | Path |
|------|------|
| Production multi-brand, 3 VPS, growth routines | **A** Hybrid 4.7 |
| New site / changing DOM / LLM planning | **B** stack hybrid |
| Own single account, schedule posts, human review | **C** social-autopilot |
| Fast research without browser | Slim-search / `--factual` |
### Slim hard rules
**Automate**
## Hard rules
1. self-check PASS before publish/grow
2. NO platform APIs
3. ≥11 hashtags
4. CircuitBreaker on 3 failures
5. Factual → unified-slim-search first
6. Never publish on gate_blocked
7. Save cookies after auth
8. BANNED_WORDS filter active
---
**Search**
## Hard rules (strengths only)
1. **Never search raw query** — DEFINE → WIDEN (≥6 angles) → NARROW first.
2. **Phantom default** — triangulation + CLEAR before synthesis.
3. **Nexus when** — tool hunt, broad discovery, or Phantom returns <3 sources.
4. **Dual gate** — triangulation PASS + CLEAR ≥0.70 before handoff.
5. **Tag claims** — `[VERIFIED]` / `[SINGLE-SOURCE]` / `[OPINION]` / `[SPECULATIVE]`.
6. **Contrarian pass** on every synthesis.
7. **Version-aware** — `npm ls` / `pip show` before accepting snippets.
8. **Vault-first** — read `findings.jsonl` before live fetch.
9. **Passive only** — ethics.yaml.
10. **Archive on PASS** — append findings.jsonl.
---
**Media**
## Hard rules
1. ffprobe first — never guess durations
2. OpenCV always; cloud only for heroes
3. Cloud failure never blocks pipeline
4. Audio sync verify ±0.3s
5. Max 3 fix attempts per failure
6. Attach qa_report.json before done
---
---
## 3. Stack Decision Tree (Ultimate Hybrid Stack)
```
Q1: Strong bot detection (CF / DataDome / reCAPTCHA v3)?
YES → Patchright or CloakBrowser or nodriver + residential proxy
NO → Playwright or browser-use default browser
Q2: Needs NL reasoning / multi-step / adaptive UI?
YES → browser-use Agent (brain) on stealth engine
NO → deterministic Playwright/Patchright steps only
Q3: Social media multi-account live?
YES → Hybrid 4.7 ServerRouter + 1:1 IP + encrypted cookies
NO → social-autopilot single profile is enough
```
### Install (engine layer)
```bash
pip install browser-use patchright playwright cryptography pyyaml
patchright install chromium
# Path A also: aiohttp + optional local Ollama
```
### Anti-bot shipping checklist
```
- [ ] Passes bot.sannysoft.com / browserscan / fingerprintjs
- [ ] navigator.webdriver === false (not undefined)
- [ ] TLS fingerprint matches real Chrome
- [ ] Proxy IP matches browser timezone + locale
- [ ] Mouse paths curved (Bezier), not straight lines
- [ ] Scroll before click
- [ ] Session cookies persist between runs
- [ ] Rate: ≥ 2–5s between actions + jitter
- [ ] UA matches engine version
- [ ] No deep-night spam on cold accounts
```
### Hybrid stack master pattern (browser-use + Patchright)
```python
import asyncio
import os
from patchright.async_api import async_playwright
from browser_use import Agent, Browser, BrowserConfig
from langchain_openai import ChatOpenAI
# 1. SETUP YOUR RESIDENTIAL PROXY (CRITICAL FOR SOCIAL MEDIA)
# Never use datacenter IPs for IG/TikTok. Use residential.
PROXY_URL = "http://user:pass@residential-proxy-provider:port"
async def run_hybrid_social_agent():
# 2. INITIALIZE THE STEALTH BROWSER (Patchright)
# Patchright bypasses Runtime.enable leaks and basic bot detection
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=False, # Headed mode is safer for social media
proxy={"server": PROXY_URL},
args=[
"--disable-blink-features=AutomationControlled",
"--no-sandbox"
]
)
# Create a persistent context to save cookies/sessions (Memory)
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
locale="en-US",
timezone_id="America/New_York"
)
# Block images/media to speed up AI processing (optional but saves tokens)
# await context.route("**/*.{png,jpg,jpeg,gif,svg,mp4}", lambda route: route.abort())
page = await context.new_page()
# 3. HAND THE STEALTH BROWSER TO BROWSER-USE (The AI Brain)
# We configure browser-use to use our existing stealth page
browser_config = BrowserConfig(
# browser-use allows injecting custom contexts in advanced setups
# or we just pass the initialized page directly to the agent
)
llm = ChatOpenAI(model="gpt-4o") # or Claude 3.5 Sonnet
agent = Agent(
task="""
Go to instagram.com.
Wait for the page to load. If there is a login screen, stop and tell me.
If I am logged in, search for the hashtag '#AIStartups'.
Like the first 3 posts, and leave a thoughtful, 1-sentence comment on each
about how interesting their technology is.
Act like a human: wait 3-5 seconds between actions.
""",
llm=llm,
page=page, # <--- INJECTING THE STEALTH PAGE HERE
)
print("🤖 AI Agent taking control of Stealth Browser...")
history = await agent.run()
print("✅ Task Completed. Agent History:")
print(history)
# Save cookies for next time so you don't have to log in again
await context.storage_state(path="ig_session.json")
await browser.close()
asyncio.run(run_hybrid_social_agent())
```
### social-autopilot browser session (Path C)
```python
"""
Browser session factory.
Wires up a single browser instance shared by both the deterministic
Playwright steps (core/publisher.py) and the browser-use AI agent, running
on patchright's stealth-patched Chromium so ordinary automation doesn't
get mistaken for a bot and blocked outright.
This manages ONE account/profile at a time, on purpose. If you need
several accounts, run several separate profiles/processes rather than
routing many accounts through shared infrastructure designed to disguise
them as unrelated users -- that's a different, much riskier category of
tool than this one.
"""
from pathlib import Path
from browser_use import BrowserSession
def get_browser_session(cfg: dict) -> BrowserSession:
browser_cfg = cfg["browser"]
profile_dir = Path(browser_cfg["user_data_dir"]).expanduser()
profile_dir.mkdir(parents=True, exist_ok=True)
kwargs = dict(
user_data_dir=str(profile_dir),
headless=browser_cfg.get("headless", False),
)
executable = browser_cfg.get("patchright_executable")
if executable:
kwargs["executable_path"] = executable
# If left null, browser-use falls back to its own bundled Chromium.
# Run `patchright install chromium` and point this at the resulting
# binary for better resistance to being auto-flagged as a bot.
return BrowserSession(**kwargs)
```
### Dual-path publisher (deterministic → agent fallback)
```python
"""
Publishes one item from the content queue.
Two paths, same browser session:
1. Deterministic (fast, free, reliable): if `publishing.steps` is filled
in for your target site, we drive them directly with Playwright.
2. Adaptive (flexible, costs a few LLM calls): if no steps are defined,
or the deterministic path throws, we hand the page to a browser-use
Agent and describe the goal in plain language.
This file ships with NO site-specific selectors. Fill in `steps` (config)
or rely on the adaptive prompt below, after checking that automating your
target site doesn't conflict with its terms of service -- some platforms
require using their official API instead of browser automation, even for
your own account.
"""
from typing import Any, Dict, List
from browser_use import Agent, ChatAnthropic
async def publish(item: Dict[str, Any], cfg: Dict[str, Any], browser_session) -> Dict[str, Any]:
steps = cfg.get("publishing", {}).get("steps") or []
if steps:
try:
return await _deterministic_publish(item, steps, browser_session)
except Exception as e:
print(f"[publisher] deterministic path failed ({e}); falling back to agent")
return await _adaptive_publish(item, cfg, browser_session)
async def _deterministic_publish(item: Dict[str, Any], steps: List[Dict[str, Any]],
browser_session) -> Dict[str, Any]:
page = await browser_session.get_current_page()
for step in steps:
action = step["action"]
selector = step.get("selector")
if action == "goto":
await page.goto(step["url"])
elif action == "click":
await page.click(selector)
elif action == "fill":
text = step["value"].format(**item)
await page.fill(selector, text)
elif action == "upload":
await page.set_input_files(selector, item[step["field"]])
elif action == "wait":
await page.wait_for_timeout(step.get("ms", 1000))
else:
raise ValueError(f"Unknown step action: {action}")
return {"status": "published", "method": "deterministic"}
async def _adaptive_publish(item: Dict[str, Any], cfg: Dict[str, Any], browser_session) -> Dict[str, Any]:
media_note = (
f"Attach the file at {item['media_path']} if there's an upload option. "
if item.get("media_path") else ""
)
agent = Agent(
task=(
f"Go to {cfg['account']['target_url']}, make sure you're logged in, "
f"and publish a new post with this caption: {item['caption']!r}. "
f"{media_note}"
"Only publish this one post -- do not follow, like, or interact "
"with anyone else's content along the way."
),
llm=ChatAnthropic(model=cfg.get("replies", {}).get("llm_model", "claude-sonnet-5")),
browser_session=browser_session,
)
result = await agent.run()
return {"status": "published", "method": "agent", "result": str(result)}
```
> **Working note:** Fill `publishing.steps` once the DOM is stable → zero LLM cost on repeat. Keep agent path for UI changes and first-run discovery.
---
## 4. Accounts, Cookies, Servers (Mode C)
### Naming
| Token | Pattern | Example |
|-------|---------|---------|
| platform | lowercase | `instagram`, `twitter`, `tiktok` |
| account | `{platform}_{purpose}` | `instagram_growth01` |
| server | VPS alias | `hetzner`, `contabo`, `hostinger` |
### Cookie import checklist
1. Detect platform from cookie domains
2. Optional staging: `config/cookies/{account}.json`
3. Encrypt → `sessions/{account}.enc`
4. Delete plaintext · `chmod 600` on `.enc`
5. Patch `config/servers.json` `account_routing` (1 account → 1 VPS)
6. Inject cookies **before** navigate; save **after** success
7. `refresh_silently()` every ~45m on long engage runs
### servers.json template
```json
{
"servers": [
{"id": "hetzner", "host": "46.62.228.173", "timezone": "Europe/Berlin", "proxy": null, "ssh": "ssh -i ~/.ssh/hetzner_dokploy root@46.62.228.173"},
{"id": "contabo", "host": "149.102.150.185", "timezone": "Europe/Berlin", "proxy": null, "ssh": "ssh -i ~/.ssh/contabo2_new1 root@149.102.150.185"},
{"id": "hostinger", "host": "31.97.122.87", "timezone": "Africa/Cairo", "proxy": null, "ssh": "ssh hostinger"}
],
"account_routing": {"instagram_growth01": "hetzner", "twitter_brand_main": "hostinger"},
"defaults": {"timezone": "UTC", "server": "hetzner"}
}
```
### Encrypted session store (Path C — Fernet)
```python
"""
Login-session persistence for a single account.
Uses Playwright's standard storage_state export (cookies + localStorage)
so you log in manually once, and every later run reuses that session.
Optionally encrypted at rest -- this is just "don't leave your session
token sitting around as a plain-text file," not anything platform-facing.
"""
import json
import os
from pathlib import Path
from typing import Optional
from cryptography.fernet import Fernet
def _get_key() -> bytes:
env_key = os.getenv("SESSION_ENCRYPTION_KEY")
if env_key:
return env_key.encode()
key_path = Path(".session_key")
if key_path.exists():
return key_path.read_bytes()
key = Fernet.generate_key()
key_path.write_bytes(key)
key_path.chmod(0o600)
return key
def save_storage_state(state: dict, path: str, encrypt: bool = True) -> None:
out = Path(path)
out.parent.mkdir(parents=True, exist_ok=True)
raw = json.dumps(state).encode()
if encrypt:
raw = Fernet(_get_key()).encrypt(raw)
out.write_bytes(raw)
out.chmod(0o600)
def load_storage_state(path: str, encrypt: bool = True) -> Optional[dict]:
p = Path(path)
if not p.exists():
return None
raw = p.read_bytes()
if encrypt:
raw = Fernet(_get_key()).decrypt(raw)
return json.loads(raw)
```
### Cookie manager core (Path A)
```python
# core/cookie_manager.py
import json, os, hashlib, base64
from pathlib import Path
from typing import List, Dict, Optional
class EncryptedCookieManager:
def __init__(self):
self.key = os.environ.get("ENCRYPTION_KEY")
if self.key:
# Derive 32-byte URL-safe base64 key from hex or raw string via SHA-256
derived = hashlib.sha256(self.key.encode()).digest()
self.key_b64 = base64.urlsafe_b64encode(derived)
else:
self.key_b64 = None
self.dir = Path("sessions")
self.dir.mkdir(exist_ok=True)
def save(self, account_id: str, cookies: List[Dict]) -> bool:
sanitized = [self.sanitize_cookie(c) for c in cookies]
raw_data = json.dumps(sanitized).encode("utf-8")
if self.key_b64:
try:
from cryptography.fernet import Fernet
cipher = Fernet(self.key_b64)
data = cipher.encrypt(raw_data)
suffix = ".enc"
except ImportError:
data = raw_data
suffix = ".json"
else:
data = raw_data
suffix = ".json"
(self.dir / f"{account_id}{suffix}").write_bytes(data)
return True
def load(self, account_id: str) -> Optional[List[Dict]]:
enc_path = self.dir / f"{account_id}.enc"
json_path = self.dir / f"{account_id}.json"
try:
if enc_path.exists() and self.key_b64:
from cryptography.fernet import Fernet
cipher = Fernet(self.key_b64)
raw = cipher.decrypt(enc_path.read_bytes())
return json.loads(raw.decode("utf-8"))
elif json_path.exists():
return json.loads(json_path.read_text())
except Exception:
pass
return None
def quarantine(self, account_id: str) -> bool:
enc_path = self.dir / f"{account_id}.enc"
json_path = self.dir / f"{account_id}.json"
quarantine_enc = self.dir / f"{account_id}_expired.enc"
quarantine_json = self.dir / f"{account_id}_expired.json"
try:
if enc_path.exists():
enc_path.rename(quarantine_enc)
return True
elif json_path.exists():
json_path.rename(quarantine_json)
return True
except Exception:
pass
return False
def sanitize_cookie(self, c: Dict) -> Dict:
# Replaced docstring with comments to prevent triple quote nesting issues
# Playwright strict type formatter: resolves case sensitivity in keys.
out = {"name": c.get("name", ""), "value": c.get("value", ""),
"domain": c.get("domain", ""), "path": c.get("path", "/")}
if "expires" in c: out["expires"] = c["expires"]
elif "expirationDate" in c: out["expires"] = c["expirationDate"]
if "httpOnly" in c: out["httpOnly"] = c["httpOnly"]
if "secure" in c: out["secure"] = c["secure"]
# Sanitize sameSite value to capitalize
ss = c.get("sameSite", "Lax")
if isinstance(ss, str):
ss = ss.lower()
if ss == "no_restriction": out["sameSite"] = "None"
elif ss == "lax": out["sameSite"] = "Lax"
elif ss == "strict": out["sameSite"] = "Strict"
return out
```
### Server router
```python
# core/server_router.py
import json
from pathlib import Path
from typing import Dict, Optional
class ServerRouter:
def __init__(self, path: str = "config/servers.json"):
self.path = Path(path)
self._data = self._load()
def _load(self) -> dict:
if self.path.exists():
try:
return json.loads(self.path.read_text())
except Exception:
pass
return {"servers": [], "account_routing": {}}
def server_for(self, account_id: str) -> Optional[dict]:
sid = self._data.get("account_routing", {}).get(account_id)
for s in self._data.get("servers", []):
if s.get("id") == sid:
return s
return None
def proxy_for(self, account_id: str) -> Optional[Dict]:
srv = self.server_for(account_id)
if not srv or not srv.get("proxy"):
return None
return {"server": srv["proxy"]}
def timezone_for(self, account_id: str) -> str:
srv = self.server_for(account_id)
return (srv or {}).get("timezone", "UTC")
def assign_account(self, account_id: str, preferred: str = None) -> str:
"""Load-balance: pick server with fewest routed accounts."""
routing = self._data.get("account_routing", {})
if preferred:
routing[account_id] = preferred
else:
counts = {s["id"]: 0 for s in self._data.get("servers", [])}
for sid in routing.values():
counts[sid] = counts.get(sid, 0) + 1
routing[account_id] = min(counts, key=counts.get) if counts else "hetzner"
self._data["account_routing"] = routing
self.path.parent.mkdir(parents=True, exist_ok=True)
self.path.write_text(json.dumps(self._data, indent=2))
return routing[account_id]
```
### Dispatcher
```python
# core/dispatcher.py
import asyncio, logging
from typing import Dict, Optional
log = logging.getLogger("dispatcher")
PLATFORM_URLS = {
"instagram": "https://www.instagram.com/",
"facebook": "https://www.facebook.com/",
"twitter": "https://x.com/",
"tiktok": "https://www.tiktok.com/",
"youtube": "https://www.youtube.com/",
"linkedin": "https://www.linkedin.com/",
}
class HybridDispatcher:
# Replaced docstring with comments to prevent triple quote nesting issues
# Cascade: cookies -> credentials -> realtime re-auth (No WebBridge dependency).
def __init__(self, account_id: str, platform: str, headless: bool = False):
self.account_id = account_id
self.platform = platform
self.headless = headless
self.priority = ["cookies", "credentials"]
async def get_session(self, prefer: str = "auto") -> Dict:
methods = [prefer] if prefer != "auto" else self.priority
last_err = None
for method in methods:
try:
session = await getattr(self, f"_session_{method}")()
if session and session.get("success"):
log.info(f"auth={method} account={self.account_id}")
return session
except Exception as e:
last_err = e
continue
raise RuntimeError(f"No auth for {self.account_id}: {last_err}")
async def _session_cookies(self) -> Dict:
from core.cookie_manager import EncryptedCookieManager
from core.stealth_browser import StealthBrowser
mgr = EncryptedCookieManager()
cookies = mgr.load(self.account_id) or mgr.load(f"{self.platform}_{self.account_id}")
if not cookies:
raise RuntimeError("no cookies found")
p, ctx, page, fp = await StealthBrowser().launch(self.account_id, self.platform, self.headless)
await ctx.add_cookies(cookies)
return {"success": True, "type": "playwright", "playwright": p, "context": ctx,
"page": page, "fingerprint": fp, "auth_method": "cookies"}
async def _session_credentials(self) -> Dict:
from core.stealth_browser import StealthBrowser
from core.cookie_manager import EncryptedCookieManager
p, ctx, page, fp = await StealthBrowser().launch(self.account_id, self.platform, headless=False)
await page.goto(PLATFORM_URLS.get(self.platform, PLATFORM_URLS["instagram"]))
log.warning(f"Manual login required. Waiting 120s for {self.account_id}...")
await asyncio.sleep(120)
mgr = EncryptedCookieManager()
mgr.save(self.account_id, await ctx.cookies())
return {"success": True, "type": "playwright", "playwright": p, "context": ctx,
"page": page, "fingerprint": fp, "auth_method": "credentials"}
async def close(self, session: Dict):
if session.get("type") == "playwright":
ctx = session.get("context")
pw = session.get("playwright")
if ctx:
await ctx.close()
if pw:
await pw.stop()
```
> **Working note — why cookies fail:** Domain mismatch, expired session, `SameSite=None` without Secure, inject *after* navigate, or sharing one IP across unrelated accounts. Fix order: re-export on **same** VPS IP → re-encrypt → dry-run.
---
## 5. Boot, Secrets, Self-Check
```python
# core/boot.py
import json, os, shutil, subprocess
from pathlib import Path
class AutoDiscovery:
TOOLS = ("ffmpeg", "ffprobe", "ollama", "node", "curl")
COOKIE_DIRS = (Path("config/cookies"), Path("cookies"), Path.home() / "Desktop" / "cookies")
@staticmethod
def detect_tools() -> dict:
tools = {t: shutil.which(t) is not None for t in AutoDiscovery.TOOLS}
tools["webbridge"] = False # Disabled per directive
try:
import curl_cffi
tools["curl_cffi"] = True
except ImportError:
tools["curl_cffi"] = False
return tools
@staticmethod
def find_cookies() -> dict:
found = {}
for d in AutoDiscovery.COOKIE_DIRS:
if not d.is_dir(): continue
for f in d.glob("*.json"):
try:
data = json.loads(f.read_text())
if isinstance(data, list) and data and "name" in data[0]:
found[f.name.split("_")[0].lower()] = str(f)
except Exception: pass
return found
@staticmethod
def platform_health(platform: str) -> bool:
urls = {
"instagram": "https://www.instagram.com", "facebook": "https://www.facebook.com",
"twitter": "https://x.com", "tiktok": "https://www.tiktok.com",
"youtube": "https://www.youtube.com", "linkedin": "https://www.linkedin.com"
}
try:
r = subprocess.run(["curl", "-sI", "-o", "/dev/null", "-w", "%{http_code}",
urls.get(platform, urls["instagram"])],
capture_output=True, text=True, timeout=10)
return r.stdout.strip().startswith(("2", "3"))
except Exception:
return False
def _ensure_encryption_key() -> bool:
if os.getenv("ENCRYPTION_KEY"):
return True
try:
from cryptography.fernet import Fernet
key = Fernet.generate_key().decode()
with open(Path.cwd() / ".env", "a") as f:
f.write(f"\nENCRYPTION_KEY={key}\n")
os.environ["ENCRYPTION_KEY"] = key
return True
except ImportError:
return False
def boot() -> dict:
_ensure_encryption_key()
for d in ("sessions", "proofs", "proofs/optimized", "logs", "checkpoints", "content", "config/cookies", "cache"):
Path(d).mkdir(parents=True, exist_ok=True)
# Initialize basic configuration placeholders
totp_cfg = Path("config/totp.json")
if not totp_cfg.exists():
totp_cfg.parent.mkdir(parents=True, exist_ok=True)
totp_cfg.write_text(json.dumps({
"_comment": "Base32 TOTP secrets per platform/account",
"instagram": "", "tiktok": "", "twitter": "", "facebook": "", "youtube": ""
}, indent=2))
niche_cfg = Path("config/niches.json")
if not niche_cfg.exists():
niche_cfg.write_text(json.dumps({
"general": ["content", "tips", "growth", "community", "share"],
"ai_marketing": ["ai", "marketing", "automation", "growth", "content", "tools"]
}, indent=2))
growth_cfg = Path("config/growth_targets.json")
if not growth_cfg.exists():
growth_cfg.write_text(json.dumps({
p: {"accounts": [], "hashtags": [], "groups": [], "comment_templates": [],
"dm_templates": ["Hey {{name}}, loved your content on {{topic}}!"],
"engage_mode": "balanced"}
for p in ("instagram", "twitter", "facebook", "tiktok", "linkedin")
}, indent=2))
health = {p: AutoDiscovery.platform_health(p) for p in ("instagram", "twitter", "tiktok")}
from core.secrets_scanner import SecretsScanner
secrets = SecretsScanner().run()
encryption_ok = bool(os.getenv("ENCRYPTION_KEY"))
boot_warnings = _collect_boot_warnings(secrets, encryption_ok)
perm_errors = _harden_sensitive_permissions()
from core.ip_reputation import check_ip_risk
ip_risk = check_ip_risk()
if ip_risk.get("risk_level") in ("HIGH", "MEDIUM"):
boot_warnings.append(f"IP_RISK: {ip_risk.get('summary', 'datacenter/proxy flags')}")
ethics = _check_ethics_yaml()
if ethics.get("blocked"):
boot_warnings.append(f"ETHICS_BLOCKED: {ethics['blocked']}")
ops = ops_hardening_check() if os.getenv("OMNI_DEPLOY") == "production" else None
return {
"tools": AutoDiscovery.detect_tools(),
"cookies": AutoDiscovery.find_cookies(),
"ready": any(health.values()),
"health": health,
"secrets_scan": secrets,
"encryption_key_set": encryption_ok,
"boot_warnings": boot_warnings,
"permission_hardening": {"adjusted": True, "errors": perm_errors},
"ip_reputation": ip_risk,
"ethics": ethics,
"ops_hardening": ops,
}
def _harden_sensitive_permissions() -> list:
import platform
errors = []
if platform.system() not in ("Linux", "Darwin"):
return errors
targets = [
(Path(".env"), 0o600), (Path("config/totp.json"), 0o600),
(Path("config/cookies"), 0o700), (Path("sessions"), 0o700),
(Path("logs"), 0o700), (Path("cache"), 0o700),
]
for path, mode in targets:
if not path.exists():
continue
try:
os.chmod(path, mode)
except OSError as e:
errors.append(f"{path}: {e}")
return errors
def _check_ethics_yaml() -> dict:
path = Path("config/ethics.yaml")
if not path.exists():
return {"present": False, "passive_only": True, "blocked": None}
try:
import yaml
data = yaml.safe_load(path.read_text()) or {}
except Exception:
return {"present": True, "passive_only": True, "blocked": "parse_error"}
blocked = []
if data.get("passive_only") and data.get("allow_dm_automation"):
blocked.append("allow_dm_automation conflicts with passive_only")
if data.get("allow_platform_apis"):
blocked.append("allow_platform_apis violates Directive #3")
return {"present": True, "passive_only": data.get("passive_only", True), "blocked": blocked or None}
def _collect_boot_warnings(secrets: dict, encryption_ok: bool) -> list:
warnings = []
if secrets.get("verdict") == "FAIL":
warnings.append("HIGH_RISK: plaintext cookies or secrets detected — encrypt before publish")
if not encryption_ok:
warnings.append("ENCRYPTION_KEY auto-generated — set explicitly in production")
if secrets.get("verdict") == "WARN":
# ... [truncated — full in new-hybrid-4.7.md] ...
```
### Onboarding gate (all must PASS before live)
| Step | Pass condition |
|------|----------------|
| `boot()` | `ready: true`, secrets_scan ≠ FAIL |
| Encrypted session | `sessions/{account}.enc` exists |
| Routing | `account_routing[account]` set |
| Dry-run | `publish --dry-run` → rejected false |
| Virality (if on) | score ≥ 60 |
| Ethics | `passive_only: true` |
### ethics.yaml
```yaml
Boot checks `ethics.yaml`; `allow_platform_apis: true` → `boot_warnings` block.
## §18 Ops Hardening (VPS Deployment Appendix)
> Apply when `OMNI_DEPLOY=production`. Source: Mastering Linux Security + Linux Server Security summaries.
### 18.1 Service isolation
- Run as user `omni`, never root · `chmod 700 sessions/ config/cookies/ logs/ cache/`
### 18.2 SSH + firewall
```
### Stealth browser (Path A excerpt)
```python
# core/stealth_browser.py
import hashlib, random
from pathlib import Path
from typing import Tuple, Dict, Optional
class StealthBrowser:
USER_AGENTS = [
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 Chrome/134.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/134.0.0.0 Safari/537.36"
]
VIEWPORTS = [(1366, 768), (1440, 900), (1920, 1080)]
# Escaped triple quotes to compile properly inside outer string
STEALTH_INIT_SCRIPT = """
// Spoof Canvas Fingerprint
// Deterministic Canvas Seed Injected via add_init_script dynamically
// Spoof WebGL Renderer
const getParameter = WebGLRenderingContext.prototype.getParameter;
WebGLRenderingContext.prototype.getParameter = function(parameter) {
if (parameter === 37445) return 'Intel Inc.';
if (parameter === 37446) return 'Intel Iris OpenGL Engine';
return getParameter.apply(this, arguments);
};
// Hide automation flags
Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3]});
window.chrome = window.chrome || {};
"""
async def launch(self, account_id: str, platform: str, headless: bool = False,
proxy: Optional[Dict] = None) -> Tuple:
from playwright.async_api import async_playwright
# Deterministic seeding based on account + platform
seed = int(hashlib.md5(f"{account_id}_{platform}".encode()).hexdigest()[:8], 16)
rng = random.Random(seed)
ua = rng.choice(self.USER_AGENTS)
vp = rng.choice(self.VIEWPORTS)
p = await async_playwright().start()
args = {
"headless": headless,
"channel": "chrome",
"args": ["--disable-blink-features=AutomationControlled"]
}
if proxy:
args["proxy"] = proxy
profile = Path("sessions") / f"profile_{account_id}_{platform}"
profile.mkdir(exist_ok=True, parents=True)
ctx = await p.chromium.launch_persistent_context(
str(profile),
**args,
user_agent=ua,
viewport={"width": vp[0], "height": vp[1]},
timezone_id="Africa/Cairo",
locale="en-US"
)
page = ctx.pages[0] if ctx.pages else await ctx.new_page()
hw_cores = [2, 4, 6, 8][seed % 4]
dynamic_stealth = self.STEALTH_INIT_SCRIPT + f"""
(() => {{
Object.defineProperty(navigator, 'hardwareConcurrency', {{ get: () => {hw_cores} }});
const originalToDataURL = HTMLCanvasElement.prototype.toDataURL;
HTMLCanvasElement.prototype.toDataURL = function(type, ...args) {{
const ctx = this.getContext('2d');
if (ctx && this.width > 0) {{
const imgData = ctx.getImageData(0, 0, this.width, this.height);
for (let i = 0; i < imgData.data.length; i += {seed % 8 + 4}) {{
imgData.data[i] ^= {seed % 3 + 1};
}}
ctx.putImageData(imgData, 0, 0);
}}
return originalToDataURL.apply(this, [type, ...args]);
}};
}})();
"""
await page.add_init_script(dynamic_stealth)
return p, ctx, page, {"ua": ua, "vp": vp}
```
### Quick commands
```bash
# Path A
python self_check.py
python publish.py --dry-run --account instagram_growth01
python publish.py publish --account instagram_growth01
python publish.py grow --account instagram_growth01
python publish.py shadowban --account instagram_growth01
# Path C
python run.py login
python run.py once
python run.py serve
python run.py replies
python run.py analytics
```
---
## 6. Stealth, Humanization, Resilience
### Rules that prevent bans more than any proxy
1. Headed mode for social when possible
2. Persistent profile / storage_state
3. Bezier mouse + Fitts delays + Poisson burst/rest
4. Inject cookies before first navigation
5. Warm-up scroll/feed before publish
6. First 60 minutes: reply to comments (strong signal)
7. Never sudden engagement spikes on cold accounts
### Rate limiter (Path C)
```python
"""
Simple pacing so the automation behaves like a person clicking around
now and then, not a script firing actions back-to-back. This is about
not tripping generic bot-detection or rate limits on your own account.
There's no day-by-day ramp here -- just steady, configurable limits you
set once. If you find yourself wanting to schedule the limits themselves
to slowly increase over weeks, that's the growth-ramp pattern from the
uploaded file, and it's out of scope for this tool on purpose.
"""
import random
import time
from collections import deque
from datetime import datetime, timedelta
from typing import Optional
class RateLimiter:
def __init__(self, min_seconds_between_actions: int = 60,
max_per_day: Optional[int] = None):
self.min_gap = min_seconds_between_actions
self.max_per_day = max_per_day
self._last_action: Optional[float] = None
self._today_log: deque = deque()
def _prune(self) -> None:
cutoff = datetime.now() - timedelta(days=1)
while self._today_log and self._today_log[0] < cutoff:
self._today_log.popleft()
def can_act(self) -> bool:
self._prune()
if self.max_per_day is not None and len(self._today_log) >= self.max_per_day:
return False
if self._last_action is not None:
elapsed = time.monotonic() - self._last_action
if elapsed < self.min_gap:
return False
return True
def wait_if_needed(self) -> None:
if self._last_action is not None:
elapsed = time.monotonic() - self._last_action
remaining = self.min_gap - elapsed
if remaining > 0:
time.sleep(remaining + random.uniform(1, 5)) # a little jitter
def record_action(self) -> None:
self._last_action = time.monotonic()
self._today_log.append(datetime.now())
```
### Circuit breaker (Path A)
```python
# core/circuit_breaker.py
import time
from typing import Callable, Any
class CircuitBreaker:
def __init__(self, max_failures: int = 3, cooldown_period: int = 300):
self.max_failures = max_failures
self.cooldown_period = cooldown_period
self.failures = 0
self.state = "CLOSED"
self.last_failure_time = 0.0
def execute(self, action_fn: Callable, *args, **kwargs) -> Any:
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.cooldown_period:
self.state = "HALF-OPEN"
logger.info("Circuit breaker entering HALF-OPEN state, testing system health.")
else:
raise RuntimeError("Circuit breaker is OPEN. Execution blocked.")
try:
res = action_fn(*args, **kwargs)
if self.state == "HALF-OPEN":
self.state = "CLOSED"
self.failures = 0
logger.info("Circuit breaker reset to CLOSED after successful execution.")
return res
except Exception as e:
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.max_failures:
self.state = "OPEN"
logger.critical(f"Circuit breaker tripped to OPEN! Failures: {self.failures}")
raise e
async def execute_async(self, action_coro) -> Any:
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.cooldown_period:
self.state = "HALF-OPEN"
logger.info("Circuit breaker entering HALF-OPEN state (async).")
else:
raise RuntimeError("Circuit breaker is OPEN. Async execution blocked.")
try:
res = await action_coro
if self.state == "HALF-OPEN":
self.state = "CLOSED"
self.failures = 0
return res
except Exception as e:
self.failures += 1
self.last_failure_time = time.time()
if self.failures >= self.max_failures:
self.state = "OPEN"
raise e
```
### Rate governor (Path A)
```python
# core/rate_governor.py
import time, json
from datetime import datetime
from pathlib import Path
PLATFORM_LIMITS = {
"instagram": {"posts_per_day": 3, "min_gap_min": 120, "comments_per_day": 15,
"follows_per_day": 30, "likes_per_day": 60, "dms_per_day": 10},
"facebook": {"posts_per_day": 3, "min_gap_min": 90, "comments_per_day": 10,
"follows_per_day": 25, "likes_per_day": 50, "dms_per_day": 8},
"tiktok": {"posts_per_day": 2, "min_gap_min": 180, "comments_per_day": 8,
"follows_per_day": 20, "likes_per_day": 40, "dms_per_day": 5},
"twitter": {"posts_per_day": 6, "min_gap_min": 30, "comments_per_day": 20,
"follows_per_day": 40, "likes_per_day": 80, "dms_per_day": 15},
"youtube": {"posts_per_day": 1, "min_gap_min": 1440, "comments_per_day": 5,
"follows_per_day": 10, "likes_per_day": 20, "dms_per_day": 3},
}
class BioMimeticScheduler:
@staticmethod
def get_session_multiplier() -> float:
hour = datetime.now().hour
if 7 <= hour <= 9: return 1.5 # Morning commute
elif 12 <= hour <= 14: return 1.0 # Lunch break
elif 19 <= hour <= 23: return 0.8 # Evening peak
elif 1 <= hour <= 5: return 3.0 # Late night caution
return 1.0
class PredictiveRateGovernor:
def __init__(self, state_path: str = "cache/rate_state.json", history_path: str = "cache/rate_history.json"):
self.state_path = Path(state_path)
self.history_path = Path(history_path)
self.state_path.parent.mkdir(parents=True, exist_ok=True)
self.state = json.loads(self.state_path.read_text()) if self.state_path.exists() else {}
self.history = json.loads(self.history_path.read_text()) if self.history_path.exists() else []
self.gap_multiplier = 1.0
self._paused = {}
def _key(self, platform, account, action):
return f"{platform}:{account}:{action}"
def is_rate_limited(self, platform: str, action: str, today_count: int) -> bool:
limits = PLATFORM_LIMITS.get(platform, PLATFORM_LIMITS["instagram"])
key = f"{action}_per_day"
return today_count >= limits.get(key, 5)
def can_proceed(self, platform: str, account: str, action: str = "posts") -> bool:
if platform in self._paused and time.time() < self._paused[platform]:
return False
limits = PLATFORM_LIMITS.get(platform, PLATFORM_LIMITS["instagram"])
key = self._key(platform, account, action)
st = self.state.get(key, {"count": 0, "first_action": 0, "last_action": 0})
now = time.time()
# Reset count daily
if now - st.get("first_action", 0) > 86400:
st = {"count": 0, "first_action": now, "last_action": 0}
action_map = {"posts": "posts_per_day", "comments": "comments_per_day",
"follows": "follows_per_day", "likes": "likes_per_day", "dms": "dms_per_day"}
max_daily = limits.get(action_map.get(action, "posts_per_day"), 3)
if st["count"] >= max_daily:
return False
# Adapt gaps based on success history (stretch gaps if success rate drops)
recent = [h for h in self.history if h.get("platform") == platform
and h.get("account") == account and h.get("action") == action][-20:]
if len(recent) >= 10:
sr = sum(1 for h in recent if h.get("success")) / len(recent)
if sr < 0.6: self.gap_multiplier = min(3.0, self.gap_multiplier * 1.25)
else: self.gap_multiplier = max(1.0, self.gap_multiplier * 0.98)
bio_mult = BioMimeticScheduler.get_session_multiplier()
required_gap = limits.get("min_gap_min", 90) * 60 * self.gap_multiplier * bio_mult
if now - st.get("last_action", 0) < required_gap:
return False
return True
def record_action(self, platform: str, account: str, action: str = "posts"):
key = self._key(platform, account, action)
now = time.time()
st = self.state.get(key, {"count": 0, "first_action": now, "last_action": 0})
st["count"] += 1
st["last_action"] = now
self.state[key] = st
self.state_path.write_text(json.dumps(self.state))
def record_outcome(self, platform: str, account: str, action: str, success: bool):
if success:
self.record_action(platform, account, action)
self.history.append({"ts": time.time(), "platform": platform, "account": account,
"action": action, "success": success})
self.history = self.history[-500:]
self.history_path.write_text(json.dumps(self.history))
def pause_platform(self, platform: str, hours: int = 48):
self._paused[platform] = time.time() + hours * 3600
```
### Humanization (Bezier / smart_click excerpt)
```python
# (source block not found — open original file)
```
### Phased growth (accounts < 21 days)
| Action | Day 0–6 | Day 7–13 | Day 14–20 | Day 21+ |
|--------|---------|----------|-----------|---------|
| posts/day | 0 | 1 | 2 | 3 |
| follows/day | 0 | 5 | 15 | 30 |
| likes/day | 5 | 20 | 40 | 80 |
```python
# (source block not found — open original file)
```
### Shadowban canary
```python
# (source block not found — open original file)
```
### Incident tiers
| Code | Meaning | Action |
|------|---------|--------|
| 301 | Soft rate / friction | L1 reduce volume 50% 24h |
| 308 | Session drift | L2 re-export cookies same IP |
| 310 | Shadowban / lock | L3 halt 48h + quarantine |
```python
# (source block not found — open original file)
```
---
## 7. Pre-Publish Gates & Content Quality
```
TopicSanity → PrePublishScorer (3D) → AlgorithmMapper hooks
→ ViralityScorer ≥ 60 → ContentValidator → CaptionTransformer
```
```python
# (source block not found — open original file)
```
```python
# (source block not found — open original file)
```
```python
# (source block not found — open original file)
```
```python
# (source block not found — open original file)
```
```python
# (source block not found — open original file)
```
```python
# (source block not found — open original file)
```
```python
# (source block not found — open original file)
```
```python
# (source block not found — open original file)
```
> **Working note:** Never catch exceptions from the verification engine just to print “ok”. An unverified fix is no fix. INCONCLUSIVE blocks “done”.
---
## 7b. Extended Hybrid Modules (Extracted Best-Of Code)
> Curated working modules from `new-hybrid-4.7.md`. Truncated long files keep the public surface (classes, thresholds, gates).
### Cookie chain
```python
# core/cookie_chain.py
import time
from typing import Optional, Dict, List
class CookieChainManager:
"""Silent session refresh during active browser session."""
def __init__(self, account_id: str, refresh_interval_minutes: int = 45):
self.account_id = account_id
self.interval = refresh_interval_minutes * 60
self.last_refresh = time.time()
async def refresh_silently(self, page, refresh_url: str):
await page.evaluate(f"""
fetch('{refresh_url}', {{method: 'GET', credentials: 'include',
headers: {{'X-Refresh-Request': 'true'}}}}).catch(() => {{}});
""")
self.last_refresh = time.time()
def should_refresh(self) -> bool:
return time.time() - self.last_refresh > self.interval
```
### Secrets scanner
```python
# core/secrets_scanner.py
import json, re
from pathlib import Path
from typing import List
_SECRET_PATTERNS = [
(r"(?i)(api[_-]?key|secret|token|password)\s*[=:]\s*['\"]?[a-zA-Z0-9_\-]{8,}", "inline_secret"),
(r"sk-[a-zA-Z0-9]{20,}", "openai_key"),
(r"ghp_[a-zA-Z0-9]{20,}", "github_token"),
(r"AKIA[0-9A-Z]{16}", "aws_access_key"),
]
_SCAN_PATHS = [Path(".env"), Path("config"), Path("sessions"), Path("config/cookies")]
class SecretsScanner:
def __init__(self, root: Path = None):
self.root = root or Path.cwd()
def scan_file(self, path: Path) -> List[dict]:
findings = []
if not path.is_file() or path.stat().st_size > 500_000:
return findings
try:
text = path.read_text(encoding="utf-8", errors="ignore")
except Exception:
return findings
for pat, kind in _SECRET_PATTERNS:
for m in re.finditer(pat, text):
findings.append({"file": str(path.relative_to(self.root)), "kind": kind,
"line_hint": text[:m.start()].count("\n") + 1})
return findings
def scan_plaintext_cookies(self) -> List[dict]:
findings = []
for d in [Path("sessions"), Path("config/cookies")]:
if not d.is_dir():
continue
for f in d.glob("*.json"):
if "_expired" in f.name:
continue
try:
data = json.loads(f.read_text())
if isinstance(data, list) and data and "value" in data[0]:
findings.append({"file": str(f), "kind": "plaintext_cookie_store", "severity": "HIGH"})
except Exception:
pass
return findings
def run(self) -> dict:
findings = []
for base in _SCAN_PATHS:
p = self.root / base
if p.is_file():
findings.extend(self.scan_file(p))
elif p.is_dir():
for f in p.rglob("*"):
if f.is_file() and f.suffix in (".json", ".env", ".yaml", ".yml", ".txt", ".py"):
findings.extend(self.scan_file(f))
findings.extend(self.scan_plaintext_cookies())
high = [f for f in findings if f.get("severity") == "HIGH" or f.get("kind") == "plaintext_cookie_store"]
return {"total": len(findings), "high_risk": len(high), "findings": findings[:50],
"verdict": "FAIL" if high else ("WARN" if findings else "PASS")}
```
### Error handler recovery
```python
# core/error_handler.py
import sys, time, traceback
from enum import IntEnum
from core.automator_config import AutomatorConfig
from core.structured_logger import logger
class ErrorCode(IntEnum):
SUCCESS = 0
NETWORK_TIMEOUT = 201
NETWORK_CONNECTION_REFUSED = 202
NETWORK_PROXY_FAILED = 205
PLATFORM_RATE_LIMITED = 301
PLATFORM_AUTH_FAILED = 302
PLATFORM_CAPTCHA = 306
PLATFORM_BOT_DETECTED = 307
PLATFORM_SESSION_EXPIRED = 308
SHADOWBAN_DETECTED = 310
STATE_RESOURCE_EXHAUSTED = 404
UNKNOWN = 999
class ErrorHandler:
def __init__(self, config: AutomatorConfig = None):
self.config = config or AutomatorConfig()
self.consecutive_failures = 0
def classify_exception(self, exc: Exception) -> ErrorCode:
msg = str(exc).lower()
if "timeout" in msg or "timed out" in msg:
return ErrorCode.NETWORK_TIMEOUT
if "connection refused" in msg:
return ErrorCode.NETWORK_CONNECTION_REFUSED
if "proxy" in msg:
return ErrorCode.NETWORK_PROXY_FAILED
if "rate" in msg or "429" in msg:
return ErrorCode.PLATFORM_RATE_LIMITED
if "auth" in msg or "login" in msg:
return ErrorCode.PLATFORM_AUTH_FAILED
if "captcha" in msg or "challenge" in msg:
return ErrorCode.PLATFORM_CAPTCHA
if "bot" in msg or "webdriver" in msg:
return ErrorCode.PLATFORM_BOT_DETECTED
if "session" in msg or "cookie" in msg:
return ErrorCode.PLATFORM_SESSION_EXPIRED
if "shadowban" in msg:
return ErrorCode.SHADOWBAN_DETECTED
return ErrorCode.UNKNOWN
def handle(self, exc: Exception, account_id: str = "default") -> ErrorCode:
code = self.classify_exception(exc)
self.consecutive_failures += 1
logger.error(f"Error {code.name} ({code.value}) encountered: {exc}",
account_id=account_id, traceback=traceback.format_exc())
return code
def wait_cooldown(self, code: ErrorCode):
if code == ErrorCode.PLATFORM_RATE_LIMITED:
time.sleep(120)
elif code == ErrorCode.PLATFORM_BOT_DETECTED:
time.sleep(300)
def auto_recover(component_name: str = "generic"):
def decorator(func):
import asyncio
if asyncio.iscoroutinefunction(func):
async def async_wrapper(*args, **kwargs):
handler = ErrorHandler()
try:
return await func(*args, **kwargs)
except Exception as e:
code = handler.handle(e, component_name)
handler.wait_cooldown(code)
if handler.config.auto_recovery_enabled:
logger.info("Attempting auto-recovery execution...", component=component_name)
return await func(*args, **kwargs)
raise
return async_wrapper
else:
def sync_wrapper(*args, **kwargs):
handler = ErrorHandler()
try:
return func(*args, **kwargs)
except Exception as e:
code = handler.handle(e, component_name)
handler.wait_cooldown(code)
if handler.config.auto_recovery_enabled:
logger.info("Attempting auto-recovery execution...", component=component_name)
return func(*args, **kwargs)
raise
return sync_wrapper
return decorator
```
### Omni memory FTS
```python
# core/omni_memory.py
import asyncio, json, re, hashlib
from datetime import datetime
from pathlib import Path
from urllib.parse import quote_plus
import sqlite3
class OmniMemory:
def __init__(self, db_path: str = "cache/omni_memory.db"):
self.db_path = db_path
Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
self._init_db()
def _init_db(self):
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS research_fts USING fts5(
query, snippet, domain, scores, ts
)
""")
conn.commit()
conn.close()
def index_finding(self, query: str, snippet: str, domain: str, scores: str):
conn = sqlite3.connect(self.db_path)
conn.execute(
"INSERT INTO research_fts (query,snippet,domain,scores,ts) VALUES (?,?,?,?,?)",
(query, snippet[:500], domain, scores, datetime.now().isoformat())
)
conn.commit()
conn.close()
def prune_cache(self, days: int = 1):
"""Urgent Fix: Prune web research older than X days to avoid high cache."""
try:
conn = sqlite3.connect(self.db_path)
from datetime import timedelta, datetime
cutoff = (datetime.now() - timedelta(days=days)).isoformat()
conn.execute("DELETE FROM research_fts WHERE ts < ?", (cutoff,))
conn.execute("INSERT INTO research_fts(research_fts) VALUES('optimize')")
conn.commit()
conn.execute("VACUUM")
conn.close()
except: pass
def search_local(self, query: str, limit: int = 8) -> list:
conn = sqlite3.connect(self.db_path)
rows = conn.execute(
"SELECT query,snippet,domain,scores FROM research_fts WHERE research_fts MATCH ? LIMIT ?",
(query, limit)
).fetchall()
conn.close()
return [{"query": r[0], "snippet": r[1], "domain": r[2], "scores": r[3]} for r in rows]
# core/omni_search.py
from urllib.request import Request, urlopen
class OmniSearchEngine:
async def run_osint(self, query: str) -> list:
tasks = [
self._search_arxiv(query),
self._search_semantic_scholar(query)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
flat = []
for batch in results:
if isinstance(batch, list): flat.extend(batch)
return flat
async def _search_arxiv(self, q: str) -> list:
try:
url = f"http://export.arxiv.org/api/query?search_query=all:{quote_plus(q)}&max_results=3"
r = await asyncio.to_thread(urlopen, Request(url), 10)
text = r.read().decode()
out = []
for entry in re.findall(r"(.*?)", text, re.DOTALL):
link = re.search(r"(.*?)", entry)
title = re.search(r"(.*?)", entry, re.DOTALL)
summary = re.search(r"(.*?)", entry, re.DOTALL)
out.append({
"url": link.group(1).strip() if link else "",
"title": title.group(1).strip() if title else "",
"snippet": summary.group(1).strip()[:200] if summary else "",
"source": "arxiv"
})
return out
except Exception: return []
async def _search_semantic_scholar(self, q: str) -> list:
try:
url = f"https://api.semanticscholar.org/graph/v1/paper/search?query={quote_plus(q)}&limit=3&fields=title,url,abstract"
r = await asyncio.to_thread(urlopen, Request(url), 10)
data = json.loads(r.read().decode())
return [{
"url": p.get("url", ""), "title": p.get("title", ""),
"snippet": p.get("abstract", "")[:200] if p.get("abstract") else "",
"source": "semantic_scholar"
} for p in data.get("data", [])]
except Exception: return []
class ResearchPipelineV43:
# Integrates OmniMemory and OmniSearchEngine
@classmethod
def maybe_search(cls, content: dict, flags: dict = None) -> dict:
flags = flags or {}
if flags.get("factual") or flags.get("needs_search"): return {"run": True}
return {"run": False}
```
### Omni search OSINT
```python
# core/omni_search.py
from urllib.request import Request, urlopen
class OmniSearchEngine:
async def run_osint(self, query: str) -> list:
tasks = [
self._search_arxiv(query),
self._search_semantic_scholar(query)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
flat = []
for batch in results:
if isinstance(batch, list): flat.extend(batch)
return flat
async def _search_arxiv(self, q: str) -> list:
try:
url = f"http://export.arxiv.org/api/query?search_query=all:{quote_plus(q)}&max_results=3"
r = await asyncio.to_thread(urlopen, Request(url), 10)
text = r.read().decode()
out = []
for entry in re.findall(r"(.*?)", text, re.DOTALL):
link = re.search(r"(.*?)", entry)
title = re.search(r"(.*?)", entry, re.DOTALL)
summary = re.search(r"(.*?)", entry, re.DOTALL)
out.append({
"url": link.group(1).strip() if link else "",
"title": title.group(1).strip() if title else "",
"snippet": summary.group(1).strip()[:200] if summary else "",
"source": "arxiv"
})
return out
except Exception: return []
async def _search_semantic_scholar(self, q: str) -> list:
try:
url = f"https://api.semanticscholar.org/graph/v1/paper/search?query={quote_plus(q)}&limit=3&fields=title,url,abstract"
r = await asyncio.to_thread(urlopen, Request(url), 10)
data = json.loads(r.read().decode())
return [{
"url": p.get("url", ""), "title": p.get("title", ""),
"snippet": p.get("abstract", "")[:200] if p.get("abstract") else "",
"source": "semantic_scholar"
} for p in data.get("data", [])]
except Exception: return []
class ResearchPipelineV43:
# Integrates OmniMemory and OmniSearchEngine
@classmethod
def maybe_search(cls, content: dict, flags: dict = None) -> dict:
flags = flags or {}
if flags.get("factual") or flags.get("needs_search"): return {"run": True}
return {"run": False}
```
### Platform hacks
```python
# core/platform_hacks.py
from dataclasses import dataclass, field
from typing import Dict, List
@dataclass
class HackResult:
hack_name: str
platform: str
success: bool
expected_boost: float
risk_level: str
applied_content: Dict = field(default_factory=dict)
class PlatformHacksEngine:
def __init__(self):
self.hack_history: List[HackResult] = []
self.instagram_hacks = {
"carousel_boost": {"boost": 2.5, "risk": "low", "mod": {"format": "carousel", "slides": 10}},
"save_bait": {"boost": 3.0, "risk": "low", "mod": {"format": "carousel", "style": "cheat_sheet"}},
"reel_remix_boost": {"boost": 3.0, "risk": "low", "mod": {"remix": True}},
}
self.tiktok_hacks = {
"false_loop": {"boost": 3.0, "risk": "low", "mod": {"loop_type": "perfect"}},
"sound_hijacking": {"boost": 3.5, "risk": "low", "mod": {"trending_sound": True}},
}
self.twitter_hacks = {
"thread_funnel": {"boost": 3.0, "risk": "low", "mod": {"format": "thread", "tweet_count": 5}},
}
self.facebook_hacks = {
"group_posting": {"boost": 10.0, "risk": "medium", "mod": {"target": "group"}},
}
self.youtube_hacks = {
"thumbnail_optimization": {"boost": 3.0, "risk": "low", "mod": {"custom_thumbnail": True}},
}
self.recovery_steps = {
"instagram": ["Pause 48h", "Carousel only", "Post 07:30 Cairo"],
"tiktok": ["1 post/day max", "Photo mode"],
"twitter": ["No links 48h", "Threads only"]
}
def _hacks_for(self, platform: str) -> Dict:
return getattr(self, f"{platform}_hacks", self.instagram_hacks)
def apply_hack(self, platform: str, hack_name: str, content: Dict) -> HackResult:
hacks = self._hacks_for(platform)
if hack_name not in hacks:
return HackResult(hack_name, platform, False, 0, "unknown", content)
h = hacks[hack_name]
modified = {**content, **h["mod"]}
result = HackResult(hack_name, platform, True, h["boost"], h["risk"], modified)
self.hack_history.append(result)
return result
def get_hacks(self, platform: str) -> Dict:
return self._hacks_for(platform)
def get_recovery_steps(self, platform: str) -> List[str]:
base = ["Pause automation", "Mobile app only", "Manual engage", "Wait 48-72h"]
return base + self.recovery_steps.get(platform, [])
class AdaptiveHackEngine:
def __init__(self, engine: PlatformHacksEngine):
self.engine = engine
self.performance: Dict[str, float] = {}
def select_hack(self, platform: str, content: Dict) -> str:
hacks = self.engine.get_hacks(platform)
if not hacks: return "carousel_boost"
fmt = content.get("format", "post")
best, best_score = None, -999.0
for name, meta in hacks.items():
score = meta.get("boost", 1.0) + self.performance.get(name, 0.0) * 2
mod = meta.get("mod", {})
if fmt == mod.get("format"): score += 3
risk = {"low": 0, "medium": 1, "high": 2}.get(meta.get("risk", "low"), 1)
score -= risk
if score > best_score:
best_score, best = score, name
return best or next(iter(hacks))
def record_result(self, hack_name: str, engagement_delta: float):
self.performance[hack_name] = self.performance.get(hack_name, 0.0) * 0.7 + engagement_delta * 0.3
```
### Anti-detection / session health
```python
# core/anti_detection.py
import asyncio, base64, hashlib, hmac, logging, platform as platmod, struct, subprocess, time, re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple
log = logging.getLogger("anti_detection")
CHALLENGE_INDICATORS = {
"instagram": {
"url": [r"challenge", r"checkpoint", r"suspicious_login"],
"text": ["confirm your identity", "are you a robot", "أدخل الرمز"],
"sel": ['iframe[src*="captcha"]', '[data-testid="challenge"]'],
},
"facebook": {"url": [r"checkpoint"], "text": ["security check"], "sel": ['iframe[src*="captcha"]']},
"tiktok": {"url": [r"captcha", r"verify"], "text": ["slide to verify"], "sel": ['.captcha-verify-container']},
"twitter": {"url": [r"account/access", r"locked"], "text": ["verify your identity"], "sel": ['iframe[src*="captcha"]']},
}
@dataclass
class ChallengeResult:
detected: bool = False
challenge_type: str = ""
severity: str = "low"
requires_human: bool = False
evidence: List[str] = field(default_factory=list)
def alert_human_for_puzzle(platform_name: str, challenge_type: str):
msg = f"Solve {challenge_type} on {platform_name} NOW!"
try:
if platmod.system() == "Darwin":
subprocess.run(["osascript", "-e",
f'display notification "{msg}" with title "Bot Alert" sound name "Glass"'], check=False)
elif platmod.system() == "Linux":
subprocess.run(["notify-send", "Bot Alert", msg], check=False)
except Exception:
print(f"ALERT: {msg}")
class ChallengeDetector:
def __init__(self, page, platform: str):
self.page = page
self.platform = platform
self.sig = CHALLENGE_INDICATORS.get(platform, {})
async def scan(self) -> ChallengeResult:
r = ChallengeResult()
url = self.page.url
for p in self.sig.get("url", []):
if re.search(p, url, re.I):
r.detected = True
r.evidence.append(f"url:{p}")
try:
text = (await self.page.evaluate("() => document.body?.innerText?.toLowerCase() || ''"))
for p in self.sig.get("text", []):
if p.lower() in text:
r.detected = True
r.evidence.append(f"text:{p[:40]}")
except Exception: pass
for sel in self.sig.get("sel", []):
try:
if await self.page.locator(sel).count() > 0:
r.detected = True
r.evidence.append(f"sel:{sel}")
except Exception: pass
if r.detected:
ev = " ".join(r.evidence).lower()
if "captcha" in ev: r.challenge_type, r.severity, r.requires_human = "captcha", "critical", True
elif "locked" in ev: r.challenge_type, r.severity, r.requires_human = "account_locked", "critical", True
else: r.challenge_type, r.severity, r.requires_human = "checkpoint", "high", True
return r
class ZeroConfigCaptchaHandler:
@staticmethod
def generate_totp(secret_b32: str) -> str:
pad = "=" * ((8 - len(secret_b32) % 8) % 8)
key = base64.b32decode((secret_b32.upper() + pad).encode())
msg = struct.pack(">Q", int(time.time()) // 30)
h = hmac.new(key, msg, hashlib.sha1).digest()
o = h[-1] & 0x0F
code = struct.unpack(">I", h[o:o + 4])[0] & 0x7FFFFFFF
return str(code % 1000000).zfill(6)
@classmethod
async def try_inject_totp(cls, page, platform: str, human=None, account_id: str = "") -> bool:
path = Path("config/totp.json")
if not path.exists(): return False
data = json.loads(path.read_text())
secret = data.get(account_id) or data.get(platform)
if not secret: return False
code = cls.generate_totp(secret)
inputs = ['input[autocomplete="one-time-code"]', 'input[name="verificationCode"]',
'input[type="tel"]', 'input[inputmode="numeric"]']
for sel in inputs:
loc = page.locator(sel).first
if await loc.count() == 0: continue
if human: await human.burst_type(page, sel, code)
else: await loc.fill(code)
await page.keyboard.press("Enter")
await asyncio.sleep(3)
return True
return False
@classmethod
async def rescue_handoff(cls, page, platform: str, timeout_ms: int = 300000) -> bool:
await page.evaluate("document.body.style.border = '15px solid #ff0000'; document.body.style.boxShadow = 'inset 0 0 50px rgba(255,0,0,0.5)';")
await page.bring_to_front()
alert_human_for_puzzle(platform, "rescue_window")
try: await page.screenshot(path=f"proofs/rescue_{platform}_{int(time.time())}.png")
except Exception: pass
try:
# Wait for challenge URLs to disappear (changed nested triple-quotes to single double-quotes)
await page.wait_for_function(
"() => !/challenge|checkpoint|captcha|verify/i.test(window.location.href)",
timeout=timeout_ms
)
await page.evaluate("document.body.style.border=''; document.body.style.boxShadow='';")
# ... [truncated — full in new-hybrid-4.7.md] ...
```
### DOM self-healing locator
```python
# core/dom_locator.py
import json, re
from pathlib import Path
from typing import List
INTENT_HINTS = {
"create_post": {"texts": ["New post", "Create", "Post", "إنشاء"], "svg": ['path[d*="v16m-8"]', 'path[d*="M12 4v16"]']},
"next": {"texts": ["Next", "Continue", "التالي"], "svg": []},
"share": {"texts": ["Share", "Post", "Publish", "نشر", "مشاركة"], "svg": []},
"caption": {"texts": ["caption", "Write a caption", "وصف"], "svg": []},
"compose": {"texts": ["Tweet", "What's happening", "compose"], "svg": []},
}
class SelfHealingLocator:
CACHE_FILE = Path("cache/dom_healing.json")
@classmethod
def _load_cache(cls) -> dict:
if cls.CACHE_FILE.exists():
try: return json.loads(cls.CACHE_FILE.read_text())
except Exception: pass
return {}
@classmethod
def _save_cache(cls, key: str, selector: str):
cls.CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
cache = cls._load_cache()
cache[key] = selector
cls.CACHE_FILE.write_text(json.dumps(cache, indent=2))
@classmethod
async def _extract_selector(cls, loc) -> str:
# Escaped double quotes inside python string to avoid breaking string literal
return await loc.evaluate("el => { if (el.dataset && el.dataset.testid) return '[data-testid=\"' + el.dataset.testid + '\"]'; const al = el.getAttribute('aria-label'); if (al) return '[aria-label=\"' + al.replace(/'/g, '') + '\"]'; if (el.id) return '#' + el.id; return el.tagName.toLowerCase(); }")
@classmethod
async def locate(cls, page, intent: str, platform: str, primary: str = None,
fallback_texts: List[str] = None):
cache = cls._load_cache()
key = f"{platform}:{intent}"
hints = INTENT_HINTS.get(intent, {})
texts = fallback_texts or hints.get("texts", [])
# 1. Try cache & primary selector
for sel in [cache.get(key), primary]:
if sel:
loc = page.locator(sel).first
try:
if await loc.count() > 0:
await loc.wait_for(state="visible", timeout=3000)
return loc
except Exception: pass
# 2. Try semantic text search
if texts:
pattern = re.compile("|".join(re.escape(t) for t in texts), re.I)
for role in ("button", "link"):
loc = page.get_by_role(role, name=pattern).first
try:
await loc.wait_for(state="visible", timeout=4000)
healed = await cls._extract_selector(loc)
cls._save_cache(key, healed)
return loc
except Exception: pass
# 3. Fallback to SVG signature matching
for svg_pat in hints.get("svg", []):
loc = page.locator(f"svg {svg_pat}").first
if await loc.count() > 0:
parent = loc.locator("xpath=ancestor::*[@role='button' or self::button or self::a][1]")
target = parent if await parent.count() > 0 else loc
healed = await cls._extract_selector(target)
cls._save_cache(key, healed)
return target
# 4. Fallback to CLI AI Agent (avoiding ollama, using agent/openclaw)
try:
import subprocess
html_snippet = await page.evaluate("() => document.body.innerHTML.substring(0, 3000)")
prompt = f"Find the CSS selector for '{intent}' on {platform}. HTML snippet: {html_snippet}. Reply ONLY with the valid CSS selector string."
proc = subprocess.run(["agent", prompt], capture_output=True, text=True, timeout=20)
if proc.returncode == 0 and proc.stdout.strip():
sel = proc.stdout.strip()
ai_loc = page.locator(sel).first
if await ai_loc.count() > 0:
cls._save_cache(key, sel)
return ai_loc
except Exception: pass
raise RuntimeError(f"SelfHealingLocator failed to locate {platform}:{intent}")
```
### Media prep
```python
# core/media_prep.py
import json, shutil, subprocess
from pathlib import Path
class MediaPrepEngine:
@staticmethod
def ensure_vertical_916(media_path: str) -> str:
if not media_path or not Path(media_path).exists():
return media_path
if not shutil.which("ffmpeg") or not shutil.which("ffprobe"):
return media_path
out = Path("proofs/optimized") / f"{Path(media_path).stem}_916.mp4"
if out.exists() and out.stat().st_size > 1000:
return str(out)
try:
probe = subprocess.run(
["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", media_path],
capture_output=True, text=True, timeout=30)
if probe.returncode != 0: return media_path
info = json.loads(probe.stdout or "{}")
stream = next((s for s in info.get("streams", []) if s.get("codec_type") == "video"), None)
if not stream: return media_path
w, h = int(stream["width"]), int(stream["height"])
if w <= h: return media_path # already vertical
# Scaler complex logic: blur fill sides
out.parent.mkdir(parents=True, exist_ok=True)
fg = "[0:v]scale=1080:1920:force_original_aspect_ratio=decrease[fg]"
bg = "[0:v]scale=1080:1920:force_original_aspect_ratio=increase,crop=1080:1920,boxblur=25:5[bg]"
filt = f"{bg};{fg};[bg][fg]overlay=(W-w)/2:(H-h)/2"
enc = subprocess.run(
["ffmpeg", "-y", "-i", media_path, "-filter_complex", filt,
"-c:v", "libx264", "-preset", "fast", "-crf", "23", "-c:a", "aac", "-b:a", "128k", str(out)],
capture_output=True, timeout=180)
if enc.returncode == 0 and out.exists():
return str(out)
except Exception: pass
return media_path
```
### Topic sanity
```python
# core/topic_sanity.py
import json, re
from pathlib import Path
from typing import Dict, List
_DEFAULT_NICHES = {
"general": ["content", "tips", "growth", "community", "share"],
"ai_marketing": ["ai", "marketing", "automation", "growth", "content", "tools"],
"tech": ["technology", "software", "coding", "startup", "innovation"],
}
def topic_niche_check(content_topics: List[str], declared_niche: str,
caption: str = "", hashtags: List = None) -> Dict:
path = Path("config/niches.json")
if path.exists():
try:
data = json.loads(path.read_text())
keywords = [str(k).lower() for k in data.get(declared_niche, [])]
except Exception:
keywords = _DEFAULT_NICHES.get(declared_niche, _DEFAULT_NICHES["general"])
else:
keywords = _DEFAULT_NICHES.get(declared_niche, _DEFAULT_NICHES["general"])
topics = [t.lower().strip() for t in content_topics if t]
for word in re.findall(r"[a-zA-Z]{3,}", (caption or "").lower()): topics.append(word)
for tag in hashtags or []: topics.append(str(tag).lstrip("#").lower())
topics = list(set(topics))
overlap = len(set(topics) & set(keywords))
alignment = overlap / max(len(keywords), 1)
return {
"niche_alignment": round(alignment, 2),
"overlap_count": overlap,
"warn": overlap == 0,
"action": "review" if overlap == 0 else "proceed",
"declared_niche": declared_niche,
"matched": sorted(set(topics) & set(keywords))
}
```
### Pre-publish scorer
```python
# core/prepublish_scorer.py
import re
from typing import Dict, Optional
from core.algorithm_signal_scorer import AlgorithmSignalScorer
BANNED_PHRASES = ["buy followers", "guaranteed viral", "bot service", "hack instagram"]
class PrePublishScorer:
def __init__(self):
self.algo = AlgorithmSignalScorer()
def score(self, content: dict, virality: int = 0,
niche_check: Optional[dict] = None,
circuit_state: str = "CLOSED",
platform: str = "instagram",
account_id: str = "default") -> dict:
compliance = self._compliance(content, niche_check)
authenticity = self._authenticity(content)
operations = self._operations(content, circuit_state)
algo = self.algo.composite(platform, account_id,
content.get("caption") or content.get("text") or "")
dims = {"compliance": compliance, "authenticity": authenticity,
"operations": operations, "algorithm": algo["algorithm_score"] / 100.0}
overall = round(sum(dims.values()) / len(dims) * 100, 1)
rejected = (compliance < 0.5 or (niche_check and niche_check.get("warn"))
or virality < 60 or operations < 0.5
or algo["recommendation"] == "defer" or algo["algorithm_score"] < 55)
reason = None
if niche_check and niche_check.get("warn"): reason = "niche_mismatch"
elif compliance < 0.5: reason = "compliance_fail"
elif virality < 60: reason = "virality_low"
elif operations < 0.5: reason = "ops_unsafe"
elif algo["recommendation"] == "defer": reason = "algorithm_signal_low"
return {"overall": overall, "score": overall,
"dimensions": {k: round(v, 2) for k, v in dims.items()},
"virality": virality, "algorithm": algo,
"rejected": rejected, "reason": reason,
"weaknesses": [k for k, v in dims.items() if v < 0.6]}
def _compliance(self, content: dict, niche_check: Optional[dict]) -> float:
caption = (content.get("caption") or content.get("text") or "").lower()
tags = len(re.findall(r"#\w+", caption))
tag_ok = 1.0 if tags >= 11 else tags / 11
banned = sum(1 for p in BANNED_PHRASES if p in caption)
banned_ok = 1.0 if banned == 0 else max(0.0, 1.0 - banned * 0.5)
niche_ok = 1.0 if not niche_check or not niche_check.get("warn") else 0.0
return (tag_ok * 0.4 + banned_ok * 0.4 + niche_ok * 0.2)
def _authenticity(self, content: dict) -> float:
tool = content.get("_tool_used", "ai")
template_penalty = 0.6 if tool == "template" else 0.9
caption = content.get("caption") or content.get("text") or ""
length_ok = min(1.0, len(caption) / 80) if caption else 0.3
return (template_penalty * 0.6 + length_ok * 0.4)
def _operations(self, content: dict, circuit_state: str) -> float:
if circuit_state == "OPEN": return 0.0
if circuit_state == "HALF-OPEN": return 0.5
return 1.0
```
### Algorithm signal scorer
```python
# core/algorithm_signal_scorer.py
import json, re
from datetime import datetime
from pathlib import Path
from typing import Dict, List
class AlgorithmSignalScorer:
POPULAR_TAG_POOL = {
"instagram": ["love", "instagood", "photooftheday", "fashion", "beautiful"],
"twitter": ["trending", "news", "tech", "ai", "startup"],
"tiktok": ["fyp", "foryou", "viral", "trending", "tiktok"],
}
def __init__(self, analytics_path: str = "cache/engagement_analytics.json"):
self.path = Path(analytics_path)
self._cache = self._load()
def _load(self) -> dict:
if self.path.exists():
try: return json.loads(self.path.read_text())
except Exception: pass
return {}
def _extract_hashtags(self, caption: str) -> List[str]:
return [t.lower() for t in re.findall(r"#(\w+)", caption or "")]
def hashtag_mix_score(self, platform: str, caption: str) -> float:
tags = self._extract_hashtags(caption)
if not tags: return 0.0
popular = set(self.POPULAR_TAG_POOL.get(platform, []))
pop_count = sum(1 for t in tags if t in popular)
niche_count = len(tags) - pop_count
if pop_count == 0 and len(tags) >= 5: return 0.7
ratio = pop_count / max(len(tags), 1)
if 0.2 <= ratio <= 0.5 and niche_count >= 3: return 1.0
if 0.1 <= ratio <= 0.6: return 0.75
return 0.4
def cadence_score(self, platform: str, account_id: str) -> float:
key = f"{platform}:{account_id}"
history = self._cache.get(key, {}).get("post_timestamps", [])
if len(history) < 2: return 0.5
stamps = sorted(datetime.fromisoformat(t) for t in history[-14:])
gaps = [(stamps[i] - stamps[i-1]).total_seconds() / 3600 for i in range(1, len(stamps))]
avg_gap = sum(gaps) / len(gaps)
target = {"instagram": 12, "twitter": 4, "tiktok": 8}.get(platform, 12)
if 0.5 * target <= avg_gap <= 2.0 * target: return 1.0
if avg_gap <= 3 * target: return 0.6
return 0.3
def engagement_velocity_score(self, platform: str, account_id: str) -> float:
key = f"{platform}:{account_id}"
metrics = self._cache.get(key, {}).get("recent_posts", [])
if not metrics: return 0.5
scores = []
for m in metrics[-5:]:
reach = max(m.get("reach", 1), 1)
engagement_rate = (m.get("likes", 0) + m.get("comments", 0) * 2 + m.get("shares", 0) * 3) / reach
scores.append(min(1.0, engagement_rate * 10))
return sum(scores) / len(scores)
def timing_score(self, platform: str, scheduled_at: datetime = None) -> float:
from core.content_calendar import ContentCalendar
now = scheduled_at or datetime.now()
peaks = ContentCalendar.PEAK_HOURS.get(platform, [9, 12, 17])
if now.hour in peaks: return 1.0
for h in peaks:
if abs(now.hour - h) <= 1: return 0.8
return 0.4
def composite(self, platform: str, account_id: str, caption: str, scheduled_at: datetime = None) -> dict:
dims = {"hashtag_mix": self.hashtag_mix_score(platform, caption),
"cadence": self.cadence_score(platform, account_id),
"engagement_velocity": self.engagement_velocity_score(platform, account_id),
"timing": self.timing_score(platform, scheduled_at)}
weights = {"engagement_velocity": 0.35, "cadence": 0.25, "timing": 0.25, "hashtag_mix": 0.15}
overall = sum(dims[k] * weights[k] for k in dims)
return {"algorithm_score": round(overall * 100, 1),
"dimensions": {k: round(v, 2) for k, v in dims.items()},
"recommendation": "proceed" if overall >= 0.6 else "defer",
"defer_hours": 2 if overall < 0.6 else 0}
```
### Session health coordinator
```python
# core/session_health.py
from core.cookie_manager import EncryptedCookieManager
from core.structured_logger import logger
class SessionHealthCoordinator:
QUARANTINE_TRIGGERS = {"shadowban", "session_expired", "account_locked", "checkpoint", "rate_limited"}
def __init__(self):
self.cookies = EncryptedCookieManager()
def on_detection(self, account_id: str, platform: str, trigger: str) -> dict:
if trigger not in self.QUARANTINE_TRIGGERS:
return {"quarantined": False}
ok = self.cookies.quarantine(account_id) or self.cookies.quarantine(f"{platform}_{account_id}")
logger.warning("session_quarantined", account=account_id, platform=platform, trigger=trigger)
return {"quarantined": ok, "trigger": trigger, "account_id": account_id}
```
### Algorithm mapper
```python
# core/algorithm_mapper.py
import sqlite3, hashlib, random
from datetime import datetime
from typing import List
class AlgorithmMapper:
def __init__(self, platform: str, db_path: str = "algorithm_intel.db"):
self.platform = platform
self.conn = sqlite3.connect(db_path)
self.conn.execute("""CREATE TABLE IF NOT EXISTS content_performance (
post_id TEXT, content_hash TEXT, posted_at TEXT, hook_type TEXT,
video_length_sec INTEGER, hashtag_count INTEGER, posted_hour INTEGER,
initial_velocity_5min REAL, velocity_30min REAL, reach_1h INTEGER,
reach_24h INTEGER, algorithm_boost_detected INTEGER)""")
self.conn.commit()
def log_post(self, post_data: dict):
content_hash = hashlib.sha256(
f"{post_data.get('caption','')}{post_data.get('thumbnail','')}".encode()
).hexdigest()[:16]
self.conn.execute("""INSERT INTO content_performance VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", (
post_data["post_id"], content_hash, datetime.now().isoformat(),
post_data.get("hook_type", "unknown"), post_data.get("video_length", 0),
len(post_data.get("hashtags", [])), datetime.now().hour,
post_data.get("velocity_5min", 0), post_data.get("velocity_30min", 0),
post_data.get("reach_1h", 0), post_data.get("reach_24h", 0),
int(post_data.get("boost_detected", False)),
))
self.conn.commit()
def get_winning_hook_types(self) -> List[str]:
cur = self.conn.execute("""
SELECT hook_type, AVG(reach_24h) as avg_reach, COUNT(*) as n
FROM content_performance GROUP BY hook_type HAVING n >= 3
ORDER BY avg_reach DESC LIMIT 5""")
return [f"{row[0]} (avg reach: {row[1]:,.0f}, n={row[2]})" for row in cur.fetchall()]
class HookABTester:
TEMPLATES = [
"🚨 {topic}", "Most people get {topic} wrong",
"Here's how I {topic} in 60s", "🤯 {topic} (you won't believe #3)",
"{topic} — save this for later", "Stop doing {topic} like this",
]
def generate_variants(self, topic: str, count: int = 5) -> list:
templates = random.sample(self.TEMPLATES, min(count, len(self.TEMPLATES)))
return [t.format(topic=topic) for t in templates]
def select_winner(self, metrics: dict) -> str:
scored = [(h, d.get("views", 0) * 0.4 + d.get("retention", 0) * 30 + d.get("engagement", 0) * 30)
for h, d in metrics.items()]
scored.sort(key=lambda x: x[1], reverse=True)
return scored[0][0] if scored else ""
```
### Engagement warm-up hook
```python
# core/engagement_hook.py
import asyncio, random, logging
log = logging.getLogger("engagement_hook")
FEED_URLS = {
"instagram": "https://www.instagram.com/",
"twitter": "https://x.com/home",
"tiktok": "https://www.tiktok.com/",
"facebook": "https://www.facebook.com/",
"linkedin": "https://www.linkedin.com/feed/",
}
LIKE_SELECTORS = {
"instagram": ['svg[aria-label="Like"]', '[aria-label*="Like"]'],
"twitter": ['[data-testid="like"]', '[aria-label*="Like"]'],
"tiktok": ['[data-e2e="like-icon"]', '[aria-label*="Like"]'],
}
class EngagementHook:
def __init__(self, governor=None, human=None, challenge_detector=None):
self.gov = governor
self.human = human
self.ch = challenge_detector
async def warm_up(self, page, platform: str, account_id: str, likes: int = 2) -> dict:
if self.gov and not self.gov.can_proceed(platform, account_id, "likes"):
return {"status": "rate_limited", "phase": "warm_up"}
url = FEED_URLS.get(platform)
if not url:
return {"status": "skipped", "reason": "unsupported_platform"}
try:
await page.goto(url, wait_until="domcontentloaded", timeout=30000)
await asyncio.sleep(random.uniform(2, 4))
for _ in range(random.randint(2, 4)):
await page.evaluate(f"window.scrollBy(0, {random.randint(400, 900)})")
await asyncio.sleep(random.uniform(1.5, 3.5))
liked = 0
for sel in LIKE_SELECTORS.get(platform, LIKE_SELECTORS["instagram"]):
if liked >= likes:
break
loc = page.locator(sel)
count = await loc.count()
for i in range(min(count, likes - liked)):
if self.gov and not self.gov.can_proceed(platform, account_id, "likes"):
break
if self.human and self.ch:
ok = await self.human.smart_click(page, sel, self.ch, index=i)
else:
ok = await loc.nth(i).click(timeout=3000)
if ok:
liked += 1
if self.gov:
self.gov.record_action(platform, account_id, "likes")
await asyncio.sleep(random.uniform(8, 20))
return {"status": "ok", "phase": "warm_up", "likes": liked}
except Exception as e:
return {"status": "error", "phase": "warm_up", "error": str(e)}
```
### Humanization (Bezier smart_click)
```python
# core/humanization.py
import asyncio, logging, math, random, time
from typing import Dict, List, Tuple
log = logging.getLogger("humanization")
MIN_REACTION = 0.18
class BioMimeticMouse:
@staticmethod
def generate_bezier_path(start_x: int, start_y: int, end_x: int, end_y: int,
points: int = 25) -> List[Tuple[int, int]]:
cp1_x = start_x + random.randint(-80, 80)
cp1_y = start_y + random.randint(-80, 80)
cp2_x = end_x + random.randint(-50, 50)
cp2_y = end_y + random.randint(-50, 50)
path = []
for i in range(points + 1):
t = i / points
x = (1-t)**3 * start_x + 3*(1-t)**2*t * cp1_x + 3*(1-t)*t**2 * cp2_x + t**3 * end_x
y = (1-t)**3 * start_y + 3*(1-t)**2*t * cp1_y + 3*(1-t)*t**2 * cp2_y + t**3 * end_y
x += random.uniform(-1.5, 1.5)
y += random.uniform(-1.5, 1.5)
if random.random() < 0.05:
overshoot = random.randint(3, 8)
x += overshoot if random.random() > 0.5 else -overshoot
y += overshoot if random.random() > 0.5 else -overshoot
path.append((int(x), int(y)))
if random.random() < 0.08:
path.append(path[-1])
return path
@staticmethod
def calculate_path_quality(path: List[Tuple[int, int]], target_box: dict) -> float:
if len(path) < 10: return 0.2
tremors = sum(1 for i in range(1, len(path))
if abs(path[i][0] - path[i-1][0]) < 3 and abs(path[i][1] - path[i-1][1]) < 3)
tremor_ratio = tremors / len(path)
final = path[-1]
cx = target_box.get("x", 0) + target_box.get("width", 100) / 2
overshot = abs(final[0] - cx) > (target_box.get("width", 100) * 0.4)
variance = sum(abs(path[i][0] - path[i-1][0]) + abs(path[i][1] - path[i-1][1])
for i in range(1, min(10, len(path)))) / max(len(path) - 1, 1)
straight_penalty = 0.0 if variance > 5 else 0.3
score = 0.5 + (tremor_ratio * 0.3) + (0.15 if overshot else 0.0) - straight_penalty
return min(1.0, max(0.1, score))
class QuantumBehavioralFingerprint:
def __init__(self):
self._state = self._fresh()
def _fresh(self):
seed = time.time() + random.random()
r = random.Random(int(seed * 1000) % (2**32))
return {
'mouse_speed': 280 + r.randint(0, 520),
'typing_variance': 0.45 + r.random() * 1.1,
'hesitation_chance': 0.06 + r.random() * 0.18,
'scroll_momentum': 0.7 + r.random() * 0.55,
'pause_freq': 0.07 + r.random() * 0.25,
}
def get(self, action: str, platform: str = "") -> Dict:
s = self._state.copy()
if action in ("follow", "like"):
s['mouse_speed'] *= 0.85
s['hesitation_chance'] *= 1.4
elif action in ("comment", "reply", "dm"):
s['typing_variance'] *= 1.35
s['pause_freq'] *= 1.6
if platform == "tiktok": s['scroll_momentum'] *= 1.4
elif platform == "instagram": s['pause_freq'] *= 1.2
return s
def rotate(self):
if random.random() < 0.08:
self._state = self._fresh()
class ActionQualityScorer:
def __init__(self):
self.recent_scores: List[float] = []
def evaluate_typing_pattern(self, delays: List[float]) -> float:
if len(delays) < 3: return 0.5
mean_d = sum(delays) / len(delays)
variance = sum((d - mean_d) ** 2 for d in delays) / len(delays)
cv = (variance ** 0.5) / mean_d if mean_d > 0 else 0
score = min(1.0, 0.4 + cv * 0.6)
self.recent_scores.append(score)
self.recent_scores = self.recent_scores[-50:]
return score
def get_ai_feedback(self) -> str:
if not self.recent_scores: return ""
recent = self.recent_scores[-10:]
score = sum(recent) / len(recent)
if score < 0.5:
return f"SYSTEM NOTICE: Human-likeness scored {score:.2f}. Add more timing variance."
return ""
class HumanizationEngine:
def __init__(self):
self._action_count = 0
self.qbf = QuantumBehavioralFingerprint()
self.scorer = ActionQualityScorer()
async def warm_session(self, page, platform: str):
homes = {"instagram": "https://www.instagram.com/", "tiktok": "https://www.tiktok.com/",
"twitter": "https://x.com/home", "facebook": "https://www.facebook.com/",
"youtube": "https://www.youtube.com/"}
await page.goto(homes.get(platform, homes["instagram"]), wait_until="domcontentloaded")
await asyncio.sleep(random.uniform(3, 6))
behavior = self.qbf.get("scroll", platform)
for _ in range(random.randint(2, 4)):
momentum = behavior.get('scroll_momentum', 1.0)
await page.mouse.wheel(0, int(random.randint(200, 600) * momentum))
await asyncio.sleep(random.uniform(0.5, 1.5) * behavior.get('pause_freq', 1.0))
async def bio_mimetic_click(self, page, tx: int, ty: int, action: str = "click",
platform: str = "") -> List[Tuple[int, int]]:
# ... [truncated — full in new-hybrid-4.7.md] ...
```
### Shadowban detector
```python
# core/shadowban.py
import asyncio, logging, time
from pathlib import Path
from typing import Dict, Optional
log = logging.getLogger("shadowban")
CANARY_HASHTAGS = ["canarytest2026", "shadowbanprobe", "algotest"]
class ShadowbanDetector:
def __init__(self, hacks_engine=None):
self.hacks = hacks_engine
async def canary_visible(self, page, platform: str, tag: str) -> bool:
tag = tag.lstrip("#")
urls = {
"instagram": f"https://www.instagram.com/explore/tags/{tag}/",
"tiktok": f"https://www.tiktok.com/tag/{tag}",
"twitter": f"https://x.com/search?q=%23{tag}&f=live",
}
if platform not in urls: return True
await page.goto(urls[platform], wait_until="domcontentloaded")
await asyncio.sleep(3)
if platform == "instagram":
return await page.locator("article").count() > 0
content = await page.content()
return tag.lower() in content.lower()
async def check_shadowban(self, page, platform: str, username: str = "",
post_reach: Optional[int] = None) -> Dict:
methods = []
for tag in CANARY_HASHTAGS[:2]:
visible = await self.canary_visible(page, platform, tag)
methods.append({"method": "hashtag_search", "tag": tag, "visible": visible,
"confidence": 0.85 if not visible else 0.2})
if post_reach is not None and post_reach < 50:
methods.append({"method": "reach_drop", "reach": post_reach, "confidence": 0.7})
conf = sum(m["confidence"] for m in methods) / max(len(methods), 1)
shadowbanned = conf > 0.55
recovery = self.hacks.get_recovery_steps(platform) if self.hacks else []
if shadowbanned:
from core.session_health import SessionHealthCoordinator
SessionHealthCoordinator().on_detection(username or account_id, platform, "shadowban")
return {
"shadowbanned": shadowbanned,
"confidence": round(conf, 2),
"methods": methods,
"recovery_steps": recovery,
"recommendations": ["Pause automation 48h", "Manual engagement"] if shadowbanned else ["Normal"]
}
async def trigger_recovery(self, platform: str, governor=None):
log.warning(f"Shadowban recovery activated on {platform}")
if governor: governor.pause_platform(platform, hours=48)
```
### Phased growth
```python
# core/phased_growth.py
from datetime import datetime, timezone
class PhasedGrowth:
def __init__(self, account_created: str = None):
self.created = datetime.fromisoformat(account_created) if account_created else datetime.now(timezone.utc)
def age_days(self) -> int:
return (datetime.now(timezone.utc) - self.created).days
def can_action(self, action: str) -> bool:
d = self.age_days()
caps = {
"posts": {0: 0, 7: 1, 14: 2, 21: 3},
"follows": {0: 0, 7: 5, 14: 15, 21: 30},
"comments": {0: 0, 7: 3, 14: 10, 21: 15},
}
limits = caps.get(action, caps["posts"])
allowed = 0
for day, cap in sorted(limits.items()):
if d >= day: allowed = cap
return allowed > 0
```
### Virality scorer
```python
# agent/virality_scorer.py
class ViralityScorer:
CHECKS = [
("hook_first_3s", 20), ("virality_trigger", 20), ("engagement_cta", 15),
("format_match", 10), ("trending_element", 10), ("visual_quality", 10),
("caption_seo", 5), ("hashtag_strategy", 5), ("hashtag_min_11", 10),
("dm_share_hook", 5),
]
@classmethod
def score(cls, passed: list) -> int:
return sum(w for k, w in cls.CHECKS if k in passed)
@classmethod
def report(cls, passed: list) -> str:
s = cls.score(passed)
return f"Virality {s}/100 {'PUBLISH' if s >= 60 else 'REVISE'}"
```
### Content validator
```python
# core/content_validator.py
import re
class ContentValidator:
BANNED_WORDS = {"delve", "testament", "tapestry", "moreover", "leverage", "in conclusion"}
AI_PATTERNS = re.compile(r"\b(as an ai|i cannot|i don't have personal|it'?s important to note)\b", re.I)
GENERIC_TAGS = ["#content", "#tips", "#growth", "#learn", "#trending", "#share"]
@classmethod
def validate_caption(cls, caption: str, min_hashtags: int = 11) -> dict:
cleaned = caption.lower()
found = [w for w in cls.BANNED_WORDS if w in cleaned]
ai_flags = cls.AI_PATTERNS.findall(caption)
hashtags = re.findall(r"#\w+", caption)
return {
"valid": len(found) == 0 and len(ai_flags) == 0 and len(hashtags) >= min_hashtags,
"banned_found": found,
"ai_phrases_found": ai_flags,
"hashtag_count": len(hashtags)
}
@classmethod
def strip_banned(cls, caption: str) -> str:
for word in cls.BANNED_WORDS:
caption = re.sub(re.escape(word), "", caption, flags=re.IGNORECASE)
caption = cls.AI_PATTERNS.sub("", caption)
return re.sub(r"\s{2,}", " ", caption).strip()
@classmethod
def pad_hashtags(cls, content: dict, min_tags: int = 11) -> dict:
caption = content.get("caption", "")
tags = re.findall(r"#\w+", caption)
if len(tags) < min_tags:
extra = [t for t in cls.GENERIC_TAGS if t.lower() not in caption.lower()]
caption = caption + " " + " ".join(extra[:min_tags - len(tags)])
content["caption"] = caption.strip()
return content
```
### Browser automation core
```python
# core/browser_automation.py
import asyncio, time
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional
@dataclass
class ContentPackage:
platform: str
content_type: str = "post"
text: str = ""
caption: str = ""
hashtags: List[str] = field(default_factory=list)
media_path: Optional[str] = None
title: str = ""
@dataclass
class PublishResult:
success: bool
platform: str
proof_path: str = ""
post_url: str = ""
error: str = ""
class BrowserAutomation:
async def publish(self, content: ContentPackage, account_id: str,
page, human, challenge_handler) -> PublishResult:
publishers = {
"instagram": self._ig, "tiktok": self._tiktok, "twitter": self._twitter,
"facebook": self._facebook, "youtube": self._youtube,
}
pub = publishers.get(content.platform)
if not pub: return PublishResult(False, content.platform, error="unsupported platform")
try:
if content.media_path:
from core.media_prep import MediaPrepEngine
content.media_path = MediaPrepEngine.ensure_vertical_916(content.media_path)
res = await pub(page, content, human, challenge_handler)
Path("proofs").mkdir(exist_ok=True)
proof = f"proofs/{content.platform}_{int(time.time())}.png"
await page.screenshot(path=proof)
return PublishResult(res.get("success", False), content.platform, proof_path=proof, post_url=res.get("post_url", ""))
except Exception as e:
return PublishResult(False, content.platform, error=str(e)[:200])
async def _ig(self, page, c: ContentPackage, human, ch):
await human.warm_session(page, "instagram")
safe, _ = await ch.pre_action_check()
if not safe: return {"success": False}
await human.smart_click(page, 'svg[aria-label="New post"]', ch, intent="create_post",
platform="instagram", fallback_texts=["New post", "Create", "إنشاء"])
await asyncio.sleep(2)
if c.media_path:
fi = page.locator('input[type="file"]')
await fi.set_input_files(c.media_path)
await asyncio.sleep(4)
await human.smart_click(page, 'div[role="button"]:has-text("Next")', ch, intent="next",
platform="instagram", fallback_texts=["Next", "Continue"])
await asyncio.sleep(2)
cap = f"{c.caption or c.text}\n" + " ".join(f"#{h.lstrip('#')}" for h in c.hashtags)
box = 'div[aria-label="Write a caption…"], textarea[aria-label*="caption"]'
await human.smart_click(page, box, ch, intent="caption", platform="instagram",
fallback_texts=["caption", "Write a caption"])
await human.burst_type(page, box, cap)
await human.smart_click(page, 'div[role="button"]:has-text("Share")', ch, intent="share",
platform="instagram", fallback_texts=["Share", "Post"])
await asyncio.sleep(5)
return {"success": True, "post_url": page.url}
async def _tiktok(self, page, c, human, ch):
await page.goto("https://www.tiktok.com/upload", wait_until="domcontentloaded")
await asyncio.sleep(4)
if c.media_path:
await page.locator('input[type="file"]').set_input_files(c.media_path)
await asyncio.sleep(8)
cap = f"{c.caption or c.text} " + " ".join(f"#{h.lstrip('#')}" for h in c.hashtags)
box = 'div[contenteditable="true"]'
await human.smart_click(page, box, ch, intent="caption", platform="tiktok",
fallback_texts=["caption", "description"])
await human.burst_type(page, box, cap)
await human.smart_click(page, 'button:has-text("Post")', ch, intent="share",
platform="tiktok", fallback_texts=["Post", "Publish"])
return {"success": True}
async def _twitter(self, page, c, human, ch):
await page.goto("https://x.com/compose/tweet", wait_until="domcontentloaded")
await asyncio.sleep(3)
box = 'div[data-testid="tweetTextarea_0"]'
text = f"{c.caption or c.text} " + " ".join(f"#{h.lstrip('#')}" for h in c.hashtags)
await human.smart_click(page, box, ch, intent="compose", platform="twitter",
fallback_texts=["Tweet", "What's happening"])
await human.burst_type(page, box, text[:280])
await human.smart_click(page, 'button[data-testid="tweetButton"]', ch, intent="share",
platform="twitter", fallback_texts=["Post", "Tweet"])
return {"success": True}
async def _facebook(self, page, c, human, ch):
await human.warm_session(page, "facebook")
box = 'div[role="textbox"][contenteditable="true"]'
await human.smart_click(page, box, ch)
await human.burst_type(page, box, c.caption or c.text)
if c.media_path:
await page.locator('input[type="file"]').set_input_files(c.media_path)
await asyncio.sleep(4)
await human.smart_click(page, 'div[aria-label="Post"]', ch, intent="share",
platform="facebook", fallback_texts=["Post", "Publish"])
return {"success": True}
async def _youtube(self, page, c, human, ch):
await page.goto("https://www.youtube.com/upload", wait_until="domcontentloaded")
await asyncio.sleep(4)
if c.media_path:
await page.locator('input[type="file"]').set_input_files(c.media_path)
await asyncio.sleep(10)
if c.title:
await human.smart_click(page, 'input[aria-label="Add a title"]', ch)
await human.burst_type(page, 'input[aria-label="Add a title"]', c.title)
desc = c.caption or c.text
await human.smart_click(page, 'div[aria-label="Tell viewers about your video"]', ch)
await human.burst_type(page, 'div[aria-label="Tell viewers about your video"]', desc)
await human.smart_click(page, 'button:has-text("Publish")', ch, intent="share",
platform="youtube", fallback_texts=["Publish", "Upload"])
return {"success": True}
```
### Browser scraper
```python
# core/browser_scraper.py
import asyncio, json, re
from typing import Dict, List
class BrowserScraper:
@staticmethod
async def scrape_hashtag_posts(page, platform: str, tag: str, limit: int = 10) -> List[str]:
tag = tag.lstrip("#")
urls = {
"instagram": f"https://www.instagram.com/explore/tags/{tag}/",
"tiktok": f"https://www.tiktok.com/tag/{tag}",
"twitter": f"https://x.com/search?q=%23{tag}&f=live",
}
await page.goto(urls.get(platform, urls["instagram"]), wait_until="domcontentloaded")
await asyncio.sleep(4)
for _ in range(3):
await page.mouse.wheel(0, 800)
await asyncio.sleep(1.5)
# JS evaluation is wrapped cleanly in single quotes inside raw JS logic
links = await page.evaluate("() => { const out = new Set(); document.querySelectorAll('a[href]').forEach(a => { const h = a.href; if (h.includes('/p/') || h.includes('/reel/') || h.includes('/status/') || h.includes('/video/') || (h.includes('tiktok.com') && h.includes('/video'))) out.add(h.split('?')[0]); }); return [...out]; }")
return list(links)[:limit]
@staticmethod
async def scrape_profile_links(page, profile_url: str, limit: int = 10) -> List[str]:
await page.goto(profile_url, wait_until="domcontentloaded")
await asyncio.sleep(3)
links = await page.evaluate("() => { const out = new Set(); document.querySelectorAll('a[href]').forEach(a => { const h = a.href; if (h.includes('/p/') || h.includes('/reel/') || h.includes('/status/')) out.add(h.split('?')[0]); }); return [...out]; }")
return list(links)[:limit]
@staticmethod
async def scrape_comment_threads(page, post_url: str, limit: int = 15) -> List[Dict]:
await page.goto(post_url, wait_until="domcontentloaded")
await asyncio.sleep(3)
# Fixed nested quotes syntax by using single quotes inside Javascript string
raw = await page.evaluate("(lim) => { const items = []; document.querySelectorAll('[role=\'article\'], li, div[data-testid=\'tweet\']').forEach(el => { const t = (el.innerText || '').trim(); if (t.length > 5 && t.length < 500) items.push({text: t.slice(0, 300)}); }); return items.slice(0, lim); }", limit)
return raw or []
@staticmethod
async def scrape_profile_bio(page, profile_url: str) -> str:
try:
await page.goto(profile_url, wait_until="domcontentloaded")
await asyncio.sleep(2)
bio = await page.evaluate("() => { const el = document.querySelector('[role=\'main\'], header, .bio, [data-testid=\'UserProfile-bio\']'); return el ? el.innerText.substring(0, 300) : ''; }")
return bio or ""
except Exception:
return ""
```
### NSRE / v38 hybrid
```python
# core/v38_hybrid.py
import asyncio, base64, json, math, random, re, sqlite3, time
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Any
class _AsyncStore:
def __init__(self, name: str):
self.path = Path(f"cache/{name}.db")
self.path.parent.mkdir(parents=True, exist_ok=True)
async def execute(self, sql, params=()):
def _run():
conn = sqlite3.connect(str(self.path))
try:
conn.execute(sql, params)
conn.commit()
finally:
conn.close()
await asyncio.to_thread(_run)
async def fetch(self, sql, params=()):
def _run():
conn = sqlite3.connect(str(self.path))
try:
return conn.execute(sql, params).fetchall()
finally:
conn.close()
return await asyncio.to_thread(_run)
@dataclass
class Relationship:
partner_id: str
trust: float = 0.35
reciprocity: float = 1.0
interactions: int = 0
last_touch: str = ""
symbiosis: str = "commensalism"
niche_align: float = 0.5
benefit_to_partner: float = 0.0
benefit_to_agent: float = 0.0
class NeuroSymbioticRelationshipEngine:
def __init__(self):
self.db = _AsyncStore("relationships_v38")
async def init(self):
await self.db.execute("""
CREATE TABLE IF NOT EXISTS rels (
partner_id TEXT PRIMARY KEY, trust REAL, reciprocity REAL,
interactions INT, last_touch TEXT, symbiosis TEXT,
niche_align REAL, benefit_to_partner REAL, benefit_to_agent REAL
)
""")
async def get(self, pid: str, niche_align: float = 0.5) -> Relationship:
rows = await self.db.fetch("SELECT * FROM rels WHERE partner_id=?", (pid,))
if rows:
r = rows[0]
return Relationship(partner_id=r[0], trust=r[1], reciprocity=r[2],
interactions=r[3], last_touch=r[4], symbiosis=r[5],
niche_align=r[6], benefit_to_partner=r[7], benefit_to_agent=r[8])
rel = Relationship(pid, niche_align=niche_align)
await self._save(rel)
return rel
async def _save(self, rel: Relationship):
await self.db.execute("""
INSERT OR REPLACE INTO rels VALUES (?,?,?,?,?,?,?,?,?)
""", (rel.partner_id, rel.trust, rel.reciprocity, rel.interactions,
rel.last_touch, rel.symbiosis, rel.niche_align,
rel.benefit_to_partner, rel.benefit_to_agent))
async def record(self, pid: str, action: str, success: bool,
benefit_given: float = 1.0, benefit_received: float = 0.0):
rel = await self.get(pid)
rel.interactions += 1
rel.last_touch = datetime.now().isoformat()
rel.benefit_to_partner += benefit_given
rel.benefit_to_agent += benefit_received
rel.trust = min(1.0, rel.trust + 0.025) if success else max(0.1, rel.trust - 0.02)
if rel.benefit_to_partner > 0:
rel.reciprocity = rel.benefit_to_agent / rel.benefit_to_partner
if rel.trust > 0.65 and 0.75 < rel.reciprocity < 1.35: rel.symbiosis = "mutualism"
elif rel.trust < 0.3 or rel.reciprocity < 0.5 or rel.reciprocity > 2.0: rel.symbiosis = "parasitic_risk"
else: rel.symbiosis = "commensalism"
await self._save(rel)
async def should_engage(self, pid: str, min_trust: float = 0.25) -> bool:
rel = await self.get(pid)
return rel.symbiosis != "parasitic_risk" and rel.trust >= min_trust
async def calculate_niche_alignment(self, page, profile_url: str, ai_client, target_niche: str) -> float:
try:
from core.browser_scraper import BrowserScraper
bio = await BrowserScraper.scrape_profile_bio(page, profile_url)
if not bio: return 0.5
prompt = f"Score alignment (0.0 to 1.0) between Target Niche: '{target_niche}' and Bio: '{bio}'. Output JSON: {{'score': 0.8}}"
resp = await ai_client.generate(prompt, platform="general", format_type="json")
data = resp.parse_json()
return float(data.get("score", 0.5)) if data else 0.5
except Exception:
return 0.5
```
### Engagement engine
```python
# core/engagement_engine.py
import asyncio, random
from dataclasses import dataclass, field
from datetime import datetime
from core.browser_scraper import BrowserScraper
from core.engage_prompts import ENGAGE_PROMPTS
from core.v38_hybrid import NeuroSymbioticRelationshipEngine
from core.structured_logger import logger
@dataclass
class GrowthState:
follows: int = 0
comments: int = 0
likes: int = 0
dms: int = 0
replies: int = 0
day: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d"))
def reset_if_new_day(self):
today = datetime.now().strftime("%Y-%m-%d")
if self.day != today:
self.follows = self.comments = self.likes = self.dms = self.replies = 0
self.day = today
class EngagementEngine:
FOLLOW_SEL = {
"instagram": ['button:has-text("Follow")', 'button:has-text("متابعة")'],
"tiktok": ['button[data-e2e="follow-button"]'],
"twitter": ['button[data-testid$="-follow"]'],
}
LIKE_SEL = {
"instagram": ['svg[aria-label="Like"]'],
"tiktok": ['button[data-e2e="like-icon"]'],
"twitter": ['button[data-testid="like"]'],
}
COMMENT_SEL = 'textarea[placeholder*="comment" i], textarea[placeholder*="تعليق"], div[contenteditable="true"][role="textbox"]'
DM_SEL = {
"instagram": ['div[role="textbox"][contenteditable="true"]', 'textarea[placeholder*="Message"]'],
"twitter": ['div[data-testid="dmComposerTextInput"]'],
}
def __init__(self, page, platform: str, human, challenge_handler, governor, ai=None, rels_engine=None):
self.page = page
self.platform = platform
self.human = human
self.ch = challenge_handler
self.gov = governor
self.ai = ai
self.rels = rels_engine or NeuroSymbioticRelationshipEngine()
self.scraper = BrowserScraper()
self.state = GrowthState()
async def follow_profile(self, url: str, account: str, trust_check: bool = True) -> dict:
if not self.gov.can_proceed(self.platform, account, "follows"):
return {"status": "rate_limited", "action": "follow"}
pid = url.rstrip("/").split("/")[-1][:40]
if trust_check and self.rels:
if not await self.rels.should_engage(pid):
return {"status": "skipped_parasitic_risk", "partner": pid}
await self.page.goto(url, wait_until="domcontentloaded")
await asyncio.sleep(random.uniform(2, 5))
for sel in self.FOLLOW_SEL.get(self.platform, []):
if await self.human.smart_click(self.page, sel, self.ch, intent="share",
platform=self.platform, fallback_texts=["Follow"]):
self.state.follows += 1
self.gov.record_action(self.platform, account, "follows")
if self.rels:
await self.rels.record(pid, "follow", True, benefit_given=2.0)
return {"status": "ok", "action": "follow"}
return {"status": "not_found", "action": "follow"}
async def like_post(self, post_url: str, account: str) -> dict:
if not self.gov.can_proceed(self.platform, account, "likes"):
return {"status": "rate_limited", "action": "like"}
await self.page.goto(post_url, wait_until="domcontentloaded")
await asyncio.sleep(random.uniform(2, 4))
for sel in self.LIKE_SEL.get(self.platform, []):
if await self.human.smart_click(self.page, sel, self.ch):
self.state.likes += 1
self.gov.record_action(self.platform, account, "likes")
return {"status": "ok", "action": "like", "url": post_url}
return {"status": "not_found", "action": "like"}
async def comment_on_post(self, post_url: str, text: str, account: str) -> dict:
if not self.gov.can_proceed(self.platform, account, "comments"):
return {"status": "rate_limited", "action": "comment"}
await self.page.goto(post_url, wait_until="domcontentloaded")
await asyncio.sleep(random.uniform(2, 4))
if await self.human.smart_click(self.page, self.COMMENT_SEL, self.ch, intent="caption",
platform=self.platform, fallback_texts=["comment"]):
await self.human.burst_type(self.page, self.COMMENT_SEL, text[:500], action="comment", platform=self.platform)
await self.human.smart_click(self.page, 'button:has-text("Post")', self.ch, intent="share", platform=self.platform)
self.state.comments += 1
self.gov.record_action(self.platform, account, "comments")
return {"status": "ok", "action": "comment", "text": text[:80]}
return {"status": "not_found", "action": "comment"}
async def ai_comment(self, post_url: str, topic: str, account: str) -> dict:
text = f"Great point about {topic}!"
if self.ai:
prompt = ENGAGE_PROMPTS["comment"].format(platform=self.platform, topic=topic)
resp = await self.ai.generate(prompt, self.platform)
data = resp.parse_json()
if data and data.get("text"): text = data["text"]
return await self.comment_on_post(post_url, text, account)
async def respond_to_post(self, post_url: str, account: str, topic: str = "your post") -> dict:
threads = await self.scraper.scrape_comment_threads(self.page, post_url, limit=8)
results = []
for item in threads[:3]:
if not self.gov.can_proceed(self.platform, account, "comments"): break
text = "Thanks for engaging!"
if self.ai:
ctx = f"Comment: {item.get('text','')[:200]}"
resp = await self.ai.generate(ENGAGE_PROMPTS["reply"].format(platform=self.platform) + "\n" + ctx, self.platform)
data = resp.parse_json()
if data and data.get("text"): text = data["text"]
r = await self.comment_on_post(post_url, text, account)
# ... [truncated — full in new-hybrid-4.7.md] ...
```
### Strategy engines
```python
# core/strategy_engines.py
import asyncio, base64, json, random, re, time
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, List, Optional
@dataclass
class PlatformVariant:
platform: str
caption: str
hashtags: List[str]
hook: str
title: Optional[str] = None
media_path: Optional[str] = None
cross_link_cta: Optional[str] = None
class SyndicateEngine:
CONSTRAINTS = {
"instagram": {"max_len": 2200, "tags_max": 30, "style": "visual_first_hook"},
"twitter": {"max_len": 280, "tags_max": 3, "style": "thread_hook"},
"tiktok": {"max_len": 2200, "tags_max": 5, "style": "script_3s_hook"},
"facebook": {"max_len": 5000, "tags_max": 10, "style": "discussion_question"},
"youtube": {"max_len": 5000, "tags_max": 15, "style": "seo_description"},
}
def __init__(self, ai): self.ai = ai
async def transmute(self, seed_topic: str, seed_media: Optional[str] = None, niche: str = "general") -> Dict[str, PlatformVariant]:
master = await self.ai.generate(f"Master narrative about '{seed_topic}' for niche '{niche}'. Include: hook, 3 key points, CTA, 20 hashtags. Output JSON.", "general", "json")
master_data = master.parse_json() or {}
variants = {}
for plat, rules in self.CONSTRAINTS.items():
prompt = (f"Adapt this master content for {plat}. Rules: max {rules['max_len']} chars, ≤{rules['tags_max']} hashtags, style: {rules['style']}. Include subtle cross-reference hint. Master: {json.dumps(master_data)}. Output JSON: caption, hashtags, hook, title, cross_link_cta.")
resp = await self.ai.generate(prompt, plat, "json")
d = resp.parse_json() or {}
variants[plat] = PlatformVariant(plat, d.get("caption", "")[:rules["max_len"]], d.get("hashtags", [])[:rules["tags_max"]], d.get("hook", ""), d.get("title"), seed_media, d.get("cross_link_cta"))
return variants
def inject_cross_links(self, variants: Dict[str, PlatformVariant], published_urls: Dict[str, str]) -> Dict[str, PlatformVariant]:
for plat, v in variants.items():
cta = [f"{v.cross_link_cta} {url}" for target, url in published_urls.items() if target != plat and v.cross_link_cta]
if cta: v.caption += "\n\n" + "\n".join(cta)
return variants
class PulseScanner:
TREND_PAGES = {
"instagram": "https://www.instagram.com/explore/",
"tiktok": "https://www.tiktok.com/foryou",
"twitter": "https://x.com/explore/tabs/trending"
}
async def scan(self, page, platform: str, niche_keywords: List[str]) -> List[Dict]:
url = self.TREND_PAGES.get(platform, self.TREND_PAGES["instagram"])
await page.goto(url, wait_until="domcontentloaded")
await asyncio.sleep(3)
# Changed nested triple-quotes to single double-quotes
tokens = await page.evaluate("() => { const out = new Set(); document.querySelectorAll('span, div, a, h2, h3').forEach(el => { const t = el.innerText?.trim(); if (t && t.length > 2 && t.length < 80) out.add(t); }); return [...out]; }")
trends = []
for token in tokens:
tags = re.findall(r'#\w+', token)
if tags and any(k.lower() in token.lower() for k in niche_keywords):
trends.append({"type": "hashtag", "value": tags[0], "heat": len(token)})
elif any(k.lower() in token.lower() for k in niche_keywords):
trends.append({"type": "topic", "value": token[:60], "heat": 1})
seen, out = set(), []
for t in sorted(trends, key=lambda x: -x["heat"]):
if t["value"] not in seen:
seen.add(t["value"])
out.append(t)
return out[:8]
async def rider_prompt(self, trend: Dict, niche: str, platform: str, ai) -> Dict:
resp = await ai.generate(f"Trending NOW: {trend['value']}. Create a {platform} post bridging this trend to '{niche}'. Output JSON: caption, hashtags, hook.", platform, "json")
return resp.parse_json() or {}
class PhoenixProtocol:
STRATEGIES = ["shorten_hook", "swap_cta", "replace_hashtags", "add_urgency", "storytelling_open"]
def __init__(self, ai):
self.ai = ai
self.morph_log = {}
def _fp(self, content: Dict) -> str:
import hashlib
return hashlib.sha256(json.dumps(content, sort_keys=True).encode()).hexdigest()[:12]
async def morph(self, content: Dict, platform: str, failure_reason: str) -> Optional[Dict]:
fp = self._fp(content)
if self.morph_log.get(fp, 0) >= 3: return None
strategy = random.choice(self.STRATEGIES)
resp = await self.ai.generate(f"Content failed to publish. Reason: {failure_reason}. Morph strategy: {strategy}. Original: {json.dumps(content)}. Output revised JSON for {platform}.", platform, "json")
new_data = resp.parse_json()
if new_data:
self.morph_log[fp] = self.morph_log.get(fp, 0) + 1
new_data["_morph"] = {"count": self.morph_log[fp], "strategy": strategy}
return new_data
class MirrorProtocol:
def __init__(self, ai): self.ai = ai
async def verify_publish(self, page, platform: str, expected_topic: str) -> Dict:
Path("proofs").mkdir(parents=True, exist_ok=True)
proof_path = f"proofs/mirror_{platform}_{int(time.time())}.png"
await page.screenshot(path=proof_path, full_page=False)
with open(proof_path, "rb") as f: b64 = base64.b64encode(f.read()).decode()
resp = await self.ai.generate_from_image(f"Verify this {platform} post screenshot. Does it contain content about '{expected_topic}'? Output JSON: {{'valid':bool, 'issues':[...], 'confidence':0-1}}", b64, platform)
data = resp.parse_json() or {}
return {"valid": data.get("valid", True), "issues": data.get("issues", []), "confidence": data.get("confidence", 0.5), "proof": proof_path}
class DMWeaver:
async def weave(self, page, profile_url: str, platform: str, ai) -> str:
from core.browser_scraper import BrowserScraper
try:
posts = await BrowserScraper.scrape_profile_links(page, profile_url, limit=3)
snippets = []
for post in posts[:3]:
await page.goto(post, wait_until="domcontentloaded")
await asyncio.sleep(1)
text = await page.evaluate("() => document.body.innerText") or ""
snippets.append(text[:180])
name = profile_url.rstrip("/").split("/")[-1]
resp = await ai.generate(f"Write a personalized {platform} DM to '{name}'. Their recent themes: {json.dumps(snippets)}. Reference something specific. Output JSON: {{'text':'...'}}", platform, "json")
data = resp.parse_json()
return data.get("text", f"Hey {name}, loved your post!") if data else f"Hey {name}, loved your post!"
# ... [truncated — full in new-hybrid-4.7.md] ...
```
### Proxy rotator
```python
# core/proxy_rotator.py
import time
from typing import Optional
from urllib.request import urlopen, Request
class ProxyRotator:
def __init__(self, proxy_list: list[dict] = None):
self.proxies = proxy_list or []
self._health = {}
self._blacklist = set()
def add_proxy(self, server: str, username: str = "", password: str = ""):
proxy = {"server": server}
if username:
proxy["username"] = username
proxy["password"] = password
self.proxies.append(proxy)
def health_check(self, proxy: dict, test_url: str = "https://httpbin.org/ip", timeout: int = 5) -> bool:
server = proxy["server"]
try:
import urllib.request
handler = urllib.request.ProxyHandler({"https": server, "http": server})
opener = urllib.request.build_opener(handler)
start = time.time()
opener.open(test_url, timeout=timeout)
latency = (time.time() - start) * 1000
self._health[server] = {"alive": True, "latency_ms": latency, "last_check": time.time()}
self._blacklist.discard(server)
return True
except Exception:
self._health[server] = {"alive": False, "latency_ms": -1, "last_check": time.time()}
self._blacklist.add(server)
return False
def get_best_proxy(self) -> Optional[dict]:
candidates = [p for p in self.proxies if p["server"] not in self._blacklist]
if not candidates:
if self.proxies:
self.health_check(self.proxies[0])
if self.proxies[0]["server"] not in self._blacklist:
return self.proxies[0]
return None
candidates.sort(key=lambda p: self._health.get(p["server"], {}).get("latency_ms", 9999))
return candidates[0]
def mark_failed(self, proxy: dict):
self._blacklist.add(proxy["server"])
```
### IP reputation
```python
# core/ip_reputation.py
import json
from urllib.request import urlopen, Request
def check_ip_risk() -> dict:
"""ponytail: ip-api free tier; upgrade to scamalytics if limits hit."""
try:
req = Request("http://ip-api.com/json/?fields=status,country,hosting,proxy,query",
headers={"User-Agent": "omni-boot/4.6"})
data = json.loads(urlopen(req, timeout=8).read().decode())
if data.get("status") != "success":
return {"risk_level": "UNKNOWN", "summary": "lookup failed"}
hosting, proxy = data.get("hosting", False), data.get("proxy", False)
risk = "HIGH" if hosting and proxy else ("MEDIUM" if hosting else "LOW")
return {"risk_level": risk, "ip": data.get("query"), "hosting": hosting,
"proxy": proxy, "summary": f"hosting={hosting} proxy={proxy}"}
except Exception as e:
return {"risk_level": "UNKNOWN", "summary": str(e)}
class ZeroCostProxyManager:
@staticmethod
def get_proxy(strategy: str = "direct", boot_warnings: list = None):
if strategy == "direct" and not any("IP_RISK" in w for w in (boot_warnings or [])):
return None
if strategy == "warp" or any("IP_RISK" in w for w in (boot_warnings or [])):
return {"server": "socks5://127.0.0.1:8080"}
strategies = {"ssh": "socks5://127.0.0.1:1080", "tor": "socks5://127.0.0.1:9050"}
return {"server": strategies[strategy]} if strategy in strategies else None
```
### Content calendar
```python
# core/content_calendar.py
import json, random
from datetime import datetime, timedelta
from pathlib import Path
class ContentCalendar:
PEAK_HOURS = {
"instagram": [7, 8, 11, 12, 13, 17, 18, 19],
"twitter": [8, 9, 12, 13, 17, 18],
"tiktok": [7, 9, 12, 15, 19, 21, 22],
"linkedin": [7, 8, 10, 12, 17],
"facebook": [9, 11, 12, 13, 16, 17],
"youtube": [12, 14, 15, 16, 17],
}
DAILY_CADENCE = {
"instagram": 2, "twitter": 5, "tiktok": 3,
"linkedin": 1, "facebook": 2, "youtube": 1,
}
def __init__(self, schedule_path: str = "cache/content_schedule.json"):
self.path = Path(schedule_path)
self.path.parent.mkdir(parents=True, exist_ok=True)
self.schedule = self._load()
def _load(self) -> list:
if self.path.exists():
try: return json.loads(self.path.read_text())
except Exception: pass
return []
def _save(self):
self.path.write_text(json.dumps(self.schedule, indent=2, default=str))
def get_audience_active_hours(self, platform: str, account_id: str = "default") -> list:
path = Path("cache/engagement_analytics.json")
if path.exists():
try:
data = json.loads(path.read_text())
hours = data.get(f"{platform}:{account_id}", {}).get("audience_active_hours")
if hours: return hours
except Exception: pass
return self.PEAK_HOURS.get(platform, [9, 12, 17])
def get_next_slot(self, platform: str, after: datetime = None, account_id: str = "default") -> datetime:
now = after or datetime.now()
peaks = self.get_audience_active_hours(platform, account_id)
for hour in sorted(peaks):
candidate = now.replace(hour=hour, minute=random.randint(0, 29), second=0)
if candidate > now: return candidate
tomorrow = now + timedelta(days=1)
first_peak = sorted(peaks)[0]
return tomorrow.replace(hour=first_peak, minute=random.randint(0, 29), second=0)
def queue_post(self, platform: str, caption: str, media_path: str = None, account_id: str = "default"):
slot = self.get_next_slot(platform)
entry = {
"platform": platform, "account_id": account_id,
"caption": caption, "media": media_path,
"scheduled_at": slot.isoformat(), "status": "queued"
}
self.schedule.append(entry)
self._save()
return entry
def get_pending(self) -> list:
now = datetime.now().isoformat()
return [e for e in self.schedule if e["status"] == "queued" and e["scheduled_at"] <= now]
def mark_done(self, index: int, success: bool = True):
if 0 <= index < len(self.schedule):
self.schedule[index]["status"] = "done" if success else "failed"
self._save()
```
### Multi-account orchestrator
```python
# core/orchestrator.py
import asyncio
from typing import Dict, List
from core.dispatcher import HybridDispatcher
from core.circuit_breaker import CircuitBreaker
from core.rate_governor import PredictiveRateGovernor
from core.v38_hybrid import NeuroSymbioticRelationshipEngine
class MultiAccountOrchestrator:
"""Runs actions across multiple accounts with isolation and rate limiting.
Each account gets its own circuit breaker and session.
ponytail: sequential account iteration, O(n) on account count.
Upgrade path: asyncio.gather with semaphore for parallel execution.
"""
def __init__(self, accounts: List[Dict]):
# accounts: [{"id": "acc1", "platform": "instagram"}, ...]
self.accounts = accounts
self.breakers = {a["id"]: CircuitBreaker() for a in accounts}
self.governor = PredictiveRateGovernor()
self.db = NeuroSymbioticRelationshipEngine()
self._sessions = {}
async def authenticate_all(self, headless: bool = False):
"""Opens sessions for all accounts, skipping those with open circuit breakers."""
for account in self.accounts:
aid = account["id"]
if self.breakers[aid].state == "OPEN":
continue
try:
dispatcher = HybridDispatcher(aid, account["platform"], headless)
session = await dispatcher.get_session()
self._sessions[aid] = {"session": session, "dispatcher": dispatcher}
except Exception as e:
self.breakers[aid].failures = getattr(self.breakers[aid], "failures", 0) + 1
async def execute_action(self, account_id: str, action_type: str, action_coro):
"""Wraps an action with circuit breaking, rate limiting, and logging."""
if self.breakers[account_id].state == "OPEN":
return {"status": "circuit_open", "account": account_id}
account = next((a for a in self.accounts if a["id"] == account_id), None)
if not account:
return {"status": "unknown_account"}
platform = account["platform"]
if not self.governor.can_proceed(platform, account_id, action_type):
return {"status": "rate_limited"}
import time
start = time.time()
try:
result = await self.breakers[account_id].execute_async(action_coro)
duration = time.time() - start
self.governor.record_outcome(platform, account_id, action_type, success=True)
return {"status": "ok", "result": result, "duration": duration}
except Exception as e:
duration = time.time() - start
self.governor.record_outcome(platform, account_id, action_type, success=False)
return {"status": "error", "error": str(e)}
async def close_all(self):
for aid, data in self._sessions.items():
try:
await data["dispatcher"].close(data["session"])
except Exception: pass
self._sessions.clear()
```
### Caption transformer
```python
# core/caption_transformer.py
import re, random
class CaptionTransformer:
PLATFORM_LIMITS = {
"instagram": {"max_chars": 2200, "max_hashtags": 30, "optimal_hashtags": 15},
"twitter": {"max_chars": 280, "max_hashtags": 3, "optimal_hashtags": 2},
"tiktok": {"max_chars": 2200, "max_hashtags": 5, "optimal_hashtags": 4},
"linkedin": {"max_chars": 3000, "max_hashtags": 5, "optimal_hashtags": 3},
"facebook": {"max_chars": 63206, "max_hashtags": 5, "optimal_hashtags": 3},
"youtube": {"max_chars": 5000, "max_hashtags": 15, "optimal_hashtags": 8},
}
@classmethod
def apply_evasion_filters(cls, text: str) -> str:
"""Advanced Evasion: Spam-Filter Language Swapper to bypass ML toxicity/crypto filters."""
banned = {r'\bcrypto\b': 'cr.ypto', r'\bbuy\b': 'b.uy', r'\bsell\b': 's.ell',
r'\bfree\b': 'fr.ee', r'\blink in bio\b': 'l.ink in b.io',
r'\bsubscribe\b': 's.ubscribe', r'\bmoney\b': 'm.oney', r'\binvest\b': 'i.nvest'}
for pat, rep in banned.items():
import re
text = re.sub(pat, rep, text, flags=re.IGNORECASE)
return text
@classmethod
def transform(cls, caption: str, platform: str) -> str:
limits = cls.PLATFORM_LIMITS.get(platform, cls.PLATFORM_LIMITS["instagram"])
hashtags = re.findall(r"#\w+", caption)
body = re.sub(r"\s*#\w+", "", caption).strip()
body = cls.apply_evasion_filters(body)
max_body = limits["max_chars"] - (limits["optimal_hashtags"] * 15)
if len(body) > max_body:
body = body[:max_body].rsplit(" ", 1)[0] + "..."
selected = hashtags[:limits["optimal_hashtags"]]
# Fixed newline formatting with double-escapes to prevent f-string crashes in output file
if platform == "twitter": return f"{body} {' '.join(selected)}"
elif platform == "linkedin": return f"{body}\n\n{' '.join(selected)}"
else: return f"{body}\n\n{' '.join(selected)}"
@classmethod
def add_call_to_action(cls, caption: str, platform: str) -> str:
ctas = {
"instagram": ["💬 What do you think?", "👇 Drop your thoughts below!"],
"twitter": ["💬 Reply with your take", "❤️ Like if this resonates"],
"linkedin": ["💡 Agree? Share your perspective.", "💬 What's your experience?"],
}
options = ctas.get(platform, ctas["instagram"])
# Fixed newline formatting with double-escapes
return f"{caption}\n\n{random.choice(options)}"
```
### Analytics
```python
# core/analytics.py
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path
class EngagementAnalytics:
def __init__(self, db_path: str = "cache/relationships_v38.db"):
self.path = Path(db_path)
def get_daily_summary(self, platform: str, account_id: str, days: int = 7) -> list[dict]:
cutoff = (datetime.now() - timedelta(days=days)).isoformat()
if not self.path.exists(): return []
with sqlite3.connect(self.path) as conn:
conn.row_factory = sqlite3.Row
cur = conn.cursor()
try:
cur.execute("""
SELECT DATE(last_touch) as day, symbiosis, COUNT(*) as count
FROM rels
WHERE last_touch > ?
GROUP BY day, symbiosis
""", (cutoff,))
return [dict(r) for r in cur.fetchall()]
except Exception: return []
def detect_anomaly(self, platform: str, account_id: str) -> dict:
return {
"recent_rate": 1.0,
"baseline_rate": 1.0,
"drop": 0.0,
"anomaly": False,
"recommendation": "Normal"
}
```
### Verify engine (3-state)
```python
# core/verify_engine.py
import hashlib, json, subprocess
from enum import Enum
from pathlib import Path
class Status(str, Enum):
PASS = "PASS"
FAIL = "FAIL"
INCONCLUSIVE = "INCONCLUSIVE"
class VerificationEngine:
def __init__(self, work_dir: str = "./work"):
self.work_dir = Path(work_dir)
self.work_dir.mkdir(parents=True, exist_ok=True)
def verify_file(self, path: str, expected_duration: float = None) -> tuple[Status, str]:
p = Path(path)
if not p.exists() or p.stat().st_size == 0:
return Status.FAIL, "File missing or empty"
ok, out, err = self._run_ffprobe(path)
if not ok:
size_kb = p.stat().st_size / 1024
if size_kb > 10:
return Status.INCONCLUSIVE, f"FFprobe unavailable, file size {size_kb:.0f}KB seems valid"
return Status.FAIL, f"FFprobe failed and file too small ({size_kb:.0f}KB): {err}"
try:
data = json.loads(out)
duration = float(data["format"]["duration"])
if expected_duration and abs(duration - expected_duration) > 1.0:
return Status.FAIL, f"Duration mismatch: {duration} != {expected_duration}"
return Status.PASS, f"Duration: {duration}s"
except Exception:
return Status.INCONCLUSIVE, "Could not determine duration"
def verify_image(self, path: str) -> tuple[Status, str]:
"""Quick image validation — checks file header magic bytes."""
p = Path(path)
if not p.exists() or p.stat().st_size == 0:
return Status.FAIL, "Image missing or empty"
header = p.read_bytes()[:8]
if header[:3] == b'\xff\xd8\xff':
return Status.PASS, "Valid JPEG"
if header[:8] == b'\x89PNG\r\n\x1a\n':
return Status.PASS, "Valid PNG"
if header[:4] == b'RIFF' and header[8:12] == b'WEBP' if len(header) > 11 else False:
return Status.PASS, "Valid WebP"
return Status.INCONCLUSIVE, f"Unknown image format: {header[:4]}"
def _run_ffprobe(self, path: str) -> tuple[bool, str, str]:
import shutil
if not shutil.which("ffprobe"):
return False, "", "ffprobe not installed"
cmd = ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "json", path]
res = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
return res.returncode == 0, res.stdout, res.stderr
"""framemd5 hash — deterministic for same decoded frames."""
import shutil
if not shutil.which("ffmpeg"):
return Status.INCONCLUSIVE, "ffmpeg not installed"
cmd = ["ffmpeg", "-i", path, "-map", "0", "-f", "framemd5", "-"]
res = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
if res.returncode != 0:
return Status.FAIL, res.stderr[:200]
sig = hashlib.sha256(res.stdout.encode()).hexdigest()[:16]
return Status.PASS, f"framemd5_sig={sig}"
```
### V38 system orchestrator
```python
# core/v38_system.py
import asyncio, json, re
from pathlib import Path
from typing import Dict, List, Optional
from core.ai_cli import AICLIIntegration
from core.platform_hacks import PlatformHacksEngine, AdaptiveHackEngine
from core.browser_automation import BrowserAutomation, ContentPackage
from core.shadowban import ShadowbanDetector
from core.rate_governor import PredictiveRateGovernor, BioMimeticScheduler
from core.circuit_breaker import CircuitBreaker
from core.error_handler import error_handler
from core.structured_logger import logger
from core.telemetry import TelemetryEngine
from core.topic_sanity import topic_niche_check
from core.prepublish_scorer import PrePublishScorer
from core.content_validator import ContentValidator
from core.engagement_engine import EngagementEngine, daily_growth_routine, first_60_minutes
from core.browser_scraper import BrowserScraper
from core.v38_hybrid import (
QuantumBehavioralFingerprint,
NeuroSymbioticRelationshipEngine,
CausalTemporalActionPlanner
)
from core.v38_hybrid_planner import CausalTemporalActionPlanner
from core.strategy_engines import (
SyndicateEngine,
PulseScanner,
PhoenixProtocol,
MirrorProtocol,
DMWeaver,
TagMiner
)
class V38AutomatorSystem:
def __init__(self, niche: str = "general", agent_id: str = "default"):
self.niche = niche
self.agent_id = agent_id
self.ai = AICLIIntegration(agent_id=agent_id)
self.hacks = PlatformHacksEngine()
self.adaptive = AdaptiveHackEngine(self.hacks)
self.browser = BrowserAutomation()
self.shadowban = ShadowbanDetector(self.hacks)
self.gov = PredictiveRateGovernor()
self.breaker = CircuitBreaker()
self.errors = error_handler
self.log = logger
self.telemetry = TelemetryEngine()
self.pub_scorer = PrePublishScorer()
self.ContentPackage = ContentPackage
self.qbf = QuantumBehavioralFingerprint()
self.rels = NeuroSymbioticRelationshipEngine()
self.planner = CausalTemporalActionPlanner()
self.syndicate = SyndicateEngine(self.ai)
self.pulse = PulseScanner()
self.phoenix = PhoenixProtocol(self.ai)
self.mirror = MirrorProtocol(self.ai)
self.weaver = DMWeaver()
self.miner = TagMiner()
async def boot_hybrid(self):
await self.rels.init()
await self.planner.init()
await self.telemetry.init()
async def generate(self, topic: str, platform: str, hack: str = "auto") -> dict:
resp = await self.ai.generate(f"Create {platform} post about {topic}", platform)
data = resp.parse_json() or {"caption": resp.content, "hashtags": []}
data["_tool_used"] = resp.tool_used
check = topic_niche_check([topic], self.niche, data.get("caption", ""), data.get("hashtags"))
if check["warn"]:
resp = await self.ai.generate(
f"Create {platform} post about {topic} strictly for '{self.niche}' niche. "
f"Use niche keywords: {', '.join(check.get('matched') or [])}.", platform)
data = resp.parse_json() or data
data["_tool_used"] = resp.tool_used
check = topic_niche_check([topic], self.niche, data.get("caption", ""), data.get("hashtags"))
data["_niche_check"] = check
hack_name = self.adaptive.select_hack(platform, data) if hack in (None, "auto") else hack
data = self.hacks.apply_hack(platform, hack_name, data).applied_content
data["_hack_used"] = hack_name
# Validate & Pad
data = ContentValidator.pad_hashtags(data)
return data
def score(self, content: dict) -> dict:
virality = self._virality_score(content)
result = self.pub_scorer.score(
content, virality=virality,
niche_check=content.get("_niche_check"),
circuit_state=getattr(self.breaker, "state", "CLOSED"))
result["virality"] = virality
return result
def _virality_score(self, content: dict) -> int:
from agent.virality_scorer import ViralityScorer
passed = []
if content.get("caption") or content.get("hook"): passed.append("hook_first_3s")
tags = len(re.findall(r"#\w+", content.get("caption", "")))
if tags >= 11: passed.append("hashtag_min_11")
if tags >= 3: passed.append("hashtag_strategy")
passed.append("engagement_cta")
return ViralityScorer.score(passed)
async def _publish_once(self, platform: str, account_id: str, topic: str, page, human, hack: str) -> dict:
from core.anti_detection import ChallengeHandler
from core.phased_growth import PhasedGrowth
ch = ChallengeHandler(page, platform, account_id)
phased = PhasedGrowth()
if not self.gov.can_proceed(platform, account_id, "posts"): return {"status": "rate_limited"}
if not phased.can_action("posts"): return {"status": "phased_growth_block"}
content = await self.generate(topic, platform, hack)
scored = self.score(content)
if scored.get("rejected"):
return {"status": "rejected", "score": scored, "niche": content.get("_niche_check")}
sb = await self.shadowban.check_shadowban(page, platform)
if sb["shadowbanned"]:
await self.shadowban.trigger_recovery(platform, self.gov)
return {"status": "shadowban", "details": sb}
media = content.get("media_path")
pkg = self.ContentPackage(platform=platform, caption=content.get("caption", ""),
hashtags=content.get("hashtags", []), media_path=media,
title=content.get("title", topic))
res = await self.browser.publish(pkg, account_id, page, human, ch)
ok = res.success
self.gov.record_outcome(platform, account_id, "posts", ok)
await self.telemetry.log(self.agent_id, "publish", ok, "NONE" if ok else "LOGIC", res.error)
# ... [truncated — full in new-hybrid-4.7.md] ...
```
### publish.py CLI
```python
# publish.py
import argparse, asyncio, json, logging, sys
from pathlib import Path
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("publish")
async def _with_session(platform, account, fn):
from core.dispatcher import HybridDispatcher
from core.cookie_manager import EncryptedCookieManager
from core.anti_detection import SessionHealthMonitor
disp = HybridDispatcher(account, platform, headless=False)
session = await disp.get_session("auto")
page = session["page"]
try:
out = await fn(page, session)
monitor = SessionHealthMonitor(page, platform, account)
is_healthy, reason = await monitor.is_healthy()
if is_healthy:
mgr = EncryptedCookieManager()
mgr.save(account, await session["context"].cookies())
else:
log.warning(f"Session unhealthy ({reason}), skipping cookie save.")
return out
finally:
await disp.close(session)
async def main():
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=[
"boot", "publish", "shadowban", "grow", "scrape", "engage",
"comment", "respond", "group", "syndicate", "pulse", "smart_dm", "mine_tags"
])
parser.add_argument("--platform", default="instagram")
parser.add_argument("--account", default="account_1")
parser.add_argument("--topic", default="AI tools")
parser.add_argument("--hack", default="auto")
parser.add_argument("--tag", default="", help="hashtag for scrape")
parser.add_argument("--profile", default="", help="profile URL")
parser.add_argument("--post-url", default="", dest="post_url")
parser.add_argument("--text", default="", help="comment/dm text")
parser.add_argument("--group-url", default="", dest="group_url")
parser.add_argument("--group-action", default="post", choices=["post", "invite"], dest="group_action")
parser.add_argument("--mode", default="balanced", choices=["light", "balanced", "aggressive"])
parser.add_argument("--limit", type=int, default=10)
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--platforms", default="instagram,twitter,tiktok", help="comma list for syndicate")
args = parser.parse_args()
if args.command == "boot":
from core.boot import boot
print(json.dumps(boot(), indent=2))
return
from core.v38_system import V38AutomatorSystem
from core.humanization import HumanizationEngine
from core.anti_detection import ChallengeHandler
from core.supervisor import ResilientSupervisor
system = V38AutomatorSystem(niche="general", agent_id=args.account)
await system.boot_hybrid()
human = HumanizationEngine()
if args.dry_run:
content = await system.generate(args.topic, args.platform, args.hack)
scored = system.score(content)
print(json.dumps({"dry_run": True, "content": content, "score": scored,
"niche": content.get("_niche_check")}, indent=2))
return
if args.command == "publish":
async def run(page, _sess):
return await system.full_workflow(args.platform, args.account, args.topic, page, human, args.hack)
async def supervised_publish():
return await _with_session(args.platform, args.account, run)
sup = ResilientSupervisor(agent_id=args.account, max_restarts=3)
print(json.dumps(await sup.run_loop(supervised_publish, "publish"), indent=2))
elif args.command == "shadowban":
from core.shadowban import ShadowbanDetector
async def run(page, _sess):
return await ShadowbanDetector(system.hacks).check_shadowban(page, args.platform)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "grow" or args.command == "engage":
async def run(page, _sess):
return await system.engage(page, args.platform, args.account, human, args.mode)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "scrape":
async def run(page, _sess):
return await system.scrape(page, args.platform, args.tag, args.profile, args.limit)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "comment":
async def run(page, _sess):
return await system.comment(page, args.platform, args.account, human, args.post_url, args.text)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "respond":
async def run(page, _sess):
return await system.respond(page, args.platform, args.account, human, args.post_url, args.topic)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "group":
async def run(page, _sess):
return await system.group_action(page, args.platform, args.account, human,
args.group_url, args.group_action, args.text, args.profile)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "syndicate":
platforms = [p.strip() for p in args.platforms.split(",") if p.strip()]
async def run(page, _sess):
return await system.syndicate(page, human, args.topic, args.account, platforms)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "pulse":
async def run(page, _sess):
return await system.pulse(page, args.platform)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "smart_dm":
if not args.profile:
print(json.dumps({"status": "error", "message": "--profile required for smart_dm"}, indent=2))
return
async def run(page, _sess):
text = await system.smart_dm(page, args.profile, args.platform)
return {"status": "ok", "dm_text": text, "target": args.profile}
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
elif args.command == "mine_tags":
seed = args.tag or args.topic.replace(" ", "")
async def run(page, _sess):
return await system.mine_tags(page, args.platform, seed)
print(json.dumps(await _with_session(args.platform, args.account, run), indent=2))
if __name__ == "__main__":
asyncio.run(main())
```
### Incident response L1–L3
```python
# core/incident_response.py
import asyncio, json, logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import IntEnum
from pathlib import Path
from typing import List
log = logging.getLogger("incident")
class IncidentTier(IntEnum):
L1_WARNING = 1
L2_CONTAIN = 2
L3_CRITICAL = 3
@dataclass
class IncidentRecord:
tier: IncidentTier
platform: str
account_id: str
trigger: str
actions_taken: List[str] = field(default_factory=list)
timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
class IncidentResponseEngine:
LOG_PATH = Path("logs/incidents.jsonl")
def __init__(self, governor=None, circuit_breaker=None, webhook_url: str = None):
self.governor = governor
self.cb = circuit_breaker
self.webhook_url = webhook_url
self.LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
def classify(self, trigger: str) -> IncidentTier:
t = trigger.lower()
if any(x in t for x in ("shadowban", "locked", "banned", "suspended")):
return IncidentTier.L3_CRITICAL
if any(x in t for x in ("checkpoint", "session_expired", "captcha", "challenge")):
return IncidentTier.L2_CONTAIN
return IncidentTier.L1_WARNING
async def execute(self, platform: str, account_id: str, trigger: str, page=None) -> dict:
from core.session_health import SessionHealthCoordinator
tier = self.classify(trigger)
record = IncidentRecord(tier=tier, platform=platform, account_id=account_id, trigger=trigger)
actions = []
if tier >= IncidentTier.L1_WARNING:
if self.governor:
self.governor.pause_platform(platform, minutes=30)
actions.append("governor_pause_30m")
if page:
await page.goto("about:blank")
await asyncio.sleep(60)
actions.append("cooldown_60s")
if tier >= IncidentTier.L2_CONTAIN:
q = SessionHealthCoordinator().on_detection(account_id, platform, trigger)
if q.get("quarantined"): actions.append("cookie_quarantine")
if self.cb:
self.cb.force_open(platform)
actions.append("circuit_breaker_open")
if tier >= IncidentTier.L3_CRITICAL:
if self.governor:
self.governor.pause_platform(platform, hours=48)
actions.append("governor_pause_48h")
self._notify_webhook(record, actions)
actions.append("webhook_fired")
record.actions_taken = actions
with open(self.LOG_PATH, "a") as f:
f.write(json.dumps(record.__dict__, default=str) + "\n")
return {"tier": tier.name, "trigger": trigger, "actions": actions,
"manual_steps": self._manual_steps(tier)}
def _manual_steps(self, tier: IncidentTier) -> List[str]:
if tier == IncidentTier.L3_CRITICAL:
return ["Pause automation 48h", "Manual engagement only", "Re-run shadowban canary before resume"]
if tier == IncidentTier.L2_CONTAIN:
return ["Solve challenge on real device", "Re-export cookies after login"]
return ["Reduce action frequency", "Check proxy/IP reputation"]
def _notify_webhook(self, record, actions):
if not self.webhook_url: return
try:
from core.structured_logger import StructuredLogger
StructuredLogger().fire_webhook("detection_triggered", {
"tier": record.tier.name, "platform": record.platform,
"account": record.account_id, "trigger": record.trigger, "actions": actions})
except Exception as e:
log.warning("webhook_failed: %s", e)
# core/emergency_tactics.py — backward-compatible alias
class BanEvasionEngine:
def __init__(self, governor=None, circuit_breaker=None, webhook_url=None):
self._engine = IncidentResponseEngine(governor, circuit_breaker, webhook_url)
async def cooldown_session(self, page, minutes: int = 15, platform: str = "",
account_id: str = "", trigger: str = "rate_limited"):
return await self._engine.execute(platform or "unknown", account_id or "default", trigger, page=page)
```
### Self-check
```python
# self_check.py
import sys, base64, hashlib, hmac, struct, time, os
from pathlib import Path
def check(name, condition, detail=""):
status = "PASS" if condition else "FAIL"
print(f" [{status}] {name}" + (f" — {detail}" if detail else ""))
return condition
def _totp(secret_b32: str) -> str:
pad = "=" * ((8 - len(secret_b32) % 8) % 8)
key = base64.b32decode((secret_b32.upper() + pad).encode())
msg = struct.pack(">Q", int(time.time()) // 30)
h = hmac.new(key, msg, hashlib.sha1).digest()
o = h[-1] & 0x0F
code = struct.unpack(">I", h[o:o + 4])[0] & 0x7FFFFFFF
return str(code % 1000000).zfill(6)
def run_checks():
passed = 0
total = 0
print("=== OMNI-SOCIAL v4.3-SLIM-V2 Self-Check ===\n")
# §1 Boot
total += 1
try:
from core.boot import AutoDiscovery, boot
tools = AutoDiscovery.detect_tools()
passed += check("§1 Boot/AutoDiscovery", isinstance(tools, dict), f"found {sum(tools.values())} tools")
except Exception as e:
check("§1 Boot/AutoDiscovery", False, str(e))
# §2 Logger
total += 1
try:
from core.structured_logger import StructuredLogger
lg = StructuredLogger()
lg.info("self-check")
recent = lg.flush_recent(1)
passed += check("§2 StructuredLogger", len(recent) >= 1, f"buffer has {len(recent)} entries")
except Exception as e:
check("§2 Logger", False, str(e))
# §2b Error Handler
total += 1
try:
from core.error_handler import ErrorHandler, ErrorCode
eh = ErrorHandler()
code = eh.classify_exception(RuntimeError("timeout"))
passed += check("§2b ErrorHandler", code == ErrorCode.NETWORK_TIMEOUT, f"classified as {code.name}")
except Exception as e:
check("§2b ErrorHandler", False, str(e))
# §3 Cookie Manager
total += 1
try:
from core.cookie_manager import EncryptedCookieManager
mgr = EncryptedCookieManager()
cookie = {"sameSite": "lax", "name": "test"}
sanitized = mgr.sanitize_cookie(cookie.copy())
passed += check("§3 CookieManager", sanitized["sameSite"] == "Lax", f"sameSite={sanitized['sameSite']}")
except Exception as e:
check("§3 CookieManager", False, str(e))
# §5 Circuit Breaker
total += 1
try:
from core.circuit_breaker import CircuitBreaker
cb = CircuitBreaker(max_failures=2)
try: cb.execute(lambda: 1/0)
except: pass
try: cb.execute(lambda: 1/0)
except: pass
passed += check("§5 CircuitBreaker", cb.state == "OPEN", f"state={cb.state} after 2 failures")
except Exception as e:
check("§5 CircuitBreaker", False, str(e))
# §7 Content Validator
total += 1
try:
from core.content_validator import ContentValidator
result = ContentValidator.validate_caption("Great post! #a #b #c #d #e #f #g #h #i #j #k")
passed += check("§7 ContentValidator", result["valid"], f"hashtags={result['hashtag_count']}")
bad = ContentValidator.validate_caption("This post delves into the tapestry")
passed_bad = not bad["valid"]
total += 1
passed += check("§7 BannedWordDetection", passed_bad, f"banned={bad['banned_found']}")
except Exception as e:
check("§7 ContentValidator", False, str(e))
# §8 Humanization
total += 1
try:
from core.humanization import BioMimeticMouse
path = BioMimeticMouse.generate_bezier_path(0, 0, 100, 100)
passed += check("§8 BezierPath", len(path) > 20, f"points={len(path)}")
except Exception as e:
check("§8 BezierPath", False, str(e))
# §9 Research Pipeline
total += 1
try:
from core.research_pipeline_v43 import ResearchPipelineV43
rp = ResearchPipelineV43(findings_path="cache/test_findings.jsonl")
local = rp.search_local("nonexistent query 12345")
passed += check("§9 ResearchPipeline", isinstance(local, list), f"local results={len(local)}")
except Exception as e:
check("§9 ResearchPipeline", False, str(e))
# §11 NSRE Database
total += 1
try:
from core.v38_hybrid import Relationship
rel = Relationship("test_partner")
passed += check("§11 NSRE Instance", rel.partner_id == "test_partner", f"partner={rel.partner_id}")
except Exception as e:
check("§11 NSRE Instance", False, str(e))
# §12g Multi-Account Orchestrator
# ... [truncated — full in new-hybrid-4.7.md] ...
```
---
## 8. Growth Playbook (Merged Overgrowth)
### 8.1 Foundation (before posting)
- One clear niche + 3–5 content pillars
- Identical name, handle, visual identity, keyword bio across platforms
- One lead platform 30–60 days (TikTok or IG Reels for zero→audience)
- Optimize for **shares, saves, comments, completion** — not likes
- Native only: no watermarks, no pure copy-paste
### 8.2 High-retention video blueprint
1. **0–3s pattern interrupt** — motion + bold text + high-stakes line (never “Hey guys”)
2. **3–10s open loop** — information gap / conditional framing
3. **Middle pacing** — visual change every 2–3s; punch-in 10–15%
4. **End cold-cut / seamless loop** — no “thanks for watching”
**Edit hacks:** punch-in on sentence change · J-cuts 0.2–0.5s · word-by-word captions · 1.1–1.2× speech · cut the second the retention graph dips.
### 8.3 Hub-and-spoke engine
```
1 hub (talking head / tutorial / strong take)
→ 3–6 vertical shorts (different hooks)
→ 1 X/Threads post or thread
→ 1–2 carousels
→ Stories / newsletter takeaway
```
Repurpose rule: **1 long → 8–12 shorts → 4–6 carousels → 2–3 threads**.
### 8.4 Universal actions
- Hook in first 0.5–3s (or first line of text)
- 80/20 value-to-promotion
- Batch weekly; ride trends in 2–24h through niche filter
- Reply every comment first 30–60 min
- Engage 20–40 min pre/post in niche
- Pin a high-value or polarizing follow-up comment
### 8.5 Platform quick playbooks
| Platform | Cadence & tactics |
|----------|-------------------|
| **TikTok** | 3–5/wk · 7–30s viral / ≤60s authority · trend sound 24–72h · 3–5 niche tags · stitch/duet · series |
| **Instagram** | Reels 3–7/wk · carousels 2–4/wk (saves) · Stories daily + stickers · Collab weekly · design for DM shares |
| **YouTube** | Shorts → long funnel · chapters · first 24h engagement · result-first 5s · SEO transcript |
| **X/Twitter** | Value replies under big accounts · threads with payoff · avoid spammy auto-engage |
| **Facebook** | Groups value-first · live ≥10 min · meaningful comments · repurpose live → reel |
| **LinkedIn** | Document carousels · founder POV · comment early on niche posts |
### 8.6 Algorithm-aware tactics (ethical only)
- **3-second score** — pattern interrupt or die
- **Mid-roll second hook** at ~50%
- **Loop / cliffhanger** for rewatches (retention craft, not deception)
- **Carousel save bait** — last slide “save this”
- **Remix / Add Yours** to seed chain reactions
- **Purge dead followers** for engagement rate (hygiene)
- **Peak-time post** + first-hour reply storm
### 8.7 30-day growth skeleton
| Week | Focus |
|------|--------|
| 1 | Algorithm testing — 1 platform, 5 hooks, retention graphs |
| 2 | Content scaling — hub-spoke batch, double output of winners |
| 3 | Collab / cross-promo / stitch-duet / guest posts |
| 4 | Cut losers, 2× winners, systemize calendar + Path C/A automation |
### 8.8 Never do
Fake engagement · mass follow/unfollow bots · purchased followers · scraped spam DMs · captcha farms · multi-account IP pools pretending to be strangers · copyright music on TikTok.
---
## 9. Lead Generation Playbook (Ethical)
Source condensed from `Lead generation and media automation.md`.
### 11 free lead methods (easiest → hardest)
1. **LinkedIn** — connections export / intentional outreach + personalization (no bulk scrape bots)
2. **X/Twitter bios** — public emails only; careful public-data collection
3. **Reddit pain mining** — answer high-intent threads; soft CTA to lead magnet
4. **YouTube competitor comments** — public problem statements you solve
5. **G2/Capterra** — reviewers of adjacent tools
6. **Podcast guests** — show notes emails
7. **Quora** — answer → magnet funnel
8. **Slack/Discord** — value first, no spam blasts
9. **Maps B2B** — public business contacts where permitted
10. **Webinar/workshop** — co-host or attend + offer
11. **SEO + lead magnet engine** — owned media compounder
### Telegram
Prefer organic admin/value presence over mass Telethon spam. Account safety > volume.
### Meta ads (3-3-3 beginner)
- 3 creatives × 3 audiences × 3-day read
- Kill losers fast; scale winners ~20–30%
- Use Ad Library for public competitor creative research
### Tool stack
| Layer | Tools |
|-------|--------|
| Creative | CapCut / native editors / Apex editor pipeline |
| Schedule | Path C queue or Path A hybrid scheduler |
| CRM/leads | Sheets + email tool of choice |
| Ads | Meta Ads Manager native |
| Research | Slim-search / CLEAR gate |
---
## 10. Research & Search Extension (CLEAR / Lite)
### When search runs
Only with `--factual`, user verify request, caption stats needing sources, or `needs_search: true`.
### Evidence contract
- Confidence ≥ 65
- ≥ 2 independent domains
- Verdicts: VERIFIED | LIKELY_TRUE | UNCERTAIN | FALSE
- Store in OmniMemory FTS when available
### Omni search hook (hybrid)
```python
# (source block not found — open original file)
```
### Confidence scoring pattern (pro-mistral)
```python
class ConfidenceScorer:
def __init__(self):
self.weights = {
'source_authority': 0.30,
'cross_references': 0.25,
'expert_consensus': 0.20,
'temporal_freshness': 0.15,
'logical_consistency': 0.10
}
def score(self, source):
scores = {
'source_authority': min(source.get('domain_authority', 0), 100),
'cross_references': min(source.get('cross_references', 0) / 3 * 100, 100),
'expert_consensus': min(source.get('expert_endorsements', 0) / 5 * 100, 100),
'temporal_freshness': max(0, 100 - (source.get('age_days', 0) / 180 * 100)),
'logical_consistency': source.get('logical_score', 100)
}
total = sum(scores[k] * self.weights[k] for k in self.weights)
if total >= 90: grade = 'A+ (Exceptional)'
elif total >= 80: grade = 'A (Excellent)'
elif total >= 70: grade = 'B (Good)'
elif total >= 60: grade = 'C (Fair)'
else: grade = 'F (Unreliable)'
return {'scores': scores, 'total': total, 'grade': grade}
```
### Triangulation pattern
```python
def triangulate(claims: list[str], sources_by_claim: dict) -> str:
"""Require agreement across independent sources before asserting fact."""
supported = [c for c in claims if len(sources_by_claim.get(c, [])) >= 2]
if len(supported) == len(claims) and claims:
return "VERIFIED"
if supported:
return "LIKELY_TRUE"
return "UNCERTAIN"
```
### 5-layer research method
1. Define question + **disconfirming** query
2. Primary sources first (docs, papers, official)
3. Secondary synthesis (reputable press)
4. Contrarian pass (what would make this wrong?)
5. CLEAR gate + citation pack before publish
### Rate-safe research fetch
```python
# ponytail: simple token bucket; upgrade to Redis if multi-process
import time, random
from collections import deque
class FetchPacer:
def __init__(self, max_per_minute: int = 20):
self.max = max_per_minute
self.hits: deque = deque()
def wait(self) -> None:
now = time.time()
while self.hits and now - self.hits[0] > 60:
self.hits.popleft()
if len(self.hits) >= self.max:
time.sleep(60 - (now - self.hits[0]) + random.uniform(0.2, 1.0))
self.hits.append(time.time())
```
---
## 11. Mastery Takeaways (2026)
### ✅ **Top 7 Insights for 2026**
| Rank | Insight | Impact Level | Implementation Priority |
| ---- | ----------------------------------------------------------------------------------- | ------------ | ----------------------- |
| 1 | **API-First with Stealth Fallback** is the only sustainable architecture | 🔴 Critical | **DO FIRST** |
| 2 | **Deterministic Fingerprinting** (seeded per account) defeats correlation detection | 🔴 Critical | **DO FIRST** |
| 3 | **Rate Limits Are Non-Negotiable** - Conservative pacing keeps accounts alive | 🔴 Critical | **DO FIRST** |
| 4 | **Cookie + Proxy + UA Matching** is the holy trinity for session validity | 🟡 High | Week 1 |
| 5 | **Screenshot Verification** is the only way to confirm "really done" | 🟡 High | Week 1 |
| 6 | **AI Agent Orchestration** (Claude/Mistral/Grok) multiplies output 10x | 🟢 Medium | Week 2 |
| 7 | **Anti-Shadowban Protocols** can recover 80% of flagged accounts | 🟢 Medium | Week 2 |
---
### Six signal layers platforms track
1. Browser / TLS / canvas / WebGL fingerprint
2. IP / ASN / geo consistency
3. Behavioral timing (mouse, scroll, key)
4. Graph / social graph anomalies
5. Content similarity / spam templates
6. Session / cookie / device binding
**Defense:** be indistinguishable from a careful human on **one** device identity.
---
## 12. Platform Rate & Media Specs (Working Defaults)
> Re-check live platform docs; these are operational starting ceilings.
| Platform | Cold posts/day | Warm posts/day | Media notes |
|----------|----------------|----------------|-------------|
| TikTok | 1 | 3–5 | 9:16, prefer native sounds |
| Instagram | 1 Reel | 1–2 Reels + 1 carousel | 9:16 reels; carousels for saves |
| X/Twitter | 3–5 | 8–15 | Threads > single spam |
| Facebook | 1 | 2–3 | Groups value-first |
| LinkedIn | 1 | 1–2 | Documents / carousels |
| YouTube Shorts | 1 | 1–3 | Hook 0–1s; funnel to long |
### first_60_minutes checklist
```
[ ] Post published, URL captured
[ ] Pinned comment with CTA or extra value
[ ] Reply to every early comment
[ ] Share to 1–3 owned channels (Story, newsletter) — not spam groups
[ ] Log metrics baseline (0 / 15 / 60 min)
[ ] If zero distribution after 2h on TT/IG → review sound/hook/tags, not bot boosts
```
---
## 13. CLI Surface (Unified Mental Model)
| Intent | Path A (hybrid) | Path C (autopilot) |
|--------|-----------------|--------------------|
| Login / session | cookie import + enter_account | `run.py login` |
| Publish due items | `publish.py publish` | `run.py once` / `serve` |
| Growth routine | `publish.py grow` | manual + calendar |
| Shadowban check | `publish.py shadowban` | n/a (manual) |
| Replies | engage / smart_dm (gated) | `run.py replies` (draft) |
| Analytics | analytics module | `run.py analytics` |
| Verify | `--verify` / verify_engine | dry-run + human |
| Research | `--factual` | external slim-search |
### Path C settings skeleton
```yaml
# Copy this file to config/settings.yaml and fill in your own values.
# One account per copy of this repo/config -- that's intentional.
account:
platform: "example" # just a label, used in logs/paths
target_url: "https://example.com/login" # the site you're automating
storage_state_path: "sessions/example.enc"
encrypt_storage_state: true
browser:
patchright_executable: null # path patchright installed chromium to; leave null to use browser-use's bundled default
headless: false
user_data_dir: "profiles/example"
rate_limits:
min_seconds_between_actions: 90 # simple pacing between actions, not a growth ramp
max_posts_per_day: 5
max_replies_per_hour: 10
publishing:
# Optional deterministic step list, e.g.:
# steps:
# - {action: goto, url: "https://example.com/compose"}
# - {action: fill, selector: "#caption", value: "{caption}"}
# - {action: upload, selector: "input[type=file]", field: "media_path"}
# - {action: click, selector: "button.publish"}
# Leave empty and the agent will work out the page itself.
steps: []
replies:
auto_send: false # drafts only, until you flip this on purpose
llm_model: "claude-sonnet-5"
analytics:
post_urls_file: "content/post_urls.json"
output_path: "analytics/metrics.csv"
```
### Path C entrypoint
```python
"""
CLI entrypoint.
python run.py login # open a browser window, log in by hand, save the session
python run.py once # publish anything currently due in the queue
python run.py serve # keep running, checking the queue every few minutes
python run.py replies # check for new comments/DMs and draft replies
python run.py analytics # pull metrics for your own posts
"""
import argparse
import asyncio
import yaml
from core.analytics import pull_analytics
from core.browser_session import get_browser_session
from core.replier import check_and_draft_replies
from core.scheduler import run_forever, run_once
from core.session_store import save_storage_state
def load_config(path: str = "config/settings.yaml") -> dict:
with open(path) as f:
return yaml.safe_load(f)
async def cmd_login(cfg: dict, session) -> None:
page = await session.get_current_page()
await page.goto(cfg["account"]["target_url"])
input("Log in by hand in the opened browser window, then press Enter here to save the session...")
state = await session.browser_context.storage_state()
save_storage_state(
state,
cfg["account"]["storage_state_path"],
cfg["account"].get("encrypt_storage_state", True),
)
print("Session saved.")
async def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=["login", "once", "serve", "replies", "analytics"])
args = parser.parse_args()
cfg = load_config()
session = get_browser_session(cfg)
if args.command == "login":
await cmd_login(cfg, session)
elif args.command == "once":
n = await run_once(cfg, session)
print(f"Published {n} item(s).")
elif args.command == "serve":
await run_forever(cfg, session)
elif args.command == "replies":
drafts = await check_and_draft_replies(cfg, session)
plural = "y" if len(drafts) == 1 else "ies"
print(f"Drafted {len(drafts)} repl{plural} -> replies/pending.json")
elif args.command == "analytics":
rows = await pull_analytics(cfg, session)
print(f"Pulled metrics for {len(rows)} post(s).")
if __name__ == "__main__":
asyncio.run(main())
```
---
## 14. VPS Ops Notes (Mode C)
| Alias | Role |
|-------|------|
| hostinger | VPS A |
| contabo | VPS B |
| hetzner | VPS C |
| ai-developer | spare / build |
Hardening:
```
- ssh keys only, no password auth
- fail2ban + unattended-upgrades
- per-account process isolation
- rotated logs; agent_monitor.db per VPS
- never leave plaintext cookies after .enc
```
Cron sketch:
```cron
# Active hours only — timezone of account's VPS
0 9,13,17 * * * cd /opt/omni && python publish.py once --account $ACC
30 10 * * * cd /opt/omni && python publish.py grow --account $ACC
0 3 * * 0 cd /opt/omni && python publish.py shadowban --account $ACC
```
---
## 15. Execution Checklists
### Every publish
```
[ ] boot() PASS
[ ] session not quarantined
[ ] PhasedGrowth allows action
[ ] PrePublish + virality PASS
[ ] Caption GPT-ism strip
[ ] Media aspect/safe zone OK
[ ] Dry-run once on new templates
[ ] Live publish
[ ] first_60_minutes engagement
[ ] Log post_id + metrics seed
```
### Weekly ops
```
[ ] Cookie freshness / silent refresh health
[ ] Shadowban canary all accounts
[ ] Kill low-retention formats
[ ] 2× top 20% hooks
[ ] Review incident log L1–L3
[ ] Disk / log rotation on VPS
```
### Research publish (`--factual`)
```
[ ] Query + disconfirming query
[ ] ≥2 domains
[ ] Confidence ≥65
[ ] Contrarian note stored
[ ] Citations in caption or pinned comment
[ ] Three-state gate PASS
```
---
## 16. Working Notes — Pro Tips That Survive Reality
1. **Stealth without content quality still dies.** Automation multiplies both hits and misses.
2. **Deterministic steps beat agents for known DOM** — agents for recovery and discovery.
3. **SameSite Titlecase** is a stupid-high failure rate for cookie inject; sanitize always.
4. **1:1 account:IP** is cheaper than recovering bans.
5. **Quarantine is success** — better 48h pause than permanent loss.
6. **Retention graph is the editor** — cut the dip second.
7. **Hub-spoke prevents burnout** more than any AI tool.
8. **Draft replies default** — auto-send only after trust.
9. **INCONCLUSIVE means stop** — never ship “probably fine.”
10. **Search is a spice** — default automate mode stays offline-of-web.
11. **ponytail:** Prefer one Fernet session file over a framework when Path C is enough.
12. **Upgrade path:** Path C → B (stealth engine) → A (multi-account product) only when pain is real.
---
## 17. Self-Check Snippet (Doc Integrity)
```python
#!/usr/bin/env python3
"""Minimal integrity check for this unified doc + key sources."""
from pathlib import Path
BASE = Path(__file__).resolve().parent if "__file__" in dir() else Path(".")
# when run from this folder:
BASE = Path("/Users/khaledahmedmohamed/designs-content/vip/social media agent/the full merged manager hybrid")
required = [
"FULL-MERGED-MANAGER-HYBRID-UNIFIED.md",
"new-hybrid-4.7.md",
"Ultimate hybrid stack-media automation.md",
"social media overgrowth.md",
"Lead generation and media automation.md",
]
missing = [r for r in required if not (BASE / r).exists()]
assert not missing, f"Missing: {missing}"
text = (BASE / "FULL-MERGED-MANAGER-HYBRID-UNIFIED.md").read_text()
for needle in [
"Operating Contract",
"smart_click",
"PhasedGrowth",
"PASS | FAIL | INCONCLUSIVE",
"browser-use",
"hub-and-spoke",
"passive_only",
]:
assert needle in text, f"Missing section marker: {needle}"
print("PASS: unified merge integrity markers present")
```
---
## 18. AI-Agent Acting Logic — Golden Rules, Pro-Tips & Error-Avoidance Prompts
> **Purpose:** How the AI agent must think and act on hybrid manager tasks so it avoids the common failure modes (ban loops, false “done”, cookie death, search-on-by-default, centroid clicks, skipped gates).
>
> **Audience:** Any agent (Grok / Claude / AGY / gocode) executing publish · grow · engage · cookie · research · VPS ops from this unified spec.
### 18.0 Scan coverage certificate (this merge)
All **40** source files under `the full merged manager hybrid/` were inventoried and tip-scanned before this section was written:
| Category | Files |
|----------|--------|
| Root specs | `new-hybrid-4.7.md`, `Ultimate hybrid stack-media automation.md`, `Social Media Mastery Guide.md`, `Advanced media overgrowth.md`, `social media overgrowth.md` (+ duplicate 2), `Lead generation and media automation.md`, `pro-mistreal-searcher-agent-extension.md` |
| Path C | `social-autopilot/**` (README, run.py, core/*, config, queue) |
| Slim package | 3 SKILL.md, SLIM-V2/V3, z-kimi graft, upgrade/comparison/lecture, 7 adoption patches |
| Live skill memory (cross-check) | `hybrid_47_social_memory.json` + `hybrid-47-social/SKILL.md` |
| Prior unified | this file + `check_unified_merge.py` |
**Also mined:** all **43** `PRO TIP §…` blocks from hybrid 4.7, permanent JSON `directives` / `anti_block` / `auto_login` / `pro_tips` / `expert_notes.never`, stack “4 Golden Rules”, slim hard rules + error-fixation, mastery cookie failure modes, research prompts (pro-mistral).
---
### 18.1 Pre-flight brain loop (run before EVERY action)
```
PERCEIVE → What is the task? Which path (A/B/C)? Which account + VPS?
CLASSIFY → automate | verify | search | ops | content-only
GATE → boot() / secrets / ethics / quarantine / PhasedGrowth
ACT → minimal steps; smart_click; rate limits
VERIFY → PASS|FAIL|INCONCLUSIVE (never promote)
DECIDE → success log | self-heal ≤3 | quarantine+stop | ask human
```
**Hard stop conditions (do not continue):**
- `boot_warnings` non-empty without `--force`
- session status in `{shadowban, checkpoint, session_expired, account_locked}`
- CircuitBreaker open (3 consecutive fails)
- PrePublish / virality gate `rejected` or score < 60 when required
- Verification `INCONCLUSIVE` or `FAIL`
- ethics `passive_only` would be violated
- PhasedGrowth blocks the action for account age
---
### 18.2 Golden rules (logic law — memorize)
#### G0 — Identity & path
1. Prefer **Path C** for one owned account; **Path A** for multi-account Mode C; **Path B** when DOM is unknown/adaptive.
2. **One account → one VPS IP forever.** Never “rotate IPs to look like different people” for linked brands on the same home IP.
3. Load skill + permanent JSON **before** browser if prompt matches hybrid keywords.
#### G1 — Browser & stealth
4. **Never** `page.click()` / centroid click → **always** `smart_click()` cubic Bezier + padding offsets.
5. **Never** platform APIs (instagrapi, tweepy, Graph API, unofficial SDKs).
6. Headed mode preferred for social; cold publish needs **warm_up** first.
7. Scroll before click; Poisson burst-rest; no fixed-interval spam cron.
8. Proxy geo **must match** browser `timezone_id` + `locale` (Germany IP ≠ America/New_York).
#### G2 — Cookies & auth
9. Encrypt at rest → `sessions/{account}.enc` · `chmod 600` · delete plaintext.
10. **Sanitize SameSite to Titlecase** (`Lax`/`Strict`/`None`) or Playwright context dies.
11. Inject cookies **before** first navigate; save **after** success.
12. IG/TT: include `local_storage` in blob; `refresh_silently()` every ~45m on long engage.
13. On fail: **quarantine** to `*_expired.enc` — **never ban-loop retry**.
14. 2FA / captcha → Rescue Window 5m **manual** only (no captcha bypass automation).
#### G3 — Publish pipeline order
```
boot → secrets_scan → ServerRouter → Dispatcher
→ warm_up → PrePublish + AlgorithmMapper (+ TopicSanity)
→ maybe_search ONLY if --factual/needs_search
→ CaptionTransformer → publish
→ first_60_minutes → log_post → analytics
```
15. **Never skip warm_up** on cold publish.
16. **Never auto-invoke search** from publish/grow/engage.
17. Strip GPT-isms: delve, tapestry, unlock, landscape, elevate, moreover, testament.
18. Absolute media paths only; verify headers/magic bytes before upload.
19. Deterministic steps when DOM known; agent fallback only when steps fail/unknown.
#### G4 — Growth & rates
20. Accounts < 21 days: **PhasedGrowth** caps (day 0–6: posts=0).
21. Virality ≥ **60** before live publish when scorer enabled.
22. CircuitBreaker: abort after **3** consecutive failures.
23. NSRE: skip `parasitic_risk` targets.
24. Incident map: **301→L1** reduce · **308→L2** re-export cookies same IP · **310→L3** halt 48h.
#### G5 — Verification (anti false-positive)
25. Three-state only: **PASS | FAIL | INCONCLUSIVE**.
26. **INCONCLUSIVE ≠ PASS** — blocks “done” exactly like FAIL.
27. Re-verify after every fix; max **3** self-heal retries with evidence.
28. Never weaken checks, hardcode PASS, or catch-and-ignore verification errors.
29. Factual content: confidence ≥ **65**, ≥ **2** domains, contrarian pass.
#### G6 — Research (lite)
30. Default search **OFF**.
31. Never parse Google HTML with regex; never f-string URL-encode (`quote_plus`).
32. Never block event loop with `subprocess.run(ollama)` — aiohttp `/api/generate`.
33. Tag claims: `[VERIFIED]` / `[SINGLE-SOURCE]` / `[OPINION]` / `[SPECULATIVE]`.
34. DEFINE → WIDEN (≥6 angles) → NARROW before live fetch; vault/findings first.
#### G7 — Content growth craft
35. Hook in first 0.5–3s; never “Hey guys…”.
36. Cut the second the retention graph dips.
37. Hub-spoke: native per platform; no watermarked cross-posts.
38. Reply first 30–60 minutes after post.
39. AI multiplies drafts — **human/agent gate** on voice, ethics, final ship.
#### G8 — Path C honesty
40. Draft replies default; `auto_send: false` until trusted.
41. No multi-account disguise, no captcha bypass, no ban-evasion auto-resume in Path C.
42. One profile/process per account.
---
### 18.3 Never / Always card (print this)
| NEVER | ALWAYS |
|-------|--------|
| raw `page.click()` | `smart_click()` Bezier |
| platform APIs | browser + cookies |
| treat INCONCLUSIVE as PASS | three-state + re-verify |
| ban-loop retry after quarantine | stop → human → re-export same IP |
| inject cookies after navigate | inject **before** navigate |
| share one IP across many live accounts | 1:1 account:VPS |
| auto-search on publish | `--factual` / `needs_search` only |
| fixed 50ms clicks / rigid cron | Poisson + jitter + offsets |
| geo mismatch (proxy ≠ TZ/locale) | match fingerprint stack |
| publish on `gate_blocked` / boot_warnings | dry-run + fix gates |
| captcha auto-bypass | Rescue Window manual 5m |
| promote speculative facts | ≥2 domains + confidence ≥65 |
| GPT-ism captions | strip banned words |
| sync SQLite inside Playwright loop | `asyncio.to_thread` |
| headless file upload on social | headed + absolute paths |
---
### 18.4 Failure mode → classify → act (error map)
| Symptom | Likely class | First fix | Stop? |
|---------|--------------|-----------|-------|
| Login redirect / home not feed | session_expired | quarantine cookies; re-export **same VPS IP** | Yes until fresh |
| Empty tag chrono / not in canary | shadowban | L3 48h halt; manual only | Yes |
| Selector timeout on known UI | DOM drift or cookie half-dead | check cookie health **first**; then SelfHealingLocator | Retry ≤3 |
| 429 / rate friction | rate_limited (301) | L1 reduce 50% 24h; backoff | Soft |
| Checkpoint / puzzle | challenge | Rescue Window manual | Yes automation |
| Upload fails | media path / corrupt / aspect | absolute path; magic-byte check; re-export media | Retry ≤2 |
| Context create error on cookies | SameSite case / domain | `sanitize_cookie` Titlecase | Fix & retry |
| Low reach after publish | content not bot | retention hook / sound / timing — not more bots | No ban loop |
| Agent “done” without proof | false positive | force verify_engine / screenshot / URL | Block done |
| Ollama hang | blocked event loop | aiohttp generate; cascade CLI | Fix pattern |
| Async crash with SQLite | sync DB in async | `asyncio.to_thread` | Fix pattern |
| Boot warnings | secrets/ethics | clear warnings or `--force` with reason | Block publish |
| gate_blocked | PrePublish/virality | rewrite hook/caption; rescore | No publish |
| Cross-account link suspicion | IP sharing | split VPS routing | Halt multi |
**Error fixation protocol (slim + verification layer):**
```
Reproduce → Isolate → Classify (TRANSIENT|SESSION|RATE|BAN|DOM|CONTENT|ENV)
→ Fix root cause → Re-verify (must PASS) → log evidence
Max 3 self-heal cycles. TRANSIENT → restart. FATAL/BAN → halt.
```
---
### 18.5 Task playbooks (how to act)
#### A) Publish
```
1. boot() PASS; no quarantine; PhasedGrowth allows posts
2. Resolve ServerRouter → open dispatcher for account
3. warm_up (feed scroll/likes light)
4. PrePublishScorer + virality ≥60 + TopicSanity
5. CaptionTransformer + GPT-ism strip + hashtag rules
6. Upload with absolute media path; headed
7. Capture post URL / id
8. first_60_minutes: pin comment, reply, optional story share
9. log_post → AlgorithmMapper
10. Verdict PASS only with URL + non-empty proof
```
#### B) Grow / engage
```
1. boot + shadowban canary if due
2. RateGovernor + PhasedGrowth caps
3. NSRE filter (skip parasitic_risk)
4. Bezier interactions only; Poisson rests
5. refresh_silently every 45m if long session
6. CircuitBreaker; stop on 3 fails or checkpoint
```
#### C) Cookie import / auto-login
```
1. Detect platform from domains
2. Stage JSON → encrypt .enc → chmod 600 → delete plaintext
3. Patch account_routing (1:1 VPS)
4. Inject BEFORE navigate
5. Headful if 2FA; save AFTER success
6. Fail → quarantine; never tight retry
```
#### D) Research / research_publish
```
1. Mode = search only with --factual or needs_search
2. DEFINE → WIDEN → NARROW; vault first
3. ≥2 domains; confidence ≥65; contrarian
4. Tag claims; archive findings
5. Handoff package to caption — never invent stats
```
#### E) Fix / debug automation
```
1. Read last 5 errors from agent_monitor.db (not full logs)
2. Classify via error map above
3. Do NOT weaken gates to “make green”
4. Fix → re-run self_check → dry-run → limited live
```
#### F) Content growth (no browser)
```
1. Niche + pillars clear
2. Retention blueprint for video
3. Hub-spoke native variants
4. Schedule + first-hour reply plan
5. AI drafts only; final gate human/agent quality
```
---
### 18.6 System prompts for the acting agent
#### Master system prompt (drop into agent session)
```
You are the Hybrid Manager Operator Agent for browser-first social automation.
LAW:
1) boot()/self-check before any browser context.
2) Never page.click() — only smart_click() cubic Bezier with padding offsets.
3) No platform APIs. Cookies + human-mimetic browser only.
4) Encrypt cookies; SameSite Titlecase; inject BEFORE navigate; save AFTER success.
5) One account → one VPS IP forever (Mode C).
6) Pipeline: warm_up → PrePublish → publish → first_60_minutes.
7) Search OFF unless --factual or needs_search:true.
8) PASS|FAIL|INCONCLUSIVE only. INCONCLUSIVE blocks done. Never promote.
9) Quarantine on shadowban|checkpoint|session_expired|account_locked — no ban loops.
10) CircuitBreaker aborts after 3 consecutive failures.
11) PhasedGrowth for accounts <21 days; virality ≥60 when scoring.
12) ethics passive_only: true — no captcha bypass automation.
13) Strip GPT-isms from captions.
14) Evidence for factual claims: ≥2 domains, confidence ≥65.
15) Prefer smallest path: Path C single-account before Path A multi-account.
When uncertain: dry-run, verify, or STOP and report INCONCLUSIVE — never invent success.
After every material action: record status, evidence path/URL, and next safe step.
```
#### Publish operator prompt
```
Task: Publish for account={ACCOUNT} platform={PLATFORM} path={A|B|C}.
Steps you must execute in order:
1. Run boot/self-check. If warnings and no --force → STOP.
2. Confirm sessions/{ACCOUNT}.enc exists and is not quarantined.
3. Confirm PhasedGrowth allows a post today.
4. warm_up on home feed (scroll + light human actions).
5. Score content (PrePublish + virality). If <60 or rejected → rewrite, do not force.
6. Transform caption for platform; strip banned GPT-isms.
7. Publish with absolute media path; headed browser.
8. Capture canonical post URL.
9. first_60_minutes engagement routine.
10. Return: {status, post_url, scores, evidence, incidents}.
Forbidden: platform APIs, raw clicks, skipping warm_up, claiming done without URL.
```
#### Cookie / session recovery prompt
```
Task: Restore session for {ACCOUNT} on VPS {SERVER}.
1. Do NOT retry login more than once with the same dead cookies.
2. If redirect to login or checkpoint → quarantine to {ACCOUNT}_expired.enc.
3. Request fresh cookie export from browser on the SAME VPS IP.
4. Sanitize SameSite Titlecase; encrypt; chmod 600; delete plaintext.
5. Inject before navigate; verify feed/home loaded.
6. Save storage_state / cookie chain after success.
7. Report PASS with timestamp or FAIL with quarantine path.
```
#### Grow safely prompt
```
Task: Growth session for {ACCOUNT}, mode={balanced|conservative}.
Constraints:
- Respect RateGovernor + PhasedGrowth ceilings.
- NSRE: skip parasitic_risk.
- Bezier mouse only; Poisson burst-rest.
- No mass follow/unfollow; no purchased engagement.
- On 301 reduce; on 308 quarantine; on 310 halt 48h.
- refresh_silently every 45m if session >45m.
- Max actions = playbook ceiling for platform × account age factor.
End with action counts, skips, incidents.
```
#### Research / factual caption prompt
```
Task: Research for factual post on {TOPIC}.
Process:
1. DEFINE question + disconfirming query.
2. WIDEN ≥6 angles; NARROW to best sources.
3. Collect ≥2 independent domains.
4. Score confidence; require ≥65 for fact claims.
5. Contrarian pass: what would falsify this?
6. Output: Finding, Sources, Verification, Confidence/10, Caveats, Caption-ready bullets.
7. Tag each claim [VERIFIED]|[SINGLE-SOURCE]|[OPINION]|[SPECULATIVE].
Never invent statistics. If thin sources → INCONCLUSIVE, do not publish as fact.
```
#### Debug / fix automation prompt
```
Task: Fix failure {ERROR} for {ACCOUNT}.
Protocol:
1. Reproduce once with logs (last 5 agent_monitor errors).
2. Classify: TRANSIENT | SESSION | RATE | BAN | DOM | CONTENT | ENV.
3. Isolate minimal failing step.
4. Fix root cause (not symptom). Do not disable gates.
5. Re-run self_check → dry-run → one live action max.
6. Verdict PASS only with evidence. Else FAIL/INCONCLUSIVE + stop reason.
Max 3 fix attempts then STOP with report.
```
#### Content strategy prompt (growth, no browser)
```
Task: Plan 7-day hub-spoke content for niche={NICHE}.
Requirements:
- One hub piece + spoke matrix (shorts, carousel, thread, story).
- Hooks for 0–3s pattern interrupt; mid-roll hook; cold-cut ending.
- Platform-native captions (no watermarked cross-post).
- First-hour reply plan; save/share CTAs.
- Mark what AI can draft vs what needs human film.
Deliver calendar table + top 5 hooks + kill criteria for losers.
```
#### Lead-gen ethical prompt
```
Task: Lead pipeline for offer={OFFER} audience={AUDIENCE}.
Use only public, consent-respecting methods (LinkedIn intentional outreach,
Reddit value answers, podcast pitches, SEO magnets). No scrape-spam bots,
no Telethon mass spam from personal numbers, no purchased lists.
Output: channel ranking, weekly action counts, message templates, compliance notes.
```
#### Anti-bot shipping prompt (stack)
```
Before shipping any automation change, verify checklist:
[ ] fingerprint tests (sannysoft/browserscan)
[ ] webdriver === false
[ ] geo/timezone/locale match proxy
[ ] Bezier mouse + scroll-before-click
[ ] session persistence
[ ] rate jitter ≥2–5s
[ ] headed social uploads
[ ] warm-up days for new accounts
If any box fails → do not deploy live.
```
---
### 18.7 Hybrid 4.7 Pro-Tips — FULL ADOPTION (competitive power pack)
> **Source of truth:** `/Users/khaledahmedmohamed/designs-content/vip/new-hybrid-4.7.md`
> **Adopted:** 42 PRO TIP blocks, untruncated, with **agent actions** + **20 power rules**.
> **Purpose:** Make this UNIFIED file competitive with (and operationally stronger than) raw hybrid-4.7 for CLI AI agents.
### Competitive power rules (Hybrid 4.7 → agent MUST)
These turn tips into **hard execution policy** (what makes UNIFIED beat a raw code-spec dump):
| # | Power rule | Source tips |
|---|------------|-------------|
| P1 | **Search is opt-in only** — publish/grow/engage never open research | §0b |
| P2 | **Boot warnings block ship** — no silent live with dirty boot | §1b, §14 |
| P3 | **SameSite Titlecase or die** — cookie inject fails without sanitize | §3 |
| P4 | **IG/TT = cookies + local_storage blob** + 45m silent refresh on long runs | §3v2 |
| P5 | **Bezier or STEALTH_GAP** — raw centroid click is a ban signal | §8, §8b |
| P6 | **PhasedGrowth is law for <21d** — playbook numbers are ceilings | §10, §19 |
| P7 | **Canary empty = L3** — never “try again harder” on shadowban | §9, §17 |
| P8 | **first_60_minutes is part of publish** — not optional polish | §12 |
| P9 | **Absolute media path + header check** before upload | §11, §12h |
| P10 | **1 browser / 1 account process** — sequential orchestrator | §12g |
| P11 | **Async DB off event loop** | §4c |
| P12 | **Incident log every 301/308/310** to `logs/incidents.jsonl` | §17 |
| P13 | **Last 5 errors from agent_monitor.db** before inventing fixes | §2c |
| P14 | **Phoenix once** then halt — no morph spam | §12b |
| P15 | **IP risk at boot** on each VPS; prefer native IP | §12c2 |
| P16 | **Reject out-of-niche before footprint** | §7c |
| P17 | **Jitter schedules** — never rigid cron grid | §12d |
| P18 | **NSRE skip parasitic_risk** | §11c |
| P19 | **Rescue Window manual only** — no captcha automation | §7 |
| P20 | **Dual health** — self_check (import) + runtime health_check | §14, §14b |
### 18.7.1 Battle cards by domain (full tip text + agent action)
#### Boot, secrets & self-check
**§1**
- **Tip:** Add a `COOKIE_DIRS` env-var override so CI and Docker can point to mounted volumes without forking the code: `extra = os.environ.get("OMNI_COOKIE_PATH"); dirs = list(COOKIE_DIRS) + ([Path(extra)] if extra else [])`. This costs zero config for default users but unblocks containerized deployments.
- **Agent action:** Honor OMNI_COOKIE_PATH / COOKIE_DIRS env for CI/Docker/VPS mounts.
**§1b**
- **Tip:** Block `publish` when `boot_warnings` is non-empty unless `--force` is passed.
- **Agent action:** Abort publish if boot reports warnings unless user passed `--force`.
**§14**
- **Tip:** Testing suite exercises modules in isolation. Invoking `python self_check.py` parses compiler trees and checks basic instance creations.
- **Agent action:** Run self_check/import tests before live; health_check for runtime.
**§14b**
- **Tip:** Complements `self_check.py` (import test) with runtime ops monitoring.
- **Agent action:** Run self_check/import tests before live; health_check for runtime.
#### Cookies, auth & sessions
**§3**
- **Tip:** Platform cookies are volatile. Storing raw cookies without sanitizing SameSite case triggers Playwright runtime context creation exceptions. Always process using the `sanitize_cookie` method.
- **Agent action:** Sanitize cookies to Titlecase SameSite before context inject.
**§3v2**
- **Tip:** Instagram/TikTok need `local_storage` in encrypted blob. Call `refresh_silently()` every 45 min during engage sessions.
- **Agent action:** Persist local_storage for IG/TT; refresh_silently every ~45m on long engage.
**§4**
- **Tip:** Chrome persistent context profiles must reside under the workspace (`sessions/profile_...`) to keep user permissions and ensure the sandbox is local to prevent global OS lockups.
- **Agent action:** Keep browser profiles under workspace sessions/ only.
**§4b**
- **Tip:** In local systems, fallback auth to manual GUI window allows users to solve 2FA and visual puzzles directly without needing complex browser automation bypasses.
- **Agent action:** Headful manual window for 2FA; never automate captcha solve.
#### Stealth, humanization & rates
**§5**
- **Tip:** Adaptive rate regulation prevents platforms from detecting automation via fixed interaction frequencies. By combining Fitts's law multipliers and success rate backoffs, the governor dynamically behaves like a human user during normal vs. suspicious intervals.
- **Agent action:** Apply tip before related action; log evidence on fail.
**§8**
- **Tip:** Smart clicks implement Bezier interpolation path generation. Playwright's native mouse clicks click coordinate centroids exactly, which acts as a bot indicator. Smart clicks add padding offsets and trace realistic human curves.
- **Agent action:** Never page.click; use smart_click Bezier + padding offsets.
**§8b**
- **Tip:** Replace uniform `random.uniform` for scroll sessions with burst_rest_pattern.
- **Agent action:** Use Poisson burst_rest for scrolls, not uniform random.
**§11b2**
- **Tip:** Use for cold profile visits during engage; skip for own feed/home.
- **Agent action:** Referrer nav for cold profiles; skip on own feed.
#### Pre-publish gates & media
**§7c**
- **Tip:** Alignment scoring filters spam topics before sending media payload commands. Rejecting out-of-niche drafts avoids generating useless browser footprint data.
- **Agent action:** Reject out-of-niche drafts before browser upload footprint.
**§7d**
- **Tip:** Feed `cache/engagement_analytics.json` from §12f with `post_timestamps`, `recent_posts`, `audience_active_hours`.
- **Agent action:** Feed analytics JSON into algorithm mapper for timing hooks.
**§7f**
- **Tip:** Closes the feedback loop v4.5 `AlgorithmSignalScorer` only scores pre-publish.
- **Agent action:** Pre-publish scorer only; log outcomes after 24h for feedback.
**§11**
- **Tip:** Media file uploads must be tested with absolute workspace paths. The media preparation engine automatically outputs vertical optimized assets to `proofs/optimized/` and replaces the parameter context references.
- **Agent action:** Use absolute media paths; verify magic bytes/headers before upload.
**§12h**
- **Tip:** Media uploading engines can verify file headers before sending raw payloads. This avoids platform upload exceptions from corrupted assets.
- **Agent action:** Apply tip before related action; log evidence on fail.
**§12e**
- **Tip:** Formatting clean captions per platform layout rules ensures optimal text wrap displays on feeds.
- **Agent action:** Run CaptionTransformer per platform limits before post.
#### DOM, rescue & challenges
**§7**
- **Tip:** The Rescue Window draws a red 15px border on the page, alerts the OS sound system, and pauses script execution to give you 5 minutes to solve manual CAPTCHAs directly on the screen.
- **Agent action:** On captcha/2FA: Rescue Window 5m manual only — no bypass bots.
**§7b**
- **Tip:** Platform layouts update frequently. `SelfHealingLocator` caches corrected selectors in `cache/dom_healing.json` to skip semantic scraping on subsequent workflows.
- **Agent action:** Cache healed selectors in cache/dom_healing.json.
#### AI router & platform hacks
**§5c**
- **Tip:** Wire into `AICLIIntegration.generate_caption()` and `BrowserScraper.extract_safe_text()`.
- **Agent action:** Apply tip before related action; log evidence on fail.
**§6**
- **Tip:** If the local Ollama instance does not have mixtral/pixtral models installed, the AI router automatically cascades through the CLI Claude executor, Grok web endpoint, and finally falls back to local templates.
- **Agent action:** Cascade LLM: Ollama → CLI Claude/Grok → templates; no event-loop block.
**§6b**
- **Tip:** Adaptive selection automatically scores hacks based on engagement outcomes. If a "carousel_boost" gets flagged or brings poor reach, the algorithm dampens its boost score and selects alternate hacks like "save_bait".
- **Agent action:** Damp failing platform hacks; switch to save_bait/alt tactics.
#### Publish, first hour & growth
**§0b**
- **Tip:** `publish/grow/engage` never auto-invoke search. Pass `--factual` or set `needs_search: true` in content brief.
- **Agent action:** Only pass `--factual` / `needs_search` when research is required.
**§10**
- **Tip:** Newly created accounts run on tighter limits. The `PhasedGrowth` module restricts all posting, follow, and comment limits to zero for the first 7 days, allowing account warmups before starting automated scheduling.
- **Agent action:** Enforce PhasedGrowth caps; treat playbooks as ceilings not floors.
**§12**
- **Tip:** Executing `first_60_minutes` post-publishing simulates standard human habits. Instantly sharing the newly uploaded post to stories boosts initial organic reach scores.
- **Agent action:** After publish: pin comment, reply, optional story; log metrics 0/15/60m.
**§12b**
- **Tip:** Phoenix Protocol maps execution failures. If a caption triggers anti-spam blocks, the strategy engine morphs the hook or removes links automatically.
- **Agent action:** On caption block: morph hook/remove links once, then stop.
**§11c**
- **Tip:** The relationship engine gates automated growth. Accounts matching "parasitic_risk" are blacklisted from follows and likes to minimize warning risks.
- **Agent action:** Skip NSRE parasitic_risk targets on grow.
**§19**
- **Tip:** New accounts use PhasedGrowth caps (§0c) until day 21 — playbooks are ceilings, not floors.
- **Agent action:** Enforce PhasedGrowth caps; treat playbooks as ceilings not floors.
#### Shadowban, IP & incidents
**§9**
- **Tip:** Check shadowbans using tag exploration pages. If tag queries return empty sets or fail to display the post within the chronological tab, the detector pauses the platform queue for 48 hours.
- **Agent action:** Run tag chrono canary; empty → L3 48h quarantine, no retry loop.
**§12c2**
- **Tip:** Run `check_ip_risk()` at boot on each VPS. Native IP (`proxy: null`) needs no WARP unless IP_RISK warning.
- **Agent action:** Run IP risk at boot; keep proxy:null native unless risk flagged.
**§12c**
- **Tip:** Using free testing nodes like `httpbin.org` prevents latency inflation during initial validation routines.
- **Agent action:** Validate proxies with free test nodes before social traffic.
**§17**
- **Tip:** ErrorCode 301→L1, 308→L2, 310→L3. Logs to `logs/incidents.jsonl`.
- **Agent action:** Map errors to L1 reduce / L2 re-export cookies same IP / L3 halt 48h.
#### Orchestration, schedule & analytics
**§12d**
- **Tip:** Randomized scheduling offsets prevent platform firewalls from flagging regular cron behaviors.
- **Agent action:** Apply tip before related action; log evidence on fail.
**§12g**
- **Tip:** The orchestrator coordinates all standard execution loops: circuit breakers per account and rate limit validations. Sequential execution prevents browser memory spikes.
- **Agent action:** Open circuit after consecutive_failures ≥ max_retry; abort.
**§12f**
- **Tip:** Analytics tracking measures the counts of healthy symbiosis partners over time to check campaign efficiency.
- **Agent action:** Track healthy partner engagement counts for campaign health.
**§13**
- **Tip:** The upgraded CLI supports 13 sub-command modules. The V38 supervisor handles transient reconnection faults via custom exponential backoff delays.
- **Agent action:** Use V38 CLI subcommands; supervisor backoff on transient faults.
#### Telemetry & infra ops
**§2b**
- **Tip:** Circuit breaker state can query `consecutive_failures` counter to automatically open if threshold exceeds `config.max_retry_attempts` without importing complex logic.
- **Agent action:** Open circuit after consecutive_failures ≥ max_retry; abort.
**§2c**
- **Tip:** One `agent_monitor.db` per VPS. AI reads last 5 errors without loading full logs.
- **Agent action:** Read last 5 errors from agent_monitor.db — not full logs.
**§2d**
- **Tip:** 5 MB × 5 backups per VPS. SIGTERM saves checkpoint before browser close.
- **Agent action:** On shutdown: checkpoint then close browser; rotate logs 5×5MB.
**§4c**
- **Tip:** Telemetry logging to AsyncSQLite isolates database calls to separate threads using `asyncio.to_thread`. This guarantees that heavy read/write metrics do not starve Playwright's network loops.
- **Agent action:** Never sync SQLite on Playwright event loop; use to_thread.
#### Additional hybrid tips (not bucketed above)
**§11b**
- **Tip:** Raw page evaluations parse DOM tags efficiently. The scraper returns plain string arrays containing relative matching post pathways.
- **Agent action:** Apply tip before related action; log evidence on fail.
### 18.7.2 Master quick-reference table (all IDs)
| ID | One-line tip (full) |
|----|---------------------|
| §0b | `publish/grow/engage` never auto-invoke search. Pass `--factual` or set `needs_search: true` in content brief. |
| §1 | Add a `COOKIE_DIRS` env-var override so CI and Docker can point to mounted volumes without forking the code: `extra = os.environ.get("OMNI_COOKIE_PATH"); dirs = list(COOKIE_DIRS) + ([Path(extra)] if extra else [])`. This costs zero config for default users but unblocks containerized deployments. |
| §1b | Block `publish` when `boot_warnings` is non-empty unless `--force` is passed. |
| §2c | One `agent_monitor.db` per VPS. AI reads last 5 errors without loading full logs. |
| §2d | 5 MB × 5 backups per VPS. SIGTERM saves checkpoint before browser close. |
| §2b | Circuit breaker state can query `consecutive_failures` counter to automatically open if threshold exceeds `config.max_retry_attempts` without importing complex logic. |
| §3 | Platform cookies are volatile. Storing raw cookies without sanitizing SameSite case triggers Playwright runtime context creation exceptions. Always process using the `sanitize_cookie` method. |
| §3v2 | Instagram/TikTok need `local_storage` in encrypted blob. Call `refresh_silently()` every 45 min during engage sessions. |
| §4 | Chrome persistent context profiles must reside under the workspace (`sessions/profile_...`) to keep user permissions and ensure the sandbox is local to prevent global OS lockups. |
| §4b | In local systems, fallback auth to manual GUI window allows users to solve 2FA and visual puzzles directly without needing complex browser automation bypasses. |
| §4c | Telemetry logging to AsyncSQLite isolates database calls to separate threads using `asyncio.to_thread`. This guarantees that heavy read/write metrics do not starve Playwright's network loops. |
| §5 | Adaptive rate regulation prevents platforms from detecting automation via fixed interaction frequencies. By combining Fitts's law multipliers and success rate backoffs, the governor dynamically behaves like a human user during normal vs. suspicious intervals. |
| §5c | Wire into `AICLIIntegration.generate_caption()` and `BrowserScraper.extract_safe_text()`. |
| §6 | If the local Ollama instance does not have mixtral/pixtral models installed, the AI router automatically cascades through the CLI Claude executor, Grok web endpoint, and finally falls back to local templates. |
| §6b | Adaptive selection automatically scores hacks based on engagement outcomes. If a "carousel_boost" gets flagged or brings poor reach, the algorithm dampens its boost score and selects alternate hacks like "save_bait". |
| §7 | The Rescue Window draws a red 15px border on the page, alerts the OS sound system, and pauses script execution to give you 5 minutes to solve manual CAPTCHAs directly on the screen. |
| §7b | Platform layouts update frequently. `SelfHealingLocator` caches corrected selectors in `cache/dom_healing.json` to skip semantic scraping on subsequent workflows. |
| §7f | Closes the feedback loop v4.5 `AlgorithmSignalScorer` only scores pre-publish. |
| §7d | Feed `cache/engagement_analytics.json` from §12f with `post_timestamps`, `recent_posts`, `audience_active_hours`. |
| §7c | Alignment scoring filters spam topics before sending media payload commands. Rejecting out-of-niche drafts avoids generating useless browser footprint data. |
| §8 | Smart clicks implement Bezier interpolation path generation. Playwright's native mouse clicks click coordinate centroids exactly, which acts as a bot indicator. Smart clicks add padding offsets and trace realistic human curves. |
| §8b | Replace uniform `random.uniform` for scroll sessions with burst_rest_pattern. |
| §9 | Check shadowbans using tag exploration pages. If tag queries return empty sets or fail to display the post within the chronological tab, the detector pauses the platform queue for 48 hours. |
| §10 | Newly created accounts run on tighter limits. The `PhasedGrowth` module restricts all posting, follow, and comment limits to zero for the first 7 days, allowing account warmups before starting automated scheduling. |
| §11 | Media file uploads must be tested with absolute workspace paths. The media preparation engine automatically outputs vertical optimized assets to `proofs/optimized/` and replaces the parameter context references. |
| §11b2 | Use for cold profile visits during engage; skip for own feed/home. |
| §11b | Raw page evaluations parse DOM tags efficiently. The scraper returns plain string arrays containing relative matching post pathways. |
| §11c | The relationship engine gates automated growth. Accounts matching "parasitic_risk" are blacklisted from follows and likes to minimize warning risks. |
| §12 | Executing `first_60_minutes` post-publishing simulates standard human habits. Instantly sharing the newly uploaded post to stories boosts initial organic reach scores. |
| §12b | Phoenix Protocol maps execution failures. If a caption triggers anti-spam blocks, the strategy engine morphs the hook or removes links automatically. |
| §12c2 | Run `check_ip_risk()` at boot on each VPS. Native IP (`proxy: null`) needs no WARP unless IP_RISK warning. |
| §12c | Using free testing nodes like `httpbin.org` prevents latency inflation during initial validation routines. |
| §12d | Randomized scheduling offsets prevent platform firewalls from flagging regular cron behaviors. |
| §12g | The orchestrator coordinates all standard execution loops: circuit breakers per account and rate limit validations. Sequential execution prevents browser memory spikes. |
| §12e | Formatting clean captions per platform layout rules ensures optimal text wrap displays on feeds. |
| §12f | Analytics tracking measures the counts of healthy symbiosis partners over time to check campaign efficiency. |
| §12h | Media uploading engines can verify file headers before sending raw payloads. This avoids platform upload exceptions from corrupted assets. |
| §14b | Complements `self_check.py` (import test) with runtime ops monitoring. |
| §13 | The upgraded CLI supports 13 sub-command modules. The V38 supervisor handles transient reconnection faults via custom exponential backoff delays. |
| §14 | Testing suite exercises modules in isolation. Invoking `python self_check.py` parses compiler trees and checks basic instance creations. |
| §17 | ErrorCode 301→L1, 308→L2, 310→L3. Logs to `logs/incidents.jsonl`. |
| §19 | New accounts use PhasedGrowth caps (§0c) until day 21 — playbooks are ceilings, not floors. |
*Catalog count: **42** — full text retained (no ellipsis truncation).*
### 18.7.3 How agents should use this pack
```
1. Load power rules P1–P20 into working memory for every social task.
2. Before action domain X → open matching battle card only (not whole hybrid file).
3. On failure → map to §17 incident + agent_monitor last 5 errors (§2c).
4. Prefer omni_cli gates for publish/grow; apply P5 STEALTH_GAP if Omni still uses page.click.
5. Module source of truth remains new-hybrid-4.7.md; tips here are the operational edge.
```
### 18.8 Stack “4 Golden Rules” (Ultimate hybrid stack)
1. **Warm-Up Rule** — Never run an autonomous agent on a brand-new account. 3–7 days manual human use first.
2. **Proxy-Geo Match** — Proxy country ≡ browser timezone ≡ locale. Mismatch = instant risk.
3. **Jitter Rule** — Humans don’t click in 50ms. Scroll past, scroll back, pause ~2.4s, then click.
4. **Session Persistence** — Always `storage_state` / encrypted cookies. Login/logout loops trigger CAPTCHA.
Plus: residential proxies for social (not datacenter); headed uploads; deterministic steps when known, agent when not.
---
### 18.9 Slim skill error-fixation (automate + search + media)
**Common loop:** Reproduce → Isolate → Fix → Verify.
| Domain | Fast self-check | Top cause |
|--------|-----------------|-----------|
| Automate | `publish.py self-check` / `self_check.py` | selector timeout = cookie expiry first |
| Search | `python3 -m validate.clear_gate` | version mismatch / raw query |
| Media | ffprobe first; never claim done without file + qa_report | VFR/A-V drift if editing |
**Handoff:** Search writes `cache/handoff_package.json` / `research_*.json` → automate merges into caption — agent must **read handoff**, not re-hallucinate facts.
---
### 18.10 Decision trees (quick)
#### Should I open a browser?
```
Need live site action? NO → do research/content offline
YES → boot PASS? NO → fix boot
YES → session healthy? NO → cookie recovery path
YES → continue pipeline
```
#### Should I publish now?
```
quarantined? → NO
PhasedGrowth allows? → else wait
PrePublish + virality OK? → else rewrite
boot_warnings empty? → else fix or --force with reason
dry-run once on new template? → then live
```
#### Should I call search?
```
User --factual OR needs_search OR verify claim? → YES (mode search)
Else → NO (stay automate)
```
#### Can I say “done”?
```
All required checks PASS?
Evidence exists (URL, path, screenshot, score)?
No INCONCLUSIVE remaining?
→ only then report done with evidence pack
```
---
### 18.11 Agent self-talk checklist (paste into chain-of-thought)
```
[ ] Which path A/B/C?
[ ] Account + VPS pin known?
[ ] boot/self-check result?
[ ] Session encrypted + not quarantined?
[ ] Using smart_click not page.click?
[ ] Warm_up done if publishing?
[ ] Gates scored (virality/prepublish)?
[ ] Search justified if used?
[ ] Caption cleaned of GPT-isms?
[ ] Media absolute path validated?
[ ] Rate/PhasedGrowth allows action?
[ ] After act: evidence captured?
[ ] Verification three-state recorded?
[ ] If fail: classified + not ban-looping?
```
---
### 18.12 Permanent memory directives (cross-check)
From `hybrid_47_social_memory.json` (live operator memory):
1. Self-check boot() before any browser context
2. Never raw page.click() — always smart_click() with cubic Bezier paths
3. No platform APIs — browser-only (cookies → credentials headful → manual 2FA)
4. CircuitBreaker: abort after 3 consecutive publish/grow failures
5. Encrypt cookies at rest; sanitize sameSite to Titlecase (Lax/Strict/None)
6. Session quarantine on shadowban|checkpoint|session_expired|account_locked
7. Strip GPT-isms: delve, testament, tapestry, moreover, unlock, landscape, elevate
8. Three-state verdict: PASS|FAIL|INCONCLUSIVE — never treat INCONCLUSIVE as PASS
9. Lite search OFF by default; only with --factual or needs_search:true
10. ethics.yaml passive_only:true blocks gray-hat at boot
11. One account → one VPS IP forever (anti-link); Mode C server entry for automation
**anti_block.never:**
- Burst retry on 429
- Share home IP across many accounts for live automation
- Continue after ban/checkpoint without human solve
- Use platform official APIs
- Skip warm_up before cold publish
**expert_notes.never:**
- Platform APIs (instagrapi/tweepy/Graph API)
- Raw page.click()
- Treat INCONCLUSIVE as PASS
- Captcha bypass automation
- Share one home IP across unlinked brand accounts for live publish
- Retry 429/shadowban in tight loops
---
### 18.13 Micro-prompts (one-liners to steer the agent)
```
BOOT FIRST: "Run boot and report ready/warnings before any browser."
CLICK LAW: "Replace every page.click with smart_click Bezier."
NO API: "Refuse platform API SDKs; browser cookies only."
GATE: "Do not publish if virality < 60 or PrePublish rejects."
SEARCH OFF: "Do not search unless I pass --factual."
QUARANTINE: "On checkpoint/shadowban stop and quarantine; no retry loop."
EVIDENCE: "Done requires post URL + three-state PASS."
DRY RUN: "Dry-run once before first live template."
PATH C: "Single account only; drafts not auto-send."
FIX ROOT: "Classify error; fix root; re-verify; max 3 attempts."
```
---
### 18.14 How this prevents the usual agent mistakes
| Usual agent mistake | Rule that blocks it |
|---------------------|---------------------|
| “Published” without URL | Evidence pack required |
| Infinite login retries | Quarantine + no ban loop |
| Spams search every publish | Mode router lite search OFF |
| Clicks like a robot | smart_click / jitter / warm_up |
| Uses instagrapi “because easier” | No platform APIs |
| Treats timeout as success | INCONCLUSIVE blocks done |
| Burns new accounts day 1 | PhasedGrowth posts=0 |
| Mixes DE proxy + US timezone | Geo match golden rule |
| Auto-sends bad DM replies | Path C draft default |
| Hallucinates stats in caption | CLEAR ≥65 / ≥2 domains |
| Disables scorer to ship | Anti-mutation / gate law |
---
## 19. OMNI + HYBRID Runtime Merge (IMPLEMENTED)
> **Package:** `omni-hybrid-merged/` next to this file
> **Hands:** `../social media omni credentials/media omni agents/`
> **Brain:** this document (§1, §4, §6–7, §18)
> **Glue:** thin gates + `omni_cli.py` + `SYSTEM_PROMPT.md` + `smart_click.py`
### 19.1 What was merged
| Layer | Location | Role |
|-------|----------|------|
| Constitution | this file | Laws, prompts, playbooks |
| Executors | media omni agents `*.py` | Real Playwright publish/reply/keep-alive/grow |
| Gates | `omni-hybrid-merged/gates.py` | PhasedGrowth, quarantine, caption, routing, 3-state |
| CLI | `omni-hybrid-merged/omni_cli.py` | Single entry: gate then subprocess executor |
| AI prompt | `omni-hybrid-merged/SYSTEM_PROMPT.md` | Sticky system prompt for CLI agents |
| Stealth helper | `omni-hybrid-merged/smart_click.py` | Bezier click (patch Omni over time) |
| Comparison | `COMPARISON-OMNI-AGENTS-vs-UNIFIED.md` | Score tables |
### 19.2 Commands (run from package dir)
```bash
cd ".../the full merged manager hybrid/omni-hybrid-merged"
python3 omni_cli.py boot
python3 omni_cli.py publish --account master1_vip1 --platform tiktok --now
python3 omni_cli.py grow --account acc1 --platform twitter
python3 omni_cli.py keep-alive
python3 omni_cli.py dashboard # :8080
python3 omni_cli.py responder # :8081
python3 omni_cli.py prompt # print agent system prompt
python3 check_merge.py
```
Env: `OMNI_AGENTS_DIR` · `OMNI_COOKIE_DIR` · `OMNI_PROJECT_DIR`
Omni `hybrid_scheduler.py` / `session_sync_keep_alive.py` / `master_fix.py` honor these env vars after merge patch.
### 19.3 Agent rule
```
Load SYSTEM_PROMPT.md → boot → ONE omni_cli command → evidence → PASS|FAIL|INCONCLUSIVE
```
Never bypass gates by calling Omni scripts directly for publish/grow unless debugging; prefer `omni_cli.py` so PhasedGrowth and quarantine apply.
### 19.4 Remaining gaps (honest)
- Omni scripts still contain raw `page.click` in places — flag STEALTH_GAP; migrate to `smart_click` incrementally.
- Plaintext `credentials.json` must be rotated off disk (security).
- Full Hybrid 4.7 modules are not all running as live Python — gates are the thin bridge; expand only when pain is real.
---
## Appendix A — Full Hybrid Module Index (pointer)
Complete sources live in `new-hybrid-4.7.md`:
`boot · secrets_scanner · cookie_manager · cookie_chain · server_router · stealth_browser · dispatcher · telemetry · circuit_breaker · rate_governor · omni_memory · omni_search · ai_cli · platform_hacks · anti_detection · dom_locator · media_prep · topic_sanity · prepublish_scorer · algorithm_signal_scorer · session_health · algorithm_mapper · engagement_hook · humanization · shadowban · phased_growth · virality_scorer · content_validator · browser_automation · browser_scraper · v38_hybrid · engagement_engine · strategy_engines · proxy_rotator · ip_reputation · content_calendar · orchestrator · caption_transformer · analytics · verify_engine · v38_system · publish.py · self_check · incident_response · ethics.yaml`
## Appendix B — Social-Autopilot Layout
```
social-autopilot/
run.py
config/settings.example.yaml
content/queue.example.json
core/
browser_session.py publisher.py scheduler.py
rate_limiter.py session_store.py
replier.py analytics.py
```
## Appendix C — Related source files in this folder
| File | Role |
|------|------|
| `new-hybrid-4.7.md` | Full product code spec |
| `Ultimate hybrid stack-media automation.md` | Engine decision tree + hybrid install |
| `Social Media Mastery Guide.md` | Research synthesis + defense |
| `Advanced media overgrowth.md` | Deep growth tactics |
| `social media overgrowth.md` | Compact viral playbook |
| `Lead generation and media automation.md` | Leads + ads |
| `pro-mistreal-searcher-agent-extension.md` | Research agent extension |
| `unified-slim-package/*` | Slim skill gates + SLIM V2/V3 |
| `social-autopilot/*` | Minimal runnable Path C |
| `Content_Growth_Tracker.xlsx` | Metrics tracker companion |
## Appendix D — Document history
| Ver | Date | Change |
|-----|------|--------|
| 1.0 | 2026-08-01 | Initial best-of merge of manager hybrid folder into one extended operator manual |
| 1.1 | 2026-08-01 | Full rescan of 40 sources; §18 agent golden rules, error map, task prompts, PRO TIP catalog |
| 1.2 | 2026-08-01 | Runtime merge: omni-hybrid-merged CLI + gates + SYSTEM_PROMPT; Omni env path patches |
| 1.3 | 2026-08-01 | Full Hybrid 4.7 PRO TIP adoption: 20 power rules + untruncated battle cards + agent actions |
---
*End of FULL-MERGED-MANAGER-HYBRID-UNIFIED.md — use Path C for honest single-account ops; Path A when multi-account productization is required; always keep the Operating Contract above any growth hack.*