${esc(d.field_count)} fields analyzed
Selected
Entity: ${esc((selected.entity || []).join(', ') || '-')}
Time: ${esc((selected.time || []).join(', ') || '-')}
Categorical: ${esc((selected.categorical || []).join(', ') || '-')}
Numeric: ${esc((selected.numeric || []).join(', ') || '-')}
Shared across streams
${esc(shared || '-')}
Advisor roles
${esc(fieldRoles || '-')}
${esc(correlationRoles || '')}
Top fields
${esc(top || '-')}
Rejected
${esc(rejected || '-')}
${esc(reasons || '')}
`;
+ return `${esc(d.field_count)} fields analyzed
Selected
Entity: ${esc((selected.entity || []).join(', ') || '-')}
Time: ${esc((selected.time || []).join(', ') || '-')}
Categorical: ${esc((selected.categorical || []).join(', ') || '-')}
Numeric: ${esc((selected.numeric || []).join(', ') || '-')}
Relationships: ${esc((selected.relationships || []).join(', ') || '-')}
Shared across streams
${esc(shared || '-')}
Advisor roles
${esc(fieldRoles || '-')}
${esc(correlationRoles || '')}
Top fields
${esc(top || '-')}
Rejected
${esc(rejected || '-')}
${esc(reasons || '')}
`;
}
function sharedFieldLabel(item) {
const aliases = (item.aliases || []).filter(alias => alias !== item.field).slice(0,4);
@@ -614,6 +614,16 @@ function renderStreamPicker(errorText='') {
}
function mergeProfile(existing, recommended) {
const mergeList = (left, right) => [...new Set([...(left || []), ...(right || [])].filter(Boolean))];
+ const mergeRelationships = (left, right) => {
+ const rows = [...(left || []), ...(right || [])].filter(item => item && item.left && item.right);
+ const seen = new Set();
+ return rows.filter(item => {
+ const key = `${String(item.left).toLowerCase()}->${String(item.right).toLowerCase()}`;
+ if (seen.has(key)) return false;
+ seen.add(key);
+ return true;
+ });
+ };
const detectors = {...(recommended.detectors || {}), ...(existing.detectors || {})};
return {
...recommended,
@@ -626,7 +636,7 @@ function mergeProfile(existing, recommended) {
numeric_fields: mergeList(existing.numeric_fields, recommended.numeric_fields),
detectors,
field_weights: {...(recommended.field_weights || {}), ...(existing.field_weights || {})},
- relationship_fields: existing.relationship_fields || recommended.relationship_fields || []
+ relationship_fields: mergeRelationships(existing.relationship_fields, recommended.relationship_fields)
};
}
async function applySuggestedProfile(streamId) {
diff --git a/src/fgai/profile_suggestions.py b/src/fgai/profile_suggestions.py
index c9a7106..40b8aff 100644
--- a/src/fgai/profile_suggestions.py
+++ b/src/fgai/profile_suggestions.py
@@ -421,6 +421,95 @@ def _generic_numeric_fields(coverage: Counter[str], numeric_counts: Counter[str]
return output
+def _relationship_name(left: str, right: str) -> str:
+ return f"{left} to {right}"
+
+
+def _field_matches(field: str, groups: set[str], tokens: set[str]) -> bool:
+ lower = field.lower()
+ semantic = set(_semantic_field_groups(lower))
+ return bool(semantic & groups) or any(token in lower for token in tokens)
+
+
+def _relationship_candidates(
+ coverage: Counter[str],
+ unique_values: dict[str, set[str]],
+ total: int,
+ entity_fields: list[str],
+ categorical_fields: list[str],
+ numeric_fields: list[str],
+ *,
+ limit: int = 6,
+) -> list[dict[str, str]]:
+ def usable(field: str) -> bool:
+ return (
+ bool(field)
+ and field in coverage
+ and field not in IGNORED_DISCOVERY_FIELDS
+ and coverage[field] / max(1, total) >= 0.02
+ and len(unique_values[field]) > 1
+ )
+
+ candidates: list[tuple[str, str]] = []
+ entities = [field for field in entity_fields if usable(field)]
+ categorical = [field for field in categorical_fields if usable(field)]
+ numeric = [field for field in numeric_fields if usable(field)]
+
+ identity_fields = [
+ field for field in entities
+ if _field_matches(field, {"username", "*_user"}, {"user", "account", "actor", "principal", "login", "identity"})
+ ]
+ source_fields = [
+ field for field in entities
+ if _field_matches(field, {"srcip", "*_ip"}, {"src", "source", "client", "remote", "ipaddress", "clientip"})
+ ]
+ host_fields = [
+ field for field in entities
+ if _field_matches(field, {"*_host"}, {"host", "hostname", "computer", "workstation", "device"})
+ ]
+ destination_fields = [
+ field for field in [*entities, *categorical]
+ if _field_matches(field, {"dstip", "*_ip"}, {"dst", "dest", "destination", "server"})
+ ]
+ destination_fields.extend(field for field in categorical if field.lower() in {"dstport", "destination_port", "dest_port", "service", "application", "app", "url", "query_domain", "dns_query"})
+
+ for left in identity_fields:
+ for right in [*source_fields, *host_fields]:
+ if left != right:
+ candidates.append((left, right))
+ for left in source_fields:
+ for right in destination_fields:
+ if left != right:
+ candidates.append((left, right))
+ for left in host_fields:
+ for right in [field for field in categorical if any(token in field.lower() for token in ("process", "service", "app", "action", "status", "event"))]:
+ if left != right:
+ candidates.append((left, right))
+
+ behavior_fields = [
+ field for field in [*categorical, *numeric]
+ if field not in entities
+ and not _looks_like_time_field(field)
+ and not any(token in field.lower() for token in ("gl2_", "message_id", "sort_field", "accounted_message_size", "processing_timestamp"))
+ ]
+ for left in entities:
+ for right in behavior_fields[:4]:
+ if left != right:
+ candidates.append((left, right))
+
+ output = []
+ seen = set()
+ for left, right in candidates:
+ key = (left.lower(), right.lower())
+ if key in seen:
+ continue
+ seen.add(key)
+ output.append({"left": left, "right": right, "name": _relationship_name(left, right)})
+ if len(output) >= limit:
+ break
+ return output
+
+
def _shared_profile_fields(
coverage: Counter[str],
unique_values: dict[str, set[str]],
@@ -723,11 +812,12 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
detectors.setdefault("auth_failure", {"enabled": True, "minimum": 5, "z_threshold": 3.0})
if not detectors:
detectors = {"deny_action": {"enabled": True, "minimum": 10, "z_threshold": 3.0}}
+ relationship_fields = _relationship_candidates(coverage, unique_values, total, entity_fields, categorical, numeric)
high_coverage = [
{"field": field, "coverage": round(count / max(1, total), 2), "unique_values": len(unique_values[field])}
for field, count in coverage.most_common(12)
]
- selected_fields = {timestamp, *entity_fields, *categorical, *numeric}
+ selected_fields = {timestamp, *entity_fields, *categorical, *numeric, *(item["left"] for item in relationship_fields), *(item["right"] for item in relationship_fields)}
discovery = {
"log_type": log_type,
"field_count": len([field for field in coverage if field not in IGNORED_DISCOVERY_FIELDS]),
@@ -736,13 +826,14 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
"time": [timestamp] if timestamp else [],
"categorical": categorical,
"numeric": numeric,
+ "relationships": [f"{item['left']}->{item['right']}" for item in relationship_fields],
},
"top_fields": _field_stats(coverage, unique_values, numeric_counts, total)[:16],
"shared_fields": shared_stats,
"rejected_fields": _rejected_fields(coverage, unique_values, numeric_counts, total, selected_fields),
"reasons": [
f"detected {log_type} log pattern",
- f"selected {len(entity_fields)} entity field(s), {len(categorical)} categorical field(s), and {len(numeric)} numeric field(s)",
+ f"selected {len(entity_fields)} entity field(s), {len(categorical)} categorical field(s), {len(numeric)} numeric field(s), and {len(relationship_fields)} relationship(s)",
f"found {len(shared_stats)} field(s) that also appear in other enabled streams",
"ignored raw/internal, constant, sparse, and very high-cardinality fields",
],
@@ -760,6 +851,7 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
"timestamp_field": timestamp,
"categorical_fields": categorical,
"numeric_fields": numeric,
+ "relationship_fields": relationship_fields,
"detectors": detectors,
"common_fields": high_coverage,
"shared_fields": shared_stats,
@@ -775,7 +867,7 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
"numeric_fields": numeric,
"detectors": detectors,
"field_weights": {},
- "relationship_fields": [],
+ "relationship_fields": relationship_fields,
},
"reason": "selected common entity, time, categorical, and numeric fields from observed events; non-priority fields are included when coverage and cardinality look useful",
})
diff --git a/tests/test_profile_suggestions.py b/tests/test_profile_suggestions.py
index 0d3a6e4..eec1166 100644
--- a/tests/test_profile_suggestions.py
+++ b/tests/test_profile_suggestions.py
@@ -221,6 +221,34 @@ class ProfileSuggestionTests(unittest.TestCase):
self.assertIn("*_status", shared)
self.assertIn("*_name", shared)
+ def test_heuristic_profile_recommends_behavior_relationships(self):
+ events = [
+ parse_log_line(
+ f"fgai_stream_id=windows fgai_stream=Windows username=user{index % 3} srcip=10.0.0.{index % 5} hostname=host{index % 4} action=success eventid=4624"
+ )
+ for index in range(1, 40)
+ ]
+
+ suggestion = suggest_stream_profiles(events)[0]
+ relationships = {(item["left"], item["right"]) for item in suggestion["relationship_fields"]}
+
+ self.assertIn(("username", "srcip"), relationships)
+ self.assertIn("username->srcip", suggestion["discovery"]["selected_fields"]["relationships"])
+ self.assertEqual(suggestion["profile"]["relationship_fields"], suggestion["relationship_fields"])
+
+ def test_heuristic_profile_recommends_generic_entity_behavior_relationships(self):
+ events = [
+ parse_log_line(
+ f"fgai_stream_id=app fgai_stream=App entry_context_clientipaddress=10.0.0.{index % 5} entry_context_type={'login' if index % 2 else 'update'} houston_category=cat{index % 3}"
+ )
+ for index in range(1, 40)
+ ]
+
+ suggestion = suggest_stream_profiles(events)[0]
+ relationships = {(item["left"], item["right"]) for item in suggestion["relationship_fields"]}
+
+ self.assertIn(("entry_context_clientipaddress", "entry_context_type"), relationships)
+
def test_applies_valid_llm_advice_and_rejects_unknown_fields(self):
suggestion = suggest_stream_profiles([
parse_log_line("fgai_stream_id=windows fgai_stream=Windows username=alice hostname=host01 eventid=4625 action=failure")